NK-WP-0058-T11: value-free OpenBao audit read for the executor write role, with synthetic-log tests

write-role-audit.py reads the file audit log on stdin and reports each interval in which
the write role existed, with logins, write-policy requests and failed kubernetes logins.
The pilot's own login is the positive control. Receipt holds times, counts and SA names only.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 943911@bnt-lap001
Assistant-Session: fc322940-1c8e-4bed-ab39-86cec99f23a0

Makefile

@@ -213,7 +213,7 @@ execution-attribution-receipt-test: ## Receipt schema v0.1 validator; fixtures m
 	python3 -m pytest tools/execution-attribution-receipt/tests
 
 bounded-executor-test: ## Bounded executor RBAC and declaration tests; the pilot admission inside its declared grants (NK-WP-0056-T03)
-	python3 -m pytest sso-mfa/k8s/bounded-executor bounded-executor/tests tests/test_executor_lanes.py
+	python3 -m pytest sso-mfa/k8s/bounded-executor bounded-executor/tests tests/test_executor_lanes.py tests/test_write_role_audit.py
 	python3 tools/service-identity/validate.py tools/service-identity/declarations/bounded-executor.yaml \
 	  --admissions examples/bounded-execution-admission/bind-rotation-svc-authelia.yaml
 

WORK-RECORDS.md

@@ -371,15 +371,15 @@
 | task | NK-WP-0057-T03 | todo | — | workplans/NK-WP-0057-decision-routing-by-responsibility.md |
 | task | NK-WP-0057-T04 | todo | — | workplans/NK-WP-0057-decision-routing-by-responsibility.md |
 | task | NK-WP-0058-T01 | todo | — | workplans/NK-WP-0058-bounded-execution-operability.md |
-| task | NK-WP-0058-T02 | todo | — | workplans/NK-WP-0058-bounded-execution-operability.md |
+| task | NK-WP-0058-T02 | progress | — | workplans/NK-WP-0058-bounded-execution-operability.md |
 | task | NK-WP-0058-T03 | todo | — | workplans/NK-WP-0058-bounded-execution-operability.md |
 | task | NK-WP-0058-T04 | todo | — | workplans/NK-WP-0058-bounded-execution-operability.md |
 | task | NK-WP-0058-T05 | todo | — | workplans/NK-WP-0058-bounded-execution-operability.md |
-| task | NK-WP-0058-T06 | todo | — | workplans/NK-WP-0058-bounded-execution-operability.md |
+| task | NK-WP-0058-T06 | progress | — | workplans/NK-WP-0058-bounded-execution-operability.md |
-| task | NK-WP-0058-T07 | todo | — | workplans/NK-WP-0058-bounded-execution-operability.md |
+| task | NK-WP-0058-T07 | progress | — | workplans/NK-WP-0058-bounded-execution-operability.md |
 | task | NK-WP-0058-T08 | todo | — | workplans/NK-WP-0058-bounded-execution-operability.md |
 | task | NK-WP-0058-T09 | todo | — | workplans/NK-WP-0058-bounded-execution-operability.md |
-| task | NK-WP-0058-T10 | todo | — | workplans/NK-WP-0058-bounded-execution-operability.md |
+| task | NK-WP-0058-T10 | progress | — | workplans/NK-WP-0058-bounded-execution-operability.md |
 | task | NK-WP-0058-T11 | todo | — | workplans/NK-WP-0058-bounded-execution-operability.md |
 | intake | NK-IN-0001 | closed | blue | docs/intakes/activity-core-ops-sso-operators.md |
 | intake | NK-IN-0002 | closed | blue | docs/intakes/activity-core-ops-sso-operators.md |

tests/test_write_role_audit.py

