Merge 54cedbd into d24d121

CHANGES.txt

@@ -5,6 +5,9 @@ Changes for crash
 Unreleased
 ==========
 
+ - Check if certificate related files exist and have read permissions before
+   entering Crash prompt.
+
 2017/06/27 0.21.0
 =================
 

src/crate/crash/command.py

@@ -99,8 +99,8 @@ def parse_args(parser):
         argcomplete.autocomplete(parser)
     except ImportError:
         pass
-    args = parser.parse_args()
-    return args
+
+    return parser.parse_args()
 
 
 def boolean(v):
@@ -151,11 +151,11 @@ def _conf_or_default(key, value):
                         help='the crate hosts to connect to', metavar='HOST')
     parser.add_argument('--verify-ssl', type=boolean, default=True,
                         help='force verification of the SSL certificate of the server')
-    parser.add_argument('--cert-file', type=str,
+    parser.add_argument('--cert-file', type=file_with_permissions,
                         help='path to the client certificate file')
-    parser.add_argument('--key-file', type=str,
+    parser.add_argument('--key-file', type=file_with_permissions,
                         help='path to the key file of the client certificate')
-    parser.add_argument('--ca-cert-file', type=str,
+    parser.add_argument('--ca-cert-file', type=file_with_permissions,
                         help='path to the CA certificate file')
     parser.add_argument('--format', type=str,
                         default=_conf_or_default('format', 'tabular'),
@@ -442,7 +442,11 @@ def main():
         parser.print_usage()
         sys.exit(1)
     parser = get_parser(output_writer.formats, conf=conf)
-    args = parse_args(parser)
+    try:
+        args = parse_args(parser)
+    except Exception as e:
+        printer.warn(str(e))
+        sys.exit(1)
     output_writer.output_format = args.format
 
     if args.version:
@@ -499,5 +503,9 @@ def _create_cmd(crate_hosts, error_trace, output_writer, is_tty, args):
                     ca_cert_file=args.ca_cert_file,
                     username=args.username)
 
+def file_with_permissions(path):
+    open(path, 'r').close()
+    return path
+
 if __name__ == '__main__':
     main()

src/crate/crash/test_command.py

@@ -532,12 +532,21 @@ def test_username_param(self):
         self.assertEqual(crateCmd.connection.client.username, "testUser")
 
     def test_ssl_params(self):
+        tmpdirname = tempfile.mkdtemp()
+        cert_filename = os.path.join(tmpdirname, "cert_file")
+        key_filename = os.path.join(tmpdirname, "key_file")
+        ca_cert_filename = os.path.join(tmpdirname, "ca_cert_file")
+
+        open(cert_filename, 'a').close()
+        open(key_filename, 'a').close()
+        open(ca_cert_filename, 'a').close()
+
         sys.argv = ["testcrash",
                     "--hosts", self.crate_host,
                     "--verify-ssl", "false",
-                    "--cert-file", "cert_file",
-                    "--key-file", "key_file",
-                    "--ca-cert-file", "ca_cert_file"
+                    "--cert-file", cert_filename,
+                    "--key-file", key_filename,
+                    "--ca-cert-file", ca_cert_filename
                     ]
         parser = get_parser()
         args = parse_args(parser)
@@ -548,14 +557,55 @@ def test_ssl_params(self):
         self.assertEqual(crateCmd.verify_ssl, False)
         self.assertEqual(crateCmd.connection.client._pool_kw['cert_reqs'], ssl.CERT_NONE)
 
-        self.assertEqual(crateCmd.cert_file, 'cert_file')
-        self.assertEqual(crateCmd.connection.client._pool_kw['cert_file'], 'cert_file')
+        self.assertEqual(crateCmd.cert_file, cert_filename)
+        self.assertEqual(crateCmd.connection.client._pool_kw['cert_file'], cert_filename)
+
+        self.assertEqual(crateCmd.key_file, key_filename)
+        self.assertEqual(crateCmd.connection.client._pool_kw['key_file'], key_filename)
+
+        self.assertEqual(crateCmd.ca_cert_file, ca_cert_filename)
+        self.assertEqual(crateCmd.connection.client._pool_kw['ca_certs'], ca_cert_filename)
+
+
+    def test_ssl_params_missing_file(self):
+        sys.argv = ["testcrash",
+                    "--hosts", self.crate_host,
+                    "--verify-ssl", "false",
+                    "--key-file", "wrong_file",
+                    "--ca-cert-file", "ca_cert_file"
+                    ]
+        parser = get_parser()
+
+        # Python 2
+        try:
+            FileNotFoundError
+        except NameError:
+            FileNotFoundError = IOError
+
+        with self.assertRaises(FileNotFoundError):
+            parse_args(parser)
+
+    def test_ssl_params_wrong_permision_file(self):
+        tmpdirname = tempfile.mkdtemp()
+        ca_cert_filename = os.path.join(tmpdirname, "ca_cert_file")
+        open(ca_cert_filename, 'a').close()
+        os.chmod(ca_cert_filename, 0000)
+
+        sys.argv = ["testcrash",
+                    "--hosts", self.crate_host,
+                    "--verify-ssl", "false",
+                    "--ca-cert-file", ca_cert_filename
+                    ]
+        parser = get_parser()
 
-        self.assertEqual(crateCmd.key_file, 'key_file')
-        self.assertEqual(crateCmd.connection.client._pool_kw['key_file'], 'key_file')
+        # Python 2
+        try:
+            PermissionError
+        except NameError:
+            PermissionError = IOError
 
-        self.assertEqual(crateCmd.ca_cert_file, 'ca_cert_file')
-        self.assertEqual(crateCmd.connection.client._pool_kw['ca_certs'], 'ca_cert_file')
+        with self.assertRaises(PermissionError):
+            parse_args(parser)
 
 
 class TestGetInformationSchemaQuery(TestCase):