MB-65941 Do not send request if ssl-manage --pkey_settings not provided

Do not send empty requests for `ssl-manage --set-client-certificate`
and `ssl-manage --set-node-certificate` when `--pkey_settings` is not
provided.

Change-Id: I799e516e00762a1063e5f39f610284871d0f1a08
Reviewed-on: https://review.couchbase.org/c/couchbase-cli/+/225298
Reviewed-by: Matt Hall <matt.hall@couchbase.com>
Tested-by: Build Bot <build@couchbase.com>

Author: lubomarinski

cbmgr.py

@@ -4367,11 +4367,19 @@ def val_or_unknown(ca, key):
             _exit_if_errors(errors)
             _success(f'Uploaded cluster certificate to {opts.cluster}')
         elif opts.set_cert:
-            _, errors = self.rest.set_certificate(_read_json_file_if_provided(opts.pkey_settings), is_client_cert=False)
+            if opts.pkey_settings is None:
+                _exit_if_errors(['--pkey-passphrase-settings is required'])
+
+            _, errors = self.rest.set_certificate(_exit_on_json_file_read_failure(opts.pkey_settings),
+                                                  is_client_cert=False)
             _exit_if_errors(errors)
             _success("Node certificate set")
         elif opts.set_client_cert:
-            _, errors = self.rest.set_certificate(_read_json_file_if_provided(opts.pkey_settings), is_client_cert=True)
+            if opts.pkey_settings is None:
+                _exit_if_errors(['--pkey-passphrase-settings is required'])
+
+            _, errors = self.rest.set_certificate(_exit_on_json_file_read_failure(opts.pkey_settings),
+                                                  is_client_cert=True)
             _exit_if_errors(errors)
             _success("Internal client certificate set")
         elif opts.client_auth_path:

test/test_cli.py

@@ -619,8 +619,8 @@ def test_bucket_create_encryption(self):
         expected_params = [
             'bucketType=couchbase', 'name=name', 'evictionPolicy=fullEviction', 'replicaNumber=0',
             'ramQuotaMB=100', 'storageBackend=magma', 'rank=3', 'numVBuckets=128',
-            'encryptionAtRestKeyId=2', f'encryptionAtRestDekRotationInterval={30*24*60*60}',
-            f'encryptionAtRestDekLifetime={60*24*60*60}',
+            'encryptionAtRestKeyId=2', f'encryptionAtRestDekRotationInterval={30 * 24 * 60 * 60}',
+            f'encryptionAtRestDekLifetime={60 * 24 * 60 * 60}',
         ]
         self.rest_parameter_match(expected_params)
 
@@ -793,8 +793,8 @@ def test_bucket_edit_encryption_settings(self):
         ]
         self.no_error_run(self.command + self.command_args + args, self.server_args)
         expected_params = [
-            'encryptionAtRestKeyId=2', f'encryptionAtRestDekRotationInterval={30*24*60*60}',
-            f'encryptionAtRestDekLifetime={60*24*60*60}',
+            'encryptionAtRestKeyId=2', f'encryptionAtRestDekRotationInterval={30 * 24 * 60 * 60}',
+            f'encryptionAtRestDekLifetime={60 * 24 * 60 * 60}',
         ]
         self.rest_parameter_match(expected_params)
 
@@ -1360,15 +1360,15 @@ def test_set_encryption(self):
                 '--dek-lifetime', '60']
         self.no_error_run(self.command + args, self.server_args)
         expected_params = ['log.encryptionMethod=encryptionKey', 'log.encryptionKeyId=2',
-                           f'log.dekRotationInterval={30*24*60*60}', f'log.dekLifetime={60*24*60*60}']
+                           f'log.dekRotationInterval={30 * 24 * 60 * 60}', f'log.dekLifetime={60 * 24 * 60 * 60}']
         self.rest_parameter_match(expected_params)
 
     def test_set_encryption_master_password(self):
         args = ['--set', '--type', 'master-password', '--target', 'log', '--dek-rotate-every', '30',
                 '--dek-lifetime', '60']
         self.no_error_run(self.command + args, self.server_args)
         expected_params = ['log.encryptionMethod=nodeSecretManager',
-                           f'log.dekRotationInterval={30*24*60*60}', f'log.dekLifetime={60*24*60*60}']
+                           f'log.dekRotationInterval={30 * 24 * 60 * 60}', f'log.dekLifetime={60 * 24 * 60 * 60}']
         self.rest_parameter_match(expected_params)
 
     def test_add_edit_key_no_name(self):
@@ -2310,27 +2310,73 @@ def test_regenerate_cert_not_init(self):
         self.assertIn('POST:/controller/regenerateCertificate', self.server.trace)
         self.assertIn('Certificate regenerate and copied to `node1.pem`', self.str_output)
 
