diff --git a/keepercommander/commands/aram.py b/keepercommander/commands/aram.py index 3a0a3d99f..ce9cf1fb4 100644 --- a/keepercommander/commands/aram.py +++ b/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') diff --git a/unit-tests/test_command_enterprise.py b/unit-tests/test_command_enterprise.py index ef4a18cfd..6743ad428 100644 --- a/unit-tests/test_command_enterprise.py +++ b/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()