Bind deliver-secret reference to the admitted case (NK48-R2-F2)
build() now takes case_source, checks it against admitted.pins.case, refuses
non-confidential cases and a client_id differing from the admitted target, and
derives sso/keycape-{client_id}-client (key client-secret) from
platform/workloads/keycape/{client_id} per type 1.2. No caller-chosen
reference remains. Fixtures use the 1.2 forgejo example and real names;
regression tests cover the reviewer's forgejo/grafana demonstration, a public
client and a case-pin mismatch.
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
68f4bbb8ca
commit
640292ccea
3 changed files with 86 additions and 22 deletions
|
|
@ -11,8 +11,14 @@ owner `railiance-platform`, lane `platform-secret-delivery`, target parameter
|
||||||
`client_id`, allowed created object `delivered-secret`, `retry_safe: false`,
|
`client_id`, allowed created object `delivered-secret`, `retry_safe: false`,
|
||||||
revocation scope `parent-session`.
|
revocation scope `parent-session`.
|
||||||
|
|
||||||
It binds one admitted invocation to one Secret reference (namespace, name,
|
It binds one admitted invocation to one Secret reference. `build()` takes
|
||||||
custody *reference*, key names). It never reads, mints or templates a value.
|
`case_source` and refuses unless its digest equals `admitted.pins.case`, the
|
||||||
|
case is `client_type: confidential`, and its `client_id` equals the admitted
|
||||||
|
target (as `mint-secret` and `bind-env` do). The reference is then derived from
|
||||||
|
the admitted target per type 1.2, never chosen by the caller: ExternalSecret
|
||||||
|
`sso/keycape-{client_id}-client`, key `client-secret`, from custody
|
||||||
|
`platform/workloads/keycape/{client_id}` (NK48-R2-F2). Owner receipts must name
|
||||||
|
that same reference. It never reads, mints or templates a value.
|
||||||
Delivery, removal, evidence publication/authentication and parent-session
|
Delivery, removal, evidence publication/authentication and parent-session
|
||||||
revocation are injected owner capabilities; the module itself has no CLI and
|
revocation are injected owner capabilities; the module itself has no CLI and
|
||||||
calls no kubectl, OpenBao or registry.
|
calls no kubectl, OpenBao or registry.
|
||||||
|
|
|
||||||
|
|
@ -13,6 +13,8 @@ import json
|
||||||
import re
|
import re
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
|
import yaml
|
||||||
|
|
||||||
from warden.case_run import Adapter, Descriptor, Invocation, OwnerResult, contained_output, digest, IDENTIFIER
|
from warden.case_run import Adapter, Descriptor, Invocation, OwnerResult, contained_output, digest, IDENTIFIER
|
||||||
from warden.case_receipt_state import Outcome, Refused
|
from warden.case_receipt_state import Outcome, Refused
|
||||||
|
|
||||||
|
|
@ -20,7 +22,8 @@ ROOT = Path(__file__).resolve().parent
|
||||||
SOURCE_FILES = ('deliver-secret-adapter.py',)
|
SOURCE_FILES = ('deliver-secret-adapter.py',)
|
||||||
REFERENCE = 'net-kingdom/deliver-secret@v1'
|
REFERENCE = 'net-kingdom/deliver-secret@v1'
|
||||||
CREATED = 'delivered-secret'
|
CREATED = 'delivered-secret'
|
||||||
NAME = re.compile(r'[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?')
|
NAMESPACE = 'sso'
|
||||||
|
KEY_NAMES = ('client-secret',)
|
||||||
RECEIPT_FIELDS = {'schema', 'operation', 'namespace', 'name', 'custody_reference',
|
RECEIPT_FIELDS = {'schema', 'operation', 'namespace', 'name', 'custody_reference',
|
||||||
'key_names', 'values_emitted', 'changed', 'before', 'after'}
|
'key_names', 'values_emitted', 'changed', 'before', 'after'}
|
||||||
|
|
||||||
|
|
@ -41,10 +44,21 @@ def _revision(value):
|
||||||
and all(isinstance(v, str) and 0 < len(v) <= 192 for v in value.values()))
|
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,
|
def secret_name(client_id):
|
||||||
custody_reference: str, key_names: tuple, deliver, remove, publish,
|
return 'keycape-' + client_id + '-client'
|
||||||
authenticate, revoke_session):
|
|
||||||
"""Bind one admitted case to one delivered Secret reference; return a kernel Adapter.
|
|
||||||
|
def custody_path(client_id):
|
||||||
|
return 'platform/workloads/keycape/' + client_id
|
||||||
|
|
||||||
|
|
||||||
|
def build(*, admitted: Invocation, case_source: bytes, case_acceptance: str,
|
||||||
|
deliver, remove, publish, authenticate, revoke_session):
|
||||||
|
"""Bind one admitted confidential-client case; return a kernel Adapter.
|
||||||
|
|
||||||
|
The Secret reference is derived from the admitted target per type 1.2:
|
||||||
|
ExternalSecret sso/keycape-{client_id}-client (key client-secret) from
|
||||||
|
bao:platform/workloads/keycape/{client_id}. No caller chooses a reference.
|
||||||
|
|
||||||
deliver(inv, binding)->receipt and remove(inv, binding, after)->receipt are
|
deliver(inv, binding)->receipt and remove(inv, binding, after)->receipt are
|
||||||
owner capabilities (e.g. ESO sync with readback); receipts carry metadata only.
|
owner capabilities (e.g. ESO sync with readback); receipts carry metadata only.
|
||||||
|
|
@ -62,22 +76,27 @@ def build(*, admitted: Invocation, case_acceptance: str, namespace: str, name: s
|
||||||
raise Refused('delivery invocation boundary differs')
|
raise Refused('delivery invocation boundary differs')
|
||||||
if admitted.operation != 'admission' or admitted.attempt_id:
|
if admitted.operation != 'admission' or admitted.attempt_id:
|
||||||
raise Refused('expected admission invocation')
|
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()
|
desc_bytes = json.dumps(asdict(desc), sort_keys=True, separators=(',', ':')).encode()
|
||||||
if admitted.pins.adapter != digest(desc_bytes):
|
if admitted.pins.adapter != digest(desc_bytes):
|
||||||
raise Refused('descriptor pin differs')
|
raise Refused('descriptor pin differs')
|
||||||
if not (isinstance(case_acceptance, str) and 0 < len(case_acceptance) <= 192
|
if not (isinstance(case_acceptance, str) and 0 < len(case_acceptance) <= 192
|
||||||
and digest(case_acceptance.encode()) == admitted.pins.acceptance):
|
and digest(case_acceptance.encode()) == admitted.pins.acceptance):
|
||||||
raise Refused('acceptance evidence binding differs')
|
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)):
|
if not all(callable(x) for x in (deliver, remove, publish, authenticate, revoke_session)):
|
||||||
raise Refused('owner delivery, evidence and revocation capabilities required')
|
raise Refused('owner delivery, 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 not isinstance(params, dict) or params.get('client_type') != 'confidential':
|
||||||
|
raise Refused('deliver-secret applies only to confidential clients')
|
||||||
|
if params.get('client_id') != admitted.target:
|
||||||
|
raise Refused('case target differs')
|
||||||
|
namespace, name = NAMESPACE, secret_name(admitted.target)
|
||||||
|
custody_reference, key_names = custody_path(admitted.target), KEY_NAMES
|
||||||
|
if not re.fullmatch(r'[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?', name):
|
||||||
|
raise Refused('bounded secret reference required')
|
||||||
binding_ref = {'namespace': namespace, 'name': name, 'custody_reference': custody_reference,
|
binding_ref = {'namespace': namespace, 'name': name, 'custody_reference': custody_reference,
|
||||||
'key_names': list(key_names), 'client_id': admitted.target}
|
'key_names': list(key_names), 'client_id': admitted.target}
|
||||||
records = {}
|
records = {}
|
||||||
|
|
|
||||||
|
|
@ -14,13 +14,18 @@ a = importlib.util.module_from_spec(spec)
|
||||||
spec.loader.exec_module(a)
|
spec.loader.exec_module(a)
|
||||||
|
|
||||||
KEYS = ('client-secret',)
|
KEYS = ('client-secret',)
|
||||||
|
EXAMPLES = ROOT.parents[2] / 'security-case-type-revisions' / '1.2' / 'oidc-browser-relying-party' / 'examples'
|
||||||
|
CASE = (EXAMPLES / 'forgejo.yaml').read_bytes()
|
||||||
|
PUBLIC_CASE = (EXAMPLES / 'example-public-portal.yaml').read_bytes()
|
||||||
|
SECRET_NAME = 'keycape-forgejo-client'
|
||||||
|
CUSTODY = 'platform/workloads/keycape/forgejo'
|
||||||
AFTER = {'uid': 'uid-1', 'resource_version': '3'}
|
AFTER = {'uid': 'uid-1', 'resource_version': '3'}
|
||||||
|
|
||||||
|
|
||||||
class DeliverSecretTests(unittest.TestCase):
|
class DeliverSecretTests(unittest.TestCase):
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
desc = json.dumps(asdict(a.descriptor()), sort_keys=True, separators=(',', ':')).encode()
|
desc = json.dumps(asdict(a.descriptor()), sort_keys=True, separators=(',', ':')).encode()
|
||||||
pins = Pins(digest(b'case'), digest(b'type'), digest(desc), digest(b'catalog'),
|
pins = Pins(digest(CASE), digest(b'type'), digest(desc), digest(b'catalog'),
|
||||||
digest(b'case-acceptance:test'), digest(b'run sheet'))
|
digest(b'case-acceptance:test'), digest(b'run sheet'))
|
||||||
self.admitted = Invocation('deliver-secret', 'railiance-platform', 'forgejo', 'human',
|
self.admitted = Invocation('deliver-secret', 'railiance-platform', 'forgejo', 'human',
|
||||||
'founder', 'platform-secret-delivery', pins)
|
'founder', 'platform-secret-delivery', pins)
|
||||||
|
|
@ -28,16 +33,15 @@ class DeliverSecretTests(unittest.TestCase):
|
||||||
self.rb = replace(self.admitted, attempt_id='attempt-two', operation='rollback')
|
self.rb = replace(self.admitted, attempt_id='attempt-two', operation='rollback')
|
||||||
self.evidence, self.calls, self.revoked = {}, [], []
|
self.evidence, self.calls, self.revoked = {}, [], []
|
||||||
self.before = None
|
self.before = None
|
||||||
self.kw = dict(admitted=self.admitted, case_acceptance='case-acceptance:test',
|
self.kw = dict(admitted=self.admitted, case_source=CASE,
|
||||||
namespace='sso', name='keycape-client-forgejo',
|
case_acceptance='case-acceptance:test',
|
||||||
custody_reference='kv/keycape/forgejo', key_names=KEYS,
|
|
||||||
deliver=self.deliver, remove=self.remove, publish=self.publish,
|
deliver=self.deliver, remove=self.remove, publish=self.publish,
|
||||||
authenticate=lambda ref, p: self.evidence.get(ref) == p,
|
authenticate=lambda ref, p: self.evidence.get(ref) == p,
|
||||||
revoke_session=lambda inv: self.revoked.append(inv) or 'revocation-' + inv.attempt_id)
|
revoke_session=lambda inv: self.revoked.append(inv) or 'revocation-' + inv.attempt_id)
|
||||||
|
|
||||||
def receipt(self, operation, changed, before, after, **extra):
|
def receipt(self, operation, changed, before, after, **extra):
|
||||||
r = {'schema': 'netkingdom.deliver-secret.v1', 'operation': operation, 'namespace': 'sso',
|
r = {'schema': 'netkingdom.deliver-secret.v1', 'operation': operation, 'namespace': 'sso',
|
||||||
'name': 'keycape-client-forgejo', 'custody_reference': 'kv/keycape/forgejo',
|
'name': SECRET_NAME, 'custody_reference': CUSTODY,
|
||||||
'key_names': list(KEYS), 'values_emitted': False, 'changed': changed,
|
'key_names': list(KEYS), 'values_emitted': False, 'changed': changed,
|
||||||
'before': before, 'after': after}
|
'before': before, 'after': after}
|
||||||
r.update(extra)
|
r.update(extra)
|
||||||
|
|
@ -92,14 +96,49 @@ class DeliverSecretTests(unittest.TestCase):
|
||||||
a.build(**self.kw).invoke(self.inv)
|
a.build(**self.kw).invoke(self.inv)
|
||||||
|
|
||||||
def test_build_refusals(self):
|
def test_build_refusals(self):
|
||||||
cases = [dict(namespace='SSO!'), dict(key_names=['client-secret']), dict(key_names=()),
|
cases = [dict(case_acceptance='other'), dict(case_source=CASE.decode()), dict(publish=None),
|
||||||
dict(case_acceptance='other'), dict(publish=None),
|
|
||||||
dict(admitted=replace(self.admitted, lane='other-lane')),
|
dict(admitted=replace(self.admitted, lane='other-lane')),
|
||||||
dict(admitted=replace(self.admitted, pins=replace(self.admitted.pins, adapter=digest(b'x'))))]
|
dict(admitted=replace(self.admitted, pins=replace(self.admitted.pins, adapter=digest(b'x'))))]
|
||||||
for change in cases:
|
for change in cases:
|
||||||
with self.subTest(change=change), self.assertRaises(Refused):
|
with self.subTest(change=change), self.assertRaises(Refused):
|
||||||
a.build(**{**self.kw, **change})
|
a.build(**{**self.kw, **change})
|
||||||
|
|
||||||
|
def test_reference_derived_from_admitted_target_per_type(self):
|
||||||
|
a.build(**self.kw).invoke(self.inv)
|
||||||
|
self.assertEqual(self.calls[0], ('deliver', {
|
||||||
|
'namespace': 'sso', 'name': SECRET_NAME, 'custody_reference': CUSTODY,
|
||||||
|
'key_names': list(KEYS), 'client_id': 'forgejo'}))
|
||||||
|
for caller_ref in ('namespace', 'name', 'custody_reference', 'key_names'):
|
||||||
|
with self.subTest(caller_ref=caller_ref), self.assertRaises(TypeError):
|
||||||
|
a.build(**{**self.kw, caller_ref: 'x'})
|
||||||
|
|
||||||
|
def test_reviewer_demo_admitted_forgejo_grafana_delivery_refused(self):
|
||||||
|
grafana = CASE.replace(b'client_id: forgejo', b'client_id: grafana')
|
||||||
|
# Same admitted forgejo case, grafana case source: pin differs.
|
||||||
|
with self.assertRaises(Refused):
|
||||||
|
a.build(**{**self.kw, 'case_source': grafana})
|
||||||
|
# Admission re-pinned to the grafana source but still targeting forgejo.
|
||||||
|
repinned = replace(self.admitted, pins=replace(self.admitted.pins, case=digest(grafana)))
|
||||||
|
with self.assertRaises(Refused):
|
||||||
|
a.build(**{**self.kw, 'admitted': repinned, 'case_source': grafana})
|
||||||
|
# A grafana-shaped receipt under the forgejo admission is refused.
|
||||||
|
self.kw['deliver'] = lambda inv, b: self.receipt(
|
||||||
|
'deliver', True, None, dict(AFTER), name='keycape-grafana-client',
|
||||||
|
custody_reference='platform/workloads/keycape/grafana')
|
||||||
|
with self.assertRaises(Refused):
|
||||||
|
a.build(**self.kw).invoke(self.inv)
|
||||||
|
|
||||||
|
def test_public_client_case_refused(self):
|
||||||
|
admitted = replace(self.admitted, target='example-portal',
|
||||||
|
pins=replace(self.admitted.pins, case=digest(PUBLIC_CASE)))
|
||||||
|
with self.assertRaises(Refused):
|
||||||
|
a.build(**{**self.kw, 'admitted': admitted, 'case_source': PUBLIC_CASE})
|
||||||
|
|
||||||
|
def test_case_pin_mismatch_refused(self):
|
||||||
|
with self.assertRaises(Refused):
|
||||||
|
a.build(**{**self.kw, 'case_source': CASE + b'\n# edited\n'})
|
||||||
|
self.assertEqual(self.calls, [])
|
||||||
|
|
||||||
def test_attempt_binding_and_reused_evidence_refused(self):
|
def test_attempt_binding_and_reused_evidence_refused(self):
|
||||||
adapter = a.build(**self.kw)
|
adapter = a.build(**self.kw)
|
||||||
with self.assertRaises(Refused):
|
with self.assertRaises(Refused):
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue