NK-WP-0048-T09: keycape-client entry point and declaration
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s

One case-driven entry point for the live KeyCape client registry, with the
existing safeguards; apply requires the case acceptance record. Declaration
net-kingdom.keycape-client-registration (implemented). The type pins it and
adds the confidential delivery and env-binding steps. 48 + 71 tests pass.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 5457@bnt-lap001
Assistant-Session: 792bf84a-3783-4f5d-a811-cded61dd2012
This commit is contained in:
tegwick 2026-10-08 23:03:00 +02:00
parent 2720c05102
commit 6a5857746d
7 changed files with 683 additions and 21 deletions

View file

@ -0,0 +1,117 @@
apiVersion: netkingdom.io/playbook-capability/v0.1
kind: PlaybookCapabilityDeclaration
metadata:
id: net-kingdom.keycape-client-registration
name: KeyCape OIDC client registration (live registry)
owner: net-kingdom
repo: net-kingdom
domain: infotech
contract_version: "0.1"
source_links:
- label: Entry point
path: sso-mfa/k8s/keycape/keycape-client.py
- label: Tests
path: sso-mfa/k8s/keycape/test_keycape_client.py
- label: SecurityCaseType
path: security-case-types/oidc-browser-relying-party/type.yaml
spec:
playbook:
path: sso-mfa/k8s/keycape/keycape-client.py
type: cli-sequence
invocation: keycape-client.py apply --client-file <client.yaml> --case-acceptance <record> --expected-uid <uid> --expected-resource-version <rv>
description: >-
Add or remove exactly one client in the live KeyCape client registry
without exporting secret values: contained kubectl, cluster and issuer
pins, byte-preserving merge, compare-and-swap, read-back. Replaces the
per-client helpers beside it. Field rules follow key-cape's canonical model
(KEY-WP-0028). Modes: inspect, dry-run, apply; operations add and remove.
capabilities:
- id: c1.lightweight-sso
tier: C1
resource_kinds:
- identities
- roles_scopes_policies
description: Register one OIDC relying party with the running lightweight issuer.
parameters:
- name: client_id
type: string
required: true
constraints:
pattern: ^[a-z0-9][a-z0-9-]{1,62}$
sensitivity: operational
tuning_authority: netkingdom_tunable
description: OAuth2 client_id, unique in the registry.
- name: display_name
type: string
required: true
constraints:
pattern: ^[^\n\r]{1,80}$
sensitivity: public
tuning_authority: netkingdom_tunable
description: Human-readable client name shown at sign-in.
- name: redirect_uris
type: string
required: true
constraints:
pattern: ^https://[^*\s,]+(,https://[^*\s,]+)*$
sensitivity: security_sensitive
tuning_authority: netkingdom_tunable
description: Comma-separated exact redirect URIs; wildcards are never accepted.
- name: allowed_scopes
type: string
required: true
constraints:
pattern: ^openid(,(profile|email|groups))*$
sensitivity: security_sensitive
tuning_authority: netkingdom_tunable
description: Comma-separated scopes the client may request; openid first.
- name: client_type
type: string
required: true
constraints:
pattern: ^(confidential|public)$
sensitivity: security_sensitive
tuning_authority: netkingdom_tunable
description: confidential (server-side, holds a secret) or public (PKCE only).
- name: mfa_required
type: string
required: true
constraints:
pattern: ^(true|false)$
sensitivity: security_sensitive
tuning_authority: netkingdom_tunable
description: Pin MFA on this client instead of relying on the provider default.
responsibilities:
- resource_kind: identities
owner: net-kingdom
resources:
- oidc:static-clients
repo_owns: Live registry change for exactly this client, its idempotency and removal; field rules are key-cape's canonical model.
netkingdom_orchestrates: Supply the case's validated, non-secret client fields.
- resource_kind: roles_scopes_policies
owner: key-cape
resources:
- iam-profile:client-scope-allowlists
repo_owns: Enforce exact redirect URIs, allowed scopes and the MFA pin at authorization time.
netkingdom_orchestrates: Choose scopes and MFA posture per SecurityCaseType.
trust:
requires:
- state: runtime_secret_trust
readiness_checks: []
- state: runtime_identity_trust
readiness_checks: []
satisfies:
- state: tenant_onboarding_trust
readiness_checks:
- id: client-authorize-accepts-registered-redirect
description: The authorization endpoint accepts the client with each registered redirect URI.
evidence: GET /authorize with client_id and each redirect_uri returns the login page, not an error.
- id: client-authorize-refuses-other-redirect
description: The authorization endpoint refuses an unregistered redirect URI for the client.
evidence: GET /authorize with an unregistered redirect_uri returns an OAuth error.
catalog:
publish: capabilities/playbooks/net-kingdom.keycape-client-registration.yaml
maturity: implemented
consumers:
- netkingdom-security-scenario-composer
- netkingdom-security-case-types

