ops-warden/src/warden/proxy.py

"""Operator access proxy — transparent, audited fetch of a non-SSH credential.

WP-0014 T3. ops-warden does not own these secrets; the proxy lane lets an operator
obtain one *through* the `warden access` front door while keeping the security model
intact. Three guardrails are enforced here in code:

* **G1 — caller identity, never warden's.** The proxy runs the owner's tool with the
  caller's own environment. ops-warden injects no token of its own; if the caller has
  no credential, the underlying tool fails and we surface the auth pointer. We never
  add a `*_TOKEN` warden owns to the child environment.
* **G2 — transit only, no persistence/logging of values.** ``proxy_fetch`` runs the
  tool with **inherited** stdout/stderr (never a pipe), so the value streams to the
  caller and never enters warden's memory. ``proxy_exec`` reads the value solely to
  place it in a child process's environment (the accepted proxy tradeoff) and never
  writes it to disk or log. The audit record is metadata only.
* **G3 — policy gate before fetch.** The CLI runs ``check_fetch_policy`` before
  calling anything here; this module refuses to run an unresolved command template.

This module shells out but never *interprets* secret bytes in the ``--fetch`` path.
"""
from __future__ import annotations

import json
import os
import re
import shlex
import subprocess
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import List, Optional

from warden.routing.models import RouteEntry

_PLACEHOLDER = re.compile(r"<[^>]+>")


@dataclass(frozen=True)
class ResolvedFetch:
    """A catalog fetch command ready to run — either argv or a shell pipeline."""

    argv: Optional[List[str]] = None
    shell_cmd: Optional[str] = None

    def __post_init__(self) -> None:
        if bool(self.argv) == bool(self.shell_cmd):
            raise ValueError("exactly one of argv or shell_cmd must be set")


class ProxyError(Exception):
    """Raised when a proxy fetch cannot be performed safely."""


def _has_shell_pipe(cmd: str) -> bool:
    """True when ``cmd`` contains an unquoted shell pipe operator."""
    in_single = in_double = False
    for ch in cmd:
        if ch == "'" and not in_double:
            in_single = not in_single
        elif ch == '"' and not in_single:
            in_double = not in_double
        elif ch == "|" and not in_single and not in_double:
            return True
    return False


def resolve_fetch_command(
    entry: RouteEntry,
    *,
    domain: Optional[str] = None,
    field: Optional[str] = None,
    path: Optional[str] = None,
) -> ResolvedFetch:
    """Build the concrete fetch command for an entry, or raise if under-specified.

    Starts from the catalog ``fetch_command`` template (with ``<path_template>``
    inlined), substitutes ``<domain>``/``<FIELD>`` and an explicit ``--path`` override,
    then **refuses** if any ``<…>`` placeholder remains. We never run a half-templated
    command — an unresolved placeholder means the operator has not named the owner-side
    resource, and guessing it is exactly the failure mode we avoid.
    """
    if not entry.exec_capable or not entry.fetch_command:
        raise ProxyError(
            f"{entry.id!r} is not exec_capable — it has no proxyable fetch command. "
            "Use `warden access` (advisory) and obtain it from the owner directly."
        )

    cmd = entry.fetch_command
    if entry.path_template and "<path_template>" in cmd:
        cmd = cmd.replace("<path_template>", path or entry.path_template)
    elif path:
        # No <path_template> token but caller supplied a path — append/override is
        # ambiguous, so require the template to carry the token.
        raise ProxyError(
            f"{entry.id!r} fetch_command has no <path_template> token to override with --path."
        )

    if domain:
        cmd = cmd.replace("<domain>", domain)
    if field:
        cmd = cmd.replace("<FIELD>", field)

    leftover = _PLACEHOLDER.findall(cmd)
    if leftover:
        raise ProxyError(
            f"unresolved placeholder(s) {', '.join(sorted(set(leftover)))} in fetch command. "
            "Supply --domain/--field (and --path for owner-side names) — warden will not "
            "guess owner-confirmed resource names."
        )
    if _has_shell_pipe(cmd):
        # Catalog-reviewed pipelines (e.g. kubectl | base64 -d) need a shell.
        return ResolvedFetch(shell_cmd=cmd)
    return ResolvedFetch(argv=shlex.split(cmd))


