execution_backends/cli_executor.py

"""Safe CLI executor backend — real read-only `gh` / `docker`, gated writes.

WHAT THIS IS
------------
This is the fourth and most operationally sensitive execution backend. It stands
in for an operator at a terminal running `gh` (GitHub CLI) and `docker`. In a real
SOC that is exactly the kind of power you want on the tightest possible leash, so
this backend splits its actions sharply by blast radius:

    * READS (low-tier)        — docker_ps, docker_logs,      (REALLY EXECUTE;
                                 gh_run_list, gh_pr_list        they change nothing)
    * STATE-CHANGES (L3)      — docker_stop, docker_restart,  (approval-gated AND
                                 gh_issue_create                simulated; never run)
    * DESTRUCTIVE (always denied) — docker_rm, rm_rf, ...     (refused outright)

WHAT REALLY RUNS, AND WHAT NEVER DOES
-------------------------------------
Read-only inspections **actually execute** as subprocesses and return the tool's
real output — `docker ps` really lists containers, `gh pr list` really queries
GitHub. They are safe to run for real precisely because they only *read*: there is
no state to change and nothing to undo.

State-changing commands are the opposite. They are approval-gated at L3 (see
``policies/agt_policy.yaml``), and even once an operator approves them this backend
**does not shell out** — it returns a simulated/dry-run result. So nothing here can
ever stop a real container or open a real issue. (This backend never
performs a real *write*.)

TWO SAFETY PROPERTIES OF THE REAL EXECUTION PATH
------------------------------------------------
1. **No shell.** Commands are run as an argument vector (``["gh", "pr", "list", ...]``)
   with ``shell=False``. There is no shell to interpret metacharacters, so a value
   can never break out into a second command. This is a deliberate improvement over
   the naive ``shell=True`` pattern.
2. **Validated arguments.** Any value interpolated into a command (e.g. a container
   id) is checked against a conservative pattern *before* execution, and a value
   that looks like a flag or contains shell metacharacters is refused. So even the
   argv path cannot be used to smuggle in an unexpected flag or token.

WHERE IT SITS IN THE TRUST MODEL (same as every backend)
--------------------------------------------------------
    gate decides ─► dispatcher verifies grant ─► THIS backend executes

This backend never decides *whether* an action is allowed — that already happened
upstream. By the time :meth:`run` is called, the gate has said ``ALLOW`` and the
dispatcher has verified the gate's signed grant. It still keeps its own
defense-in-depth guards: an allow-list of handlers, an explicit destructive
deny-list, and the argument validation above.
"""

from __future__ import annotations

import re
import shutil
import subprocess
from typing import Any, Callable

from control_plane.schema import ProposedAction


class CliExecutorError(Exception):
    """Base error for anything this backend refuses or cannot do."""


class CliUnsupportedActionError(CliExecutorError):
    """The requested action is not in this backend's allow-list of safe handlers.

    Defense in depth: anything that isn't one of the allowlisted reads/state-changes
    below lands here and is refused rather than executed.
    """


class CliDestructiveActionError(CliExecutorError):
    """The requested action is a known destructive command and is always denied.

    Refused *before* any execution and with a distinct error type, so a destructive
    verb (e.g. ``docker_rm``, ``rm_rf``) can never be quietly treated as a
    merely-unknown action. No tier or approval unlocks these.
    """


class CliArgumentError(CliExecutorError):
    """A command argument was missing or failed validation (refused before execution).

    Raised for a value that looks like a flag, contains shell metacharacters, or is
    otherwise outside the safe pattern — caught here so it can never reach the real
    subprocess.
    """


class CliToolUnavailableError(CliExecutorError):
    """A read was requested but its CLI tool (``gh`` / ``docker``) is not installed."""


# Known-destructive verbs this backend refuses outright (destructive
# set, plus CLI-specific equivalents). Belt-and-braces: none of these has a handler
# either, so they would be refused regardless — naming them here just produces a
# clearer, security-relevant error. Illustrative, not exhaustive; the real
# guarantee is the allow-list (anything not explicitly handled is denied).
_DESTRUCTIVE_ACTIONS: frozenset[str] = frozenset(
    {
        "docker_rm",            # delete a container
        "docker_kill",          # force-kill (vs. graceful stop)
        "docker_system_prune",  # bulk-delete containers/images/volumes
        "gh_repo_delete",       # delete a repository
        "rm_rf",                # the classic "rm -rf"
        "delete_logs",          # also in the policy's always_denied set
        "remove_audit_records",
        "exfiltrate_data",
        "disable_security_controls",
        "destructive_wildcard",
    }
)

