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'])