"""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, receipt): observed = receipt.setdefault("observed_http_statuses", []) 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['phase'] = sys.argv[1] receipt['http_status'] = delivery_phase(config,sys.argv[1],dbpath,receipt) receipt['status'] = 'passed' 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'}} receipt['phase'] = 'bootstrap' envelope = bootstrap(config,dbpath,envelope) phases=[] for phase in ['lost-receipt','restart-retry']: receipt['phase'] = phase child = subprocess.run([sys.executable,__file__,phase],capture_output=True,timeout=30) child_receipt = json.loads(child.stdout) if child.returncode: receipt['child_error'] = child_receipt.get('error','contained_probe_failed') receipt['observed_http_statuses'] = child_receipt.get('observed_http_statuses',[]) raise Refused(phase+'_failed') phases.append(child_receipt['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())