Plan Qonto Kubernetes credential lane

src/ops_mason/executor.py

@@ -55,6 +55,17 @@ class AppRoleKVSpec:
     audit_log_path: Path | None = None
 
 
+@dataclass
+class KubernetesKVSpec:
+    policy_name: str
+    role_name: str
+    service_account_names: tuple[str, ...]
+    service_account_namespaces: tuple[str, ...]
+    token_ttl: str = "15m"
+    audit_log_path: Path | None = None
+    bao_bin: str = "bao"
+
+
 def _run(bao_bin: str, args: list[str], input_text: str | None = None) -> str:
     proc = subprocess.run(
         [bao_bin, *args],
@@ -151,3 +162,44 @@ def build_approle_kv_lane(plan: ConstructionPlan, spec: AppRoleKVSpec) -> dict[s
         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."""
+    if not plan.is_approved():
+        raise BuildRefused(
+            f"plan {plan.id!r} is not approved "
+            f"(status={plan.status!r}, approved_by={plan.approved_by!r}, "
+            f"approved_at={plan.approved_at!r}) — refusing to build"
+        )
+    if not spec.service_account_names or not spec.service_account_namespaces:
+        raise BuildRefused("Kubernetes role requires explicit service account bindings")
+
+    _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 = {
+        "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