import importlib.util import json import os from pathlib import Path import shlex import tempfile import time import unittest from unittest.mock import patch spec=importlib.util.spec_from_file_location('launcher',Path(__file__).with_name('openbao_negative_private_launcher.py')) l=importlib.util.module_from_spec(spec);spec.loader.exec_module(l) spec=importlib.util.spec_from_file_location('fixtures',Path(__file__).with_name('test_openbao_negative_private_handoff.py')) f=importlib.util.module_from_spec(spec);spec.loader.exec_module(f) class FakeProcess: def __init__(self,argv,contract,envelope_hash,envelope,marker='valid'): self.argv=argv;self.rc=None;self.terminated=False;self.waits=[] self.read_data,w=os.pipe();self.stdin=os.fdopen(w,'wb',buffering=0) r,cw=os.pipe();self.stdout=os.fdopen(r,'rb',buffering=0) args=shlex.split(argv[-1]);nonce=args[args.index('--session-nonce')+1] ready={'schema':'key39-private-input-ready/v1','session_nonce':nonce, 'proposal_id':l.PROPOSAL,'target':l.TARGET,'executor_sha256':contract['remote_child_sha256'], 'envelope_sha256':envelope_hash,'bao_health_identity_sha256':envelope['bao_health_identity_sha256'], 'valid_until_epoch':envelope['valid_until_epoch'],'receive_budget_seconds':15} if marker=='wrong':ready['session_nonce']='0'*64 if marker!='missing':os.write(cw,json.dumps(ready).encode()+b'\n') os.close(cw) def wait(self,timeout): self.waits.append(timeout);self.rc=-15 if self.terminated else 0;return self.rc def poll(self):return self.rc def terminate(self):self.terminated=True def kill(self):self.terminated=True class Tests(unittest.TestCase): def setUp(self): self.tmp=tempfile.TemporaryDirectory();self.dir=Path(self.tmp.name) self.h=self.dir/'human.json';self.a=self.dir/'agent.json' for p,token in ((self.h,f.jwt()),(self.a,f.jwt('agent'))): p.write_text(json.dumps({'access_token':token,'token_type':'Bearer'}));p.chmod(0o600) self.envelope={'proposal_id':l.PROPOSAL,'target':l.TARGET,'disposition':'approved', 'founder_decision_id':'fixture-founder','valid_until_epoch':time.time()+600, 'maximum_session_seconds':900,'bao_health_identity_sha256':'4'*64, 'source_hashes':{'/run/key39/child.py':'2'*64,'/run/key39/wrapper.py':'3'*64}, 'owner_acknowledgments':{k:'fixture-source' for k in ('key-cape','net-kingdom','railiance-platform')}} self.e=self.dir/'envelope.json';self.e.write_text(json.dumps(self.envelope)) self.contract={'schema':'key39-private-launch-contract/v1','proposal_id':l.PROPOSAL, 'target':l.TARGET,'remote_host':'railiance01','remote_wrapper_path':'/run/key39/wrapper.py', 'remote_wrapper_sha256':'3'*64,'remote_child_path':'/run/key39/child.py', 'remote_child_sha256':'2'*64,'remote_envelope_path':'/run/key39/approved.json', 'local_envelope_path':str(self.e),'remote_receipt_path':'/run/key39/receipt.json', 'founder_decision_id':'fixture-founder','private_tmpfs_file_allowance_receipt':'fixture-file-allowance', 'local_launcher_sha256':l.digest(l.__file__),'local_adapter_sha256':l.digest(l.ADAPTER)} self.processes=[] def tearDown(self): for p in self.processes: os.close(p.read_data) self.tmp.cleanup() def run_launcher(self,marker='valid'): def popen(argv,**kwargs): _,envelope=l.read_public(self.e) self.assertEqual(kwargs['stdin'],l.subprocess.PIPE) self.assertEqual(kwargs['stdout'],l.subprocess.PIPE) self.assertEqual(kwargs['stderr'],l.subprocess.DEVNULL) self.assertNotIn('KEYCAPE_CODEX_RAILIANCE_PLATFORM_CLIENT_SECRET',kwargs['env']) self.assertIn('-oStrictHostKeyChecking=yes',argv);self.assertIn('-T',argv) p=FakeProcess(argv,self.contract,l.digest(self.e),envelope,marker);self.processes.append(p);return p with patch.object(l.subprocess,'Popen',side_effect=popen) as spawn, \ patch.dict(os.environ,{'KEYCAPE_CODEX_RAILIANCE_PLATFORM_CLIENT_SECRET':'PRIVATE-FIXTURE'}): receipt=l.launch_and_deliver(self.contract,str(self.h),str(self.a),consume_owner_files=True) return receipt,spawn.call_count def test_complete_pinned_child_marker_then_private_pipe_delivery_no_actual_ssh(self): receipt,count=self.run_launcher();self.assertEqual(count,1) self.assertTrue(receipt['child_ready_matched']);self.assertTrue(receipt['consumer_process_completed']) self.assertFalse(receipt['native_accepted']);self.assertFalse(receipt['authenticated_authority_established_by_helper']) raw=os.read(self.processes[0].read_data,65536);self.assertEqual(set(json.loads(raw)),{'human_token','coding_agent_token'}) self.assertNotIn('fixture-person',json.dumps(receipt));self.assertFalse(self.h.exists());self.assertFalse(self.a.exists()) def test_missing_or_wrong_marker_retains_files_and_writes_zero_tokens(self): for marker in ('missing','wrong'): receipt,count=self.run_launcher(marker) self.assertEqual(count,1);self.assertFalse(receipt['child_ready_matched']) self.assertEqual(os.read(self.processes[-1].read_data,65536),b'') self.assertTrue(self.h.exists());self.assertTrue(self.a.exists()) def test_changed_local_source_pin_refuses_before_transport_and_file_read(self): self.contract['local_adapter_sha256']='0'*64 receipt,count=self.run_launcher();self.assertEqual(count,0) self.assertEqual(receipt['failure_code'],'local_source_pin_changed');self.assertTrue(self.h.exists()) def test_wrong_host_or_remote_command_path_refuses_before_transport(self): for field,value in (('remote_host','other-node'),('remote_wrapper_path','/tmp/x;cat'), ('remote_child_path','/run/key39/../child.py')): old=self.contract[field];self.contract[field]=value receipt,count=self.run_launcher();self.contract[field]=old self.assertEqual(count,0);self.assertFalse(receipt['private_input_delivery_confirmed']);self.assertTrue(self.h.exists()) def test_expired_or_wrong_founder_source_contract_no_spawn_no_file_access(self): for key,value in (('valid_until_epoch',time.time()-1),('founder_decision_id','other')): current=json.loads(self.e.read_text());current[key]=value;self.e.write_text(json.dumps(current)) receipt,count=self.run_launcher();self.assertEqual(count,0);self.assertTrue(self.h.exists()) self.e.write_text(json.dumps(self.envelope)) def test_full_pipe_payload_then_confirmation_timeout_keeps_remote_unknown(self): original_deliver=l.handoff.deliver original_write=os.write def interrupted_delivery(*args,**kwargs): kwargs['budget']=0.01 def write_then_stall(fd,payload): written=original_write(fd,payload) time.sleep(0.1) return written with patch.object(l.handoff.os,'write',side_effect=write_then_stall): return original_deliver(*args,**kwargs) with patch.object(l.handoff,'deliver',side_effect=interrupted_delivery): receipt,count=self.run_launcher() self.assertEqual(count,1) data=os.read(self.processes[0].read_data,65536) self.assertEqual(set(json.loads(data)),{'human_token','coding_agent_token'}) self.assertTrue(receipt['private_input_delivery_unknown']) self.assertFalse(receipt['private_input_delivery_confirmed']) self.assertTrue(receipt['consumer_outcome_unknown']) self.assertFalse(receipt['consumer_process_completed']) self.assertTrue(self.processes[0].terminated) self.assertFalse(self.h.exists());self.assertFalse(self.a.exists()) self.assertFalse(receipt['native_accepted']) def test_real_partial_pipe_write_unknown_is_sticky_after_local_terminate(self): import fcntl for p,token in ((self.h,f.jwt(padding='h'*7000)),(self.a,f.jwt('agent',padding='a'*7000))): p.write_text(json.dumps({'access_token':token,'token_type':'Bearer'}));p.chmod(0o600) original_process=FakeProcess original_deliver=l.handoff.deliver def small_pipe(*args,**kwargs): proc=original_process(*args,**kwargs) fcntl.fcntl(proc.stdin.fileno(),fcntl.F_SETPIPE_SZ,4096) return proc def bounded(*args,**kwargs): kwargs['budget']=0.015 return original_deliver(*args,**kwargs) with patch.dict(globals(),{'FakeProcess':small_pipe}),patch.object(l.handoff,'deliver',side_effect=bounded): receipt,count=self.run_launcher() data=os.read(self.processes[0].read_data,65536) self.assertEqual(len(data),4096) self.assertTrue(receipt['private_input_delivery_unknown']) self.assertTrue(receipt['consumer_outcome_unknown']) self.assertFalse(receipt['consumer_process_completed']) self.assertTrue(self.processes[0].terminated) self.assertFalse(receipt['native_accepted']) if __name__=='__main__':unittest.main()