ops-warden/src/warden/caller_identity.py

"""Caller identity for ops-warden's outbound flex-auth policy calls.

flex-auth's `flex-auth-ops-warden` pin (FLEX-WP-0016) authenticates the *caller*
before it evaluates the request: `Authorization: Bearer <token>` is passed to a
Kubernetes TokenReview, and `resource.system: ops-warden` is bound to the
principal `system:serviceaccount:ops-warden:ops-warden`. Until ops-warden sends
that header, the pin logs `caller authentication warning` and can only run in
`warn` mode — which is why `policy.enabled` cannot flip.

This module resolves the token at call time and hands it straight to the request.
Nothing is cached to disk, logged, or echoed: ops-warden carries the value, it
does not hold it (ADR-0002).
"""
from __future__ import annotations

import os
import subprocess

from warden.config import CallerAuthConfig


class CallerIdentityError(Exception):
    """Raised when a caller token was configured but could not be obtained."""


def resolve_caller_token(cfg: CallerAuthConfig) -> str | None:
    """Return the bearer token for flex-auth, or None when mode is ``none``.

    Raises CallerIdentityError when a token was configured but is unavailable.
    The token itself never appears in an exception message.
    """
    mode = cfg.mode
    if mode == "none":
        return None

    if mode == "file":
        if cfg.token_path is None:
            raise CallerIdentityError("caller_auth mode 'file' has no token_path")
        try:
            token = cfg.token_path.read_text()
        except OSError as e:
            raise CallerIdentityError(
                f"caller token file unreadable: {cfg.token_path} ({e.strerror})"
            ) from e
    elif mode == "env":
        token = os.environ.get(cfg.token_env, "")
        if not token.strip():
            raise CallerIdentityError(
                f"caller token env {cfg.token_env} is unset or empty"
            )
    elif mode == "command":
        if not cfg.command:
            raise CallerIdentityError("caller_auth mode 'command' has no command")
        try:
            result = subprocess.run(
                cfg.command,
                capture_output=True,
                text=True,
                timeout=30,
                check=False,
            )
        except FileNotFoundError as e:
            raise CallerIdentityError(
                f"caller token command not found: {cfg.command[0]}"
            ) from e
        except subprocess.TimeoutExpired as e:
            raise CallerIdentityError("caller token command timed out") from e
        if result.returncode != 0:
            stderr = (result.stderr or "").strip().splitlines()
            detail = stderr[-1] if stderr else f"exit {result.returncode}"
            raise CallerIdentityError(f"caller token command failed: {detail}")
        token = result.stdout
    else:
        raise CallerIdentityError(f"unsupported caller_auth mode {mode!r}")

    token = token.strip()
    if not token:
        raise CallerIdentityError(f"caller_auth mode {mode!r} produced an empty token")
    if any(ch.isspace() for ch in token):
        # flex-auth rejects a bearer token containing whitespace outright.
        raise CallerIdentityError(
            f"caller_auth mode {mode!r} produced a token containing whitespace"
        )
    return token


def caller_auth_headers(cfg: CallerAuthConfig) -> dict[str, str]:
    """Headers to attach to a flex-auth /v1/check call ({} when unauthenticated)."""
    token = resolve_caller_token(cfg)
    if token is None:
        return {}
    return {"Authorization": f"Bearer {token}"}