Separate producer checks from attended audit readback

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

docs/credential-lane-designs/factory-native-acceptance.md

@@ -27,7 +27,7 @@ bearer refusal, one own-source reconciliation count and sibling-count refusal.
 No domain approval/disposition API is called, and no production heartbeat is
 emitted. Domain-transaction atomicity is not retested by synthetic outbox seeding.
 
-The attended parent reads the authoritative registry only in memory, verifies
+The optional attended readback parent reads the authoritative registry only in memory, verifies
 the two exact sender scopes, selects an unambiguous existing independent
 read-only full-tenant operator, and retrieves only the two named synthetic
 events and chain-integrity metadata through a private loopback port-forward.
@@ -51,7 +51,7 @@ python3 scripts/native_factory_acceptance.py run \
   --confirm 'VERIFY CCR-2026-0021 CCR-2026-0022 PRODUCERS'
 ```
 
-Four preparation/guard/integration tests pass. The integration test runs both
+Five preparation/guard/integration tests pass. The integration test runs both
 real source outboxes in the pinned image against actual local Audit Core and
 preserves exactly two events with an intact chain. All six native objects pass
 server dry-run. Neither rehearsal result is claimed as native evidence.
@@ -62,3 +62,23 @@ bearer. CCRs stay applied until their remaining lifecycle acceptance exists;
 rotation/revocation must use its separately reviewed owner procedure. Service
 startup, human binding, native policy/caller admission, attestation/offsite
 operation and factory execution remain in their existing owner records.
+
+## Consume existing producer credentials without an operator login
+
+The first combined attempt failed at OIDC before command handoff; no native
+job ran. Split normal producer consumption from privileged operator readback.
+`jobs` uses only Kubernetes metadata and the existing approved Secret references;
+it never reads OpenBao or obtains an operator token. The audit credential is
+read solely inside its own source-pinned producer Job. The same packet, exact
+confirmation, native receiver check and bounded resource/cleanup gates apply.
+
+Run `jobs` with the same options as `run`. A successful receipt has status
+`native_sender_checks_passed_pending_independent_readback`. Use a fresh receipt
+path, and preserve it. After successful jobs, do not use `run` again: repeat
+sends must not be mistaken for a new first acceptance.
+
+For independent readback, use `readback` inside a fresh attended envelope with
+`--producer-receipt <successful-jobs-receipt>` and a new `--receipt`. It verifies
+the exact packet/run/event identities and reads only those already stored
+probe records and chain metadata. It creates no Job and emits no new event.
+The operator login remains required for that separate registry-backed reader.

scripts/native_factory_acceptance.py

@@ -137,35 +137,17 @@ def operator_identity(rows):
     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')
+def jobs(packet, kube, receipt, save):
+    """Consume only the producer's already-delivered credential inside its Job."""
+    receipt['receiver_before']=receiver_check(kube,RECEIVER)
+    for lane in contracts(approved=True):
         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)
+    receipt.update(phase='native_jobs',operator_registry_reads=0);save()
     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))
@@ -188,22 +170,13 @@ def run(packet, kube, receipt, save):
             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')
+            require(any(o['uid']==job['metadata']['uid'] and o.get('controller') is True for o in pods[0]['metadata'].get('ownerReferences',[])), 'probe_pod_owner_mismatch')
+            require(producer['event_id']==packet['run_id']+'-'+sender,'probe_event_id_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()
+        receipt.update(status='native_sender_checks_passed_pending_independent_readback',phase='sender_checks_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]
@@ -216,23 +189,79 @@ def run(packet, kube, receipt, save):
         receipt['cleanup_requested']=True;save()
 
 
+
+def readback(packet, kube, receipt, save):
+    """Operator-only archive readback; never rerun jobs or expose the registry."""
+    require(receipt['status']=='native_sender_checks_passed_pending_independent_readback'
+            and receipt['cleanup_requested'] is True and receipt['run_id']==packet['run_id'], 'successful_sender_receipt_required')
+    require({p['sender'] for p in receipt['producers']}==set(SOURCES) and len(receipt['producers'])==2,'exact_producer_receipts_required')
+    for producer in receipt['producers']:
+        require(producer['event_id']==packet['run_id']+'-'+producer['sender']
+                and producer['source_zip_sha256']==packet['source_zip_sha256']
+                and producer['status']=='passed' and producer['http_statuses']==[202,200],'producer_receipt_mismatch')
+    receipt['receiver_at_readback']=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 contracts(approved=True):
+        registered=next(r for r in rows if r['name']==lane['name'])
+        require(registered==desired(lane,registered['tokens'][0]),'sender_registry_scope_changed')
+    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='independent_readback');save()
+    with socket.socket() as sock:sock.bind(('127.0.0.1',0));port=sock.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)
+    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)
+        for producer in receipt['producers']:
+            record=read_http(origin,reader_token,'/v1/events/'+producer['event_id'])
+            require(record['event_id']==producer['event_id'] and record['source']==producer['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};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']>=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)
+
+
 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)
+    for name in ['run','jobs','readback']:
+        native=sub.add_parser(name);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)
+        if name=='readback':native.add_argument('--producer-receipt',type=Path,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')
+        if a.action!='jobs':
+            require(Path.home().parent.name=='.warden-attended-login' and not os.getenv('BAO_TOKEN') and not os.getenv('VAULT_TOKEN'),'attended_warden_envelope_required')
+        if a.action=='readback':
+            prior=json.loads(a.producer_receipt.read_text())
+            require(prior['packet_sha256']==a.packet_sha256 and prior['run_id']==packet['run_id'],'producer_packet_mismatch')
+            receipt.update(prior)
+            receipt['readback_started_at']=datetime.now(timezone.utc).isoformat()
         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)
+        kube=['kubectl','--kubeconfig',a.kubeconfig,'--server',a.server,'--request-timeout=20s']
+        if a.action in {'run','jobs'}:jobs(packet,kube,receipt,save)
+        if a.action in {'run','readback'}:readback(packet,kube,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')

tests/test_factory_native_acceptance.py

@@ -9,6 +9,7 @@ import sys
 import tempfile
 import threading
 import unittest
+from unittest.mock import patch
 from wsgiref.simple_server import make_server, WSGIRequestHandler
 
 sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts'))
@@ -16,6 +17,14 @@ import native_factory_acceptance as native
 
 
 class Contracts(unittest.TestCase):
+    def test_jobs_do_not_read_operator_credentials_and_refuse_additive_egress(self):
+        result=subprocess.CompletedProcess([],0,b'{"items":[{"metadata":{"name":"existing-allow"}}]}',b'')
+        with patch.object(native,'receiver_check',return_value={}),patch.object(native,'command',return_value=result) as cmd,patch.object(native,'bao',side_effect=AssertionError('no OpenBao call permitted')):
+            with self.assertRaisesRegex(native.LaneError,'existing_egress_policy_requires_review'):
+                native.jobs({},['kubectl'],{},lambda:None)
+            self.assertEqual(cmd.call_count,1)
+            self.assertNotIn('create',cmd.call_args.args[0])
+
     def test_existing_independent_reader_must_be_unambiguous_and_read_only(self):
         reader={'name':'operator','may_read':True,'may_write':False,'tenants':['*']}
         self.assertEqual(native.operator_identity([reader]),reader)