ops-warden/src/warden/taint.py

"""EXPOSED taint convention for OpenBao KV secrets (WARDEN-WP-0026 T05).

Convention (KV v2 ``custom_metadata`` on the secret, never on secret *data*):

* ``exposed_at`` — ISO-8601 UTC datetime when disclosure was recognized
* ``exposed_version`` — KV version that was (or may have been) disclosed
* ``exposed_reason`` — short machine-safe reason slug (optional)
* ``exposed_ref`` — pointer to a lessons note / CCR / incident doc (optional)

A lane is **tainted** when ``exposed_at`` is set and non-empty. Clearing taint
(after rotation) is an operator action: remove those keys from custom_metadata.
ops-warden only *reports* taint — it never auto-rotates (Strand B / WP-0027).

This module only shells out to ``bao kv metadata get`` (or equivalent). It never
reads secret data values.
"""
from __future__ import annotations

import json
import os
import subprocess
from dataclasses import dataclass
from typing import Any, Optional

from warden.routing.models import RouteEntry

# Canonical custom_metadata keys (WP-0026 T05).
EXPOSED_AT = "exposed_at"
EXPOSED_VERSION = "exposed_version"
EXPOSED_REASON = "exposed_reason"
EXPOSED_REF = "exposed_ref"

_TAINT_KEYS = (EXPOSED_AT, EXPOSED_VERSION, EXPOSED_REASON, EXPOSED_REF)


@dataclass(frozen=True)
class TaintStatus:
    """Advisory taint view for a lane — no secret values."""

    lane_id: str
    path: str
    tainted: bool
    exposed_at: Optional[str] = None
    exposed_version: Optional[str] = None
    exposed_reason: Optional[str] = None
    exposed_ref: Optional[str] = None
    current_version: Optional[int] = None
    error: Optional[str] = None

    def to_dict(self) -> dict[str, Any]:
        return {
            "id": self.lane_id,
            "path": self.path,
            "tainted": self.tainted,
            "exposed_at": self.exposed_at,
            "exposed_version": self.exposed_version,
            "exposed_reason": self.exposed_reason,
            "exposed_ref": self.exposed_ref,
            "current_version": self.current_version,
            **({"error": self.error} if self.error else {}),
        }


class TaintError(Exception):
    """Raised when taint status cannot be determined (auth, path, tool)."""


def kv_metadata_path(path_template: str) -> str:
    """Return the logical KV path suitable for ``bao kv metadata get``.

    Catalog paths are logical (``platform/workloads/...``), not API data paths.
    """
    return path_template.strip().strip("/")


def parse_custom_metadata(meta: dict[str, Any]) -> TaintStatus:
    """Build a TaintStatus from a ``bao kv metadata get -format=json`` data blob.

    ``meta`` is the ``data`` object (with ``custom_metadata``, ``current_version``).
    """
    custom = meta.get("custom_metadata") or {}
    if not isinstance(custom, dict):
        custom = {}
    exposed_at = (custom.get(EXPOSED_AT) or "").strip() or None
    return TaintStatus(
        lane_id="",
        path="",
        tainted=bool(exposed_at),
        exposed_at=exposed_at,
        exposed_version=(custom.get(EXPOSED_VERSION) or "").strip() or None,
        exposed_reason=(custom.get(EXPOSED_REASON) or "").strip() or None,
        exposed_ref=(custom.get(EXPOSED_REF) or "").strip() or None,
        current_version=meta.get("current_version"),
    )


def fetch_taint_status(entry: RouteEntry, *, bao_bin: str = "bao") -> TaintStatus:
    """Query OpenBao metadata for a catalog entry (never reads secret data).

    Uses the caller's ``BAO_TOKEN`` / ``VAULT_TOKEN`` / ``~/.vault-token`` — same
    G1 rule as the access proxy. Requires ``path_template`` on the entry.
    """
    if not entry.path_template or "<" in entry.path_template:
        raise TaintError(
            f"{entry.id!r} has no concrete path_template — cannot query taint metadata."
        )
    path = kv_metadata_path(entry.path_template)
    try:
        proc = subprocess.run(
            [bao_bin, "kv", "metadata", "get", "-format=json", path],
            capture_output=True,
            text=True,
            env=os.environ.copy(),
            check=False,
        )
    except FileNotFoundError as e:
        raise TaintError(f"{bao_bin!r} not found on PATH") from e

    if proc.returncode != 0:
        err = (proc.stderr or proc.stdout or "metadata get failed").strip().splitlines()
        # Never echo tokens if somehow present.
        safe = " ".join(err[:3])[:300]
        return TaintStatus(
            lane_id=entry.id,
            path=path,
            tainted=False,
            error=safe or f"bao exit {proc.returncode}",
        )

    try:
        payload = json.loads(proc.stdout)
    except json.JSONDecodeError as e:
        raise TaintError(f"invalid JSON from bao metadata get: {e}") from e

    data = payload.get("data") if isinstance(payload, dict) else None
    if not isinstance(data, dict):
        raise TaintError("bao metadata response missing data object")

    status = parse_custom_metadata(data)
    return TaintStatus(
        lane_id=entry.id,
        path=path,
        tainted=status.tainted,
        exposed_at=status.exposed_at,
        exposed_version=status.exposed_version,
        exposed_reason=status.exposed_reason,
        exposed_ref=status.exposed_ref,
        current_version=status.current_version,
    )