diff --git a/docs/credential-lane-designs/factory-native-acceptance.md b/docs/credential-lane-designs/factory-native-acceptance.md index 02f009b..864f4ee 100644 --- a/docs/credential-lane-designs/factory-native-acceptance.md +++ b/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 ` 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. diff --git a/scripts/native_factory_acceptance.py b/scripts/native_factory_acceptance.py index 0b5ce50..a05b423 100644 --- a/scripts/native_factory_acceptance.py +++ b/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') diff --git a/tests/test_factory_native_acceptance.py b/tests/test_factory_native_acceptance.py index d20d70d..df1096e 100644 --- a/tests/test_factory_native_acceptance.py +++ b/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)