feat: verify live KeyCape custody and preserve versions on resume
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 2s

Assistant: codex
Assistant-Model: gpt-5.6-luna
Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
This commit is contained in:
codex 2026-09-09 02:18:12 +02:00
parent b7861de20e
commit c6dc4286e2
8 changed files with 382 additions and 57 deletions

View file

@ -1,6 +1,8 @@
"""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
@ -78,6 +80,7 @@ class CustodyExercise(unittest.TestCase):
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'):
@ -110,5 +113,26 @@ class CustodyExercise(unittest.TestCase):
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()