From 7442d212ec7d5fcf1f13e3a2f82d9ac97e028beb Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 17 Oct 2026 08:02:15 +0200 Subject: [PATCH] NK-WP-0048-T27W: wrap bind-env as a full Warden Adapter Source-digest, descriptor, case and acceptance pins; owner-published and authenticated evidence; parent-session revocation in finally. Existing reference-only refusals and created-only rollback kept. Synthetic tests 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 | 24 ++- sso-mfa/k8s/keycape/bind-env-adapter.py | 173 +++++++++++++++---- sso-mfa/k8s/keycape/test_bind_env_adapter.py | 153 ++++++++++++---- 3 files changed, 282 insertions(+), 68 deletions(-) diff --git a/sso-mfa/k8s/keycape/bind-env-adapter.md b/sso-mfa/k8s/keycape/bind-env-adapter.md index df78b7e..3af06e2 100644 --- a/sso-mfa/k8s/keycape/bind-env-adapter.md +++ b/sso-mfa/k8s/keycape/bind-env-adapter.md @@ -14,13 +14,27 @@ bound to anything else (including a literal), is refused. 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. +- `build(...)` returns a closed Warden v1 `Adapter` (T13 pattern). It pins the + owner source digest over `bind-env-adapter.py` + `keycape-client.py`, the + descriptor digest (`Pins.adapter`), case source and acceptance digests, and the + admitted invocation (step `bind-env`, actor `human`, lane `key-cape-oidc-login`, + created object `keycape-env-binding`, not retry-safe, `parent-session` scope). + Every attempt is rebound to that invocation and re-checks source drift. +- `publish`, `authenticate` and `revoke_session` are mandatory injected + capabilities. The parent session is revoked in `finally`, also on refusal; a + missing revocation or invalid/reused evidence identifier is never success. + Evidence payload is closed metadata (`netkingdom.keycape-bind-env.attempt.v1`: + invocation, source, outcome, env name, before/after revision) with no values. - An identical existing binding is `no-change` and is not rollbackable. -- Rollback removes only the entry this process created, and only while the +- Rollback removes only the entry this process created, whose evidence still + authenticates (a fresh process has no creation authority), 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). +Tests: `PYTHONPATH=~/ops-warden/src ~/ops-warden/.venv/bin/python -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. +Not supplied: kernel-level admission/dispatch tests with fixture trust (as in T13), +owner confirmation of the proposed step/lane/schema/object names, NetKingdom +runtime owner source review before pinning, the real owner evidence and +revocation capabilities, 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 index f1b2d53..2e61140 100644 --- a/sso-mfa/k8s/keycape/bind-env-adapter.py +++ b/sso-mfa/k8s/keycape/bind-env-adapter.py @@ -1,29 +1,36 @@ -"""Unexercised source candidate: bind one issuer env reference, never a value (NK-WP-0048-T27). +"""Unexercised, injection-only bind-env seam for Warden's v1 kernel (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. +Only an entry this process created, authenticated and still byte-identical, may +be rolled back. Owner publication/authentication and parent-session revocation +are mandatory injected capabilities; no evidence trust root is installed here. """ from __future__ import annotations import copy +from dataclasses import asdict, replace import importlib.util +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 REFERENCE = 'net-kingdom/keycape-bind-env@v1' +SOURCE_FILES = ('bind-env-adapter.py', 'keycape-client.py') +CREATED = 'keycape-env-binding' _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' @@ -66,45 +73,145 @@ def plan(deployment, 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 source_digest(): + """Pin adapter plus its local helper import closure, in filename order.""" + composition = {'files': {name: digest((ROOT / name).read_bytes()) for name in SOURCE_FILES}} + return digest(json.dumps(composition, sort_keys=True, separators=(',', ':')).encode()) - def run(): + +def descriptor(): + return Descriptor('warden.case.execution.v1', 'bind-env', 'net-kingdom', REFERENCE, + source_digest(), 'human', 'key-cape-oidc-login', 'client_id', 'change', + (CREATED,), False, ('parent-session',)) + + +def build(*, admitted: Invocation, case_source: bytes, expected_resource_version: str, + case_acceptance: str, read, write, publish, authenticate, revoke_session, + allow_unimplemented=False): + """Bind one admitted case and observed revision; return a closed kernel Adapter. + + read()->deployment dict; write(deployment, expected_rv)->new rv (CAS, raises + on conflict). publish(bytes)->identifier retains only the closed metadata + payload; authenticate(identifier, bytes)->True verifies owner custody; + revoke_session(invocation)->identifier revokes this attempt's parent session. + Process loss, readback ambiguity and publication/revocation failure require + owner reconciliation; none are converted to retry-safe failures. + """ + reviewed_source = source_digest() + desc = descriptor() + if type(admitted) is not Invocation or not all( + isinstance(value, str) and IDENTIFIER.fullmatch(value) + for value 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('bind-env 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') + # Pins.adapter is the descriptor digest, not the owner source digest. + descriptor_bytes = json.dumps(asdict(desc), sort_keys=True, separators=(',', ':')).encode() + if admitted.pins.adapter != digest(descriptor_bytes): + raise Refused('descriptor pin differs') + if not all(isinstance(x, str) and 0 < len(x) <= 192 + for x in (expected_resource_version, case_acceptance)): + raise Refused('observed resourceVersion and acceptance required') + if digest(case_acceptance.encode()) != admitted.pins.acceptance: + raise Refused('acceptance evidence binding differs') + if not all(callable(x) for x in (read, write, publish, authenticate, revoke_session)): + raise Refused('read/write, owner evidence and revocation capabilities required') + with contained_output(): + case = yaml.safe_load(case_source) + client = helper.client_from_case(case) + binding = binding_for_case(case, allow_unimplemented=allow_unimplemented) + if admitted.target != client['clientId']: + raise Refused('case target differs') + records = {} + creation = None + + def bound(inv, operation): + if (type(inv) is not Invocation + or 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_source: + raise Refused('owner source changed') + + def add(): current = read() - rv = current['metadata'].get('resourceVersion') - if rv != expected_resource_version: + before = current['metadata'].get('resourceVersion') + if before != 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) + return Outcome.NO_CHANGE, (), before, before + after = write(new, before) back = read() - if back['metadata'].get('resourceVersion') != after or plan(back, binding)[1]: + if (not isinstance(after, str) or not 0 < len(after) <= 192 + or 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} + return Outcome.SUCCESS, (CREATED,), before, after - def rollback(): - if state['created_rv'] is None: - raise Refused('no case-created binding to roll back') + def remove(created): + if created != (CREATED,) or creation is None or not verify(*creation): + raise Refused('no authenticated case-created binding') + created_rv = json.loads(records[creation[1].evidence][1])['after'] current = read() - rv = current['metadata'].get('resourceVersion') - if rv != state['created_rv']: + before = current['metadata'].get('resourceVersion') + if before != created_rv: raise Refused('deployment changed since creation; owner reconciliation required') new = copy.deepcopy(current) - env = _container(new)['env'] + env = _container(new).get('env', []) matches = [i for i, e in enumerate(env) if e == binding] - if len(matches) != 1: + if len(matches) != 1 or [e for e in env if e.get('name') == binding['name']] != [binding]: 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} + after = write(new, before) + back = read() + if (not isinstance(after, str) or back['metadata'].get('resourceVersion') != after + or any(e.get('name') == binding['name'] for e in _container(back).get('env', []))): + raise Refused('rollback readback ambiguous; owner reconciliation required') + return Outcome.ROLLED_BACK, (), before, after - return REFERENCE, run, rollback + def execute(inv, operation, created=()): + nonlocal creation + try: + with contained_output(): + bound(inv, 'run' if operation == 'add' else 'rollback') + outcome, objects, before, after = add() if operation == 'add' else remove(created) + finally: + # Even failure must end the parent session; failures propagate to the + # kernel as uncertain, never safe to replay. + 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.keycape-bind-env.attempt.v1', + 'invocation': asdict(inv), 'source': reviewed_source, 'outcome': outcome.value, + 'created': objects, 'env_name': binding['name'], 'before': before, 'after': after, + 'values_emitted': False, '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_source, lambda inv: execute(inv, 'add'), + verify, lambda inv, created: execute(inv, 'remove', created)) diff --git a/sso-mfa/k8s/keycape/test_bind_env_adapter.py b/sso-mfa/k8s/keycape/test_bind_env_adapter.py index 0812acc..fe35626 100644 --- a/sso-mfa/k8s/keycape/test_bind_env_adapter.py +++ b/sso-mfa/k8s/keycape/test_bind_env_adapter.py @@ -1,8 +1,15 @@ """Synthetic-only tests for the T27 bind-env candidate; no cluster, no values.""" import copy +from dataclasses import asdict, replace +import json import importlib.util from pathlib import Path import unittest +from unittest.mock import patch + +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('bind_env_adapter', ROOT / 'bind-env-adapter.py') @@ -38,65 +45,151 @@ class Store: class BindEnvTests(unittest.TestCase): + def setUp(self): + self.case = yaml.safe_dump(CASE).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'case-acceptance:test'), digest(b'run sheet')) + self.admitted = Invocation('bind-env', 'net-kingdom', 'forgejo', 'human', + 'founder', 'key-cape-oidc-login', pins) + self.inv = replace(self.admitted, attempt_id='attempt-one', operation='run') + self.undo = replace(self.admitted, attempt_id='attempt-two', operation='rollback') + self.evidence, self.revocations = {}, [] + + def publish(self, payload): + self.assertNotIn(b'synthetic', payload) + ref = 'owner-evidence:attempt-' + str(len(self.evidence)) + self.evidence[ref] = payload + return ref + + def revoke(self, inv): + self.revocations.append(inv) + return 'revocation:' + inv.attempt_id + + def adapter(self, store, **kw): + args = dict(admitted=self.admitted, case_source=self.case, expected_resource_version='1', + case_acceptance='case-acceptance:test', read=store.read, write=store.write, + publish=self.publish, authenticate=lambda r, p: self.evidence.get(r) == p, + revoke_session=self.revoke) + args.update(kw) + return a.build(**args) + + def env(self, store): + return store.d['spec']['template']['spec']['containers'][0]['env'] + 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): + def test_run_verify_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() + ad = self.adapter(s) + self.assertEqual((ad.reference, ad.source), (a.REFERENCE, a.source_digest())) + r = ad.invoke(self.inv) + self.assertEqual((r.outcome, r.created, r.session_revocation_evidence), (Outcome.SUCCESS, (a.CREATED,), 'revocation:attempt-one')) + self.assertTrue(ad.verify(self.inv, r)) + payload = json.loads(self.evidence[r.evidence]) + self.assertEqual((payload['before'], payload['after'], payload['values_emitted']), ('1', '2', False)) + self.assertEqual(len(self.env(s)), 2) + u = ad.rollback(self.undo, (a.CREATED,)) + self.assertEqual(u.outcome, Outcome.ROLLED_BACK) + self.assertEqual(self.env(s), deployment()['spec']['template']['spec']['containers'][0]['env']) + self.assertEqual(self.revocations, [self.inv, self.undo]) + with self.assertRaises(Refused): + ad.rollback(replace(self.undo, attempt_id='attempt-three'), (a.CREATED,)) + self.assertEqual(len(self.revocations), 3) 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') + ad = self.adapter(s) + self.assertEqual(ad.invoke(self.inv).outcome, Outcome.NO_CHANGE) self.assertEqual(s.writes, []) - with self.assertRaises(a.Refused): - rollback() + with self.assertRaises(Refused): + ad.rollback(self.undo, (a.CREATED,)) - def test_refusals(self): + def test_refusals_still_revoke_session(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() + with self.assertRaises(Refused): + self.adapter(s).invoke(self.inv) 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) + with self.assertRaises(Refused): + self.adapter(s, expected_resource_version='0').invoke(self.inv) + self.assertEqual((s.writes, len(self.revocations), self.evidence), ([], 3, {})) 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): + with self.assertRaises(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): + with self.assertRaises(Refused): a.plan(dict(deployment(), metadata={'name': 'keycape', 'namespace': 'other'}), a.binding_for_case(CASE)) - with self.assertRaises(a.Refused): + with self.assertRaises(Refused): a.plan(deployment(), dict(a.binding_for_case(CASE), value='synthetic')) + def test_build_pins_and_capabilities_required(self): + s = Store(deployment()) + pins = self.admitted.pins + for kw in (dict(allow_unimplemented=True), + dict(case_source=self.case + b' '), + dict(case_acceptance='case-acceptance:other'), + dict(admitted=replace(self.admitted, pins=replace(pins, adapter=digest(b'x')))), + dict(admitted=replace(self.admitted, step='register-client')), + dict(admitted=replace(self.admitted, target='other')), + dict(admitted=self.inv), + dict(publish=None), dict(revoke_session='cmd'), dict(write=None)): + with self.assertRaises(Exception, msg=kw): + self.adapter(s, **kw) + + def test_rebinding_and_source_drift_refused_before_write(self): + s = Store(deployment()) + ad = self.adapter(s) + with self.assertRaises(Refused): + ad.invoke(replace(self.inv, principal='someone-else')) + with patch.object(a, 'source_digest', return_value=digest(b'drift')): + with self.assertRaises(Refused): + ad.invoke(self.inv) + self.assertEqual(s.writes, []) + self.assertEqual(len(self.revocations), 2) + + def test_tampered_or_unauthenticated_evidence_never_authorizes_undo(self): + s = Store(deployment()) + ad = self.adapter(s) + r = ad.invoke(self.inv) + self.assertFalse(ad.verify(self.undo, r)) + self.assertFalse(ad.verify(self.inv, replace(r, evidence='owner-evidence:forged'))) + self.evidence[r.evidence] = b'tampered' + self.assertFalse(ad.verify(self.inv, r)) + with self.assertRaises(Refused): + ad.rollback(self.undo, (a.CREATED,)) + self.assertEqual(len(self.env(s)), 2) + + def test_process_loss_does_not_reconstruct_created_authority(self): + s = Store(deployment()) + self.adapter(s).invoke(self.inv) + with self.assertRaises(Refused): + self.adapter(s, expected_resource_version='2').rollback(self.undo, (a.CREATED,)) + self.assertEqual(len(self.env(s)), 2) + + def test_revocation_or_publication_failure_is_never_success(self): + for kw in (dict(revoke_session=lambda inv: None), dict(publish=lambda p: 'bad id!')): + s = Store(deployment()) + with self.assertRaises(Refused): + self.adapter(s, **kw).invoke(self.inv) + 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() + ad = self.adapter(s) + ad.invoke(self.inv) s.d['metadata']['resourceVersion'] = '9' - with self.assertRaises(a.Refused): - rollback() + with self.assertRaises(Refused): + ad.rollback(self.undo, (a.CREATED,)) + self.assertEqual(len(self.env(s)), 2) if __name__ == '__main__':