Revise KeyCape secrets-engine-openbao custody: OpenBao-only, operator reader CCR-2026-0034 (source only)
Founder direction 2026-10-06 and ops-mason plan keycape-secrets-engine-openbao-client-custody (658cdbc, MASON-WP-0009): the client secret lives only in OpenBao. - CCR-2026-0033: consumer file and secrets-engine D1-D4 removed; KV metadata declared (cas_required, max_versions=2, delete_version_after=0s, custom_metadata); target.domain infotech; live order and roll-forward rollback revised; status stays proposed. - New sibling CCR-2026-0034 + policy: attended OIDC operator reader on auth/netkingdom, read data path + revoke-self only, 5m/4 uses, no default policy, bound groups=net-kingdom-admins and /assurance/level=aal2 (KeyCape 7d96ee4). Proposed, bound claims unconfirmed (founder item for both readers). - credential-change.py: OIDC roles carry token_max_ttl/explicit_max_ttl/no_default_policy/ num_uses/type and bound_claims_type (string only); token_revoke_self policy option; validated openbao.kv_metadata. Tests added. - Custody child: consumer write and check-secret-file removed; cas=0 create / --rotate cas=<current>; compare-before-write metadata put; refusals unchanged. - Founder runbook (keycape/founder-runbook.md), window docs, RPF-WP-0035-T09 (waits on A1/A2/A3). 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
This commit is contained in:
codex
parent
9e23eaa067
commit
4a284f0a3b
11 changed files with 734 additions and 307 deletions
|
|
@ -453,6 +453,65 @@ class CredentialChangeTests(unittest.TestCase):
|
|||
["openid", "profile", "email", "groups"],
|
||||
)
|
||||
|
||||
def test_oidc_operator_reader_carries_token_bounds_and_assurance(self) -> None:
|
||||
path = REPO_DIR / "credential-change-requests/CCR-2026-0034.yaml"
|
||||
ccr, errors, _warnings = credential_change.validate_ccr(path)
|
||||
self.assertEqual(errors, [])
|
||||
payload = credential_change.auth_payload(ccr)
|
||||
self.assertEqual(payload["token_num_uses"], 4)
|
||||
self.assertEqual((payload["ttl"], payload["token_max_ttl"], payload["token_explicit_max_ttl"]),
|
||||
("5m", "5m", "5m"))
|
||||
self.assertIs(payload["token_no_default_policy"], True)
|
||||
self.assertEqual(payload["token_type"], "service")
|
||||
self.assertEqual(payload["bound_claims_type"], "string")
|
||||
self.assertEqual(payload["bound_claims"],
|
||||
{"groups": ["net-kingdom-admins"], "/assurance/level": "aal2"})
|
||||
body = credential_change.generated_policy_hcl(ccr)
|
||||
self.assertIn('path "auth/token/revoke-self"', body)
|
||||
self.assertNotIn("lookup-self", body)
|
||||
self.assertNotIn("/metadata/", body)
|
||||
self.assertEqual(credential_change.policy_artifact_diff(ccr)["status"], "matches")
|
||||
self.assertEqual(credential_change.applier_policy_violations(ccr), [])
|
||||
|
||||
def test_oidc_token_bounds_and_kv_metadata_are_validated(self) -> None:
|
||||
import yaml
|
||||
source = yaml.safe_load((REPO_DIR / "credential-change-requests/CCR-2026-0034.yaml").read_text())
|
||||
cases = [
|
||||
(("openbao", "auth", "token_num_uses"), 0, "token_num_uses must be a positive integer"),
|
||||
(("openbao", "auth", "token_num_uses"), True, "token_num_uses must be a positive integer"),
|
||||
(("openbao", "auth", "token_type"), "root", "token_type must be one of"),
|
||||
(("openbao", "auth", "bound_claims_type"), "glob", "glob matching is not admitted"),
|
||||
(("openbao", "auth", "token_max_ttl"), "5 minutes", "token_max_ttl must be a TTL string"),
|
||||
(("openbao", "kv_metadata"), {"max_versions": 0}, "max_versions must be a positive integer"),
|
||||
(("openbao", "kv_metadata"), {"delete_version_after": "never"}, "delete_version_after"),
|
||||
(("openbao", "kv_metadata"), {"versions": 2}, "unknown keys"),
|
||||
(("openbao", "kv_metadata"), {"custom_metadata": {"a": 1}}, "custom_metadata must map"),
|
||||
]
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
for keys, value, message in cases:
|
||||
ccr = yaml.safe_load(yaml.safe_dump(source))
|
||||
target = ccr
|
||||
for key in keys[:-1]:
|
||||
target = target[key]
|
||||
target[keys[-1]] = value
|
||||
path = Path(tmp) / "CCR-2026-0034.yaml"
|
||||
path.write_text(yaml.safe_dump(ccr))
|
||||
_ccr, errors, _ = credential_change.validate_ccr(path)
|
||||
self.assertTrue(any(message in error for error in errors), (keys, value, errors))
|
||||
|
||||
def test_keycape_secrets_engine_openbao_ccr_declares_kv_metadata_without_consumer_copy(self) -> None:
|
||||
path = REPO_DIR / "credential-change-requests/CCR-2026-0033.yaml"
|
||||
ccr, errors, _ = credential_change.validate_ccr(path)
|
||||
self.assertEqual(errors, [])
|
||||
self.assertNotIn("consumer_copy", ccr)
|
||||
self.assertEqual(ccr["target"]["domain"], "infotech")
|
||||
self.assertEqual(ccr["status"], "proposed")
|
||||
meta = ccr["openbao"]["kv_metadata"]
|
||||
self.assertEqual((meta["cas_required"], meta["max_versions"], meta["delete_version_after"]),
|
||||
(True, 2, "0s"))
|
||||
self.assertEqual(meta["custom_metadata"]["readers"], "eso-sso-issuer,oidc-secrets-engine-operator")
|
||||
self.assertIs(ccr["openbao"]["auth"]["bound_claims_confirmed"], False)
|
||||
|
||||
def test_apply_plan_refuses_unapproved_ccr(self) -> None:
|
||||
with self.assertRaises(SystemExit):
|
||||
credential_change.command_apply_plan(type("Args", (), {"ref": str(self.issue_core)})())
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
"""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
|
||||
|
|
@ -25,19 +26,31 @@ def result(code=0, stdout=b'', stderr=b''):
|
|||
|
||||
|
||||
class FakeBao:
|
||||
def __init__(self, current=None, fail_write=False):
|
||||
"""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())
|
||||
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:
|
||||
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')
|
||||
|
|
@ -46,30 +59,27 @@ class FakeBao:
|
|||
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)
|
||||
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.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, 'CONSUMER_FILE', self.consumer),
|
||||
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:
|
||||
|
|
@ -85,16 +95,17 @@ class CustodyChild(unittest.TestCase):
|
|||
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]
|
||||
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)):
|
||||
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), receipt
|
||||
return code, json.loads(text)
|
||||
|
||||
def test_refuses_without_or_with_wrong_decision(self):
|
||||
with self.assertRaises(SystemExit), contextlib.redirect_stderr(io.StringIO()):
|
||||
|
|
@ -102,92 +113,95 @@ class CustodyChild(unittest.TestCase):
|
|||
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)
|
||||
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):
|
||||
def test_committed_proposed_ccrs_are_refused(self):
|
||||
with patch.object(child, 'CCR_PATH', ROOT / 'credential-change-requests/CCR-2026-0033.yaml'):
|
||||
code, receipt, _ = self.run_child()
|
||||
code, receipt = self.run_child()
|
||||
self.assertEqual((code, receipt['error']), (1, 'approved_ccr_required'))
|
||||
with patch.object(child, 'SIBLING_PATH', ROOT / 'credential-change-requests/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_outside_attended_envelope(self):
|
||||
def test_refuses_ambient_token_or_outside_envelope(self):
|
||||
with patch.object(child, 'attended_envelope', lambda: False):
|
||||
code, receipt, _ = self.run_child()
|
||||
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_provision_writes_kv_and_consumer_in_one_process(self):
|
||||
code, receipt, _ = self.run_child()
|
||||
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']), (1, 0))
|
||||
self.assertTrue(receipt['consumer_written'] and receipt['consumer_check_ok'])
|
||||
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.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])
|
||||
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_custody_or_consumer_file_is_refused(self):
|
||||
def test_existing_entry_requires_rotate_and_rotation_uses_current_cas(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.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_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):
|
||||
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['consumer_written']), (1, False, False))
|
||||
self.assertFalse(self.consumer.exists())
|
||||
self.assertEqual(list(self.consumer.parent.iterdir()), [])
|
||||
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_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_metadata_readback_mismatch_fails_but_records_written_version(self):
|
||||
real = self.bao.__call__
|
||||
bao = self.bao
|
||||
|
||||
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 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()
|
||||
_, 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'})
|
||||
'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__':
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue