railiance-platform/tests/test_factory_audit_custody.py

"""Disposable OpenBao and exact receiver compatibility tests; no native credentials."""
import copy
import importlib.util
import json
import os
from pathlib import Path
import secrets
import shutil
import socket
import subprocess
import sys
import time
import unittest
from unittest.mock import patch
import yaml

ROOT=Path(__file__).resolve().parents[1]
sys.path.insert(0,str(ROOT/'scripts'))
import factory_audit_custody as lane


class Contracts(unittest.TestCase):
    def test_proposed_requests_cannot_seed(self):
        self.assertEqual(len(lane.contracts()),2)
        with patch.object(lane,'bao',side_effect=AssertionError('must not contact Bao')):
            with self.assertRaisesRegex(lane.LaneError,'approved_ccrs_required'):
                lane.contracts(approved=True)

    def test_each_named_owner_approval_is_required(self):
        module=lane.credential_module()
        original=module.validate_ccr
        for missing in ['platform-operator','audit-core-owner','approval-engine-owner','informed-decision-owner']:
            def candidate(path):
                c,errors,warnings=original(path);c=copy.deepcopy(c);c['status']='approved'
                owners=['platform-operator','audit-core-owner',c['target']['workload']+'-owner']
                c['review']['comments']=[{'reviewer':owner,'decision':'approved'} for owner in owners if owner!=missing]
                return c,errors,warnings
            with patch.object(module,'validate_ccr',side_effect=candidate),patch.object(lane,'credential_module',return_value=module):
                with self.assertRaisesRegex(lane.LaneError,'named_owner_reviews_required'):lane.contracts(approved=True)

    def test_altered_credential_coordinates_refuse_even_with_approvals(self):
        module=lane.credential_module();original=module.validate_ccr
        def candidate(path):
            c,errors,warnings=original(path);c=copy.deepcopy(c);c['status']='approved'
            c['review']['comments']=[{'reviewer':owner,'decision':'approved'} for owner in ['platform-operator','audit-core-owner',c['target']['workload']+'-owner']]
            c['openbao']['kv_path']='platform/workloads/audit-core/senders'
            return c,errors,warnings
        with patch.object(module,'validate_ccr',side_effect=candidate),patch.object(lane,'credential_module',return_value=module):
            with self.assertRaisesRegex(lane.LaneError,'exact_contract_required'):lane.contracts(approved=True)

    def test_projection_only_delivers_one_field_per_exact_namespace(self):
        objects=list(yaml.safe_load_all((ROOT/'manifests/factory-audit-senders.yaml').read_text()))
        self.assertEqual([x['kind'] for x in objects],['ClusterSecretStore','ExternalSecret']*2)
        for contract,store,es in zip(lane.contracts(),objects[::2],objects[1::2]):
            self.assertEqual(store['spec']['conditions'],[{'namespaces':[contract['name']]}])
            self.assertEqual(es['metadata']['namespace'],contract['name'])
            self.assertEqual(es['spec']['data'],[{'secretKey':contract['secret_key'],'remoteRef':{'key':contract['kv'].removeprefix('platform/data/'),'property':'AUDIT_TOKEN'}}])
            self.assertEqual(es['spec']['secretStoreRef']['name'],store['metadata']['name'])
            self.assertNotIn('senders.json',json.dumps(es))

    def test_current_receiver_source_supports_exact_contract(self):
        source=os.environ.get('AUDIT_CORE_SOURCE')
        if not source:self.skipTest('set AUDIT_CORE_SOURCE for actual receiver contract')
        result=subprocess.run([sys.executable,'-c',lane.RECEIVER_PROBE],cwd=source,capture_output=True,text=True,check=True)
        self.assertTrue(all(json.loads(result.stdout).values()))

    def test_legacy_receiver_is_refused_before_credentials(self):
        dep={'metadata':{'uid':'d','generation':1,'resourceVersion':'1'},'spec':{'replicas':1,'template':{'spec':{'containers':[{'name':'audit-core','image':'forgejo.coulomb.social/coulomb/audit-core@sha256:'+'a'*64}]}}},'status':{'observedGeneration':1,'readyReplicas':1}}
        pod={'metadata':{'name':'audit-core-test','uid':'p'},'spec':dep['spec']['template']['spec'],'status':{'containerStatuses':[{'ready':True}]}}
        responses=[dep,{'items':[pod]},{'load_bearing':False,'redact':True,'write_only':True,'source_exact':True,'tenant_exact':True}]
        with patch.object(lane,'assert_cluster'),patch.object(lane,'command',side_effect=[subprocess.CompletedProcess([],0,json.dumps(x).encode(),b'') for x in responses]),patch.object(lane,'bao',side_effect=AssertionError('no credentials')):
            with self.assertRaisesRegex(lane.LaneError,'receiver_lacks_sender_contract'):
                lane.receiver_check(['kubectl'],dep['spec']['template']['spec']['containers'][0]['image'])


