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 <noreply@anthropic.com>

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

@@ -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.

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)

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()