secrets-engine/src/secrets_engine/lane_state.py

"""Persistent non-secret lane lifecycle state.

State lives under the evidence directory, never in Git, and never holds a
secret value. It does not recreate OpenBao objects; ``apply`` remains the
metadata path. Delivery commands consult this overlay and fail closed.
"""
from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
import yaml

from secrets_engine.errors import DecisionError, PolicyGuardError
from secrets_engine.redact import looks_secret, redact_text

STATES = ("active", "suspended", "deactivated", "compromised")
DELIVERY_BLOCKED = frozenset({"suspended", "deactivated", "compromised"})
PROVISION_BLOCKED = frozenset({"suspended", "deactivated", "compromised"})


@dataclass(frozen=True)
class LaneState:
    catalog_id: str
    state: str
    updated_at: str = ""
    last_operation: str = ""
    reason: str = ""

    def as_dict(self) -> dict[str, str]:
        payload = {
            "catalog_id": self.catalog_id,
            "state": self.state,
            "updated_at": self.updated_at,
            "last_operation": self.last_operation,
        }
        if self.reason:
            payload["reason"] = self.reason
        return payload


def state_dir(evidence_dir: Path) -> Path:
    return Path(evidence_dir) / "lane-state"


def state_path(evidence_dir: Path, catalog_id: str) -> Path:
    return state_dir(evidence_dir) / f"{catalog_id}.yaml"


def load_lane_state(evidence_dir: Path, catalog_id: str) -> LaneState:
    path = state_path(evidence_dir, catalog_id)
    if not path.is_file():
        return LaneState(catalog_id=catalog_id, state="active")
    try:
        data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    except (OSError, yaml.YAMLError) as exc:
        raise PolicyGuardError(f"unable to load lane state for '{catalog_id}'") from exc
    if not isinstance(data, dict):
        raise PolicyGuardError(f"lane state for '{catalog_id}' is invalid")
    state = str(data.get("state") or "active")
    if state not in STATES:
        raise PolicyGuardError(f"lane '{catalog_id}' has unknown state '{state}'")
    return LaneState(
        catalog_id=str(data.get("catalog_id") or catalog_id),
        state=state,
        updated_at=str(data.get("updated_at") or ""),
        last_operation=str(data.get("last_operation") or ""),
        reason=str(data.get("reason") or ""),
    )


def save_lane_state(
    evidence_dir: Path,
    catalog_id: str,
    state: str,
    *,
    operation: str,
    reason: str = "",
    now: datetime | None = None,
) -> LaneState:
    if state not in STATES:
        raise PolicyGuardError(f"unknown lane state '{state}'")
    cleaned = _clean_reason(reason)
    record = LaneState(
        catalog_id=catalog_id,
        state=state,
        updated_at=(now or datetime.now(timezone.utc)).astimezone(timezone.utc).isoformat(),
        last_operation=operation,
        reason=cleaned,
    )
    path = state_path(evidence_dir, catalog_id)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(yaml.safe_dump(record.as_dict(), sort_keys=True), encoding="utf-8")
    return record


def _clean_reason(reason: str) -> str:
    text = (reason or "").strip()
    if not text:
        return ""
    if len(text) > 200:
        raise PolicyGuardError("lane-state reason must be at most 200 characters")
    if looks_secret(text) or redact_text(text) != text:
        raise PolicyGuardError("lane-state reason must not contain secret-like material")
    return text


def require_delivery_state(evidence_dir: Path, catalog_id: str, action: str) -> LaneState:
    """Refuse exec/wrap/handoff when the lane is not active."""
    current = load_lane_state(evidence_dir, catalog_id)
    if current.state in DELIVERY_BLOCKED:
        raise DecisionError(
            f"lane '{catalog_id}' is {current.state}; "
            f"refusing {action} until lifecycle reactivate"
        )
    return current


def require_provision_state(evidence_dir: Path, catalog_id: str) -> LaneState:
    current = load_lane_state(evidence_dir, catalog_id)
    if current.state in PROVISION_BLOCKED:
        hint = "rotate" if current.state == "compromised" else "lifecycle reactivate"
        raise DecisionError(
            f"lane '{catalog_id}' is {current.state}; refusing provision; use {hint}"
        )
    return current


def operation_state(operation: str) -> str | None:
    """Return the state persisted after a successful lifecycle operation."""
    return {
        "suspend": "suspended",
        "deactivate": "deactivated",
        "compromise": "compromised",
        "reactivate": "active",
        "revoke": "deactivated",
    }.get(operation)