Prepare scoped factory audit custody and enforce receiver compatibility
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
This commit is contained in:
codex
parent
d3a502b45c
commit
bfe65a5b05
11 changed files with 1053 additions and 1 deletions
227
tests/test_factory_audit_custody.py
Normal file
227
tests/test_factory_audit_custody.py
Normal file
|
|
@ -0,0 +1,227 @@
|
|||
"""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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue