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

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 903097@bnt-lap001
Assistant-Session: aa320c8f-c3fa-46e7-8541-0b58e1e45d23

sso-mfa/k8s/keycape/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.

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

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__':