diff --git a/sso-mfa/k8s/keycape/mint-secret-adapter.md b/sso-mfa/k8s/keycape/mint-secret-adapter.md new file mode 100644 index 0000000..febb615 --- /dev/null +++ b/sso-mfa/k8s/keycape/mint-secret-adapter.md @@ -0,0 +1,23 @@ +# Mint-secret adapter (source candidate, NK-WP-0048-T25) + +`mint-secret-adapter.py` is an unexercised, injection-only seam for the type 1.2 +`mint-secret` step (owner `railiance-platform`, lane +`keycape-client-secret-generate`, `value_handling: generated_in_custody`). It is +modelled on the T13 registration adapter. + +- **No secret values.** The secret is generated inside custody by an injected + Platform `mint(path)` capability. Receipts must be exactly + `{schema, path, operation, version, created, values_emitted: false}`; any other + field (including a value) is refused before evidence is published. +- **Value-free evidence.** Published evidence carries invocation, source digest, + custody path, version number and both revocation identifiers. +- **Created-only rollback.** Only a version this adapter minted and can + authenticate is destroyed, by exact version; otherwise rollback is refused. +- **Revocation.** Descriptor effect `mint` with scopes `parent-session` and + `broker-grant`; both are revoked after every attempt, including failures. +- Confidential clients only; case, descriptor, acceptance and attempt pins bound + as in T13. Not retry-safe. + +Not supplied here: Platform owner source review before pinning, Warden admission, +type 1.2 selection (canonical 1.1 stays default; no `--allow-unimplemented`), and +any native mint. Tests: `PYTHONPATH=~/ops-warden/src python3 -m pytest test_mint_secret_adapter.py`. diff --git a/sso-mfa/k8s/keycape/mint-secret-adapter.py b/sso-mfa/k8s/keycape/mint-secret-adapter.py new file mode 100644 index 0000000..25f6115 --- /dev/null +++ b/sso-mfa/k8s/keycape/mint-secret-adapter.py @@ -0,0 +1,146 @@ +"""Unexercised, injection-only mint-secret seam (type 1.2 step `mint-secret`). + +Source candidate for Platform owner review; not pinned, admitted or run. The +client secret is generated inside custody by an injected Platform capability. +This module never receives, reads, templates or returns a secret value: owner +receipts are closed value-free metadata and any other field is refused. +Canonical 1.1 stays the default; nothing here selects 1.2 or bypasses admission. +""" +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 = ('mint-secret-adapter.py',) +REFERENCE = 'net-kingdom/keycape-mint-secret@v1' +RECEIPT_SCHEMA = 'railiance-platform.custody-mint.v1' +RECEIPT_KEYS = {'schema', 'path', 'operation', 'version', 'created', 'values_emitted'} +CREATED = ('custody-secret-version',) + + +def source_digest(): + composition = {'files': {n: digest((ROOT / n).read_bytes()) for n in SOURCE_FILES}} + return digest(json.dumps(composition, sort_keys=True, separators=(',', ':')).encode()) + + +def descriptor(): + return Descriptor('warden.case.execution.v1', 'mint-secret', 'railiance-platform', + REFERENCE, source_digest(), 'warden', 'keycape-client-secret-generate', + 'client_id', 'mint', CREATED, False, ('parent-session', 'broker-grant')) + + +def custody_path(client_id): + return 'platform/workloads/keycape/' + client_id + + +def build(*, admitted: Invocation, case_source: bytes, case_acceptance: str, + mint, destroy, publish, authenticate, revoke_session, revoke_grant): + """Bind one admitted confidential-client case; return a closed kernel Adapter. + + mint(path)->receipt generates a new secret version in custody, value never returned. + destroy(path, version)->receipt removes exactly that version (created-only rollback). + publish(bytes)->identifier, authenticate(identifier, bytes)->True: owner evidence. + revoke_session(inv)/revoke_grant(inv)->identifier end this attempt's parent + session and broker grant. All are capabilities, never command strings. + """ + 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('mint 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 isinstance(case_acceptance, str) or digest(case_acceptance.encode()) != admitted.pins.acceptance: + raise Refused('acceptance evidence binding differs') + if not all(callable(x) for x in (mint, destroy, publish, authenticate, revoke_session, revoke_grant)): + raise Refused('owner custody, evidence and revocation capabilities required') + with contained_output(): + case = yaml.safe_load(case_source) + params = (case or {}).get('spec', {}).get('parameters', {}) if isinstance(case, dict) else {} + if params.get('client_type') != 'confidential': + raise Refused('mint-secret applies only to confidential clients') + if params.get('client_id') != admitted.target: + raise Refused('case target differs') + path = custody_path(admitted.target) + records, creation = {}, None + + def check(receipt, operation, version=None): + if (type(receipt) is not dict or set(receipt) != RECEIPT_KEYS + or receipt['schema'] != RECEIPT_SCHEMA or receipt['path'] != path + or receipt['operation'] != operation or receipt['values_emitted'] is not False + or type(receipt['created']) is not bool + or type(receipt['version']) is not int or receipt['version'] < 1 + or (version is not None and receipt['version'] != version)): + raise Refused('owner value-free receipt contract differs') + + def execute(inv, operation, created=()): + nonlocal creation + receipt = None + try: + with contained_output(): + if (replace(inv, attempt_id='', operation='admission') != admitted + or inv.operation != ('run' if operation == 'mint' else 'rollback') + or not IDENTIFIER.fullmatch(inv.attempt_id)): + raise Refused('attempt binding differs') + if source_digest() != reviewed: + raise Refused('owner source changed') + if operation == 'mint': + receipt = mint(path) + check(receipt, 'mint') + if not receipt['created']: + raise Refused('custody did not create a version') + else: + if created != CREATED or creation is None or not verify(*creation): + raise Refused('no authenticated case-created version') + version = json.loads(records[creation[1].evidence][1])['version'] + receipt = destroy(path, version) + check(receipt, 'destroy', version) + finally: + with contained_output(): + session, grant = revoke_session(inv), revoke_grant(inv) + for ref in (session, grant): + if not isinstance(ref, str) or not IDENTIFIER.fullmatch(ref): + raise Refused('revocation evidence missing') + outcome = Outcome.SUCCESS if operation == 'mint' else Outcome.ROLLED_BACK + objects = CREATED if operation == 'mint' else () + payload = json.dumps({'schema': 'netkingdom.keycape-mint-secret.attempt.v1', + 'invocation': asdict(inv), 'source': reviewed, 'outcome': outcome.value, + 'created': objects, 'path': path, 'version': receipt['version'], + 'values_emitted': False, 'session_revocation': session, 'grant_revocation': grant}, + 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, session, grant, + attempt_id=inv.attempt_id, operation=inv.operation) + records[evidence] = (inv, payload, result) + creation = (inv, result) if operation == 'mint' else 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, 'mint'), + verify, lambda inv, created: execute(inv, 'rollback', created)) diff --git a/sso-mfa/k8s/keycape/test_mint_secret_adapter.py b/sso-mfa/k8s/keycape/test_mint_secret_adapter.py new file mode 100644 index 0000000..c50f78d --- /dev/null +++ b/sso-mfa/k8s/keycape/test_mint_secret_adapter.py @@ -0,0 +1,137 @@ +"""Synthetic custody capabilities only; no secret value exists anywhere in these tests.""" +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 +import yaml + +ROOT = Path(__file__).parent +spec = importlib.util.spec_from_file_location('mint_secret_adapter', ROOT / 'mint-secret-adapter.py') +a = importlib.util.module_from_spec(spec) +spec.loader.exec_module(a) +PATH = 'platform/workloads/keycape/forgejo' + + +def receipt(op, version=3, **over): + r = {'schema': a.RECEIPT_SCHEMA, 'path': PATH, 'operation': op, 'version': version, + 'created': op == 'mint', 'values_emitted': False} + r.update(over) + return r + + +class MintSecretTests(unittest.TestCase): + def setUp(self): + self.case = yaml.safe_dump({'spec': {'type': 'oidc-browser-relying-party@1.2', 'parameters': + {'client_id': 'forgejo', 'client_type': 'confidential'}}}).encode() + desc = json.dumps(asdict(a.descriptor()), sort_keys=True, separators=(',', ':')).encode() + pins = Pins(digest(self.case), digest(b'type'), digest(desc), digest(b'catalog'), + digest(b'case-acceptance:test'), digest(b'run sheet')) + self.admitted = Invocation('mint-secret', 'railiance-platform', 'forgejo', 'warden', + 'founder', 'keycape-client-secret-generate', pins) + self.run = replace(self.admitted, attempt_id='attempt-one', operation='run') + self.back = replace(self.admitted, attempt_id='attempt-one', operation='rollback') + self.calls, self.evidence, self.revoked = [], {}, [] + self.mint_receipt = receipt('mint') + self.kw = dict(admitted=self.admitted, case_source=self.case, + case_acceptance='case-acceptance:test', mint=self.mint, destroy=self.destroy, + publish=self.publish, authenticate=lambda ref, p: self.evidence.get(ref) == p, + revoke_session=lambda inv: self.revoke('session', inv), + revoke_grant=lambda inv: self.revoke('grant', inv)) + + def mint(self, path): + self.calls.append(('mint', path)) + return self.mint_receipt + + def destroy(self, path, version): + self.calls.append(('destroy', path, version)) + return receipt('destroy', version) + + def publish(self, payload): + ref = 'owner-evidence:' + str(len(self.evidence)) + self.evidence[ref] = payload + return ref + + def revoke(self, kind, inv): + self.revoked.append((kind, inv.operation)) + return kind + '-revoked:' + inv.attempt_id + + def test_descriptor_is_mint_with_both_revocations(self): + d = a.descriptor() + self.assertEqual((d.effect, d.revocation_scopes, d.retry_safe), + ('mint', ('parent-session', 'broker-grant'), False)) + + def test_mint_records_value_free_evidence_and_verifies(self): + adapter = a.build(**self.kw) + result = adapter.invoke(self.run) + self.assertEqual((result.outcome, result.created), (Outcome.SUCCESS, a.CREATED)) + self.assertTrue(adapter.verify(self.run, result)) + payload = json.loads(self.evidence[result.evidence]) + self.assertEqual((payload['path'], payload['version'], payload['values_emitted']), (PATH, 3, False)) + self.assertEqual(self.revoked, [('session', 'run'), ('grant', 'run')]) + self.assertNotIn('value', json.dumps(payload).replace('values_emitted', '')) + + def test_created_only_rollback_destroys_exact_version(self): + adapter = a.build(**self.kw) + adapter.invoke(self.run) + result = adapter.rollback(self.back, a.CREATED) + self.assertEqual(result.outcome, Outcome.ROLLED_BACK) + self.assertEqual(self.calls[-1], ('destroy', PATH, 3)) + with self.assertRaises(Refused): # second rollback has nothing case-created + adapter.rollback(replace(self.back, attempt_id='attempt-two'), a.CREATED) + + def test_rollback_without_creation_refused_and_revokes(self): + adapter = a.build(**self.kw) + with self.assertRaises(Refused): + adapter.rollback(self.back, a.CREATED) + self.assertEqual([c[0] for c in self.calls], []) + self.assertEqual(self.revoked, [('session', 'rollback'), ('grant', 'rollback')]) + + def test_receipt_with_value_or_extra_field_refused(self): + for bad in (receipt('mint', value='x'), receipt('mint', values_emitted=True), + receipt('mint', path='platform/workloads/keycape/other'), + receipt('mint', created=False), receipt('mint', version='3'), 'opaque'): + self.mint_receipt = bad + with self.assertRaises(Refused): + a.build(**self.kw).invoke(self.run) + self.assertEqual(self.evidence, {}) + + def test_public_client_and_target_mismatch_refused(self): + for params in ({'client_id': 'forgejo', 'client_type': 'public'}, + {'client_id': 'other', 'client_type': 'confidential'}): + case = yaml.safe_dump({'spec': {'parameters': params}}).encode() + pins = replace(self.admitted.pins, case=digest(case)) + with self.assertRaises(Refused): + a.build(**dict(self.kw, case_source=case, admitted=replace(self.admitted, pins=pins))) + + def test_binding_refusals(self): + cases = [dict(case_source=self.case + b'#'), dict(case_acceptance='other'), + dict(admitted=replace(self.admitted, lane='key-cape-oidc-login')), + dict(admitted=replace(self.admitted, pins=replace(self.admitted.pins, adapter=digest(b'x')))), + dict(admitted=self.run), dict(mint='bao kv put')] + for over in cases: + with self.assertRaises(Refused): + a.build(**dict(self.kw, **over)) + + def test_attempt_binding_and_reused_evidence_refused(self): + adapter = a.build(**self.kw) + with self.assertRaises(Refused): + adapter.invoke(replace(self.run, target='other')) + self.kw['publish'] = lambda payload: 'owner-evidence:fixed' + adapter = a.build(**self.kw) + adapter.invoke(self.run) + with self.assertRaises(Refused): + adapter.invoke(replace(self.run, attempt_id='attempt-two')) + + def test_verify_rejects_foreign_result(self): + adapter = a.build(**self.kw) + result = adapter.invoke(self.run) + self.assertFalse(adapter.verify(replace(self.run, attempt_id='attempt-two'), result)) + self.assertFalse(a.build(**self.kw).verify(self.run, result)) + + +if __name__ == '__main__': + unittest.main()