key-cape/scripts/test_openbao_negative_private_handoff.py

import base64
import importlib.util
import json
import os
from pathlib import Path
import signal
import tempfile
import time
import unittest
from unittest.mock import patch
REAL_WRITE=os.write
spec = importlib.util.spec_from_file_location('handoff',Path(__file__).with_name('openbao_negative_private_handoff.py'))
s = importlib.util.module_from_spec(spec);spec.loader.exec_module(s)

def jwt(kind='human', **changes):
    obj={'iss':'https://kc.coulomb.social','iat':time.time()-1,'exp':time.time()+700,
         'principal_type':'human','aud':'netkingdom-bootstrap-console','sub':'fixture-person',
         'assurance':{'level':'aal2','mfa':True}}
    if kind == 'agent':
        obj.update(principal_type='service',aud='codex-railiance-platform',
                   sub='service:codex:railiance-platform',tenant='tenant:platform',
                   roles=['coding-agent'],scope='openbao:login')
    obj.update(changes)
    body=base64.urlsafe_b64encode(json.dumps(obj).encode()).decode().rstrip('=')
    return 'fixture.'+body+'.fixture'

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'
        self.write(self.h,jwt());self.write(self.a,jwt('agent'))
    def tearDown(self):
        self.tmp.cleanup()
    def write(self,p,token):
        p.write_text(json.dumps({'access_token':token,'token_type':'Bearer','expires_in':700}));p.chmod(0o600)
    def run_delivery(self,**kw):
        r,w=os.pipe();cr,cw=os.pipe()
        expected=self.expected()
        REAL_WRITE(cw,json.dumps(dict(expected,schema=s.READY_SCHEMA,receive_budget_seconds=10)).encode()+b'\n')
        os.close(cw)
        receipt=s.deliver(kw.pop("human_path",str(self.h)),str(self.a),w,ready_fd=cr,expected=expected,consume_owner_files=True,**kw)
        raw=os.read(r,65536);os.close(r)
        return receipt,raw
    def expected(self):
        return {'session_nonce':'1'*64,'proposal_id':'KEY-WP-0039-T02/attended-lifecycle-v9-ready-input',
                'target':'sso/keycape','executor_sha256':'2'*64,'envelope_sha256':'3'*64,
                'bao_health_identity_sha256':'4'*64,'valid_until_epoch':time.time()+600}
    def test_success_only_private_pipe_and_consumes_owned_files(self):
        receipt,raw=self.run_delivery()
        obj=json.loads(raw)
        self.assertEqual(set(obj),{'human_token','coding_agent_token'})
        self.assertTrue(receipt['private_input_delivery_confirmed'])
        self.assertFalse(receipt['native_accepted'])
        self.assertNotIn('fixture',json.dumps(receipt))
        self.assertFalse(self.h.exists());self.assertFalse(self.a.exists())
    def test_stdout_stderr_and_regular_fd_refused_before_file_consumption(self):
        for fd in (1,2):
            receipt=s.deliver(str(self.h),str(self.a),fd,consume_owner_files=True)
            self.assertEqual(receipt['failure_code'],'private_pipe_fd_required')
        with tempfile.TemporaryFile() as f:
            dup=os.dup(f.fileno())
            receipt=s.deliver(str(self.h),str(self.a),dup,consume_owner_files=True)
            self.assertEqual(receipt['failure_code'],'private_pipe_fd_required')
            self.assertEqual(f.tell(),0)
        self.assertTrue(self.h.exists());self.assertTrue(self.a.exists())
    def test_read_only_pipe_refused_before_consuming_files(self):
        r,w=os.pipe()
        receipt=s.deliver(str(self.h),str(self.a),r,consume_owner_files=True)
        os.close(w)
        self.assertEqual(receipt['failure_code'],'private_pipe_fd_required')
        self.assertTrue(self.h.exists());self.assertTrue(self.a.exists())
    def test_git_worktree_refused_before_delivery(self):
        marker=self.dir/'.git';marker.write_text('gitdir: fixture')
        receipt,raw=self.run_delivery()
        self.assertEqual(receipt['failure_code'],'git_path_refused');self.assertFalse(raw)
    def test_symlinked_parent_refused(self):
        alias=self.dir/'alias';alias.symlink_to(self.dir,target_is_directory=True)
        receipt,raw=self.run_delivery(human_path=str(alias/'human.json'))
        self.assertEqual(receipt['failure_code'],'private_path_required')
        self.assertFalse(raw);self.assertTrue(self.h.exists())

    def test_explicit_consumption_required(self):
        r,w=os.pipe();receipt=s.deliver(str(self.h),str(self.a),w)
        self.assertEqual(receipt['failure_code'],'explicit_owner_file_consumption_required')
        self.assertEqual(os.read(r,10),b'');os.close(r);self.assertTrue(self.h.exists())
    def test_symlink_refused_no_delivery(self):
        target=self.dir/'target';self.h.rename(target);self.h.symlink_to(target)
        receipt,raw=self.run_delivery();self.assertFalse(raw)
        self.assertFalse(receipt['private_input_delivery_confirmed']);self.assertTrue(target.exists())
    def test_hardlinked_file_refused(self):
        os.link(self.h,self.dir/'extra');receipt,raw=self.run_delivery()
        self.assertEqual(receipt['failure_code'],'private_file_required');self.assertFalse(raw)
    def test_public_mode_refused(self):
        self.a.chmod(0o644);receipt,raw=self.run_delivery()
        self.assertEqual(receipt['failure_code'],'private_file_required');self.assertFalse(raw)
    def test_public_directory_refused(self):
        self.dir.chmod(0o755);receipt,raw=self.run_delivery()
        self.assertEqual(receipt['failure_code'],'private_directory_required');self.assertFalse(raw)
    def test_oversize_file_refused_before_delivery(self):
        self.a.write_bytes(b'x'*(s.MAX_FILE+1));receipt,raw=self.run_delivery()
        self.assertEqual(receipt['failure_code'],'private_file_required');self.assertFalse(raw)
    def test_wrong_human_audience_or_missing_mfa_refused(self):
        for changes in ({'aud':'openbao-admin'},{'assurance':{'level':'aal1','mfa':False}}):
            self.write(self.h,jwt(**changes));receipt,raw=self.run_delivery()
            self.assertFalse(raw);self.assertFalse(receipt['private_input_delivery_confirmed'])
    def test_approle_string_or_wrong_coding_claim_refused(self):
        for token in ('hvs.not-a-jwt',jwt('agent',tenant='tenant:coulomb'),jwt('agent',roles=['secrets-engine'])):
            self.write(self.a,token);receipt,raw=self.run_delivery()
            self.assertFalse(raw);self.assertFalse(receipt['private_input_delivery_confirmed'])
    def test_expired_bool_or_infinite_timestamp_refused(self):
        for expiry in (time.time()-1,True,float('inf')):
            self.write(self.a,jwt('agent',exp=expiry));receipt,raw=self.run_delivery()
            self.assertFalse(raw);self.assertFalse(receipt['private_input_delivery_confirmed'])
    def test_duplicate_json_refused(self):
        self.a.write_text('{"token_type":"Bearer","access_token":"a","access_token":"b"}')
        receipt,raw=self.run_delivery();self.assertEqual(receipt['failure_code'],'duplicate_json_key');self.assertFalse(raw)
    def test_file_replacement_before_consumption_not_unlinked(self):
        original=s.consume
        def replace(item):
            self.h.unlink();self.write(self.h,jwt())
            return original(item)
        with patch.object(s,'consume',side_effect=replace):receipt,raw=self.run_delivery()
        self.assertEqual(receipt['failure_code'],'private_file_changed');self.assertFalse(raw);self.assertTrue(self.h.exists())
    def test_blocked_partial_write_is_bounded_unknown_and_timer_restored(self):
        previous=signal.getsignal(signal.SIGALRM);started=time.monotonic()
        with patch.object(s.os,'write',side_effect=lambda *args:time.sleep(0.1)):
            receipt,raw=self.run_delivery(budget=0.01)
        self.assertLess(time.monotonic()-started,0.06)
        self.assertEqual(receipt['failure_code'],'delivery_timeout')
        self.assertTrue(receipt['private_input_delivery_unknown']);self.assertFalse(raw)
        self.assertFalse(self.h.exists());self.assertFalse(self.a.exists())
        self.assertEqual(signal.getitimer(signal.ITIMER_REAL),(0.0,0.0));self.assertIs(signal.getsignal(signal.SIGALRM),previous)
    def test_broken_pipe_fixed_error_no_raw_exception(self):
        with patch.object(s.os,'write',side_effect=RuntimeError('protected raw value')):
            receipt,raw=self.run_delivery()
        self.assertEqual(receipt['failure_code'],'private_delivery_failed')
        self.assertTrue(receipt['private_input_delivery_unknown']);self.assertNotIn('protected',json.dumps(receipt));self.assertFalse(raw)

if __name__ == '__main__':unittest.main()

class ReadyTests(unittest.TestCase):
    setUp=Tests.setUp
    tearDown=Tests.tearDown
    write=Tests.write
    expected=Tests.expected
    run_delivery=Tests.run_delivery
    # Concrete ready-channel counterexamples; base safety cases run only once.
    def with_marker(self, marker, budget=0.05):
        r,w=os.pipe();cr,cw=os.pipe();expected=self.expected()
        raw=marker(expected) if callable(marker) else marker
        REAL_WRITE(cw,raw);os.close(cw)
        with patch.object(s,'load',wraps=s.load) as load:
            receipt=s.deliver(str(self.h),str(self.a),w,ready_fd=cr,expected=expected,
                              consume_owner_files=True,budget=budget)
            calls=load.call_count
        data=os.read(r,65536);os.close(r)
        return receipt,data,calls
    def test_no_child_ready_or_mismatch_means_zero_file_load_and_private_bytes(self):
        for marker in (b'', b'not-json\n', lambda e:json.dumps(dict(e,schema=s.READY_SCHEMA,
                receive_budget_seconds=10,session_nonce='0'*64)).encode()+b'\n',
                lambda e:json.dumps(dict(e,schema=s.READY_SCHEMA,receive_budget_seconds=True)).encode()+b'\n'):
            receipt,data,calls=self.with_marker(marker)
            self.assertFalse(receipt['private_input_delivery_confirmed']);self.assertFalse(data)
            self.assertEqual(calls,0);self.assertTrue(self.h.exists());self.assertTrue(self.a.exists())
    def test_no_ready_eof_hard_timeout_zero_file_load_or_consumption(self):
        r,w=os.pipe();cr,cw=os.pipe();started=time.monotonic()
        try:
            with patch.object(s,'load') as load:
                receipt=s.deliver(str(self.h),str(self.a),w,ready_fd=cr,expected=self.expected(),
                                  consume_owner_files=True,budget=0.01)
                load.assert_not_called()
            self.assertLess(time.monotonic()-started,0.06)
            self.assertEqual(receipt['failure_code'],'delivery_timeout');self.assertFalse(os.read(r,10))
            self.assertTrue(self.h.exists());self.assertTrue(self.a.exists())
        finally:os.close(cw);os.close(r)
    def test_stdout_and_stderr_pipe_duplicates_refused(self):
        for standard in (1,2):
            r,w=os.pipe();backup=os.dup(standard)
            try:
                os.dup2(w,standard);duplicate=os.dup(standard)
                receipt=s.deliver(str(self.h),str(self.a),duplicate,consume_owner_files=True)
            finally:os.dup2(backup,standard);os.close(backup);os.close(w)
            self.assertEqual(receipt['failure_code'],'stdio_pipe_alias_refused')
            self.assertEqual(os.read(r,10),b'');os.close(r)
            self.assertTrue(self.h.exists());self.assertTrue(self.a.exists())
    def test_data_and_control_same_pipe_refused(self):
        r,w=os.pipe();readback=os.dup(r)
        receipt=s.deliver(str(self.h),str(self.a),w,ready_fd=r,expected=self.expected(),consume_owner_files=True)
        self.assertEqual(receipt['failure_code'],'data_control_pipe_alias_refused')
        self.assertEqual(os.read(readback,10),b'');os.close(readback)
    def test_oversize_ready_refused_before_load(self):
        receipt,data,calls=self.with_marker(b'x'*4096+b'\n')
        self.assertEqual(receipt['failure_code'],'child_ready_oversized');self.assertFalse(data);self.assertEqual(calls,0)
    def test_received_short_budget_clips_timer_before_token_load(self):
        marker=lambda e:json.dumps(dict(e,schema=s.READY_SCHEMA,receive_budget_seconds=0.01)).encode()+b'\n'
        original=s.load
        def delayed(*args):time.sleep(0.1);return original(*args)
        started=time.monotonic()
        with patch.object(s,'load',side_effect=delayed):receipt,data,calls=self.with_marker(marker)
        self.assertLess(time.monotonic()-started,0.06);self.assertFalse(data)
        self.assertEqual(receipt['failure_code'],'delivery_timeout');self.assertTrue(self.h.exists())
    def test_real_partial_write_stays_unknown_and_conservative_cleanup(self):
        import fcntl
        self.write(self.h,jwt(padding='h'*7000));self.write(self.a,jwt('agent',padding='a'*7000))
        r,w=os.pipe();fcntl.fcntl(w,fcntl.F_SETPIPE_SZ,4096)
        cr,cw=os.pipe();e=self.expected();REAL_WRITE(cw,json.dumps(dict(e,schema=s.READY_SCHEMA,
            receive_budget_seconds=10)).encode()+b'\n');os.close(cw)
        receipt=s.deliver(str(self.h),str(self.a),w,ready_fd=cr,expected=e,consume_owner_files=True,budget=0.015)
        data=os.read(r,65536);os.close(r)
        self.assertEqual(len(data),4096);self.assertTrue(receipt['private_input_delivery_unknown'])
        self.assertFalse(receipt['private_input_delivery_confirmed'])
        self.assertTrue(receipt['credential_files_remain_possible'])
        self.assertFalse(receipt['credential_copy_cleanup_verified'])
    def test_stat_unlink_race_retained_as_exclusive_custody_violation_not_atomic_claim(self):
        original=s.os.stat;replaced=[]
        def race(path,*args,**kwargs):
            current=original(path,*args,**kwargs)
            if path=='human.json' and kwargs.get('dir_fd') is not None and not replaced:
                self.h.unlink();self.write(self.h,jwt(sub='replacement-fixture'))
                replaced.append(True)
            return current
        with patch.object(s.os,'stat',side_effect=race):receipt,data=self.run_delivery()
        self.assertTrue(replaced);self.assertFalse(self.h.exists())
        self.assertTrue(receipt['caller_path_removal_completed'])
        self.assertFalse(receipt['atomic_inode_delete_guaranteed'])
        self.assertTrue(receipt['exclusive_same_uid_custody_required'])
        self.assertTrue(receipt['credential_files_remain_possible'])
        self.assertFalse(receipt['credential_copy_cleanup_verified'])