ops-warden/src/warden/config.py

"""Config loading for OpsWarden."""
from __future__ import annotations

import os
import shlex
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional

import yaml


class ConfigError(Exception):
    """Raised when config is invalid or missing."""


@dataclass
class CallerAuthConfig:
    """How ops-warden proves *its own* identity to flex-auth (FLEX-WP-0016).

    flex-auth's ops-warden pin authenticates the caller with a Kubernetes
    TokenReview and binds ``resource.system: ops-warden`` to the principal
    ``system:serviceaccount:ops-warden:ops-warden``. A workstation ``warden
    sign`` is not a ServiceAccount, so the token has to come from somewhere:

    ``none``     send no ``Authorization`` header (pre-FLEX-WP-0016 behaviour;
                 accepted only while that pin runs ``callerAuth.mode: warn``)
    ``file``     read a projected ServiceAccount token from ``token_path``
                 (in-cluster PEP, audience-bound by the projection)
    ``env``      read the token from ``token_env``
    ``command``  run ``command`` and use its stdout, e.g.
                 ``kubectl create token ops-warden -n ops-warden
                 --audience flex-auth --duration 10m``

    ops-warden never stores the token: it is read, sent, and dropped
    (ADR-0002 — transparent conduit, not a broker).
    """

    mode: str = "none"
    token_path: Optional[Path] = None
    token_env: str = "WARDEN_POLICY_CALLER_TOKEN"
    command: Optional[List[str]] = None
    audience: str = "flex-auth"


@dataclass
class PolicyConfig:
    # SSH signing pin (flex-auth-ops-warden). Admits only resource.system ops-warden.
    flex_auth_url: Optional[str] = None
    # Secret-read pin (flex-auth-ops-warden-fetch, CUST-WP-0075). Used only by
    # check_fetch_policy; there is no fallback to flex_auth_url.
    fetch_flex_auth_url: Optional[str] = None
    zone_registry_path: Optional[Path] = None
    failure_modes: Dict[str, str] = field(
        default_factory=lambda: {
            "z0-experimental": "fail_open",
            "z1-operational": "fail_open",
            "z2-protected": "fail_open",
            "z2-continuity": "fail_open",
            "z3-critical": "fail_closed",
            "unknown": "fail_open",
            "not-applicable": "fail_closed",
        }
    )
    tenant: str = "tenant:platform"
    subject_env: str = "WARDEN_POLICY_SUBJECT"
    system: str = "ops-warden"
    caller_auth: "CallerAuthConfig" = field(default_factory=lambda: CallerAuthConfig())


@dataclass
class VaultConfig:
    addr: str
    role_map: Dict[str, str]  # ActorType.value -> vault role name
    token_env: str = "VAULT_TOKEN"  # env var holding the Vault token
    mount: str = "ssh"  # Vault secrets engine mount path


@dataclass
class WardenConfig:
    backend: str  # "local" or "vault"
    ca_key: Optional[Path] = None  # required for local backend
    vault: Optional[VaultConfig] = None  # required for vault backend
    inventory_path: Path = field(
        default_factory=lambda: Path.home() / ".config" / "warden" / "inventory.yaml"
    )
    state_dir: Path = field(
        default_factory=lambda: Path.home() / ".local" / "state" / "warden"
    )
    policy: PolicyConfig = field(default_factory=PolicyConfig)


def _default_config_path() -> Path:
    return Path.home() / ".config" / "warden" / "warden.yaml"


