KC-1462: Validate record ownership in audit-log command

keepercommander/commands/aram.py

@@ -339,6 +339,19 @@ def load_syslog_templates(params):
                 syslog_templates[name] = syslog
 
 
+def _is_owned_audit_log_record(params: KeeperParams, record_uid: str) -> bool:
+    """True if the current account owns this record.
+
+    Uses ``record_owner_cache`` first, then ``meta_data_cache`` from
+    recordMetaData (server-asserted). Unknown ownership fails closed.
+    """
+    owner = (params.record_owner_cache or {}).get(record_uid)
+    if owner and owner.owner:
+        return True
+    meta = (getattr(params, 'meta_data_cache', None) or {}).get(record_uid) or {}
+    return bool(meta.get('owner'))
+
+
 class AuditLogBaseExport(abc.ABC):
     def __init__(self):
         self.store_record = False
@@ -494,7 +507,7 @@ def get_properties(self, record, props):
                     for test_url in ['https://{0}/services/collector'.format(address), 'http://{0}/services/collector'.format(address)]:
                         try:
                             print('Testing \'{0}\' ...'.format(test_url), file=sys.stderr, end='', flush=True)
-                            rs = requests.post(test_url, json='', verify=False)
+                            rs = requests.post(test_url, json='')
                             if rs.status_code == 401:
                                 js = rs.json()
                                 if 'code' in js:
@@ -519,7 +532,7 @@ def get_properties(self, record, props):
                         return
                     try:
                         auth={'Authorization': 'Splunk {0}'.format(test_token)}
-                        rs = requests.post(url, json='', headers=auth, verify=False)
+                        rs = requests.post(url, json='', headers=auth)
                         if rs.status_code == 400:
                             js = rs.json()
                             if 'code' in js:
@@ -553,7 +566,7 @@ def export_events(self, props, events):
         auth = { 'Authorization': 'Splunk {0}'.format(props['token']) }
         try:
             logging.captureWarnings(True)
-            rs = requests.post(props['hec_url'], data='\n'.join(events), headers=auth, verify=False)
+            rs = requests.post(props['hec_url'], data='\n'.join(events), headers=auth)
         finally:
             logging.captureWarnings(False)
 
@@ -942,7 +955,11 @@ def execute(self, params, **kwargs):
         for r_uid in params.record_cache:
             rec = vault.KeeperRecord.load(params, r_uid)
             if record_name in [rec.record_uid, rec.title]:
+                if not _is_owned_audit_log_record(params, r_uid):
+                    print(f'Note: Record "{rec.title}" is not owned by you and will not be used.')
+                    continue
                 record = rec
+                break
         if record is None:
             answer = user_choice('Do you want to create a Keeper record to store audit log settings?', 'yn', 'n')
             if answer.lower() == 'y':
@@ -957,6 +974,8 @@ def execute(self, params, **kwargs):
         if record is None:
             raise CommandError('audit-log', 'Record not found')
 
+        print(f'Export Destination: "{record.title}" ({record.record_uid})')
+
         shared_folder_uids = kwargs.get('shared_folder_uid')
         node_ids = kwargs.get('node_id')
         days = kwargs.get('days')

unit-tests/test_command_enterprise.py

@@ -527,6 +527,97 @@ def test_audit_log_splunk_convert_event(self):
         }
         splunk.convert_event(props, self.get_audit_event())
 
+    def test_audit_log_ownership_skips_non_owned_record(self):
+        """When both shared and owned records exist, only owned is selected."""
+        from keepercommander.params import RecordOwner
+
+        shared = mock.MagicMock(spec=vault.PasswordRecord)
+        shared.record_uid = 'SHARED_UID'
+        shared.title = 'Audit Log: Sumologic'
+
+        owned = mock.MagicMock(spec=vault.PasswordRecord)
+        owned.record_uid = 'OWNED_UID'
+        owned.title = 'Audit Log: Sumologic'
+
+        params = mock.MagicMock(spec=KeeperParams)
+        params.record_cache = {'SHARED_UID': {}, 'OWNED_UID': {}}
+        params.record_owner_cache = {
+            'SHARED_UID': RecordOwner(False, 'attacker'),
+            'OWNED_UID': RecordOwner(True, 'operator'),
+        }
+        params.meta_data_cache = {}
+
+        with mock.patch('keepercommander.commands.aram.vault.KeeperRecord.load') as mock_load:
+            def load_side_effect(_params, uid):
+                return {'SHARED_UID': shared, 'OWNED_UID': owned}[uid]
+            mock_load.side_effect = load_side_effect
+
+            record = None
+            record_name = 'Audit Log: Sumologic'
+            for r_uid in params.record_cache:
+                rec = mock_load(params, r_uid)
+                if record_name in [rec.record_uid, rec.title]:
+                    if not aram._is_owned_audit_log_record(params, r_uid):
+                        continue
+                    record = rec
+                    break
+
+            self.assertEqual(record.record_uid, 'OWNED_UID')
+
+    def test_audit_log_ownership_returns_none_when_only_shared(self):
+        """When only a shared record matches, none is returned."""
+        from keepercommander.params import RecordOwner
+
+        shared = mock.MagicMock(spec=vault.PasswordRecord)
+        shared.record_uid = 'SHARED_UID'
+        shared.title = 'Audit Log: Sumologic'
+
+        params = mock.MagicMock(spec=KeeperParams)
+        params.record_cache = {'SHARED_UID': {}}
+        params.record_owner_cache = {'SHARED_UID': RecordOwner(False, 'attacker')}
+        params.meta_data_cache = {}
+
+        with mock.patch('keepercommander.commands.aram.vault.KeeperRecord.load') as mock_load:
+            mock_load.return_value = shared
+
+            record = None
+            record_name = 'Audit Log: Sumologic'
+            for r_uid in params.record_cache:
+                rec = mock_load(params, r_uid)
+                if record_name in [rec.record_uid, rec.title]:
+                    if not aram._is_owned_audit_log_record(params, r_uid):
+                        continue
+                    record = rec
+                    break
+
+            self.assertIsNone(record)
+
+    def test_audit_log_ownership_uses_meta_data_cache_fallback(self):
+        """Ownership check falls back to meta_data_cache if record_owner_cache missing."""
+        from keepercommander.params import RecordOwner
+
+        owned = mock.MagicMock(spec=vault.PasswordRecord)
+        owned.record_uid = 'OWNED_UID'
+        owned.title = 'Audit Log: Sumologic'
+
+        params = mock.MagicMock(spec=KeeperParams)
+        params.record_cache = {'OWNED_UID': {}}
+        params.record_owner_cache = {}
+        params.meta_data_cache = {'OWNED_UID': {'owner': True}}
+
+        result = aram._is_owned_audit_log_record(params, 'OWNED_UID')
+        self.assertTrue(result)
+
+    def test_audit_log_ownership_fails_closed_on_unknown(self):
+        """Ownership check returns False when ownership is unknown."""
+        params = mock.MagicMock(spec=KeeperParams)
+        params.record_cache = {}
+        params.record_owner_cache = {}
+        params.meta_data_cache = {}
+
+        result = aram._is_owned_audit_log_record(params, 'UNKNOWN_UID')
+        self.assertFalse(result)
+
     def test_audit_audit_report_parse_date_filter(self):
         cmd = aram.AuditReportCommand()