View file

@ -16,9 +16,9 @@ Until now every browser client was registered by its own script in
`register-user-engine-portal.py`, `vergabe-client-rollout.py`,
`openbao-client-config.py`), each with its own test and an attended founder run.
The scripts share one pattern: a non-exporting merge into the live client
registry, with inspect, dry-run, apply and remove modes. key-cape now declares
that pattern once as the playbook capability `key-cape.oidc-client-registration`.
This type composes it.
registry, with inspect, dry-run, apply and remove modes. net-kingdom now declares
that pattern once as the playbook capability `net-kingdom.keycape-client-registration`,
implemented by `sso-mfa/k8s/keycape/keycape-client.py`. This type composes it.
### Reference instances and fit
@ -69,9 +69,13 @@ roles there), token lifetime policy (KeyCape's), and federation (C5).
## Attended steps
`register-client` is attended. KeyCape's live client registry is a Kubernetes
Secret, and agents do not read or write Secrets. ops-warden groups every attended
step of a run into one operator session (WARDEN-WP-0044). The way to remove this
`register-client` is attended: KeyCape's live client registry is a Kubernetes
Secret, and agents do not read or write Secrets. For confidential clients,
`bind-env` is attended as well, being a change to the identity provider's
Deployment. KeyCape resolves a confidential client's secret from an environment
variable at start, so the binding (and `deliver-secret` before it) must exist
before registration. ops-warden groups both attended steps into one operator
session (WARDEN-WP-0044). The way to remove this
step is a KeyCape change: load clients from a non-secret registry and reference
confidential secrets from custody. That would be a 1.1 or 2.0 of this type.

View file

@ -42,6 +42,18 @@ def with_params(case, **changes):
return c
def test_keycape_client_record_matches_the_accepted_request():
spec_ = importlib.util.spec_from_file_location(
"keycape_client", REPO / "sso-mfa" / "k8s" / "keycape" / "keycape-client.py")
kc = importlib.util.module_from_spec(spec_)
spec_.loader.exec_module(kc)
accepted = yaml.safe_load((REPO.parent / "key-cape" / "docs" / "forgejo-oidc-client-request.yaml").read_text())
record = kc.client_from_case(FORGEJO)
for key in ("clientId", "redirectUris", "allowedScopes", "clientType", "mfaRequired"):
assert record[key] == accepted["client"][key], key
assert record["secretRef"] == accepted["client"]["secretRef"]
def test_forgejo_case_matches_the_accepted_request():
accepted = yaml.safe_load((REPO.parent / "key-cape" / "docs" / "forgejo-oidc-client-request.yaml").read_text())
p = FORGEJO["spec"]["parameters"]
@ -84,17 +96,20 @@ def test_public_case_renders_without_custody_steps():
issues, r = v.render_case(PUBLIC, as_implemented(), DECLS)
assert msgs(issues) == []
assert [m["id"] for m in r["materialization"]] == ["catalog-entry"]
assert "mint-secret" not in [s["id"] for s in r["follow_through"]]
assert not {"mint-secret", "deliver-secret", "bind-env"} & {s["id"] for s in r["follow_through"]}
assert {c["id"] for c in r["verification"]} >= {"authorize-refuses-other-redirect"}
assert not {c["id"] for c in r["verification"]} & {"custody-denied-to-default", "custody-allowed-to-consumer"}
def test_confidential_case_renders_custody_and_one_attended_step():
def test_confidential_case_renders_custody_delivery_and_one_attended_session():
issues, r = v.render_case(FORGEJO, as_implemented(), DECLS)
assert msgs(issues) == []
order = [s["id"] for s in r["follow_through"]]
assert order == ["ensure-groups", "mint-secret", "deliver-secret", "bind-env", "register-client", "hand-over"]
steps = {s["id"]: s for s in r["follow_through"]}
assert set(steps) == {"ensure-groups", "mint-secret", "register-client", "hand-over"}
assert [s for s in steps.values() if s.get("attended")] == [steps["register-client"]]
assert [s["id"] for s in r["follow_through"] if s.get("attended")] == ["bind-env", "register-client"]
assert "keycape-client.py apply" in steps["register-client"]["entry_point"]
assert "sso/keycape-forgejo-client" in steps["deliver-secret"]["entry_point"]
assert steps["mint-secret"]["entry_point"].endswith("bao:platform/workloads/forgejo/netkingdom-oidc-client")
assert steps["hand-over"]["owner"] == "railiance-apps"
plan = r["composition"]

View file

@ -95,14 +95,14 @@ spec:
initial_trust: [runtime_secret_trust, runtime_identity_trust]
capabilities: [c1.lightweight-sso]
providers:
c1.lightweight-sso: key-cape.oidc-client-registration
c1.lightweight-sso: net-kingdom.keycape-client-registration
parameter_bindings:
key-cape.oidc-client-registration.client_id: client_id
key-cape.oidc-client-registration.display_name: display_name
key-cape.oidc-client-registration.redirect_uris: redirect_uris
key-cape.oidc-client-registration.allowed_scopes: allowed_scopes
key-cape.oidc-client-registration.client_type: client_type
key-cape.oidc-client-registration.mfa_required: mfa_required
net-kingdom.keycape-client-registration.client_id: client_id
net-kingdom.keycape-client-registration.display_name: display_name
net-kingdom.keycape-client-registration.redirect_uris: redirect_uris
net-kingdom.keycape-client-registration.allowed_scopes: allowed_scopes
net-kingdom.keycape-client-registration.client_type: client_type
net-kingdom.keycape-client-registration.mfa_required: mfa_required
materialization:
- id: client-secret-path
@ -135,9 +135,28 @@ spec:
auth: "credential broker issuer, scoped to the case's KV path"
warden_executes: true
attended: false
- id: deliver-secret
owner: railiance-platform
when: "client_type == confidential"
entry_point: "ExternalSecret sso/keycape-{client_id}-client (key client-secret) from {secret_ref}"
auth: "railiance-platform ESO delivery through the platform ClusterSecretStore; no value leaves custody"
warden_executes: false
attended: false
- id: bind-env
owner: net-kingdom
when: "client_type == confidential"
entry_point: "patch Deployment sso/keycape: env KEYCAPE_<CLIENT_ID>_CLIENT_SECRET from sso/keycape-{client_id}-client"
auth: "operator kube credentials for namespace sso"
warden_executes: false
attended: true
attended_justification: >-
A change to the live identity provider's Deployment. It runs in the same
attended operator session as register-client, before it, because
KeyCape resolves a confidential client's secret at start and fails
closed if the binding is missing.
- id: register-client
owner: key-cape
entry_point: "keycape-client register (key-cape.oidc-client-registration) for {client_id}"
owner: net-kingdom
entry_point: "sso-mfa/k8s/keycape/keycape-client.py apply --client-file <client-from-case> --case-acceptance <record> (net-kingdom.keycape-client-registration)"
auth: "operator write access to the live KeyCape client registry in namespace sso"
warden_executes: false
attended: true
@ -182,7 +201,13 @@ spec:
- reverses: hand-over
action: "remove the OIDC source for {client_id} from {consuming_service} (owner {consumer_owner_repo})"
- reverses: register-client
action: "keycape-client register --remove {client_id}"
action: "keycape-client.py apply --operation remove for {client_id}"
- reverses: bind-env
when: "client_type == confidential"
action: "remove env KEYCAPE_<CLIENT_ID>_CLIENT_SECRET from Deployment sso/keycape, after register-client is reversed"
- reverses: deliver-secret
when: "client_type == confidential"
action: "delete ExternalSecret sso/keycape-{client_id}-client and its target"
- reverses: mint-secret
when: "client_type == confidential"
action: "destroy the secret at {secret_ref}"

View file

@ -0,0 +1,271 @@
#!/usr/bin/env python3
"""Register or remove exactly one KeyCape client from a SecurityCase, without exporting Secret values.
Entry point of the playbook capability net-kingdom.keycape-client-registration,
composed by the SecurityCaseType oidc-browser-relying-party
(security-case-types/, canon security-case-types_v0.1). It generalizes the
per-client helpers in this directory (portal-client-rollout,
register-coulomb-social, vergabe-client-rollout,
secrets-engine-openbao-client-rollout) and keeps their safeguards:
- contained kubectl, cluster-identity pin and verified upstream issuer pin
- byte-preserving merge: only this client's lines change; every other config
byte, the signing key and unrelated clients are preserved
- exact match: absent -> add; identical -> noop; drift, duplicates or a shared
secretRef -> refuse (update = remove, then add, each accepted)
- compare-and-swap on the Secret's uid and resourceVersion, then read-back
- confidential clients: the KeyCape Deployment must already bind the client's
secret env var (delivered from custody), or the next pod start fails closed
- no Secret value, parser message or Kubernetes error text leaves the process
Modes: inspect (default), dry-run (server-side), apply. apply requires the
case's acceptance record instead of a per-client founder decision id: the case
was accepted under an acknowledged type.
keycape-client.py client-from-case <case.yaml>
keycape-client.py inspect --client-file client.yaml [--operation remove]
keycape-client.py apply --client-file client.yaml --case-acceptance <record> \\
--expected-uid <uid> --expected-resource-version <rv>
"""
from __future__ import annotations
import argparse
import base64
import copy
import importlib.util
import json
import re
import sys
from pathlib import Path
import yaml
ROOT = Path(__file__).resolve().parent
def module(name, filename):
spec = importlib.util.spec_from_file_location(name, ROOT / filename)
result = importlib.util.module_from_spec(spec)
spec.loader.exec_module(result)
return result
lane = module('portal_rollout', 'portal-client-rollout.py')
pin = lane.pin
Refused = lane.Refused
require = lane.require
CASE_TYPE = re.compile(r'^oidc-browser-relying-party@1\.\d+$')
CLIENT_ID = re.compile(r'^[a-z0-9][a-z0-9-]{1,62}$')
REDIRECT = re.compile(r'^https://[^*\s,]+$')
SCOPES = {'openid', 'profile', 'email', 'groups'}
def env_name(client_id):
return 'KEYCAPE_' + client_id.upper().replace('-', '_') + '_CLIENT_SECRET'
def env_secret_name(client_id):
return f'keycape-{client_id}-client'
def client_from_params(params):
"""The KeyCape client record for a case of oidc-browser-relying-party."""
cid = params['client_id']
require(CLIENT_ID.fullmatch(cid or '') is not None, 'invalid_client_id')
redirects = [r for r in params['redirect_uris'].split(',')]
require(redirects and all(REDIRECT.fullmatch(r) for r in redirects), 'redirect_uris_must_be_exact_https')
scopes = params['allowed_scopes'].split(',')
require(scopes[0] == 'openid' and set(scopes) <= SCOPES and len(set(scopes)) == len(scopes),
'allowed_scopes_invalid')
require(params['client_type'] in ('confidential', 'public'), 'client_type_invalid')
require(params['mfa_required'] in ('true', 'false'), 'mfa_required_invalid')
client = {
'clientId': cid,
'displayName': params['display_name'],
'redirectUris': redirects,
'allowedScopes': scopes,
'grantTypes': ['authorization_code'],
'clientType': params['client_type'],
}
if params['client_type'] == 'confidential':
require(str(params.get('secret_ref', '')).startswith('bao:'), 'confidential_client_needs_custody_reference')
client['secretRef'] = 'env:' + env_name(cid)
if params['mfa_required'] == 'true':
client['mfaRequired'] = True
return client
def client_from_case(case):
spec = case.get('spec') or {}
require(CASE_TYPE.fullmatch(str(spec.get('type', ''))) is not None, 'case_type_not_oidc_browser_relying_party')
return client_from_params(spec.get('parameters') or {})
def env_binding(client):
return {'name': env_name(client['clientId']), 'valueFrom': {
'secretKeyRef': {'name': env_secret_name(client['clientId']), 'key': 'client-secret'}}}
def _client_node(raw):
root = yaml.compose(raw)
node = next(value for key, value in root.value if key.value == 'clients')
require(isinstance(node, yaml.nodes.SequenceNode) and not node.flow_style,
'block_client_sequence_required')
return node
def _line_start(raw, index):
start = raw.rfind('\n', 0, index) + 1
return start if not raw[start:index].strip() else index
def _inspect(secret, client):
raw, config, _ = pin.issuer_document(secret)
require(config['authelia'].get('issuer') == pin.VERIFIED_UPSTREAM_ISSUER, 'verified_issuer_pin_required')
clients = config.get('clients')
require(isinstance(clients, list) and clients, 'client_sequence_required')
require(all(isinstance(c, dict) and 'clientId' in c for c in clients), 'client_shape_unexpected')
ids = [c['clientId'] for c in clients]
require(len(ids) == len(set(ids)), 'duplicate_client_id')
if 'secretRef' in client:
others = [c for c in clients if c['clientId'] != client['clientId']]
require(all(c.get('secretRef') != client['secretRef'] for c in others), 'secret_ref_already_bound')
return raw, config, ids
def replacement(secret, client):
"""Return (base64 config, changed) with only this client added."""
raw, config, ids = _inspect(secret, client)
if client['clientId'] in ids:
require(config['clients'][ids.index(client['clientId'])] == client,
'existing_registration_differs_requires_reconciliation')
return secret['data']['config.yaml'], False
node = _client_node(raw)
index = _line_start(raw, node.end_mark.index)
indent = ' ' * node.start_mark.column
fragment = yaml.safe_dump([client], sort_keys=False, width=120)
fragment = ''.join(indent + line if line.strip() else line for line in fragment.splitlines(True))
if index and raw[index - 1] != '\n':
fragment = '\n' + fragment
updated = raw[:index] + fragment + raw[index:]
expected = copy.deepcopy(config)
expected['clients'].append(client)
require(yaml.load(updated, Loader=pin.UniqueLoader) == expected, 'unrelated_configuration_changed')
return base64.b64encode(updated.encode()).decode(), True
def removal(secret, client):
"""Rollback: return (base64 config, changed) with only this exact client removed."""
raw, config, ids = _inspect(secret, client)
if client['clientId'] not in ids:
return secret['data']['config.yaml'], False
position = ids.index(client['clientId'])
require(config['clients'][position] == client, 'existing_registration_differs_requires_reconciliation')
require(len(ids) > 1, 'removal_would_empty_client_sequence')
node = _client_node(raw)
item = node.value[position]
start = raw.rfind('\n', 0, item.start_mark.index) + 1
require(raw[start:item.start_mark.index].strip() == '-', 'unexpected_item_layout')
if position + 1 < len(node.value):
end = raw.rfind('\n', 0, node.value[position + 1].start_mark.index) + 1
else:
end = _line_start(raw, node.end_mark.index)
updated = raw[:start] + raw[end:]
expected = copy.deepcopy(config)
del expected['clients'][position]
require(yaml.load(updated, Loader=pin.UniqueLoader) == expected, 'unrelated_configuration_changed')
return base64.b64encode(updated.encode()).decode(), True
def deployment_bound(deployment, client):
"""Whether the live Deployment carries exactly this client's secret binding."""
if 'secretRef' not in client:
return None
containers = deployment['spec']['template']['spec']['containers']
container = next(c for c in containers if c['name'] == 'keycape')
matches = [e for e in container.get('env', []) if e.get('name') == env_name(client['clientId'])]
require(len(matches) <= 1 and all(e == env_binding(client) for e in matches), 'env_binding_drift')
return bool(matches)
def run(args, client):
kubectl = lane.kubectl
cluster = kubectl(['get', 'namespace', 'kube-system', '-o', 'json'])
require(cluster['metadata']['uid'] == lane.CLUSTER_UID, 'cluster_identity_changed')
before = kubectl(['-n', 'sso', 'get', 'secret', 'keycape-config', '-o', 'json'])
require(not before['metadata'].get('ownerReferences'), 'controller_owned_secret')
metadata = pin.safe_metadata(before)
encoded, changed = (replacement if args.operation == 'add' else removal)(before, client)
bound = None
if 'secretRef' in client:
deployment = kubectl(['-n', 'sso', 'get', 'deployment', 'keycape', '-o', 'json'])
bound = deployment_bound(deployment, client)
receipt = {'schema': 'netkingdom.keycape-client.v1', 'target': 'sso/keycape-config',
'before': metadata, 'operation': args.operation, 'client_id': client['clientId'],
'client_type': client['clientType'], 'change_needed': changed, 'mode': args.mode,
'deployment_env_bound': bound, 'values_emitted': False, 'changed': False}
if args.mode == 'inspect':
return receipt
if args.mode == 'apply':
require(bool(args.case_acceptance), 'case_acceptance_record_required')
receipt['case_acceptance'] = args.case_acceptance
if args.operation == 'add' and bound is not None:
require(bound, 'deployment_env_binding_required_first')
require(metadata == {'uid': args.expected_uid, 'resource_version': args.expected_resource_version},
'observed_revision_changed')
if changed:
patch = [
{'op': 'test', 'path': '/metadata/uid', 'value': metadata['uid']},
{'op': 'test', 'path': '/metadata/resourceVersion', 'value': metadata['resource_version']},
{'op': 'replace', 'path': '/data/config.yaml', 'value': encoded},
]
command = ['-n', 'sso', 'patch', 'secret', 'keycape-config', '--type=json',
'--patch-file=/dev/stdin', '-o', 'json']
if args.mode == 'dry-run':
command += ['--dry-run=server']
result = kubectl(command, patch)
require(result['data'] == dict(before['data'], **{'config.yaml': encoded}), 'patch_response_mismatch')
if args.mode == 'apply':
after = kubectl(['-n', 'sso', 'get', 'secret', 'keycape-config', '-o', 'json'])
require(after['metadata']['uid'] == metadata['uid']
and after['data'] == dict(before['data'], **{'config.yaml': encoded}),
'readback_mismatch_stop_without_stale_replay')
receipt.update(changed=changed, after=pin.safe_metadata(after), restart_required=changed)
receipt.update(unrelated_config_bytes_preserved=True, other_secret_data_unchanged=True)
return receipt
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__.split('\n', 1)[0])
parser.add_argument('mode', nargs='?', default='inspect',
choices=['inspect', 'dry-run', 'apply', 'client-from-case'])
parser.add_argument('case', nargs='?', help='case file, for client-from-case')
parser.add_argument('--client-file', help='KeyCape client record (output of client-from-case)')
parser.add_argument('--operation', default='add', choices=['add', 'remove'])
parser.add_argument('--expected-uid')
parser.add_argument('--expected-resource-version')
parser.add_argument('--case-acceptance', help='the case acceptance record; required for apply')
args = parser.parse_args(argv)
try:
if args.mode == 'client-from-case':
require(bool(args.case), 'case_file_required')
print(yaml.safe_dump(client_from_case(yaml.safe_load(Path(args.case).read_text())), sort_keys=False), end='')
return 0
require(bool(args.client_file), 'client_file_required')
client = yaml.safe_load(Path(args.client_file).read_text())
require(isinstance(client, dict) and 'clientId' in client, 'client_file_shape')
print(json.dumps(run(args, client), sort_keys=True))
return 0
except Refused as error:
print(json.dumps({'status': 'refused', 'reason': str(error)}))
except pin.IssuerPinError as error:
print(json.dumps({'status': 'refused', 'reason': str(error)}))
except Exception:
# Parser and Kubernetes errors can contain full Secret material.
print(json.dumps({'status': 'failed', 'reason': 'contained_operation_failed'}))
return 1
if __name__ == '__main__':
raise SystemExit(main())

