8cab281ffc
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
70 lines
4 KiB
Python
70 lines
4 KiB
Python
#!/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
|
|
(privacyIDEA's LDAP resolver bind) that a person operates. 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.")
|