Prepare compatible audit receiver with verified container lifecycle

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc

scripts/release_container_check.py

@@ -0,0 +1,127 @@
+#!/usr/bin/env python3
+"""Exercise a candidate against a labelled disposable local PostgreSQL container.
+
+No native credentials or cluster access. Requires psycopg on the host. Keeps
+only a JSON receipt; deletes only the receiver and database schema it creates.
+"""
+import argparse
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime, timezone
+import json
+from pathlib import Path
+import subprocess
+import time
+import uuid
+
+import psycopg
+
+
+def run(argv, **kwargs):
+    return subprocess.run(argv, check=True, capture_output=True, text=True, **kwargs).stdout
+
+
+def main():
+    p = argparse.ArgumentParser(description=__doc__)
+    p.add_argument('--image', required=True)
+    p.add_argument('--postgres-container', required=True)
+    p.add_argument('--postgres-port', type=int, required=True)
+    p.add_argument('--receipt', type=Path, required=True)
+    a = p.parse_args()
+    pg = json.loads(run(['docker', 'inspect', a.postgres_container]))[0]
+    assert pg['Config']['Labels'].get('hfact.task') == 'audit-release-20260911', 'local test fixture required'
+    suffix = uuid.uuid4().hex[:12]
+    schema = 'release_' + suffix
+    name = 'audit-release-check-' + suffix
+    dsn = f'postgresql://postgres:local-test-only@127.0.0.1:{a.postgres_port}/audit_core'
+    registry = [{'name': n, 'tokens': [n+'-fixture'], 'sources': [n],
+                 'tenants': ['tenant:platform'], 'may_write': True, 'may_read': False,
+                 'evidence_kind': 'load-bearing', 'secret_policy': 'redact'}
+                for n in ['approval-engine', 'informed-decision']]
+    registry.append({'name':'independent-reader','tokens':['reader-fixture'],'sources':['*'],
+                     'tenants':['*'],'may_write':False,'may_read':True})
+    proof = {'image': a.image, 'native_credentials_used': False, 'checks': {},
+             'started_at': datetime.now(timezone.utc).isoformat()}
+    created = False
+    def request(method, path, token='', body=None):
+        # Values here are local synthetic fixtures, never native credentials.
+        payload=json.dumps({'method':method,'path':path,'token':token,'body':body})
+        code='''import json,sys,urllib.request,urllib.error
+x=json.load(sys.stdin); b=x['body']
+r=urllib.request.Request('http://127.0.0.1:8080'+x['path'],method=x['method'],data=json.dumps(b).encode() if b is not None else None,headers={'Content-Type':'application/json','Authorization':'Bearer '+x['token'],'Idempotency-Key':b['id'] if b else ''})
+try:
+ with urllib.request.urlopen(r,timeout=15) as f: print(json.dumps([f.status,json.load(f)]))
+except urllib.error.HTTPError as e: print(json.dumps([e.code,json.load(e)]))
+'''
+        return json.loads(run(['docker','exec','-i',name,'python','-c',code],input=payload))
+    def event(sender, suffix):
+        return {'id':suffix,'source':sender,'type':'factory.custody-check',
+                'subject':'synthetic-release-check','tenant':'tenant:platform','correlation_id':'release-'+suffix,
+                'occurred_at':datetime.now(timezone.utc).isoformat(),
+                'data':{'synthetic':True,'auth_token':'fixture-must-be-redacted'}}
+    try:
+        run(['docker','run','-d','--name',name,'--network','container:'+a.postgres_container,
+             '--read-only','--tmpfs','/tmp','--cap-drop','ALL','--security-opt','no-new-privileges',
+             '-e','AUDIT_CORE_DATABASE_URL=postgresql://postgres:local-test-only@127.0.0.1:5432/audit_core',
+             '-e','AUDIT_CORE_DATABASE_SCHEMA='+schema,'-e','AUDIT_CORE_REQUIRE_CUSTODY_CLASS=operational',
+             '-e','AUDIT_CORE_SENDERS='+json.dumps(registry),a.image])
+        created = True
+        for _ in range(50):
+            try:
+                status,body=request('GET','/readyz')
+                if status==200: break
+            except (subprocess.CalledProcessError, ValueError): time.sleep(.2)
+        else: raise AssertionError('receiver readiness timed out')
+        proof['checks']['operational_ready']=body['custody_class']=='operational'
+        assert proof['checks']['operational_ready']
+        uid=run(['docker','exec',name,'id','-u']).strip(); assert uid=='10001'
+        proof['checks']['nonroot_readonly']=True
+        for sender in ['approval-engine','informed-decision']:
+            e=event(sender,sender+'-'+suffix); token=sender+'-fixture'
+            assert request('POST','/v1/events',token,e)[0]==202
+            assert request('POST','/v1/events',token,e)[0]==200
+            assert request('POST','/v1/events',token,dict(e,source='sibling'))[0]==400
+            assert request('POST','/v1/events',token,dict(e,tenant='tenant:sibling'))[0]==400
+            for path in ['/v1/events/'+e['id'],'/v1/events?correlation_id=fixture','/v1/stats','/v1/integrity','/v1/dead-letters','/v1/secret-findings','/v1/stream-findings']:
+                assert request('GET',path,token)[0]==403, path
+            status,stored=request('GET','/v1/events/'+e['id'],'reader-fixture')
+            assert status==200 and 'fixture-must-be-redacted' not in json.dumps(stored)
+            with psycopg.connect(dsn) as c:
+                assert c.execute(f'SELECT count(*) FROM {schema}.events WHERE event_id=%s',(e['id'],)).fetchone()[0]==1
+            proof['checks'][sender]={'accepted':202,'duplicate':200,'wrong_source':400,'wrong_tenant':400,'read_routes_denied':7,'independent_read':200,'redacted':True,'stored_once':True}
+        e=event('approval-engine','shutdown-'+suffix)
+        with psycopg.connect(dsn,autocommit=True) as lock, ThreadPoolExecutor() as pool:
+            lock.execute('SELECT pg_advisory_lock(%s)',(0xA0D17007,))
+            pending=pool.submit(request,'POST','/v1/events','approval-engine-fixture',e)
+            for _ in range(100):
+                if lock.execute("SELECT count(*) FROM pg_locks WHERE locktype='advisory' AND NOT granted").fetchone()[0]: break
+                time.sleep(.05)
+            else: raise AssertionError('in-flight database write not observed')
+            stop=pool.submit(run,['docker','stop','--timeout','10',name])
+            time.sleep(.5)
+            lock.execute('SELECT pg_advisory_unlock(%s)',(0xA0D17007,))
+            try: pending.result()
+            except subprocess.CalledProcessError: pass  # Lost acknowledgement requires retry.
+            stop.result()
+        code=int(run(['docker','inspect',name,'--format','{{.State.ExitCode}}']).strip()); assert code==0, code
+        with psycopg.connect(dsn) as c:
+            assert c.execute(f'SELECT count(*) FROM {schema}.events WHERE event_id=%s',(e['id'],)).fetchone()[0]==1
+        run(['docker','start',name])
+        for _ in range(50):
+            try:
+                if request('GET','/readyz')[0]==200: break
+            except subprocess.CalledProcessError: pass
+            time.sleep(.2)
+        assert request('POST','/v1/events','approval-engine-fixture',e)[0]==200
+        status,integrity=request('GET','/v1/integrity','reader-fixture')
+        assert status==200 and integrity['intact'] and integrity['events']==3
+        proof['checks']['sigterm']={'exit_code':code,'inflight_committed_once':True,'restart_retry_duplicate':True,'chain_intact':True}
+        proof['status']='passed'
+    finally:
+        if created: run(['docker','rm','-f',name])
+        with psycopg.connect(dsn) as c: c.execute(f'DROP SCHEMA IF EXISTS {schema} CASCADE')
+        proof['own_fixture_cleanup']=True
+        with a.receipt.open('x') as f: json.dump(proof,f,indent=2); f.write('\n')
+    print(json.dumps(proof,indent=2))
+
+
+if __name__=='__main__': main()