Merge NK48 r2 finding fixes (R2-F1 mint revocation, R2-F2 deliver-secret case binding, R2-F3 bind-env closure)

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

@@ -15,7 +15,10 @@ bound to anything else (including a literal), is 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
+  owner source digest over the full local import closure (`bind-env-adapter.py`,
+  `keycape-client.py`, `portal-client-rollout.py`, `openbao-client-config.py`,
+  `register-user-engine-portal.py` — closure checked by
+  `test_adapter_source_closure.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).

sso-mfa/k8s/keycape/bind-env-adapter.py

@@ -24,7 +24,11 @@ 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')
+# keycape-client.py loads portal-client-rollout.py at import, which loads
+# openbao-client-config.py and register-user-engine-portal.py; all of them run.
+SOURCE_FILES = ('bind-env-adapter.py', 'keycape-client.py',
+                'portal-client-rollout.py', 'openbao-client-config.py',
+                'register-user-engine-portal.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)
@@ -74,7 +78,7 @@ def plan(deployment, binding):
 
 
 def source_digest():
-    """Pin adapter plus its local helper import closure, in filename order."""
+    """Pin adapter plus the complete 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())
 

sso-mfa/k8s/keycape/deliver-secret-adapter.md

@@ -11,8 +11,14 @@ 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.
+It binds one admitted invocation to one Secret reference. `build()` takes
+`case_source` and refuses unless its digest equals `admitted.pins.case`, the
+case is `client_type: confidential`, and its `client_id` equals the admitted
+target (as `mint-secret` and `bind-env` do). The reference is then derived from
+the admitted target per type 1.2, never chosen by the caller: ExternalSecret
+`sso/keycape-{client_id}-client`, key `client-secret`, from custody
+`platform/workloads/keycape/{client_id}` (NK48-R2-F2). Owner receipts must name
+that same reference. 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.

sso-mfa/k8s/keycape/deliver-secret-adapter.py

@@ -13,6 +13,8 @@ import json
 import re
 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
 
@@ -20,7 +22,8 @@ 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])?')
+NAMESPACE = 'sso'
+KEY_NAMES = ('client-secret',)
 RECEIPT_FIELDS = {'schema', 'operation', 'namespace', 'name', 'custody_reference',
                   'key_names', 'values_emitted', 'changed', 'before', 'after'}
 
@@ -41,10 +44,21 @@ def _revision(value):
             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.
+def secret_name(client_id):
+    return 'keycape-' + client_id + '-client'
+
+
+def custody_path(client_id):
+    return 'platform/workloads/keycape/' + client_id
+
+
+def build(*, admitted: Invocation, case_source: bytes, case_acceptance: str,
+          deliver, remove, publish, authenticate, revoke_session):
+    """Bind one admitted confidential-client case; return a kernel Adapter.
+
+    The Secret reference is derived from the admitted target per type 1.2:
+    ExternalSecret sso/keycape-{client_id}-client (key client-secret) from
+    bao:platform/workloads/keycape/{client_id}. No caller chooses a reference.
 
     deliver(inv, binding)->receipt and remove(inv, binding, after)->receipt are
     owner capabilities (e.g. ESO sync with readback); receipts carry metadata only.
@@ -62,22 +76,27 @@ def build(*, admitted: Invocation, case_acceptance: str, namespace: str, name: s
         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 (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')
+    with contained_output():
+        case = yaml.safe_load(case_source)
+    params = (case or {}).get('spec', {}).get('parameters', {}) if isinstance(case, dict) else {}
+    if not isinstance(params, dict) or params.get('client_type') != 'confidential':
+        raise Refused('deliver-secret applies only to confidential clients')
+    if params.get('client_id') != admitted.target:
+        raise Refused('case target differs')
+    namespace, name = NAMESPACE, secret_name(admitted.target)
+    custody_reference, key_names = custody_path(admitted.target), KEY_NAMES
+    if not re.fullmatch(r'[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?', name):
+        raise Refused('bounded secret reference required')
     binding_ref = {'namespace': namespace, 'name': name, 'custody_reference': custody_reference,
                    'key_names': list(key_names), 'client_id': admitted.target}
     records = {}

sso-mfa/k8s/keycape/mint-secret-adapter.md

@@ -15,6 +15,8 @@ modelled on the T13 registration adapter.
   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.
+  Each revocation is attempted independently: if one raises, the other is still
+  called, and the attempt is refused rather than reported as success.
 - Confidential clients only; case, descriptor, acceptance and attempt pins bound
   as in T13. Not retry-safe.
 

sso-mfa/k8s/keycape/mint-secret-adapter.py

@@ -113,8 +113,21 @@ def build(*, admitted: Invocation, case_source: bytes, case_acceptance: str,
                     receipt = destroy(path, version)
                     check(receipt, 'destroy', version)
         finally:
+            # Each revocation is attempted independently: a failing session
+            # revocation must not leave the broker grant alive, or the reverse.
+            # The first failure is re-raised only after both were attempted.
             with contained_output():
-                session, grant = revoke_session(inv), revoke_grant(inv)
+                session = grant = failure = None
+                try:
+                    session = revoke_session(inv)
+                except BaseException as exc:
+                    failure = exc
+                try:
+                    grant = revoke_grant(inv)
+                except BaseException as exc:
+                    failure = failure or exc
+                if failure is not None:
+                    raise failure
         for ref in (session, grant):
             if not isinstance(ref, str) or not IDENTIFIER.fullmatch(ref):
                 raise Refused('revocation evidence missing')

sso-mfa/k8s/keycape/test_adapter_source_closure.py

@@ -0,0 +1,82 @@
+"""Every adapter's source_digest must cover every local file it executes.
+
+Each *-adapter.py is imported in a fresh interpreter with
+importlib.util.spec_from_file_location instrumented; the local .py files
+actually loaded (sys.modules __file__ under the repo plus every
+spec_from_file_location target) must be a subset of its SOURCE_FILES.
+"""
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import unittest
+
+ROOT = Path(__file__).resolve().parent
+REPO = ROOT.parents[2]
+
+PROBE = r'''
+import importlib.util, json, sys
+from pathlib import Path
+root, repo, adapter = Path(sys.argv[1]), Path(sys.argv[2]), sys.argv[3]
+seen = set()
+_orig = importlib.util.spec_from_file_location
+def _spy(name, location=None, *a, **k):
+    if location is not None:
+        seen.add(str(Path(location).resolve()))
+    return _orig(name, location, *a, **k)
+importlib.util.spec_from_file_location = _spy
+spec = _orig('adapter_under_probe', root / adapter)
+mod = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(mod)
+for m in list(sys.modules.values()):
+    f = getattr(m, '__file__', None)
+    if f:
+        seen.add(str(Path(f).resolve()))
+local = sorted(p for p in seen
+               if p.endswith('.py') and Path(p).is_relative_to(repo)
+               and '.venv' not in Path(p).parts and 'site-packages' not in Path(p).parts)
+local.append(str((root / adapter).resolve()))
+pinned = []
+for entry in mod.SOURCE_FILES:
+    pinned.append(str((repo / entry[0] / entry[1]).resolve()) if isinstance(entry, tuple)
+                  else str((root / entry).resolve()))
+print(json.dumps({'loaded': sorted(set(local)), 'pinned': sorted(pinned)}))
+'''
+
+# Adapters outside this fix's scope that fail the check (reported, not fixed).
+KNOWN_FAILING = set()
+
+
+def probe(adapter):
+    out = subprocess.run([sys.executable, '-c', PROBE, str(ROOT), str(REPO), adapter],
+                         capture_output=True, text=True, env=dict(os.environ), cwd=ROOT)
+    if out.returncode:
+        raise AssertionError(f'{adapter} probe failed: {out.stderr[-2000:]}')
+    return json.loads(out.stdout.strip().splitlines()[-1])
+
+
+def unpinned(adapter):
+    r = probe(adapter)
+    return sorted(set(r['loaded']) - set(r['pinned']))
+
+
+class AdapterSourceClosure(unittest.TestCase):
+    pass
+
+
+def _make(adapter):
+    def test(self):
+        missing = unpinned(adapter)
+        self.assertEqual(missing, [], f'{adapter} loads files its source_digest does not pin')
+    if adapter in KNOWN_FAILING:
+        test = unittest.expectedFailure(test)
+    return test
+
+
+for _p in sorted(ROOT.glob('*-adapter.py')):
+    setattr(AdapterSourceClosure, 'test_' + _p.stem.replace('-', '_'), _make(_p.name))
+
+
+if __name__ == '__main__':
+    unittest.main()

sso-mfa/k8s/keycape/test_bind_env_adapter.py

@@ -157,6 +157,21 @@ class BindEnvTests(unittest.TestCase):
         self.assertEqual(s.writes, [])
         self.assertEqual(len(self.revocations), 2)
 
+    def test_source_digest_covers_executed_helper_closure(self):
+        # NK48-R2-F3: keycape-client.py executes portal-client-rollout.py,
+        # which executes openbao-client-config.py and register-user-engine-portal.py.
+        import shutil, tempfile
+        before = a.source_digest()
+        for name in ('portal-client-rollout.py', 'openbao-client-config.py',
+                     'register-user-engine-portal.py'):
+            with tempfile.TemporaryDirectory() as tmp:
+                for f in a.ROOT.glob('*.py'):
+                    shutil.copy(f, Path(tmp) / f.name)
+                with open(Path(tmp) / name, 'a') as fh:
+                    fh.write('\n# drift\n')
+                with patch.object(a, 'ROOT', Path(tmp)):
+                    self.assertNotEqual(a.source_digest(), before, name)
+
     def test_tampered_or_unauthenticated_evidence_never_authorizes_undo(self):
         s = Store(deployment())
         ad = self.adapter(s)

sso-mfa/k8s/keycape/test_deliver_secret_adapter.py

@@ -14,13 +14,18 @@ a = importlib.util.module_from_spec(spec)
 spec.loader.exec_module(a)
 
 KEYS = ('client-secret',)
+EXAMPLES = ROOT.parents[2] / 'security-case-type-revisions' / '1.2' / 'oidc-browser-relying-party' / 'examples'
+CASE = (EXAMPLES / 'forgejo.yaml').read_bytes()
+PUBLIC_CASE = (EXAMPLES / 'example-public-portal.yaml').read_bytes()
+SECRET_NAME = 'keycape-forgejo-client'
+CUSTODY = 'platform/workloads/keycape/forgejo'
 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'),
+        pins = Pins(digest(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)
@@ -28,16 +33,15 @@ class DeliverSecretTests(unittest.TestCase):
         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,
+        self.kw = dict(admitted=self.admitted, case_source=CASE,
+                       case_acceptance='case-acceptance:test',
                        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',
+             'name': SECRET_NAME, 'custody_reference': CUSTODY,
              'key_names': list(KEYS), 'values_emitted': False, 'changed': changed,
              'before': before, 'after': after}
         r.update(extra)
@@ -92,14 +96,49 @@ class DeliverSecretTests(unittest.TestCase):
                 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),
+        cases = [dict(case_acceptance='other'), dict(case_source=CASE.decode()), 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_reference_derived_from_admitted_target_per_type(self):
+        a.build(**self.kw).invoke(self.inv)
+        self.assertEqual(self.calls[0], ('deliver', {
+            'namespace': 'sso', 'name': SECRET_NAME, 'custody_reference': CUSTODY,
+            'key_names': list(KEYS), 'client_id': 'forgejo'}))
+        for caller_ref in ('namespace', 'name', 'custody_reference', 'key_names'):
+            with self.subTest(caller_ref=caller_ref), self.assertRaises(TypeError):
+                a.build(**{**self.kw, caller_ref: 'x'})
+
+    def test_reviewer_demo_admitted_forgejo_grafana_delivery_refused(self):
+        grafana = CASE.replace(b'client_id: forgejo', b'client_id: grafana')
+        # Same admitted forgejo case, grafana case source: pin differs.
+        with self.assertRaises(Refused):
+            a.build(**{**self.kw, 'case_source': grafana})
+        # Admission re-pinned to the grafana source but still targeting forgejo.
+        repinned = replace(self.admitted, pins=replace(self.admitted.pins, case=digest(grafana)))
+        with self.assertRaises(Refused):
+            a.build(**{**self.kw, 'admitted': repinned, 'case_source': grafana})
+        # A grafana-shaped receipt under the forgejo admission is refused.
+        self.kw['deliver'] = lambda inv, b: self.receipt(
+            'deliver', True, None, dict(AFTER), name='keycape-grafana-client',
+            custody_reference='platform/workloads/keycape/grafana')
+        with self.assertRaises(Refused):
+            a.build(**self.kw).invoke(self.inv)
+
+    def test_public_client_case_refused(self):
+        admitted = replace(self.admitted, target='example-portal',
+                           pins=replace(self.admitted.pins, case=digest(PUBLIC_CASE)))
+        with self.assertRaises(Refused):
+            a.build(**{**self.kw, 'admitted': admitted, 'case_source': PUBLIC_CASE})
+
+    def test_case_pin_mismatch_refused(self):
+        with self.assertRaises(Refused):
+            a.build(**{**self.kw, 'case_source': CASE + b'\n# edited\n'})
+        self.assertEqual(self.calls, [])
+
     def test_attempt_binding_and_reused_evidence_refused(self):
         adapter = a.build(**self.kw)
         with self.assertRaises(Refused):

sso-mfa/k8s/keycape/test_mint_secret_adapter.py

@@ -90,6 +90,37 @@ class MintSecretTests(unittest.TestCase):
         self.assertEqual([c[0] for c in self.calls], [])
         self.assertEqual(self.revoked, [('session', 'rollback'), ('grant', 'rollback')])
 
+    def _failing(self, kind):
+        def revoke(inv):
+            self.revoked.append((kind, inv.operation))
+            raise RuntimeError(kind + ' revocation unavailable')
+        return revoke
+
+    def test_failing_session_revocation_still_revokes_grant(self):
+        self.kw['revoke_session'] = self._failing('session')
+        adapter = a.build(**self.kw)
+        with self.assertRaises(Exception):
+            adapter.invoke(self.run)
+        self.assertEqual(self.revoked, [('session', 'run'), ('grant', 'run')])
+        self.assertEqual(self.evidence, {})   # never published as success
+
+    def test_failing_grant_revocation_still_revokes_session(self):
+        self.kw['revoke_grant'] = self._failing('grant')
+        adapter = a.build(**self.kw)
+        with self.assertRaises(Exception):
+            adapter.invoke(self.run)
+        self.assertEqual(self.revoked, [('session', 'run'), ('grant', 'run')])
+        self.assertEqual(self.evidence, {})
+
+    def test_both_revocations_failing_attempts_both_and_refuses(self):
+        self.kw['revoke_session'] = self._failing('session')
+        self.kw['revoke_grant'] = self._failing('grant')
+        adapter = a.build(**self.kw)
+        with self.assertRaises(Exception):
+            adapter.rollback(self.back, a.CREATED)
+        self.assertEqual(self.revoked, [('session', 'rollback'), ('grant', 'rollback')])
+        self.assertEqual(self.evidence, {})
+
     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'),