tegwick
061461e4a4
build() accepts paired holder/observe read-only capabilities and installs Adapter.reconcile_original; verify accepts the published reconcile record. Fixture now exercises the adapter's own path. No add replay, no undo after restart; missing/ambiguous/changed state stays uncertain. Holder choice remains an owner gate. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 903097@bnt-lap001 Assistant-Session: aa320c8f-c3fa-46e7-8541-0b58e1e45d23
221 lines
11 KiB
Python
221 lines
11 KiB
Python
"""NK-WP-0048-T31: restart and lost-receipt fixtures for the T13 registration
|
|
adapter against Warden's original-intent interface (ops-warden a45cf47:
|
|
OriginalIntent, Journal.original_intent, Adapter.reconcile_original).
|
|
|
|
T15 source: reconciliation now runs through the adapter's own
|
|
reconcile_original (build(holder=..., observe=...)). The original-attempt holder below is the test's own in-memory
|
|
evidence store standing in for an owner-authenticated holder; Warden's intent
|
|
row is Warden's dispatch record, not created-only provenance. Selecting the
|
|
actual authenticated holder remains a T15 owner gate. Synthetic Kubernetes
|
|
objects only; nothing here is native admission or a receipt.
|
|
"""
|
|
from dataclasses import replace
|
|
import copy
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from warden.case_run import OwnerResult, IDENTIFIER
|
|
from warden.case_receipt_state import Outcome, Refused
|
|
|
|
ROOT = Path(__file__).parent
|
|
spec = importlib.util.spec_from_file_location('registration_adapter_tests',
|
|
ROOT / 'test_keycape_registration_adapter.py')
|
|
base = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(base)
|
|
a = base.a
|
|
|
|
|
|
def capture_reasons(adapter, reasons):
|
|
"""Fixture-only: Warden contains the owner's refusal, so keep the message."""
|
|
reconcile = adapter.reconcile_original
|
|
|
|
def wrapped(ctx):
|
|
try:
|
|
return reconcile(ctx)
|
|
except Exception as exc:
|
|
reasons.append(f'{type(exc).__name__}: {exc}')
|
|
raise
|
|
|
|
return replace(adapter, reconcile_original=wrapped)
|
|
|
|
|
|
class OriginalIntentReconcileTests(unittest.TestCase):
|
|
setUp = base.RegistrationTests.setUp
|
|
publish = base.RegistrationTests.publish
|
|
revoke = base.RegistrationTests.revoke
|
|
kube = base.RegistrationTests.kube
|
|
kernel_fixture = base.RegistrationTests.kernel_fixture
|
|
|
|
reasons = None
|
|
|
|
def holder(self, attempt_id):
|
|
if getattr(self, 'holder_down', False):
|
|
raise ConnectionError('synthetic holder outage')
|
|
return [(ref, raw) for ref, raw in self.evidence.items()
|
|
if json.loads(raw).get('invocation', {}).get('attempt_id') == attempt_id]
|
|
|
|
def restart(self, executor):
|
|
"""Fresh process: new journal handle, new adapter (empty records), readmission."""
|
|
executor.journal.db.close()
|
|
self.reasons = []
|
|
# The adapter's own reconcile_original path, given test-only holder/readback.
|
|
self.kw = dict(self.kw, holder=self.holder, observe=lambda: copy.deepcopy(self.secret))
|
|
fresh, admitted = self.kernel_fixture()
|
|
fresh.adapters[a.REFERENCE] = capture_reasons(fresh.adapters[a.REFERENCE], self.reasons)
|
|
return fresh, admitted
|
|
|
|
def case_source(self, admitted):
|
|
root = ROOT.parents[2]
|
|
case = base.yaml.safe_load((root / 'security-cases/forgejo-netkingdom-oidc.yaml').read_bytes())
|
|
case['spec']['status'] = 'materialized'
|
|
return base.yaml.safe_dump(case).encode()
|
|
|
|
def lose_receipt(self):
|
|
"""Owner completed and published, then the kernel lost the result."""
|
|
executor, admitted = self.kernel_fixture()
|
|
adapter = executor.adapters[a.REFERENCE]
|
|
def lost(inv):
|
|
adapter.invoke(inv)
|
|
raise TimeoutError('synthetic lost receipt')
|
|
executor.adapters[a.REFERENCE] = replace(adapter, invoke=lost)
|
|
self.assertEqual(executor.run(*admitted).last.outcome, Outcome.UNCERTAIN)
|
|
self.assertEqual(len(self.effects), 1)
|
|
return executor, admitted
|
|
|
|
def state(self, executor, admitted):
|
|
binding = f'{admitted[1].step}:{admitted[1].target}'
|
|
return executor.journal.load(binding, admitted[2]).last
|
|
|
|
def test_restart_reconciles_from_separate_original_intent_without_replay(self):
|
|
executor, _ = self.lose_receipt()
|
|
executor, admitted = self.restart(executor)
|
|
seen = []
|
|
rec = executor.adapters[a.REFERENCE]
|
|
executor.adapters[a.REFERENCE] = replace(
|
|
rec, reconcile_original=lambda ctx: (seen.append(ctx), rec.reconcile_original(ctx))[1])
|
|
state = executor.reconcile(*admitted)
|
|
self.assertEqual(state.last.outcome, Outcome.SUCCESS)
|
|
self.assertEqual(state.last.created, ('keycape-client',))
|
|
(ctx,) = seen
|
|
self.assertEqual(ctx.original.invocation.operation, 'run')
|
|
self.assertEqual(ctx.attempt.operation, 'reconcile')
|
|
self.assertNotEqual(ctx.original.invocation.attempt_id, ctx.attempt.attempt_id)
|
|
self.assertEqual(ctx.original.invocation.principal, 'founder')
|
|
self.assertEqual(len(self.effects), 1) # no add replay
|
|
self.assertEqual(self.revocations[-1], ctx.attempt) # reconcile session revoked
|
|
# T13 deliberately refuses post-restart undo even after reconciliation.
|
|
self.assertEqual(executor.rollback(*admitted).last.outcome, Outcome.ROLLBACK_UNCERTAIN)
|
|
self.assertEqual(len(self.effects), 1)
|
|
|
|
def assert_stays_uncertain(self, executor, admitted, match=None):
|
|
with self.assertRaises(Refused) as caught:
|
|
executor.reconcile(*admitted)
|
|
self.assertRegex(self.reasons[-1] if self.reasons else str(caught.exception), match or '.')
|
|
self.assertEqual(self.state(executor, admitted).outcome, Outcome.UNCERTAIN)
|
|
self.assertEqual(len(self.effects), 1) # no add replay, no undo
|
|
with self.assertRaises(Refused):
|
|
executor.run(*admitted, retry=True)
|
|
self.assertEqual(len(self.effects), 1)
|
|
|
|
def test_holder_unavailable_refuses_and_stays_uncertain(self):
|
|
executor, admitted = self.restart(self.lose_receipt()[0])
|
|
self.holder_down = True
|
|
self.assert_stays_uncertain(executor, admitted)
|
|
|
|
def test_missing_warden_intent_refuses_before_owner(self):
|
|
executor, admitted = self.restart(self.lose_receipt()[0])
|
|
executor.journal.db.execute('DELETE FROM intents')
|
|
before = len(self.revocations)
|
|
self.assert_stays_uncertain(executor, admitted, 'original intent record missing')
|
|
self.assertEqual(len(self.revocations), before)
|
|
|
|
def test_missing_or_ambiguous_owner_record_stays_uncertain(self):
|
|
for mutate in ('drop', 'duplicate'):
|
|
with self.subTest(mutate=mutate):
|
|
self.tearDown_and_setUp()
|
|
executor, admitted = self.restart(self.lose_receipt()[0])
|
|
(ref,) = list(self.evidence)
|
|
if mutate == 'drop':
|
|
del self.evidence[ref]
|
|
else:
|
|
self.evidence[ref + '-copy'] = self.evidence[ref]
|
|
self.assert_stays_uncertain(executor, admitted, 'missing or ambiguous')
|
|
|
|
def test_lost_readback_without_owner_record_stays_uncertain(self):
|
|
executor, admitted = self.kernel_fixture()
|
|
original = self.kube
|
|
def lost(args, payload=None):
|
|
if self.effects and args[:5] == ['-n', 'sso', 'get', 'secret', 'keycape-config']:
|
|
raise RuntimeError('synthetic')
|
|
return original(args, payload)
|
|
with patch.object(a.helper.lane, 'kubectl', side_effect=lost):
|
|
self.assertEqual(executor.run(*admitted).last.outcome, Outcome.UNCERTAIN)
|
|
executor, admitted = self.restart(executor)
|
|
self.assert_stays_uncertain(executor, admitted, 'missing or ambiguous')
|
|
|
|
def test_wrong_attempt_case_source_principal_records_refuse(self):
|
|
fields = {'attempt': ('invocation', 'attempt_id', 'attempt-other'),
|
|
'case': ('invocation', 'pins', None),
|
|
'principal': ('invocation', 'principal', 'someone-else'),
|
|
'source': ('source', None, 'sha256:' + '0' * 64)}
|
|
for name, (key, sub, value) in fields.items():
|
|
with self.subTest(name=name):
|
|
self.tearDown_and_setUp()
|
|
executor, admitted = self.restart(self.lose_receipt()[0])
|
|
(ref,) = list(self.evidence)
|
|
record = json.loads(self.evidence[ref])
|
|
if key == 'source':
|
|
record['source'] = value
|
|
elif sub == 'pins':
|
|
record['invocation']['pins']['case'] = 'sha256:' + '1' * 64
|
|
else:
|
|
record['invocation'][sub] = value
|
|
raw = json.dumps(record, sort_keys=True, separators=(',', ':')).encode()
|
|
self.evidence[ref] = raw # still "authenticated" by the fixture holder
|
|
if name == 'attempt':
|
|
self.assert_stays_uncertain(executor, admitted, 'missing or ambiguous')
|
|
else:
|
|
self.assert_stays_uncertain(executor, admitted, 'another attempt')
|
|
|
|
def test_warden_refuses_mismatched_case_principal_and_source(self):
|
|
executor, admitted = self.restart(self.lose_receipt()[0])
|
|
binding = f'{admitted[1].step}:{admitted[1].target}'
|
|
inv, source = admitted[1], executor.adapters[a.REFERENCE].source
|
|
for current, src, match in (
|
|
(replace(inv, principal='someone-else'), source, 'principal mismatch'),
|
|
(replace(inv, pins=replace(inv.pins, case='sha256:' + '1' * 64)), source, 'pins mismatch'),
|
|
(inv, 'other-source', 'source mismatch')):
|
|
with self.subTest(match=match), self.assertRaisesRegex(Refused, match):
|
|
executor.journal.original_intent(binding, current, src)
|
|
|
|
def test_no_change_original_or_changed_revision_stays_uncertain(self):
|
|
for case in ('no_change', 'revision', 'entry'):
|
|
with self.subTest(case=case):
|
|
self.tearDown_and_setUp()
|
|
executor, admitted = self.restart(self.lose_receipt()[0])
|
|
if case == 'no_change':
|
|
(ref,) = list(self.evidence)
|
|
record = json.loads(self.evidence[ref])
|
|
record.update(outcome='no_change', created=[])
|
|
self.evidence[ref] = json.dumps(record, sort_keys=True,
|
|
separators=(',', ':')).encode()
|
|
match = 'not created-only'
|
|
elif case == 'revision':
|
|
self.secret['metadata']['resourceVersion'] = '99'
|
|
match = 'revision changed'
|
|
else:
|
|
with patch.object(a.helper, 'replacement', return_value=('x', True)):
|
|
self.assert_stays_uncertain(executor, admitted, 'exact client entry absent')
|
|
continue
|
|
self.assert_stays_uncertain(executor, admitted, match)
|
|
|
|
def tearDown_and_setUp(self):
|
|
self.doCleanups()
|
|
self.setUp()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|