Release 18.0.7

keepercommander/__init__.py

@@ -10,4 +10,4 @@
 # Contact: commander@keepersecurity.com
 #
 
-__version__ = '18.0.6'
+__version__ = '18.0.7'

keepercommander/commands/pam_import/kcm_import.py

@@ -1367,6 +1367,7 @@ def execute(self, params, **kwargs):
             fd = os.open(output_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
             with os.fdopen(fd, 'w') as f:
                 json.dump(out_data, f, indent=2)
+            utils.set_file_permissions(output_file)
             redact_note = '' if include_creds else ' (credentials redacted)'
             logging.warning('JSON written to %s (%d resources, %d users)%s',
                             output_file, num_resources, num_users, redact_note)

unit-tests/pam/test_dag_layer_b_set_record_rotation.py

@@ -83,18 +83,30 @@ def test_set_record_rotation_hits_correct_url():
 
 def test_set_record_rotation_sends_protobuf_body():
     from keepercommander.commands.pam.router_helper import router_set_record_rotation_information
+    from keepercommander import crypto, utils
     rq = router_pb2.RouterRecordRotationRequest(
         recordUid=RECORD_UID,
         configurationUid=CONFIG_UID,
         resourceUid=RESOURCE_UID,
         schedule='0 0 * * *',
     )
+    transmission_key = utils.generate_aes_key()
     with patch(REQUESTS_TARGET, return_value=_ok_router_response()) as mock_req:
-        router_set_record_rotation_information(_mock_params(), rq)
+        router_set_record_rotation_information(_mock_params(), rq, transmission_key=transmission_key)
     body = mock_req.call_args.kwargs.get('data')
     assert isinstance(body, (bytes, bytearray))
-    # Encrypted blob, not JSON
-    assert not body.startswith(b'{')
+    # Body is the AES-GCM-encrypted protobuf — not the plaintext proto, and not
+    # JSON. Decrypt with the known transmission key and confirm it round-trips.
+    # (A first-byte heuristic like `not body.startswith(b'{')` is flaky: ~1/256
+    # of ciphertexts legitimately start with 0x7b.)
+    assert body != rq.SerializeToString()
+    decrypted = crypto.decrypt_aes_v2(body, transmission_key)
+    parsed = router_pb2.RouterRecordRotationRequest()
+    parsed.ParseFromString(decrypted)
+    assert parsed.recordUid == RECORD_UID
+    assert parsed.configurationUid == CONFIG_UID
+    assert parsed.resourceUid == RESOURCE_UID
+    assert parsed.schedule == '0 0 * * *'
 
 
 # --------------------------------------------------------------------------- #

unit-tests/pam/test_kcm_import.py

@@ -11,6 +11,8 @@
 
 import json
 import os
+import platform
+import subprocess
 import sys
 import tempfile
 import unittest
@@ -516,9 +518,8 @@ def test_output_file_pipeline(self, mock_getpass, MockConnector):
         cmd = PAMProjectKCMImportCommand()
         params = MagicMock()
 
-        import tempfile
-        with tempfile.NamedTemporaryFile(suffix='.json', delete=False) as f:
-            output_path = f.name
+        tmp_dir = tempfile.mkdtemp()
+        output_path = os.path.join(tmp_dir, 'output.json')
 
         try:
             # With --include-credentials, passwords are preserved
@@ -547,6 +548,7 @@ def test_output_file_pipeline(self, mock_getpass, MockConnector):
         finally:
             if os.path.exists(output_path):
                 os.unlink(output_path)
+            os.rmdir(tmp_dir)
 
     @patch('keepercommander.commands.pam_import.kcm_import.KCMDatabaseConnector')
     @patch('keepercommander.commands.pam_import.kcm_import.getpass.getpass',
@@ -562,9 +564,8 @@ def test_output_file_redacted_by_default(self, mock_getpass, MockConnector):
         cmd = PAMProjectKCMImportCommand()
         params = MagicMock()
 
-        import tempfile
-        with tempfile.NamedTemporaryFile(suffix='.json', delete=False) as f:
-            output_path = f.name
+        tmp_dir = tempfile.mkdtemp()
+        output_path = os.path.join(tmp_dir, 'output.json')
 
         try:
             cmd.execute(params,
@@ -588,6 +589,7 @@ def test_output_file_redacted_by_default(self, mock_getpass, MockConnector):
         finally:
             if os.path.exists(output_path):
                 os.unlink(output_path)
+            os.rmdir(tmp_dir)
 
     @patch('keepercommander.commands.pam_import.kcm_import.KCMDatabaseConnector')
     @patch('keepercommander.commands.pam_import.kcm_import.getpass.getpass',
@@ -1290,17 +1292,28 @@ def test_output_file_owner_only(self, mock_getpass, MockConnector):
         cmd = PAMProjectKCMImportCommand()
         params = MagicMock()
 
-        import stat
         tmp_dir = tempfile.mkdtemp()
         output_path = os.path.join(tmp_dir, 'test_output.json')
         try:
             cmd.execute(params,
                         db_host='127.0.0.1',
                         output=output_path)
 
-            file_mode = os.stat(output_path).st_mode & 0o777
-            self.assertEqual(file_mode, 0o600,
-                             f'Expected 0o600, got {oct(file_mode)}')
+            # POSIX: verifies mode == 0o600 via os.stat.
+            # Windows: os.stat().st_mode reads DOS attributes, not the NTFS
+            # DACL, so verify the icacls-applied ACL has the principals
+            # stripped by utils.set_file_permissions (SYSTEM, Administrators).
+            if platform.system() == 'Windows':
+                acl = subprocess.run(['icacls', output_path],
+                                     capture_output=True, text=True, check=True).stdout
+                self.assertNotIn('NT AUTHORITY\\SYSTEM', acl,
+                                 f'SYSTEM should have been removed from ACL:\n{acl}')
+                self.assertNotIn('BUILTIN\\Administrators', acl,
+                                 f'Administrators should have been removed from ACL:\n{acl}')
+            else:
+                file_mode = os.stat(output_path).st_mode & 0o777
+                self.assertEqual(file_mode, 0o600,
+                                 f'Expected 0o600, got {oct(file_mode)}')
         finally:
             if os.path.exists(output_path):
                 os.unlink(output_path)