secrets-engine/src/secrets_engine/lifecycle.py

"""Non-destructive lifecycle plans for native secrets-engine access.

The legacy KV revoke path deleted all KV metadata while leaving delivery auth
behind.  That is not revocation and is not reversible.  Ordinary revoke now
means *deactivate native access*: remove only policy/AppRole objects managed by
secrets-engine and preserve OpenBao custody plus externally managed workload
delivery.
"""
from __future__ import annotations

from dataclasses import dataclass

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

LIFECYCLE_OPERATIONS = ("suspend", "deactivate", "destroy")


@dataclass(frozen=True)
class LifecycleAction:
    kind: str
    target: str
    mutation: bool
    reason: str
    mount: str = ""
    path: str = ""

    def render(self) -> str:
        mode = "mutation" if self.mutation else "no mutation"
        return f"  [{self.kind}] {self.target}  ({mode}; {self.reason})"


@dataclass(frozen=True)
class LifecyclePlan:
    catalog_id: str
    operation: str
    actions: tuple[LifecycleAction, ...]

    def render(self) -> str:
        lines = [
            f"Lifecycle plan for lane '{self.catalog_id}'",
            f"  operation: {self.operation}",
            "  actions:",
        ]
        lines.extend(action.render() for action in self.actions)
        return "\n".join(lines)


@dataclass(frozen=True)
class LifecycleResult:
    applied: tuple[str, ...]
    preserved: tuple[str, ...]

    def render(self) -> str:
        lines = [f"  applied: {target}" for target in self.applied]
        lines.extend(f"  preserved: {target}" for target in self.preserved)
        return "\n".join(lines) or "  (nothing to do)"


def build_lifecycle_plan(entry: CatalogEntry, operation: str) -> LifecyclePlan:
    """Plan an explicit lifecycle operation with no hidden mutations."""
    if operation not in LIFECYCLE_OPERATIONS:
        raise PolicyGuardError(
            f"unknown lifecycle operation '{operation}'; allowed {LIFECYCLE_OPERATIONS}"
        )
    if operation == "destroy" and not entry.stores_kv_value():
        raise PolicyGuardError(
            f"lane '{entry.id}' is {entry.kind}; it has no KV custody to destroy"
        )
    actions: list[LifecycleAction] = []
    if entry.manages_delivery_auth:
        actions.append(
            LifecycleAction(
                "delete-approle",
                entry.role_name,
                True,
                "stop new native delivery logins",
            )
        )
        if operation in {"deactivate", "destroy"}:
            actions.append(
                LifecycleAction(
                    "delete-policy",
                    entry.policy_name,
                    True,
                    "remove native consumer policy",
                )
            )
        else:
            actions.append(
                LifecycleAction(
                    "preserve-policy",
                    entry.policy_name,
                    False,
                    "suspension keeps policy for reviewed rollback",
                )
            )
    elif entry.has_delivery_auth:
        actions.extend(
            (
                LifecycleAction(
                    "preserve-external-approle",
                    entry.role_name,
                    False,
                    "delivery auth is externally managed",
                ),
                LifecycleAction(
                    "preserve-external-policy",
                    entry.policy_name,
                    False,
                    "delivery auth is externally managed",
                ),
            )
        )
    else:
        actions.append(
            LifecycleAction(
                "no-native-delivery-auth",
                entry.id,
                False,
                "lane declares no native delivery auth",
            )
        )

    if entry.stores_kv_value():
        if operation == "destroy":
            actions.append(
                LifecycleAction(
                    "delete-kv-metadata",
                    f"{entry.mount}/{entry.path}",
                    True,
                    "irreversibly delete all KV versions and metadata",
                    mount=entry.mount,
                    path=entry.path,
                )
            )
        else:
            actions.append(
                LifecycleAction(
                    "preserve-kv-custody",
                    f"{entry.mount}/{entry.path}",
                    False,
                    "suspend/deactivate never delete KV metadata or values",
                )
            )
    if entry.workload_delivery:
        actions.append(
            LifecycleAction(
                "preserve-workload-delivery",
                entry.id,
                False,
                "workload delivery is externally managed",
            )
        )
    return LifecyclePlan(
        catalog_id=entry.id,
        operation=operation,
        actions=tuple(actions),
    )


def build_native_deactivation_plan(entry: CatalogEntry) -> LifecyclePlan:
    """Compatibility plan for ordinary revoke: safe native deactivation."""
    return build_lifecycle_plan(entry, "deactivate")


def require_destroy_confirmation(entry: CatalogEntry, confirmation: str) -> None:
    """Require an exact lane id before an irreversible custody operation."""
    if confirmation != entry.id:
        raise PolicyGuardError(
            "destroy requires --confirm-destroy with the exact catalog id; "
            "no custody mutation performed"
        )


def apply_lifecycle_plan(
    client: OpenBaoClient, plan: LifecyclePlan
) -> LifecycleResult:
    """Apply exactly the mutations described by a lifecycle plan."""
    applied: list[str] = []
    preserved: list[str] = []
    for action in plan.actions:
        if not action.mutation:
            preserved.append(action.target)
            continue
        if action.kind == "delete-approle":
            client.delete_approle(action.target)
        elif action.kind == "delete-policy":
            client.delete_policy(action.target)
        elif action.kind == "delete-kv-metadata":
            client.kv_delete_metadata(action.mount, action.path)
        else:  # defensive: a rendered plan must not grow hidden mutations
            raise ValueError(f"unsupported lifecycle mutation '{action.kind}'")
        applied.append(action.target)
    return LifecycleResult(applied=tuple(applied), preserved=tuple(preserved))