CCR-2026-0055 (proposed) and request type attended-service-credential-delivery
A value held in platform custody that no workload reads, delivered into a service by an attended, MFA-gated procedure (privacyIDEA resolver bind for NetKingdom NK-WP-0056-T04). The contract forbids any policy, auth role, ESO object or runtime frontdoor. Modelled on attended-host-key-delivery. 4 tests. The 3 failures in test_credential_change.py also occur on main. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 896183@bnt-lap001 Assistant-Session: 3a18c01a-3205-4d02-94db-9efff25649a2
This commit is contained in:
parent
674a7132d0
commit
2d4a3838bf
5 changed files with 235 additions and 0 deletions
70
scripts/attended_service_delivery.py
Normal file
70
scripts/attended_service_delivery.py
Normal file
|
|
@ -0,0 +1,70 @@
|
|||
#!/usr/bin/env python3
|
||||
"""attended-service-credential-delivery CCRs: a value in platform custody that no workload reads.
|
||||
|
||||
NetKingdom NK-WP-0056-T04 (svc-privacyidea). Some services keep a credential in
|
||||
their own encrypted store and accept it only through an administrator API that
|
||||
requires a person's MFA (privacyIDEA's LDAP resolver bind). The value is held in
|
||||
OpenBao, written once with CAS=0 from the attended platform-admin session, and
|
||||
delivered by an attended procedure that reads it in memory and writes it into
|
||||
the service. No Kubernetes or workload auth role, policy or ESO object exists,
|
||||
and the runtime frontdoor stays disabled. Modelled on attended-host-key-delivery
|
||||
(CCR-2026-0028). Nothing here reads, prints or accepts a secret value.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
REQUEST_TYPE = "attended-service-credential-delivery"
|
||||
APPROVED = {"applied", "verified", "active"} # approve sets "approved" after the first reviewer
|
||||
KV_RE = re.compile(r"^platform/workloads/[a-z0-9][a-z0-9-]*(/[a-z0-9][a-z0-9-]*)+$")
|
||||
FIELD_RE = re.compile(r"^[A-Z][A-Z0-9_]*$")
|
||||
AUTH = {"method": "attended-oidc", "mount": "netkingdom", "role": "platform-admin"}
|
||||
DELIVERY_KEYS = ("surface", "service", "target", "procedure", "gate")
|
||||
DOC = "docs/credential-lane-designs/attended-service-credential-delivery.md"
|
||||
|
||||
|
||||
def validate_contract(ccr: dict[str, Any]) -> list[str]:
|
||||
errors: list[str] = []
|
||||
if ccr.get("request_type") != REQUEST_TYPE:
|
||||
errors.append("unsupported attended service contract")
|
||||
openbao = ccr.get("openbao")
|
||||
if not isinstance(openbao, dict):
|
||||
return errors + ["openbao must be an object"]
|
||||
if openbao.get("mount") != "platform" or not KV_RE.match(str(openbao.get("kv_path", ""))):
|
||||
errors.append("openbao.kv_path must be a platform/workloads/... path on mount platform")
|
||||
fields = openbao.get("fields")
|
||||
if not isinstance(fields, list) or not fields or not all(isinstance(f, str) and FIELD_RE.match(f) for f in fields):
|
||||
errors.append("openbao.fields must list upper-case field names")
|
||||
if openbao.get("auth") != AUTH:
|
||||
errors.append("openbao.auth must be the attended platform-admin OIDC session only")
|
||||
if openbao.get("runtime_access") is not False or openbao.get("initial_cas") != 0:
|
||||
errors.append("runtime_access must be false and initial_cas 0")
|
||||
extra = set(openbao) - {"mount", "kv_path", "fields", "auth", "runtime_access", "initial_cas"}
|
||||
if extra:
|
||||
errors.append("openbao must not declare policies, roles or other access: " + ", ".join(sorted(extra)))
|
||||
if (ccr.get("access_frontdoor") or {}).get("resolvable") is not False:
|
||||
errors.append("runtime credential frontdoor must remain disabled")
|
||||
delivery = ccr.get("delivery")
|
||||
if not isinstance(delivery, dict) or any(not isinstance(delivery.get(k), str) or not delivery[k] for k in DELIVERY_KEYS):
|
||||
errors.append("delivery needs surface, service, target, procedure and gate")
|
||||
elif delivery["surface"] != "attended-service-config":
|
||||
errors.append("delivery.surface must be attended-service-config")
|
||||
if ccr.get("status") in APPROVED:
|
||||
comments = (ccr.get("review") or {}).get("comments") or []
|
||||
approved = {str(c.get("reviewer", "")).removeprefix("User (").removesuffix(")")
|
||||
for c in comments if c.get("decision") == "approved"}
|
||||
for role in (ccr.get("review") or {}).get("required_approvers") or []:
|
||||
if role not in approved:
|
||||
errors.append("explicit role approval missing: " + role)
|
||||
for section, keys in {"verification": ["positive", "negative", "activation_conditions"],
|
||||
"lifecycle": ["deactivate", "rotate", "compromised"]}.items():
|
||||
for key in keys:
|
||||
if not (ccr.get(section) or {}).get(key):
|
||||
errors.append(f"{section}.{key} required")
|
||||
return errors
|
||||
|
||||
|
||||
def procedure_note(ccr: dict[str, Any]) -> str:
|
||||
return (f"Attended delivery into {ccr['delivery']['service']} through {ccr['delivery']['procedure']} "
|
||||
f"({DOC}). No workload policy, auth role or ESO object is created.")
|
||||
|
|
@ -518,6 +518,8 @@ def validate_ccr(path: Path) -> tuple[dict[str, Any], list[str], list[str]]:
|
|||
owner = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(owner)
|
||||
errors.extend(owner.validate_contract(ccr))
|
||||
elif request_type == "attended-service-credential-delivery":
|
||||
errors.extend(attended_service_module().validate_contract(ccr))
|
||||
elif request_type in LANE_MODULES:
|
||||
errors.extend(lane_module(ccr).validate(ccr, REPO_DIR))
|
||||
else:
|
||||
|
|
@ -724,6 +726,8 @@ def auth_payload(ccr: dict[str, Any]) -> dict[str, Any]:
|
|||
def render_plan(ccr: dict[str, Any]) -> str:
|
||||
if ccr.get("request_type") == "attended-host-key-delivery":
|
||||
return "Use docs/credential-lane-designs/railiance-clock-host-delivery.md through the attended owner procedure. No workload policy or auth role is created."
|
||||
if ccr.get("request_type") == "attended-service-credential-delivery":
|
||||
return attended_service_module().procedure_note(ccr)
|
||||
openbao = ccr["openbao"]
|
||||
auth = openbao["auth"]
|
||||
payload = auth_payload(ccr)
|
||||
|
|
@ -755,6 +759,8 @@ def render_plan(ccr: dict[str, Any]) -> str:
|
|||
def render_operator_commands(ccr: dict[str, Any]) -> str:
|
||||
if ccr.get("request_type") == "attended-host-key-delivery":
|
||||
return "Use docs/credential-lane-designs/railiance-clock-host-delivery.md through the attended owner procedure. No workload policy or auth role is created."
|
||||
if ccr.get("request_type") == "attended-service-credential-delivery":
|
||||
return attended_service_module().procedure_note(ccr)
|
||||
openbao = ccr["openbao"]
|
||||
auth = openbao["auth"]
|
||||
auth_path = f"auth/{auth['mount']}/role/{auth['role']}"
|
||||
|
|
@ -835,6 +841,15 @@ LANE_MODULES = {
|
|||
}
|
||||
|
||||
|
||||
def attended_service_module():
|
||||
"""Owner contract for attended-service-credential-delivery (no workload access)."""
|
||||
import importlib.util
|
||||
spec = importlib.util.spec_from_file_location("attended_service_delivery", REPO_DIR / "scripts/attended_service_delivery.py")
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
def lane_module(ccr: dict[str, Any]):
|
||||
"""Lazy-load the applier module for a lane request type, or None."""
|
||||
name = LANE_MODULES.get(ccr.get("request_type"))
|
||||
|
|
@ -1115,6 +1130,8 @@ def render_applier_apply_plan(ccr: dict[str, Any], warnings: list[str]) -> str:
|
|||
def runbook_readiness_blockers(ccr: dict[str, Any]) -> list[str]:
|
||||
if ccr.get("request_type") == "attended-host-key-delivery":
|
||||
return ["use the attended owner procedure in docs/credential-lane-designs/railiance-clock-host-delivery.md"]
|
||||
if ccr.get("request_type") == "attended-service-credential-delivery":
|
||||
return ["use the attended procedure: " + attended_service_module().procedure_note(ccr)]
|
||||
blockers: list[str] = []
|
||||
status = ccr.get("status")
|
||||
if status not in RUNBOOK_ALLOWED_STATUSES:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue