NK-WP-0048-T27: bind-env adapter source candidate (references 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

@@ -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>_CLIENT_SECRET`) to `valueFrom.secretKeyRef` on the delivered
+secret reference (`keycape-<client>-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.

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

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