View file

@ -0,0 +1,203 @@
import argparse
import base64
import contextlib
import copy
import importlib.util
import io
import json
from pathlib import Path
import unittest
import yaml
spec = importlib.util.spec_from_file_location('keycape_client', Path(__file__).with_name('keycape-client.py'))
kc = importlib.util.module_from_spec(spec)
spec.loader.exec_module(kc)
vspec = importlib.util.spec_from_file_location('vergabe_rollout', Path(__file__).with_name('vergabe-client-rollout.py'))
UID = '00000000-0000-4000-8000-000000000001'
EXISTING = [
{'clientId': 'openbao-admin', 'clientType': 'public',
'redirectUris': ['http://localhost:8250/oidc/callback']},
{'clientId': 'secrets-engine-approval', 'clientType': 'confidential',
'secretRef': 'env:KEYCAPE_SECRETS_ENGINE_APPROVAL_CLIENT_SECRET'},
]
FORGEJO_PARAMS = {
'client_id': 'forgejo', 'display_name': 'Forgejo source forge',
'redirect_uris': 'https://forgejo.coulomb.social/user/oauth2/netkingdom/callback',
'allowed_scopes': 'openid,profile,email,groups', 'client_type': 'confidential',
'mfa_required': 'true', 'secret_ref': 'bao:platform/workloads/forgejo/netkingdom-oidc-client',
}
FORGEJO = kc.client_from_params(FORGEJO_PARAMS)
def fixture(clients=None, tail='tokenLifetime: 15m\n'):
# Synthetic marker values only; no live Secret is loaded by this suite.
raw = ('# preserve this comment\nissuer: https://kc.coulomb.social\n'
'authelia:\n issuer: https://auth.coulomb.social\n clientSecret: synthetic-marker\n'
'clients:\n' + yaml.safe_dump(json.loads(json.dumps(EXISTING if clients is None else clients)),
sort_keys=False) + tail)
secret = {'metadata': {'uid': UID, 'resourceVersion': '7'},
'data': {'config.yaml': base64.b64encode(raw.encode()).decode(),
'signing-key.pem': 'c3ludGhldGljLXNpZ25pbmcta2V5'}}
return secret, raw
def decode(encoded):
return base64.b64decode(encoded).decode()
class ClientRecordTests(unittest.TestCase):
def test_confidential_record_binds_env_secret_and_pins_mfa(self):
self.assertEqual(FORGEJO['secretRef'], 'env:KEYCAPE_FORGEJO_CLIENT_SECRET')
self.assertIs(FORGEJO['mfaRequired'], True)
self.assertEqual(FORGEJO['grantTypes'], ['authorization_code'])
self.assertEqual(kc.env_binding(FORGEJO)['valueFrom']['secretKeyRef']['name'], 'keycape-forgejo-client')
def test_public_record_has_no_secret(self):
params = dict(FORGEJO_PARAMS, client_type='public', mfa_required='false')
params.pop('secret_ref')
client = kc.client_from_params(params)
self.assertNotIn('secretRef', client)
self.assertNotIn('mfaRequired', client)
def test_invalid_params_refused(self):
for change in [dict(redirect_uris='https://x.example/*'), dict(redirect_uris='http://x.example/cb'),
dict(allowed_scopes='profile,openid'), dict(allowed_scopes='openid,admin'),
dict(client_type='weird'), dict(secret_ref='plain-value'), dict(client_id='Bad_Id')]:
with self.subTest(change=change), self.assertRaises(kc.Refused):
kc.client_from_params(dict(FORGEJO_PARAMS, **change))
def test_vergabe_case_reproduces_the_hand_built_registration(self):
vergabe = importlib.util.module_from_spec(vspec)
vspec.loader.exec_module(vergabe)
derived = kc.client_from_params({
'client_id': 'vergabe-demo-company', 'display_name': 'Vergabe Demo Company',
'redirect_uris': vergabe.CLIENT['redirectUris'][0],
'allowed_scopes': ','.join(vergabe.CLIENT['allowedScopes']),
'client_type': 'public', 'mfa_required': 'false'})
self.assertEqual(derived, vergabe.CLIENT)
def test_client_from_case_requires_the_type(self):
case = {'spec': {'type': 'oidc-browser-relying-party@1.0', 'parameters': FORGEJO_PARAMS}}
self.assertEqual(kc.client_from_case(case), FORGEJO)
with self.assertRaises(kc.Refused):
kc.client_from_case({'spec': {'type': 'other@1.0', 'parameters': FORGEJO_PARAMS}})
class MergeTests(unittest.TestCase):
def test_add_appends_only_this_client_and_preserves_bytes(self):
secret, raw = fixture()
before = copy.deepcopy(secret)
encoded, changed = kc.replacement(secret, FORGEJO)
updated = decode(encoded)
self.assertTrue(changed)
head = raw[:raw.index('tokenLifetime:')]
self.assertTrue(updated.startswith(head))
expected = yaml.safe_load(raw)
expected['clients'].append(FORGEJO)
self.assertEqual(yaml.safe_load(updated), expected)
self.assertEqual(secret, before)
def test_identical_is_noop_and_drift_refused(self):
secret, _ = fixture(EXISTING + [FORGEJO])
self.assertEqual(kc.replacement(secret, FORGEJO), (secret['data']['config.yaml'], False))
for change in [dict(redirectUris=['https://evil.example/cb']), dict(clientType='public'),
dict(allowedScopes=['openid'])]:
with self.subTest(change=change), self.assertRaises(kc.Refused):
kc.replacement(fixture(EXISTING + [dict(FORGEJO, **change)])[0], FORGEJO)
def test_shared_secret_ref_and_duplicates_refused(self):
other = dict(EXISTING[1], secretRef=FORGEJO['secretRef'])
with self.assertRaises(kc.Refused):
kc.replacement(fixture([EXISTING[0], other])[0], FORGEJO)
with self.assertRaises(kc.Refused):
kc.replacement(fixture(EXISTING + [EXISTING[0]])[0], FORGEJO)
def test_removal_removes_exactly_this_client(self):
secret, raw = fixture(EXISTING + [FORGEJO])
encoded, changed = kc.removal(secret, FORGEJO)
self.assertTrue(changed)
self.assertEqual(yaml.safe_load(decode(encoded))['clients'], EXISTING)
self.assertEqual(kc.removal(fixture()[0], FORGEJO), (fixture()[0]['data']['config.yaml'], False))
class FakeKube:
def __init__(self, secret, bound=True):
self.secret = secret
self.bound = bound
self.patches = []
def __call__(self, arguments, payload=None):
if arguments[:2] == ['get', 'namespace']:
return {'metadata': {'uid': kc.lane.CLUSTER_UID}}
if 'deployment' in arguments:
env = [kc.env_binding(FORGEJO)] if self.bound else []
return {'spec': {'template': {'spec': {'containers': [{'name': 'keycape', 'env': env}]}}}}
if 'patch' in arguments:
self.patches.append(arguments)
data = dict(self.secret['data'], **{'config.yaml': payload[2]['value']})
if '--dry-run=server' not in arguments:
self.secret = dict(self.secret, data=data)
return {'data': data}
return copy.deepcopy(self.secret)
def args(mode, **kw):
base = dict(mode=mode, operation='add', expected_uid=UID, expected_resource_version='7', case_acceptance=None)
base.update(kw)
return argparse.Namespace(**base)
class RunTests(unittest.TestCase):
def setUp(self):
self.original = kc.lane.kubectl
def tearDown(self):
kc.lane.kubectl = self.original
def test_inspect_changes_nothing_and_emits_no_values(self):
fake = FakeKube(fixture()[0])
kc.lane.kubectl = fake
receipt = kc.run(args('inspect'), FORGEJO)
self.assertTrue(receipt['change_needed'])
self.assertFalse(receipt['values_emitted'])
self.assertEqual([], fake.patches)
self.assertNotIn('synthetic-marker', json.dumps(receipt))
def test_apply_needs_case_acceptance(self):
kc.lane.kubectl = FakeKube(fixture()[0])
with self.assertRaises(kc.Refused):
kc.run(args('apply'), FORGEJO)
def test_apply_needs_env_binding_for_confidential_clients(self):
kc.lane.kubectl = FakeKube(fixture()[0], bound=False)
with self.assertRaises(kc.Refused):
kc.run(args('apply', case_acceptance='decision:x'), FORGEJO)
def test_apply_refuses_a_changed_revision(self):
kc.lane.kubectl = FakeKube(fixture()[0])
with self.assertRaises(kc.Refused):
kc.run(args('apply', case_acceptance='decision:x', expected_resource_version='6'), FORGEJO)
def test_apply_patches_reads_back_and_records_acceptance(self):
fake = FakeKube(fixture()[0])
kc.lane.kubectl = fake
receipt = kc.run(args('apply', case_acceptance='decision:x'), FORGEJO)
self.assertTrue(receipt['changed'])
self.assertTrue(receipt['restart_required'])
self.assertEqual('decision:x', receipt['case_acceptance'])
self.assertEqual(1, len(fake.patches))
self.assertIn(FORGEJO, yaml.safe_load(decode(fake.secret['data']['config.yaml']))['clients'])
def test_main_contains_failures(self):
out = io.StringIO()
with contextlib.redirect_stdout(out):
code = kc.main(['inspect'])
self.assertEqual(1, code)
self.assertEqual('refused', json.loads(out.getvalue())['status'])
if __name__ == '__main__':
unittest.main()

View file

@ -189,7 +189,7 @@ moves to `acknowledged`.
```task
id: NK-WP-0048-T06
status: wait
depends_on: [NK-WP-0048-T05, NK-WP-0048-T08, KEY-WP-0038-T04, MASON-WP-0010-T03, WARDEN-WP-0044-T03]
depends_on: [NK-WP-0048-T05, NK-WP-0048-T08, NK-WP-0048-T09, MASON-WP-0010-T03, WARDEN-WP-0044-T03]
blocking_reason: "Implementation needs the acknowledged type and both executors' light paths."
priority: high
state_hub_task_id: "d9b26bcf-a7c6-59a2-92a0-f913d76a203c"
@ -247,3 +247,30 @@ asked and empty. 7 new tests (39 in the provisioner suite). The type's
`ensure-groups` step and its rollback now name these endpoints. The ops-warden
route for the provisioner's service token belongs to WARDEN-WP-0044-T01.
```task
id: NK-WP-0048-T09
status: done
priority: high
```
### T09 — The `keycape-client` entry point and its declaration
Planned first as key-cape KEY-WP-0038-T04. Moved here because the machinery a
live registration needs already lives in net-kingdom, where KeyCape is deployed
(`sso-mfa/k8s/keycape/`): the contained kubectl, the cluster-identity pin and the
issuer pin. A copy in key-cape would drift.
2026-10-08: done. `sso-mfa/k8s/keycape/keycape-client.py` derives the KeyCape
client record from a case (`client-from-case`). It adds or removes exactly one
client with the existing safeguards (byte-preserving merge, exact match or
refuse, shared-reference refusal, compare-and-swap, read-back, contained errors),
requires the Deployment env binding first for confidential clients, and requires
the case acceptance record for `apply` instead of a per-client founder decision
id. 15 tests, including a regression showing that the vergabe case reproduces
its hand-built registration exactly; all 71 helper tests pass. Declaration
`capabilities/playbooks/net-kingdom.keycape-client-registration.yaml` (contract
PASS, `maturity: implemented`) replaces key-cape's draft. The type now pins it,
and for confidential clients adds `deliver-secret` (delivery from custody into
the sso namespace) and `bind-env` (Deployment env), attended in the same session
as `register-client`. The per-client helpers stay until their clients are
adopted as cases.