net-kingdom/sso-mfa/k8s/keycape/test_openbao_private_input.py

"""Actual main bounded private receive; synthetic local binary pipes only."""
import importlib.util
from pathlib import Path
import io
import json
import os
import signal
import sys
import tempfile
import time
import unittest
from unittest.mock import patch

spec=importlib.util.spec_from_file_location('routes',Path(__file__).with_name('test_openbao_node_route.py'))
r=importlib.util.module_from_spec(spec);spec.loader.exec_module(r);s=r.s;f=r.f
PAYLOAD=b'{"human_token":"synthetic-human","coding_agent_token":"synthetic-agent"}'

class PrivateInputTests(unittest.TestCase):
    def setUp(self):
        s.CLEANING=False;s.REQUEST_DEADLINE=None
        s.BUSINESS_MONOTONIC_DEADLINE=None;s.CLEANUP_MONOTONIC_DEADLINE=None

    def main_with(self,stream,*,lifetime=0.05,health=None):
        previous=signal.getsignal(signal.SIGTERM)
        try:
            with tempfile.TemporaryDirectory() as folder:
                envelope=f.envelope();envelope['valid_until_epoch']=time.time()+lifetime
                envelope['bao_health_identity_sha256']=r.PIN
                path=Path(folder)/'approved.json';path.write_text(json.dumps(envelope));path.chmod(0o600)
                receipt=Path(folder)/'receipt.json'
                argv=['session','execute','--envelope',str(path),'--receipt',str(receipt)]
                with patch.object(sys,'argv',argv),patch.object(sys,'stdin',stream), \
                     patch.object(s,'SessionTransport') as factory,patch.object(s,'emit_private_ready'),patch.object(s,'execute') as execute, \
                     patch.object(s,'k') as k,patch.object(s,'get') as get:
                    factory.return_value.health_identity.side_effect=health
                    factory.return_value.health_identity.return_value=r.PIN
                    rc=s.main()
                    k.assert_not_called();get.assert_not_called()
                    return rc,json.loads(receipt.read_text()),execute.call_args,execute.call_count
        finally:signal.signal(signal.SIGTERM,previous)

    def test_actual_main_real_pipe_stall_has_original_window_total_bound(self):
        read_fd,write_fd=os.pipe()
        binary=os.fdopen(read_fd,'rb',buffering=0)
        stream=type('Input',(),{'buffer':binary})()
        started=time.monotonic();previous=signal.getsignal(signal.SIGALRM)
        try:
            rc,receipt,args,count=self.main_with(stream,lifetime=0.02)
        finally:os.close(write_fd);binary.close()
        self.assertEqual(rc,1);self.assertEqual(count,0)
        self.assertEqual(receipt['reason'],'request_total_timeout')
        self.assertLess(time.monotonic()-started,0.15)
        self.assertEqual(signal.getitimer(signal.ITIMER_REAL),(0.0,0.0))
        self.assertEqual(signal.getsignal(signal.SIGALRM),previous)

    def test_actual_main_complete_json_without_eof_still_refuses(self):
        read_fd,write_fd=os.pipe();os.write(write_fd,PAYLOAD)
        binary=os.fdopen(read_fd,'rb',buffering=0);stream=type('Input',(),{'buffer':binary})()
        try:rc,receipt,args,count=self.main_with(stream,lifetime=0.02)
        finally:os.close(write_fd);binary.close()
        self.assertEqual(rc,1);self.assertEqual(count,0)
        self.assertEqual(receipt['reason'],'request_total_timeout')

    def test_actual_main_oversized_receive_refuses_without_execution(self):
        stream=type('Input',(),{'buffer':io.BytesIO(b'x'*(s.PRIVATE_INPUT_MAX_BYTES+1))})()
        rc,receipt,args,count=self.main_with(stream)
        self.assertEqual(rc,1);self.assertEqual(count,0)
        self.assertEqual(receipt['reason'],'private_input_oversized')

    def test_actual_main_empty_partial_invalid_or_excessive_token_refuses(self):
        inputs=[(b'','private_input_empty'),(b'{"human_token":','private_input_invalid'),
                (b'not-json','private_input_invalid'),(b'{"human_token":true,"coding_agent_token":"a"}','private_input_invalid'),
                (json.dumps({'human_token':'x'*(s.PRIVATE_TOKEN_MAX_BYTES+1),'coding_agent_token':'a'}).encode(),'private_input_invalid')]
        for raw,reason in inputs:
            with self.subTest(reason=reason,length=len(raw)):
                stream=type('Input',(),{'buffer':io.BytesIO(raw)})()
                rc,receipt,args,count=self.main_with(stream)
                self.assertEqual(rc,1);self.assertEqual(count,0);self.assertEqual(receipt['reason'],reason)

    def test_actual_main_expiry_after_health_never_starts_private_receive(self):
        stream=type('Input',(),{'buffer':io.BytesIO(PAYLOAD)})()
        def late_health():
            s.REQUEST_DEADLINE=time.time()-1
            return r.PIN
        with patch.object(stream.buffer,'read',wraps=stream.buffer.read) as read:
            rc,receipt,args,count=self.main_with(stream,health=late_health)
        self.assertEqual(rc,1);self.assertEqual(count,0)
        self.assertEqual(receipt['reason'],'approved_window_expired');read.assert_not_called()

    def test_actual_main_late_eof_never_decodes_or_executes(self):
        class LateInput:
            reads=0
            def read(self,size):
                self.reads+=1
                if self.reads==1:return PAYLOAD
                s.BUSINESS_MONOTONIC_DEADLINE=time.monotonic()-1
                return b''
        stream=type('Input',(),{'buffer':LateInput()})()
        rc,receipt,args,count=self.main_with(stream)
        self.assertEqual(rc,1);self.assertEqual(count,0)
        self.assertEqual(receipt['reason'],'session_business_budget_expired')

    def test_actual_main_complete_bounded_json_plus_eof_reaches_execution_once(self):
        stream=type('Input',(),{'buffer':io.BytesIO(PAYLOAD)})()
        rc,receipt,args,count=self.main_with(stream)
        self.assertEqual(rc,0);self.assertEqual(count,1)
        self.assertEqual(args.args[1],json.loads(PAYLOAD))
        self.assertTrue(args.kwargs['window_started'])
        self.assertNotIn('synthetic-human',json.dumps(receipt))

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