"""Opt-in local OpenBao exercise; no production API or credential helper used.""" import importlib.util import json import copy import tempfile 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 ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / 'scripts')) import keycape_approval_custody as lane class CustodyExercise(unittest.TestCase): @classmethod def setUpClass(cls): if not shutil.which('bao'): raise unittest.SkipTest('OpenBao executable required') sock = socket.socket(); sock.bind(('127.0.0.1', 0)); port = sock.getsockname()[1]; sock.close() cls.env = os.environ.copy() token = secrets.token_urlsafe(32) for name in ('BAO_ADDR', 'VAULT_ADDR'): cls.env[name] = f'http://127.0.0.1:{port}' for name in ('BAO_TOKEN', 'VAULT_TOKEN', 'BAO_DEV_ROOT_TOKEN_ID'): cls.env[name] = token cls.server = subprocess.Popen(['bao', 'server', '-dev', '-dev-no-store-token', f'-dev-listen-address=127.0.0.1:{port}'], env=cls.env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) cls.environment = patch.dict(os.environ, cls.env, clear=True); cls.environment.start() try: for _ in range(40): if lane.bao(['status'], allow_failure=True).returncode == 0: break time.sleep(.25) else: raise RuntimeError('local_dev_server_not_ready') lane.bao(['secrets', 'enable', '-path=platform', 'kv-v2']) lane.bao(['auth', 'enable', 'kubernetes']) cls.lanes = lane.contracts() except Exception: 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 test_01_first_provision_and_real_acl_boundaries(self): lane.bao(['write', lane.BOUNDARY, '-'], payload={'policy': 'path "protected/fixture" { capabilities = ["deny"] }\n'}) receipt = {'lanes': []} lane.provision(self.lanes, receipt) values = [] for item in self.lanes: value = lane.data(lane.bao(['read', '-format=json', item['kv']]))['data']['data']['CLIENT_SECRET'] self.assertTrue(len(value) == 64) values.append(value) auth = lane.data(lane.bao(['token', 'create', '-format=json', '-policy=' + item['policy'], '-ttl=60s']))['auth'] token = auth['client_token'] try: self.assertEqual(lane.bao(['read', '-format=json', item['kv']], token=token).returncode, 0) sibling = next(x['kv'] for x in self.lanes if x != item) lane.require_denied(lane.bao(['read', sibling], token=token, allow_failure=True)) lane.require_denied(lane.bao(['list', 'platform/metadata/workloads'], token=token, allow_failure=True)) finally: lane.revoke(token) lane.require_denied(lane.bao(['token', 'lookup'], token=token, allow_failure=True)) child = lane.data(lane.bao(['token', 'create', '-format=json', '-policy=' + item['policy'], '-policy=agent-high-risk-boundary', '-ttl=60s']))['auth']['client_token'] try: lane.require_denied(lane.bao(['read', item['kv']], token=child, allow_failure=True)) finally: lane.revoke(child) self.assertTrue(values[0] != values[1]) self.assertTrue(all(x['kv_version'] == 1 for x in receipt['lanes'])) self.assertFalse(any(value in json.dumps(receipt) for value in values)) type(self).initial_receipt = copy.deepcopy(receipt) def test_02_retry_refuses_existing_custody(self): with self.assertRaisesRegex(lane.LaneError, 'existing_custody_requires'): lane.provision(self.lanes, {'lanes': []}) self.assertTrue(all(lane.read_optional(x['metadata'])['current_version'] == 1 for x in self.lanes)) def test_03_denials_must_be_authoritative(self): for result in [subprocess.CompletedProcess([], 0, b'', b''), subprocess.CompletedProcess([], 1, b'', b'404 Not Found'), subprocess.CompletedProcess([], 1, b'', b'503 permission denied'), subprocess.CompletedProcess([], 1, b'', b'403 unrelated refusal')]: with self.assertRaises(lane.LaneError): lane.require_denied(result) def test_04_missing_approval_refused(self): with patch.object(lane, 'ROOT', Path('/nonexistent-keycape-contract')): with self.assertRaises(FileNotFoundError): lane.contracts() def test_05_role_drift_refused(self): item = self.lanes[0] actual = lane.read_optional('auth/kubernetes/role/' + item['role']) self.assertTrue(lane.role_matches(actual, item)) for field, value in [('token_policies', [item['policy'], 'extra']), ('bound_service_account_namespaces', ['*']), ('bound_service_account_names', ['*']), ('token_ttl', 3600), ('token_period', 900)]: self.assertFalse(lane.role_matches(dict(actual, **{field: value}), item)) def test_06_direct_execution_refused(self): with self.assertRaisesRegex(lane.LaneError, 'attended_warden_envelope_required'): lane.run(None, {}) def test_07_resume_reattaches_readers_without_rewriting_values(self): before = [lane.read_optional(x['metadata']) for x in self.lanes] for item in self.lanes: lane.bao(['write', 'auth/kubernetes/role/' + item['role'], '-'], payload=dict(item['role_payload'], policies=[])) prior = dict(self.initial_receipt, status='failed', verifier_delivery_disabled_custody_versions_retained=True, keycape={'compatible_pair_restored': True}) with tempfile.TemporaryDirectory() as directory: path = Path(directory) / 'prior.json'; path.write_text(json.dumps(prior)) receipt = {'lanes': []} lane.resume_provision(self.lanes, receipt, path) self.assertTrue(receipt['custody_versions_unchanged']) self.assertTrue(all(x['existing_version_reused'] and not x['custody_seeded'] for x in receipt['lanes'])) self.assertEqual(before, [lane.read_optional(x['metadata']) for x in self.lanes]) self.assertTrue(all(lane.role_matches(lane.read_optional('auth/kubernetes/role/' + x['role']), x) for x in self.lanes)) def test_08_resume_requires_completed_rollback(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / 'prior.json'; path.write_text(json.dumps({'status':'failed'})) with self.assertRaisesRegex(lane.LaneError, 'verified_partial_activation_receipt_required'): lane.resume_provision(self.lanes, {'lanes': []}, path) if __name__ == '__main__': unittest.main()