secrets-engine/src/secrets_engine/roles.py

"""Stage role/policy generation and the safety guards that keep them narrow.

Three stage roles exist: secrets-engine-build, -test, -prod. Each is confined to
its own KV prefix and a small, explicit capability set. The guards in this module
are the heart of the product promise: a generated plan that would grant broad
power (root, sudo, sys/, auth/ admin, wildcard mounts, cross-stage paths) is
rejected before it can ever reach OpenBao.
"""
from __future__ import annotations

from dataclasses import dataclass

from secrets_engine.catalog import CatalogEntry
from secrets_engine.errors import PolicyGuardError

STAGES = ("build", "test", "prod")

# Per-stage KV path prefix each role is allowed to touch. A lane whose path does
# not sit under its stage prefix is out of bounds.
STAGE_PREFIX = {
    "build": "build/",
    "test": "test/",
    "prod": "",  # prod lanes use their own owner-scoped paths (no shared prefix)
}

# Capabilities a stage role's *own* policy may carry. Anything else is broad.
ALLOWED_CAPABILITIES = {"create", "read", "update", "delete", "list"}

# Substrings that, if they appear in a policy path, mean the plan is too broad.
FORBIDDEN_PATH_MARKERS = (
    "sys/",
    "auth/token/",
    "identity/",
    "sudo",
    "+/",   # single-level wildcard
)

# Capability names that confer admin/root and must never appear in a stage policy.
FORBIDDEN_CAPABILITIES = {"sudo", "root", "deny-all-bypass"}

# Policy/role names that smell like broad admin and are refused outright.
FORBIDDEN_NAME_MARKERS = ("root", "admin", "superuser", "platform-admin", "sys")


@dataclass(frozen=True)
class StageRole:
    stage: str
    policy_name: str
    role_name: str
    prefix: str

    @classmethod
    def for_stage(cls, stage: str) -> "StageRole":
        if stage not in STAGES:
            raise PolicyGuardError(f"unknown stage '{stage}'; allowed {STAGES}")
        return cls(
            stage=stage,
            policy_name=f"secrets-engine-{stage}",
            role_name=f"secrets-engine-{stage}",
            prefix=STAGE_PREFIX[stage],
        )


def assert_path_in_stage(entry: CatalogEntry) -> None:
    """Reject a lane whose path is wildcarded or outside its stage prefix."""
    if "*" in entry.path or "+" in entry.path:
        raise PolicyGuardError(
            f"lane '{entry.id}': wildcard path '{entry.path}' is not allowed"
        )
    prefix = STAGE_PREFIX[entry.stage]
    if entry.stage in ("build", "test") and not entry.path.startswith(prefix):
        raise PolicyGuardError(
            f"lane '{entry.id}': {entry.stage} path must start with '{prefix}' "
            f"(got '{entry.path}')"
        )
    if entry.stage in ("build", "test"):
        # A build/test lane must not reach into another stage's prefix.
        for other, oprefix in STAGE_PREFIX.items():
            if other != entry.stage and oprefix and entry.path.startswith(oprefix):
                raise PolicyGuardError(
                    f"lane '{entry.id}': {entry.stage} lane reaches into "
                    f"'{oprefix}' ({other} territory)"
                )


def assert_policy_safe(policy_name: str, paths: dict[str, list[str]]) -> None:
    """Reject a policy document that is too broad to be a stage policy."""
    lowered = policy_name.lower()
    for marker in FORBIDDEN_NAME_MARKERS:
        if marker in lowered:
            raise PolicyGuardError(
                f"policy name '{policy_name}' resembles broad admin (marker '{marker}')"
            )
    for path, caps in paths.items():
        for marker in FORBIDDEN_PATH_MARKERS:
            if marker in path:
                raise PolicyGuardError(
                    f"policy '{policy_name}': path '{path}' is out of bounds "
                    f"(marker '{marker}')"
                )
        if path.strip() in ("*", "/", "secret/*", "+"):
            raise PolicyGuardError(
                f"policy '{policy_name}': wildcard path '{path}' not allowed"
            )
        bad_caps = set(caps) - ALLOWED_CAPABILITIES
        if bad_caps & FORBIDDEN_CAPABILITIES or bad_caps:
            raise PolicyGuardError(
                f"policy '{policy_name}': capabilities {sorted(bad_caps)} not allowed "
                f"(allowed {sorted(ALLOWED_CAPABILITIES)})"
            )


def lane_policy_paths(entry: CatalogEntry) -> dict[str, list[str]]:
    """The minimal KV v2 paths + capabilities a consumer policy needs for a lane."""
    data_path = f"{entry.mount}/data/{entry.path}"
    meta_path = f"{entry.mount}/metadata/{entry.path}"
    return {
        data_path: ["read"],
        meta_path: ["read"],
    }


def render_policy_hcl(policy_name: str, paths: dict[str, list[str]]) -> str:
    """Render an OpenBao ACL policy in HCL. Validates safety first."""
    assert_policy_safe(policy_name, paths)
    blocks = [f'# Generated by secrets-engine for policy "{policy_name}"']
    for path, caps in paths.items():
        cap_list = ", ".join(f'"{c}"' for c in caps)
        blocks.append(f'path "{path}" {{\n  capabilities = [{cap_list}]\n}}')
    return "\n\n".join(blocks) + "\n"


def consumer_policy_for(entry: CatalogEntry) -> tuple[str, str]:
    """Return (policy_name, hcl) for the lane's approved consumer."""
    assert_path_in_stage(entry)
    paths = lane_policy_paths(entry)
    name = entry.policy_name
    return name, render_policy_hcl(name, paths)