def caller_auth_present(token_envs: tuple[str, ...] = ("VAULT_TOKEN", "BAO_TOKEN")) -> bool:
    """True if the *caller* appears to hold an auth token (G1 sanity check).

    Best-effort: also accepts a ``~/.vault-token`` file. We do not validate it — the
    owner's tool does that — we only avoid proxying when the caller clearly has no
    credential, so the failure is a clear auth pointer rather than a confusing tool error.
    """
    if any(os.environ.get(e, "").strip() for e in token_envs):
        return True
    return (Path.home() / ".vault-token").exists()


def write_audit(
    state_dir: Path,
    *,
    need_id: str,
    owner_repo: str,
    domain: Optional[str],
    action: str,
    decision_id: Optional[str],
    exit_code: Optional[int] = None,
) -> Path:
    """Append a metadata-only audit record. Never contains a secret value (G2)."""
    state_dir.mkdir(parents=True, exist_ok=True)
    log_path = state_dir / "access-audit.log"
    record = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "action": action,  # "fetch" | "exec"
        "need_id": need_id,
        "owner_repo": owner_repo,
        "domain": domain,
        "subject": os.environ.get("WARDEN_POLICY_SUBJECT", "").strip() or "operator",
        "policy_decision_id": decision_id,
        "exit_code": exit_code,
    }
    with log_path.open("a") as f:
        f.write(json.dumps(record) + "\n")
    try:
        from warden.audit import record_event

        record_event(
            state_dir,
            kind="access",
            action=action,
            subject=record["subject"],
            target=need_id,
            decision_id=decision_id,
            outcome="ok" if exit_code in (None, 0) else "error",
            source="access",
            owner_repo=owner_repo,
            domain=domain,
        )
    except Exception:
        pass
    return log_path


def _caller_env() -> dict:
    """The child environment = the caller's own env. warden adds no credential (G1)."""
    return dict(os.environ)


def proxy_fetch(resolved: ResolvedFetch) -> int:
    """Run the owner's tool, streaming its output straight to the caller.

    stdout/stderr are **inherited** (``None``), never piped — the secret value flows
    subsystem → caller and is never read into warden's memory, buffer, or log (G2).
    Returns the tool's exit code.
    """
    env = _caller_env()
    if resolved.argv is not None:
        completed = subprocess.run(  # noqa: S603 — argv is shlex-split from a validated template
            resolved.argv,
            stdout=None,
            stderr=None,
            stdin=None,
            env=env,
            check=False,
        )
    else:
        completed = subprocess.run(  # noqa: S602 — shell_cmd is catalog-reviewed, not user input
            resolved.shell_cmd,
            shell=True,
            stdout=None,
            stderr=None,
            stdin=None,
            env=env,
            check=False,
        )
    return completed.returncode


def _capture_value(resolved: ResolvedFetch) -> str:
    """Run the fetch and return its stdout (the value) minus one trailing newline.

    The value transits warden's memory (the accepted proxy tradeoff for the
    non-stdout transports) but is never written to disk or log by this function.
    """
    env = _caller_env()
    if resolved.argv is not None:
        fetched = subprocess.run(  # noqa: S603
            resolved.argv, stdout=subprocess.PIPE, stderr=None, stdin=None,
            env=env, check=False, text=True,
        )
    else:
        fetched = subprocess.run(  # noqa: S602
            resolved.shell_cmd, shell=True, stdout=subprocess.PIPE, stderr=None,
            stdin=None, env=env, check=False, text=True,
        )
    if fetched.returncode != 0:
        raise ProxyError(
            f"fetch failed (exit {fetched.returncode}) — check caller auth and the path."
        )
    value = fetched.stdout
    if value.endswith("\n"):
        value = value[:-1]
    return value


