"""CCR-2026-0033 custody child: synthetic values and fakes only; no OpenBao, cluster or real file.""" import contextlib import io import json import os from pathlib import Path import sys import tempfile from types import SimpleNamespace import unittest from unittest.mock import patch import yaml ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / 'scripts')) import keycape_secrets_engine_openbao_custody as child # noqa: E402 DECISION = '00000000-0000-4000-8000-000000000033' SYNTHETIC = 'synthetic-test-value-' + 'x' * 43 def result(code=0, stdout=b'', stderr=b''): return SimpleNamespace(returncode=code, stdout=stdout, stderr=stderr) class FakeBao: def __init__(self, current=None, fail_write=False): self.current, self.fail_write, self.calls, self.payloads = current, fail_write, [], [] def __call__(self, args, *, payload=None, token=None, allow_failure=False): self.calls.append(list(args)) if args[:2] == ['token', 'lookup']: return result(stdout=json.dumps({'data': {'policies': ['default', 'platform-admin']}}).encode()) if args[0] == 'read' and args[-1] == child.KV_METADATA: if self.current is None: return result(2, stderr=b'No value found at ' + child.KV_METADATA.encode()) return result(stdout=json.dumps({'data': {'current_version': self.current, 'versions': {str(self.current): {'destroyed': False, 'deletion_time': ''}}}}).encode()) if args[0] == 'write' and args[2] == child.KV_DATA: self.payloads.append(payload) if self.fail_write or payload['options']['cas'] != (self.current or 0): raise child.LaneError('command_failed') self.current = (self.current or 0) + 1 return result(stdout=json.dumps({'request_id': 'req-1', 'data': {'version': self.current}}).encode()) raise AssertionError('unexpected bao call: ' + ' '.join(args[:2])) class CustodyChild(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() base = Path(self.tmp.name) os.chmod(base, 0o755) self.consumer = base / 'home/.secrets-engine/keycape/secrets-engine-openbao.client-secret' self.consumer.parent.parent.mkdir(parents=True) ccr = yaml.safe_load((ROOT / 'credential-change-requests/CCR-2026-0033.yaml').read_text()) ccr['status'] = 'approved' ccr['openbao']['auth']['bound_claims_confirmed'] = True ccr['consumer_copy']['file'] = str(self.consumer) ccr['review']['comments'] = [ {'reviewer': r, 'decision': 'approved', 'date': '2026-10-06', 'comment': 'custody child approved by decision ' + DECISION} for r in child.APPROVERS] self.ccr = base / 'CCR-2026-0033.yaml' self.ccr.write_text(yaml.safe_dump(ccr, sort_keys=False)) checker = base / 'check.py' checker.write_text('#!' + sys.executable + '\nimport json,os,stat,sys\np=sys.argv[sys.argv.index("--path")+1]\n' 'ok=os.path.isfile(p) and stat.S_IMODE(os.stat(p).st_mode)==0o600\n' 'print(json.dumps({"ok":ok}))\nsys.exit(0 if ok else 1)\n') checker.chmod(0o700) self.checker = str(checker) self.bao = FakeBao() self.patches = [patch.object(child, 'CCR_PATH', self.ccr), patch.object(child, 'CONSUMER_FILE', self.consumer), patch.object(child, 'bao', self.bao), patch.object(child, 'attended_envelope', lambda: True), patch.object(child.secrets, 'token_urlsafe', lambda n: SYNTHETIC)] for p in self.patches: p.start() self.umask = os.umask(0o022) def tearDown(self): os.umask(self.umask) for p in reversed(self.patches): p.stop() self.tmp.cleanup() def run_child(self, *extra, decision=DECISION): receipt = Path(self.tmp.name) / ('receipt-%d.json' % len(list(Path(self.tmp.name).glob('receipt-*')))) out, err = io.StringIO(), io.StringIO() argv = ['provision', '--receipt', str(receipt), '--secrets-engine', self.checker, *extra] if decision is not None: argv += ['--approval-decision', decision] with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): code = child.main(argv) text = receipt.read_text() for blob in (text, out.getvalue(), err.getvalue(), json.dumps(self.bao.calls)): self.assertNotIn(SYNTHETIC, blob) self.assertEqual(oct(receipt.stat().st_mode & 0o777), oct(0o600)) return code, json.loads(text), receipt def test_refuses_without_or_with_wrong_decision(self): with self.assertRaises(SystemExit), contextlib.redirect_stderr(io.StringIO()): child.main(['provision', '--receipt', str(Path(self.tmp.name) / 'r.json')]) for decision, reason in (('not-a-uuid', 'approval_decision_id_required'), (child.BOUNDARY_DECISION, 'boundary_decision_does_not_approve_custody_child'), ('11111111-1111-4111-8111-111111111111', 'custody_child_decision_not_recorded_in_ccr')): code, receipt, _ = self.run_child(decision=decision) self.assertEqual((code, receipt['error']), (1, reason)) self.assertEqual(self.bao.calls, []) def test_committed_proposed_ccr_is_refused(self): with patch.object(child, 'CCR_PATH', ROOT / 'credential-change-requests/CCR-2026-0033.yaml'): code, receipt, _ = self.run_child() self.assertEqual((code, receipt['error']), (1, 'approved_ccr_required')) self.assertEqual(self.bao.calls, []) def test_refuses_outside_attended_envelope(self): with patch.object(child, 'attended_envelope', lambda: False): code, receipt, _ = self.run_child() self.assertEqual(receipt['error'], 'attended_warden_envelope_required') self.assertEqual(self.bao.calls, []) def test_provision_writes_kv_and_consumer_in_one_process(self): code, receipt, _ = self.run_child() self.assertEqual(code, 0) self.assertEqual(receipt['status'], 'custody_written_pending_issuer_delivery') self.assertEqual((receipt['kv_version'], receipt['kv_cas']), (1, 0)) self.assertTrue(receipt['consumer_written'] and receipt['consumer_check_ok']) self.assertEqual(self.bao.payloads, [{'options': {'cas': 0}, 'data': {'CLIENT_SECRET': SYNTHETIC}}]) self.assertNotIn(['read', '-format=json', child.KV_DATA], self.bao.calls) self.assertEqual(self.consumer.read_text(), SYNTHETIC + '\n') self.assertEqual(self.consumer.stat().st_mode & 0o777, 0o600) self.assertEqual(self.consumer.parent.stat().st_mode & 0o777, 0o700) self.assertEqual(sorted(p.name for p in self.consumer.parent.iterdir()), [self.consumer.name]) def test_existing_custody_or_consumer_file_is_refused(self): self.bao.current = 1 code, receipt, _ = self.run_child() self.assertEqual(receipt['error'], 'existing_custody_requires_retry_receipt_or_rotation_approval') self.bao.current = None self.consumer.parent.mkdir(mode=0o700) self.consumer.write_text('old\n') code, receipt, _ = self.run_child() self.assertEqual(receipt['error'], 'consumer_file_exists_rotation_needs_separate_approval') self.assertEqual(self.bao.payloads, []) def test_shared_parent_is_refused(self): self.consumer.parent.mkdir(mode=0o755) os.chmod(self.consumer.parent, 0o755) code, receipt, _ = self.run_child() self.assertEqual(receipt['error'], 'consumer_parent_must_be_owned_0700_directory') self.assertEqual(self.bao.payloads, []) def test_failed_kv_write_writes_no_consumer_file(self): self.bao.fail_write = True code, receipt, _ = self.run_child() self.assertEqual((code, receipt['kv_written'], receipt['consumer_written']), (1, False, False)) self.assertFalse(self.consumer.exists()) self.assertEqual(list(self.consumer.parent.iterdir()), []) def test_consumer_failure_cleans_temp_and_retry_preserves_version(self): with patch.object(child.os, 'rename', side_effect=OSError('synthetic')): code, first, path = self.run_child() self.assertEqual((code, first['kv_written'], first['kv_version'], first['consumer_written']), (1, True, 1, False)) self.assertTrue(first['consumer_temp_removed']) self.assertEqual(list(self.consumer.parent.iterdir()), []) code, receipt, _ = self.run_child() self.assertEqual(receipt['error'], 'existing_custody_requires_retry_receipt_or_rotation_approval') code, second, _ = self.run_child('--prior-receipt', str(path)) self.assertEqual((code, second['kv_version'], second['kv_cas'], second['retry_of_kv_version']), (0, 2, 1, 1)) self.assertEqual([p['options']['cas'] for p in self.bao.payloads], [0, 1]) self.bao.current = 3 Path(self.consumer).unlink() code, third, _ = self.run_child('--prior-receipt', str(path)) self.assertEqual(third['error'], 'kv_version_changed_since_prior_receipt') def test_failed_consumer_check_fails_receipt(self): bad = Path(self.tmp.name) / 'bad.py' bad.write_text('#!' + sys.executable + '\nimport sys\nprint(\'{"ok": false}\')\nsys.exit(1)\n') bad.chmod(0o700) self.checker = str(bad) code, receipt, _ = self.run_child() self.assertEqual((code, receipt['consumer_written'], receipt['consumer_check_ok'], receipt['error']), (1, True, False, 'consumer_check_failed')) def test_receipt_contains_only_metadata_keys(self): _, receipt, _ = self.run_child() self.assertTrue(set(receipt) <= { 'schema', 'ccr', 'status', 'phase', 'started_at', 'finished_at', 'credential_values_emitted', 'client_side_read_admitted', 'kv_written', 'consumer_written', 'approval_decision_id', 'ccr_sha256', 'kv_version', 'kv_request_id', 'kv_cas', 'consumer_check_ok', 'retry_of_kv_version', 'consumer_temp_removed', 'error'}) if __name__ == '__main__': unittest.main()