diff --git a/sso-mfa/k8s/keycape/bind-env-adapter.md b/sso-mfa/k8s/keycape/bind-env-adapter.md index 3af06e2..39b899d 100644 --- a/sso-mfa/k8s/keycape/bind-env-adapter.md +++ b/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). diff --git a/sso-mfa/k8s/keycape/bind-env-adapter.py b/sso-mfa/k8s/keycape/bind-env-adapter.py index 2e61140..817fd0f 100644 --- a/sso-mfa/k8s/keycape/bind-env-adapter.py +++ b/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()) diff --git a/sso-mfa/k8s/keycape/test_adapter_source_closure.py b/sso-mfa/k8s/keycape/test_adapter_source_closure.py new file mode 100644 index 0000000..d44ac8b --- /dev/null +++ b/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() diff --git a/sso-mfa/k8s/keycape/test_bind_env_adapter.py b/sso-mfa/k8s/keycape/test_bind_env_adapter.py index fe35626..b80af14 100644 --- a/sso-mfa/k8s/keycape/test_bind_env_adapter.py +++ b/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)