From 6fbce73b87efeaaea19c6e3f1a084afa5e02c6b7 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 17 Oct 2026 07:55:04 +0200 Subject: [PATCH 1/7] NK-WP-0048-T27: bind-env adapter source candidate (references only) 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/bind-env-adapter.md | 26 +++++ sso-mfa/k8s/keycape/bind-env-adapter.py | 110 +++++++++++++++++++ sso-mfa/k8s/keycape/test_bind_env_adapter.py | 103 +++++++++++++++++ 3 files changed, 239 insertions(+) create mode 100644 sso-mfa/k8s/keycape/bind-env-adapter.md create mode 100644 sso-mfa/k8s/keycape/bind-env-adapter.py create mode 100644 sso-mfa/k8s/keycape/test_bind_env_adapter.py diff --git a/sso-mfa/k8s/keycape/bind-env-adapter.md b/sso-mfa/k8s/keycape/bind-env-adapter.md new file mode 100644 index 0000000..df78b7e --- /dev/null +++ b/sso-mfa/k8s/keycape/bind-env-adapter.md @@ -0,0 +1,26 @@ +# Bind-env adapter source candidate (NK-WP-0048-T27) + +Status: **unexercised source candidate**. Not reviewed, pinned, admitted or run. + +`bind-env-adapter.py` adds one `env` entry to the `keycape` container of the +`sso/keycape` Deployment, binding the helper-derived name +(`KEYCAPE__CLIENT_SECRET`) to `valueFrom.secretKeyRef` on the delivered +secret reference (`keycape--client`, key `client-secret`). It handles +references only: a binding carrying `value`, or an existing entry of that name +bound to anything else (including a literal), is refused. + +- Case parsing reuses `keycape-client.py` (`client_from_case`, `env_binding`), so + canonical `oidc-browser-relying-party@1.1` stays the default; public clients + and `allow_unimplemented` are refused. +- `read`/`write` are injected capabilities; `write` is compare-and-swap on the + observed `resourceVersion`. No kubectl, credential or secret value is used. +- An identical existing binding is `no-change` and is not rollbackable. +- Rollback removes only the entry this process created, and only while the + Deployment is still at the revision that creation produced; otherwise owner + reconciliation is required. + +Tests: `python3 -m pytest -q test_bind_env_adapter.py` (synthetic only). + +Not supplied: Warden `Adapter`/descriptor wrapping and evidence publication (see +T13 `keycape-registration-adapter.py`), NetKingdom runtime owner source review +before pinning, and any native action. Those remain separate gates. diff --git a/sso-mfa/k8s/keycape/bind-env-adapter.py b/sso-mfa/k8s/keycape/bind-env-adapter.py new file mode 100644 index 0000000..f1b2d53 --- /dev/null +++ b/sso-mfa/k8s/keycape/bind-env-adapter.py @@ -0,0 +1,110 @@ +"""Unexercised source candidate: bind one issuer env reference, never a value (NK-WP-0048-T27). + +Given an admitted oidc-browser-relying-party case (canonical 1.1 by default via the +maintained helper), add exactly one `env` entry to the `keycape` container of the +issuer Deployment: name from the helper, valueFrom.secretKeyRef to the delivered +secret *reference*. No CLI, kubectl, credential, or value handling lives here; +read/write are injected capabilities with compare-and-swap on resourceVersion. +Only an entry this process created, still byte-identical, may be rolled back. +""" +from __future__ import annotations + +import copy +import importlib.util +from pathlib import Path + +ROOT = Path(__file__).resolve().parent +REFERENCE = 'net-kingdom/keycape-bind-env@v1' +_spec = importlib.util.spec_from_file_location('bind_env_owner_helper', ROOT / 'keycape-client.py') +helper = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(helper) + + +class Refused(Exception): + pass + + +def _container(deployment): + if (deployment.get('kind') != 'Deployment' + or deployment.get('metadata', {}).get('name') != 'keycape' + or deployment.get('metadata', {}).get('namespace') != 'sso'): + raise Refused('issuer deployment sso/keycape required') + found = [c for c in deployment['spec']['template']['spec'].get('containers', []) + if c.get('name') == 'keycape'] + if len(found) != 1: + raise Refused('exactly one keycape container required') + return found[0] + + +def binding_for_case(case, *, allow_unimplemented=False): + if allow_unimplemented: + raise Refused('allow-unimplemented admission is not supported') + client = helper.client_from_case(case) + if client.get('clientType') != 'confidential': + raise Refused('public client has no secret reference to bind') + return helper.env_binding(client) + + +def _check_reference_only(entry): + if 'value' in entry or set(entry) != {'name', 'valueFrom'} or set(entry['valueFrom']) != {'secretKeyRef'}: + raise Refused('reference-only env binding required') + + +def plan(deployment, binding): + """Return (new_deployment, created). Never reads or emits a secret value.""" + _check_reference_only(binding) + new = copy.deepcopy(deployment) + env = _container(new).setdefault('env', []) + same = [e for e in env if e.get('name') == binding['name']] + if len(same) > 1: + raise Refused('duplicate env name present') + if same: + if same[0] != binding: + raise Refused('env name already bound to a different source') + return new, False + env.append(copy.deepcopy(binding)) + return new, True + + +def build(*, case, expected_resource_version, read, write, allow_unimplemented=False): + """read()->deployment dict; write(deployment, expected_rv)->new rv (CAS, raises on conflict).""" + binding = binding_for_case(case, allow_unimplemented=allow_unimplemented) + if not (callable(read) and callable(write)): + raise Refused('read/write capabilities required') + if not isinstance(expected_resource_version, str) or not expected_resource_version: + raise Refused('observed resourceVersion required') + state = {'created_rv': None} + + def run(): + current = read() + rv = current['metadata'].get('resourceVersion') + if rv != expected_resource_version: + raise Refused('deployment revision differs from observed') + new, created = plan(current, binding) + if not created: + return {'outcome': 'no-change', 'created': (), 'values_emitted': False} + after = write(new, rv) + back = read() + if back['metadata'].get('resourceVersion') != after or plan(back, binding)[1]: + raise Refused('readback ambiguous; owner reconciliation required') + state['created_rv'] = after + return {'outcome': 'success', 'created': (binding['name'],), 'values_emitted': False} + + def rollback(): + if state['created_rv'] is None: + raise Refused('no case-created binding to roll back') + current = read() + rv = current['metadata'].get('resourceVersion') + if rv != state['created_rv']: + raise Refused('deployment changed since creation; owner reconciliation required') + new = copy.deepcopy(current) + env = _container(new)['env'] + matches = [i for i, e in enumerate(env) if e == binding] + if len(matches) != 1: + raise Refused('created binding not found unchanged') + del env[matches[0]] + write(new, rv) + state['created_rv'] = None + return {'outcome': 'rolled-back', 'created': (), 'values_emitted': False} + + return REFERENCE, run, rollback diff --git a/sso-mfa/k8s/keycape/test_bind_env_adapter.py b/sso-mfa/k8s/keycape/test_bind_env_adapter.py new file mode 100644 index 0000000..0812acc --- /dev/null +++ b/sso-mfa/k8s/keycape/test_bind_env_adapter.py @@ -0,0 +1,103 @@ +"""Synthetic-only tests for the T27 bind-env candidate; no cluster, no values.""" +import copy +import importlib.util +from pathlib import Path +import unittest + +ROOT = Path(__file__).parent +spec = importlib.util.spec_from_file_location('bind_env_adapter', ROOT / 'bind-env-adapter.py') +a = importlib.util.module_from_spec(spec) +spec.loader.exec_module(a) +spec = importlib.util.spec_from_file_location('bind_env_fixture', ROOT / 'test_keycape_client.py') +f = importlib.util.module_from_spec(spec) +spec.loader.exec_module(f) + +CASE = {'spec': {'type': 'oidc-browser-relying-party@1.1', 'parameters': f.FORGEJO_PARAMS}} + + +def deployment(env=None): + return {'kind': 'Deployment', 'metadata': {'name': 'keycape', 'namespace': 'sso', 'resourceVersion': '1'}, + 'spec': {'template': {'spec': {'containers': [ + {'name': 'keycape', 'env': list(env or [{'name': 'KEYCAPE_CONFIG', 'value': '/etc/keycape/config.yaml'}])}]}}}} + + +class Store: + def __init__(self, d): + self.d, self.writes = d, [] + + def read(self): + return copy.deepcopy(self.d) + + def write(self, new, rv): + if rv != self.d['metadata']['resourceVersion']: + raise RuntimeError('conflict') + self.d = copy.deepcopy(new) + self.d['metadata']['resourceVersion'] = str(int(rv) + 1) + self.writes.append(self.d) + return self.d['metadata']['resourceVersion'] + + +class BindEnvTests(unittest.TestCase): + def test_binding_is_reference_only(self): + b = a.binding_for_case(CASE) + self.assertEqual(b['name'], 'KEYCAPE_FORGEJO_CLIENT_SECRET') + self.assertEqual(b['valueFrom']['secretKeyRef'], {'name': 'keycape-forgejo-client', 'key': 'client-secret'}) + self.assertNotIn('value', b) + + def test_run_then_created_only_rollback(self): + s = Store(deployment()) + ref, run, rollback = a.build(case=CASE, expected_resource_version='1', read=s.read, write=s.write) + self.assertEqual(ref, a.REFERENCE) + r = run() + self.assertEqual((r['outcome'], r['created'], r['values_emitted']), ('success', ('KEYCAPE_FORGEJO_CLIENT_SECRET',), False)) + self.assertEqual(len(s.d['spec']['template']['spec']['containers'][0]['env']), 2) + self.assertEqual(rollback()['outcome'], 'rolled-back') + self.assertEqual(s.d['spec']['template']['spec']['containers'][0]['env'], deployment()['spec']['template']['spec']['containers'][0]['env']) + with self.assertRaises(a.Refused): + rollback() + + def test_existing_identical_binding_is_no_change_and_not_rollbackable(self): + s = Store(deployment([a.binding_for_case(CASE)])) + _, run, rollback = a.build(case=CASE, expected_resource_version='1', read=s.read, write=s.write) + self.assertEqual(run()['outcome'], 'no-change') + self.assertEqual(s.writes, []) + with self.assertRaises(a.Refused): + rollback() + + def test_refusals(self): + other = dict(a.binding_for_case(CASE), valueFrom={'secretKeyRef': {'name': 'x', 'key': 'client-secret'}}) + literal = {'name': 'KEYCAPE_FORGEJO_CLIENT_SECRET', 'value': 'synthetic'} + for env in ([other], [literal]): + s = Store(deployment(env)) + _, run, _ = a.build(case=CASE, expected_resource_version='1', read=s.read, write=s.write) + with self.assertRaises(a.Refused): + run() + self.assertEqual(s.writes, []) + s = Store(deployment()) + _, run, _ = a.build(case=CASE, expected_resource_version='0', read=s.read, write=s.write) + with self.assertRaises(a.Refused): + run() + with self.assertRaises(a.Refused): + a.build(case=CASE, expected_resource_version='1', read=s.read, write=s.write, allow_unimplemented=True) + public = {'spec': {'type': 'oidc-browser-relying-party@1.1', + 'parameters': dict(f.FORGEJO_PARAMS, client_type='public', mfa_required='false')}} + with self.assertRaises(a.Refused): + a.binding_for_case(public) + with self.assertRaises(Exception): + a.binding_for_case({'spec': {'type': 'other@1', 'parameters': f.FORGEJO_PARAMS}}) + with self.assertRaises(a.Refused): + a.plan(dict(deployment(), metadata={'name': 'keycape', 'namespace': 'other'}), a.binding_for_case(CASE)) + with self.assertRaises(a.Refused): + a.plan(deployment(), dict(a.binding_for_case(CASE), value='synthetic')) + + def test_rollback_refused_after_drift(self): + s = Store(deployment()) + _, run, rollback = a.build(case=CASE, expected_resource_version='1', read=s.read, write=s.write) + run() + s.d['metadata']['resourceVersion'] = '9' + with self.assertRaises(a.Refused): + rollback() + + +if __name__ == '__main__': + unittest.main() From 1b60f6e9c6f83ff278ed352648617016cfe1b432 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 17 Oct 2026 07:55:24 +0200 Subject: [PATCH 2/7] NK-WP-0048-T26: deliver-secret adapter source candidate Injection-only, value-free Platform delivery seam with exact metadata-only receipt contract, created-only rollback and synthetic positive/refusal tests. Unexercised; Platform owner source review and native action stay separate gates. 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/deliver-secret-adapter.md | 44 +++++ sso-mfa/k8s/keycape/deliver-secret-adapter.py | 158 ++++++++++++++++++ .../keycape/test_deliver_secret_adapter.py | 130 ++++++++++++++ 3 files changed, 332 insertions(+) create mode 100644 sso-mfa/k8s/keycape/deliver-secret-adapter.md create mode 100644 sso-mfa/k8s/keycape/deliver-secret-adapter.py create mode 100644 sso-mfa/k8s/keycape/test_deliver_secret_adapter.py diff --git a/sso-mfa/k8s/keycape/deliver-secret-adapter.md b/sso-mfa/k8s/keycape/deliver-secret-adapter.md new file mode 100644 index 0000000..a7e9444 --- /dev/null +++ b/sso-mfa/k8s/keycape/deliver-secret-adapter.md @@ -0,0 +1,44 @@ +# Deliver-secret adapter source candidate (NK-WP-0048-T26) + +Status: **source candidate, unexercised.** Not reviewed, not pinned, not admitted. +Modelled on the T13 registration adapter (`keycape-registration-adapter.py`). + +## What it is + +`deliver-secret-adapter.py` is an injection-only Warden v1 kernel adapter for the +Platform-owned delivery seam (ESO/custody). Descriptor: step `deliver-secret`, +owner `railiance-platform`, lane `platform-secret-delivery`, target parameter +`client_id`, allowed created object `delivered-secret`, `retry_safe: false`, +revocation scope `parent-session`. + +It binds one admitted invocation to one Secret reference (namespace, name, +custody *reference*, key names). It never reads, mints or templates a value. +Delivery, removal, evidence publication/authentication and parent-session +revocation are injected owner capabilities; the module itself has no CLI and +calls no kubectl, OpenBao or registry. + +## Value-free evidence contract + +Owner receipts must have exactly the fields `schema, operation, namespace, name, +custody_reference, key_names, values_emitted, changed, before, after`, with +`values_emitted: false` and `before`/`after` limited to `{uid, resource_version}`. +Any extra field (for example a value) is refused before evidence is published. +The parent session is revoked even when the owner capability fails. + +## Rollback: created-only + +Only a delivery this process verified as created (`before: null`, `changed: true`, +evidence authenticated by the owner) may be rolled back, and rollback must remove +exactly the created revision. A pre-existing Secret is `NO_CHANGE` and is never +removed. Failures are not converted to retry-safe outcomes. + +## Tests + +`PYTHONPATH=~/ops-warden/src python3 -m pytest -q test_deliver_secret_adapter.py` +— synthetic positive and refusal fixtures only. + +## Not supplied by this candidate + +Canonical 1.1 stays the default; no `--allow-unimplemented` admission. Platform +owner source review before pinning, the real ESO/custody capability +implementations, and any native action remain separate owner gates. diff --git a/sso-mfa/k8s/keycape/deliver-secret-adapter.py b/sso-mfa/k8s/keycape/deliver-secret-adapter.py new file mode 100644 index 0000000..408ccc0 --- /dev/null +++ b/sso-mfa/k8s/keycape/deliver-secret-adapter.py @@ -0,0 +1,158 @@ +"""Unexercised, injection-only deliver-secret seam (NK-WP-0048-T26) for Warden's v1 kernel. + +Platform-owned delivery evidence (ESO/custody), value-free. No CLI, credential, +kubectl, OpenBao or registry access lives here: delivery, removal, evidence +publication/authentication and parent-session revocation are injected owner +capabilities. Receipts must match an exact metadata-only field set, so a +capability that returns a value (or any unexpected field) is refused. +""" +from __future__ import annotations + +from dataclasses import asdict, replace +import json +import re +from pathlib import Path + +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 = ('deliver-secret-adapter.py',) +REFERENCE = 'net-kingdom/deliver-secret@v1' +CREATED = 'delivered-secret' +NAME = re.compile(r'[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?') +RECEIPT_FIELDS = {'schema', 'operation', 'namespace', 'name', 'custody_reference', + 'key_names', 'values_emitted', 'changed', 'before', 'after'} + + +def source_digest(): + files = {name: digest((ROOT / name).read_bytes()) for name in SOURCE_FILES} + return digest(json.dumps({'files': files}, sort_keys=True, separators=(',', ':')).encode()) + + +def descriptor(): + return Descriptor('warden.case.execution.v1', 'deliver-secret', 'railiance-platform', + REFERENCE, source_digest(), 'human', 'platform-secret-delivery', + 'client_id', 'change', (CREATED,), False, ('parent-session',)) + + +def _revision(value): + return (isinstance(value, dict) and set(value) == {'uid', 'resource_version'} + and all(isinstance(v, str) and 0 < len(v) <= 192 for v in value.values())) + + +def build(*, admitted: Invocation, case_acceptance: str, namespace: str, name: str, + custody_reference: str, key_names: tuple, deliver, remove, publish, + authenticate, revoke_session): + """Bind one admitted case to one delivered Secret reference; return a kernel Adapter. + + deliver(inv, binding)->receipt and remove(inv, binding, after)->receipt are + owner capabilities (e.g. ESO sync with readback); receipts carry metadata only. + Only a delivery verified as created in this process may be rolled back. + """ + 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('delivery invocation boundary differs') + if admitted.operation != 'admission' or admitted.attempt_id: + raise Refused('expected admission invocation') + 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) and 0 < len(case_acceptance) <= 192 + and digest(case_acceptance.encode()) == admitted.pins.acceptance): + raise Refused('acceptance evidence binding differs') + if not all(isinstance(v, str) and NAME.fullmatch(v) for v in (namespace, name)): + raise Refused('bounded secret reference required') + if not (isinstance(custody_reference, str) and 0 < len(custody_reference) <= 192 + and IDENTIFIER.fullmatch(custody_reference.replace('/', '-'))): + raise Refused('bounded custody reference required') + if (type(key_names) is not tuple or not key_names or len(set(key_names)) != len(key_names) + or not all(isinstance(k, str) and re.fullmatch(r'[A-Za-z0-9._-]{1,64}', k) for k in key_names)): + raise Refused('bounded key names required') + if not all(callable(x) for x in (deliver, remove, publish, authenticate, revoke_session)): + raise Refused('owner delivery, evidence and revocation capabilities required') + binding_ref = {'namespace': namespace, 'name': name, 'custody_reference': custody_reference, + 'key_names': list(key_names), 'client_id': admitted.target} + records = {} + creation = None + + def check(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 execute(inv, operation, created=()): + nonlocal creation + try: + with contained_output(): + check(inv, 'run' if operation == 'deliver' else 'rollback') + if operation == 'deliver': + receipt = deliver(inv, dict(binding_ref)) + else: + if created != (CREATED,) or creation is None or not verify(*creation): + raise Refused('no authenticated case-created delivery') + prior = json.loads(records[creation[1].evidence][1])['after'] + receipt = remove(inv, dict(binding_ref), dict(prior)) + if not (isinstance(receipt, dict) and set(receipt) == RECEIPT_FIELDS + and receipt['schema'] == 'netkingdom.deliver-secret.v1' + and receipt['operation'] == operation + and (receipt['namespace'], receipt['name'], receipt['custody_reference']) + == (namespace, name, custody_reference) + and receipt['key_names'] == list(key_names) + and receipt['values_emitted'] is False + and type(receipt['changed']) is bool): + raise Refused('owner delivery receipt contract differs') + if operation == 'deliver': + if not _revision(receipt['after']) or not ( + receipt['before'] is None or _revision(receipt['before'])): + raise Refused('owner revision contract differs') + if receipt['changed'] != (receipt['before'] is None): + raise Refused('pre-existing secret must not be claimed as created') + elif not (receipt['changed'] is True and receipt['after'] is None + and receipt['before'] == prior): + raise Refused('rollback must remove exactly the created revision') + changed = receipt['changed'] + outcome = (Outcome.ROLLED_BACK if operation == 'remove' else + Outcome.SUCCESS if changed else Outcome.NO_CHANGE) + objects = (CREATED,) if operation == 'deliver' and changed else () + finally: + with contained_output(): + revocation = revoke_session(inv) + if not isinstance(revocation, str) or not IDENTIFIER.fullmatch(revocation): + raise Refused('session revocation evidence missing') + payload = json.dumps({'schema': 'netkingdom.deliver-secret.attempt.v1', + 'invocation': asdict(inv), 'source': reviewed, 'outcome': outcome.value, + 'created': objects, 'binding': binding_ref, 'before': receipt['before'], + 'after': receipt['after'], 'session_revocation': revocation}, + 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, True, revocation, + attempt_id=inv.attempt_id, operation=inv.operation) + records[evidence] = (inv, payload, result) + if objects: + creation = (inv, result) + elif operation == 'remove': + creation = None + return result + + 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, lambda inv: execute(inv, 'deliver'), + verify, lambda inv, created: execute(inv, 'remove', created)) diff --git a/sso-mfa/k8s/keycape/test_deliver_secret_adapter.py b/sso-mfa/k8s/keycape/test_deliver_secret_adapter.py new file mode 100644 index 0000000..1b3f8a5 --- /dev/null +++ b/sso-mfa/k8s/keycape/test_deliver_secret_adapter.py @@ -0,0 +1,130 @@ +"""Synthetic, value-free fixtures for the T26 deliver-secret seam. No live calls.""" +from dataclasses import asdict, replace +import importlib.util +import json +from pathlib import Path +import unittest + +from warden.case_run import Invocation, digest +from warden.case_receipt_state import Outcome, Pins, Refused + +ROOT = Path(__file__).parent +spec = importlib.util.spec_from_file_location('deliver_secret_adapter', ROOT / 'deliver-secret-adapter.py') +a = importlib.util.module_from_spec(spec) +spec.loader.exec_module(a) + +KEYS = ('client-secret',) +AFTER = {'uid': 'uid-1', 'resource_version': '3'} + + +class DeliverSecretTests(unittest.TestCase): + def setUp(self): + desc = json.dumps(asdict(a.descriptor()), sort_keys=True, separators=(',', ':')).encode() + pins = Pins(digest(b'case'), digest(b'type'), digest(desc), digest(b'catalog'), + digest(b'case-acceptance:test'), digest(b'run sheet')) + self.admitted = Invocation('deliver-secret', 'railiance-platform', 'forgejo', 'human', + 'founder', 'platform-secret-delivery', 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.calls, self.revoked = {}, [], [] + self.before = None + self.kw = dict(admitted=self.admitted, case_acceptance='case-acceptance:test', + namespace='sso', name='keycape-client-forgejo', + custody_reference='kv/keycape/forgejo', key_names=KEYS, + deliver=self.deliver, remove=self.remove, publish=self.publish, + authenticate=lambda ref, p: self.evidence.get(ref) == p, + revoke_session=lambda inv: self.revoked.append(inv) or 'revocation-' + inv.attempt_id) + + def receipt(self, operation, changed, before, after, **extra): + r = {'schema': 'netkingdom.deliver-secret.v1', 'operation': operation, 'namespace': 'sso', + 'name': 'keycape-client-forgejo', 'custody_reference': 'kv/keycape/forgejo', + 'key_names': list(KEYS), 'values_emitted': False, 'changed': changed, + 'before': before, 'after': after} + r.update(extra) + return r + + def deliver(self, inv, binding): + self.calls.append(('deliver', binding)) + return self.receipt('deliver', self.before is None, self.before, + self.before or dict(AFTER)) + + def remove(self, inv, binding, after): + self.calls.append(('remove', after)) + return self.receipt('remove', True, after, None) + + def publish(self, payload): + ref = 'owner-evidence-' + str(len(self.evidence)) + self.evidence[ref] = payload + return ref + + def test_created_delivery_verifies_and_rolls_back(self): + adapter = a.build(**self.kw) + result = adapter.invoke(self.inv) + self.assertEqual((result.outcome, result.created), (Outcome.SUCCESS, (a.CREATED,))) + self.assertTrue(adapter.verify(self.inv, result)) + self.assertNotIn(b'secret-value', b''.join(self.evidence.values())) + undone = adapter.rollback(self.rb, (a.CREATED,)) + self.assertEqual(undone.outcome, Outcome.ROLLED_BACK) + self.assertEqual(self.calls[-1], ('remove', AFTER)) + self.assertEqual(len(self.revoked), 2) + + def test_pre_existing_secret_is_no_change_and_never_rolled_back(self): + self.before = dict(AFTER) + 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.CREATED,)) + self.assertNotIn('remove', [c[0] for c in self.calls]) + + def test_receipt_carrying_value_is_refused_and_session_still_revoked(self): + self.deliver = lambda inv, b: self.receipt('deliver', True, None, dict(AFTER), data='secret-value') + self.kw['deliver'] = self.deliver + with self.assertRaises(Refused): + a.build(**self.kw).invoke(self.inv) + self.assertEqual(len(self.revoked), 1) + self.assertEqual(self.evidence, {}) + + def test_values_emitted_or_wrong_reference_refused(self): + for bad in (dict(values_emitted=True), dict(name='other'), dict(key_names=['x'])): + self.kw['deliver'] = lambda inv, b, bad=bad: self.receipt('deliver', True, None, dict(AFTER), **bad) + with self.assertRaises(Refused): + a.build(**self.kw).invoke(self.inv) + + def test_build_refusals(self): + cases = [dict(namespace='SSO!'), dict(key_names=['client-secret']), dict(key_names=()), + dict(case_acceptance='other'), dict(publish=None), + dict(admitted=replace(self.admitted, lane='other-lane')), + dict(admitted=replace(self.admitted, pins=replace(self.admitted.pins, adapter=digest(b'x'))))] + for change in cases: + with self.subTest(change=change), self.assertRaises(Refused): + a.build(**{**self.kw, **change}) + + def test_attempt_binding_and_reused_evidence_refused(self): + adapter = a.build(**self.kw) + with self.assertRaises(Refused): + adapter.invoke(replace(self.inv, target='other')) + self.kw['publish'] = lambda payload: 'same-ref' + adapter = a.build(**self.kw) + adapter.invoke(self.inv) + with self.assertRaises(Refused): + adapter.invoke(replace(self.inv, attempt_id='attempt-three')) + + def test_unverified_or_foreign_creation_cannot_roll_back(self): + self.kw['authenticate'] = lambda ref, p: False + adapter = a.build(**self.kw) + adapter.invoke(self.inv) + with self.assertRaises(Refused): + adapter.rollback(self.rb, (a.CREATED,)) + adapter = a.build(**{**self.kw, 'authenticate': lambda ref, p: True}) + with self.assertRaises(Refused): + adapter.rollback(self.rb, (a.CREATED,)) + + def test_descriptor_keeps_non_retry_safe_and_parent_session_scope(self): + d = a.descriptor() + self.assertEqual((d.step, d.owner, d.retry_safe, d.revocation_scopes), + ('deliver-secret', 'railiance-platform', False, ('parent-session',))) + + +if __name__ == '__main__': + unittest.main() From 558c5bcbf6de38d64772d3bf5eb13c81345b7309 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 17 Oct 2026 07:55:29 +0200 Subject: [PATCH 3/7] NK-WP-0048-T25: mint-secret adapter source candidate (value-free) Injection-only seam for the type 1.2 mint-secret step: closed value-free custody receipts, created-only rollback by exact version, parent-session and broker-grant revocation. Source candidate only; no owner review or admission. 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/mint-secret-adapter.md | 23 +++ sso-mfa/k8s/keycape/mint-secret-adapter.py | 146 ++++++++++++++++++ .../k8s/keycape/test_mint_secret_adapter.py | 137 ++++++++++++++++ 3 files changed, 306 insertions(+) create mode 100644 sso-mfa/k8s/keycape/mint-secret-adapter.md create mode 100644 sso-mfa/k8s/keycape/mint-secret-adapter.py create mode 100644 sso-mfa/k8s/keycape/test_mint_secret_adapter.py diff --git a/sso-mfa/k8s/keycape/mint-secret-adapter.md b/sso-mfa/k8s/keycape/mint-secret-adapter.md new file mode 100644 index 0000000..febb615 --- /dev/null +++ b/sso-mfa/k8s/keycape/mint-secret-adapter.md @@ -0,0 +1,23 @@ +# Mint-secret adapter (source candidate, NK-WP-0048-T25) + +`mint-secret-adapter.py` is an unexercised, injection-only seam for the type 1.2 +`mint-secret` step (owner `railiance-platform`, lane +`keycape-client-secret-generate`, `value_handling: generated_in_custody`). It is +modelled on the T13 registration adapter. + +- **No secret values.** The secret is generated inside custody by an injected + Platform `mint(path)` capability. Receipts must be exactly + `{schema, path, operation, version, created, values_emitted: false}`; any other + field (including a value) is refused before evidence is published. +- **Value-free evidence.** Published evidence carries invocation, source digest, + custody path, version number and both revocation identifiers. +- **Created-only rollback.** Only a version this adapter minted and can + authenticate is destroyed, by exact version; otherwise rollback is refused. +- **Revocation.** Descriptor effect `mint` with scopes `parent-session` and + `broker-grant`; both are revoked after every attempt, including failures. +- Confidential clients only; case, descriptor, acceptance and attempt pins bound + as in T13. Not retry-safe. + +Not supplied here: Platform owner source review before pinning, Warden admission, +type 1.2 selection (canonical 1.1 stays default; no `--allow-unimplemented`), and +any native mint. Tests: `PYTHONPATH=~/ops-warden/src python3 -m pytest test_mint_secret_adapter.py`. diff --git a/sso-mfa/k8s/keycape/mint-secret-adapter.py b/sso-mfa/k8s/keycape/mint-secret-adapter.py new file mode 100644 index 0000000..25f6115 --- /dev/null +++ b/sso-mfa/k8s/keycape/mint-secret-adapter.py @@ -0,0 +1,146 @@ +"""Unexercised, injection-only mint-secret seam (type 1.2 step `mint-secret`). + +Source candidate for Platform owner review; not pinned, admitted or run. The +client secret is generated inside custody by an injected Platform capability. +This module never receives, reads, templates or returns a secret value: owner +receipts are closed value-free metadata and any other field is refused. +Canonical 1.1 stays the default; nothing here selects 1.2 or bypasses admission. +""" +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 = ('mint-secret-adapter.py',) +REFERENCE = 'net-kingdom/keycape-mint-secret@v1' +RECEIPT_SCHEMA = 'railiance-platform.custody-mint.v1' +RECEIPT_KEYS = {'schema', 'path', 'operation', 'version', 'created', 'values_emitted'} +CREATED = ('custody-secret-version',) + + +def source_digest(): + composition = {'files': {n: digest((ROOT / n).read_bytes()) for n in SOURCE_FILES}} + return digest(json.dumps(composition, sort_keys=True, separators=(',', ':')).encode()) + + +def descriptor(): + return Descriptor('warden.case.execution.v1', 'mint-secret', 'railiance-platform', + REFERENCE, source_digest(), 'warden', 'keycape-client-secret-generate', + 'client_id', 'mint', CREATED, False, ('parent-session', 'broker-grant')) + + +def custody_path(client_id): + return 'platform/workloads/keycape/' + client_id + + +def build(*, admitted: Invocation, case_source: bytes, case_acceptance: str, + mint, destroy, publish, authenticate, revoke_session, revoke_grant): + """Bind one admitted confidential-client case; return a closed kernel Adapter. + + mint(path)->receipt generates a new secret version in custody, value never returned. + destroy(path, version)->receipt removes exactly that version (created-only rollback). + publish(bytes)->identifier, authenticate(identifier, bytes)->True: owner evidence. + revoke_session(inv)/revoke_grant(inv)->identifier end this attempt's parent + session and broker grant. All are capabilities, never command strings. + """ + 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('mint 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 (mint, destroy, publish, authenticate, revoke_session, revoke_grant)): + raise Refused('owner custody, evidence and revocation capabilities required') + with contained_output(): + case = yaml.safe_load(case_source) + params = (case or {}).get('spec', {}).get('parameters', {}) if isinstance(case, dict) else {} + if params.get('client_type') != 'confidential': + raise Refused('mint-secret applies only to confidential clients') + if params.get('client_id') != admitted.target: + raise Refused('case target differs') + path = custody_path(admitted.target) + records, creation = {}, None + + def check(receipt, operation, version=None): + if (type(receipt) is not dict or set(receipt) != RECEIPT_KEYS + or receipt['schema'] != RECEIPT_SCHEMA or receipt['path'] != path + or receipt['operation'] != operation or receipt['values_emitted'] is not False + or type(receipt['created']) is not bool + or type(receipt['version']) is not int or receipt['version'] < 1 + or (version is not None and receipt['version'] != version)): + raise Refused('owner value-free receipt contract differs') + + def execute(inv, operation, created=()): + nonlocal creation + receipt = None + try: + with contained_output(): + if (replace(inv, attempt_id='', operation='admission') != admitted + or inv.operation != ('run' if operation == 'mint' else 'rollback') + or not IDENTIFIER.fullmatch(inv.attempt_id)): + raise Refused('attempt binding differs') + if source_digest() != reviewed: + raise Refused('owner source changed') + if operation == 'mint': + receipt = mint(path) + check(receipt, 'mint') + if not receipt['created']: + raise Refused('custody did not create a version') + else: + if created != CREATED or creation is None or not verify(*creation): + raise Refused('no authenticated case-created version') + version = json.loads(records[creation[1].evidence][1])['version'] + receipt = destroy(path, version) + check(receipt, 'destroy', version) + finally: + with contained_output(): + session, grant = revoke_session(inv), revoke_grant(inv) + for ref in (session, grant): + if not isinstance(ref, str) or not IDENTIFIER.fullmatch(ref): + raise Refused('revocation evidence missing') + outcome = Outcome.SUCCESS if operation == 'mint' else Outcome.ROLLED_BACK + objects = CREATED if operation == 'mint' else () + payload = json.dumps({'schema': 'netkingdom.keycape-mint-secret.attempt.v1', + 'invocation': asdict(inv), 'source': reviewed, 'outcome': outcome.value, + 'created': objects, 'path': path, 'version': receipt['version'], + 'values_emitted': False, 'session_revocation': session, 'grant_revocation': grant}, + 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, True, session, grant, + attempt_id=inv.attempt_id, operation=inv.operation) + records[evidence] = (inv, payload, result) + creation = (inv, result) if operation == 'mint' else None + return result + + 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, lambda inv: execute(inv, 'mint'), + verify, lambda inv, created: execute(inv, 'rollback', created)) diff --git a/sso-mfa/k8s/keycape/test_mint_secret_adapter.py b/sso-mfa/k8s/keycape/test_mint_secret_adapter.py new file mode 100644 index 0000000..c50f78d --- /dev/null +++ b/sso-mfa/k8s/keycape/test_mint_secret_adapter.py @@ -0,0 +1,137 @@ +"""Synthetic custody capabilities only; no secret value exists anywhere in these tests.""" +from dataclasses import asdict, replace +import importlib.util +import json +from pathlib import Path +import unittest + +from warden.case_run import Invocation, digest +from warden.case_receipt_state import Outcome, Pins, Refused +import yaml + +ROOT = Path(__file__).parent +spec = importlib.util.spec_from_file_location('mint_secret_adapter', ROOT / 'mint-secret-adapter.py') +a = importlib.util.module_from_spec(spec) +spec.loader.exec_module(a) +PATH = 'platform/workloads/keycape/forgejo' + + +def receipt(op, version=3, **over): + r = {'schema': a.RECEIPT_SCHEMA, 'path': PATH, 'operation': op, 'version': version, + 'created': op == 'mint', 'values_emitted': False} + r.update(over) + return r + + +class MintSecretTests(unittest.TestCase): + def setUp(self): + self.case = yaml.safe_dump({'spec': {'type': 'oidc-browser-relying-party@1.2', 'parameters': + {'client_id': 'forgejo', 'client_type': 'confidential'}}}).encode() + desc = json.dumps(asdict(a.descriptor()), sort_keys=True, separators=(',', ':')).encode() + pins = Pins(digest(self.case), digest(b'type'), digest(desc), digest(b'catalog'), + digest(b'case-acceptance:test'), digest(b'run sheet')) + self.admitted = Invocation('mint-secret', 'railiance-platform', 'forgejo', 'warden', + 'founder', 'keycape-client-secret-generate', pins) + self.run = replace(self.admitted, attempt_id='attempt-one', operation='run') + self.back = replace(self.admitted, attempt_id='attempt-one', operation='rollback') + self.calls, self.evidence, self.revoked = [], {}, [] + self.mint_receipt = receipt('mint') + self.kw = dict(admitted=self.admitted, case_source=self.case, + case_acceptance='case-acceptance:test', mint=self.mint, destroy=self.destroy, + publish=self.publish, authenticate=lambda ref, p: self.evidence.get(ref) == p, + revoke_session=lambda inv: self.revoke('session', inv), + revoke_grant=lambda inv: self.revoke('grant', inv)) + + def mint(self, path): + self.calls.append(('mint', path)) + return self.mint_receipt + + def destroy(self, path, version): + self.calls.append(('destroy', path, version)) + return receipt('destroy', version) + + def publish(self, payload): + ref = 'owner-evidence:' + str(len(self.evidence)) + self.evidence[ref] = payload + return ref + + def revoke(self, kind, inv): + self.revoked.append((kind, inv.operation)) + return kind + '-revoked:' + inv.attempt_id + + def test_descriptor_is_mint_with_both_revocations(self): + d = a.descriptor() + self.assertEqual((d.effect, d.revocation_scopes, d.retry_safe), + ('mint', ('parent-session', 'broker-grant'), False)) + + def test_mint_records_value_free_evidence_and_verifies(self): + adapter = a.build(**self.kw) + result = adapter.invoke(self.run) + self.assertEqual((result.outcome, result.created), (Outcome.SUCCESS, a.CREATED)) + self.assertTrue(adapter.verify(self.run, result)) + payload = json.loads(self.evidence[result.evidence]) + self.assertEqual((payload['path'], payload['version'], payload['values_emitted']), (PATH, 3, False)) + self.assertEqual(self.revoked, [('session', 'run'), ('grant', 'run')]) + self.assertNotIn('value', json.dumps(payload).replace('values_emitted', '')) + + def test_created_only_rollback_destroys_exact_version(self): + adapter = a.build(**self.kw) + adapter.invoke(self.run) + result = adapter.rollback(self.back, a.CREATED) + self.assertEqual(result.outcome, Outcome.ROLLED_BACK) + self.assertEqual(self.calls[-1], ('destroy', PATH, 3)) + with self.assertRaises(Refused): # second rollback has nothing case-created + adapter.rollback(replace(self.back, attempt_id='attempt-two'), a.CREATED) + + def test_rollback_without_creation_refused_and_revokes(self): + adapter = a.build(**self.kw) + with self.assertRaises(Refused): + adapter.rollback(self.back, a.CREATED) + self.assertEqual([c[0] for c in self.calls], []) + self.assertEqual(self.revoked, [('session', 'rollback'), ('grant', 'rollback')]) + + def test_receipt_with_value_or_extra_field_refused(self): + for bad in (receipt('mint', value='x'), receipt('mint', values_emitted=True), + receipt('mint', path='platform/workloads/keycape/other'), + receipt('mint', created=False), receipt('mint', version='3'), 'opaque'): + self.mint_receipt = bad + with self.assertRaises(Refused): + a.build(**self.kw).invoke(self.run) + self.assertEqual(self.evidence, {}) + + def test_public_client_and_target_mismatch_refused(self): + for params in ({'client_id': 'forgejo', 'client_type': 'public'}, + {'client_id': 'other', 'client_type': 'confidential'}): + case = yaml.safe_dump({'spec': {'parameters': params}}).encode() + pins = replace(self.admitted.pins, case=digest(case)) + with self.assertRaises(Refused): + a.build(**dict(self.kw, case_source=case, admitted=replace(self.admitted, pins=pins))) + + def test_binding_refusals(self): + cases = [dict(case_source=self.case + b'#'), dict(case_acceptance='other'), + dict(admitted=replace(self.admitted, lane='key-cape-oidc-login')), + dict(admitted=replace(self.admitted, pins=replace(self.admitted.pins, adapter=digest(b'x')))), + dict(admitted=self.run), dict(mint='bao kv put')] + for over in cases: + with self.assertRaises(Refused): + a.build(**dict(self.kw, **over)) + + def test_attempt_binding_and_reused_evidence_refused(self): + adapter = a.build(**self.kw) + with self.assertRaises(Refused): + adapter.invoke(replace(self.run, target='other')) + self.kw['publish'] = lambda payload: 'owner-evidence:fixed' + adapter = a.build(**self.kw) + adapter.invoke(self.run) + with self.assertRaises(Refused): + adapter.invoke(replace(self.run, attempt_id='attempt-two')) + + def test_verify_rejects_foreign_result(self): + adapter = a.build(**self.kw) + result = adapter.invoke(self.run) + self.assertFalse(adapter.verify(replace(self.run, attempt_id='attempt-two'), result)) + self.assertFalse(a.build(**self.kw).verify(self.run, result)) + + +if __name__ == '__main__': + unittest.main() From ec635393b68809336d3f8b8fc996a6b7f3258534 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 17 Oct 2026 07:55:36 +0200 Subject: [PATCH 4/7] 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() From 650c1ac57c56ec1b8d5badcaa0a01055b8e8e341 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 17 Oct 2026 07:55:47 +0200 Subject: [PATCH 5/7] NK-WP-0048-T28: deliver-consumer-secret adapter source candidate Unexercised, injection-only, value-free evidence seam for type 1.2 with positive/refusal fixtures and created-only rollback. Canonical 1.1 default unchanged; no owner review or native action supplied. 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 --- .../deliver-consumer-secret-adapter.md | 38 ++++ .../deliver-consumer-secret-adapter.py | 193 ++++++++++++++++++ .../test_deliver_consumer_secret_adapter.py | 138 +++++++++++++ 3 files changed, 369 insertions(+) create mode 100644 sso-mfa/k8s/keycape/deliver-consumer-secret-adapter.md create mode 100644 sso-mfa/k8s/keycape/deliver-consumer-secret-adapter.py create mode 100644 sso-mfa/k8s/keycape/test_deliver_consumer_secret_adapter.py diff --git a/sso-mfa/k8s/keycape/deliver-consumer-secret-adapter.md b/sso-mfa/k8s/keycape/deliver-consumer-secret-adapter.md new file mode 100644 index 0000000..bf229a0 --- /dev/null +++ b/sso-mfa/k8s/keycape/deliver-consumer-secret-adapter.md @@ -0,0 +1,38 @@ +# Deliver-consumer-secret adapter candidate + +Exercise status: unexercised + +Source-only, injection-only evidence seam for the type 1.2 step +`deliver-consumer-secret` (NK-WP-0048-T28), modelled on the T13 registration +adapter. Warden does not execute the consumer owner's apply +(`warden_executes: false`); the adapter binds one admitted case and accepts a +closed, value-free owner status receipt. It installs no command, kubectl call, +credential provider, OpenBao access or evidence trust root. Canonical 1.1 stays +the default: a 1.1 case, or a public client, is refused because the step does not +exist there. No `--allow-unimplemented` admission is used or implied. + +Binding: descriptor (`agent` actor, lane `consumer-owner-deployment`, effect +`change`, not retry-safe, `parent-session` revocation) pins this adapter source; +case, descriptor, target (`client_id`) and owner (`consumer_owner_repo`) must match +the admitted invocation; each attempt is rebound and the source rehashed. + +Receipt contract (`netkingdom.consumer-secret-delivery.v1`): per object +(`consumer-eso-sa`, `consumer-secretstore`, `consumer-externalsecret`) a uid, a +`created` flag and fixed metadata that must equal what the case parameters derive: +SA automount false with no legacy token refs; SecretStore with the dedicated +serviceAccountRef, audiences `[openbao]`, the case role and Ready; ExternalSecret +with Owner creation, Retain deletion, path `workloads/keycape/{client_id}`, only +key `CLIENT_SECRET`, Ready. Any value-bearing key (`data`, `stringData`, `value`, +`token`, `secret`, ...), `values_emitted` other than false, or metadata drift is +refused. Published evidence carries status, uids and revocation only. + +Rollback: only objects the owner reported created in this process, with +authenticated evidence and the exact created set, are removed — ExternalSecret, +then SecretStore, then SA — and only after the owner reports the consumer secret +reference reverted. Removed uids must equal the created uids. Preexisting-matched +objects are never deleted. Removing SecretStore/SA is not OpenBao token revocation. + +Tests: `PYTHONPATH=~/ops-warden/src python3 -m pytest test_deliver_consumer_secret_adapter.py` +(synthetic fixtures only). Not supplied here: Platform and Apps owner source +review before pinning, ESO TokenRequest/RBAC admission validation, and any native +action — each a separate gate. diff --git a/sso-mfa/k8s/keycape/deliver-consumer-secret-adapter.py b/sso-mfa/k8s/keycape/deliver-consumer-secret-adapter.py new file mode 100644 index 0000000..f3b592d --- /dev/null +++ b/sso-mfa/k8s/keycape/deliver-consumer-secret-adapter.py @@ -0,0 +1,193 @@ +"""Unexercised, injection-only consumer-secret delivery evidence seam (type 1.2). + +Source candidate for step deliver-consumer-secret. Warden does not execute the +consumer owner's apply (warden_executes: false); this adapter only binds one +admitted case, invokes an injected owner capability, and accepts a closed, +value-free status receipt. No CLI, kubectl, credential, OpenBao call or evidence +trust root is installed here. Canonical type 1.1 remains the default. +""" +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 = ('deliver-consumer-secret-adapter.py',) +REFERENCE = 'net-kingdom/deliver-consumer-secret@v1' +TYPE = 'oidc-browser-relying-party@1.2' +OBJECTS = ('consumer-eso-sa', 'consumer-secretstore', 'consumer-externalsecret') +# Rollback order: ExternalSecret/owned target, then SecretStore, then SA. +ROLLBACK_ORDER = tuple(reversed(OBJECTS)) +VALUE_KEYS = {'data', 'stringData', 'value', 'values', 'token', 'secret', 'clientSecret'} +RECEIPT_KEYS = {'schema', 'operation', 'objects', 'values_emitted', 'consumer_reference_reverted'} + + +def source_digest(): + files = {name: digest((ROOT / name).read_bytes()) for name in SOURCE_FILES} + return digest(json.dumps({'files': files}, sort_keys=True, separators=(',', ':')).encode()) + + +def descriptor(owner): + if not isinstance(owner, str) or not IDENTIFIER.fullmatch(owner): + raise Refused('bounded consumer owner required') + return Descriptor('warden.case.execution.v1', 'deliver-consumer-secret', owner, + REFERENCE, source_digest(), 'agent', 'consumer-owner-deployment', + 'client_id', 'change', OBJECTS, False, ('parent-session',)) + + +def expected(params): + """Fixed metadata the owner receipt must match; derived from case parameters only.""" + cid, svc, ns = params['client_id'], params['consuming_service'], params['consumer_namespace'] + return { + 'consumer-eso-sa': {'kind': 'ServiceAccount', 'namespace': ns, + 'name': f'{svc}-keycape-{cid}-eso', + 'automount': False, 'legacy_token_refs': 0}, + 'consumer-secretstore': {'kind': 'SecretStore', 'namespace': ns, + 'name': f'openbao-keycape-{cid}-consumer', + 'service_account': f'{svc}-keycape-{cid}-eso', + 'audiences': ['openbao'], + 'role': f'external-secrets-keycape-{cid}-{svc}', 'ready': True}, + 'consumer-externalsecret': {'kind': 'ExternalSecret', 'namespace': ns, + 'name': f'{svc}-netkingdom-oidc-client', + 'creation_policy': 'Owner', 'deletion_policy': 'Retain', + 'remote_path': f'workloads/keycape/{cid}', + 'keys': ['CLIENT_SECRET'], 'ready': True}, + } + + +def _value_free(obj): + if isinstance(obj, dict): + return not (VALUE_KEYS & set(obj)) and all(_value_free(v) for v in obj.values()) + if isinstance(obj, list): + return all(_value_free(v) for v in obj) + return obj is None or isinstance(obj, (bool, int, str)) and (not isinstance(obj, str) or len(obj) <= 253) + + +def build(*, admitted: Invocation, case_source: bytes, owner_apply, owner_remove, + publish, authenticate, revoke_session): + """Bind one admitted 1.2 confidential case; return a closed kernel Adapter. + + owner_apply(inv, expected)->receipt and owner_remove(inv, created)->receipt are + consumer-owner capabilities (never command strings). Receipts carry status and + fixed metadata only; any value-bearing key is refused. Only objects the owner + reports as created in this attempt (uid-bound) are recorded and may be removed; + preexisting-matched objects are never deleted. + """ + reviewed = source_digest() + 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') + desc = descriptor(admitted.owner) + if (admitted.step, admitted.actor, admitted.lane) != (desc.step, desc.actor, desc.lane): + raise Refused('delivery 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 all(callable(x) for x in (owner_apply, owner_remove, publish, authenticate, revoke_session)): + raise Refused('owner capabilities required') + with contained_output(): + 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') != TYPE: + raise Refused('type 1.2 case required; canonical 1.1 has no delivery step') + if params.get('client_type') != 'confidential': + raise Refused('delivery applies to confidential clients only') + for key in ('client_id', 'consuming_service', 'consumer_namespace', 'consumer_owner_repo'): + if not isinstance(params.get(key), str) or not IDENTIFIER.fullmatch(params[key]): + raise Refused('bounded case parameter required') + if admitted.target != params['client_id'] or admitted.owner != params['consumer_owner_repo']: + raise Refused('case target or owner differs') + want = expected(params) + records, state = {}, {'created': 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 check(receipt, operation, scope): + if (type(receipt) is not dict or set(receipt) != RECEIPT_KEYS or not _value_free(receipt) + or receipt['schema'] != 'netkingdom.consumer-secret-delivery.v1' + or receipt['operation'] != operation or receipt['values_emitted'] is not False + or type(receipt['objects']) is not dict or set(receipt['objects']) != set(scope)): + raise Refused('owner delivery receipt contract differs') + created = [] + for name in scope: + obj = receipt['objects'][name] + if type(obj) is not dict or set(obj) != {'uid', 'created', 'metadata'}: + raise Refused('owner object receipt differs') + if not isinstance(obj['uid'], str) or not 0 < len(obj['uid']) <= 64: + raise Refused('object uid binding required') + if operation == 'apply': + if obj['metadata'] != want[name] or type(obj['created']) is not bool: + raise Refused('delivered metadata differs from case') + if obj['created']: + created.append(name) + return {n: receipt['objects'][n]['uid'] for n in scope}, tuple(created) + + def execute(inv, operation, created=()): + try: + with contained_output(): + if operation == 'apply': + binding(inv, 'run') + uids, objects = check(owner_apply(inv, want), 'apply', OBJECTS) + outcome = Outcome.SUCCESS if objects else Outcome.NO_CHANGE + else: + binding(inv, 'rollback') + prior = state['created'] + if prior is None or not created or set(created) != set(prior[1]) or not verify(*prior[0]): + raise Refused('no authenticated case-created delivery objects') + order = tuple(n for n in ROLLBACK_ORDER if n in created) + receipt = owner_remove(inv, {n: prior[1][n] for n in order}) + if type(receipt) is dict and receipt.get('consumer_reference_reverted') is not True: + raise Refused('consumer secret reference must be reverted first') + uids, _ = check(receipt, 'remove', order) + if uids != {n: prior[1][n] for n in order}: + raise Refused('removed object uid differs from created') + objects, outcome = (), Outcome.ROLLED_BACK + finally: + with contained_output(): + revocation = revoke_session(inv) + if not isinstance(revocation, str) or not IDENTIFIER.fullmatch(revocation): + raise Refused('session revocation evidence missing') + payload = json.dumps({'schema': 'netkingdom.consumer-secret-delivery.attempt.v1', + 'invocation': asdict(inv), 'source': reviewed, 'outcome': outcome.value, + 'created': objects, 'uids': uids, 'session_revocation': revocation}, + 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, True, revocation, + attempt_id=inv.attempt_id, operation=inv.operation) + records[evidence] = (inv, payload, result) + if operation == 'apply' and objects: + state['created'] = ((inv, result), {n: uids[n] for n in objects}) + elif operation == 'remove': + state['created'] = None + return result + + 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, lambda inv: execute(inv, 'apply'), verify, + lambda inv, created: execute(inv, 'remove', created)) diff --git a/sso-mfa/k8s/keycape/test_deliver_consumer_secret_adapter.py b/sso-mfa/k8s/keycape/test_deliver_consumer_secret_adapter.py new file mode 100644 index 0000000..d4371b8 --- /dev/null +++ b/sso-mfa/k8s/keycape/test_deliver_consumer_secret_adapter.py @@ -0,0 +1,138 @@ +"""Synthetic positive/refusal fixtures for the deliver-consumer-secret seam. No live calls.""" +from dataclasses import asdict, replace +import copy +import importlib.util +import json +from pathlib import Path +import unittest + +import yaml +from warden.case_run import Invocation, digest +from warden.case_receipt_state import Outcome, Pins, Refused + +ROOT = Path(__file__).parent +spec = importlib.util.spec_from_file_location('delivery_adapter', ROOT / 'deliver-consumer-secret-adapter.py') +a = importlib.util.module_from_spec(spec) +spec.loader.exec_module(a) + +PARAMS = {'client_id': 'forgejo', 'client_type': 'confidential', 'consuming_service': 'forgejo', + 'consumer_namespace': 'forgejo', 'consumer_owner_repo': 'forgejo-ops'} + + +class DeliveryTests(unittest.TestCase): + def setUp(self): + self.case = self.case_bytes() + self.effects, self.published = [], {} + self.created = {n: True for n in a.OBJECTS} + self.reverted = True + self.admitted = self.invocation(self.case) + self.inv = replace(self.admitted, attempt_id='attempt-one', operation='run') + self.rinv = replace(self.admitted, attempt_id='attempt-two', operation='rollback') + + def case_bytes(self, type_='oidc-browser-relying-party@1.2', **over): + return yaml.safe_dump({'spec': {'type': type_, 'parameters': {**PARAMS, **over}}}).encode() + + def invocation(self, case, owner='forgejo-ops'): + desc = json.dumps(asdict(a.descriptor(owner)), sort_keys=True, separators=(',', ':')).encode() + pins = Pins(digest(case), digest(b'type'), digest(desc), digest(b'catalog'), + digest(b'acceptance'), digest(b'run sheet')) + return Invocation('deliver-consumer-secret', owner, 'forgejo', 'agent', 'founder', + 'consumer-owner-deployment', pins) + + def receipt(self, operation, names, want=None): + objs = {n: {'uid': f'uid-{n}', 'created': self.created[n], + 'metadata': copy.deepcopy(want[n]) if want else {}} for n in names} + return {'schema': 'netkingdom.consumer-secret-delivery.v1', 'operation': operation, + 'objects': objs, 'values_emitted': False, 'consumer_reference_reverted': self.reverted} + + def adapter(self, case=None, admitted=None, apply=None): + def owner_apply(inv, want): + self.effects.append('apply') + return apply(want) if apply else self.receipt('apply', a.OBJECTS, want) + + def owner_remove(inv, targets): + self.effects.append(('remove', tuple(targets))) + return self.receipt('remove', tuple(targets)) + + def publish(payload): + ident = f'evidence-{len(self.published)}' + self.published[ident] = payload + return ident + + return a.build(admitted=admitted or self.admitted, case_source=case or self.case, + owner_apply=owner_apply, owner_remove=owner_remove, publish=publish, + authenticate=lambda i, p: self.published.get(i) == p, + revoke_session=lambda inv: f'revoked-{inv.attempt_id}') + + def test_positive_apply_verify_and_created_only_rollback(self): + self.created['consumer-eso-sa'] = False # preexisting-matched: must be preserved + ad = self.adapter() + result = ad.invoke(self.inv) + self.assertEqual(result.outcome, Outcome.SUCCESS) + self.assertEqual(result.created, ('consumer-secretstore', 'consumer-externalsecret')) + self.assertTrue(ad.verify(self.inv, result)) + payload = self.published[result.evidence].decode() + self.assertNotIn('CLIENT_SECRET', payload) + back = ad.rollback(self.rinv, result.created) + self.assertEqual(back.outcome, Outcome.ROLLED_BACK) + self.assertEqual(self.effects[-1], ('remove', ('consumer-externalsecret', 'consumer-secretstore'))) + + def test_no_change_has_nothing_to_roll_back(self): + self.created = {n: False for n in a.OBJECTS} + ad = self.adapter() + self.assertEqual(ad.invoke(self.inv).outcome, Outcome.NO_CHANGE) + with self.assertRaises(Refused): + ad.rollback(self.rinv, ('consumer-externalsecret',)) + self.assertNotIn('remove', [e[0] for e in self.effects if isinstance(e, tuple)]) + + def test_refuses_canonical_11_public_and_bad_binding(self): + for case in (self.case_bytes('oidc-browser-relying-party@1.1'), + self.case_bytes(client_type='public')): + with self.assertRaises(Refused): + self.adapter(case=case, admitted=self.invocation(case)) + with self.assertRaises(Refused): + self.adapter(case=self.case + b'\n#drift') + with self.assertRaises(Refused): + self.adapter(admitted=self.invocation(self.case, owner='other-owner')) + with self.assertRaises(Refused): + self.adapter().invoke(replace(self.inv, target='other')) + self.assertEqual(self.effects, []) + + def test_refuses_value_bearing_or_mismatched_receipts(self): + def value(want): + r = self.receipt('apply', a.OBJECTS, want) + r['objects']['consumer-externalsecret']['metadata']['data'] = {'CLIENT_SECRET': 'x'} + return r + + def drift(want): + r = self.receipt('apply', a.OBJECTS, want) + r['objects']['consumer-externalsecret']['metadata']['keys'] = ['CLIENT_SECRET', 'OTHER'] + return r + + def automount(want): + r = self.receipt('apply', a.OBJECTS, want) + r['objects']['consumer-eso-sa']['metadata']['automount'] = True + return r + + def emitted(want): + return {**self.receipt('apply', a.OBJECTS, want), 'values_emitted': True} + + for bad in (value, drift, automount, emitted): + ad = self.adapter(apply=bad) + with self.assertRaises(Refused): + ad.invoke(self.inv) + + def test_rollback_requires_reverted_reference_and_exact_created_set(self): + ad = self.adapter() + result = ad.invoke(self.inv) + with self.assertRaises(Refused): + ad.rollback(self.rinv, ('consumer-externalsecret',)) + self.reverted = False + with self.assertRaises(Refused): + ad.rollback(self.rinv, result.created) + forged = replace(result, evidence='evidence-forged') + self.assertFalse(ad.verify(self.inv, forged)) + + +if __name__ == '__main__': + unittest.main() From 5a2e7d67a2d9fcbe3239080f23c0e18156b5b13d Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 17 Oct 2026 07:56:25 +0200 Subject: [PATCH 6/7] NK-WP-0048-T24: ensure-groups adapter source candidate Binds group creation to the T14/T21 PostgreSQL journal seam (exact case, source and principal binding; refuses untrusted journals), canonical 1.1 case only, created-only rollback via the journal receipt. Source candidate; owner review and native action remain separate gates. 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/ensure-groups-adapter.md | 47 +++++ sso-mfa/k8s/keycape/ensure-groups-adapter.py | 183 +++++++++++++++++ .../k8s/keycape/test_ensure_groups_adapter.py | 186 ++++++++++++++++++ 3 files changed, 416 insertions(+) create mode 100644 sso-mfa/k8s/keycape/ensure-groups-adapter.md create mode 100644 sso-mfa/k8s/keycape/ensure-groups-adapter.py create mode 100644 sso-mfa/k8s/keycape/test_ensure_groups_adapter.py diff --git a/sso-mfa/k8s/keycape/ensure-groups-adapter.md b/sso-mfa/k8s/keycape/ensure-groups-adapter.md new file mode 100644 index 0000000..0453440 --- /dev/null +++ b/sso-mfa/k8s/keycape/ensure-groups-adapter.md @@ -0,0 +1,47 @@ +# Ensure-groups adapter source candidate (NK-WP-0048-T24) + +Source candidate only, modelled on the T13 registration adapter. It is +unexercised against any live system and is not pinned, admitted or reviewed. + +## Binding + +- `build()` takes an admitted `ensure-groups` invocation, the exact case bytes, + an injected provisioner, and owner `publish`/`authenticate` evidence + capabilities. +- Group creation binds to the T14/T21 journal seam. The provisioner's + `group_provenance` must be a `PostgresGroupProvenance`. Its `GroupJournalBinding` + must name this case pin, the reviewed adapter source digest and the admitted + principal. A missing journal, a local SQLite journal or a differing binding is + refused before any call. +- The case must be canonical `oidc-browser-relying-party@1.1`. `access_groups` + and `initial_members` are parsed exactly, and members outside `access_groups` + are refused. +- The idempotency key and correlation id are derived from the case pin, so + every attempt resumes the same journal receipt. A replay returns `no_change` + and never recreates removed state. +- The source digest pins the adapter plus `provisioner.py`, + `group_provenance.py` and `group_provenance_postgres.py`. + +## Rollback + +Rollback calls `revoke_service_groups` against the original journal receipt. +It accepts only `service-group`/`service-group-membership` created objects. +The journal, not the caller, decides what was created, and preexisting or +shared state is preserved. `delete_created_if_empty` defaults to `false`. +Enabling it requires an owner fact that no other case uses the groups. This +adapter does not supply that fact. + +`verify` requires the authenticated owner evidence and a journal readback +state (`ensured` / `revoked`) matching the outcome. It fails closed on any +journal error. + +## Not supplied here + +- NetKingdom (directory) owner source review before pinning. +- Native CNPG journal admission, trusted continuity/checkpoint callbacks and + protected `GroupJournalBinding` construction (T14). +- Writer isolation or CAS against external LDAP writers. +- Any native action. + +Tests use synthetic fakes only (no SQL, no LLDAP): +`PYTHONPATH=~/ops-warden/src ~/ops-warden/.venv/bin/python -m pytest sso-mfa/k8s/keycape/test_ensure_groups_adapter.py` diff --git a/sso-mfa/k8s/keycape/ensure-groups-adapter.py b/sso-mfa/k8s/keycape/ensure-groups-adapter.py new file mode 100644 index 0000000..e19124f --- /dev/null +++ b/sso-mfa/k8s/keycape/ensure-groups-adapter.py @@ -0,0 +1,183 @@ +"""Unexercised, injection-only ensure-groups seam for Warden's v1 kernel (T24). + +Source candidate only. No CLI, credentials, LLDAP session, SQL connection or +evidence trust root is installed here. Group creation is bound to T14's +journal/provenance seam: the injected provisioner must carry a +PostgresGroupProvenance whose GroupJournalBinding names this exact case, +principal and reviewed adapter source, with independent continuity callbacks. +SQLite/local journals, a missing journal or a differing binding are refused. +Owner source review and every native action remain separate gates. +""" +from __future__ import annotations + +from dataclasses import asdict, replace +import json +import sys +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 +PROVISIONER = ROOT.parents[2] / 'identity-provisioner' +SOURCE_FILES = (('sso-mfa/k8s/keycape', 'ensure-groups-adapter.py'), + ('identity-provisioner', 'provisioner.py'), + ('identity-provisioner', 'group_provenance.py'), + ('identity-provisioner', 'group_provenance_postgres.py')) +REFERENCE = 'net-kingdom/ensure-groups@v1' +CASE_TYPE = 'oidc-browser-relying-party@1.1' # canonical 1.1 stays the default +CREATED = ('service-group', 'service-group-membership') + +if str(PROVISIONER) not in sys.path: + sys.path.insert(0, str(PROVISIONER)) +from group_provenance_postgres import PostgresGroupProvenance # noqa: E402 (no I/O on import) + + +def source_digest(): + """Pin adapter plus the provisioner/journal import closure, in path order.""" + base = ROOT.parents[2] + files = {f'{d}/{n}': digest((base / d / n).read_bytes()) for d, n in SOURCE_FILES} + return digest(json.dumps({'files': files}, sort_keys=True, separators=(',', ':')).encode()) + + +def descriptor(): + return Descriptor('warden.case.execution.v1', 'ensure-groups', 'net-kingdom', REFERENCE, + source_digest(), 'human', 'key-cape-oidc-login', 'client_id', 'change', + CREATED, False, ()) + + +def _request(case): + spec = case.get('spec') if isinstance(case, dict) else None + if not isinstance(spec, dict) or spec.get('type') != CASE_TYPE: + raise Refused('canonical oidc-browser-relying-party@1.1 case required') + params = spec.get('parameters') + if not isinstance(params, dict) or not isinstance(params.get('access_groups'), str): + raise Refused('access_groups required') + groups = params['access_groups'].split(',') + if len(set(groups)) != len(groups): + raise Refused('duplicate access group') + members = {} + raw = params.get('initial_members', '') + if not isinstance(raw, str): + raise Refused('initial_members must be a string') + for entry in filter(None, raw.split(';')): + group, _, users = entry.partition('=') + if group not in groups or group in members or not users: + raise Refused('initial_members names a group outside access_groups') + members[group] = users.split('+') + return params.get('client_id'), groups, members + + +def build(*, admitted: Invocation, case_source: bytes, provisioner, publish, authenticate, + delete_created_if_empty: bool = False): + """Bind one admitted case to the owner journal; return a closed kernel Adapter. + + provisioner exposes ensure_service_groups/revoke_service_groups and carries + the trusted owner journal as group_provenance. publish(bytes)->identifier and + authenticate(identifier, bytes)->True are owner evidence capabilities. + + The idempotency key and correlation derive from the case pin, so every + attempt resumes the one journal receipt and never recreates removed state. + Rollback reverses only journal-confirmed creations; preexisting/shared + groups and memberships are preserved. Ambiguity propagates as uncertain. + """ + 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('ensure-groups 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') + descriptor_bytes = json.dumps(asdict(desc), sort_keys=True, separators=(',', ':')).encode() + if admitted.pins.adapter != digest(descriptor_bytes): + raise Refused('descriptor pin differs') + journal = getattr(provisioner, 'group_provenance', None) + if type(journal) is not PostgresGroupProvenance: + raise Refused('trusted owner journal callbacks required') + binding = journal.binding + if (binding.case_digest, binding.source_digest, binding.principal) != ( + admitted.pins.case, reviewed, admitted.principal): + raise Refused('owner journal binding differs from this case') + if not all(callable(getattr(provisioner, m, None)) for m in ('ensure_service_groups', 'revoke_service_groups')): + raise Refused('owner group operations required') + if not all(callable(x) for x in (publish, authenticate)) or type(delete_created_if_empty) is not bool: + raise Refused('owner evidence capabilities required') + with contained_output(): + client_id, groups, members = _request(yaml.safe_load(case_source)) + if client_id != admitted.target: + raise Refused('case target differs') + key = 'warden-ensure:' + admitted.pins.case + request = {'groups': groups, 'members': members, 'correlation_id': admitted.pins.case} + journal_binding = provisioner._group_binding(groups, members, admitted.pins.case) + records = {} + + def check_attempt(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 execute(inv, rollback, created=()): + with contained_output(): + check_attempt(inv, 'rollback' if rollback else 'run') + if rollback: + if not created or set(created) - set(CREATED): + raise Refused('only journal-confirmed creations may be rolled back') + result = provisioner.revoke_service_groups(dict( + request, idempotency_key=key + ':revoke', ensure_idempotency_key=key, + ensure_correlation_id=admitted.pins.case, delete_if_empty=delete_created_if_empty)) + expected = {'groups_revoked', 'groups_revoked_and_deleted'} + else: + result = provisioner.ensure_service_groups(dict(request, idempotency_key=key)) + expected = {'groups_ensured'} + if (result.status not in expected or result.ensure_idempotency_key != key + or result.ensure_correlation_id != admitted.pins.case + or result.external_subject != ','.join(sorted(groups))): + raise Refused('owner journal result contract differs') + objects = tuple(name for name, items in zip(CREATED, (result.created_groups, result.created_memberships)) if items) + if rollback: + outcome = Outcome.ROLLED_BACK + elif result.resumed or not objects: + outcome, objects = Outcome.NO_CHANGE, () + else: + outcome = Outcome.SUCCESS + # Names and ids only: the journal holds no secret values. + payload = json.dumps({'schema': 'netkingdom.ensure-groups.attempt.v1', + 'invocation': asdict(inv), 'source': reviewed, 'outcome': outcome.value, + 'created': objects, 'journal_key': key, 'status': result.status, + 'created_groups': list(result.created_groups), + 'created_memberships': list(result.created_memberships)}, + 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') + owner = OwnerResult(outcome, evidence, objects, rollback, + attempt_id=inv.attempt_id, operation=inv.operation) + records[evidence] = (inv, payload, owner) + return owner + + 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(): + try: + state = journal.get(key, journal_binding)['state'] + except Exception: + return False + want = 'revoked' if result.outcome is Outcome.ROLLED_BACK else 'ensured' + return state == want and authenticate(result.evidence, record[1]) is True + + return Adapter(REFERENCE, reviewed, lambda inv: execute(inv, False), verify, + lambda inv, created: execute(inv, True, created)) diff --git a/sso-mfa/k8s/keycape/test_ensure_groups_adapter.py b/sso-mfa/k8s/keycape/test_ensure_groups_adapter.py new file mode 100644 index 0000000..d6105c3 --- /dev/null +++ b/sso-mfa/k8s/keycape/test_ensure_groups_adapter.py @@ -0,0 +1,186 @@ +"""Ensure-groups adapter against synthetic journal and provisioner fakes only. + +No SQL connection, LLDAP session or native action happens here; the fake +connector raises if anything tries to open it. +""" +from dataclasses import asdict, replace +import importlib.util +import json +from pathlib import Path +import unittest +from unittest.mock import patch +import uuid + +from warden.case_run import Invocation, digest +from warden.case_receipt_state import Outcome, Pins, Refused +import yaml + +ROOT = Path(__file__).parent +spec = importlib.util.spec_from_file_location('ensure_groups_adapter', ROOT / 'ensure-groups-adapter.py') +a = importlib.util.module_from_spec(spec) +spec.loader.exec_module(a) +from group_provenance_postgres import GroupJournalBinding, PostgresGroupProvenance # noqa: E402 +from group_provenance import GroupProvenance # noqa: E402 +from provisioner import GroupResult, LLDAPProvisioner # noqa: E402 + + +class Continuity: + def check(self, *args): + return True + + def advance(self, *args): + return True + + +def no_connect(): + raise AssertionError('no SQL connection in source tests') + + +class FakeProvisioner: + _group_binding = staticmethod(LLDAPProvisioner._group_binding) + + def __init__(self, journal): + self.group_provenance = journal + self.calls = [] + self.state = None + self.created_groups = ({'name': 'forgejo-users', 'id': 7},) + self.created_memberships = ({'group': 'forgejo-users', 'user': 'alice'},) + + def _result(self, payload, status, resumed): + return GroupResult('netkingdom-lldap', ','.join(sorted(payload['groups'])), status, resumed, + payload.get('ensure_idempotency_key', payload['idempotency_key']), + payload.get('ensure_correlation_id', payload['correlation_id']), + self.created_groups, self.created_memberships) + + def ensure_service_groups(self, payload): + self.calls.append(('ensure', payload)) + resumed = self.state == 'ensured' + self.state = 'ensured' + return self._result(payload, 'groups_ensured', resumed) + + def revoke_service_groups(self, payload): + self.calls.append(('revoke', payload)) + if self.state != 'ensured': + raise RuntimeError('group provenance unresolved') + self.state = 'revoked' + return self._result(payload, 'groups_revoked', False) + + +class EnsureGroupsTests(unittest.TestCase): + def setUp(self): + self.case = yaml.safe_dump({'spec': {'type': 'oidc-browser-relying-party@1.1', 'parameters': { + 'client_id': 'forgejo', 'access_groups': 'forgejo-users,forgejo-admins', + 'initial_members': 'forgejo-users=alice'}}}).encode() + 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'acceptance'), digest(b'run sheet')) + self.admitted = Invocation('ensure-groups', 'net-kingdom', 'forgejo', 'human', + 'founder', 'key-cape-oidc-login', pins) + self.inv = replace(self.admitted, attempt_id='attempt-one', operation='run') + self.journal = self.make_journal() + self.prov = FakeProvisioner(self.journal) + self.evidence = {} + self.kw = dict(admitted=self.admitted, case_source=self.case, provisioner=self.prov, + publish=self.publish, authenticate=lambda r, p: self.evidence.get(r) == p) + get = patch.object(PostgresGroupProvenance, 'get', lambda s, k, b: {'state': self.prov.state}) + get.start() + self.addCleanup(get.stop) + + def make_journal(self, **override): + values = dict(tenant_scope='platform', directory='netkingdom-lldap', + case_digest=digest(self.case), source_digest=a.source_digest(), + principal='founder', epoch=str(uuid.uuid4())) + values.update(override) + return PostgresGroupProvenance(connect=no_connect, binding=GroupJournalBinding(**values), + continuity=Continuity()) + + def publish(self, payload): + ref = 'owner-evidence:' + str(len(self.evidence)) + self.evidence[ref] = payload + return ref + + def test_run_creates_verifies_and_rolls_back_only_creations(self): + adapter = a.build(**self.kw) + result = adapter.invoke(self.inv) + self.assertEqual(result.outcome, Outcome.SUCCESS) + self.assertEqual(result.created, ('service-group', 'service-group-membership')) + self.assertTrue(adapter.verify(self.inv, result)) + ensure = self.prov.calls[0][1] + self.assertEqual(ensure['idempotency_key'], 'warden-ensure:' + digest(self.case)) + self.assertEqual(ensure['members'], {'forgejo-users': ['alice']}) + rb_inv = replace(self.inv, attempt_id='attempt-two', operation='rollback') + rolled = adapter.rollback(rb_inv, result.created) + self.assertEqual(rolled.outcome, Outcome.ROLLED_BACK) + revoke = self.prov.calls[1][1] + self.assertEqual(revoke['ensure_idempotency_key'], ensure['idempotency_key']) + self.assertIs(revoke['delete_if_empty'], False) + self.assertTrue(adapter.verify(rb_inv, rolled)) + self.assertFalse(adapter.verify(self.inv, result)) # journal now says revoked + + def test_replay_resumes_without_creation(self): + adapter = a.build(**self.kw) + adapter.invoke(self.inv) + again = adapter.invoke(replace(self.inv, attempt_id='attempt-two')) + self.assertEqual((again.outcome, again.created), (Outcome.NO_CHANGE, ())) + + def test_refuses_untrusted_or_differing_journal(self): + for journal in (None, GroupProvenance.__new__(GroupProvenance), + self.make_journal(case_digest=digest(b'other')), + self.make_journal(principal='someone-else'), + self.make_journal(source_digest=digest(b'other source'))): + self.prov.group_provenance = journal + with self.assertRaises(Refused): + a.build(**self.kw) + + def test_refuses_case_and_invocation_drift(self): + bad = [dict(case_source=self.case + b'\n'), + dict(admitted=replace(self.admitted, target='gitea')), + dict(admitted=replace(self.admitted, step='mint-secret')), + dict(publish=None), + dict(delete_created_if_empty='yes')] + for override in bad: + with self.assertRaises(Refused): + a.build(**{**self.kw, **override}) + + def test_refuses_non_canonical_or_malformed_case(self): + for params, kind in (({'client_id': 'forgejo', 'access_groups': 'g'}, 'oidc-browser-relying-party@1.2'), + ({'client_id': 'forgejo', 'access_groups': 'g', 'initial_members': 'h=alice'}, None), + ({'client_id': 'forgejo'}, None)): + case = yaml.safe_dump({'spec': {'type': kind or a.CASE_TYPE, 'parameters': params}}).encode() + pins = replace(self.admitted.pins, case=digest(case)) + self.prov.group_provenance = self.make_journal(case_digest=digest(case)) + with self.assertRaises(Refused): + a.build(**{**self.kw, 'case_source': case, 'admitted': replace(self.admitted, pins=pins)}) + + def test_rollback_refuses_unconfirmed_or_foreign_objects(self): + adapter = a.build(**self.kw) + rb_inv = replace(self.inv, operation='rollback') + for created in ((), ('keycape-client',)): + with self.assertRaises(Refused): + adapter.rollback(rb_inv, created) + with self.assertRaises(Refused): # journal has no ensured receipt (contained owner failure) + adapter.rollback(rb_inv, ('service-group',)) + self.assertEqual([c[0] for c in self.prov.calls], ['revoke']) + + def test_attempt_binding_and_contract_drift(self): + adapter = a.build(**self.kw) + with self.assertRaises(Refused): + adapter.invoke(replace(self.inv, principal='other')) + with self.assertRaises(Refused): + adapter.invoke(replace(self.inv, operation='rollback')) + self.prov.ensure_service_groups = lambda p: GroupResult( + 'netkingdom-lldap', 'x', 'groups_ensured', False, 'k', 'c', (), ()) + with self.assertRaises(Refused): + adapter.invoke(self.inv) + + def test_verify_fails_closed_on_journal_or_evidence_failure(self): + adapter = a.build(**self.kw) + result = adapter.invoke(self.inv) + self.evidence.clear() + self.assertFalse(adapter.verify(self.inv, result)) + with patch.object(PostgresGroupProvenance, 'get', side_effect=RuntimeError): + self.assertFalse(adapter.verify(self.inv, result)) + + +if __name__ == '__main__': + unittest.main() From 74adcd7260bfca678e02edc5419d476b65a18f9a Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 17 Oct 2026 07:57:13 +0200 Subject: [PATCH 7/7] NK-WP-0048-T31: restart and lost-receipt fixtures against Warden original intent Fixture-only reconcile_original owner over the T13 registration adapter, driven through Warden's actual Executor/Journal (ops-warden a45cf47): original intent kept separate from the reconcile attempt across restart; holder-unavailable, missing/ambiguous record, wrong attempt/case/source/ principal, NO_CHANGE original, changed revision and absent client entry all refuse and stay uncertain with no add replay or undo. The holder is fixture-only; actual authenticated holder selection stays a T15 owner 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 --- ..._registration_original_intent_reconcile.py | 280 ++++++++++++++++++ 1 file changed, 280 insertions(+) create mode 100644 sso-mfa/k8s/keycape/test_registration_original_intent_reconcile.py 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()