"""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()