diff --git a/sso-mfa/k8s/keycape/test_registration_original_intent_reconcile.py b/sso-mfa/k8s/keycape/test_registration_original_intent_reconcile.py new file mode 100644 index 0000000..30e9844 --- /dev/null +++ b/sso-mfa/k8s/keycape/test_registration_original_intent_reconcile.py @@ -0,0 +1,280 @@ +"""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). + +Fixture-only. 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 original_reconciler(adapter, admitted, client, *, holder, authenticate, + observe, publish, revoke_session, reasons): + """Fixture owner reconciler: read-only, never adds, never removes. + + holder(original_attempt_id) -> [(evidence_ref, payload_bytes)] is the + authenticated original-attempt readback; observe() -> current secret. + Confirms creation only from one authenticated original created-only record + whose post-creation uid/resourceVersion and exact client entry still hold. + Anything else raises, so Warden keeps the step uncertain. + """ + confirmed = {} + + def reconcile(ctx): + try: + return confirm(ctx) + except Exception as exc: # Warden contains the message; keep it for fixtures + reasons.append(f'{type(exc).__name__}: {exc}') + raise + + def confirm(ctx): + o, attempt = ctx.original.invocation, ctx.attempt + try: + if (o.operation != 'run' or attempt.operation != 'reconcile' + or o.attempt_id == attempt.attempt_id + or not IDENTIFIER.fullmatch(o.attempt_id)): + raise Refused('original intent not separate from reconcile attempt') + for inv in (o, attempt): + if replace(inv, attempt_id='', operation='admission') != admitted: + raise Refused('invocation differs from admitted case') + if ctx.original.source != adapter.source: + raise Refused('original source differs') + try: + records = holder(o.attempt_id) + except Exception: + raise Refused('original-attempt holder unavailable') from None + if len(records) != 1: + raise Refused('original record missing or ambiguous') + ref, raw = records[0] + if authenticate(ref, raw) is not True: + raise Refused('original record unauthenticated') + record = json.loads(raw) + if (record.get('schema') != 'netkingdom.keycape-registration.attempt.v1' + or record.get('invocation') != json.loads(json.dumps(base.asdict(o))) + or record.get('source') != adapter.source): + raise Refused('original record belongs to another attempt/case/source/principal') + if record.get('outcome') != Outcome.SUCCESS.value or record.get('created') != ['keycape-client']: + raise Refused('original record is not created-only provenance') + secret = observe() + meta = secret['metadata'] + if {'uid': meta['uid'], 'resource_version': meta['resourceVersion']} != record['after']: + raise Refused('native revision changed since creation') + if a.helper.replacement(secret, client)[1] is not False: + raise Refused('exact client entry absent') + finally: + revocation = revoke_session(attempt) + evidence = publish(json.dumps({'schema': 'netkingdom.keycape-registration.reconcile.v1', + 'original': ref, 'attempt': attempt.attempt_id}, sort_keys=True).encode()) + result = OwnerResult(Outcome.SUCCESS, evidence, ('keycape-client',), True, revocation, + attempt_id=attempt.attempt_id, operation='reconcile') + confirmed[evidence] = (attempt, result) + return result + + def verify(inv, result): + if confirmed.get(getattr(result, 'evidence', None)) == (inv, result): + return True + return adapter.verify(inv, result) + + return replace(adapter, reconcile_original=reconcile, verify=verify) + + +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 = [] + fresh, admitted = self.kernel_fixture() + adapter = fresh.adapters[a.REFERENCE] + client = a.helper.client_from_case(base.yaml.safe_load(self.case_source(admitted))) + fresh.adapters[a.REFERENCE] = original_reconciler( + adapter, admitted[1], client, holder=self.holder, + authenticate=lambda ref, raw: self.evidence.get(ref) == raw, + observe=lambda: copy.deepcopy(self.secret), publish=self.publish, + revoke_session=self.revoke, reasons=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()