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

NK-WP-0048-T28: deliver-consumer-secret adapter source candidate

Browse source 
Unexercised, injection-only, value-free evidence seam for type 1.2 with
positive/refusal fixtures and created-only rollback. Canonical 1.1 default
unchanged; no owner review or native action supplied.

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:47 +02:00
parent 2019e823f0
commit 650c1ac57c
3 changed files with 369 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

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

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

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

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

138
sso-mfa/k8s/keycape/test_deliver_consumer_secret_adapter.py Normal file
View file

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