workflow_planning.py (13316B)
1 #!/usr/bin/env python3 2 """Independent plan/trace oracles and fail-closed local authority controls.""" 3 import argparse 4 from dataclasses import asdict 5 import hashlib 6 import json 7 import os 8 from pathlib import Path 9 import resource 10 import subprocess 11 import tempfile 12 from unittest.mock import patch 13 14 from tricu_workflow_demo import Session, WorkspaceExecutor, ProtocolError 15 16 17 def tree(roots, value): 18 if isinstance(value, str): 19 return roots.string(value) 20 result = roots.leaf() 21 for child in reversed(value): 22 result = roots.fork(tree(roots, child), result) 23 return result 24 25 26 def identity(stages): 27 payload = json.dumps(stages, ensure_ascii=False, sort_keys=True, separators=(',', ':')).encode() 28 return hashlib.sha256(b'tricu.workflow-demo.plan.v1\0'+payload).hexdigest() 29 30 31 def main(): 32 parser = argparse.ArgumentParser(description=__doc__) 33 parser.add_argument('--app', type=Path, required=True) 34 parser.add_argument('--report', type=Path, required=True) 35 args = parser.parse_args() 36 checks = [] 37 text = 'built by tricu 🌳\n' 38 expected_stages = [ 39 dict(name='build', action=['write','candidate.txt',text], approval=False), 40 dict(name='test', action=['check','candidate.txt',text], approval=False), 41 dict(name='deploy', action=['publish','candidate.txt','release.txt'], approval=False)] 42 protected_stages = [dict(stage, approval=stage['name']=='deploy') for stage in expected_stages] 43 expected_explanation = ('build: write candidate.txt\ntest: check candidate.txt\n' 44 'approval required: deploy\ndeploy: publish candidate.txt -> release.txt\n') 45 with tempfile.TemporaryDirectory(prefix='workflow-lab-') as directory: 46 root = Path(directory) 47 caller = root/'unrelated-cwd'; caller.mkdir(); caller.chmod(0o555) 48 home = root/'empty-home'; home.mkdir(); home.chmod(0o555) 49 environment = dict(os.environ, HOME=str(home), XDG_CACHE_HOME=str(home/'cache'), PATH='/no-tools') 50 for name in ('TRICU_LIB','ARBORICX_LIB','PYTHONPATH'): 51 environment.pop(name,None) 52 def run(*options, code=0): 53 result = subprocess.run([str(args.app), *options], cwd=caller, env=environment, 54 capture_output=True, text=True, timeout=10) 55 assert result.returncode == code, (options,result.returncode,result.stdout,result.stderr) 56 assert not result.stderr, result.stderr 57 return json.loads(result.stdout) 58 plain = run('explain','--message',text) 59 protected = run('explain','--message',text,'--approval') 60 assert plain['stages'] == expected_stages 61 assert plain['definition_id'] == identity(expected_stages) 62 assert protected['stages'] == protected_stages 63 assert protected['definition_id'] == identity(protected_stages) 64 assert protected['explanation'] == expected_explanation 65 assert protected['definition_id'] != plain['definition_id'] 66 assert not list(caller.iterdir()) and not list(home.iterdir()) 67 checks.append('independent explanation, definition hash, and approval transformation') 68 capabilities = ['--allow','write','--allow','check','--allow','publish'] 69 def execute(name, guarded=True, grants=(), caps=capabilities, code=1): 70 workspace = root/name; workspace.mkdir() 71 options = ['execute','--workspace',str(workspace),'--message',text,*caps] 72 if guarded: options += ['--approval'] 73 for grant in grants: options += ['--approve',grant] 74 return workspace,run(*options,code=code) 75 workspace, denied = execute('no-capabilities',caps=[]) 76 assert denied['events'] == [dict(stage='build',action=expected_stages[0]['action'],status='denied')] 77 assert not list(workspace.iterdir()) 78 workspace, denied = execute('no-approval') 79 assert [e['action'][0] for e in denied['events']] == ['write','check','approve'] 80 assert [e['status'] for e in denied['events']] == ['ok','ok','denied'] 81 assert (workspace/'candidate.txt').read_text() == text and not (workspace/'release.txt').exists() 82 workspace, denied = execute('stale-approval',grants=[plain['definition_id']+':deploy']) 83 assert denied['events'][-1]['status']=='denied' and not (workspace/'release.txt').exists() 84 workspace, denied = execute('wrong-stage',grants=[protected['definition_id']+':test']) 85 assert denied['events'][-1]['status']=='denied' and not (workspace/'release.txt').exists() 86 workspace, denied = execute('no-boundary',guarded=False,grants=[plain['definition_id']+':deploy']) 87 assert denied['events'][-1]['action'][0]=='publish' and denied['events'][-1]['status']=='denied' 88 assert not (workspace/'release.txt').exists() 89 workspace, denied = execute('no-publish-capability',grants=[protected['definition_id']+':deploy'],caps=capabilities[:4]) 90 assert [e['status'] for e in denied['events']] == ['ok','ok','ok','denied'] 91 assert not (workspace/'release.txt').exists() 92 workspace, completed = execute('approved',grants=[protected['definition_id']+':deploy'],code=0) 93 assert completed == dict(outcome='completed',definition_id=protected['definition_id'],events=[ 94 dict(stage='build',action=expected_stages[0]['action'],status='ok'), 95 dict(stage='test',action=expected_stages[1]['action'],status='ok'), 96 dict(stage='deploy',action=['approve'],status='ok'), 97 dict(stage='deploy',action=expected_stages[2]['action'],status='ok')]) 98 assert (workspace/'release.txt').read_text()==text 99 checks.append('capabilities and exact-definition/stage approval are independent; no missing-gate bypass') 100 checks.append('source interpreter stops after refusal and executes approved pipeline in declared order') 101 workspace=root/'changed-payload'; workspace.mkdir() 102 changed=run('execute','--workspace',str(workspace),'--message',text+'changed','--approval', 103 *capabilities,'--approve',protected['definition_id']+':deploy',code=1) 104 assert changed['events'][-1]['status']=='denied' and not (workspace/'release.txt').exists() 105 with Session() as session: 106 with session.context.scope() as roots: 107 plan = session.plan(roots,text) 108 # Neither interpreter calls the built-in native IO driver. 109 with patch.object(session.context,'run_io',side_effect=AssertionError('native IO invoked')): 110 assert session.describe(roots,plan)['definition_id']==plain['definition_id'] 111 once = session.guard(roots,plan,'deploy') 112 twice = session.guard(roots,once,'deploy') 113 assert session.describe(roots,once)==session.describe(roots,twice) 114 changed = session.plan(roots,text+'changed',True) 115 assert session.describe(roots,changed)['definition_id'] != protected['definition_id'] 116 try: session.guard(roots,plan,'missing') 117 except ProtocolError: pass 118 else: raise AssertionError('missing transform target accepted') 119 noio = root/'pure-execution'; noio.mkdir() 120 with patch.object(session.context,'run_io',side_effect=AssertionError('native IO invoked')): 121 with WorkspaceExecutor(noio,('write','check','publish'),[(protected['definition_id'],'deploy')]) as executor: 122 assert session.execute(roots,once,executor)['outcome']=='completed' 123 checks.append('idempotent structural transform; payload-sensitive identity; pure interaction interpreter') 124 valid = ['stage','build',['write','candidate.txt',text]] 125 invalid = [[], [valid,valid], [['stage','bad',['shell','sh','-c']]], 126 [valid,['stage','later',['write','../escape','x']]], 127 [['stage','bad',['write','/absolute','x']]], 128 [['approval','other',valid]], [['approval','build',['approval','build',valid]]], 129 [['stage','bad',['write','candidate.txt','x'*65537]]]] 130 for index, model in enumerate(invalid): 131 workspace=root/f'invalid-{index}'; workspace.mkdir() 132 with session.context.scope() as roots, WorkspaceExecutor(workspace,('write','check','publish')) as executor: 133 try: session.execute(roots,tree(roots,model),executor) 134 except ProtocolError: pass 135 else: raise AssertionError('invalid declaration accepted') 136 assert not list(workspace.iterdir()) 137 # Fabricate a dishonest execution message, independently of the 138 # production tricu interpreter: dispatcher must compare declarations. 139 workspace=root/'mismatched-request'; workspace.mkdir() 140 with session.context.scope() as roots, WorkspaceExecutor(workspace,('write','check','publish')) as executor: 141 plan=session.plan(roots,text,True) 142 dishonest=tree(roots,['request','build',['write','different.txt','oops'],[]]) 143 with patch.object(session,'_start',return_value=dishonest): 144 try: session.execute(roots,plan,executor) 145 except ProtocolError: pass 146 else: raise AssertionError('undeclared effect accepted') 147 assert not list(workspace.iterdir()) 148 with patch.object(session,'_start',return_value=tree(roots,['done'])): 149 try: session.execute(roots,plan,executor) 150 except ProtocolError: pass 151 else: raise AssertionError('premature completion accepted') 152 checks.append('whole-plan validation before effects; independent dishonest-request/completion controls') 153 outside=root/'outside'; outside.write_text('untouched') 154 workspace=root/'symlink'; workspace.mkdir() 155 (workspace/'candidate.txt').symlink_to(outside) 156 with session.context.scope() as roots, WorkspaceExecutor(workspace,('write','check','publish')) as executor: 157 result=session.execute(roots,session.plan(roots,text,True),executor) 158 assert result['outcome']=='failed' and result['events'][0]['status']=='error' 159 assert outside.read_text()=='untouched' 160 assert not (workspace/'release.txt').exists() 161 # Direct source-interpreter check stages exercise no-follow reads of 162 # symlinks and nonregular files without an earlier write masking it. 163 for name,kind in [('link','symlink'),('directory','directory'),('pipe','fifo')]: 164 workspace=root/name; workspace.mkdir() 165 path=workspace/'candidate.txt' 166 if kind=='symlink': path.symlink_to(outside) 167 elif kind=='directory': path.mkdir() 168 else: os.mkfifo(path) 169 with session.context.scope() as roots, WorkspaceExecutor(workspace,('check',)) as executor: 170 plan=tree(roots,[['stage','test',['check','candidate.txt','untouched']]]) 171 result=session.execute(roots,plan,executor) 172 assert result['outcome']=='failed' and result['events'][0]['status']=='error' 173 checks.append('exclusive writes, anchored no-follow reads, nonregular-file rejection') 174 workspace=root/'closed-executor'; workspace.mkdir() 175 executor=WorkspaceExecutor(workspace,('write','check','publish')) 176 executor.close() 177 with session.context.scope() as roots: 178 result=session.execute(roots,session.plan(roots,text),executor) 179 assert result['events'][0]['status']=='denied' and result['reason']=='executor is closed' 180 assert not list(workspace.iterdir()) 181 checks.append('closed executor cannot fall back to process-relative file access') 182 # Reuse the same definitions/context; per-run gates and handles do not 183 # inherit previous approvals or grow the descriptor registry. 184 baseline=len(list(Path('/proc/self/fd').iterdir())) 185 for index in range(32): 186 workspace=root/f'repeated-{index}'; workspace.mkdir() 187 with session.context.scope() as roots, WorkspaceExecutor(workspace,('write','check','publish')) as executor: 188 result=session.execute(roots,session.plan(roots,text,True),executor) 189 assert result['events'][-1]['action']==['approve'] and result['events'][-1]['status']=='denied' 190 assert not (workspace/'release.txt').exists() 191 assert len(list(Path('/proc/self/fd').iterdir()))==baseline 192 checks.append('32 reused-context runs with fresh approvals and stable descriptors') 193 caller.chmod(0o755); home.chmod(0o755) 194 report=dict(checks=checks,plain_definition=plain['definition_id'],guarded_definition=protected['definition_id'], 195 peak_rss_kib=resource.getrusage(resource.RUSAGE_SELF).ru_maxrss) 196 args.report.write_text(json.dumps(report,ensure_ascii=False,indent=2)+'\n') 197 print(json.dumps(report,ensure_ascii=False)) 198 199 200 if __name__=='__main__': main()