From 1770a6067530677c7ebb6531175e9615cbca8bd5 Mon Sep 17 00:00:00 2001 From: codex Date: Fri, 11 Sep 2026 11:31:35 +0200 Subject: [PATCH] feat: verify native factory sender delivery and outbox replay Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc --- .../factory-native-acceptance.md | 64 +++++ scripts/native_factory_acceptance.py | 245 ++++++++++++++++++ scripts/native_factory_probe/producer.py | 189 ++++++++++++++ tests/test_factory_native_acceptance.py | 93 +++++++ 4 files changed, 591 insertions(+) create mode 100644 docs/credential-lane-designs/factory-native-acceptance.md create mode 100644 scripts/native_factory_acceptance.py create mode 100644 scripts/native_factory_probe/producer.py create mode 100644 tests/test_factory_native_acceptance.py diff --git a/docs/credential-lane-designs/factory-native-acceptance.md b/docs/credential-lane-designs/factory-native-acceptance.md new file mode 100644 index 0000000..02f009b --- /dev/null +++ b/docs/credential-lane-designs/factory-native-acceptance.md @@ -0,0 +1,64 @@ +# Bounded native factory sender acceptance + +RPF-WP-0035-T08 and AUDIT-WP-0009-T09/T11 retain this acceptance under approved +CCR-2026-0021/0022. Custody and delivery are already complete. This operation +creates no credential, registry entry, approval, disposition or UI service. + +The six exclusive resources are one immutable ConfigMap, one probe-only +NetworkPolicy and one Job in each producer namespace. Jobs mount only their +own delivered audit field, have no API token, run as UID 10001 with read-only +root and no capabilities, and use disposable private SQLite state. Egress is +restricted to Audit Core's receiver and cluster DNS; existing namespace +NetworkPolicies refuse preparation for native execution because grants add. +Jobs have a 150-second deadline and no retry. Cleanup is UID guarded and waits +for each Job and its pods to disappear before removing its isolation policy. + +The runtime image is the already published Approval Engine 251941a5 digest. +The immutable ZIP contains 31 actual producer Python files from Approval +Engine a0a60297 and Informed Decision bda9381f. Both sources and the full packet +are hash pinned. This proves source adapter/outbox integration on Railiance; +it does not publish or admit Informed Decision's full service image. + +Each job seeds exactly one explicitly synthetic outbox record in its disposable +store. It sends through the real audit adapter, loses the successful 202 receipt, +starts a fresh Python process, and drains the preserved outbox with a 200 duplicate. +It checks exact source/tenant refusals, seven evidence-read refusals, invalid +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 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. +Credentials never appear in arguments, stdout, logs or receipts. This uses the +reviewed platform-admin envelope; no producer receives the operator registry. + +Prepare and inspect before opening the attended window: + +```sh +python3 scripts/native_factory_acceptance.py prepare \ + --source-root /home/worsch --packet /operator/unique-packet.json +``` + +The result names the packet digest. Under `scripts/openbao-attended-exec.py`: + +```sh +python3 scripts/native_factory_acceptance.py run \ + --packet /operator/unique-packet.json --packet-sha256 REVIEWED_SHA256 \ + --kubeconfig /operator/railiance-kubeconfig --server https://127.0.0.1:16444 \ + --receipt /operator/unique-native-receipt.json \ + --confirm 'VERIFY CCR-2026-0021 CCR-2026-0022 PRODUCERS' +``` + +Four 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. + +Success remains `native_producer_delivery_verified_pending_bearer_revocation_and_service_admission`. +Invalid bearer refusal does not prove revocation of a previously admitted audit +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. diff --git a/scripts/native_factory_acceptance.py b/scripts/native_factory_acceptance.py new file mode 100644 index 0000000..0b5ce50 --- /dev/null +++ b/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()) diff --git a/scripts/native_factory_probe/producer.py b/scripts/native_factory_probe/producer.py new file mode 100644 index 0000000..3f61b3f --- /dev/null +++ b/scripts/native_factory_probe/producer.py @@ -0,0 +1,189 @@ +"""Bounded synthetic audit probe; credentials stay in the producer process.""" +import hashlib +import json +import os +from pathlib import Path +import sqlite3 +import subprocess +import sys +import time +from datetime import datetime, timezone, timedelta +from urllib.error import HTTPError, URLError +from urllib.parse import urlencode +from urllib.request import Request, build_opener, ProxyHandler, HTTPRedirectHandler + + +class Refused(Exception): + pass + + +def require(value, code): + if not value: + raise Refused(code) + + +class NoRedirect(HTTPRedirectHandler): + def redirect_request(self, *args): + return None + + +def http(config, method, path, *, body=None, event_id=None, invalid_token=False): + token = 'deliberately-invalid-native-probe' if invalid_token else Path('/credential/token').read_text().strip() + headers = {'Authorization': 'Bearer '+token, 'Content-Type': 'application/json'} + if event_id: + headers['Idempotency-Key'] = event_id + request = Request(config['origin']+path, data=body, headers=headers, method=method) + opener = build_opener(ProxyHandler({}), NoRedirect()) + try: + response = opener.open(request, timeout=5) + except HTTPError as exc: + response = exc + with response: + raw = response.read(262145) + require(len(raw) <= 262144, 'response_too_large') + return response.status, json.loads(raw) + + +def bootstrap(config, dbpath, envelope): + # Seed only a disposable synthetic outbox, never an approval or disposition. + # This exercises recovery/delivery, not domain-transaction atomicity. + if config['sender'] == 'approval-engine': + from approval_engine.store import Engine + engine = Engine(dbpath) + payload = {'schema_version':'audit-core.event.v1alpha1', 'event_id':envelope['id'], + 'observed_at':envelope['occurred_at'], 'tenant':'tenant:platform', + 'scope':'synthetic-audit-acceptance', 'source':'approval-engine', + 'actor':None, 'action':envelope['type'], 'resource':envelope['subject'], + 'outcome':'probe', 'reason':'synthetic sender verification', + 'details':dict(envelope['data'], approval_id=envelope['correlation_id'])} + from approval_engine.audit import audit_envelope + envelope = audit_envelope(payload) + engine.close() + with sqlite3.connect(dbpath) as db: + require(db.execute('SELECT COUNT(*) FROM approvals').fetchone()[0] == 0, 'unexpected_approval') + db.execute('INSERT INTO outbox(event_id,class,approval_id,payload_json,created_at) VALUES(?,?,?,?,?)', + (envelope['id'], envelope['type'], None, json.dumps(payload, sort_keys=True), envelope['occurred_at'])) + else: + from informed_decision.store import Store + Store(dbpath) + with sqlite3.connect(dbpath) as db: + db.execute('INSERT INTO evidence VALUES(?,?,?,?,?)', + (envelope['id'], envelope['type'], envelope['occurred_at'], json.dumps(envelope,sort_keys=True), '{}')) + db.execute('INSERT INTO outbox(id) VALUES(?)', (envelope['id'],)) + return envelope + + +def delivery_phase(config, phase, dbpath): + observed = [] + lose = phase == 'lost-receipt' + if config['sender'] == 'approval-engine': + from approval_engine.store import Engine + from approval_engine.audit import AuditCoreSink + real = build_opener(ProxyHandler({}), NoRedirect()) + def opened(request, timeout): + response = real.open(request, timeout=timeout) + observed.append(response.status) + if lose: + response.close() + raise URLError('injected receipt loss after native response') + return response + engine = Engine(dbpath) + pending = engine.undrained() + require(len(pending) == 1 and pending[0]['event_id'] == config['event_id'], 'outbox_not_preserved') + result = engine.drain(AuditCoreSink(config['origin'], '/credential/token', opener=opened)) + require(result == ({'delivered':0,'failed':1} if lose else {'delivered':1,'failed':0}), 'unexpected_drain_result') + require(bool(engine.undrained()) == lose, 'unexpected_outbox_state') + engine.close() + else: + from informed_decision.store import Store + from informed_decision.audit import AuditCoreSink, OutboxWorker + from informed_decision.http_transport import JSONTransport, TransportError + class Transport(JSONTransport): + def request(self, *args, **kwargs): + status, body = super().request(*args, **kwargs) + observed.append(status) + if lose: + raise TransportError('injected receipt loss after native response') + return status, body + store = Store(dbpath) + row = store.outbox()[0] + require(row['id'] == config['event_id'] and row['state'] == 'pending', 'outbox_not_preserved') + if not lose: + time.sleep(max(0, row['next_attempt']-time.time())+.05) + sink = AuditCoreSink(config['origin'], lambda:Path('/credential/token').read_text().strip(), + transport=Transport(allow_internal_http=True), allow_internal_http=True) + result = OutboxWorker(store, sink).run_once() + require(result == ({'delivered':0,'retrying':1,'blocked':0} if lose else {'delivered':1,'retrying':0,'blocked':0}), 'unexpected_drain_result') + require(store.outbox()[0]['state'] == ('pending' if lose else 'delivered'), 'unexpected_outbox_state') + require(observed == ([202] if lose else [200]), 'native_accept_duplicate_not_observed') + return observed[0] + + +def negatives(config, envelope): + result = {} + for field, value, error in [('source','sibling-not-admitted','source_not_allowed'), + ('tenant','tenant:synthetic-denied','tenant_not_allowed')]: + body = dict(envelope, **{field:value, 'id':config['event_id']+'-'+field}) + status, reply = http(config,'POST','/v1/events',body=json.dumps(body).encode(),event_id=body['id']) + require(status == 400 and reply.get('error') == error, field+'_refusal_inconclusive') + result[field+'_denied'] = status + paths = ['/v1/events/'+config['event_id'], '/v1/events?correlation_id='+config['run_id'], + '/v1/stats','/v1/integrity','/v1/dead-letters','/v1/secret-findings','/v1/stream-findings'] + for path in paths: + status, body = http(config,'GET',path) + require(status == 403 and body.get('error') == 'read_forbidden', 'read_refusal_inconclusive') + result['read_routes_denied'] = len(paths) + status, body = http(config,'POST','/v1/events',body=json.dumps(envelope).encode(),event_id=envelope['id'],invalid_token=True) + require(status == 401, 'invalid_bearer_not_refused') + result['invalid_bearer_denied'] = 401 + since = (datetime.now(timezone.utc)-timedelta(minutes=5)).isoformat() + until = (datetime.now(timezone.utc)+timedelta(seconds=1)).isoformat() + query = {'source':config['sender'],'tenant':'tenant:platform','since':since,'until':until} + status, body = http(config,'GET','/v1/reconciliation?'+urlencode(query)) + require(status == 200 and body.get('source') == config['sender'] and body.get('tenant') == 'tenant:platform', 'own_reconciliation_failed') + require(any(x.get('class') == envelope['type'] and x.get('count') == 1 for x in body['counts']), 'probe_count_not_one') + query['source'] = 'sibling-not-admitted' + status, body = http(config,'GET','/v1/reconciliation?'+urlencode(query)) + require(status == 403 and body.get('error') == 'source_not_allowed', 'sibling_reconciliation_not_refused') + result.update(own_reconciliation=True, sibling_reconciliation_denied=True, probe_count=1, + audit_bearer_revocation_tested=False) + return result + + +def main(): + receipt = {'status':'refused', 'credential_values_emitted':False} + try: + config = json.loads(Path('/probe/config.json').read_text()) + require(hashlib.sha256(Path('/probe/source.zip').read_bytes()).hexdigest() == config['source_zip_sha256'], 'source_bundle_mismatch') + sys.path.insert(0,'/probe/source.zip') + os.umask(0o077) + directory = Path('/state/private'); directory.mkdir(mode=0o700,exist_ok=True) + dbpath = directory/'outbox.db' + if len(sys.argv)>1: + receipt = {'status':'passed','http_status':delivery_phase(config,sys.argv[1],dbpath)} + else: + require(not dbpath.exists(), 'fresh_probe_state_required') + envelope = {'id':config['event_id'], 'type':'factory.audit-probe.'+config['run_id'], + 'source':config['sender'], 'subject':'synthetic:'+config['run_id'], + 'tenant':'tenant:platform', 'correlation_id':config['run_id'], + 'occurred_at':datetime.now(timezone.utc).isoformat(), + 'data':{'synthetic':True,'no_human_action':True,'purpose':'audit-sender-acceptance'}} + envelope = bootstrap(config,dbpath,envelope) + phases=[] + for phase in ['lost-receipt','restart-retry']: + child = subprocess.run([sys.executable,__file__,phase],capture_output=True,timeout=30) + require(child.returncode == 0, phase+'_failed') + phases.append(json.loads(child.stdout)['http_status']) + receipt.update(status='passed', sender=config['sender'], event_id=config['event_id'], + run_id=config['run_id'], source_zip_sha256=config['source_zip_sha256'], + http_statuses=phases, process_restart_retry=True, synthetic_outbox_only=True, + domain_transaction_atomicity_tested=False, human_binding_tested=False, + producer_service_deployed=False, negatives=negatives(config,envelope)) + except BaseException as exc: + receipt.update(status='refused', error=str(exc) if isinstance(exc,Refused) else 'contained_probe_failed') + print(json.dumps(receipt,sort_keys=True)) + return 0 if receipt['status']=='passed' else 1 + + +if __name__=='__main__': + raise SystemExit(main()) diff --git a/tests/test_factory_native_acceptance.py b/tests/test_factory_native_acceptance.py new file mode 100644 index 0000000..d20d70d --- /dev/null +++ b/tests/test_factory_native_acceptance.py @@ -0,0 +1,93 @@ +"""Real producer adapters and outbox retries against an actual local receiver.""" +import base64 +import copy +import hashlib +import json +from pathlib import Path +import subprocess +import sys +import tempfile +import threading +import unittest +from wsgiref.simple_server import make_server, WSGIRequestHandler + +sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts')) +import native_factory_acceptance as native + + +class Contracts(unittest.TestCase): + 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) + for rows in [[],[dict(reader,may_write=True)],[dict(reader,tenants=['tenant:other'])], + [reader,dict(reader,name='second-operator')]]: + with self.assertRaises(native.LaneError):native.operator_identity(rows) + + def test_packet_cannot_redirect_secret_mount_or_add_network(self): + with tempfile.TemporaryDirectory() as tmp: + path=Path(tmp)/'packet.json';native.prepare(Path('/home/worsch'),path) + sha=lambda:hashlib.sha256(path.read_bytes()).hexdigest() + packet=native.validate_packet(path,sha()) + job=packet['objects'][2] + job['spec']['template']['spec']['volumes'][1]['secret']['secretName']='unrelated-secret' + path.write_text(json.dumps(packet)) + with self.assertRaisesRegex(native.LaneError,'resource_scope_changed'):native.validate_packet(path,sha()) + + def test_changed_probe_bytes_or_packet_digest_refuse(self): + with tempfile.TemporaryDirectory() as tmp: + path=Path(tmp)/'packet.json';meta=native.prepare(Path('/home/worsch'),path) + path.write_bytes(path.read_bytes()+b'\n') + with self.assertRaisesRegex(native.LaneError,'packet_digest_changed'): + native.validate_packet(path,meta['sha256']) + + +class Quiet(WSGIRequestHandler): + def log_message(self,*args):pass + + +class LocalReceiver(unittest.TestCase): + def test_two_real_outboxes_retry_lost_receipts_without_duplicate_records(self): + sys.path.insert(0,'/home/worsch/audit-core') + from audit_core.ingestion import IngestionApplication + from audit_core.sqlite_backend import SQLiteAuditBackend + from audit_core.senders import SenderRegistry + with tempfile.TemporaryDirectory() as tmp: + root=Path(tmp);packet_path=root/'packet.json';native.prepare(Path('/home/worsch'),packet_path) + packet=json.loads(packet_path.read_text());tokens={s:'local-fixture-'+s for s in native.SOURCES} + registry=SenderRegistry.from_env({'AUDIT_CORE_SENDERS':json.dumps([ + {'name':s,'tokens':[t],'sources':[s],'tenants':['tenant:platform'],'may_write':True, + 'may_read':False,'evidence_kind':'load-bearing','secret_policy':'redact'} for s,t in tokens.items()])}) + backend=SQLiteAuditBackend(str(root/'receiver.db')) + server=make_server('127.0.0.1',0,IngestionApplication(backend,registry),handler_class=Quiet) + thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start() + try: + for cm in packet['objects'][::3]: + sender=cm['metadata']['namespace'];d=root/sender;d.mkdir();d.chmod(0o755) + for name,body in cm['data'].items():(d/name).write_text(body) + (d/'source.zip').write_bytes(base64.b64decode(cm['binaryData']['source.zip'])) + config=json.loads((d/'config.json').read_text());config['origin']='http://127.0.0.1:'+str(server.server_port) + (d/'config.json').write_text(json.dumps(config));token=d/'token';token.write_text(tokens[sender]);token.chmod(0o444) + argv=['docker','run','--rm','--network','host','--read-only','--cap-drop=ALL', + '--security-opt=no-new-privileges','--memory=192m', + '--tmpfs','/state:rw,nosuid,nodev,size=32m,mode=1777', + '--mount','type=bind,src='+str(d)+',dst=/probe,readonly', + '--mount','type=bind,src='+str(token)+',dst=/credential/token,readonly', + '--entrypoint','python',native.IMAGE,'-I','/probe/producer.py'] + result=subprocess.run(argv,capture_output=True,timeout=60) + # Only bounded probe JSON may enter test diagnostics. + try:receipt=json.loads(result.stdout) + except ValueError:receipt={'status':'container_receipt_missing','exit':result.returncode} + self.assertEqual(receipt['status'],'passed',receipt) + self.assertEqual(receipt['http_statuses'],[202,200]) + self.assertEqual(receipt['negatives']['read_routes_denied'],7) + self.assertTrue(receipt['process_restart_retry']) + record=backend.get(config['event_id']);self.assertEqual(record['source'],sender) + print(json.dumps({'local_sender':sender,'receipt':receipt,'stored_record_keys':list(record), + 'stored_data_keys':list(record['details']['data'])})) + self.assertEqual(backend.verify_chain().events,2) + self.assertTrue(backend.verify_chain().intact) + finally: + server.shutdown();server.server_close();thread.join(timeout=5) + + +if __name__=='__main__':unittest.main()