@@ -0,0 +1,120 @@
+"""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

tools/service-binds/write-role-audit.py

@@ -0,0 +1,165 @@
+#!/usr/bin/env python3
+"""Value-free read of OpenBao's file audit log for the bounded executor's write role (NK-WP-0058-T11).
+
+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
+
+It reports every interval in which the write role existed: from a successful create/update of
+auth/kubernetes/role/<role> to its successful delete. For each interval it reports:
+
+  logins             successful auth/kubernetes/login whose response names this role (auth metadata, plaintext)
+  policy_requests    other successful requests made with a token carrying the write policy
+  failed_k8s_logins  failed kubernetes logins in the interval; the audit HMACs the requested role, so these are
+                     context, not attribution
+
+The receipt holds times, counts, operations and ServiceAccount names. It holds no request or response data,
+tokens, accessors or HMACs. Exit code 0 means the log parsed. The result line says whether a login used the role
+inside --check-from/--check-to, and whether the role still exists at the end of the log.
+"""
+import argparse
+import hashlib
+import json
+import os
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+ROLE = 'net-kingdom-bind-executor-svc-authelia-write'  # executor-lanes.py LANES['write']['role'] (= policy)
+LOGIN = 'auth/kubernetes/login'
+WRITES = {'create', 'update'}
+
+
+def when(text):
+    """Parse OpenBao's UTC time ('...Z', up to nanoseconds) into an aware datetime."""
+    if not text.endswith('Z'): raise ValueError('time_not_utc')
+    head, _, frac = text[:-1].partition('.')
+    stamp = datetime.fromisoformat(head + ('.' + frac[:6].ljust(6, '0') if frac else ''))
+    return stamp.replace(tzinfo=timezone.utc)
+
+
+def iso(stamp):
+    return stamp.strftime('%Y-%m-%dT%H:%M:%S.%fZ') if stamp else None
+
+
+def auth_of(entry):
+    response = entry.get('response') or {}
+    return response.get('auth') or entry.get('auth') or {}
+
+
+def analyse(lines, role=ROLE, policy=None):
+    policy = policy or role
+    stats = {'lines': 0, 'unparseable': 0, 'first': None, 'last': None}
+    intervals, current = [], None
+    for raw in lines:
+        raw = raw.strip()
+        if not raw: continue
+        stats['lines'] += 1
+        try:
+            entry = json.loads(raw)
+            stamp = when(entry['time'])
+        except (ValueError, KeyError, TypeError):
+            stats['unparseable'] += 1
+            continue
+        stats['first'] = min(stats['first'] or stamp, stamp)
+        stats['last'] = max(stats['last'] or stamp, stamp)
+        if entry.get('type') != 'response': continue  # each request is judged once, by its response
+        request = entry.get('request') or {}
+        path, operation = request.get('path', ''), request.get('operation', '')
+        if entry.get('error'):
+            if path == LOGIN and current is not None: current['failed_k8s_logins'] += 1
+            continue
+        if path == 'auth/kubernetes/role/' + role and (operation in WRITES or operation == 'delete'):
+            if operation in WRITES and current is None:
+                current = {'opened': stamp, 'closed': None, 'logins': [], 'policy_requests': [],
+                           'failed_k8s_logins': 0, 'service_accounts': set()}
+                intervals.append(current)
+            elif operation == 'delete' and current is not None:
+                current['closed'], current = stamp, None
+            continue
+        auth = auth_of(entry)
+        meta = auth.get('metadata') or {}
+        if path == LOGIN:
+            if meta.get('role') == role:
+                target = current if current is not None else _orphan(intervals, stamp)
+                target['logins'].append(stamp)
+                target['service_accounts'].add('%s/%s' % (meta.get('service_account_namespace', '?'),
+                                                          meta.get('service_account_name', '?')))
+            continue
+        policies = set(auth.get('policies') or []) | set(auth.get('token_policies') or [])
+        if policy in policies:
+            target = current if current is not None else _orphan(intervals, stamp)
+            target['policy_requests'].append(stamp)
+    return stats, intervals
+
+
+def _orphan(intervals, stamp):
+    """Activity under the role with no open interval: the log missed the create. Keep it visible."""
+    orphan = {'opened': None, 'closed': None, 'logins': [], 'policy_requests': [], 'failed_k8s_logins': 0,
+              'service_accounts': set(), 'orphan': True, 'first_seen': stamp}
+    intervals.append(orphan)
+    return orphan
+
+
+def overlaps(interval, start, end, log_last):
+    opened = interval.get('opened') or interval.get('first_seen')
+    closed = interval.get('closed') or log_last
+    return opened <= end and closed >= start
+
+
+def report(stats, intervals, check_from=None, check_to=None):
+    out = {'log_lines': stats['lines'], 'log_unparseable': stats['unparseable'],
+           'log_first_time': iso(stats['first']), 'log_last_time': iso(stats['last']), 'intervals': []}
+    for item in intervals:
+        out['intervals'].append({
+            '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']),
+            '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'))
+    if check_from and check_to:
+        checked = [i for i in intervals if stats['last'] and overlaps(i, check_from, check_to, stats['last'])]
+        logins = [t for i in checked for t in i['logins'] if check_from <= t <= check_to]
+        requests = sum(check_from <= t <= check_to for i in checked for t in i['policy_requests'])
+        out['check'] = {'from': iso(check_from), 'to': iso(check_to),
+                        'log_covers_check': bool(stats['first'] and stats['first'] <= check_from
+                                                 and stats['last'] >= check_to),
+                        'intervals_overlapping': len(checked), 'logins_in_check': len(logins),
+                        'policy_requests_in_check': requests,
+                        'no_login_in_check': not logins and requests == 0}
+    return out
+
+
+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('--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)
+    r = {'action': 'write-role-audit', 'role': a.role, 'status': 'failed', 'values_recorded': False,
+         'source_sha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest()}
+    try:
+        stats, intervals = analyse(sys.stdin, role=a.role)
+        if not stats['lines']: raise ValueError('empty_audit_input')
+        r.update(report(stats, intervals, a.check_from, a.check_to), status='read')
+    except Exception as error:  # value-free: name the failure, never echo input
+        r['reason'] = str(error) if isinstance(error, ValueError) and str(error).isidentifier() \
+            else type(error).__name__
+    finally:
+        with os.fdopen(fd, 'w') as out: json.dump(r, out, indent=2, sort_keys=True); out.write('\n')
+    check = r.get('check', {})
+    print(json.dumps({'status': r['status'], 'reason': r.get('reason'), 'covers': check.get('log_covers_check'),
+                      'no_login_in_check': check.get('no_login_in_check'),
+                      'intervals': len(r.get('intervals', [])),
+                      'role_present_at_log_end': r.get('role_present_at_log_end')}))
+    return int(r['status'] != 'read')
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())

workplans/NK-WP-0058-bounded-execution-operability.md

@@ -149,7 +149,7 @@ Mark the duplicated pieces, for example the two seeders and the two Secret names
 
 ```task
 id: NK-WP-0058-T10
-status: todo
+status: progress
 priority: high
 assignee: claude-code
 depends_on: [NK-WP-0058-T01]
@@ -169,7 +169,7 @@ founder for on 2026-11-04/05 (the input for T05).
 
 ```task
 id: NK-WP-0058-T02
-status: todo
+status: progress
 priority: high
 assignee: claude-code
 depends_on: [NK-WP-0058-T01, NK-WP-0058-T09, NK-WP-0058-T10]
@@ -287,7 +287,7 @@ to ask the founder for on 2026-11-04/05 is then readable, or explicitly founder-
 
 ```task
 id: NK-WP-0058-T06
-status: todo
+status: progress
 priority: medium
 assignee: claude-code
 depends_on: [NK-WP-0058-T09]
@@ -309,7 +309,7 @@ The open check from 2026-11-05 on the overnight window is T11.
 
 ```task
 id: NK-WP-0058-T07
-status: todo
+status: progress
 priority: medium
 assignee: claude-code
 state_hub_task_id: "a99424ed-ee8b-5db4-9fa7-2ae5fc32c39c"