ops-warden/src/warden/doubles.py

"""Dev-tier contract doubles for routed subsystems (WP-0015 T4).

This generalizes the "fake bao" smoke pattern into a small, hermetic library: it
materializes stand-in executables for the subsystems ops-warden *routes* to (OpenBao,
key-cape login) so that access flows (``warden access --fetch/--exec``, the login lane)
can be exercised fully offline in **dev/test** posture.

Contract, not behavior. Each double honors only the *interface contract* the proxy
relies on (argv shape, stdout, exit code) and emits **synthetic values only** — every
emitted value is prefixed ``synthetic-`` so it can never be mistaken for, or promoted
as, a real secret (Axis-A rule R3: dev touches no real data). These doubles are the
sanctioned ``backend: mock-or-contract-double`` for the ``dev`` env posture.

They are a dev/test convenience, never a runtime component: nothing here vends, stores,
or proxies a real credential.
"""
from __future__ import annotations

import os
import stat
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, List

# Marker every synthetic value carries — asserted in tests, greppable in logs.
SYNTHETIC_PREFIX = "synthetic-"


@dataclass(frozen=True)
class Double:
    """A single contract double: the command name and the script that backs it."""

    name: str          # the executable name on PATH (e.g. "bao")
    contract: str      # one-line description of the contract it honors
    script: str        # the script body (shebang included)


def _bao_script() -> str:
    # Honors: `bao kv get -field=<F> <path>` -> synthetic value on stdout, exit 0.
    #         `bao login ...`               -> token line on stdout, exit 0.
    # Any other subcommand exits 2 so contract drift surfaces loudly.
    return r"""#!/usr/bin/env bash
# Contract double for OpenBao (synthetic values only — WP-0015 T4).
set -euo pipefail
SUFFIX="${WARDEN_DOUBLE_SUFFIX:-bao}"
case "${1:-}" in
  kv)
    if [[ "${2:-}" == "get" ]]; then
      field="generic"
      for a in "$@"; do
        case "$a" in -field=*) field="${a#-field=}";; esac
      done
      echo "synthetic-${field}-${SUFFIX}"
      exit 0
    fi
    ;;
  login)
    echo "synthetic-token-${SUFFIX}"
    exit 0
    ;;
esac
echo "fake-bao: unsupported contract: $*" >&2
exit 2
"""


def _keycape_script() -> str:
    # Honors: `key-cape login ...` -> interactive-shaped success line, exit 0.
    return r"""#!/usr/bin/env bash
# Contract double for key-cape OIDC login (synthetic — WP-0015 T4).
set -euo pipefail
SUFFIX="${WARDEN_DOUBLE_SUFFIX:-keycape}"
case "${1:-}" in
  login)
    echo "synthetic-oidc-session-${SUFFIX}"
    exit 0
    ;;
esac
echo "fake-key-cape: unsupported contract: $*" >&2
exit 2
"""


# The registry of available doubles, keyed by subsystem command name.
_DOUBLES: Dict[str, Double] = {
    "bao": Double(
        name="bao",
        contract="bao kv get -field=<F> <path> | bao login",
        script=_bao_script(),
    ),
    "key-cape": Double(
        name="key-cape",
        contract="key-cape login <args>",
        script=_keycape_script(),
    ),
}


def available_doubles() -> List[str]:
    """Names of the subsystems a double can be materialized for."""
    return sorted(_DOUBLES)


def materialize_doubles(dest_dir: Path, names: List[str] | None = None) -> Dict[str, Path]:
    """Write the requested contract doubles into ``dest_dir`` as executables.

    Returns a mapping of subsystem name -> path. ``names=None`` materializes all.
    Prepend ``dest_dir`` to ``PATH`` to run an access flow fully offline against them.
    """
    dest_dir = Path(dest_dir)
    dest_dir.mkdir(parents=True, exist_ok=True)
    selected = names if names is not None else list(_DOUBLES)
    out: Dict[str, Path] = {}
    for name in selected:
        double = _DOUBLES.get(name)
        if double is None:
            raise KeyError(
                f"no contract double for {name!r}; available: {available_doubles()}"
            )
        target = dest_dir / double.name
        target.write_text(double.script)
        target.chmod(target.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
        out[name] = target
    return out


def doubles_path_prepended(dest_dir: Path, base_path: str | None = None) -> str:
    """Return a PATH string with ``dest_dir`` ahead of the current PATH.

    Convenience for spawning a subprocess that should resolve the doubles first.
    """
    base = base_path if base_path is not None else os.environ.get("PATH", "")
    return os.pathsep.join([str(Path(dest_dir)), base]) if base else str(Path(dest_dir))