def load_config(path: Optional[Path] = None) -> WardenConfig:
    """Load and validate warden.yaml. Respects WARDEN_CONFIG env var."""
    config_path = path or Path(
        os.environ.get("WARDEN_CONFIG", str(_default_config_path()))
    )
    if not config_path.exists():
        raise ConfigError(f"Config not found: {config_path}")

    try:
        with config_path.open() as f:
            raw = yaml.safe_load(f)
    except yaml.YAMLError as e:
        raise ConfigError(f"Invalid YAML in {config_path}: {e}") from e

    if not isinstance(raw, dict):
        raise ConfigError("Config must be a YAML mapping")

    backend = str(raw.get("backend", "local"))
    if backend not in ("local", "vault"):
        raise ConfigError(
            f"backend must be 'local' or 'vault', got: {backend!r}"
        )

    ca_key = None
    if "ca_key" in raw and raw["ca_key"]:
        ca_key = Path(os.path.expanduser(str(raw["ca_key"])))

    vault_cfg = None
    if backend == "vault":
        v = raw.get("vault") or {}
        if "addr" not in v:
            raise ConfigError("vault backend requires vault.addr")
        role_map = v.get("role_map") or {
            "adm": "adm-role",
            "agt": "agt-role",
            "atm": "atm-role",
        }
        vault_cfg = VaultConfig(
            addr=str(v["addr"]),
            role_map=dict(role_map),
            token_env=str(v.get("token_env", "VAULT_TOKEN")),
            mount=str(v.get("mount", "ssh")),
        )
    elif backend == "local" and ca_key is None:
        raise ConfigError("local backend requires ca_key")

    inventory_path = Path(
        os.path.expanduser(
            str(
                raw.get(
                    "inventory_path",
                    str(Path.home() / ".config" / "warden" / "inventory.yaml"),
                )
            )
        )
    )
    state_dir = Path(
        os.path.expanduser(
            str(
                raw.get(
                    "state_dir",
                    str(Path.home() / ".local" / "state" / "warden"),
                )
            )
        )
    )

    policy_raw = raw.get("policy") or {}
    retired = sorted({"enabled", "fail_closed"}.intersection(policy_raw))
    if retired:
        raise ConfigError(
            "retired policy setting(s) "
            + ", ".join(f"policy.{key}" for key in retired)
            + "; security-zones_v0.1 now selects stance and failure mode"
        )
    caller_raw = policy_raw.get("caller_auth") or {}
    caller_command = caller_raw.get("command")
    if isinstance(caller_command, str):
        caller_command = shlex.split(caller_command)
    elif caller_command is not None:
        caller_command = [str(part) for part in caller_command]
    caller_token_path = caller_raw.get("token_path")
    caller_cfg = CallerAuthConfig(
        mode=str(caller_raw.get("mode", "none")).strip().lower(),
        token_path=(
            Path(os.path.expanduser(str(caller_token_path)))
            if caller_token_path
            else None
        ),
        token_env=str(caller_raw.get("token_env", "WARDEN_POLICY_CALLER_TOKEN")),
        command=caller_command,
        audience=str(caller_raw.get("audience", "flex-auth")),
    )
    if caller_cfg.mode not in {"none", "file", "env", "command"}:
        raise ConfigError(
            f"policy.caller_auth.mode must be none|file|env|command, "
            f"got {caller_cfg.mode!r}"
        )
    if caller_cfg.mode == "file" and caller_cfg.token_path is None:
        raise ConfigError("policy.caller_auth.token_path is required for mode: file")
    if caller_cfg.mode == "command" and not caller_cfg.command:
        raise ConfigError("policy.caller_auth.command is required for mode: command")
    failure_modes = PolicyConfig().failure_modes
    configured_failure_modes = policy_raw.get("failure_modes") or {}
    if not isinstance(configured_failure_modes, dict):
        raise ConfigError("policy.failure_modes must be a mapping")
    failure_modes.update(
        {str(zone): str(mode) for zone, mode in configured_failure_modes.items()}
    )
    invalid_modes = {
        zone: mode
        for zone, mode in failure_modes.items()
        if mode not in {"fail_open", "fail_closed"}
    }
    if invalid_modes:
        raise ConfigError(
            "policy.failure_modes values must be fail_open or fail_closed: "
            f"{invalid_modes}"
        )
    zone_registry_path = policy_raw.get("zone_registry_path")
    flex_auth_url = str(policy_raw.get("flex_auth_url", "")).strip() or None
    fetch_flex_auth_url = (
        str(policy_raw.get("fetch_flex_auth_url", "") or "").strip() or None
    )
    policy_cfg = PolicyConfig(
        flex_auth_url=flex_auth_url,
        fetch_flex_auth_url=fetch_flex_auth_url,
        zone_registry_path=(
            Path(os.path.expanduser(str(zone_registry_path)))
            if zone_registry_path
            else None
        ),
        failure_modes=failure_modes,
        tenant=str(policy_raw.get("tenant", "tenant:platform")),
        subject_env=str(policy_raw.get("subject_env", "WARDEN_POLICY_SUBJECT")),
        system=str(policy_raw.get("system", "ops-warden")),
        caller_auth=caller_cfg,
    )

    return WardenConfig(
        backend=backend,
        ca_key=ca_key,
        vault=vault_cfg,
        inventory_path=inventory_path,
        state_dir=state_dir,
        policy=policy_cfg,
    )