"""Synthetic-fixture tests for scripts/nk_bind_executor_custody.py; no OpenBao, cluster or credential.""" import base64 import copy import io import json import os from pathlib import Path import subprocess import sys import tempfile import unittest from contextlib import redirect_stdout from unittest.mock import patch import yaml ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / 'scripts')) import nk_bind_executor_custody as lane # noqa: E402 import keycape_issuer_eso # noqa: E402 AUDIT_CORE = Path(os.environ.get('AUDIT_CORE_SOURCE', ROOT.parent / 'audit-core')) RAPP = ROOT.parent / 'rapp-bounded-executor' def approved_module(mutate=None, drop=None): module = lane.credential_module() original = module.validate_ccr def candidate(path): ccr, errors, warnings = original(path) ccr = copy.deepcopy(ccr) ccr['status'] = 'approved' ccr['openbao']['auth']['bound_claims_confirmed'] = True owners = next(l['approvers'] for l in lane.LANES.values() if l['ccr'] == ccr['id']) ccr['review']['comments'] = [{'reviewer': o, 'decision': 'approved'} for o in owners if o != drop] if mutate: mutate(ccr) return ccr, errors, warnings module.validate_ccr = candidate return module def ok(payload=None, stderr=b''): return subprocess.CompletedProcess([], 0, json.dumps(payload).encode() if payload is not None else b'', stderr) def missing(): return subprocess.CompletedProcess([], 2, b'', b'No value found at path') class FakeBao: """Dictionary-backed KV v2 and sys/policies with CAS; records every write.""" def __init__(self, registry_rows, boundary='path "platform/data/x" { capabilities = ["deny"] }\n'): self.kv = {lane.REGISTRY: {'version': 3, 'data': {'senders.json': json.dumps(registry_rows), 'other': 'kept'}}} self.policies = {'agent-high-risk-boundary': boundary} self.writes = [] self.fail_registry_cas = False def __call__(self, args, payload=None, token=None, allow_failure=False): if args[0] == 'read': path = args[-1] if path.startswith('sys/policies/acl/'): name = path.rsplit('/', 1)[1] return ok({'data': {'policy': self.policies[name]}}) if name in self.policies else missing() if path.startswith('platform/metadata/'): key = path.replace('platform/metadata/', 'platform/data/', 1) if key not in self.kv: return missing() v = self.kv[key]['version'] return ok({'data': {'current_version': v, 'versions': {str(v): {}}}}) if path in self.kv: entry = self.kv[path] return ok({'data': {'data': entry['data'], 'metadata': {'version': entry['version']}}}) return missing() if args[0] == 'write': path = args[-2] self.writes.append((path, copy.deepcopy(payload))) if path.startswith('sys/policies/acl/'): self.policies[path.rsplit('/', 1)[1]] = payload['policy'] return ok({}) cas = payload['options']['cas'] current = self.kv.get(path, {'version': 0})['version'] if cas != current or (path == lane.REGISTRY and self.fail_registry_cas): return subprocess.CompletedProcess([], 2, b'', b'check-and-set parameter did not match') self.kv[path] = {'version': current + 1, 'data': copy.deepcopy(payload['data'])} return ok({'data': {'version': current + 1}, 'request_id': 'req-%d' % len(self.writes)}) raise AssertionError('unexpected bao call %r' % (args,)) EXISTING = [{'name': 'approval-engine', 'tokens': ['A' * 64], 'sources': ['approval-engine'], 'tenants': ['tenant:platform'], 'may_write': True, 'may_read': False}, {'name': 'informed-decision', 'tokens': ['B' * 64], 'sources': ['informed-decision'], 'tenants': ['tenant:platform'], 'may_write': True, 'may_read': False}] class Contracts(unittest.TestCase): def test_plan_checks_the_drafts_offline(self): contract = lane.contracts(approved=False) self.assertEqual(sorted(contract), ['audit', 'consumer', 'eso']) out = io.StringIO() with patch.object(lane, 'bao', side_effect=AssertionError('plan must not contact Bao')), redirect_stdout(out): self.assertEqual(lane.main(['plan']), 0) plan = json.loads(out.getvalue()) self.assertEqual(plan['lanes']['consumer']['role'], 'keycape-nk-bind-executor-approval-bounded-executor-workload-kv-read') self.assertEqual(plan['lanes']['audit']['external_secret'], 'sso/bounded-executor-tokens') def test_unapproved_requests_refuse(self): module = lane.credential_module() original = module.validate_ccr def proposed(path): ccr, e, w = original(path) ccr = copy.deepcopy(ccr) ccr['status'] = 'proposed' return ccr, e, w module.validate_ccr = proposed with self.assertRaisesRegex(lane.LaneError, 'approved_ccr_required'): lane.contracts(approved=True, module=module, git_check=lambda p: True) def test_every_named_owner_must_approve(self): for name, spec in lane.LANES.items(): for owner in spec['approvers']: module = approved_module(drop=owner) with self.assertRaisesRegex(lane.LaneError, 'named_owner_approvals_required'): lane.contracts(approved=True, module=module, git_check=lambda p: True) def test_unconfirmed_binding_or_uncommitted_ccr_refuses(self): module = approved_module(lambda c: c['openbao']['auth'].update(bound_claims_confirmed=False)) with self.assertRaisesRegex(lane.LaneError, 'binding_unconfirmed'): lane.contracts(approved=True, module=module, git_check=lambda p: True) with self.assertRaisesRegex(lane.LaneError, 'ccr_not_committed'): lane.contracts(approved=True, module=approved_module(), git_check=lambda p: False) def test_drifted_coordinates_refuse_even_when_approved(self): drifts = [lambda c: c['openbao'].update(kv_path='platform/workloads/keycape/forgejo'), lambda c: c['openbao']['auth'].update(audience='vault'), lambda c: c['openbao']['auth'].update(ttl='1h'), lambda c: c['openbao']['auth']['bound_claims'].update(service_account_names=['default'])] for drift in drifts: with self.assertRaisesRegex(lane.LaneError, 'exact_contract_required'): lane.contracts(approved=True, module=approved_module(drift), git_check=lambda p: True) def test_approved_contract_passes(self): contract = lane.contracts(approved=True, module=approved_module(), git_check=lambda p: True) self.assertIn('auth/token/revoke-self', contract['audit']['hcl']) class Fence(unittest.TestCase): def test_appends_only_missing_denies_and_preserves_live_content(self): live = 'path "platform/data/x" { capabilities = ["deny"] }\n' + lane.deny_line(lane.FENCED[2]) + '\n' fake = FakeBao([], boundary=live) receipt = {} with patch.object(lane, 'bao', side_effect=fake): lane.fence(receipt) self.assertEqual(receipt['fence_lines_added'], 3) after = fake.policies['agent-high-risk-boundary'] self.assertTrue(after.startswith(live)) for path in lane.FENCED: self.assertIn(f'"{path}"', after) lane.fence(receipt) self.assertEqual(receipt['fence_lines_added'], 0) self.assertEqual(len(fake.writes), 1) def test_present_path_without_deny_refuses(self): live = 'path "%s" { capabilities = ["read"] }\n' % lane.FENCED[0] fake = FakeBao([], boundary=live) with patch.object(lane, 'bao', side_effect=fake), self.assertRaisesRegex(lane.LaneError, 'boundary_path_drift'): lane.fence({}) self.assertEqual(fake.writes, []) def test_source_boundary_carries_the_same_denies(self): source = (ROOT / 'openbao/policies/agent-high-risk-boundary.hcl').read_text() for path in lane.FENCED: self.assertIn(lane.deny_line(path), source) class Registry(unittest.TestCase): def test_snapshot_refuses_empty_duplicate_or_colliding_registry(self): for rows in ([], EXISTING + [copy.deepcopy(EXISTING[0])], EXISTING + [{'name': 'x', 'tokens': ['A' * 64]}]): with patch.object(lane, 'bao', side_effect=FakeBao(rows)), self.assertRaises(lane.LaneError): lane.registry_snapshot() with patch.object(lane, 'bao', side_effect=FakeBao(EXISTING)): self.assertEqual(lane.registry_snapshot()[0], 3) def test_receiver_parser_accepts_the_row_with_the_declared_overlay(self): if not (AUDIT_CORE / 'audit_core/senders.py').exists(): self.skipTest('audit-core source not checked out') row = dict(lane.SENDER, tokens=['fixture-' + 'x' * 56]) overlay = (AUDIT_CORE / 'deploy/senders-scope.json').read_text() self.assertIn(lane.SENDER['name'], [e['name'] for e in json.loads(overlay)]) probe = ('import json,sys\nfrom audit_core.senders import SenderRegistry\n' 'row,overlay=json.loads(sys.argv[1]),sys.argv[2]\n' 'r=SenderRegistry.from_env({"AUDIT_CORE_SENDERS":json.dumps([row]),"AUDIT_CORE_SENDERS_SCOPE":overlay})\n' 's=r.authenticate("Bearer "+row["tokens"][0])\n' 'print(json.dumps([s.secret_policy,s.permits_source("net-kingdom.bounded-executor"),' 's.permits_tenant("platform"),s.permits_tenant("tenant:platform"),s.may_read]))\n') result = subprocess.run([sys.executable, '-c', probe, json.dumps(row), overlay], cwd=AUDIT_CORE, capture_output=True, text=True) if result.returncode != 0: self.skipTest('audit-core dependencies unavailable: ' + result.stderr.strip().splitlines()[-1]) self.assertEqual(json.loads(result.stdout), ['reject', True, True, False, False]) class Manifests(unittest.TestCase): docs = list(yaml.safe_load_all((ROOT / 'manifests/nk-bind-executor-eso.yaml').read_text())) def test_keycape_delivery_is_the_platform_template_rendering(self): store, external = keycape_issuer_eso.documents(lane.CLIENT_ID) self.assertEqual(self.docs[0]['spec'], store['spec']) self.assertEqual(self.docs[0]['metadata']['name'], store['metadata']['name']) self.assertEqual(self.docs[1]['spec'], external['spec']) self.assertEqual(self.docs[1]['metadata']['name'], external['metadata']['name']) self.assertEqual(self.docs[1]['metadata']['namespace'], 'sso') def test_each_store_matches_its_lane_and_delivers_one_field(self): for (store, external), name in zip((self.docs[0:2], self.docs[2:4]), ('eso', 'audit')): spec = lane.LANES[name] auth = store['spec']['provider']['vault']['auth']['kubernetes'] self.assertEqual(store['metadata']['name'], spec['store']) self.assertEqual(auth['role'], spec['role']) self.assertEqual(auth['serviceAccountRef'], {'name': 'external-secrets', 'namespace': 'external-secrets', 'audiences': ['openbao']}) self.assertEqual(store['spec']['conditions'], [{'namespaces': ['sso']}]) self.assertEqual((external['metadata']['namespace'], external['metadata']['name']), spec['external_secret']) self.assertEqual(external['spec']['data'], [{'secretKey': spec['secret_key'], 'remoteRef': { 'key': spec['kv'].removeprefix('platform/'), 'property': spec['fields'][0]}}]) def test_audit_secret_is_the_one_the_executor_job_mounts(self): job = RAPP / 'templates/job.yaml' if not job.exists(): self.skipTest('rapp-bounded-executor not checked out') pod = list(yaml.safe_load_all(job.read_text()))[1]['spec']['template']['spec'] volume = next(v for v in pod['volumes'] if v['name'] == 'tokens')['secret'] self.assertEqual(volume['secretName'], lane.LANES['audit']['external_secret'][1]) self.assertEqual([i['key'] for i in volume['items']], [lane.LANES['audit']['secret_key']]) class Main(unittest.TestCase): def test_refuses_outside_approval_and_never_prints_a_value(self): with tempfile.TemporaryDirectory() as directory: receipt = Path(directory) / 'r.json' out = io.StringIO() with patch.object(lane, 'bao', side_effect=AssertionError('must not contact Bao')), redirect_stdout(out): code = lane.main(['fence', '--receipt', str(receipt)]) self.assertEqual(code, 1) body = json.loads(receipt.read_text()) self.assertEqual(body['status'], 'refused') self.assertEqual(oct(receipt.stat().st_mode & 0o777), '0o600') self.assertEqual(out.getvalue(), '') with redirect_stdout(io.StringIO()): self.assertEqual(lane.main(['verify', '--receipt', str(receipt)]), 1) # existing receipt refused class Verify(unittest.TestCase): def test_delivered_audit_copy_must_hash_equal_custody(self): token = 'T' * 64 self.assertEqual(lane.sha(token), lane.sha(base64.b64decode(base64.b64encode(token.encode())))) self.assertNotEqual(lane.sha(token), lane.sha('U' * 64)) def test_role_readback_shape(self): spec = lane.LANES['consumer'] good = {'bound_service_account_names': ['nk-bind-executor'], 'bound_service_account_namespaces': ['sso'], 'token_policies': [spec['policy']], 'audience': 'openbao', 'token_ttl': 900, 'token_max_ttl': 900, 'token_explicit_max_ttl': 900, 'token_no_default_policy': True} self.assertTrue(lane.role_ok(good, spec)) for key, value in (('audience', ''), ('token_ttl', 3600), ('token_policies', [spec['policy'], 'x']), ('bound_service_account_namespace_selector', '{}'), ('token_period', 60)): self.assertFalse(lane.role_ok(dict(good, **{key: value}), spec)) if __name__ == '__main__': unittest.main()