coulomb/net-kingdom
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

NK-WP-0048-T26: deliver-secret adapter source candidate

Browse source 
Injection-only, value-free Platform delivery seam with exact metadata-only
receipt contract, created-only rollback and synthetic positive/refusal tests.
Unexercised; Platform owner source review and native action stay separate gates.

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:
tegwick 2026-10-17 07:55:24 +02:00
parent 2019e823f0
commit 1b60f6e9c6
3 changed files with 332 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

44
sso-mfa/k8s/keycape/deliver-secret-adapter.md Normal file
View file

@ -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.

158
sso-mfa/k8s/keycape/deliver-secret-adapter.py Normal file
View file

@ -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))

130
sso-mfa/k8s/keycape/test_deliver_secret_adapter.py Normal file
View file

@ -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()