"""write-role-audit.py on a synthetic OpenBao audit log (NK-WP-0058-T11); the fixture carries fake HMACs only.""" import importlib.util import json import subprocess import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[1] TOOL = ROOT / 'tools/service-binds/write-role-audit.py' spec = importlib.util.spec_from_file_location('write_role_audit', TOOL) wra = importlib.util.module_from_spec(spec); spec.loader.exec_module(wra) ROLE = wra.ROLE SECRETISH = 'hmac-sha256:fake0123456789' def entry(time, path, operation='update', kind='response', error='', auth=None, response_auth=None): item = {'time': time, 'type': kind, 'error': error, 'request': {'path': path, 'operation': operation, 'data': {'jwt': SECRETISH, 'role': SECRETISH}}, 'auth': auth or {}} if response_auth is not None: item['response'] = {'auth': response_auth} return json.dumps(item) def role_write(time, operation='update', error=''): return [entry(time, 'auth/kubernetes/role/' + ROLE, operation, kind='request'), entry(time, 'auth/kubernetes/role/' + ROLE, operation, error=error)] def login(time, role=ROLE, error=''): meta = {'role': role, 'service_account_name': 'nk-bind-executor', 'service_account_namespace': 'sso'} return [entry(time, 'auth/kubernetes/login', kind='request'), entry(time, 'auth/kubernetes/login', error=error, response_auth=None if error else {'client_token': SECRETISH, 'policies': [ROLE], 'metadata': meta})] def used(time): return [entry(time, 'platform/data/workloads/net-kingdom/directory-binds/svc-authelia', 'read', auth={'accessor': SECRETISH, 'policies': ['default', ROLE]})] def pilot_log(overnight_login=False): lines = [entry('2026-11-04T20:00:00.000000001Z', 'sys/health', 'read')] lines += role_write('2026-11-04T23:20:01.123456789Z') # diagnostic window opened lines += login('2026-11-04T23:40:00Z', role='eso-platform') # other roles keep logging in lines += login('2026-11-05T01:00:00Z', error='permission denied') # failed, unattributed if overnight_login: lines += login('2026-11-05T02:00:00Z') + used('2026-11-05T02:00:01Z') lines += role_write('2026-11-05T08:24:10Z', 'delete') # late close lines += role_write('2026-11-05T08:25:30Z', 'update', error='denied') # a failed open is no interval lines += role_write('2026-11-05T08:29:00Z') # BEA-2026-0003 window lines += login('2026-11-05T08:30:12Z') + used('2026-11-05T08:30:20Z') # positive control: the pilot run lines += role_write('2026-11-05T08:31:00Z', 'delete') lines += ['not json', entry('2026-11-05T09:00:00Z', 'sys/health', 'read')] return lines def run(lines, tmp_path, *extra): receipt = tmp_path / 'r.json' done = subprocess.run([sys.executable, str(TOOL), '--receipt', str(receipt), *extra], input='\n'.join(lines) + '\n', capture_output=True, text=True) return done, json.loads(receipt.read_text()) def test_clean_overnight_window_and_positive_control(tmp_path): done, r = run(pilot_log(), tmp_path) assert done.returncode == 0, done.stderr assert r['status'] == 'read' and r['log_unparseable'] == 1 night, pilot = r['intervals'] assert night['opened'].startswith('2026-11-04T23:20:01.123456') assert night['closed'].startswith('2026-11-05T08:24:10') assert night['logins'] == 0 and night['policy_requests'] == 0 and night['failed_k8s_logins'] == 1 assert pilot['logins'] == 1 and pilot['policy_requests'] == 1 assert pilot['login_service_accounts'] == ['sso/nk-bind-executor'] assert r['check']['log_covers_check'] and r['check']['no_login_in_check'] assert r['role_present_at_log_end'] is False assert json.loads(done.stdout) == {'status': 'read', 'reason': None, 'covers': True, 'no_login_in_check': True, 'intervals': 2, 'role_present_at_log_end': False} def test_overnight_login_is_found(tmp_path): _, r = run(pilot_log(overnight_login=True), tmp_path) assert r['intervals'][0]['logins'] == 1 and r['intervals'][0]['policy_requests'] == 1 assert r['check']['logins_in_check'] == 1 and r['check']['no_login_in_check'] is False def test_role_left_open_and_log_gap(tmp_path): lines = pilot_log()[:-2] + role_write('2026-11-05T10:00:00Z') _, r = run(lines[3:], tmp_path) # the log starts after the check period begins assert r['role_present_at_log_end'] is True assert r['check']['log_covers_check'] is False def test_activity_without_a_seen_create_is_kept(tmp_path): _, r = run(login('2026-11-05T03:00:00Z') + [entry('2026-11-05T09:00:00Z', 'sys/health', 'read')], tmp_path, '--check-from', '2026-11-05T00:00:00Z', '--check-to', '2026-11-05T08:00:00Z') assert r['intervals'][0]['orphan'] and r['intervals'][0]['logins'] == 1 assert r['check']['no_login_in_check'] is False def test_receipt_is_value_free_and_refuses_overwrite(tmp_path): done, r = run(pilot_log(overnight_login=True), tmp_path) text = (tmp_path / 'r.json').read_text() + done.stdout assert 'hmac' not in text and 'client_token' not in text and 'accessor' not in text again = subprocess.run([sys.executable, str(TOOL), '--receipt', str(tmp_path / 'r.json')], input='', capture_output=True, text=True) assert again.returncode != 0 and 'FileExistsError' in again.stderr def test_empty_input_fails_by_name(tmp_path): done, r = run([], tmp_path) assert done.returncode == 1 and r['reason'] == 'empty_audit_input' def test_pilot_window_just_after_the_check_does_not_count(tmp_path): lines = (role_write('2026-11-04T23:20:00Z') + role_write('2026-11-05T08:24:10Z', 'delete') + role_write('2026-11-05T08:24:40Z') + login('2026-11-05T08:30:12Z') + used('2026-11-05T08:30:20Z') + role_write('2026-11-05T08:31:00Z', 'delete')) _, r = run([entry('2026-11-04T20:00:00Z', 'sys/health', 'read')] + lines, tmp_path) assert r['check']['intervals_overlapping'] == 2 # the pilot window opened before --check-to assert r['check']['logins_in_check'] == 0 and r['check']['policy_requests_in_check'] == 0 assert r['check']['no_login_in_check'] is True