"""Exercise a local image, private volumes and restore without external network. Only synthetic records/tokens are created. No native issuer, policy, audit or Approval Engine is contacted. Every container and volume is created by this process, labeled, and removed in finally; the image is retained for inspection. """ import argparse import hashlib import json import os from pathlib import Path import subprocess import tempfile import time import uuid def main(): parser=argparse.ArgumentParser(description=__doc__) parser.add_argument('--image',required=True) parser.add_argument('--receipt',type=Path,required=True) args=parser.parse_args() prefix='infd-proof-'+uuid.uuid4().hex[:12] volumes=[];containers=[];checks=[] def docker(*words,check=True): return subprocess.run(['docker',*words],capture_output=True,text=True,check=check,timeout=60) def command(name,code): return docker('exec',name,'python','-c',code).stdout.strip() def admin(name,*args): return json.loads(docker('exec',name,'informed-decision-admin',*args).stdout) def stop(name): docker('stop','--time','15',name) state=json.loads(docker('inspect',name).stdout)[0]['State'] assert state['ExitCode']==0, 'service did not stop gracefully: '+str(state['ExitCode']) def wait_health(name): for _ in range(40): result=docker('exec',name,'python','-c', "import urllib.request; assert urllib.request.urlopen('http://127.0.0.1:8080/healthz',timeout=1).status==200",check=False) if result.returncode==0:return state=json.loads(docker('inspect',name).stdout)[0]['State'] if not state['Running']:raise AssertionError('container exited before health; '+docker('logs',name).stdout) time.sleep(.2) raise AssertionError('container did not become healthy') def volume(suffix): name=prefix+'-'+suffix docker('volume','create','--label','informed-decision.fixture='+prefix,name);volumes.append(name) docker('run','--rm','--network','none','--read-only','--user','0:0','--cap-drop','ALL','--cap-add','CHOWN','--cap-add','FSETID', '--mount','type=volume,src='+name+',dst=/data','--entrypoint','python',args.image,'-c', "import os,stat; os.chown('/data',0,10001); os.chmod('/data',0o2770); assert stat.S_IMODE(os.stat('/data').st_mode)==0o2770") return name def start(suffix,data,fixture,backup=None): name=prefix+'-'+suffix;containers.append(name) words=['run','-d','--name',name,'--label','informed-decision.fixture='+prefix, '--network','none','--read-only','--cap-drop','ALL','--security-opt','no-new-privileges', '--tmpfs','/run/informed-decision:rw,nosuid,nodev,noexec,uid=10001,gid=10001,mode=0700', '--tmpfs','/tmp:rw,nosuid,nodev,noexec,mode=1777', '--mount','type=volume,src='+data+',dst=/data', '--mount','type=bind,src='+str(fixture)+',dst=/configuration,readonly', '--env','INFD_KEYCAPE_ISSUER=https://kc.coulomb.social'] if backup:words+=['--mount','type=volume,src='+backup+',dst=/backup'] docker(*words,args.image);return name snapshot="""import hashlib,json from informed_decision.store import Store s=Store('/data/private/review.sqlite') with s._connection() as db: rows={t:[list(r) for r in db.execute('SELECT * FROM '+t+' ORDER BY rowid')] for t in ['memos','presentations','dispositions','submissions','evidence','documents']} rows['documents']=[[r[0],r[1],hashlib.sha256(r[2]).hexdigest()] for r in rows['documents']] print(hashlib.sha256(json.dumps(rows,sort_keys=True).encode()).hexdigest()) """ seed="""from informed_decision.store import Store from informed_decision.memo import Memo,BindingSlice,Principal,Scope,BindingLevel,StepKind,PacketItem from informed_decision.provenance import Claim,Route from informed_decision.disposition import Actor,ActorKind,Verb s=Store('/data/private/review.sqlite');digest=s.put_document(b'Synthetic retained content; no real approval.') m=Memo(id='container-fixture',version=1,question='Synthetic custody exercise?',requested_act='deliver',binding_level=BindingLevel.ORGANIZATIONAL,brief='Fixture only',binding=BindingSlice(Principal('fixture-human','person','Fixture'),Scope('resource','fixture','Fixture')),step_kind=StepKind.APPROVE,packet=(PacketItem('doc','Fixture',digest),),approval_id='fixture',approval_binding_digest='sha256:'+'1'*64) s.save_memo(m) p=s.present(m.id,principal_sub='fixture-human',tenant=Claim('tenant:platform',Route.REGISTRATION),principal_type=Claim('human',Route.AUTHENTICATION)) d=s.record_disposition(p.id,Actor('fixture-human',ActorKind.PERSON),Verb.ACCEPT,operation_id='container-fixture-op') attempt=s.begin_submission(d.id);s.finish_submission(d.id,attempt) assert s.submission(d.id)['state']=='unresolved' """ try: with tempfile.TemporaryDirectory(prefix=prefix+'-') as temp: fixture=Path(temp);fixture.chmod(0o755) (fixture/'audit-token').write_text('synthetic-audit-token');(fixture/'caller-token').write_text('synthetic-caller-token') config={'schema':'informed-decision.review-runtime.v1','evidence_db':'/data/private/review.sqlite', 'approval_origin':'http://127.0.0.1:18082','policy':{'origin':'http://127.0.0.1:18083', 'package':'container.fixture','version':'v1','package_digest':'sha256:'+'a'*64, 'caller_token_file':'/configuration/caller-token'}, 'audit':{'origin':'http://127.0.0.1:18084','sender_token_file':'/configuration/audit-token'}} (fixture/'runtime.json').write_text(json.dumps(config)) for file in fixture.iterdir():file.chmod(0o444) data=volume('data');backup=volume('backup') first=start('first',data,fixture,backup);wait_health(first) checks.append('installed container entrypoint serves with projected configuration') info=json.loads(docker('inspect',first).stdout)[0] assert info['Config']['User']=='10001:10001' and info['HostConfig']['ReadonlyRootfs'] assert info['HostConfig']['NetworkMode']=='none' and not info['NetworkSettings']['Ports'].get('8080/tcp') checks.append('non-root read-only runtime has no external network or published port') command(first,"import os,stat,importlib.util; assert os.getuid()==10001; assert importlib.util.find_spec('pip') is None; assert stat.S_IMODE(os.stat('/data/private').st_mode)==0o700; assert stat.S_IMODE(os.stat('/data/private/review.sqlite').st_mode)==0o600; assert stat.S_IMODE(os.stat('/run/informed-decision/private/runtime.json').st_mode)==0o600") checks.append('private config/database modes enforced; package installer absent') command(first,"import urllib.request,urllib.error\nfor path,status in [('/readyz',503),('/review?memo_id=missing',401)]:\n try: urllib.request.urlopen('http://127.0.0.1:8080'+path)\n except urllib.error.HTTPError as e: assert e.code==status\n else: raise AssertionError(path)") checks.append('configured review refuses anonymous access and reports audit-unready status') command(first,"from informed_decision.container import single_writer\ntry:\n with single_writer('/data/private'): pass\nexcept ValueError: pass\nelse: raise AssertionError('second writer admitted')") checks.append('a second writer cannot acquire the active evidence volume') command(first,seed) state=admin(first,'inspect','--db','/data/private/review.sqlite') assert state['submissions']=={'unresolved':1} and state['outbox'].get('pending',0)>0 original=command(first,snapshot) checks.append('synthetic uncertain intent and undelivered evidence remain explicit') command(first,"from informed_decision.container import private_directory; private_directory('/backup/private')") result=admin(first,'backup','--db','/data/private/review.sqlite','--output','/backup/private/review.sqlite') assert result['consistent_snapshot'] is True repeat=docker('exec',first,'informed-decision-admin','backup','--db','/data/private/review.sqlite','--output','/backup/private/review.sqlite',check=False) assert repeat.returncode!=0 checks.append('consistent backup created while serving; overwrite refused') stop(first) checks.append('SIGTERM closes the serving writer without forced kill') docker('start',first);wait_health(first) assert command(first,snapshot)==original assert admin(first,'inspect','--db','/data/private/review.sqlite')['submissions']=={'unresolved':1} checks.append('restart preserves immutable content and unresolved submission without retry') stop(first) restored=start('restored',backup,fixture);wait_health(restored) assert command(restored,snapshot)==original assert admin(restored,'inspect','--db','/data/private/review.sqlite')['submissions']=={'unresolved':1} checks.append('separate restored volume preserves exact content and unresolved state') stop(restored) (fixture/'runtime.json').chmod(0o644);(fixture/'runtime.json').write_text('{}');(fixture/'runtime.json').chmod(0o444) invalid=start('invalid',data,fixture) code=int(docker('wait',invalid).stdout) assert code!=0 checks.append('incomplete owner configuration exits before serving') image=json.loads(docker('image','inspect',args.image).stdout)[0] receipt={'status':'passed','image_id':image['Id'],'repo_digests':image['RepoDigests'], 'checks_passed':len(checks),'checks':checks,'fixture_only':True,'external_network':'none', 'content_snapshot_sha256':original,'native_identity_policy_audit_proven':False, 'published':False,'deployed':False,'factory_attempts':0,'paid_model_calls':0} finally: for name in reversed(containers):docker('rm','-f',name,check=False) for name in reversed(volumes):docker('volume','rm',name,check=False) receipt['cleanup_complete']=all(docker('inspect',name,check=False).returncode!=0 for name in containers) receipt['cleanup_complete'] &= all(docker('volume','inspect',name,check=False).returncode!=0 for name in volumes) assert receipt['cleanup_complete'] args.receipt.write_text(json.dumps(receipt,indent=2)+'\n') print(json.dumps({'status':'passed','checks_passed':len(checks),'cleanup_complete':True,'image_id':receipt['image_id']})) if __name__=='__main__':main()