Pin bind-env full local source closure (NK48-R2-F3)
bind-env's source_digest covered bind-env-adapter.py and keycape-client.py only, but keycape-client.py executes portal-client-rollout.py, which executes openbao-client-config.py and register-user-engine-portal.py. Pin the same five-file closure the registration adapter pins, and add a closure-completeness test that imports every *-adapter.py in a fresh interpreter and fails on any loaded local file its SOURCE_FILES omits. 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
This commit is contained in:
parent
68f4bbb8ca
commit
d16db30282
4 changed files with 107 additions and 3 deletions
|
|
@ -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
|
- `read`/`write` are injected capabilities; `write` is compare-and-swap on the
|
||||||
observed `resourceVersion`. No kubectl, credential or secret value is used.
|
observed `resourceVersion`. No kubectl, credential or secret value is used.
|
||||||
- `build(...)` returns a closed Warden v1 `Adapter` (T13 pattern). It pins the
|
- `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
|
descriptor digest (`Pins.adapter`), case source and acceptance digests, and the
|
||||||
admitted invocation (step `bind-env`, actor `human`, lane `key-cape-oidc-login`,
|
admitted invocation (step `bind-env`, actor `human`, lane `key-cape-oidc-login`,
|
||||||
created object `keycape-env-binding`, not retry-safe, `parent-session` scope).
|
created object `keycape-env-binding`, not retry-safe, `parent-session` scope).
|
||||||
|
|
|
||||||
|
|
@ -24,7 +24,11 @@ from warden.case_receipt_state import Outcome, Refused
|
||||||
|
|
||||||
ROOT = Path(__file__).resolve().parent
|
ROOT = Path(__file__).resolve().parent
|
||||||
REFERENCE = 'net-kingdom/keycape-bind-env@v1'
|
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'
|
CREATED = 'keycape-env-binding'
|
||||||
_spec = importlib.util.spec_from_file_location('bind_env_owner_helper', ROOT / 'keycape-client.py')
|
_spec = importlib.util.spec_from_file_location('bind_env_owner_helper', ROOT / 'keycape-client.py')
|
||||||
helper = importlib.util.module_from_spec(_spec)
|
helper = importlib.util.module_from_spec(_spec)
|
||||||
|
|
@ -74,7 +78,7 @@ def plan(deployment, binding):
|
||||||
|
|
||||||
|
|
||||||
def source_digest():
|
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}}
|
composition = {'files': {name: digest((ROOT / name).read_bytes()) for name in SOURCE_FILES}}
|
||||||
return digest(json.dumps(composition, sort_keys=True, separators=(',', ':')).encode())
|
return digest(json.dumps(composition, sort_keys=True, separators=(',', ':')).encode())
|
||||||
|
|
||||||
|
|
|
||||||
82
sso-mfa/k8s/keycape/test_adapter_source_closure.py
Normal file
82
sso-mfa/k8s/keycape/test_adapter_source_closure.py
Normal file
|
|
@ -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()
|
||||||
|
|
@ -157,6 +157,21 @@ class BindEnvTests(unittest.TestCase):
|
||||||
self.assertEqual(s.writes, [])
|
self.assertEqual(s.writes, [])
|
||||||
self.assertEqual(len(self.revocations), 2)
|
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):
|
def test_tampered_or_unauthenticated_evidence_never_authorizes_undo(self):
|
||||||
s = Store(deployment())
|
s = Store(deployment())
|
||||||
ad = self.adapter(s)
|
ad = self.adapter(s)
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue