NK-WP-0048-T25: mint-secret adapter source candidate (value-free)
Injection-only seam for the type 1.2 mint-secret step: closed value-free custody receipts, created-only rollback by exact version, parent-session and broker-grant revocation. Source candidate only; no owner review or admission. 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
2019e823f0
commit
558c5bcbf6
3 changed files with 306 additions and 0 deletions
23
sso-mfa/k8s/keycape/mint-secret-adapter.md
Normal file
23
sso-mfa/k8s/keycape/mint-secret-adapter.md
Normal file
|
|
@ -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`.
|
||||
146
sso-mfa/k8s/keycape/mint-secret-adapter.py
Normal file
146
sso-mfa/k8s/keycape/mint-secret-adapter.py
Normal file
|
|
@ -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))
|
||||
137
sso-mfa/k8s/keycape/test_mint_secret_adapter.py
Normal file
137
sso-mfa/k8s/keycape/test_mint_secret_adapter.py
Normal file
|
|
@ -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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue