railiance-platform/tests/test_keycape_approval_custody.py

"""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()