key-cape/scripts/test_openbao_negative_private_launcher.py

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