tegwick
c5367a5a54
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
150 lines
11 KiB
Python
150 lines
11 KiB
Python
"""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()
|