+    def test_set_node_certificate_missing(self):
+        self.system_exit_run(self.command + ['--set-node-certificate'], self.server_args)
+        self.assertIn('--pkey-passphrase-settings is required', self.str_output)
+
+    def test_set_node_certificate_invalid_json(self):
+        with tempfile.NamedTemporaryFile() as link_options_file:
+            link_options_file.write(b'{123:123}')
+            link_options_file.flush()
+
+            self.system_exit_run(self.command + ['--set-node-certificate', '--pkey-passphrase-settings',
+                                                 link_options_file.name], self.server_args)
+            self.assertIn('does not contain valid JSON data', self.str_output)
+
     def test_set_node_certificate(self):
-        self.no_error_run(self.command + ['--set-node-certificate'], self.server_args)
-        self.assertIn('POST:/node/controller/reloadCertificate', self.server.trace)
-        self.assertIn('Node certificate set', self.str_output)
+        with tempfile.NamedTemporaryFile() as link_options_file:
+            link_options_file.write(b'{"asd":123}')
+            link_options_file.flush()
+
+            self.no_error_run(self.command + ['--set-node-certificate', '--pkey-passphrase-settings',
+                                              link_options_file.name], self.server_args)
+            self.assertIn('POST:/node/controller/reloadCertificate', self.server.trace)
+            self.assertIn('Node certificate set', self.str_output)
 
     def test_set_node_certificate_not_init(self):
-        self.server_args['init'] = False
-        self.no_error_run(self.command + ['--set-node-certificate'], self.server_args)
-        self.assertIn('POST:/node/controller/reloadCertificate', self.server.trace)
-        self.assertIn('Node certificate set', self.str_output)
+        with tempfile.NamedTemporaryFile() as link_options_file:
+            link_options_file.write(b'{"asd":123}')
+            link_options_file.flush()
+
+            self.server_args['init'] = False
+            self.no_error_run(self.command + ['--set-node-certificate', '--pkey-passphrase-settings',
+                                              link_options_file.name], self.server_args)
+            self.assertIn('POST:/node/controller/reloadCertificate', self.server.trace)
+            self.assertIn('Node certificate set', self.str_output)
+
+    def test_set_client_certificate_missing(self):
+        self.system_exit_run(self.command + ['--set-client-certificate'], self.server_args)
+        self.assertIn('--pkey-passphrase-settings is required', self.str_output)
+
+    def test_set_client_certificate_invalid_json(self):
+        with tempfile.NamedTemporaryFile() as link_options_file:
+            link_options_file.write(b'{123:123}')
+            link_options_file.flush()
+
+            self.system_exit_run(self.command + ['--set-client-certificate', '--pkey-passphrase-settings',
+                                                 link_options_file.name], self.server_args)
+            self.assertIn('does not contain valid JSON data', self.str_output)
 
     def test_set_client_certificate(self):
-        self.no_error_run(self.command + ['--set-client-certificate'], self.server_args)
-        self.assertIn('POST:/node/controller/reloadClientCertificate', self.server.trace)
-        self.assertIn('Internal client certificate set', self.str_output)
+        with tempfile.NamedTemporaryFile() as link_options_file:
+            link_options_file.write(b'{"asd":123}')
+            link_options_file.flush()
+
+            self.no_error_run(self.command + ['--set-client-certificate', '--pkey-passphrase-settings',
+                                              link_options_file.name], self.server_args)
+            self.assertIn('POST:/node/controller/reloadClientCertificate', self.server.trace)
+            self.assertIn('Internal client certificate set', self.str_output)
 
     def test_set_client_certificate_not_init(self):
-        self.server_args['init'] = False
-        self.no_error_run(self.command + ['--set-client-certificate'], self.server_args)
-        self.assertIn('POST:/node/controller/reloadClientCertificate', self.server.trace)
-        self.assertIn('Internal client certificate set', self.str_output)
+        with tempfile.NamedTemporaryFile() as link_options_file:
+            link_options_file.write(b'{"asd":123}')
+            link_options_file.flush()
+
+            self.server_args['init'] = False
+            self.no_error_run(self.command + ['--set-client-certificate', '--pkey-passphrase-settings',
+                                              link_options_file.name], self.server_args)
+            self.assertIn('POST:/node/controller/reloadClientCertificate', self.server.trace)
+            self.assertIn('Internal client certificate set', self.str_output)
 
     def test_set_node_certificate_with_pkey_settings(self):
         pkey_settings_file = tempfile.NamedTemporaryFile(delete=False)
@@ -2603,7 +2649,7 @@ def succ_cmd(self, addrFamily):
 
 def read_password(sock):
     (result, remoteaddr) = sock.recvfrom(128)
-    assert(result == b'asdasd')
+    assert (result == b'asdasd')
     sock.sendto(b'ok', remoteaddr)