diff --git a/tests/test_write_role_audit.py b/tests/test_write_role_audit.py index fd66bab..c40e2ff 100644 --- a/tests/test_write_role_audit.py +++ b/tests/test_write_role_audit.py @@ -55,6 +55,7 @@ def pilot_log(overnight_login=False): def run(lines, tmp_path, *extra): receipt = tmp_path / 'r.json' + extra = extra or ('--check-from', '2026-11-04T22:39:00Z', '--check-to', '2026-11-05T08:25:00Z') 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()) @@ -100,7 +101,8 @@ 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='', + again = subprocess.run([sys.executable, str(TOOL), '--receipt', str(tmp_path / 'r.json'), + '--check-from', '2026-11-05T00:00:00Z', '--check-to', '2026-11-05T01:00:00Z'], input='', capture_output=True, text=True) assert again.returncode != 0 and 'FileExistsError' in again.stderr @@ -118,3 +120,9 @@ def test_pilot_window_just_after_the_check_does_not_count(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 + + +def test_policy_request_times_are_reported(tmp_path): + _, r = run(pilot_log(), tmp_path) + assert r['intervals'][1]['policy_request_times'] == ['2026-11-05T08:30:20.000000Z'] + assert r['intervals'][0]['first_seen'] is None diff --git a/tools/service-binds/write-role-audit.py b/tools/service-binds/write-role-audit.py index 361a425..f2f3054 100644 --- a/tools/service-binds/write-role-audit.py +++ b/tools/service-binds/write-role-audit.py @@ -4,7 +4,7 @@ Reads the audit log as JSON lines on stdin. The founder streams it from the pod; nothing is stored on the way: kubectl -n openbao exec openbao-0 -- cat /openbao/audit/openbao-audit.log \ - | python3 tools/service-binds/write-role-audit.py --receipt EVIDENCE/NN-write-role-audit.json + | python3 tools/service-binds/write-role-audit.py --check-from T1 --check-to T2 --receipt EVIDENCE/NN-....json It reports every interval in which the write role existed: from a successful create/update of auth/kubernetes/role/ to its successful delete. For each interval it reports: @@ -116,6 +116,8 @@ def report(stats, intervals, check_from=None, check_to=None): 'opened': iso(item['opened']), 'closed': iso(item['closed']), 'orphan': item.get('orphan', False), 'logins': len(item['logins']), 'login_times': [iso(t) for t in item['logins'][:20]], 'login_service_accounts': sorted(item['service_accounts']), 'policy_requests': len(item['policy_requests']), + 'policy_request_times': [iso(t) for t in item['policy_requests'][:20]], + 'first_seen': iso(item.get('first_seen')), 'failed_k8s_logins': item['failed_k8s_logins']}) out['role_present_at_log_end'] = bool(intervals and intervals[-1]['closed'] is None and not intervals[-1].get('orphan')) @@ -135,10 +137,8 @@ def report(stats, intervals, check_from=None, check_to=None): def main(): p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) p.add_argument('--role', default=ROLE) - p.add_argument('--check-from', type=when, default=when('2026-11-04T22:39:00Z'), - help='start of the period to judge (default: BEA-2026-0002 window start)') - p.add_argument('--check-to', type=when, default=when('2026-11-05T08:25:00Z'), - help='end of the period to judge (default: just after the late close)') + p.add_argument('--check-from', type=when, required=True, help='start of the period to judge (UTC, ...Z)') + p.add_argument('--check-to', type=when, required=True, help='end of the period to judge (UTC, ...Z)') p.add_argument('--receipt', required=True) a = p.parse_args() fd = os.open(a.receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)