secrets-engine/src/secrets_engine/pep_stance.py

"""Published PEP unreachable-engine stance (security-layer-model v0.7 §6.4).

Runtime reads ``pep-stance.yaml``. ``SHIPPED_STANCE`` is the pin that makes
drift between the published map and this module fail the conformance test.
"""
from __future__ import annotations

import os
from dataclasses import dataclass, replace
from pathlib import Path
from typing import Any
from urllib.parse import urlparse

import yaml

from secrets_engine.config import repo_root
from secrets_engine.errors import DecisionError

SHIPPED_STANCE = {
    "build": "fail_open",
    "test": "fail_open",
    "prod": "fail_closed",
    "unknown": "fail_closed",
}
REQUIRED_STAGES = tuple(SHIPPED_STANCE)
VALID_MODES = frozenset({"fail_open", "fail_closed"})
LOOPBACK_HOSTS = frozenset({"127.0.0.1", "localhost", "::1"})


@dataclass(frozen=True)
class StanceApplication:
    """Named residue applied when access-engine is unreachable."""

    stage: str
    failure_mode: str
    action: str
    demo_exception: bool = False
    decision_id: str = ""

    def as_evidence(self) -> dict[str, object]:
        payload: dict[str, object] = {
            "stance_stage": self.stage,
            "stance_failure_mode": self.failure_mode,
            "stance_demo_exception": self.demo_exception,
        }
        if self.decision_id:
            payload["stance_decision_id"] = self.decision_id
        return payload


@dataclass(frozen=True)
class PepStanceMap:
    stance: dict[str, str]
    path: Path

    def for_stage(self, stage: str) -> tuple[str, str]:
        key = stage if stage in self.stance else "unknown"
        mode = self.stance.get(key, "")
        if mode not in VALID_MODES:
            raise DecisionError(
                f"pep-stance.yaml has no usable mode for stage {stage!r}"
            )
        return key, mode


def pep_stance_path() -> Path:
    override = os.environ.get("SECRETS_ENGINE_PEP_STANCE", "")
    if override:
        return Path(override)
    return repo_root() / "pep-stance.yaml"


def load_pep_stance(path: Path | None = None) -> PepStanceMap:
    target = path or pep_stance_path()
    try:
        data = yaml.safe_load(target.read_text(encoding="utf-8")) or {}
    except (OSError, yaml.YAMLError) as exc:
        raise DecisionError(f"unable to load PEP stance map {target}: {exc}") from exc
    raw = data.get("stance")
    if not isinstance(raw, dict):
        raise DecisionError(f"{target} is missing a stance map")
    stance = {str(key): str(value) for key, value in raw.items()}
    missing = [stage for stage in REQUIRED_STAGES if stage not in stance]
    if missing:
        raise DecisionError(
            f"{target} is not total; missing stages {missing}"
        )
    unknown_modes = {
        f"{stage}={mode}"
        for stage, mode in stance.items()
        if mode not in VALID_MODES
    }
    if unknown_modes:
        raise DecisionError(f"{target} has invalid modes: {sorted(unknown_modes)}")
    return PepStanceMap(stance=stance, path=target)


def demo_exception_enabled(cfg: Any) -> bool:
    """Three-factor throwaway exception; not a stance row."""
    host = (urlparse(getattr(cfg, "bao_addr", "")).hostname or "").lower()
    return (
        os.environ.get("SECRETS_ENGINE_UNSAFE_DEMO") == "1"
        and not getattr(cfg, "hub_url", "")
        and host in LOOPBACK_HOSTS
    )


def apply_unreachable_engine_stance(
    cfg: Any,
    entry: Any,
    action: str,
    *,
    stance_map: PepStanceMap | None = None,
) -> StanceApplication:
    """Apply the published unreachable-engine residue for a live action.

    ``fail_closed`` without the demo exception raises ``DecisionError`` carrying
    named stance fields. ``fail_open`` is the documented residue: continue to
    the existing lane-approval check, which is itself a gap until T02.
    """
    loaded = stance_map or load_pep_stance()
    stage, mode = loaded.for_stage(getattr(entry, "stage", "unknown"))
    demo = demo_exception_enabled(cfg)
    applied = StanceApplication(
        stage=stage,
        failure_mode=mode,
        action=action or "unknown",
        demo_exception=bool(demo and mode == "fail_closed"),
    )
    if mode == "fail_closed" and not demo:
        raise DecisionError(
            f"production action '{applied.action}' requires a durable "
            "access-engine decision record; live production remains disabled",
            stance=applied.as_evidence(),
        )
    return applied


def with_decision(stance: StanceApplication, decision: Any) -> StanceApplication:
    decision_id = str(getattr(decision, "id", "") or "")
    return replace(stance, decision_id=decision_id)