From ec635393b68809336d3f8b8fc996a6b7f3258534 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 17 Oct 2026 07:55:36 +0200 Subject: [PATCH] NK-WP-0048-T29: Apps hand-over adapter source candidate Injection-only consumer verification and hand-over receipt seam with positive/refusal fixtures and created-only rollback. Canonical 1.1 default; 1.2 only by explicit version-root selection. Unexercised; Apps owner review before pinning remains a separate gate. Co-Authored-By: Claude Opus 5.5 Assistant: claude-code Assistant-Model: opus Assistant-Process: 903097@bnt-lap001 Assistant-Session: aa320c8f-c3fa-46e7-8541-0b58e1e45d23 --- sso-mfa/k8s/keycape/handover-adapter.md | 31 ++++ sso-mfa/k8s/keycape/handover-adapter.py | 159 +++++++++++++++++++ sso-mfa/k8s/keycape/test_handover_adapter.py | 142 +++++++++++++++++ 3 files changed, 332 insertions(+) create mode 100644 sso-mfa/k8s/keycape/handover-adapter.md create mode 100644 sso-mfa/k8s/keycape/handover-adapter.py create mode 100644 sso-mfa/k8s/keycape/test_handover_adapter.py diff --git a/sso-mfa/k8s/keycape/handover-adapter.md b/sso-mfa/k8s/keycape/handover-adapter.md new file mode 100644 index 0000000..c0d2bd1 --- /dev/null +++ b/sso-mfa/k8s/keycape/handover-adapter.md @@ -0,0 +1,31 @@ +# Apps hand-over adapter (source candidate, NK-WP-0048-T29) + +`handover-adapter.py` is an unexercised, injection-only seam for Warden's v1 +kernel, modelled on the T13 registration adapter. It covers step (f), +Apps-owned consumer verification and the hand-over receipt. + +- **Descriptor:** step `hand-over`, owner `apps`, actor `human`, lane + `apps-consumer-handover`, target `client_id`, effect `change`, allowed created + object `handover-receipt`, not retry-safe, no revocation scopes. Owner and lane + names are candidate values for Apps owner review and T30 composition. +- **Case binding:** canonical `oidc-browser-relying-party@1.1` is the default; + `@1.2` is reached only through an explicit `version_root='1.2'` selection. + No `--allow-unimplemented` admission exists. Case, descriptor and acceptance + pins are checked; `access_groups` must be non-empty, because 1.1 hand-over + requires the consumer to enforce them (`consumer-enforces-groups`). +- **Injected capabilities (required, never strings or case parameters):** + `verify_consumer(client_id, groups)` must return exactly `client_id`, + `enforces_groups=True`, the case's `groups` and `login_verified=True`; + `record_handover(client_id, observation_digest) -> (identifier, created)`; + `remove_handover(identifier) -> True`; `publish(bytes) -> identifier` and + `authenticate(identifier, bytes) -> True` for owner evidence. +- **Rollback is created-only:** only a receipt this process created and whose + evidence the owner authenticates may be removed. An existing receipt is + `no_change` and never rolled back; unconfirmed removal refuses and keeps the + creation. Process loss and ambiguity stay for owner reconciliation (T31/T15). + +Tests: `python3 -m pytest sso-mfa/k8s/keycape/test_handover_adapter.py` +(synthetic fixtures only; no consumer, store or credential is touched). + +Not supplied here: Apps owner source review before pinning, actual consumer +verifier and receipt store, admission, and any native action. diff --git a/sso-mfa/k8s/keycape/handover-adapter.py b/sso-mfa/k8s/keycape/handover-adapter.py new file mode 100644 index 0000000..960bcde --- /dev/null +++ b/sso-mfa/k8s/keycape/handover-adapter.py @@ -0,0 +1,159 @@ +"""Unexercised, injection-only Apps-owned hand-over seam for Warden's v1 kernel. + +Source candidate for NK-WP-0048-T29. No CLI, credentials, consumer access or +live receipt store is installed here. Consumer verification, hand-over receipt +recording/removal and owner evidence publication/authentication are mandatory +injected capabilities. Apps owner source review precedes any pinning. +""" +from __future__ import annotations + +from dataclasses import asdict, replace +import json +from pathlib import Path + +import yaml + +from warden.case_run import Adapter, Descriptor, Invocation, OwnerResult, contained_output, digest, IDENTIFIER +from warden.case_receipt_state import Outcome, Refused + +ROOT = Path(__file__).resolve().parent +SOURCE_FILES = ('handover-adapter.py',) +REFERENCE = 'net-kingdom/apps-handover@v1' +CANONICAL_TYPE = 'oidc-browser-relying-party@1.1' +# 1.2 is reachable only by explicit version-root selection, never by default. +SELECTABLE_TYPES = {'1.1': CANONICAL_TYPE, '1.2': 'oidc-browser-relying-party@1.2'} +RECEIPT = 'handover-receipt' + + +def source_digest(): + composition = {'files': {name: digest((ROOT / name).read_bytes()) for name in SOURCE_FILES}} + return digest(json.dumps(composition, sort_keys=True, separators=(',', ':')).encode()) + + +def descriptor(): + return Descriptor('warden.case.execution.v1', 'hand-over', 'apps', REFERENCE, + source_digest(), 'human', 'apps-consumer-handover', 'client_id', + 'change', (RECEIPT,), False, ()) + + +def _case(case_source, version_root): + case = yaml.safe_load(case_source) + spec = case.get('spec') if isinstance(case, dict) else None + params = spec.get('parameters') if isinstance(spec, dict) else None + if not isinstance(params, dict) or spec.get('type') != SELECTABLE_TYPES.get(version_root): + raise Refused('case type differs from selected version root') + client_id, groups = params.get('client_id'), params.get('access_groups') + if not isinstance(client_id, str) or not IDENTIFIER.fullmatch(client_id): + raise Refused('bounded client_id required') + groups = tuple(sorted(g.strip() for g in groups.split(','))) if isinstance(groups, str) else () + if not groups or not all(IDENTIFIER.fullmatch(g) for g in groups): + raise Refused('hand-over requires access_groups the consumer enforces') + return client_id, groups + + +def build(*, admitted: Invocation, case_source: bytes, case_acceptance: str, + verify_consumer, record_handover, remove_handover, publish, authenticate, + version_root: str = '1.1'): + """Bind one admitted case; return a closed kernel Adapter. + + verify_consumer(client_id, groups)->dict observes the consumer and must report + {'client_id', 'enforces_groups', 'groups', 'login_verified'} for exactly this + case. record_handover(client_id, observation_digest)->(identifier, created) + durably records the hand-over receipt; remove_handover(identifier) removes only + that receipt. publish/authenticate are owner evidence capabilities. + + Only a receipt verifiably created by this process may be rolled back. Process + loss, ambiguity or publication failure require owner reconciliation. + """ + reviewed = source_digest() + desc = descriptor() + if type(admitted) is not Invocation or not all( + isinstance(v, str) and IDENTIFIER.fullmatch(v) for v in + (admitted.step, admitted.owner, admitted.target, admitted.actor, + admitted.principal, admitted.lane)): + raise Refused('bounded invocation required') + if (admitted.step, admitted.owner, admitted.actor, admitted.lane) != ( + desc.step, desc.owner, desc.actor, desc.lane): + raise Refused('hand-over invocation boundary differs') + if admitted.operation != 'admission' or admitted.attempt_id: + raise Refused('expected admission invocation') + if type(case_source) is not bytes or len(case_source) > 262144 or digest(case_source) != admitted.pins.case: + raise Refused('case source pin differs') + desc_bytes = json.dumps(asdict(desc), sort_keys=True, separators=(',', ':')).encode() + if admitted.pins.adapter != digest(desc_bytes): + raise Refused('descriptor pin differs') + if not isinstance(case_acceptance, str) or digest(case_acceptance.encode()) != admitted.pins.acceptance: + raise Refused('acceptance evidence binding differs') + if not all(callable(x) for x in (verify_consumer, record_handover, remove_handover, publish, authenticate)): + raise Refused('owner verification, receipt and evidence capabilities required') + with contained_output(): + client_id, groups = _case(case_source, version_root) + if admitted.target != client_id: + raise Refused('case target differs') + records = {} + creation = None + + def binding(inv, operation): + if (replace(inv, attempt_id='', operation='admission') != admitted + or inv.operation != operation or not IDENTIFIER.fullmatch(inv.attempt_id)): + raise Refused('attempt binding differs') + if source_digest() != reviewed: + raise Refused('owner source changed') + + def finish(inv, outcome, objects, body): + payload = json.dumps({'schema': 'netkingdom.apps-handover.attempt.v1', + 'invocation': asdict(inv), 'source': reviewed, + 'outcome': outcome.value, 'created': objects, **body}, + sort_keys=True, separators=(',', ':')).encode() + with contained_output(): + evidence = publish(payload) + if not isinstance(evidence, str) or not IDENTIFIER.fullmatch(evidence) or evidence in records: + raise Refused('owner evidence identifier invalid or reused') + result = OwnerResult(outcome, evidence, objects, attempt_id=inv.attempt_id, operation=inv.operation) + records[evidence] = (inv, payload, result) + return result + + def invoke(inv): + nonlocal creation + binding(inv, 'run') + with contained_output(): + seen = verify_consumer(client_id, groups) + if not (isinstance(seen, dict) and set(seen) == {'client_id', 'enforces_groups', 'groups', 'login_verified'} + and seen['client_id'] == client_id and seen['enforces_groups'] is True + and seen['login_verified'] is True and tuple(sorted(seen['groups'] or ())) == groups): + raise Refused('consumer verification contract differs') + observation = digest(json.dumps(seen, sort_keys=True, separators=(',', ':')).encode()) + with contained_output(): + receipt = record_handover(client_id, observation) + if not (isinstance(receipt, tuple) and len(receipt) == 2 and isinstance(receipt[0], str) + and IDENTIFIER.fullmatch(receipt[0]) and type(receipt[1]) is bool): + raise Refused('hand-over receipt contract differs') + ref, created = receipt + objects = (RECEIPT,) if created else () + result = finish(inv, Outcome.SUCCESS if created else Outcome.NO_CHANGE, objects, + {'observation': observation, 'receipt': ref}) + if created: + creation = (inv, result, ref) + return result + + def rollback(inv, created): + nonlocal creation + binding(inv, 'rollback') + if created != (RECEIPT,) or creation is None or not verify(*creation[:2]): + raise Refused('no authenticated case-created hand-over receipt') + ref = creation[2] + with contained_output(): + removed = remove_handover(ref) + if removed is not True: + raise Refused('hand-over receipt removal unconfirmed') + creation = None + return finish(inv, Outcome.ROLLED_BACK, (), {'removed_receipt': ref}) + + def verify(inv, result): + record = records.get(result.evidence) if type(result) is OwnerResult else None + if record is None or record[0] != inv or record[2] != result: + return False + with contained_output(): + return authenticate(result.evidence, record[1]) is True + + return Adapter(REFERENCE, reviewed, invoke, verify, rollback) diff --git a/sso-mfa/k8s/keycape/test_handover_adapter.py b/sso-mfa/k8s/keycape/test_handover_adapter.py new file mode 100644 index 0000000..97f44ca --- /dev/null +++ b/sso-mfa/k8s/keycape/test_handover_adapter.py @@ -0,0 +1,142 @@ +"""Synthetic positive and refusal fixtures for the T29 hand-over candidate.""" +from dataclasses import asdict, replace +import importlib.util +import json +from pathlib import Path +import unittest + +from warden.case_run import Invocation, OwnerResult, digest +from warden.case_receipt_state import Outcome, Pins, Refused +import yaml + +ROOT = Path(__file__).parent +spec = importlib.util.spec_from_file_location('handover_adapter', ROOT / 'handover-adapter.py') +a = importlib.util.module_from_spec(spec) +spec.loader.exec_module(a) + +PARAMS = {'client_id': 'forgejo', 'access_groups': 'forge-users,forge-admins'} + + +def case_bytes(type_=a.CANONICAL_TYPE, params=PARAMS): + return yaml.safe_dump({'spec': {'type': type_, 'parameters': params}}).encode() + + +class HandoverTests(unittest.TestCase): + def setUp(self): + self.case = case_bytes() + desc_bytes = json.dumps(asdict(a.descriptor()), sort_keys=True, separators=(',', ':')).encode() + pins = Pins(digest(self.case), digest(b'type'), digest(desc_bytes), digest(b'catalog'), + digest(b'case-acceptance:test'), digest(b'run sheet')) + self.admitted = Invocation('hand-over', 'apps', 'forgejo', 'human', 'founder', + 'apps-consumer-handover', pins) + self.inv = replace(self.admitted, attempt_id='attempt-one', operation='run') + self.rb = replace(self.admitted, attempt_id='attempt-two', operation='rollback') + self.evidence, self.receipts = {}, set() + self.observed = {'client_id': 'forgejo', 'enforces_groups': True, + 'groups': ['forge-admins', 'forge-users'], 'login_verified': True} + self.kw = dict(admitted=self.admitted, case_source=self.case, case_acceptance='case-acceptance:test', + verify_consumer=lambda c, g: dict(self.observed), record_handover=self.record, + remove_handover=self.remove, publish=self.publish, + authenticate=lambda ref, payload: self.evidence.get(ref) == payload) + + def publish(self, payload): + ref = 'owner-evidence:' + str(len(self.evidence)) + self.evidence[ref] = payload + return ref + + def record(self, client_id, observation): + ref = 'handover:' + client_id + created = ref not in self.receipts + self.receipts.add(ref) + return ref, created + + def remove(self, ref): + self.receipts.discard(ref) + return True + + def test_positive_handover_verify_and_created_only_rollback(self): + adapter = a.build(**self.kw) + result = adapter.invoke(self.inv) + self.assertEqual((result.outcome, result.created), (Outcome.SUCCESS, (a.RECEIPT,))) + self.assertTrue(adapter.verify(self.inv, result)) + self.assertNotIn(b'synthetic', self.evidence[result.evidence]) + back = adapter.rollback(self.rb, (a.RECEIPT,)) + self.assertEqual(back.outcome, Outcome.ROLLED_BACK) + self.assertEqual(self.receipts, set()) + + def test_existing_receipt_is_no_change_and_not_rollbackable(self): + self.receipts.add('handover:forgejo') + adapter = a.build(**self.kw) + result = adapter.invoke(self.inv) + self.assertEqual((result.outcome, result.created), (Outcome.NO_CHANGE, ())) + with self.assertRaises(Refused): + adapter.rollback(self.rb, (a.RECEIPT,)) + self.assertEqual(self.receipts, {'handover:forgejo'}) + + def test_rollback_refuses_unauthenticated_creation(self): + adapter = a.build(**{**self.kw, 'authenticate': lambda ref, payload: False}) + adapter.invoke(self.inv) + with self.assertRaises(Refused): + adapter.rollback(self.rb, (a.RECEIPT,)) + self.assertEqual(self.receipts, {'handover:forgejo'}) + + def test_consumer_verification_refusals(self): + for change in ({'enforces_groups': False}, {'login_verified': False}, {'groups': ['forge-users']}, + {'client_id': 'other'}, {'extra': 1}): + self.observed = {**dict(self.observed), **change} + with self.subTest(change=change), self.assertRaises(Refused): + a.build(**self.kw).invoke(self.inv) + self.observed = {k: v for k, v in self.observed.items() if k != 'extra'} + self.setUp() + self.assertEqual(self.receipts, set()) + + def test_canonical_default_and_explicit_selection_only(self): + with self.assertRaises(Refused): + a.build(**{**self.kw, 'version_root': '1.0'}) + case12 = case_bytes(a.SELECTABLE_TYPES['1.2']) + pins = replace(self.admitted.pins, case=digest(case12)) + admitted12 = replace(self.admitted, pins=pins) + with self.assertRaises(Refused): # 1.2 never reached by default + a.build(**{**self.kw, 'admitted': admitted12, 'case_source': case12}) + a.build(**{**self.kw, 'admitted': admitted12, 'case_source': case12, 'version_root': '1.2'}) + + def test_binding_refusals(self): + bad = [{'case_acceptance': 'other'}, {'case_source': self.case + b'\n#'}, + {'admitted': replace(self.admitted, owner='net-kingdom')}, + {'admitted': replace(self.admitted, target='other')}, + {'admitted': replace(self.admitted, pins=replace(self.admitted.pins, adapter=digest(b'x')))}, + {'publish': None}] + for change in bad: + with self.subTest(change=list(change)), self.assertRaises(Refused): + a.build(**{**self.kw, **change}) + no_groups = case_bytes(params={'client_id': 'forgejo'}) + with self.assertRaises(Refused): + a.build(**{**self.kw, 'case_source': no_groups, + 'admitted': replace(self.admitted, pins=replace(self.admitted.pins, case=digest(no_groups)))}) + + def test_attempt_and_contract_refusals(self): + adapter = a.build(**self.kw) + for inv in (replace(self.inv, principal='other'), replace(self.inv, operation='rollback'), + replace(self.inv, attempt_id='')): + with self.subTest(inv=inv), self.assertRaises(Refused): + adapter.invoke(inv) + for receipt in (('handover:forgejo', 'yes'), 'handover:forgejo', ('', True)): + with self.subTest(receipt=receipt), self.assertRaises(Refused): + a.build(**{**self.kw, 'record_handover': lambda c, o, r=receipt: r}).invoke(self.inv) + reuse = a.build(**{**self.kw, 'publish': lambda p: 'owner-evidence:0'}) + reuse.invoke(self.inv) + with self.assertRaises(Refused): # evidence identifier reuse + reuse.invoke(replace(self.inv, attempt_id='attempt-three')) + self.assertFalse(adapter.verify(self.inv, OwnerResult(Outcome.SUCCESS, 'forged'))) + + def test_removal_unconfirmed_keeps_creation(self): + adapter = a.build(**{**self.kw, 'remove_handover': lambda ref: None}) + adapter.invoke(self.inv) + with self.assertRaises(Refused): + adapter.rollback(self.rb, (a.RECEIPT,)) + with self.assertRaises(Refused): + adapter.rollback(self.rb, ()) + + +if __name__ == '__main__': + unittest.main()