feat: admit and exercise KeyCape verifier custody activation

Assistant: codex
Assistant-Model: gpt-5.6-luna
Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc

tests/test_keycape_approval_custody.py

@@ -0,0 +1,114 @@
+"""Opt-in local OpenBao exercise; no production API or credential helper used."""
+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
+
+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))
+
+    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, {})
+
+if __name__ == '__main__':
+    unittest.main()