informed-decision/tools/smoke_container.py

"""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()