coulomb/net-kingdom
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

NK-WP-0048-T29: Apps hand-over adapter source candidate

Browse source 
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
This commit is contained in:
tegwick 2026-10-17 07:55:36 +02:00
parent 2019e823f0
commit ec635393b6
3 changed files with 332 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

31
sso-mfa/k8s/keycape/handover-adapter.md Normal file
View file

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

159
sso-mfa/k8s/keycape/handover-adapter.py Normal file
View file

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

142
sso-mfa/k8s/keycape/test_handover_adapter.py Normal file
View file

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