tegwick
894761477f
key-cape confirmed groups_claim=groups, bound group fluid-telegram-operators, assurance aal2 and the three openbao-admin callbacks. The OIDC executor now pins /assurance/level=aal2 (bound_claims_type=string) and accepts the exact operator-tunnel UI callback; the retired public UI callback stays refused. T01 now waits on the founder go for the LLDAP group and a scoped survey grant. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 327203@bnt-lap001 Assistant-Session: 55028ae3-4914-4bd1-8f5c-815f3527aef0
485 lines
21 KiB
Python
485 lines
21 KiB
Python
"""Phase 4: execute an approved construction plan against OpenBao.
|
|
|
|
Narrow on purpose (MASON-WP-0001-T04) — implements only the operations
|
|
the first real plan (rein-openweights-openrouter-approle,
|
|
credential_type: openbao-approle-kv) needs: a read-only policy scoped to
|
|
one KV path, an AppRole bound to it, and role_id/secret_id delivery.
|
|
|
|
Never writes to the KV path itself and never touches the downstream
|
|
secret value (the OpenRouter API key). The KV path a policy references
|
|
comes into existence later, when the founder does a paste-once-provision
|
|
through ops-warden's desk (`bao kv put`) — not through this executor.
|
|
role_id/secret_id are the AppRole's own access credential, not the
|
|
downstream secret; generating and delivering them is squarely
|
|
ops-mason's job (INTENT.md: "create ... credentials, tokens"), but they
|
|
are still handled write-only — never logged, never returned in a
|
|
printable form beyond confirmation that delivery happened.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import stat
|
|
import subprocess
|
|
from contextlib import contextmanager
|
|
from dataclasses import asdict, dataclass, replace
|
|
from pathlib import Path
|
|
|
|
from ops_mason.audit import record_build
|
|
from ops_mason.plan import ConstructionPlan, PlanError
|
|
|
|
|
|
class BuildRefused(RuntimeError):
|
|
"""The plan is not approved. This is the one place a bug is a security incident."""
|
|
|
|
|
|
class BuildError(RuntimeError):
|
|
"""The plan was approved but a bao operation failed."""
|
|
|
|
|
|
@dataclass
|
|
class AppRoleKVSpec:
|
|
policy_name: str
|
|
kv_path: str
|
|
approle_name: str
|
|
# token_num_uses has no default on purpose: OpenBao's own default (0)
|
|
# means *unlimited* uses, the opposite of "bounded" -- a plan that says
|
|
# "bounded token_num_uses" (the agent-harness-binky-mail shape, which
|
|
# actually uses 8) must have that number chosen explicitly, not
|
|
# inherited silently from whatever OpenBao considers a sane default.
|
|
token_num_uses: int
|
|
kv_capabilities: tuple[str, ...] = ("read",)
|
|
token_ttl: str = "15m"
|
|
token_max_ttl: str = "30m"
|
|
secret_id_ttl: str = "0"
|
|
delivery_dir: Path | None = None
|
|
bao_bin: str = "bao"
|
|
audit_log_path: Path | None = None
|
|
# When True the named policy already exists (and may grant more than one
|
|
# KV path). Do not rewrite it — AppRole bind only.
|
|
reuse_policy: bool = False
|
|
reuse_policy_sha256: str | None = None
|
|
|
|
|
|
@dataclass
|
|
class KubernetesKVSpec:
|
|
policy_name: str
|
|
role_name: str
|
|
service_account_names: tuple[str, ...]
|
|
service_account_namespaces: tuple[str, ...]
|
|
policy_sha256: str = ""
|
|
token_ttl: str = "15m"
|
|
audit_log_path: Path | None = None
|
|
bao_bin: str = "bao"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class OIDCKVSpec:
|
|
"""Attended four-entry operator matrix; all identity evidence is reviewed input."""
|
|
policy_name: str
|
|
role_name: str
|
|
campaign_path: str
|
|
bound_group: str
|
|
allowed_redirect_uris: tuple[str, ...]
|
|
identity_confirmation_ref: str
|
|
platform_matrix_review_ref: str
|
|
auth_mount: str = "netkingdom"
|
|
token_ttl: str = "15m"
|
|
token_max_ttl: str = "30m"
|
|
bao_bin: str = "bao"
|
|
audit_log_path: Path | None = None
|
|
|
|
|
|
# KeyCape sets assurance.level=aal2 only after a verified privacyIDEA factor
|
|
# (key-cape reply 2026-10-08, as in CCR-2026-0034); every attended role pins it.
|
|
_REQUIRED_ASSURANCE = {"/assurance/level": "aal2"}
|
|
_UI_CALLBACK = "http://127.0.0.1:18200/ui/vault/auth/{mount}/oidc/callback"
|
|
|
|
_OPERATOR_MATRIX = (
|
|
("operator-app", ("read",)),
|
|
("operator-session", ("create", "read", "update")),
|
|
("bot-token", ("create", "read", "update")),
|
|
("redaction-salt", ("create", "read")),
|
|
)
|
|
|
|
|
|
def oidc_policy_hcl(spec: OIDCKVSpec) -> str:
|
|
"""Render exact KV data paths plus self-revocation; no metadata grants."""
|
|
_validate_oidc_spec(spec)
|
|
mount, rest = spec.campaign_path.split("/", 1)
|
|
stanzas = []
|
|
for entry, caps in _OPERATOR_MATRIX:
|
|
encoded = ", ".join(json.dumps(c) for c in caps)
|
|
stanzas.append(f'path "{mount}/data/{rest}/{entry}" {{\n capabilities = [{encoded}]\n}}\n')
|
|
stanzas.append('path "auth/token/revoke-self" {\n capabilities = ["update"]\n}\n')
|
|
return "\n".join(stanzas)
|
|
|
|
|
|
def _validate_oidc_spec(spec: OIDCKVSpec) -> None:
|
|
from urllib.parse import urlsplit
|
|
for value, label in ((spec.role_name, "OIDC role"), (spec.auth_mount, "auth mount"),
|
|
(spec.bound_group, "bound group")):
|
|
_name(value, label)
|
|
if not isinstance(spec.campaign_path, str):
|
|
raise BuildRefused("campaign path must be literal")
|
|
parts = spec.campaign_path.split("/")
|
|
if len(parts) != 5 or parts[0:2] != ["platform", "workloads"] or parts[3] != "fluid-telegram":
|
|
raise BuildRefused("operator matrix requires platform/workloads/<tenant>/fluid-telegram/<campaign>")
|
|
for part in parts:
|
|
_name(part, "campaign path segment")
|
|
for ref in (spec.identity_confirmation_ref, spec.platform_matrix_review_ref):
|
|
if not isinstance(ref, str) or not ref.strip() or ref.strip().lower() in {"pending", "unconfirmed", "unknown", "draft"}:
|
|
raise BuildRefused("confirmed identity and platform matrix review references are required")
|
|
ttl, maximum = _duration(spec.token_ttl, "token_ttl"), _duration(spec.token_max_ttl, "token_max_ttl")
|
|
if not 0 < ttl <= 900 or not ttl <= maximum <= 1800:
|
|
raise BuildRefused("attended OIDC token bounds must be at most 15m/30m")
|
|
if not isinstance(spec.allowed_redirect_uris, tuple) or not spec.allowed_redirect_uris:
|
|
raise BuildRefused("explicit OIDC callback tuple required")
|
|
if len(set(spec.allowed_redirect_uris)) != len(spec.allowed_redirect_uris):
|
|
raise BuildRefused("duplicate OIDC callback")
|
|
for uri in spec.allowed_redirect_uris:
|
|
if not isinstance(uri, str):
|
|
raise BuildRefused("OIDC callback must be a URI")
|
|
if uri == _UI_CALLBACK.format(mount=spec.auth_mount):
|
|
continue
|
|
try:
|
|
u = urlsplit(uri)
|
|
valid = (u.scheme == "http" and u.hostname in {"localhost", "127.0.0.1"}
|
|
and u.port == 8250 and u.path == "/oidc/callback" and not u.username
|
|
and not u.password and not u.query and not u.fragment)
|
|
except ValueError:
|
|
valid = False
|
|
if not valid:
|
|
raise BuildRefused("matrix engine accepts only the reviewed loopback CLI callbacks on port 8250 and the operator-tunnel UI callback")
|
|
|
|
|
|
def oidc_role_document(spec: OIDCKVSpec) -> dict:
|
|
_validate_oidc_spec(spec)
|
|
return {
|
|
"role_type": "oidc", "user_claim": "sub", "groups_claim": "groups",
|
|
"oidc_scopes": ["openid", "profile", "email", "groups"],
|
|
"bound_claims_type": "string",
|
|
"bound_claims": {"groups": [spec.bound_group], **_REQUIRED_ASSURANCE},
|
|
"allowed_redirect_uris": list(spec.allowed_redirect_uris),
|
|
"token_policies": [spec.policy_name], "token_no_default_policy": True,
|
|
"token_ttl": spec.token_ttl, "token_max_ttl": spec.token_max_ttl,
|
|
"token_explicit_max_ttl": spec.token_max_ttl, "token_num_uses": 0,
|
|
}
|
|
|
|
|
|
def build_oidc_kv_lane(plan: ConstructionPlan, spec: OIDCKVSpec) -> dict[str, str]:
|
|
"""Write only policy/auth structure after fresh exact-spec approval."""
|
|
spec = replace(spec)
|
|
plan = _approved_build(plan, spec)
|
|
_run(spec.bao_bin, ["policy", "write", spec.policy_name, "-"], input_text=oidc_policy_hcl(spec))
|
|
_run(spec.bao_bin, ["write", f"auth/{spec.auth_mount}/role/{spec.role_name}", "-"],
|
|
input_text=json.dumps(oidc_role_document(spec)))
|
|
objects = {"approved_spec_sha256": plan.approved_spec_sha256,
|
|
"policy_name": spec.policy_name, "oidc_role_name": spec.role_name,
|
|
"campaign_path": spec.campaign_path}
|
|
record_build(plan_id=plan.id, plan_path=str(plan.path), approved_by=plan.approved_by or "",
|
|
approved_at=plan.approved_at or "", objects=objects, log_path=spec.audit_log_path)
|
|
return objects
|
|
|
|
|
|
def _run(bao_bin: str, args: list[str], input_text: str | None = None) -> str:
|
|
try:
|
|
proc = subprocess.run(
|
|
[bao_bin, *args], input=input_text, capture_output=True, text=True, timeout=30,
|
|
)
|
|
except (OSError, UnicodeError, subprocess.TimeoutExpired):
|
|
raise BuildError("OpenBao command unavailable or timed out; output suppressed") from None
|
|
if proc.returncode != 0:
|
|
raise BuildError(f"OpenBao command failed (exit {proc.returncode}); output suppressed")
|
|
return proc.stdout
|
|
|
|
|
|
def _policy_hcl(kv_path: str, capabilities: tuple[str, ...]) -> str:
|
|
"""Policy HCL for a KV v2 path — grants on both data/ and metadata/ sub-paths.
|
|
|
|
KV v2 routes actual secret reads/writes through `<mount>/data/<path>`;
|
|
a policy written against the bare `<mount>/<path>` (the KV v1 shape)
|
|
silently denies everything on a v2 mount. Caught live during
|
|
MASON-WP-0001-T05's build — `bao token capabilities` on the bare path
|
|
reported full access, but the actual `kv get` still 403'd, because
|
|
OpenBao evaluates policy against the real `data/`-prefixed path, not
|
|
the one a caller might naively check capabilities against.
|
|
"""
|
|
if not isinstance(kv_path, str) or len(kv_path.split("/")) < 2:
|
|
raise BuildRefused("KV path must contain a mount and a literal entry")
|
|
for part in kv_path.split("/"):
|
|
_name(part, "KV path segment")
|
|
if not isinstance(capabilities, tuple) or capabilities != ("read",):
|
|
raise BuildRefused("this KV read-lane engine permits only the read capability")
|
|
mount, _, rest = kv_path.partition("/")
|
|
caps = ", ".join(f'"{c}"' for c in capabilities)
|
|
return (
|
|
f'path "{mount}/data/{rest}" {{\n capabilities = [{caps}]\n}}\n\n'
|
|
f'path "{mount}/metadata/{rest}" {{\n capabilities = [{caps}]\n}}\n'
|
|
)
|
|
|
|
|
|
def _name(value: str, label: str) -> None:
|
|
if not isinstance(value, str) or not re.fullmatch(r"[a-zA-Z0-9][a-zA-Z0-9_.-]*", value):
|
|
raise BuildRefused(f"{label} must be one literal name, without wildcards or separators")
|
|
|
|
|
|
def _duration(value: str, label: str, *, allow_zero: bool = False) -> int:
|
|
if allow_zero and value == "0":
|
|
return 0
|
|
if not isinstance(value, str) or not re.fullmatch(r"[1-9][0-9]*[smh]", value):
|
|
raise BuildRefused(f"{label} must be a positive duration in s, m or h")
|
|
return int(value[:-1]) * {"s": 1, "m": 60, "h": 3600}[value[-1]]
|
|
|
|
|
|
def _hash(value: str) -> None:
|
|
if not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{64}", value):
|
|
raise BuildRefused("reused policy requires an exact SHA-256 content pin")
|
|
|
|
|
|
def _validate_spec(spec: AppRoleKVSpec | KubernetesKVSpec | OIDCKVSpec) -> None:
|
|
_name(spec.policy_name, "policy name")
|
|
_duration(spec.token_ttl, "token_ttl")
|
|
if not isinstance(spec.bao_bin, str) or not spec.bao_bin.strip():
|
|
raise BuildRefused("bao_bin must name an executable")
|
|
if spec.audit_log_path is not None and not isinstance(spec.audit_log_path, Path):
|
|
raise BuildRefused("audit_log_path must be a Path")
|
|
if isinstance(spec, AppRoleKVSpec):
|
|
_name(spec.approle_name, "AppRole name")
|
|
_policy_hcl(spec.kv_path, spec.kv_capabilities)
|
|
if type(spec.token_num_uses) is not int or spec.token_num_uses < 0:
|
|
raise BuildRefused("token_num_uses must be an explicit nonnegative integer")
|
|
if _duration(spec.token_max_ttl, "token_max_ttl") < _duration(spec.token_ttl, "token_ttl"):
|
|
raise BuildRefused("token_max_ttl must not be shorter than token_ttl")
|
|
_duration(spec.secret_id_ttl, "secret_id_ttl", allow_zero=True)
|
|
if type(spec.reuse_policy) is not bool:
|
|
raise BuildRefused("reuse_policy must be boolean")
|
|
if spec.reuse_policy:
|
|
_hash(spec.reuse_policy_sha256)
|
|
elif spec.reuse_policy_sha256 is not None:
|
|
raise BuildRefused("policy pin is only used with reuse_policy")
|
|
if not isinstance(spec.delivery_dir, Path) or not spec.delivery_dir.is_absolute():
|
|
raise BuildRefused("credential delivery requires an explicit absolute directory")
|
|
if ".." in spec.delivery_dir.parts:
|
|
raise BuildRefused("credential delivery path must not contain parent traversal")
|
|
if spec.audit_log_path is not None and spec.audit_log_path.absolute() in {
|
|
spec.delivery_dir, spec.delivery_dir / "role_id", spec.delivery_dir / "secret_id"
|
|
}:
|
|
raise BuildRefused("audit path must not overlap credential delivery")
|
|
elif isinstance(spec, OIDCKVSpec):
|
|
_validate_oidc_spec(spec)
|
|
else:
|
|
_name(spec.role_name, "Kubernetes role name")
|
|
_hash(spec.policy_sha256)
|
|
for values in (spec.service_account_names, spec.service_account_namespaces):
|
|
if not isinstance(values, tuple) or not values:
|
|
raise BuildRefused("Kubernetes role requires explicit service account bindings")
|
|
for value in values:
|
|
if not isinstance(value, str) or not re.fullmatch(r"[a-z0-9](?:[a-z0-9-]*[a-z0-9])?", value) or len(value) > 63:
|
|
raise BuildRefused("service account bindings must be literal DNS labels")
|
|
if len(set(values)) != len(values):
|
|
raise BuildRefused("duplicate service account binding")
|
|
|
|
|
|
def build_spec_document(spec: AppRoleKVSpec | KubernetesKVSpec | OIDCKVSpec) -> dict:
|
|
"""Render the complete, value-free execution input for phase-3 review."""
|
|
_validate_spec(spec)
|
|
fields = asdict(spec)
|
|
for key, value in fields.items():
|
|
if isinstance(value, Path):
|
|
fields[key] = str(value.absolute())
|
|
elif isinstance(value, tuple):
|
|
fields[key] = list(value)
|
|
return {
|
|
"engine": "openbao-approle-kv" if isinstance(spec, AppRoleKVSpec) else "openbao-oidc-kv-read-write" if isinstance(spec, OIDCKVSpec) else "openbao-kubernetes-kv",
|
|
**fields,
|
|
}
|
|
|
|
|
|
def build_spec_digest(document: dict) -> str:
|
|
"""Hash canonical JSON; this is an integrity marker, not a signature."""
|
|
try:
|
|
encoded = json.dumps(document, sort_keys=True, separators=(",", ":"), allow_nan=False).encode()
|
|
except (TypeError, ValueError) as exc:
|
|
raise BuildRefused("build specification is not canonical JSON") from exc
|
|
return hashlib.sha256(encoded).hexdigest()
|
|
|
|
|
|
def _approved_build(plan: ConstructionPlan, spec) -> ConstructionPlan:
|
|
# The file is authoritative, so revoking approval after load takes effect.
|
|
try:
|
|
current = ConstructionPlan.load(plan.path)
|
|
except (OSError, PlanError) as exc:
|
|
raise BuildRefused("cannot reload construction-plan approval") from exc
|
|
if current.id != plan.id or not current.is_approved():
|
|
raise BuildRefused("construction plan is not approved; refusing to build")
|
|
document = build_spec_document(spec)
|
|
if current.credential_type != document["engine"]:
|
|
raise BuildRefused("credential_type does not match the build engine")
|
|
if not isinstance(current.build_spec, dict) or not current.approved_spec_sha256:
|
|
raise BuildRefused("plan requires build_spec and approved_spec_sha256")
|
|
expected = build_spec_digest(document)
|
|
if build_spec_digest(current.build_spec) != expected or current.approved_spec_sha256 != expected:
|
|
raise BuildRefused("build specification does not match the approved specification digest")
|
|
return current
|
|
|
|
|
|
def _verify_policy(bao_bin: str, name: str, expected: str) -> None:
|
|
content = _run(bao_bin, ["policy", "read", name])
|
|
if hashlib.sha256(content.encode()).hexdigest() != expected:
|
|
raise BuildRefused("live reused policy does not match the approved content pin")
|
|
|
|
|
|
@contextmanager
|
|
def _delivery_files(path: Path):
|
|
"""Reserve private files before Bao writes, without following any symlink.
|
|
|
|
Hold directory/file descriptors throughout delivery. Existing files are
|
|
never opened or overwritten. Failed issuance leaves private partial files
|
|
for explicit recovery; retries must use a newly reviewed destination.
|
|
"""
|
|
directory = os.open("/", os.O_RDONLY | os.O_DIRECTORY)
|
|
files = {}
|
|
try:
|
|
for component in path.parts[1:]:
|
|
try:
|
|
os.mkdir(component, 0o700, dir_fd=directory)
|
|
except FileExistsError:
|
|
pass
|
|
child = os.open(component, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=directory)
|
|
os.close(directory)
|
|
directory = child
|
|
info = os.fstat(directory)
|
|
if info.st_uid != os.geteuid() or stat.S_IMODE(info.st_mode) != 0o700:
|
|
raise BuildRefused("delivery directory must be owned by the caller with mode 0700")
|
|
# Check both first so a pre-existing destination never causes a write.
|
|
for name in ("role_id", "secret_id"):
|
|
try:
|
|
os.stat(name, dir_fd=directory, follow_symlinks=False)
|
|
except FileNotFoundError:
|
|
continue
|
|
raise BuildRefused("credential destination already exists; refusing overwrite")
|
|
for name in ("role_id", "secret_id"):
|
|
files[name] = os.open(name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
|
|
0o600, dir_fd=directory)
|
|
os.fchmod(files[name], 0o600)
|
|
except (OSError, BuildRefused) as exc:
|
|
for fd in files.values():
|
|
os.close(fd)
|
|
os.close(directory)
|
|
if isinstance(exc, BuildRefused):
|
|
raise
|
|
raise BuildRefused("cannot reserve safe credential delivery files") from None
|
|
try:
|
|
yield files
|
|
finally:
|
|
for fd in files.values():
|
|
os.close(fd)
|
|
os.close(directory)
|
|
|
|
|
|
def _write_credential(fd: int, value: str) -> None:
|
|
if not value or "\n" in value or "\r" in value or "\x00" in value:
|
|
raise BuildError("credential command returned an invalid response")
|
|
with os.fdopen(os.dup(fd), "w") as stream:
|
|
stream.write(value + "\n")
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
|
|
|
|
def build_approle_kv_lane(plan: ConstructionPlan, spec: AppRoleKVSpec) -> dict[str, str]:
|
|
"""Create the policy + AppRole for an `openbao-approle-kv` plan, deliver role_id/secret_id.
|
|
|
|
Refuses unless plan.is_approved(). Returns object names only (no
|
|
secret material) and appends a metadata-only audit record.
|
|
"""
|
|
spec = replace(spec)
|
|
plan = _approved_build(plan, spec)
|
|
if spec.reuse_policy:
|
|
_verify_policy(spec.bao_bin, spec.policy_name, spec.reuse_policy_sha256)
|
|
with _delivery_files(spec.delivery_dir) as files:
|
|
if not spec.reuse_policy:
|
|
policy_hcl = _policy_hcl(spec.kv_path, spec.kv_capabilities)
|
|
_run(spec.bao_bin, ["policy", "write", spec.policy_name, "-"], input_text=policy_hcl)
|
|
|
|
_run(
|
|
spec.bao_bin,
|
|
[
|
|
"write",
|
|
f"auth/approle/role/{spec.approle_name}",
|
|
f"token_policies={spec.policy_name}",
|
|
f"token_ttl={spec.token_ttl}",
|
|
f"token_max_ttl={spec.token_max_ttl}",
|
|
f"token_num_uses={spec.token_num_uses}",
|
|
f"secret_id_ttl={spec.secret_id_ttl}",
|
|
],
|
|
)
|
|
|
|
role_id = _run(
|
|
spec.bao_bin, ["read", "-field=role_id", f"auth/approle/role/{spec.approle_name}/role-id"]
|
|
).strip()
|
|
secret_id = _run(
|
|
spec.bao_bin,
|
|
["write", "-field=secret_id", "-f", f"auth/approle/role/{spec.approle_name}/secret-id"],
|
|
).strip()
|
|
|
|
_write_credential(files["role_id"], role_id)
|
|
_write_credential(files["secret_id"], secret_id)
|
|
delivered_to = str(spec.delivery_dir)
|
|
|
|
objects = {
|
|
"approved_spec_sha256": plan.approved_spec_sha256,
|
|
"policy_name": spec.policy_name,
|
|
"approle_name": spec.approle_name,
|
|
"kv_path": spec.kv_path,
|
|
"delivered_to": delivered_to,
|
|
}
|
|
record_build(
|
|
plan_id=plan.id,
|
|
plan_path=str(plan.path),
|
|
approved_by=plan.approved_by or "",
|
|
approved_at=plan.approved_at or "",
|
|
objects=objects,
|
|
log_path=spec.audit_log_path,
|
|
)
|
|
return objects
|
|
|
|
|
|
def build_kubernetes_kv_lane(
|
|
plan: ConstructionPlan, spec: KubernetesKVSpec
|
|
) -> dict[str, str]:
|
|
"""Create a policy-bound Kubernetes auth role without handling secret values."""
|
|
spec = replace(spec)
|
|
plan = _approved_build(plan, spec)
|
|
_verify_policy(spec.bao_bin, spec.policy_name, spec.policy_sha256)
|
|
|
|
_run(
|
|
spec.bao_bin,
|
|
[
|
|
"write",
|
|
f"auth/kubernetes/role/{spec.role_name}",
|
|
f"bound_service_account_names={','.join(spec.service_account_names)}",
|
|
f"bound_service_account_namespaces={','.join(spec.service_account_namespaces)}",
|
|
f"policies={spec.policy_name}",
|
|
f"ttl={spec.token_ttl}",
|
|
],
|
|
)
|
|
objects = {
|
|
"approved_spec_sha256": plan.approved_spec_sha256,
|
|
"policy_name": spec.policy_name,
|
|
"kubernetes_role_name": spec.role_name,
|
|
"service_account_names": ",".join(spec.service_account_names),
|
|
"service_account_namespaces": ",".join(spec.service_account_namespaces),
|
|
}
|
|
record_build(
|
|
plan_id=plan.id,
|
|
plan_path=str(plan.path),
|
|
approved_by=plan.approved_by or "",
|
|
approved_at=plan.approved_at or "",
|
|
objects=objects,
|
|
log_path=spec.audit_log_path,
|
|
)
|
|
return objects
|