diff --git a/sso-mfa/k8s/keycape/deliver-consumer-secret-adapter.md b/sso-mfa/k8s/keycape/deliver-consumer-secret-adapter.md new file mode 100644 index 0000000..bf229a0 --- /dev/null +++ b/sso-mfa/k8s/keycape/deliver-consumer-secret-adapter.md @@ -0,0 +1,38 @@ +# Deliver-consumer-secret adapter candidate + +Exercise status: unexercised + +Source-only, injection-only evidence seam for the type 1.2 step +`deliver-consumer-secret` (NK-WP-0048-T28), modelled on the T13 registration +adapter. Warden does not execute the consumer owner's apply +(`warden_executes: false`); the adapter binds one admitted case and accepts a +closed, value-free owner status receipt. It installs no command, kubectl call, +credential provider, OpenBao access or evidence trust root. Canonical 1.1 stays +the default: a 1.1 case, or a public client, is refused because the step does not +exist there. No `--allow-unimplemented` admission is used or implied. + +Binding: descriptor (`agent` actor, lane `consumer-owner-deployment`, effect +`change`, not retry-safe, `parent-session` revocation) pins this adapter source; +case, descriptor, target (`client_id`) and owner (`consumer_owner_repo`) must match +the admitted invocation; each attempt is rebound and the source rehashed. + +Receipt contract (`netkingdom.consumer-secret-delivery.v1`): per object +(`consumer-eso-sa`, `consumer-secretstore`, `consumer-externalsecret`) a uid, a +`created` flag and fixed metadata that must equal what the case parameters derive: +SA automount false with no legacy token refs; SecretStore with the dedicated +serviceAccountRef, audiences `[openbao]`, the case role and Ready; ExternalSecret +with Owner creation, Retain deletion, path `workloads/keycape/{client_id}`, only +key `CLIENT_SECRET`, Ready. Any value-bearing key (`data`, `stringData`, `value`, +`token`, `secret`, ...), `values_emitted` other than false, or metadata drift is +refused. Published evidence carries status, uids and revocation only. + +Rollback: only objects the owner reported created in this process, with +authenticated evidence and the exact created set, are removed — ExternalSecret, +then SecretStore, then SA — and only after the owner reports the consumer secret +reference reverted. Removed uids must equal the created uids. Preexisting-matched +objects are never deleted. Removing SecretStore/SA is not OpenBao token revocation. + +Tests: `PYTHONPATH=~/ops-warden/src python3 -m pytest test_deliver_consumer_secret_adapter.py` +(synthetic fixtures only). Not supplied here: Platform and Apps owner source +review before pinning, ESO TokenRequest/RBAC admission validation, and any native +action — each a separate gate. diff --git a/sso-mfa/k8s/keycape/deliver-consumer-secret-adapter.py b/sso-mfa/k8s/keycape/deliver-consumer-secret-adapter.py new file mode 100644 index 0000000..f3b592d --- /dev/null +++ b/sso-mfa/k8s/keycape/deliver-consumer-secret-adapter.py @@ -0,0 +1,193 @@ +"""Unexercised, injection-only consumer-secret delivery evidence seam (type 1.2). + +Source candidate for step deliver-consumer-secret. Warden does not execute the +consumer owner's apply (warden_executes: false); this adapter only binds one +admitted case, invokes an injected owner capability, and accepts a closed, +value-free status receipt. No CLI, kubectl, credential, OpenBao call or evidence +trust root is installed here. Canonical type 1.1 remains the default. +""" +from __future__ import annotations + +from dataclasses import asdict, replace +import json +from pathlib import Path + +import yaml + +from warden.case_run import Adapter, Descriptor, Invocation, OwnerResult, contained_output, digest, IDENTIFIER +from warden.case_receipt_state import Outcome, Refused + +ROOT = Path(__file__).resolve().parent +SOURCE_FILES = ('deliver-consumer-secret-adapter.py',) +REFERENCE = 'net-kingdom/deliver-consumer-secret@v1' +TYPE = 'oidc-browser-relying-party@1.2' +OBJECTS = ('consumer-eso-sa', 'consumer-secretstore', 'consumer-externalsecret') +# Rollback order: ExternalSecret/owned target, then SecretStore, then SA. +ROLLBACK_ORDER = tuple(reversed(OBJECTS)) +VALUE_KEYS = {'data', 'stringData', 'value', 'values', 'token', 'secret', 'clientSecret'} +RECEIPT_KEYS = {'schema', 'operation', 'objects', 'values_emitted', 'consumer_reference_reverted'} + + +def source_digest(): + files = {name: digest((ROOT / name).read_bytes()) for name in SOURCE_FILES} + return digest(json.dumps({'files': files}, sort_keys=True, separators=(',', ':')).encode()) + + +def descriptor(owner): + if not isinstance(owner, str) or not IDENTIFIER.fullmatch(owner): + raise Refused('bounded consumer owner required') + return Descriptor('warden.case.execution.v1', 'deliver-consumer-secret', owner, + REFERENCE, source_digest(), 'agent', 'consumer-owner-deployment', + 'client_id', 'change', OBJECTS, False, ('parent-session',)) + + +def expected(params): + """Fixed metadata the owner receipt must match; derived from case parameters only.""" + cid, svc, ns = params['client_id'], params['consuming_service'], params['consumer_namespace'] + return { + 'consumer-eso-sa': {'kind': 'ServiceAccount', 'namespace': ns, + 'name': f'{svc}-keycape-{cid}-eso', + 'automount': False, 'legacy_token_refs': 0}, + 'consumer-secretstore': {'kind': 'SecretStore', 'namespace': ns, + 'name': f'openbao-keycape-{cid}-consumer', + 'service_account': f'{svc}-keycape-{cid}-eso', + 'audiences': ['openbao'], + 'role': f'external-secrets-keycape-{cid}-{svc}', 'ready': True}, + 'consumer-externalsecret': {'kind': 'ExternalSecret', 'namespace': ns, + 'name': f'{svc}-netkingdom-oidc-client', + 'creation_policy': 'Owner', 'deletion_policy': 'Retain', + 'remote_path': f'workloads/keycape/{cid}', + 'keys': ['CLIENT_SECRET'], 'ready': True}, + } + + +def _value_free(obj): + if isinstance(obj, dict): + return not (VALUE_KEYS & set(obj)) and all(_value_free(v) for v in obj.values()) + if isinstance(obj, list): + return all(_value_free(v) for v in obj) + return obj is None or isinstance(obj, (bool, int, str)) and (not isinstance(obj, str) or len(obj) <= 253) + + +def build(*, admitted: Invocation, case_source: bytes, owner_apply, owner_remove, + publish, authenticate, revoke_session): + """Bind one admitted 1.2 confidential case; return a closed kernel Adapter. + + owner_apply(inv, expected)->receipt and owner_remove(inv, created)->receipt are + consumer-owner capabilities (never command strings). Receipts carry status and + fixed metadata only; any value-bearing key is refused. Only objects the owner + reports as created in this attempt (uid-bound) are recorded and may be removed; + preexisting-matched objects are never deleted. + """ + reviewed = source_digest() + if type(admitted) is not Invocation or not all( + isinstance(v, str) and IDENTIFIER.fullmatch(v) + for v in (admitted.step, admitted.owner, admitted.target, admitted.actor, + admitted.principal, admitted.lane)): + raise Refused('bounded invocation required') + desc = descriptor(admitted.owner) + if (admitted.step, admitted.actor, admitted.lane) != (desc.step, desc.actor, desc.lane): + raise Refused('delivery invocation boundary differs') + if admitted.operation != 'admission' or admitted.attempt_id: + raise Refused('expected admission invocation') + if type(case_source) is not bytes or len(case_source) > 262144 or digest(case_source) != admitted.pins.case: + raise Refused('case source pin differs') + desc_bytes = json.dumps(asdict(desc), sort_keys=True, separators=(',', ':')).encode() + if admitted.pins.adapter != digest(desc_bytes): + raise Refused('descriptor pin differs') + if not all(callable(x) for x in (owner_apply, owner_remove, publish, authenticate, revoke_session)): + raise Refused('owner capabilities required') + with contained_output(): + case = yaml.safe_load(case_source) + spec = case.get('spec') if isinstance(case, dict) else None + params = spec.get('parameters') if isinstance(spec, dict) else None + if not isinstance(params, dict) or spec.get('type') != TYPE: + raise Refused('type 1.2 case required; canonical 1.1 has no delivery step') + if params.get('client_type') != 'confidential': + raise Refused('delivery applies to confidential clients only') + for key in ('client_id', 'consuming_service', 'consumer_namespace', 'consumer_owner_repo'): + if not isinstance(params.get(key), str) or not IDENTIFIER.fullmatch(params[key]): + raise Refused('bounded case parameter required') + if admitted.target != params['client_id'] or admitted.owner != params['consumer_owner_repo']: + raise Refused('case target or owner differs') + want = expected(params) + records, state = {}, {'created': None} + + def binding(inv, operation): + if (replace(inv, attempt_id='', operation='admission') != admitted + or inv.operation != operation or not IDENTIFIER.fullmatch(inv.attempt_id)): + raise Refused('attempt binding differs') + if source_digest() != reviewed: + raise Refused('owner source changed') + + def check(receipt, operation, scope): + if (type(receipt) is not dict or set(receipt) != RECEIPT_KEYS or not _value_free(receipt) + or receipt['schema'] != 'netkingdom.consumer-secret-delivery.v1' + or receipt['operation'] != operation or receipt['values_emitted'] is not False + or type(receipt['objects']) is not dict or set(receipt['objects']) != set(scope)): + raise Refused('owner delivery receipt contract differs') + created = [] + for name in scope: + obj = receipt['objects'][name] + if type(obj) is not dict or set(obj) != {'uid', 'created', 'metadata'}: + raise Refused('owner object receipt differs') + if not isinstance(obj['uid'], str) or not 0 < len(obj['uid']) <= 64: + raise Refused('object uid binding required') + if operation == 'apply': + if obj['metadata'] != want[name] or type(obj['created']) is not bool: + raise Refused('delivered metadata differs from case') + if obj['created']: + created.append(name) + return {n: receipt['objects'][n]['uid'] for n in scope}, tuple(created) + + def execute(inv, operation, created=()): + try: + with contained_output(): + if operation == 'apply': + binding(inv, 'run') + uids, objects = check(owner_apply(inv, want), 'apply', OBJECTS) + outcome = Outcome.SUCCESS if objects else Outcome.NO_CHANGE + else: + binding(inv, 'rollback') + prior = state['created'] + if prior is None or not created or set(created) != set(prior[1]) or not verify(*prior[0]): + raise Refused('no authenticated case-created delivery objects') + order = tuple(n for n in ROLLBACK_ORDER if n in created) + receipt = owner_remove(inv, {n: prior[1][n] for n in order}) + if type(receipt) is dict and receipt.get('consumer_reference_reverted') is not True: + raise Refused('consumer secret reference must be reverted first') + uids, _ = check(receipt, 'remove', order) + if uids != {n: prior[1][n] for n in order}: + raise Refused('removed object uid differs from created') + objects, outcome = (), Outcome.ROLLED_BACK + finally: + with contained_output(): + revocation = revoke_session(inv) + if not isinstance(revocation, str) or not IDENTIFIER.fullmatch(revocation): + raise Refused('session revocation evidence missing') + payload = json.dumps({'schema': 'netkingdom.consumer-secret-delivery.attempt.v1', + 'invocation': asdict(inv), 'source': reviewed, 'outcome': outcome.value, + 'created': objects, 'uids': uids, 'session_revocation': revocation}, + sort_keys=True, separators=(',', ':')).encode() + with contained_output(): + evidence = publish(payload) + if not isinstance(evidence, str) or not IDENTIFIER.fullmatch(evidence) or evidence in records: + raise Refused('owner evidence identifier invalid or reused') + result = OwnerResult(outcome, evidence, objects, True, revocation, + attempt_id=inv.attempt_id, operation=inv.operation) + records[evidence] = (inv, payload, result) + if operation == 'apply' and objects: + state['created'] = ((inv, result), {n: uids[n] for n in objects}) + elif operation == 'remove': + state['created'] = None + return result + + def verify(inv, result): + record = records.get(result.evidence) if type(result) is OwnerResult else None + if record is None or record[0] != inv or record[2] != result: + return False + with contained_output(): + return authenticate(result.evidence, record[1]) is True + + return Adapter(REFERENCE, reviewed, lambda inv: execute(inv, 'apply'), verify, + lambda inv, created: execute(inv, 'remove', created)) diff --git a/sso-mfa/k8s/keycape/test_deliver_consumer_secret_adapter.py b/sso-mfa/k8s/keycape/test_deliver_consumer_secret_adapter.py new file mode 100644 index 0000000..d4371b8 --- /dev/null +++ b/sso-mfa/k8s/keycape/test_deliver_consumer_secret_adapter.py @@ -0,0 +1,138 @@ +"""Synthetic positive/refusal fixtures for the deliver-consumer-secret seam. No live calls.""" +from dataclasses import asdict, replace +import copy +import importlib.util +import json +from pathlib import Path +import unittest + +import yaml +from warden.case_run import Invocation, digest +from warden.case_receipt_state import Outcome, Pins, Refused + +ROOT = Path(__file__).parent +spec = importlib.util.spec_from_file_location('delivery_adapter', ROOT / 'deliver-consumer-secret-adapter.py') +a = importlib.util.module_from_spec(spec) +spec.loader.exec_module(a) + +PARAMS = {'client_id': 'forgejo', 'client_type': 'confidential', 'consuming_service': 'forgejo', + 'consumer_namespace': 'forgejo', 'consumer_owner_repo': 'forgejo-ops'} + + +class DeliveryTests(unittest.TestCase): + def setUp(self): + self.case = self.case_bytes() + self.effects, self.published = [], {} + self.created = {n: True for n in a.OBJECTS} + self.reverted = True + self.admitted = self.invocation(self.case) + self.inv = replace(self.admitted, attempt_id='attempt-one', operation='run') + self.rinv = replace(self.admitted, attempt_id='attempt-two', operation='rollback') + + def case_bytes(self, type_='oidc-browser-relying-party@1.2', **over): + return yaml.safe_dump({'spec': {'type': type_, 'parameters': {**PARAMS, **over}}}).encode() + + def invocation(self, case, owner='forgejo-ops'): + desc = json.dumps(asdict(a.descriptor(owner)), sort_keys=True, separators=(',', ':')).encode() + pins = Pins(digest(case), digest(b'type'), digest(desc), digest(b'catalog'), + digest(b'acceptance'), digest(b'run sheet')) + return Invocation('deliver-consumer-secret', owner, 'forgejo', 'agent', 'founder', + 'consumer-owner-deployment', pins) + + def receipt(self, operation, names, want=None): + objs = {n: {'uid': f'uid-{n}', 'created': self.created[n], + 'metadata': copy.deepcopy(want[n]) if want else {}} for n in names} + return {'schema': 'netkingdom.consumer-secret-delivery.v1', 'operation': operation, + 'objects': objs, 'values_emitted': False, 'consumer_reference_reverted': self.reverted} + + def adapter(self, case=None, admitted=None, apply=None): + def owner_apply(inv, want): + self.effects.append('apply') + return apply(want) if apply else self.receipt('apply', a.OBJECTS, want) + + def owner_remove(inv, targets): + self.effects.append(('remove', tuple(targets))) + return self.receipt('remove', tuple(targets)) + + def publish(payload): + ident = f'evidence-{len(self.published)}' + self.published[ident] = payload + return ident + + return a.build(admitted=admitted or self.admitted, case_source=case or self.case, + owner_apply=owner_apply, owner_remove=owner_remove, publish=publish, + authenticate=lambda i, p: self.published.get(i) == p, + revoke_session=lambda inv: f'revoked-{inv.attempt_id}') + + def test_positive_apply_verify_and_created_only_rollback(self): + self.created['consumer-eso-sa'] = False # preexisting-matched: must be preserved + ad = self.adapter() + result = ad.invoke(self.inv) + self.assertEqual(result.outcome, Outcome.SUCCESS) + self.assertEqual(result.created, ('consumer-secretstore', 'consumer-externalsecret')) + self.assertTrue(ad.verify(self.inv, result)) + payload = self.published[result.evidence].decode() + self.assertNotIn('CLIENT_SECRET', payload) + back = ad.rollback(self.rinv, result.created) + self.assertEqual(back.outcome, Outcome.ROLLED_BACK) + self.assertEqual(self.effects[-1], ('remove', ('consumer-externalsecret', 'consumer-secretstore'))) + + def test_no_change_has_nothing_to_roll_back(self): + self.created = {n: False for n in a.OBJECTS} + ad = self.adapter() + self.assertEqual(ad.invoke(self.inv).outcome, Outcome.NO_CHANGE) + with self.assertRaises(Refused): + ad.rollback(self.rinv, ('consumer-externalsecret',)) + self.assertNotIn('remove', [e[0] for e in self.effects if isinstance(e, tuple)]) + + def test_refuses_canonical_11_public_and_bad_binding(self): + for case in (self.case_bytes('oidc-browser-relying-party@1.1'), + self.case_bytes(client_type='public')): + with self.assertRaises(Refused): + self.adapter(case=case, admitted=self.invocation(case)) + with self.assertRaises(Refused): + self.adapter(case=self.case + b'\n#drift') + with self.assertRaises(Refused): + self.adapter(admitted=self.invocation(self.case, owner='other-owner')) + with self.assertRaises(Refused): + self.adapter().invoke(replace(self.inv, target='other')) + self.assertEqual(self.effects, []) + + def test_refuses_value_bearing_or_mismatched_receipts(self): + def value(want): + r = self.receipt('apply', a.OBJECTS, want) + r['objects']['consumer-externalsecret']['metadata']['data'] = {'CLIENT_SECRET': 'x'} + return r + + def drift(want): + r = self.receipt('apply', a.OBJECTS, want) + r['objects']['consumer-externalsecret']['metadata']['keys'] = ['CLIENT_SECRET', 'OTHER'] + return r + + def automount(want): + r = self.receipt('apply', a.OBJECTS, want) + r['objects']['consumer-eso-sa']['metadata']['automount'] = True + return r + + def emitted(want): + return {**self.receipt('apply', a.OBJECTS, want), 'values_emitted': True} + + for bad in (value, drift, automount, emitted): + ad = self.adapter(apply=bad) + with self.assertRaises(Refused): + ad.invoke(self.inv) + + def test_rollback_requires_reverted_reference_and_exact_created_set(self): + ad = self.adapter() + result = ad.invoke(self.inv) + with self.assertRaises(Refused): + ad.rollback(self.rinv, ('consumer-externalsecret',)) + self.reverted = False + with self.assertRaises(Refused): + ad.rollback(self.rinv, result.created) + forged = replace(result, evidence='evidence-forged') + self.assertFalse(ad.verify(self.inv, forged)) + + +if __name__ == '__main__': + unittest.main()