44314cad37
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
70 lines
4.5 KiB
Python
70 lines
4.5 KiB
Python
"""Delivery refuses custody/projection drift before changing a live receiver."""
|
|
import base64
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts'))
|
|
import factory_audit_delivery as delivery
|
|
|
|
|
|
class DeliveryContracts(unittest.TestCase):
|
|
def setUp(self):
|
|
self.lane=delivery.contracts()[0]
|
|
self.value='fixture_only_'+('x'*51)
|
|
self.raw={'metadata':{'version':1},'data':{'AUDIT_TOKEN':self.value,'CUSTODY_REQUEST':self.lane['ccr']}}
|
|
|
|
def test_version_one_exact_provenance_is_required(self):
|
|
self.assertEqual(delivery.custody_value(self.lane,self.raw),self.value)
|
|
for raw in [dict(self.raw,metadata={'version':2}),dict(self.raw,data=dict(self.raw['data'],CUSTODY_REQUEST='different-request')),dict(self.raw,data=dict(self.raw['data'],extra='unexpected'))]:
|
|
with self.assertRaisesRegex(delivery.LaneError,'custody_provenance_mismatch'):delivery.custody_value(self.lane,raw)
|
|
|
|
def test_bad_token_shape_never_appears_in_the_diagnostic(self):
|
|
raw=copy.deepcopy(self.raw);raw['data']['AUDIT_TOKEN']='sensitive fixture with spaces'
|
|
with self.assertRaisesRegex(delivery.LaneError,'^custody_token_shape$'):delivery.custody_value(self.lane,raw)
|
|
|
|
def test_projection_must_match_value_and_controller(self):
|
|
es={'metadata':{'uid':'expected-controller'}}
|
|
secret={'data':{self.lane['secret_key']:base64.b64encode(self.value.encode()).decode()},'metadata':{'ownerReferences':[{'uid':'expected-controller','controller':True}]}}
|
|
delivery.projection_matches(secret,es,self.lane,self.value)
|
|
with self.assertRaisesRegex(delivery.LaneError,'projection_value_mismatch'):delivery.projection_matches(secret,es,self.lane,'wrong-value')
|
|
secret['metadata']['ownerReferences'][0]['controller']=False
|
|
with self.assertRaisesRegex(delivery.LaneError,'projection_owner_mismatch'):delivery.projection_matches(secret,es,self.lane,self.value)
|
|
|
|
def test_manifest_byte_change_refuses_before_apply(self):
|
|
path,objects=delivery.guarded_manifest();self.assertEqual(len(objects),4)
|
|
with tempfile.TemporaryDirectory() as d:
|
|
target=Path(d)/'manifests';target.mkdir();(target/path.name).write_bytes(path.read_bytes()+b'\n')
|
|
with patch.object(delivery,'ROOT',Path(d)):
|
|
with self.assertRaisesRegex(delivery.LaneError,'reviewed_manifest_changed'):delivery.guarded_manifest()
|
|
|
|
def test_existing_projection_must_not_be_redirected(self):
|
|
_,objects=delivery.guarded_manifest();expected=objects[1];actual=copy.deepcopy(expected)
|
|
actual['spec']['secretStoreRef']['name']='unrelated-store'
|
|
def result(argv, **kwargs):return subprocess.CompletedProcess(argv,0,json.dumps(actual).encode(),b'')
|
|
with patch.object(delivery,'command',side_effect=result) as command:
|
|
with self.assertRaisesRegex(delivery.LaneError,'existing_projection_drift'):delivery.preflight_projection(['kubectl'],[expected])
|
|
self.assertTrue(all('apply' not in call.args[0] for call in command.call_args_list))
|
|
|
|
def test_presence_errors_are_not_treated_as_absence(self):
|
|
_,objects=delivery.guarded_manifest()
|
|
with patch.object(delivery,'command',return_value=subprocess.CompletedProcess([],1,b'',b'Forbidden')):
|
|
with self.assertRaisesRegex(delivery.LaneError,'projection_presence_unknown'):delivery.preflight_projection(['kubectl'],objects)
|
|
|
|
def test_unexpected_reader_grant_is_revoked_and_not_used(self):
|
|
lanes=delivery.contracts();lane=lanes[0]
|
|
def read(path):
|
|
if path.startswith('sys/policies'):return {'policy':lane['hcl']}
|
|
return {}
|
|
auth={'auth':{'client_token':'test-reader','token_policies':['root'],'lease_duration':900}}
|
|
with patch.object(delivery,'read_optional',side_effect=read),patch.object(delivery,'role_matches',return_value=True),patch.object(delivery,'command',return_value=subprocess.CompletedProcess([],0,b'fixture-jwt',b'')),patch.object(delivery,'bao',return_value=subprocess.CompletedProcess([],0,json.dumps(auth).encode(),b'')) as bao,patch.object(delivery,'revoke') as revoke:
|
|
with self.assertRaisesRegex(delivery.LaneError,'reader_scope_or_ttl_mismatch'):delivery.verify_readers(['kubectl'],lanes,{'lanes':[{} for x in lanes]},lambda:None)
|
|
revoke.assert_called_once_with('test-reader')
|
|
self.assertEqual(bao.call_count,1)
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|