def proxy_fetch_to_file(resolved: ResolvedFetch, out_path: Path) -> int:
    """Fetch the value and write it to ``out_path`` at mode 0600 — never to stdout.

    A sanctioned transport (WP-0026 T02): the value goes to a private file the
    caller controls, not a terminal or a logged stream. The file is created with
    O_EXCL semantics widened to truncate-if-owned so a re-fetch overwrites, but the
    mode is forced to 0600 before any bytes are written.
    """
    value = _capture_value(resolved)
    # Open with restrictive mode from the start; do not echo the value anywhere.
    fd = os.open(str(out_path), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    try:
        os.chmod(out_path, 0o600)  # enforce even if the file pre-existed with looser mode
        with os.fdopen(fd, "w") as fh:
            fh.write(value)
    finally:
        value = ""  # noqa: F841 — best-effort scrub of the local reference
    return 0


def is_bao_kv_fetch(entry: RouteEntry) -> bool:
    """True when a lane's fetch is a plain ``bao kv get`` (wrappable, WP-0026 T02)."""
    return bool(entry.fetch_command and entry.fetch_command.strip().startswith("bao kv get"))


def build_wrapped_fetch(
    entry: RouteEntry, *, path: Optional[str] = None, ttl: str = "5m"
) -> ResolvedFetch:
    """Build a response-wrapping fetch: ``bao kv get -wrap-ttl=<ttl> -format=json <path>``.

    Response wrapping returns a single-use, short-TTL *wrapping token* instead of the
    secret value — the sanctioned way to move a value between processes (WP-0026 T02).
    The caller unwraps it in their own context (`bao unwrap`). Only valid for plain
    ``bao kv get`` lanes; the whole secret is wrapped (a per-field ``-field`` read
    cannot be wrapped).
    """
    if not is_bao_kv_fetch(entry):
        raise ProxyError(
            f"{entry.id!r} is not a plain `bao kv get` lane — response wrapping "
            "(--wrap) is unavailable. Use --out FILE or --exec instead."
        )
    target = path or entry.path_template
    if not target or _PLACEHOLDER.search(target):
        raise ProxyError(
            "--wrap needs a concrete path — supply --path or a resolved path_template."
        )
    return ResolvedFetch(argv=["bao", "kv", "get", f"-wrap-ttl={ttl}", "-format=json", target])


def proxy_fetch_wrapped(resolved: ResolvedFetch) -> str:
    """Run a wrapping fetch and return the wrapping *token* (not the secret value).

    The token is single-use and short-lived; it is not itself the credential, so it
    is safe to hand back on stdout. Parses OpenBao's ``-format=json`` wrap_info.
    """
    raw = _capture_value(resolved)
    try:
        data = json.loads(raw)
        token = data["wrap_info"]["token"]
    except (json.JSONDecodeError, KeyError, TypeError) as e:
        raise ProxyError(
            "could not parse a wrapping token from the fetch output "
            "(is response wrapping supported for this path?)."
        ) from e
    if not token:
        raise ProxyError("empty wrapping token returned.")
    return str(token)


def proxy_fetch_fingerprint(resolved: ResolvedFetch):
    """Fetch the value and return a masked fingerprint — never the value (T03).

    Defense-in-depth status view: lets an operator confirm presence/length and
    compare a short non-reversible hash without disclosing the secret. The value
    transits warden's memory only to be hashed, and is scrubbed immediately.
    """
    from warden.mask import fingerprint

    value = _capture_value(resolved)
    try:
        return fingerprint(value)
    finally:
        value = ""  # noqa: F841 — best-effort scrub


def proxy_exec(resolved: ResolvedFetch, *, env_var: str, child_argv: List[str]) -> int:
    """Fetch the value and inject it into a child command's environment only.

    The value transits warden's memory here (the accepted proxy tradeoff for `--exec`)
    but is never written to disk or log and never enters the caller's own shell env.
    Captures the fetch tool's stdout to obtain the value, strips a single trailing
    newline, and runs ``child_argv`` with ``env_var`` set in its environment.
    """
    if not env_var:
        raise ProxyError("--exec requires --field (the env var name to inject), e.g. NPM_AUTH_TOKEN")

    env = _caller_env()
    if resolved.argv is not None:
        fetched = subprocess.run(  # noqa: S603
            resolved.argv,
            stdout=subprocess.PIPE,
            stderr=None,
            stdin=None,
            env=env,
            check=False,
            text=True,
        )
    else:
        fetched = subprocess.run(  # noqa: S602
            resolved.shell_cmd,
            shell=True,
            stdout=subprocess.PIPE,
            stderr=None,
            stdin=None,
            env=env,
            check=False,
            text=True,
        )
    if fetched.returncode != 0:
        raise ProxyError(
            f"fetch failed (exit {fetched.returncode}) — check caller auth and the path."
        )

    value = fetched.stdout
    if value.endswith("\n"):
        value = value[:-1]

    child_env = _caller_env()
    child_env[env_var] = value
    try:
        child = subprocess.run(  # noqa: S603
            child_argv, stdout=None, stderr=None, stdin=None, env=child_env, check=False
        )
        return child.returncode
    finally:
        # Best-effort scrub of the local reference; do not log it.
        value = ""  # noqa: F841
        del child_env[env_var]