"""Synthetic-only tests for the T27 bind-env candidate; no cluster, no values.""" import copy from dataclasses import asdict, replace import json import importlib.util from pathlib import Path import unittest from unittest.mock import patch from warden.case_run import Invocation, OwnerResult, digest from warden.case_receipt_state import Outcome, Pins, Refused import yaml 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 setUp(self): self.case = yaml.safe_dump(CASE).encode() desc_bytes = json.dumps(asdict(a.descriptor()), sort_keys=True, separators=(',', ':')).encode() pins = Pins(digest(self.case), digest(b'type'), digest(desc_bytes), digest(b'catalog'), digest(b'case-acceptance:test'), digest(b'run sheet')) self.admitted = Invocation('bind-env', 'net-kingdom', 'forgejo', 'human', 'founder', 'keycape-attended-case-change', pins) self.inv = replace(self.admitted, attempt_id='attempt-one', operation='run') self.undo = replace(self.admitted, attempt_id='attempt-two', operation='rollback') self.evidence, self.revocations = {}, [] def publish(self, payload): self.assertNotIn(b'synthetic', payload) ref = 'owner-evidence:attempt-' + str(len(self.evidence)) self.evidence[ref] = payload return ref def revoke(self, inv): self.revocations.append(inv) return 'revocation:' + inv.attempt_id def adapter(self, store, **kw): args = dict(admitted=self.admitted, case_source=self.case, expected_resource_version='1', case_acceptance='case-acceptance:test', read=store.read, write=store.write, publish=self.publish, authenticate=lambda r, p: self.evidence.get(r) == p, revoke_session=self.revoke) args.update(kw) return a.build(**args) def env(self, store): return store.d['spec']['template']['spec']['containers'][0]['env'] 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_verify_then_created_only_rollback(self): s = Store(deployment()) ad = self.adapter(s) self.assertEqual((ad.reference, ad.source), (a.REFERENCE, a.source_digest())) r = ad.invoke(self.inv) self.assertEqual((r.outcome, r.created, r.session_revocation_evidence), (Outcome.SUCCESS, (a.CREATED,), 'revocation:attempt-one')) self.assertTrue(ad.verify(self.inv, r)) payload = json.loads(self.evidence[r.evidence]) self.assertEqual((payload['before'], payload['after'], payload['values_emitted']), ('1', '2', False)) self.assertEqual(len(self.env(s)), 2) u = ad.rollback(self.undo, (a.CREATED,)) self.assertEqual(u.outcome, Outcome.ROLLED_BACK) self.assertEqual(self.env(s), deployment()['spec']['template']['spec']['containers'][0]['env']) self.assertEqual(self.revocations, [self.inv, self.undo]) with self.assertRaises(Refused): ad.rollback(replace(self.undo, attempt_id='attempt-three'), (a.CREATED,)) self.assertEqual(len(self.revocations), 3) def test_existing_identical_binding_is_no_change_and_not_rollbackable(self): s = Store(deployment([a.binding_for_case(CASE)])) ad = self.adapter(s) self.assertEqual(ad.invoke(self.inv).outcome, Outcome.NO_CHANGE) self.assertEqual(s.writes, []) with self.assertRaises(Refused): ad.rollback(self.undo, (a.CREATED,)) def test_refusals_still_revoke_session(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)) with self.assertRaises(Refused): self.adapter(s).invoke(self.inv) self.assertEqual(s.writes, []) s = Store(deployment()) with self.assertRaises(Refused): self.adapter(s, expected_resource_version='0').invoke(self.inv) self.assertEqual((s.writes, len(self.revocations), self.evidence), ([], 3, {})) public = {'spec': {'type': 'oidc-browser-relying-party@1.1', 'parameters': dict(f.FORGEJO_PARAMS, client_type='public', mfa_required='false')}} with self.assertRaises(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(Refused): a.plan(dict(deployment(), metadata={'name': 'keycape', 'namespace': 'other'}), a.binding_for_case(CASE)) with self.assertRaises(Refused): a.plan(deployment(), dict(a.binding_for_case(CASE), value='synthetic')) def test_build_pins_and_capabilities_required(self): s = Store(deployment()) pins = self.admitted.pins for kw in (dict(allow_unimplemented=True), dict(case_source=self.case + b' '), dict(case_acceptance='case-acceptance:other'), dict(admitted=replace(self.admitted, pins=replace(pins, adapter=digest(b'x')))), dict(admitted=replace(self.admitted, step='register-client')), dict(admitted=replace(self.admitted, target='other')), dict(admitted=self.inv), dict(publish=None), dict(revoke_session='cmd'), dict(write=None)): with self.assertRaises(Exception, msg=kw): self.adapter(s, **kw) def test_rebinding_and_source_drift_refused_before_write(self): s = Store(deployment()) ad = self.adapter(s) with self.assertRaises(Refused): ad.invoke(replace(self.inv, principal='someone-else')) with patch.object(a, 'source_digest', return_value=digest(b'drift')): with self.assertRaises(Refused): ad.invoke(self.inv) 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) r = ad.invoke(self.inv) self.assertFalse(ad.verify(self.undo, r)) self.assertFalse(ad.verify(self.inv, replace(r, evidence='owner-evidence:forged'))) self.evidence[r.evidence] = b'tampered' self.assertFalse(ad.verify(self.inv, r)) with self.assertRaises(Refused): ad.rollback(self.undo, (a.CREATED,)) self.assertEqual(len(self.env(s)), 2) def test_process_loss_does_not_reconstruct_created_authority(self): s = Store(deployment()) self.adapter(s).invoke(self.inv) with self.assertRaises(Refused): self.adapter(s, expected_resource_version='2').rollback(self.undo, (a.CREATED,)) self.assertEqual(len(self.env(s)), 2) def test_revocation_or_publication_failure_is_never_success(self): for kw in (dict(revoke_session=lambda inv: None), dict(publish=lambda p: 'bad id!')): s = Store(deployment()) with self.assertRaises(Refused): self.adapter(s, **kw).invoke(self.inv) def test_rollback_refused_after_drift(self): s = Store(deployment()) ad = self.adapter(s) ad.invoke(self.inv) s.d['metadata']['resourceVersion'] = '9' with self.assertRaises(Refused): ad.rollback(self.undo, (a.CREATED,)) self.assertEqual(len(self.env(s)), 2) if __name__ == '__main__': unittest.main()