# Safe shape for any value interpolated into a real command: starts with an
# alphanumeric, then alphanumerics and a few benign id characters. This rejects
# leading dashes (so a value can't pose as a flag) and every shell metacharacter.
_SAFE_ARG = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/:-]*$")

# How long a real read may run before we give up on it.
_DEFAULT_TIMEOUT_SECONDS = 15.0


class CliExecutorBackend:
    """Safe `gh`/`docker` adapter satisfying the dispatcher's ``BackendAdapter`` contract.

    Construct it once and let the dispatcher call :meth:`run` per authorized action.
    Read handlers build an argv and execute it for real; state-change handlers return
    a simulated result and never shell out.
    """

    def __init__(self, timeout_seconds: float = _DEFAULT_TIMEOUT_SECONDS) -> None:
        self._timeout = timeout_seconds

        # action_name -> argv builder. Membership here is what makes a read
        # executable, and the names match the policy's cli_executor allow list
        # (policies/agt_policy.yaml). Each builder returns a full argv (list), so
        # the command is always run without a shell.
        self._read_builders: dict[str, Callable[[dict[str, Any]], list[str]]] = {
            "docker_ps": self._build_docker_ps,
            "docker_logs": self._build_docker_logs,
            "gh_run_list": self._build_gh_run_list,
            "gh_pr_list": self._build_gh_pr_list,
        }

        # action_name -> simulated handler. These are the approval-gated
        # state-changes; they never execute, so they produce a dry-run result only.
        self._state_change_handlers: dict[str, Callable[[dict[str, Any]], Any]] = {
            "docker_stop": self._simulate_docker_stop,
            "docker_restart": self._simulate_docker_restart,
            "gh_issue_create": self._simulate_gh_issue_create,
        }

    @property
    def command_surface(self) -> frozenset[str]:
        """Every action name this backend recognises (reads + state-changes).

        Exposed so the CLI-safety test can assert, from the outside, that no
        kubernetes action exists anywhere in what this backend can be asked to do.
        """
        return frozenset(self._read_builders) | frozenset(self._state_change_handlers)

    # -- dispatcher entry point ------------------------------------------------

    def run(self, action: ProposedAction) -> Any:
        """Execute *action* and return its result.

        Resolution order, with every refusal happening *before* any command runs:

            1. a known-destructive action raises :class:`CliDestructiveActionError`;
            2. a read action builds an argv and **really executes** it;
            3. a state-change action returns a **simulated** result (never shells out);
            4. anything else raises :class:`CliUnsupportedActionError`.
        """
        # (1) Destructive verbs are denied first and loudest — no tier unlocks them.
        if action.action_name in _DESTRUCTIVE_ACTIONS:
            raise CliDestructiveActionError(
                f"cli_executor refuses destructive action {action.action_name!r} "
                f"(always denied; no approval path)"
            )

        # (2) Read-only inspections really run.
        builder = self._read_builders.get(action.action_name)
        if builder is not None:
            argv = builder(action.arguments)  # may raise CliArgumentError pre-exec
            return self._execute(argv)

        # (3) State-changes are simulated only, even though they reach here only
        #     after approval (the gate gates them; this backend still never writes).
        handler = self._state_change_handlers.get(action.action_name)
        if handler is not None:
            return handler(action.arguments)

        # (4) Unknown to this backend → refused, never guessed at.
        raise CliUnsupportedActionError(
            f"cli_executor backend cannot perform {action.action_name!r} "
            f"(not in its allow-list of safe gh/docker commands)"
        )

    # -- read argv builders (real, read-only) ----------------------------------

    def _build_docker_ps(self, args: dict[str, Any]) -> list[str]:
        """``docker ps`` — list running containers as JSON lines (read-only)."""
        return ["docker", "ps", "--format", "{{json .}}"]

    def _build_docker_logs(self, args: dict[str, Any]) -> list[str]:
        """``docker logs --tail N <container>`` — read recent logs (read-only)."""
        container_id = _safe(args, "container_id")
        return ["docker", "logs", "--tail", "50", container_id]

    def _build_gh_run_list(self, args: dict[str, Any]) -> list[str]:
        """``gh run list`` — list recent CI workflow runs (read-only)."""
        return ["gh", "run", "list", "--limit", "10"]

    def _build_gh_pr_list(self, args: dict[str, Any]) -> list[str]:
        """``gh pr list`` — list open pull requests (read-only)."""
        return ["gh", "pr", "list", "--state", "open", "--limit", "10"]

    # -- state-change simulations (approval-gated; never executed) -------------

    def _simulate_docker_stop(self, args: dict[str, Any]) -> dict[str, Any]:
        """Simulate ``docker stop <container>`` — report intent only, run nothing.

        This is the "state-change, approval-gated at L3" case. By the time it
        runs, an operator has approved it — but this backend still does not shell
        out, so **nothing is actually stopped**.
        """
        container_id = _safe(args, "container_id")
        return _simulated("docker stop", container_id, new_status="exited")

    def _simulate_docker_restart(self, args: dict[str, Any]) -> dict[str, Any]:
        """Simulate ``docker restart <container>`` — report intent only, run nothing."""
        container_id = _safe(args, "container_id")
        return _simulated("docker restart", container_id, new_status="running")

    def _simulate_gh_issue_create(self, args: dict[str, Any]) -> dict[str, Any]:
        """Simulate ``gh issue create`` — return the issue that *would* be opened.

        No issue is filed anywhere; the number is a fixed placeholder to make clear
        this is a simulation, not a real GitHub object.
        """
        title = _safe(args, "title", allow_spaces=True)
        result = _simulated("gh issue create", title)
        result.update({"issue_number": 9001, "title": title, "state": "open"})
        return result

    # -- the one real-execution helper -----------------------------------------

    def _execute(self, argv: list[str]) -> dict[str, Any]:
        """Run *argv* as a real subprocess (no shell) and return its actual output.

        A non-zero exit is **not** an exception — it is captured and returned, so the
        caller (and the agent) sees the command's real result (e.g. a `docker`
        daemon that is down). Only a missing tool or a timeout is surfaced as an
        error, because in those cases there is no command result to report.
        """
        if shutil.which(argv[0]) is None:
            raise CliToolUnavailableError(
                f"CLI tool {argv[0]!r} is not installed; cannot run {' '.join(argv)!r}"
            )
        try:
            # shell=False (the default) is the security-critical choice: argv is
            # passed straight to execve, so there is no shell to interpret any value.
            proc = subprocess.run(
                argv,
                capture_output=True,
                text=True,
                timeout=self._timeout,
                check=False,
            )
        except subprocess.TimeoutExpired:
            return {
                "command": " ".join(argv),
                "argv": argv,
                "executed": True,
                "timed_out": True,
                "returncode": None,
            }
        return {
            "command": " ".join(argv),
            "argv": argv,
            "executed": True,         # a real subprocess ran
            "timed_out": False,
            "returncode": proc.returncode,
            "stdout": proc.stdout,
            "stderr": proc.stderr,
        }


# Module-level helpers keep the handlers above short and uniform.


def _safe(args: dict[str, Any], key: str, *, allow_spaces: bool = False) -> str:
    """Return ``args[key]`` if present and safe to interpolate, else refuse.

    The value must be a non-empty string matching the conservative :data:`_SAFE_ARG`
    pattern (no leading dash, no shell metacharacters). ``allow_spaces`` relaxes the
    check for free-text fields like an issue title, while still rejecting shell
    metacharacters. Refusing here means an unsafe value never reaches the subprocess.
    """
    if key not in args or args[key] in (None, ""):
        raise CliArgumentError(f"missing required argument {key!r}")
    value = args[key]
    if not isinstance(value, str):
        raise CliArgumentError(f"argument {key!r} must be a string, got {type(value).__name__}")

    if allow_spaces:
        # Free text: forbid the shell metacharacters and control characters that
        # could matter if the value were ever logged or re-used, but allow spaces.
        if any(ch in value for ch in ";|&$`<>\n\r\\\"'") or value.startswith("-"):
            raise CliArgumentError(f"argument {key!r} contains unsafe characters: {value!r}")
        return value

    if not _SAFE_ARG.match(value):
        raise CliArgumentError(
            f"argument {key!r} is not a safe identifier: {value!r} "
            f"(must start alphanumeric; no flags or shell metacharacters)"
        )
    return value


def _simulated(command: str, target: str, **extra: Any) -> dict[str, Any]:
    """Build the standard simulated/dry-run envelope for a state-change.

    Carries ``simulated``/``executed`` flags so no consumer can mistake a simulated
    state-change for a real one; nothing here ever ran as a subprocess.
    """
    return {
        "command": command,
        "target": target,
        "executed": False,     # state-changes never shell out
        "simulated": True,
        "persisted": False,
        "note": "Approval-gated state-change; simulated only, no real command executed.",
        **extra,
    }