feat: verify native factory sender delivery and outbox replay

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

scripts/native_factory_acceptance.py

@@ -0,0 +1,245 @@
+#!/usr/bin/env python3
+"""Prepare or execute the bounded, attended native factory sender acceptance."""
+import argparse
+import base64
+from datetime import datetime, timezone
+import hashlib
+import io
+import json
+import os
+from pathlib import Path
+import secrets
+import socket
+import subprocess
+import time
+from urllib.error import HTTPError
+from urllib.request import Request, build_opener, ProxyHandler, HTTPRedirectHandler
+import zipfile
+
+from factory_audit_custody import contracts, desired, snapshot, receiver_check, require
+from state_hub_preflight_lane import ROOT, LaneError, command, bao, data
+
+IMAGE = 'forgejo.coulomb.social/coulomb/approval-engine@sha256:251941a5cb2724b57cc32cff6b693b1ab0be695bee4f56f02d51961189c0fa49'
+RECEIVER = 'forgejo.coulomb.social/coulomb/audit-core@sha256:c82e0442de0fd181342916ae9cd5d6de41d859e1efda637bd93936c67873afa5'
+SOURCES = {'approval-engine':('a0a602976eef818f36dde35f76f7f2e589bd051b','approval_engine'),
+           'informed-decision':('bda9381f070859ae997474cf44b86bf4eea9bb52','informed_decision')}
+CONFIRM = 'VERIFY CCR-2026-0021 CCR-2026-0022 PRODUCERS'
+
+
+def source_bundle(base):
+    buffer = io.BytesIO(); hashes = {}
+    with zipfile.ZipFile(buffer,'w',compression=zipfile.ZIP_DEFLATED) as out:
+        for repo,(commit,package) in SOURCES.items():
+            prefix = ['git','-C',str(base/repo)]
+            names = command(prefix+['ls-tree','-r','--name-only',commit,package]).stdout.decode().splitlines()
+            require(names and all(n.startswith(package+'/') and n.endswith('.py') for n in names), 'source_package_shape_changed')
+            for name in names:
+                body = command(prefix+['show',commit+':'+name]).stdout
+                hashes[name] = hashlib.sha256(body).hexdigest()
+                entry = zipfile.ZipInfo(name,date_time=(2026,9,11,0,0,0));entry.compress_type=zipfile.ZIP_DEFLATED
+                out.writestr(entry,body)
+    return buffer.getvalue(), hashes
+
+
+def resources(run_id, sender, bundle, script):
+    suffix = run_id.removeprefix('factory-audit-');name='factory-audit-'+suffix
+    labels={'app.kubernetes.io/name':sender,'railiance.io/factory-audit-probe':run_id}
+    config={'sender':sender,'run_id':run_id,'event_id':run_id+'-'+sender,
+            'origin':'http://audit-core.audit-core.svc.cluster.local:8080',
+            'source_zip_sha256':hashlib.sha256(bundle).hexdigest()}
+    key='audit-token' if sender=='approval-engine' else 'token'
+    meta={'name':name,'namespace':sender,'labels':labels}
+    cm={'apiVersion':'v1','kind':'ConfigMap','metadata':meta,'immutable':True,
+        'data':{'producer.py':script.decode(),'config.json':json.dumps(config,sort_keys=True)},
+        'binaryData':{'source.zip':base64.b64encode(bundle).decode()}}
+    policy={'apiVersion':'networking.k8s.io/v1','kind':'NetworkPolicy','metadata':meta,
+            'spec':{'podSelector':{'matchLabels':{'railiance.io/factory-audit-probe':run_id}},
+                    'policyTypes':['Ingress','Egress'],'ingress':[],
+                    'egress':[{'to':[{'namespaceSelector':{'matchLabels':{'kubernetes.io/metadata.name':'audit-core'}},
+                                      'podSelector':{'matchLabels':{'app.kubernetes.io/name':'audit-core','app.kubernetes.io/component':'receiver'}}}],
+                               'ports':[{'protocol':'TCP','port':8080}]},
+                              {'to':[{'namespaceSelector':{'matchLabels':{'kubernetes.io/metadata.name':'kube-system'}},
+                                      'podSelector':{'matchLabels':{'k8s-app':'kube-dns'}}}],
+                               'ports':[{'protocol':'UDP','port':53},{'protocol':'TCP','port':53}]}]}}
+    job={'apiVersion':'batch/v1','kind':'Job','metadata':meta,
+         'spec':{'backoffLimit':0,'activeDeadlineSeconds':150,'ttlSecondsAfterFinished':3600,
+                 'template':{'metadata':{'labels':labels},'spec':{
+                     'restartPolicy':'Never','automountServiceAccountToken':False,
+                     'securityContext':{'runAsNonRoot':True,'runAsUser':10001,'runAsGroup':10001,
+                                        'fsGroup':10001,'seccompProfile':{'type':'RuntimeDefault'}},
+                     'containers':[{'name':'probe','image':IMAGE,'imagePullPolicy':'IfNotPresent',
+                         'command':['python','-I','/probe/producer.py'],
+                         'securityContext':{'allowPrivilegeEscalation':False,'readOnlyRootFilesystem':True,
+                                            'capabilities':{'drop':['ALL']}},
+                         'resources':{'requests':{'cpu':'25m','memory':'64Mi'},'limits':{'cpu':'500m','memory':'192Mi'}},
+                         'volumeMounts':[{'name':'probe','mountPath':'/probe','readOnly':True},
+                                         {'name':'audit','mountPath':'/credential','readOnly':True},
+                                         {'name':'state','mountPath':'/state'}]}],
+                     'volumes':[{'name':'probe','configMap':{'name':name,'defaultMode':0o444}},
+                                {'name':'audit','secret':{'secretName':sender+'-audit','defaultMode':0o440,
+                                                         'items':[{'key':key,'path':'token'}]}},
+                                {'name':'state','emptyDir':{'sizeLimit':'32Mi'}}]}}}}
+    return [cm,policy,job]
+
+
+def prepare(base, path):
+    bundle,hashes=source_bundle(base)
+    script=(ROOT/'scripts/native_factory_probe/producer.py').read_bytes()
+    run_id='factory-audit-'+datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')+'-'+secrets.token_hex(3)
+    packet={'schema':'platform.factory-native-acceptance.v1','run_id':run_id,'source_commits':SOURCES,
+            'source_hashes':hashes,'source_zip_sha256':hashlib.sha256(bundle).hexdigest(),
+            'probe_sha256':hashlib.sha256(script).hexdigest(),'objects':sum([resources(run_id,s,bundle,script) for s in SOURCES],[])}
+    with path.open('x') as f:json.dump(packet,f,indent=2);f.write('\n')
+    return {'packet':str(path),'sha256':hashlib.sha256(path.read_bytes()).hexdigest(),
+            'run_id':run_id,'objects':len(packet['objects']),'source_files':len(hashes)}
+
+
+def validate_packet(path, sha256):
+    raw=path.read_bytes();require(hashlib.sha256(raw).hexdigest()==sha256,'packet_digest_changed')
+    packet=json.loads(raw);run_id=packet['run_id']
+    import re
+    require(re.fullmatch(r'factory-audit-[0-9]{14}-[0-9a-f]{6}',run_id),'invalid_run_id')
+    require(packet['source_commits']=={k:list(v) for k,v in SOURCES.items()},'source_pins_changed')
+    objects=packet['objects'];require(len(objects)==6,'exact_resources_required')
+    bundle=base64.b64decode(objects[0]['binaryData']['source.zip'],validate=True)
+    script=(ROOT/'scripts/native_factory_probe/producer.py').read_bytes()
+    require(hashlib.sha256(bundle).hexdigest()==packet['source_zip_sha256'],'bundle_digest_changed')
+    require(hashlib.sha256(script).hexdigest()==packet['probe_sha256'],'probe_digest_changed')
+    with zipfile.ZipFile(io.BytesIO(bundle)) as z:
+        require(set(z.namelist())==set(packet['source_hashes']),'source_inventory_changed')
+        require(all(hashlib.sha256(z.read(n)).hexdigest()==h for n,h in packet['source_hashes'].items()),'source_hash_changed')
+    require(objects==sum([resources(run_id,s,bundle,script) for s in SOURCES],[]),'resource_scope_changed')
+    return packet
+
+
+class NoRedirect(HTTPRedirectHandler):
+    def redirect_request(self,*args):return None
+
+
+def read_http(origin, token, path):
+    require(path=='/v1/integrity' or path.startswith('/v1/events/factory-audit-'),'reader_path_not_scoped_to_probe')
+    request=Request(origin+path,headers={'Authorization':'Bearer '+token})
+    try:response=build_opener(ProxyHandler({}),NoRedirect()).open(request,timeout=10)
+    except HTTPError as exc:response=exc
+    with response:
+        raw=response.read(524289);require(len(raw)<=524288,'readback_too_large')
+        require(response.status==200,'operator_readback_failed')
+        return json.loads(raw)
+
+
+def operator_identity(rows):
+    # Select an already existing independent, read-only, full-tenant operator.
+    candidates=[r for r in rows if r.get('may_read') is True and r.get('may_write',True) is False
+                and r.get('tenants')==['*'] and r['name'] not in SOURCES]
+    named=[r for r in candidates if 'operator' in r['name']]
+    if named:candidates=named
+    require(len(candidates)==1,'independent_operator_reader_ambiguous_or_absent')
+    return candidates[0]
+
+
+def run(packet, kube, receipt, save):
+    lanes=contracts(approved=True)
+    receipt['receiver_before']=receiver_check(kube,RECEIVER);save()
+    identity=data(bao(['token','lookup','-format=json']))['data']
+    require('platform-admin' in identity['policies'] and 'root' not in identity['policies'],'attended_platform_admin_required')
+    version,_,rows,_=snapshot()
+    for lane in lanes:
+        registered=next(r for r in rows if r['name']==lane['name'])
+        require(registered==desired(lane,registered['tokens'][0]),'sender_registry_scope_changed')
+        policies=data(command(kube+['-n',lane['name'],'get','networkpolicy','-o','json']))
+        require(not policies['items'],'existing_egress_policy_requires_review')
+        es=data(command(kube+['-n',lane['name'],'get','externalsecret',lane['secret'],'-o','json']))
+        require(any(c['type']=='Ready' and c['status']=='True' for c in es['status']['conditions']),'producer_projection_not_ready')
+    receipt['reader_candidates']=[{k:r.get(k) for k in ['name','may_read','may_write','tenants']} for r in rows if r.get('may_read') is True];save()
+    reader=operator_identity(rows);reader_token=(reader.get('tokens') or [reader['token']])[0]
+    receipt.update(registry_version=version,reader_name=reader['name'],phase='native_jobs');save()
+    with socket.socket() as s:s.bind(('127.0.0.1',0));port=s.getsockname()[1]
+    forward=subprocess.Popen(kube+['-n','audit-core','port-forward','--address=127.0.0.1','service/audit-core',str(port)+':8080'],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)
+    created=[]
+    try:
+        for _ in range(40):
+            if forward.poll() is not None:raise LaneError('port_forward_failed')
+            try:
+                with socket.create_connection(('127.0.0.1',port),timeout=.2):break
+            except OSError:time.sleep(.2)
+        else:raise LaneError('port_forward_not_ready')
+        origin='http://127.0.0.1:'+str(port)
+        receipt['integrity_before']=read_http(origin,reader_token,'/v1/integrity')
+        require(receipt['integrity_before']['intact'] is True,'existing_chain_not_intact');save()
+        for obj in packet['objects']:
+            # Create is exclusive; never replace another operation's object.
+            native=data(command(kube+['create','-f','-','-o','json'],payload=obj))
+            created.append({k:native[k] for k in ['apiVersion','kind','metadata']})
+            receipt['created_objects']=[{'kind':x['kind'],'name':x['metadata']['name'],'namespace':x['metadata']['namespace'],'uid':x['metadata']['uid']} for x in created];save()
+        receipt['producers']=[]
+        for sender in SOURCES:
+            name=packet['run_id'];pods=[]
+            for _ in range(80):
+                job=data(command(kube+['-n',sender,'get','job',name,'-o','json']))
+                if job.get('status',{}).get('succeeded')==1:break
+                if job.get('status',{}).get('failed'):
+                    raise LaneError('native_probe_job_failed_'+sender)
+                time.sleep(2)
+            else:raise LaneError('native_probe_job_timeout_'+sender)
+            pods=data(command(kube+['-n',sender,'get','pods','-l','job-name='+name,'-o','json']))['items']
+            require(len(pods)==1 and pods[0]['status']['phase']=='Succeeded','probe_pod_not_succeeded')
+            raw=command(kube+['-n',sender,'logs',pods[0]['metadata']['name'],'-c','probe']).stdout
+            require(len(raw)<=8192,'probe_receipt_too_large')
+            producer=json.loads(raw)
+            require(producer['status']=='passed' and producer['sender']==sender
+                    and producer['http_statuses']==[202,200] and producer['source_zip_sha256']==packet['source_zip_sha256'], 'probe_receipt_mismatch')
+            producer.update(pod_uid=pods[0]['metadata']['uid'],image=IMAGE,
+                            runtime_image_id=pods[0]['status']['containerStatuses'][0]['imageID'])
+            record=read_http(origin,reader_token,'/v1/events/'+producer['event_id'])
+            require(record['event_id']==producer['event_id'] and record['source']==sender
+                    and record['tenant']=='tenant:platform','independent_readback_scope_mismatch')
+            detail=record['details']['data']
+            require(detail.get('synthetic') is True or detail.get('details',{}).get('synthetic') is True,'probe_not_marked_synthetic')
+            producer['independent_readback']={'event_id':record['event_id'],'accepted_at':record['accepted_at'],
+                                            'source':record['source'],'tenant':record['tenant'],'synthetic':True}
+            receipt['producers'].append(producer);save()
+        receipt['integrity_after']=read_http(origin,reader_token,'/v1/integrity')
+        require(receipt['integrity_after']['intact'] is True,'chain_not_intact_after_probes')
+        require(receipt['integrity_after']['events']>=receipt['integrity_before']['events']+2,'two_probe_events_missing')
+        receipt.update(status='native_producer_delivery_verified_pending_bearer_revocation_and_service_admission',phase='complete');save()
+    finally:
+        forward.terminate();forward.wait(timeout=10)
+        # Delete only this run's objects with UID preconditions, Jobs first.
+        for obj in reversed(created):
+            meta=obj['metadata'];kind=obj['kind'];plural={'Job':'jobs','ConfigMap':'configmaps','NetworkPolicy':'networkpolicies'}[kind]
+            prefix='/api/v1' if obj['apiVersion']=='v1' else '/apis/'+obj['apiVersion']
+            uri=prefix+'/namespaces/'+meta['namespace']+'/'+plural+'/'+meta['name']
+            command(kube+['delete','--raw='+uri,'-f','/dev/stdin'],payload={'apiVersion':'v1','kind':'DeleteOptions','propagationPolicy':'Foreground','preconditions':{'uid':meta['uid']}})
+            if kind=='Job':
+                # Keep the probe's isolation policy until every dependent pod is gone.
+                command(kube+['-n',meta['namespace'],'wait','--for=delete','job/'+meta['name'],'--timeout=45s'])
+        receipt['cleanup_requested']=True;save()
+
+
+def main():
+    p=argparse.ArgumentParser(description=__doc__);sub=p.add_subparsers(dest='action',required=True)
+    prep=sub.add_parser('prepare');prep.add_argument('--source-root',type=Path,required=True);prep.add_argument('--packet',type=Path,required=True)
+    native=sub.add_parser('run');native.add_argument('--packet',type=Path,required=True);native.add_argument('--packet-sha256',required=True)
+    native.add_argument('--kubeconfig',required=True);native.add_argument('--server',required=True);native.add_argument('--receipt',type=Path,required=True);native.add_argument('--confirm',required=True)
+    a=p.parse_args()
+    if a.action=='prepare':print(json.dumps(prepare(a.source_root,a.packet)));return 0
+    receipt={'schema':'platform.factory-native-acceptance-receipt.v1','status':'refused','credential_values_emitted':False,'started_at':datetime.now(timezone.utc).isoformat()};fd=None
+    try:
+        packet=validate_packet(a.packet,a.packet_sha256)
+        require(a.confirm==CONFIRM,'exact_confirmation_required')
+        require(Path.home().parent.name=='.warden-attended-login' and not os.getenv('BAO_TOKEN') and not os.getenv('VAULT_TOKEN'),'attended_warden_envelope_required')
+        fd=os.open(a.receipt,os.O_RDWR|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
+        def save():
+            os.lseek(fd,0,os.SEEK_SET);os.ftruncate(fd,0);os.write(fd,(json.dumps(receipt,indent=2)+'\n').encode());os.fsync(fd)
+        receipt.update(run_id=packet['run_id'],packet_sha256=a.packet_sha256,source_commits=packet['source_commits']);save()
+        run(packet,['kubectl','--kubeconfig',a.kubeconfig,'--server',a.server,'--request-timeout=20s'],receipt,save)
+        receipt['completed_at']=datetime.now(timezone.utc).isoformat();save();return 0
+    except BaseException as exc:
+        receipt.update(status='refused',error=str(exc) if isinstance(exc,LaneError) else 'contained_native_acceptance_failed')
+        if fd is not None:save()
+        return 1
+    finally:
+        if fd is not None:os.close(fd)
+
+
+if __name__=='__main__':raise SystemExit(main())