class OpenBaoExercise(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        if not shutil.which('bao'):raise unittest.SkipTest('local OpenBao binary required')
        with socket.socket() as s:s.bind(('127.0.0.1',0));port=s.getsockname()[1]
        token=secrets.token_urlsafe(32)
        env=dict(os.environ,BAO_ADDR=f'http://127.0.0.1:{port}',VAULT_ADDR=f'http://127.0.0.1:{port}',BAO_TOKEN=token,VAULT_TOKEN=token,BAO_DEV_ROOT_TOKEN_ID=token)
        cls.server=subprocess.Popen(['bao','server','-dev','-dev-no-store-token',f'-dev-listen-address=127.0.0.1:{port}'],env=env,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)
        cls.environment=patch.dict(os.environ,env,clear=True);cls.environment.start()
        try:
            for _ in range(60):
                if lane.bao(['status'],allow_failure=True).returncode==0:break
                time.sleep(.1)
            else:raise RuntimeError('local server not ready')
            lane.bao(['secrets','enable','-path=platform','kv-v2'])
            lane.bao(['auth','enable','kubernetes'])
            cls.lanes=lane.contracts()
        except BaseException:
            cls.environment.stop();cls.server.terminate();cls.server.wait(timeout=10);raise

    @classmethod
    def tearDownClass(cls):
        cls.environment.stop();cls.server.terminate();cls.server.wait(timeout=10)

    def setUp(self):
        # These deletes target only the test's own loopback dev server.
        for item in self.lanes:lane.bao(['delete',item['metadata']],allow_failure=True)
        lane.bao(['delete',lane.REGISTRY.replace('/data/','/metadata/')],allow_failure=True)
        self.original=[{'name':'existing','tokens':[secrets.token_urlsafe(48)],'sources':['existing'],'tenants':['*'],'may_read':True,'may_write':False,'secret_policy':'reject'}]
        self.body={'senders.json':json.dumps(self.original),'unrelated-metadata':'preserve me'}
        lane.bao(['write',lane.REGISTRY,'-'],payload={'options':{'cas':0},'data':self.body})
        self.receipt={};self.saved=[]

    def run_seed(self,**kwargs):
        lane.seed(self.lanes,self.receipt,lambda:self.saved.append(copy.deepcopy(self.receipt)),**kwargs)

    def token(self,item):
        return lane.data(lane.bao(['read','-format=json',item['kv']]))['data']['data']['AUDIT_TOKEN']

    def test_first_provision_and_idempotent_resume(self):
        self.run_seed()
        before=[self.token(x) for x in self.lanes];self.assertNotEqual(*before)
        version,body,rows,_=lane.snapshot()
        self.assertEqual(rows[0],self.original[0]);self.assertEqual(body['unrelated-metadata'],'preserve me')
        self.assertEqual(len(rows),3)
        self.run_seed(resume=True)
        self.assertEqual(lane.snapshot()[0],version)
        self.assertEqual([self.token(x) for x in self.lanes],before)
        self.assertFalse(any(t in json.dumps(self.saved) for t in before))
        with self.assertRaisesRegex(lane.LaneError,'existing_custody_requires_reviewed_resume'):self.run_seed()

    def test_crash_after_first_write_reuses_value(self):
        real=lane.bao
        def fail_second(args,**kwargs):
            if args[:2]==['write','-format=json'] and args[2]==self.lanes[1]['kv']:raise lane.LaneError('injected_interrupt')
            return real(args,**kwargs)
        with patch.object(lane,'bao',side_effect=fail_second):
            with self.assertRaisesRegex(lane.LaneError,'injected_interrupt'):self.run_seed()
        first=self.token(self.lanes[0]);self.assertEqual(lane.snapshot()[0],1)
        self.run_seed(resume=True)
        self.assertEqual(self.token(self.lanes[0]),first);self.assertEqual(len(lane.snapshot()[2]),3)

    def test_lost_registry_write_reply_reconciles_without_rewriting(self):
        real=lane.bao
        def lost_reply(args,**kwargs):
            result=real(args,**kwargs)
            if args[:3]==['write','-format=json',lane.REGISTRY]:raise lane.LaneError('reply_lost')
            return result
        with patch.object(lane,'bao',side_effect=lost_reply):
            with self.assertRaisesRegex(lane.LaneError,'reply_lost'):self.run_seed()
        before=[self.token(x) for x in self.lanes];version=lane.snapshot()[0]
        self.run_seed(resume=True)
        self.assertEqual(lane.snapshot()[0],version);self.assertEqual([self.token(x) for x in self.lanes],before)

    def test_cas_conflict_preserves_concurrent_writer(self):
        real=lane.bao
        extra={'name':'concurrent','tokens':[secrets.token_urlsafe(48)],'sources':['concurrent']}
        def race(args,**kwargs):
            if args[:3]==['write','-format=json',lane.REGISTRY]:
                body=dict(self.body,**{'senders.json':json.dumps(self.original+[extra])})
                real(['write',lane.REGISTRY,'-'],payload={'options':{'cas':1},'data':body})
            return real(args,**kwargs)
        with patch.object(lane,'bao',side_effect=race):
            with self.assertRaises(lane.LaneError):self.run_seed()
        self.assertEqual(lane.snapshot()[2],self.original+[extra])
        self.run_seed(resume=True);self.assertIn(extra,lane.snapshot()[2])

    def test_wrong_provenance_refuses_resume(self):
        x=self.lanes[0]
        lane.bao(['write',x['kv'],'-'],payload={'options':{'cas':0},'data':{'AUDIT_TOKEN':secrets.token_urlsafe(48),'CUSTODY_REQUEST':'different-request'}})
        with self.assertRaisesRegex(lane.LaneError,'custody_provenance_mismatch'):self.run_seed(resume=True)
        self.assertIsNone(lane.read_optional(self.lanes[1]['metadata']))
        self.assertEqual(lane.snapshot()[0],1)

    def test_scope_drift_is_never_overwritten(self):
        self.run_seed();version,body,rows,_=lane.snapshot();rows[-1]['may_read']=True
        lane.bao(['write',lane.REGISTRY,'-'],payload={'options':{'cas':version},'data':dict(body,**{'senders.json':json.dumps(rows)})})
        with self.assertRaisesRegex(lane.LaneError,'registered_sender_drift'):self.run_seed(resume=True)
        self.assertTrue(lane.snapshot()[2][-1]['may_read'])

    def test_duplicate_names_refuse_before_token_creation(self):
        lane.bao(['write',lane.REGISTRY,'-'],payload={'options':{'cas':1},'data':{'senders.json':json.dumps(self.original*2)}})
        with self.assertRaisesRegex(lane.LaneError,'registry_identity_invalid_or_duplicate'):self.run_seed()
        self.assertTrue(all(lane.read_optional(x['metadata']) is None for x in self.lanes))

    def test_existing_cross_sender_token_collision_refused(self):
        row=dict(self.original[0],name='different')
        lane.bao(['write',lane.REGISTRY,'-'],payload={'options':{'cas':1},'data':{'senders.json':json.dumps(self.original+[row])}})
        with self.assertRaisesRegex(lane.LaneError,'registry_token_collision'):self.run_seed()

    def test_metadata_preserves_boundary_and_deny_overrides_reader(self):
        original='path "protected/existing" { capabilities = ["deny"] }\n'
        lane.bao(['write',lane.BOUNDARY,'-'],payload={'policy':original})
        lane.prepare_metadata(self.lanes)
        boundary=lane.read_optional(lane.BOUNDARY)['policy']
        self.assertTrue(boundary.startswith(original))
        for item in self.lanes:
            self.assertTrue(lane.role_matches(lane.read_optional('auth/kubernetes/role/'+item['role']),item))
            self.assertIn(item['kv'],boundary);self.assertIn(item['metadata'],boundary)
        self.run_seed()
        item=self.lanes[0]
        token=lane.data(lane.bao(['token','create','-format=json','-policy='+item['policy'],'-policy=agent-high-risk-boundary','-ttl=60s']))['auth']['client_token']
        try:
            result=lane.bao(['read',item['kv']],token=token,allow_failure=True)
            self.assertNotEqual(result.returncode,0);self.assertIn(b'403',result.stderr)
        finally:lane.bao(['write','auth/token/revoke','-'],payload={'token':token})
        lane.prepare_metadata(self.lanes)
        self.assertEqual(lane.read_optional(lane.BOUNDARY)['policy'],boundary)

    def test_metadata_policy_drift_refuses_before_boundary_change(self):
        item=self.lanes[0];original=lane.read_optional(lane.BOUNDARY)
        lane.bao(['write','sys/policies/acl/'+item['policy'],'-'],payload={'policy':'path "unrelated/data" { capabilities = ["read"] }'})
        try:
            with self.assertRaisesRegex(lane.LaneError,'existing_policy_drift'):lane.prepare_metadata(self.lanes)
            self.assertEqual(lane.read_optional(lane.BOUNDARY),original)
            self.assertTrue(all(lane.read_optional(x['metadata']) is None for x in self.lanes))
        finally:lane.bao(['delete','sys/policies/acl/'+item['policy']])

    def test_exact_read_policies_exclude_sibling_and_registry(self):
        self.run_seed()
        for item in self.lanes:
            lane.bao(['write','sys/policies/acl/'+item['policy'],'-'],payload={'policy':item['hcl']})
            token=lane.data(lane.bao(['token','create','-format=json','-policy='+item['policy'],'-ttl=60s']))['auth']['client_token']
            try:
                self.assertEqual(lane.bao(['read',item['kv']],token=token).returncode,0)
                targets=[lane.REGISTRY,item['metadata']]+[x['kv'] for x in self.lanes if x!=item]
                for target in targets:
                    result=lane.bao(['read',target],token=token,allow_failure=True)
                    self.assertNotEqual(result.returncode,0);self.assertIn(b'403',result.stderr)
            finally:lane.bao(['write','auth/token/revoke','-'],payload={'token':token})

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