key-cape/scripts/test_openbao_negative_private_preparation.py

import importlib.util
import io
import json
import os
from pathlib import Path
import signal
import subprocess
import sys
import tempfile
import time
import unittest
from unittest.mock import patch
spec=importlib.util.spec_from_file_location('prep_launcher',Path(__file__).with_name('openbao_negative_private_launcher.py'))
l=importlib.util.module_from_spec(spec);spec.loader.exec_module(l)

class Tests(unittest.TestCase):
    def actual_cli(self,path):
        started=time.monotonic()
        result=subprocess.run([sys.executable,l.__file__,'--contract',str(path),
            '--human-file','/NEVER-READ-HUMAN','--agent-file','/NEVER-READ-AGENT'],
            capture_output=True,timeout=0.4)
        self.assertEqual(result.returncode,1);self.assertEqual(result.stderr,b'')
        self.assertLess(time.monotonic()-started,0.25)
        value=json.loads(result.stdout);self.assertFalse(value['native_accepted'])
        self.assertFalse(value['private_input_delivery_confirmed'])
        return value
    def test_actual_main_fifo_refused_without_waiting_for_writer(self):
        with tempfile.TemporaryDirectory() as folder:
            path=Path(folder)/'contract.fifo';os.mkfifo(path,0o600)
            self.assertEqual(self.actual_cli(path)['failure_code'],'source_file_invalid')
    def test_actual_main_symlink_contract_refused(self):
        with tempfile.TemporaryDirectory() as folder:
            original=Path(folder)/'contract.json';original.write_text('{}')
            path=Path(folder)/'alias';path.symlink_to(original)
            self.assertEqual(self.actual_cli(path)['failure_code'],'source_regular_path_required')
    def test_actual_main_oversize_contract_refused(self):
        with tempfile.TemporaryDirectory() as folder:
            path=Path(folder)/'contract.json';path.write_bytes(b'x'*(l.MAX_PUBLIC_BYTES+1))
            self.assertEqual(self.actual_cli(path)['failure_code'],'source_file_invalid')
    def main_with(self,path,**patches):
        out=io.StringIO()
        argv=['launcher','--contract',str(path),'--human-file','/NEVER-READ-HUMAN',
              '--agent-file','/NEVER-READ-AGENT']
        with patch.object(sys,'argv',argv),patch.object(sys,'stdout',out), \
             patch.object(l.subprocess,'Popen') as spawn,patch.object(l.handoff,'load') as load:
            rc=l.main();spawn.assert_not_called();load.assert_not_called()
        self.assertEqual(rc,1)
        return json.loads(out.getvalue())
    def test_actual_main_slow_regular_read_has_hard_total_bound_and_signal_restore(self):
        with tempfile.TemporaryDirectory() as folder:
            path=Path(folder)/'contract.json';path.write_text('{}')
            original=l.os.read
            def slow(fd,size):time.sleep(0.1);return original(fd,size)
            previous=signal.getsignal(signal.SIGALRM);started=time.monotonic()
            with patch.object(l,'PREPARATION_TOTAL_SECONDS',0.01),patch.object(l.os,'read',side_effect=slow):
                receipt=self.main_with(path)
            self.assertEqual(receipt['failure_code'],'delivery_timeout')
            self.assertLess(time.monotonic()-started,0.06)
            self.assertEqual(signal.getitimer(signal.ITIMER_REAL),(0.0,0.0));self.assertIs(signal.getsignal(signal.SIGALRM),previous)
    def test_actual_main_launch_does_not_renew_budget_spent_acquiring_contract(self):
        with tempfile.TemporaryDirectory() as folder:
            path=Path(folder)/'contract.json';path.write_text('{}')
            original=l.read_public
            def acquire(path):time.sleep(0.02);return original(path)
            def slow_validate(contract):time.sleep(0.02);return '0'*64,{}
            started=time.monotonic()
            with patch.object(l,'PREPARATION_TOTAL_SECONDS',0.03), \
                 patch.object(l,'read_public',side_effect=acquire), \
                 patch.object(l,'validate_contract',side_effect=slow_validate):
                receipt=self.main_with(path)
            self.assertEqual(receipt['failure_code'],'delivery_timeout');self.assertLess(time.monotonic()-started,0.065)
    def test_observed_source_change_refused_and_reads_are_bounded(self):
        with tempfile.TemporaryDirectory() as folder:
            path=Path(folder)/'contract.json';path.write_text('{}')
            original=l.os.read;changed=[];sizes=[]
            def mutate(fd,size):
                sizes.append(size)
                if not changed:path.write_text('{"x":1}');changed.append(True)
                return original(fd,size)
            with patch.object(l.os,'read',side_effect=mutate):
                with self.assertRaisesRegex(l.handoff.Refused,'source_file_changed'):l.read_public(path)
            self.assertTrue(all(0<size<=4096 for size in sizes))
    def test_source_digest_fifo_and_symlink_refuse(self):
        with tempfile.TemporaryDirectory() as folder:
            path=Path(folder)/'source.fifo';os.mkfifo(path,0o600)
            with self.assertRaisesRegex(l.handoff.Refused,'source_file_invalid'):l.digest(path)
            regular=Path(folder)/'source.py';regular.write_text('fixture')
            alias=Path(folder)/'alias';alias.symlink_to(regular)
            with self.assertRaisesRegex(l.handoff.Refused,'source_regular_path_required'):l.digest(alias)

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