71607d277c
The custody refusal test now checks proposed copies instead of the committed CCRs, which are approved. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 132360@bnt-lap001 Assistant-Session: 769b1afd-4330-4c86-96b2-e510e26db42b
214 lines
10 KiB
Python
214 lines
10 KiB
Python
"""CCR-2026-0033 custody child: synthetic values and fakes only; no OpenBao, cluster or real file."""
|
|
import contextlib
|
|
import copy
|
|
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:
|
|
"""KV-v2 metadata/data behaviour for one path; refuses data reads outright."""
|
|
|
|
def __init__(self, current=None, fail_write=False, metadata=None):
|
|
self.current, self.fail_write, self.calls, self.payloads = current, fail_write, [], []
|
|
self.metadata = metadata or {'cas_required': False, 'max_versions': 0,
|
|
'delete_version_after': '0s', 'custom_metadata': None}
|
|
self.metadata_writes = []
|
|
|
|
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_DATA:
|
|
raise AssertionError('data path read attempted')
|
|
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())
|
|
body = dict(self.metadata, current_version=self.current,
|
|
versions={str(self.current): {'destroyed': False, 'deletion_time': ''}})
|
|
return result(stdout=json.dumps({'data': body}).encode())
|
|
if args[0] == 'write' and child.KV_METADATA in args:
|
|
self.metadata_writes.append(payload)
|
|
self.metadata.update(copy.deepcopy(payload))
|
|
return result()
|
|
if args[0] == 'write' and child.KV_DATA in args:
|
|
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]))
|
|
|
|
|
|
def approve(path, base, name, decision=DECISION):
|
|
ccr = yaml.safe_load(path.read_text())
|
|
ccr['status'] = 'approved'
|
|
ccr['openbao']['auth']['bound_claims_confirmed'] = True
|
|
ccr['review']['comments'] = [
|
|
{'at': '2026-10-06', 'reviewer': r, 'decision': 'approved',
|
|
'comment': 'custody child approved by decision ' + decision} for r in child.APPROVERS]
|
|
out = base / name
|
|
out.write_text(yaml.safe_dump(ccr, sort_keys=False))
|
|
return out, ccr
|
|
|
|
|
|
class CustodyChild(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
base = Path(self.tmp.name)
|
|
self.ccr, data = approve(child.CCR_PATH, base, 'CCR-2026-0033.yaml')
|
|
self.desired = data['openbao']['kv_metadata']
|
|
self.sibling, _ = approve(child.SIBLING_PATH, base, 'CCR-2026-0034.yaml')
|
|
self.bao = FakeBao()
|
|
self.patches = [patch.object(child, 'CCR_PATH', self.ccr), patch.object(child, 'SIBLING_PATH', self.sibling),
|
|
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), *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), ' '.join(argv)):
|
|
self.assertNotIn(SYNTHETIC, blob)
|
|
self.assertEqual(out.getvalue(), '')
|
|
self.assertEqual(oct(receipt.stat().st_mode & 0o777), oct(0o600))
|
|
return code, json.loads(text)
|
|
|
|
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 proposed_copy(self, name):
|
|
text = (ROOT / 'credential-change-requests' / name).read_text()
|
|
path = Path(self.tmp.name) / name
|
|
path.write_text(text.replace('\nstatus: approved\n', '\nstatus: proposed\n', 1))
|
|
return path
|
|
|
|
def test_proposed_ccrs_are_refused(self):
|
|
with patch.object(child, 'CCR_PATH', self.proposed_copy('CCR-2026-0033.yaml')):
|
|
code, receipt = self.run_child()
|
|
self.assertEqual((code, receipt['error']), (1, 'approved_ccr_required'))
|
|
with patch.object(child, 'SIBLING_PATH', self.proposed_copy('CCR-2026-0034.yaml')):
|
|
code, receipt = self.run_child()
|
|
self.assertEqual((code, receipt['error']), (1, 'approved_ccr_required'))
|
|
self.assertEqual(self.bao.calls, [])
|
|
|
|
def test_refuses_ambient_token_or_outside_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, [])
|
|
self.patches[3].stop()
|
|
try:
|
|
with patch.object(child.Path, 'home', lambda: Path('/x/.warden-attended-login/u')):
|
|
with patch.dict(os.environ, {}, clear=False):
|
|
os.environ.pop('BAO_TOKEN', None)
|
|
os.environ.pop('VAULT_TOKEN', None)
|
|
self.assertTrue(child.attended_envelope())
|
|
for var in ('BAO_TOKEN', 'VAULT_TOKEN'):
|
|
with patch.dict(os.environ, {var: 'synthetic-ambient'}):
|
|
self.assertFalse(child.attended_envelope())
|
|
code, receipt = self.run_child()
|
|
self.assertEqual(receipt['error'], 'attended_warden_envelope_required')
|
|
finally:
|
|
self.patches[3].start()
|
|
self.assertEqual(self.bao.calls, [])
|
|
|
|
def test_create_writes_kv_cas0_and_metadata(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'], receipt['kv_metadata']), (1, 0, 'applied'))
|
|
self.assertEqual(self.bao.payloads, [{'options': {'cas': 0}, 'data': {'CLIENT_SECRET': SYNTHETIC}}])
|
|
self.assertEqual(self.bao.metadata_writes, [self.desired])
|
|
self.assertTrue(self.bao.metadata['cas_required'])
|
|
self.assertEqual(self.bao.metadata['max_versions'], 2)
|
|
self.assertEqual(self.bao.metadata['custom_metadata']['ccr'], 'CCR-2026-0033')
|
|
self.assertEqual(sorted(p.name for p in Path(self.tmp.name).iterdir()),
|
|
['CCR-2026-0033.yaml', 'CCR-2026-0034.yaml', 'receipt-0.json'])
|
|
|
|
def test_existing_entry_requires_rotate_and_rotation_uses_current_cas(self):
|
|
self.bao.current = 1
|
|
self.bao.metadata.update(copy.deepcopy(self.desired))
|
|
code, receipt = self.run_child()
|
|
self.assertEqual((code, receipt['error']), (1, 'existing_custody_requires_rotate'))
|
|
self.assertEqual(self.bao.payloads, [])
|
|
code, receipt = self.run_child('--rotate')
|
|
self.assertEqual((code, receipt['kv_cas'], receipt['kv_version'], receipt['kv_metadata']), (0, 1, 2, 'unchanged'))
|
|
self.assertEqual(self.bao.metadata_writes, [])
|
|
|
|
def test_rotate_without_entry_is_refused(self):
|
|
code, receipt = self.run_child('--rotate')
|
|
self.assertEqual((code, receipt['error']), (1, 'rotation_requires_existing_custody'))
|
|
self.assertEqual(self.bao.payloads, [])
|
|
|
|
def test_failed_kv_write_records_no_version_and_no_metadata(self):
|
|
self.bao.fail_write = True
|
|
code, receipt = self.run_child()
|
|
self.assertEqual((code, receipt['kv_written'], receipt['error']), (1, False, 'command_failed'))
|
|
self.assertEqual(self.bao.metadata_writes, [])
|
|
|
|
def test_metadata_readback_mismatch_fails_but_records_written_version(self):
|
|
real = self.bao.__call__
|
|
bao = self.bao
|
|
|
|
def wrapper(args, **kw):
|
|
if args[0] == 'write' and child.KV_METADATA in args:
|
|
bao.calls.append(list(args))
|
|
return result()
|
|
return real(args, **kw)
|
|
with patch.object(child, 'bao', wrapper):
|
|
code, receipt = self.run_child()
|
|
self.assertEqual((code, receipt['kv_written'], receipt['kv_version'], receipt['error']),
|
|
(1, True, 1, 'kv_metadata_readback_mismatch'))
|
|
|
|
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', 'approval_decision_id', 'ccr_sha256',
|
|
'kv_version', 'kv_request_id', 'kv_cas', 'kv_metadata', 'error'})
|
|
self.assertFalse(any('consumer' in key for key in receipt))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|