tegwick
4a59b234b9
All checks were successful
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a10780-6d41-7952-9718-dfbc1f41236f
63 lines
4 KiB
Python
63 lines
4 KiB
Python
"""Installed dispatcher proof, isolated synthetic queue, no network or credential."""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
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()
|
|
name='memo-proof-'+uuid.uuid4().hex[:12]
|
|
def docker(*words,check=True):
|
|
return subprocess.run(['docker',*words],check=check,capture_output=True,text=True,timeout=45)
|
|
checks=[]
|
|
try:
|
|
docker('volume','create',name)
|
|
docker('run','--rm','--network','none','--user','0:0','--mount','type=volume,src='+name+',dst=/data',
|
|
'--entrypoint','python',args.image,'-c',"import os;os.chown('/data',10001,10001);os.chmod('/data',0o700)")
|
|
docker('run','-d','--name',name,'--network','none','--read-only','--cap-drop','ALL',
|
|
'--security-opt','no-new-privileges','--mount','type=volume,src='+name+',dst=/data',args.image)
|
|
for _ in range(50):
|
|
r=docker('exec',name,'python','-m','api.services.memo_dispatch','--queue','/data/private','status',check=False)
|
|
if r.returncode==0:break
|
|
time.sleep(.1)
|
|
else:raise AssertionError('worker failed startup')
|
|
checks.append('non-root read-only worker starts without external network')
|
|
ident='review-'+str(uuid.uuid4())
|
|
envelope={'kind':'informed-decision.memo-intake','version':1,'idempotency_key':ident,
|
|
'memo_class':'decision-record','review_group':'net-kingdom-admins',
|
|
'memo':{'id':ident,'question':'Synthetic fixture only?'},
|
|
'packet':[{'item_id':'proof','media_type':'text/plain','content':'Fixture only.'}],
|
|
'approval_intent':{'action':'review','principal':'pending-human-session','purpose':'Synthetic test',
|
|
'target':{'id':'fixture','type':'decision-memo','system':'informed-decision'}}}
|
|
source="from api.services.memo_dispatch import Dispatcher; d=Dispatcher('/data/private'); assert d.enqueue("+repr(envelope)+").status=='pending'; assert d.counts()=={'pending':1}"
|
|
docker('exec',name,'python','-c',source)
|
|
checks.append('explicit enqueue persists a lifecycle envelope without inventing a Hub relationship')
|
|
digest_code="from api.services.memo_dispatch import Dispatcher;import hashlib;d=Dispatcher('/data/private');row=d.ledger.pose("+repr(ident)+");print(hashlib.sha256(row['envelope_json'].encode()).hexdigest())"
|
|
before=docker('exec',name,'python','-c',digest_code).stdout.strip()
|
|
docker('stop','--time','10',name)
|
|
state=json.loads(docker('inspect',name).stdout)[0]['State'];assert state['ExitCode']==0
|
|
checks.append('SIGTERM stops without forced kill')
|
|
docker('start',name)
|
|
after=docker('exec',name,'python','-c',digest_code).stdout.strip();assert before==after
|
|
checks.append('restart retains the exact pending envelope; missing custody does not drop it')
|
|
docker('exec',name,'python','-c',"import os,stat;assert os.getuid()==10001;assert stat.S_IMODE(os.stat('/data/private').st_mode)==0o700;assert stat.S_IMODE(os.stat('/data/private/forward.sqlite').st_mode)==0o600")
|
|
checks.append('private queue and database modes enforced')
|
|
image=json.loads(docker('image','inspect',args.image).stdout)[0]
|
|
receipt={'status':'passed','image_id':image['Id'],'checks':checks,'fixture_only':True,
|
|
'external_network':'none','native_identity_proven':False,'envelope_sha256':before}
|
|
finally:
|
|
docker('rm','-f',name,check=False)
|
|
docker('volume','rm',name,check=False)
|
|
receipt['cleanup_complete']=docker('inspect',name,check=False).returncode!=0 and docker('volume','inspect',name,check=False).returncode!=0
|
|
assert receipt['cleanup_complete']
|
|
args.receipt.write_text(json.dumps(receipt,indent=2)+'\n')
|
|
print(json.dumps({'status':'passed','checks':len(checks),'cleanup_complete':True}))
|
|
|
|
if __name__=='__main__':main()
|