diff --git a/sso-mfa/k8s/keycape/deliver-secret-adapter.md b/sso-mfa/k8s/keycape/deliver-secret-adapter.md new file mode 100644 index 0000000..a7e9444 --- /dev/null +++ b/sso-mfa/k8s/keycape/deliver-secret-adapter.md @@ -0,0 +1,44 @@ +# Deliver-secret adapter source candidate (NK-WP-0048-T26) + +Status: **source candidate, unexercised.** Not reviewed, not pinned, not admitted. +Modelled on the T13 registration adapter (`keycape-registration-adapter.py`). + +## What it is + +`deliver-secret-adapter.py` is an injection-only Warden v1 kernel adapter for the +Platform-owned delivery seam (ESO/custody). Descriptor: step `deliver-secret`, +owner `railiance-platform`, lane `platform-secret-delivery`, target parameter +`client_id`, allowed created object `delivered-secret`, `retry_safe: false`, +revocation scope `parent-session`. + +It binds one admitted invocation to one Secret reference (namespace, name, +custody *reference*, key names). It never reads, mints or templates a value. +Delivery, removal, evidence publication/authentication and parent-session +revocation are injected owner capabilities; the module itself has no CLI and +calls no kubectl, OpenBao or registry. + +## Value-free evidence contract + +Owner receipts must have exactly the fields `schema, operation, namespace, name, +custody_reference, key_names, values_emitted, changed, before, after`, with +`values_emitted: false` and `before`/`after` limited to `{uid, resource_version}`. +Any extra field (for example a value) is refused before evidence is published. +The parent session is revoked even when the owner capability fails. + +## Rollback: created-only + +Only a delivery this process verified as created (`before: null`, `changed: true`, +evidence authenticated by the owner) may be rolled back, and rollback must remove +exactly the created revision. A pre-existing Secret is `NO_CHANGE` and is never +removed. Failures are not converted to retry-safe outcomes. + +## Tests + +`PYTHONPATH=~/ops-warden/src python3 -m pytest -q test_deliver_secret_adapter.py` +— synthetic positive and refusal fixtures only. + +## Not supplied by this candidate + +Canonical 1.1 stays the default; no `--allow-unimplemented` admission. Platform +owner source review before pinning, the real ESO/custody capability +implementations, and any native action remain separate owner gates. diff --git a/sso-mfa/k8s/keycape/deliver-secret-adapter.py b/sso-mfa/k8s/keycape/deliver-secret-adapter.py new file mode 100644 index 0000000..408ccc0 --- /dev/null +++ b/sso-mfa/k8s/keycape/deliver-secret-adapter.py @@ -0,0 +1,158 @@ +"""Unexercised, injection-only deliver-secret seam (NK-WP-0048-T26) for Warden's v1 kernel. + +Platform-owned delivery evidence (ESO/custody), value-free. No CLI, credential, +kubectl, OpenBao or registry access lives here: delivery, removal, evidence +publication/authentication and parent-session revocation are injected owner +capabilities. Receipts must match an exact metadata-only field set, so a +capability that returns a value (or any unexpected field) is refused. +""" +from __future__ import annotations + +from dataclasses import asdict, replace +import json +import re +from pathlib import Path + +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-secret-adapter.py',) +REFERENCE = 'net-kingdom/deliver-secret@v1' +CREATED = 'delivered-secret' +NAME = re.compile(r'[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?') +RECEIPT_FIELDS = {'schema', 'operation', 'namespace', 'name', 'custody_reference', + 'key_names', 'values_emitted', 'changed', 'before', 'after'} + + +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(): + return Descriptor('warden.case.execution.v1', 'deliver-secret', 'railiance-platform', + REFERENCE, source_digest(), 'human', 'platform-secret-delivery', + 'client_id', 'change', (CREATED,), False, ('parent-session',)) + + +def _revision(value): + return (isinstance(value, dict) and set(value) == {'uid', 'resource_version'} + and all(isinstance(v, str) and 0 < len(v) <= 192 for v in value.values())) + + +def build(*, admitted: Invocation, case_acceptance: str, namespace: str, name: str, + custody_reference: str, key_names: tuple, deliver, remove, publish, + authenticate, revoke_session): + """Bind one admitted case to one delivered Secret reference; return a kernel Adapter. + + deliver(inv, binding)->receipt and remove(inv, binding, after)->receipt are + owner capabilities (e.g. ESO sync with readback); receipts carry metadata only. + Only a delivery verified as created in this process may be rolled back. + """ + reviewed = source_digest() + desc = descriptor() + 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') + if (admitted.step, admitted.owner, admitted.actor, admitted.lane) != ( + desc.step, desc.owner, desc.actor, desc.lane): + raise Refused('delivery invocation boundary differs') + if admitted.operation != 'admission' or admitted.attempt_id: + raise Refused('expected admission invocation') + 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 (isinstance(case_acceptance, str) and 0 < len(case_acceptance) <= 192 + and digest(case_acceptance.encode()) == admitted.pins.acceptance): + raise Refused('acceptance evidence binding differs') + if not all(isinstance(v, str) and NAME.fullmatch(v) for v in (namespace, name)): + raise Refused('bounded secret reference required') + if not (isinstance(custody_reference, str) and 0 < len(custody_reference) <= 192 + and IDENTIFIER.fullmatch(custody_reference.replace('/', '-'))): + raise Refused('bounded custody reference required') + if (type(key_names) is not tuple or not key_names or len(set(key_names)) != len(key_names) + or not all(isinstance(k, str) and re.fullmatch(r'[A-Za-z0-9._-]{1,64}', k) for k in key_names)): + raise Refused('bounded key names required') + if not all(callable(x) for x in (deliver, remove, publish, authenticate, revoke_session)): + raise Refused('owner delivery, evidence and revocation capabilities required') + binding_ref = {'namespace': namespace, 'name': name, 'custody_reference': custody_reference, + 'key_names': list(key_names), 'client_id': admitted.target} + records = {} + creation = None + + def check(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 execute(inv, operation, created=()): + nonlocal creation + try: + with contained_output(): + check(inv, 'run' if operation == 'deliver' else 'rollback') + if operation == 'deliver': + receipt = deliver(inv, dict(binding_ref)) + else: + if created != (CREATED,) or creation is None or not verify(*creation): + raise Refused('no authenticated case-created delivery') + prior = json.loads(records[creation[1].evidence][1])['after'] + receipt = remove(inv, dict(binding_ref), dict(prior)) + if not (isinstance(receipt, dict) and set(receipt) == RECEIPT_FIELDS + and receipt['schema'] == 'netkingdom.deliver-secret.v1' + and receipt['operation'] == operation + and (receipt['namespace'], receipt['name'], receipt['custody_reference']) + == (namespace, name, custody_reference) + and receipt['key_names'] == list(key_names) + and receipt['values_emitted'] is False + and type(receipt['changed']) is bool): + raise Refused('owner delivery receipt contract differs') + if operation == 'deliver': + if not _revision(receipt['after']) or not ( + receipt['before'] is None or _revision(receipt['before'])): + raise Refused('owner revision contract differs') + if receipt['changed'] != (receipt['before'] is None): + raise Refused('pre-existing secret must not be claimed as created') + elif not (receipt['changed'] is True and receipt['after'] is None + and receipt['before'] == prior): + raise Refused('rollback must remove exactly the created revision') + changed = receipt['changed'] + outcome = (Outcome.ROLLED_BACK if operation == 'remove' else + Outcome.SUCCESS if changed else Outcome.NO_CHANGE) + objects = (CREATED,) if operation == 'deliver' and changed else () + 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.deliver-secret.attempt.v1', + 'invocation': asdict(inv), 'source': reviewed, 'outcome': outcome.value, + 'created': objects, 'binding': binding_ref, 'before': receipt['before'], + 'after': receipt['after'], '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 objects: + creation = (inv, result) + elif operation == 'remove': + creation = 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, 'deliver'), + verify, lambda inv, created: execute(inv, 'remove', created)) diff --git a/sso-mfa/k8s/keycape/test_deliver_secret_adapter.py b/sso-mfa/k8s/keycape/test_deliver_secret_adapter.py new file mode 100644 index 0000000..1b3f8a5 --- /dev/null +++ b/sso-mfa/k8s/keycape/test_deliver_secret_adapter.py @@ -0,0 +1,130 @@ +"""Synthetic, value-free fixtures for the T26 deliver-secret seam. No live calls.""" +from dataclasses import asdict, replace +import importlib.util +import json +from pathlib import Path +import unittest + +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('deliver_secret_adapter', ROOT / 'deliver-secret-adapter.py') +a = importlib.util.module_from_spec(spec) +spec.loader.exec_module(a) + +KEYS = ('client-secret',) +AFTER = {'uid': 'uid-1', 'resource_version': '3'} + + +class DeliverSecretTests(unittest.TestCase): + def setUp(self): + desc = json.dumps(asdict(a.descriptor()), sort_keys=True, separators=(',', ':')).encode() + pins = Pins(digest(b'case'), digest(b'type'), digest(desc), digest(b'catalog'), + digest(b'case-acceptance:test'), digest(b'run sheet')) + self.admitted = Invocation('deliver-secret', 'railiance-platform', 'forgejo', 'human', + 'founder', 'platform-secret-delivery', pins) + self.inv = replace(self.admitted, attempt_id='attempt-one', operation='run') + self.rb = replace(self.admitted, attempt_id='attempt-two', operation='rollback') + self.evidence, self.calls, self.revoked = {}, [], [] + self.before = None + self.kw = dict(admitted=self.admitted, case_acceptance='case-acceptance:test', + namespace='sso', name='keycape-client-forgejo', + custody_reference='kv/keycape/forgejo', key_names=KEYS, + deliver=self.deliver, remove=self.remove, publish=self.publish, + authenticate=lambda ref, p: self.evidence.get(ref) == p, + revoke_session=lambda inv: self.revoked.append(inv) or 'revocation-' + inv.attempt_id) + + def receipt(self, operation, changed, before, after, **extra): + r = {'schema': 'netkingdom.deliver-secret.v1', 'operation': operation, 'namespace': 'sso', + 'name': 'keycape-client-forgejo', 'custody_reference': 'kv/keycape/forgejo', + 'key_names': list(KEYS), 'values_emitted': False, 'changed': changed, + 'before': before, 'after': after} + r.update(extra) + return r + + def deliver(self, inv, binding): + self.calls.append(('deliver', binding)) + return self.receipt('deliver', self.before is None, self.before, + self.before or dict(AFTER)) + + def remove(self, inv, binding, after): + self.calls.append(('remove', after)) + return self.receipt('remove', True, after, None) + + def publish(self, payload): + ref = 'owner-evidence-' + str(len(self.evidence)) + self.evidence[ref] = payload + return ref + + def test_created_delivery_verifies_and_rolls_back(self): + adapter = a.build(**self.kw) + result = adapter.invoke(self.inv) + self.assertEqual((result.outcome, result.created), (Outcome.SUCCESS, (a.CREATED,))) + self.assertTrue(adapter.verify(self.inv, result)) + self.assertNotIn(b'secret-value', b''.join(self.evidence.values())) + undone = adapter.rollback(self.rb, (a.CREATED,)) + self.assertEqual(undone.outcome, Outcome.ROLLED_BACK) + self.assertEqual(self.calls[-1], ('remove', AFTER)) + self.assertEqual(len(self.revoked), 2) + + def test_pre_existing_secret_is_no_change_and_never_rolled_back(self): + self.before = dict(AFTER) + adapter = a.build(**self.kw) + result = adapter.invoke(self.inv) + self.assertEqual((result.outcome, result.created), (Outcome.NO_CHANGE, ())) + with self.assertRaises(Refused): + adapter.rollback(self.rb, (a.CREATED,)) + self.assertNotIn('remove', [c[0] for c in self.calls]) + + def test_receipt_carrying_value_is_refused_and_session_still_revoked(self): + self.deliver = lambda inv, b: self.receipt('deliver', True, None, dict(AFTER), data='secret-value') + self.kw['deliver'] = self.deliver + with self.assertRaises(Refused): + a.build(**self.kw).invoke(self.inv) + self.assertEqual(len(self.revoked), 1) + self.assertEqual(self.evidence, {}) + + def test_values_emitted_or_wrong_reference_refused(self): + for bad in (dict(values_emitted=True), dict(name='other'), dict(key_names=['x'])): + self.kw['deliver'] = lambda inv, b, bad=bad: self.receipt('deliver', True, None, dict(AFTER), **bad) + with self.assertRaises(Refused): + a.build(**self.kw).invoke(self.inv) + + def test_build_refusals(self): + cases = [dict(namespace='SSO!'), dict(key_names=['client-secret']), dict(key_names=()), + dict(case_acceptance='other'), dict(publish=None), + dict(admitted=replace(self.admitted, lane='other-lane')), + dict(admitted=replace(self.admitted, pins=replace(self.admitted.pins, adapter=digest(b'x'))))] + for change in cases: + with self.subTest(change=change), self.assertRaises(Refused): + a.build(**{**self.kw, **change}) + + def test_attempt_binding_and_reused_evidence_refused(self): + adapter = a.build(**self.kw) + with self.assertRaises(Refused): + adapter.invoke(replace(self.inv, target='other')) + self.kw['publish'] = lambda payload: 'same-ref' + adapter = a.build(**self.kw) + adapter.invoke(self.inv) + with self.assertRaises(Refused): + adapter.invoke(replace(self.inv, attempt_id='attempt-three')) + + def test_unverified_or_foreign_creation_cannot_roll_back(self): + self.kw['authenticate'] = lambda ref, p: False + adapter = a.build(**self.kw) + adapter.invoke(self.inv) + with self.assertRaises(Refused): + adapter.rollback(self.rb, (a.CREATED,)) + adapter = a.build(**{**self.kw, 'authenticate': lambda ref, p: True}) + with self.assertRaises(Refused): + adapter.rollback(self.rb, (a.CREATED,)) + + def test_descriptor_keeps_non_retry_safe_and_parent_session_scope(self): + d = a.descriptor() + self.assertEqual((d.step, d.owner, d.retry_safe, d.revocation_scopes), + ('deliver-secret', 'railiance-platform', False, ('parent-session',))) + + +if __name__ == '__main__': + unittest.main()