railiance-platform/tests/test_reuse_surface_reader_custody_apply.py

import importlib.util
from pathlib import Path
import unittest
import json
import yaml
import hashlib
import tempfile
from unittest.mock import patch

spec=importlib.util.spec_from_file_location('custody_apply',Path(__file__).parents[1]/'scripts/apply-reuse-surface-reader-custody.py');r=importlib.util.module_from_spec(spec);spec.loader.exec_module(r)
class Journal:
    def __init__(self): self.rows=[]
    def put(self,event,**kw): self.rows.append((event,kw))
class Tests(unittest.TestCase):
    def role(self): return {'bound_service_account_names':['reuse-surface-eso'],'bound_service_account_namespaces':['reuse'],'audience':'openbao','token_ttl':900,'token_max_ttl':900,'token_explicit_max_ttl':900,'token_no_default_policy':True,'token_policies':[r.POLICY]}
    def envelope_fixture(self,root,extra=False):
        old={'id':r.p.CCR,'status':'proposed','openbao':{'auth':{'bound_claims_confirmed':False}}}
        new={'id':r.p.CCR,'status':'approved','openbao':{'auth':{'bound_claims_confirmed':True}}}
        if extra: new['unexpected_privilege']='broader'
        before=yaml.safe_dump(old).encode();after=yaml.safe_dump(new).encode()
        files={r.SCRIPT,r.POLICY_FILE,r.CANDIDATE_FILE,r.p.CCR_FILE,r.p.PLAN_FILE,r.p.SCRIPT_FILE}
        for rel in files:
            f=root/rel;f.parent.mkdir(parents=True,exist_ok=True);f.write_text('mock')
        (root/r.p.CCR_FILE).write_bytes(before)
        wrapper={'source_yaml':before.decode(),'candidate_yaml':after.decode(),'source_sha256':hashlib.sha256(before).hexdigest(),'candidate_sha256':hashlib.sha256(after).hexdigest()}
        (root/r.CANDIDATE_FILE).write_text(json.dumps(wrapper))
        record={'ccr_id':r.p.CCR,'decision':'approved','supervisor':'founder','decision_ref':'mock','runner_artifacts':{rel:hashlib.sha256((root/rel).read_bytes()).hexdigest() for rel in (r.SCRIPT,r.POLICY_FILE,r.CANDIDATE_FILE)},'approved_artifacts':{rel:hashlib.sha256((root/rel).read_bytes()).hexdigest() for rel in (r.p.CCR_FILE,r.p.PLAN_FILE,r.p.SCRIPT_FILE,r.POLICY_FILE)}}
        record['approved_artifacts'][r.p.CCR_FILE]=wrapper['candidate_sha256']
        approval=root/'approval.json';approval.write_text(json.dumps(record));return approval,before,after
    def test_conditional_candidate_exact_future_hash_without_premature_install(self):
        with tempfile.TemporaryDirectory() as tmp:
            root=Path(tmp);approval,before,after=self.envelope_fixture(root)
            with patch.object(r,'ROOT',root): self.assertEqual(r.approval_envelope(approval),(before,after,hashlib.sha256(b'mock').hexdigest()))
            self.assertEqual((root/r.p.CCR_FILE).read_bytes(),before)
    def test_candidate_cannot_change_other_privilege_fields(self):
        with tempfile.TemporaryDirectory() as tmp:
            root=Path(tmp);approval,_,_=self.envelope_fixture(root,extra=True)
            with patch.object(r,'ROOT',root),self.assertRaisesRegex(r.p.Refused,'changes_exceed_scope'): r.approval_envelope(approval)
    def test_candidate_predecessor_changed_refused(self):
        with tempfile.TemporaryDirectory() as tmp:
            root=Path(tmp);approval,_,_=self.envelope_fixture(root);(root/r.p.CCR_FILE).write_text('different')
            with patch.object(r,'ROOT',root),self.assertRaisesRegex(r.p.Refused,'source_precondition'):r.approval_envelope(approval)
    def test_candidate_atomic_exact_transition_idempotent(self):
        with tempfile.TemporaryDirectory() as tmp:
            root=Path(tmp);_,before,after=self.envelope_fixture(root);j=Journal()
            with patch.object(r,'ROOT',root):
                r.activate_candidate(before,after,j);r.activate_candidate(before,after,j)
            self.assertEqual((root/r.p.CCR_FILE).read_bytes(),after)
            self.assertEqual(len(j.rows),2)
    def test_role_claims_policy_ttl_drift_rejected(self):
        for key,value in [('audience','other'),('bound_service_account_names',['other']),('token_ttl',0),('token_no_default_policy',False),('token_policies',[r.POLICY,'default'])]:
            data=self.role(); data[key]=value
            with self.subTest(key=key),self.assertRaisesRegex(r.p.Refused,'role_drift'): r.validate_role(data)
    def test_role_unchanged_baseline_accepted(self): r.validate_role(self.role())
    def test_unapproved_inputs_before_metadata_reads_or_writes(self):
        with patch.object(r,'approval_envelope',side_effect=r.p.Refused('runner_approval_invalid')),patch.object(r.p.Bao,'run') as run:
            with self.assertRaises(r.p.Refused): r.execute('mock','mock',Journal())
            run.assert_not_called()
    def test_installed_claim_failure_prevents_candidate_activation(self):
        with patch.object(r,'approval_envelope',return_value=(b'before',b'after',hashlib.sha256((r.ROOT/r.POLICY_FILE).read_bytes()).hexdigest())),patch.object(r,'preflight',side_effect=r.p.Refused('installed_eso_role_drift')),patch.object(r,'activate_candidate') as activate,patch.object(r.subprocess,'run') as child:
            with self.assertRaisesRegex(r.p.Refused,'role_drift'): r.execute('mock','mock',Journal())
            activate.assert_not_called(); child.assert_not_called()
    def test_policy_drift_after_preflight_prevents_write_and_credential_route(self):
        with patch.object(r,'approval_envelope',return_value=(b'before',b'after',hashlib.sha256((r.ROOT/r.POLICY_FILE).read_bytes()).hexdigest())),patch.object(r,'activate_candidate'),patch.object(r.p,'approval'),patch.object(r,'preflight',return_value=('old','new')),patch.object(r,'read_policy',return_value='changed'),patch.object(r.p.Bao,'run') as run,patch.object(r.subprocess,'run') as child:
            with self.assertRaisesRegex(r.p.Refused,'changed_since'): r.execute('mock','mock',Journal())
            run.assert_not_called(); child.assert_not_called()
    def snapshot(self): return (r.ROOT/r.POLICY_FILE).read_bytes().decode().strip()
    def test_exact_policy_only_then_existing_bootstrap_route(self):
        j=Journal(); ok=type('R',(),{'returncode':0,'stdout':b'','stderr':b''})()
        with patch.object(r,'approval_envelope',return_value=(b'before',b'after',hashlib.sha256((r.ROOT/r.POLICY_FILE).read_bytes()).hexdigest())),patch.object(r,'activate_candidate'),patch.object(r.p,'approval'),patch.object(r,'preflight',return_value=('old','new')),patch.object(r,'read_policy',side_effect=['old',self.snapshot()]),patch.object(r,'verify_ready') as verify,patch.object(r.p.Bao,'run',return_value=ok) as run,patch.object(r.subprocess,'run',return_value=ok) as child:
            r.execute('mockapproval','mockreceipt',j)
        self.assertEqual(verify.call_count,2);self.assertEqual(run.call_count,1); self.assertEqual(run.call_args.args[0],['policy','write',r.POLICY,'-']);self.assertEqual(run.call_args.kwargs['stdin'],(r.ROOT/r.POLICY_FILE).read_bytes()); self.assertIn(r.p.SCRIPT_FILE,str(child.call_args.args[0])); self.assertNotIn('auth/kubernetes',str(run.call_args))
    def test_policy_write_timeout_recorded_without_bootstrap_retry(self):
        j=Journal()
        with patch.object(r,'approval_envelope',return_value=(b'before',b'after',hashlib.sha256((r.ROOT/r.POLICY_FILE).read_bytes()).hexdigest())),patch.object(r,'activate_candidate'),patch.object(r.p,'approval'),patch.object(r,'preflight',return_value=('old','new')),patch.object(r,'read_policy',return_value='old'),patch.object(r.p.Bao,'run',side_effect=r.p.Refused('bao_observation_failed')),patch.object(r.subprocess,'run') as child:
            with self.assertRaises(r.p.Refused): r.execute('mock','mock',j)
        self.assertEqual(j.rows[-1],('outcome',{'operation':'exact_eso_policy_update','status':'ambiguous'})); child.assert_not_called()
    def test_mutable_policy_replaced_at_write_uses_approved_stdin_snapshot(self):
        with tempfile.TemporaryDirectory() as tmp:
            root=Path(tmp);f=root/r.POLICY_FILE;f.parent.mkdir(parents=True);f.write_bytes(b'approved-policy')
            j=Journal();seen=[];ok=type('R',(),{'returncode':0,'stdout':b'','stderr':b''})()
            def write(args,stdin=None):
                f.write_bytes(b'unapproved-replacement');seen.append((args,stdin));return ok
            with patch.object(r,'ROOT',root),patch.object(r,'approval_envelope',return_value=(b'before',b'after',hashlib.sha256(b'approved-policy').hexdigest())),patch.object(r,'activate_candidate'),patch.object(r.p,'approval'),patch.object(r,'preflight',return_value=('old','approved-policy')),patch.object(r,'read_policy',side_effect=['old','approved-policy']),patch.object(r.p.Bao,'run',side_effect=write),patch.object(r,'verify_ready'),patch.object(r.subprocess,'run',return_value=ok):
                r.execute('mock','mock',j)
            self.assertEqual(seen,[(['policy','write',r.POLICY,'-'],b'approved-policy')])
    def test_policy_changed_before_snapshot_refuses_before_write(self):
        with tempfile.TemporaryDirectory() as tmp:
            root=Path(tmp);f=root/r.POLICY_FILE;f.parent.mkdir(parents=True);f.write_bytes(b'approved-policy')
            def late_approval(*args):f.write_bytes(b'unapproved-replacement')
            with patch.object(r,'ROOT',root),patch.object(r,'approval_envelope',return_value=(b'before',b'after',hashlib.sha256(b'approved-policy').hexdigest())),patch.object(r,'activate_candidate'),patch.object(r.p,'approval',side_effect=late_approval),patch.object(r,'preflight',return_value=('old','approved-policy')),patch.object(r.p.Bao,'run') as write,patch.object(r.subprocess,'run') as child:
                with self.assertRaisesRegex(r.p.Refused,'snapshot_mismatch'):r.execute('mock','mock',Journal())
                write.assert_not_called();child.assert_not_called()
    def test_idempotent_acl_drift_before_bootstrap_refuses(self):
        desired=self.snapshot();j=Journal()
        with patch.object(r,'approval_envelope',return_value=(b'before',b'after',hashlib.sha256((r.ROOT/r.POLICY_FILE).read_bytes()).hexdigest())),patch.object(r,'activate_candidate'),patch.object(r.p,'approval'),patch.object(r,'preflight',side_effect=[(desired,desired),(r.BASELINE.strip(),desired)]),patch.object(r.p.Bao,'run') as write,patch.object(r.subprocess,'run') as child:
            with self.assertRaisesRegex(r.p.Refused,'not_installed'):r.execute('mock','mock',j)
            write.assert_not_called();child.assert_not_called()
        self.assertFalse(any(e=='terminal' for e,kw in j.rows))
    def test_acl_drift_after_bootstrap_prevents_verified_terminal(self):
        desired=self.snapshot();j=Journal();ok=type('R',(),{'returncode':0,'stdout':b'','stderr':b''})()
        with patch.object(r,'approval_envelope',return_value=(b'before',b'after',hashlib.sha256((r.ROOT/r.POLICY_FILE).read_bytes()).hexdigest())),patch.object(r,'activate_candidate'),patch.object(r.p,'approval'),patch.object(r,'preflight',side_effect=[(desired,desired),(desired,desired),(r.BASELINE.strip(),desired)]),patch.object(r.subprocess,'run',return_value=ok) as child:
            with self.assertRaisesRegex(r.p.Refused,'not_installed'):r.execute('mock','mock',j)
            child.assert_called_once()
        self.assertFalse(any(e=='terminal' for e,kw in j.rows))
    def test_immutable_role_drift_after_bootstrap_prevents_verified_terminal(self):
        desired=self.snapshot();j=Journal();ok=type('R',(),{'returncode':0,'stdout':b'','stderr':b''})()
        with patch.object(r,'approval_envelope',return_value=(b'before',b'after',hashlib.sha256((r.ROOT/r.POLICY_FILE).read_bytes()).hexdigest())),patch.object(r,'activate_candidate'),patch.object(r.p,'approval'),patch.object(r,'preflight',side_effect=[(desired,desired),(desired,desired),r.p.Refused('installed_eso_role_drift')]),patch.object(r.subprocess,'run',return_value=ok):
            with self.assertRaisesRegex(r.p.Refused,'role_drift'):r.execute('mock','mock',j)
        self.assertFalse(any(e=='terminal' for e,kw in j.rows))
    def test_child_output_stops_without_exposing_bytes(self):
        j=Journal(); bad=type('R',(),{'returncode':0,'stdout':b'fake-secret-provider-output','stderr':b''})()
        with patch.object(r,'approval_envelope',return_value=(b'before',b'after',hashlib.sha256((r.ROOT/r.POLICY_FILE).read_bytes()).hexdigest())),patch.object(r,'activate_candidate'),patch.object(r.p,'approval'),patch.object(r,'preflight',return_value=(self.snapshot(),self.snapshot())),patch.object(r,'verify_ready'),patch.object(r.subprocess,'run',return_value=bad):
            with self.assertRaisesRegex(r.p.Refused,'bootstrap_route_unverified'): r.execute('mock','mock',j)
        self.assertNotIn('secret',str(j.rows))
if __name__=='__main__': unittest.main()