secrets-engine/src/secrets_engine/plan.py

"""Planning: turn an approved request into a concrete, guarded set of actions.

A Plan is a human-reviewable list of OpenBao actions (policy write, approle
write, KV mount). Building a plan runs every safety guard, so a plan that exists
is, by construction, in-bounds. Apply just executes a built plan.
"""
from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any

from secrets_engine.catalog import CatalogEntry
from secrets_engine.errors import PolicyGuardError
from secrets_engine.roles import (
    StageRole,
    assert_path_in_stage,
    consumer_policy_for,
)


@dataclass
class PlanAction:
    kind: str           # "kv-mount" | "policy" | "approle"
    target: str         # human-readable target
    detail: dict[str, Any] = field(default_factory=dict)

    def render(self) -> str:
        d = ", ".join(f"{k}={v}" for k, v in self.detail.items() if k != "hcl")
        return f"  [{self.kind}] {self.target}" + (f"  ({d})" if d else "")


@dataclass
class Plan:
    catalog_id: str
    stage: str
    decision_id: str
    policy_name: str
    role_name: str
    actions: list[PlanAction]
    policy_hcl: str

    def render(self) -> str:
        lines = [
            f"Plan for lane '{self.catalog_id}' (stage={self.stage})",
            f"  decision: {self.decision_id or '<none>'}",
            f"  stage role: secrets-engine-{self.stage}",
            f"  consumer policy: {self.policy_name}",
            f"  consumer approle: {self.role_name}",
            "  actions:",
        ]
        lines.extend(a.render() for a in self.actions)
        lines.append("")
        lines.append("  generated consumer policy (HCL):")
        lines.extend("    " + ln for ln in self.policy_hcl.splitlines())
        return "\n".join(lines)


def build_plan(entry: CatalogEntry, stage: str, *, decision_id: str = "") -> Plan:
    """Construct and fully guard a plan. Raises PolicyGuardError on any violation."""
    if stage != entry.stage:
        raise PolicyGuardError(
            f"stage mismatch: lane '{entry.id}' is stage '{entry.stage}', "
            f"refusing to apply as '{stage}'"
        )
    StageRole.for_stage(stage)              # validates stage name
    assert_path_in_stage(entry)             # path must be in-stage, no wildcards
    policy_name, policy_hcl = consumer_policy_for(entry)  # runs assert_policy_safe

    actions = [
        PlanAction("kv-mount", entry.mount, {"type": "kv-v2"}),
        PlanAction(
            "policy",
            policy_name,
            {"paths": f"{entry.mount}/data/{entry.path}"},
        ),
        PlanAction(
            "approle",
            entry.role_name,
            {"token_policies": policy_name, "auth": "approle"},
        ),
    ]
    return Plan(
        catalog_id=entry.id,
        stage=stage,
        decision_id=decision_id,
        policy_name=policy_name,
        role_name=entry.role_name,
        actions=actions,
        policy_hcl=policy_hcl,
    )