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