Agent hardening

src/mergai/agents/claude_cli.py

@@ -5,9 +5,70 @@
 
 from ..utils.output import echo_err as _echo
 from .base import CliAgent
+from .env import agent_subprocess_env
 from .error import AgentError, AgentErrorType, AgentResult
 from .response_utils import parse_response_json
 
+# Tools the agent is allowed to use, auto-approved without a prompt. `Bash` is
+# allowed broadly (the agent needs arbitrary *read* shell -- `git diff`,
+# `git ls-tree`, `git show | grep`, `for` loops -- to inspect the merge), plus
+# the file tools for editing. The dangerous subset is carved out by the
+# deny-list below, which takes precedence over this allow.
+#
+# This broad allow is load-bearing, NOT decorative: in `--print` mode the CLI
+# does not auto-approve un-allowed Bash (it returns "requires approval", a
+# denial when non-interactive), so without an explicit Bash allow the agent
+# cannot run any read command. (Verified in CI: on claude 2.1.198, a deny-list-
+# only config denied git ls-tree / git diff / for-loops.)
+GUARDRAIL_ALLOWED_TOOLS = ["Bash", "Read", "Edit", "Write", "Grep", "Glob"]
+
+# Tools/commands the agent must never use: anything that could modify the remote
+# or be used to bypass the no-remote-write guarantee. Passed via --disallowedTools
+# on *every* invocation. Deny takes precedence over the broad Bash allow above,
+# so these stay blocked while all other Bash is permitted.
+#
+# NOTE: Bash deny rules are prefix-matched and best-effort (e.g. an obfuscated
+# `git -c ... push` can evade them); the credential boundary is what actually
+# guarantees no remote writes (see env.py and the workflow's
+# persist-credentials: false).
+GUARDRAIL_DISALLOWED_TOOLS = [
+    # Remote-mutating git / the GitHub CLI (which the agent has no reason to use
+    # and no credential for).
+    "Bash(git push)",
+    "Bash(git push:*)",
+    "Bash(gh:*)",
+    # Network egress: exfiltration, or fetching code/instructions to run.
+    "WebFetch",
+    "WebSearch",
+    # Spawning unconstrained work / running arbitrary skills or workflows, which
+    # could sidestep these restrictions.
+    "Task",
+    "Agent",
+    "Skill",
+    "Workflow",
+    # Harness/orchestration tools not needed for conflict resolution.
+    "CronCreate",
+    "CronDelete",
+    "CronList",
+    "ScheduleWakeup",
+    "Monitor",
+    "PushNotification",
+    "SendMessage",
+    "RemoteTrigger",
+    "DesignSync",
+    "EnterWorktree",
+    "ExitWorktree",
+    "ToolSearch",
+    "TaskCreate",
+    "TaskGet",
+    "TaskList",
+    "TaskOutput",
+    "TaskStop",
+    "TaskUpdate",
+    "ReportFindings",
+    "NotebookEdit",
+]
+
 
 class ClaudeCLIAgent(CliAgent):
     """Claude CLI Agent for running prompts via Claude Code CLI.
@@ -22,7 +83,10 @@ def __init__(self, model: str, yolo: bool = False, debug: bool = False):
 
         Args:
             model: The model to use (e.g., "claude-sonnet-4-20250514" or "sonnet").
-            yolo: Enable bypass of all permission checks (--dangerously-skip-permissions).
+            yolo: Auto-approve file edits (acceptEdits). Despite the name, this
+                does NOT enable --dangerously-skip-permissions. Remote-write /
+                network / work-spawning tools are denied on every invocation
+                regardless of this flag (see GUARDRAIL_DISALLOWED_TOOLS).
             debug: Enable debug logging.
         """
         super().__init__(model)
@@ -296,15 +360,21 @@ def build_args(
             "--verbose",  # Required for stream-json
         ]
 
-        if self.yolo:
-            args.append("--dangerously-skip-permissions")
-        elif allowed_write_paths:
-            # Use acceptEdits mode to auto-approve write operations.
-            # NOTE: Claude CLI's acceptEdits mode does not restrict edits to specific paths;
-            # it broadly auto-approves all file edits. The allowed_write_paths parameter is
-            # accepted for interface compatibility but cannot enforce path-level restrictions
-            # in print mode. For strict path isolation, consider running the agent in a
-            # sandboxed working directory.
+        # Allow Bash + file tools broadly (the agent needs arbitrary read shell
+        # to inspect the merge), then deny the remote-mutating / bypass subset.
+        # Both are passed on *every* invocation. The broad allow is required: in
+        # --print mode the CLI denies un-allowed Bash ("requires approval"), so
+        # without it the agent cannot run any read command. Deny wins over allow,
+        # so the denied commands stay blocked.
+        args.extend(["--allowedTools", ",".join(GUARDRAIL_ALLOWED_TOOLS)])
+        args.extend(["--disallowedTools", ",".join(GUARDRAIL_DISALLOWED_TOOLS)])
+
+        if self.yolo or allowed_write_paths:
+            # Auto-approve file edits in the working directory. "yolo"
+            # deliberately does NOT map to --dangerously-skip-permissions (which
+            # would also bypass the deny-list above). acceptEdits does not
+            # restrict edits to specific paths, so allowed_write_paths cannot be
+            # enforced here; the post-run validator checks for stray writes.
             args.extend(["--permission-mode", "acceptEdits"])
 
         if self.debug:
@@ -344,7 +414,9 @@ def run_prompt(
             stdout=subprocess.PIPE,
             stderr=subprocess.STDOUT,  # Combine stderr with stdout to avoid deadlock
             text=True,
-            env=os.environ.copy(),
+            # Strip GitHub write credentials so the agent cannot push or issue
+            # GitHub write APIs; local file/git operations are unaffected.
+            env=agent_subprocess_env(),
         )
 
         result: dict = {}

src/mergai/agents/env.py

@@ -0,0 +1,43 @@
+"""Environment sanitization for spawned agent subprocesses.
+
+The agent must never be able to modify remote state (push to origin or issue
+GitHub write APIs). Because the agent runs as a subprocess of mergai, it would
+otherwise inherit mergai's environment -- which in CI carries the GitHub write
+token used by mergai's own deterministic push/PR steps.
+
+`agent_subprocess_env` strips those credentials from the environment handed to
+the agent, so that even a fully compromised or prompt-injected agent cannot
+authenticate to GitHub at all. The agent only needs local access (edit files,
+local git reads and commits), none of which requires a credential.
+
+The parent mergai process keeps its own (credentialed) environment, so its
+push/PR steps continue to work -- only the spawned agent is de-privileged.
+
+NOTE: This closes the *environment* path only. `actions/checkout` also persists
+a credential into the repo's ``.git/config`` (``http.extraheader``); a bare
+`git push` authenticates via that on-disk token regardless of the environment,
+and the agent can even read it. That path must be closed separately with
+``persist-credentials: false`` in the workflow (plus a credential helper so
+mergai's own push still authenticates from the parent environment).
+"""
+
+import os
+
+# Environment variables that authenticate to GitHub: the gh CLI reads these,
+# and a credential-helper-based `git push` derives its token from them.
+_GITHUB_CREDENTIAL_VARS = ("GITHUB_TOKEN", "GH_TOKEN", "GH_ENTERPRISE_TOKEN")
+
+
+def agent_subprocess_env() -> dict[str, str]:
+    """Return a copy of the environment with GitHub credentials removed.
+
+    The agent is left with no GitHub token, so it cannot push or call GitHub
+    APIs; local file and git operations need no credential and are unaffected.
+
+    Returns:
+        A sanitized environment dict suitable for the agent subprocess.
+    """
+    env = os.environ.copy()
+    for var in _GITHUB_CREDENTIAL_VARS:
+        env.pop(var, None)
+    return env

src/mergai/agents/gemini_cli.py

@@ -5,6 +5,7 @@
 
 from ..utils.output import echo_err as _echo
 from .base import CliAgent
+from .env import agent_subprocess_env
 from .error import AgentError, AgentErrorType, AgentResult
 from .response_utils import parse_response_json
 
@@ -66,10 +67,15 @@ def parse_stats(self, session_data: dict) -> dict:
         return {"models": models}
 
     def build_args(self, prompt: str) -> list:
+        # Always use auto_edit: auto-approve file edits in the working repo, but
+        # NOT other tools. yolo deliberately does NOT map to Gemini's "yolo"
+        # approval mode (which auto-approves every tool, incl. shell commands
+        # that could push or reach the remote). This mirrors the Claude adapter,
+        # where yolo maps to acceptEdits rather than --dangerously-skip-permissions.
         args = [
             "gemini",
             "--approval-mode",
-            "yolo" if self.yolo else "auto_edit",
+            "auto_edit",
             "-o",
             "stream-json",
         ]
@@ -97,7 +103,9 @@ def run_prompt(self, prompt: str):
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             text=True,
-            env=os.environ.copy(),
+            # Strip GitHub write credentials so the agent cannot push or issue
+            # GitHub write APIs; local file/git operations are unaffected.
+            env=agent_subprocess_env(),
         )
 
         result: dict = {}

src/mergai/ci/gate.py

@@ -23,6 +23,7 @@ def _list_run_status(
     run: "github.WorkflowRun.WorkflowRun",
     *,
     check_findings: bool,
+    check_failure_kind: bool = False,
 ) -> tuple[str, str]:
     """Return ``(status, notes)`` describing what mergai sees for this run.
 
@@ -63,12 +64,15 @@ def _list_run_status(
         workflow_name=run.name,
         pr_number=_resolve_pr_number(run),
         check_findings=check_findings,
-        # Listing should stay cheap and still show a failed run as a failure
-        # entry; the non-code-failure side-calls (approvals / job steps) are
-        # only worth making on the real dispatch path. A ``cancelled`` run is
-        # the exception: classify_run always inspects its jobs to surface a
-        # fail-fast-masked failure, regardless of this flag (see classify_run).
-        check_failure_kind=False,
+        # ``ci list`` leaves this False: listing should stay cheap and still
+        # show a failed run as a failure entry; the non-code-failure side-calls
+        # (approvals / job steps) are only worth making on the real dispatch
+        # path. The gate (``_actionable_state``) passes True so an
+        # approval-rejected / infra ``failure`` is classified ``skip`` and does
+        # not spin up the privileged handler. A ``cancelled`` run is inspected
+        # regardless of this flag: classify_run always looks at its jobs to
+        # surface a fail-fast-masked failure (see classify_run).
+        check_failure_kind=check_failure_kind,
     )
 
     if decision.actionable:
@@ -153,22 +157,73 @@ def _watched_runs_for_head(
     return latest
 
 
-def _aggregate_state(
+def _actionable_state(
+    app: AppContext,
     runs_by_workflow: dict[str, "github.WorkflowRun.WorkflowRun"],
+    *,
+    check_findings: bool = True,
+    check_failure_kind: bool = True,
 ) -> Literal["in-progress", "success", "failure", "none"]:
     """Reduce the per-workflow latest runs to a single gate token.
 
+    Unlike a raw conclusion rollup, ``failure`` here means *mergai has
+    actionable work* — not merely "something didn't return success". The per-run
+    verdict comes from :func:`_list_run_status` (the same classification
+    ``ci list`` / ``ci fix all`` use), so the gate agrees with what the fixer
+    would actually do:
+
+    * a plain ``cancelled`` run (user/timeout cancel, superseded) → ``skip``;
+    * a ``failure`` from a rejected deployment approval or an infra failure with
+      no failing step → ``skip`` (``check_failure_kind``);
+    * a run mergai already handled (a ``ci_fix`` solution or a posted comment
+      exists for it) → ``applied`` / ``commented`` — not re-triggered;
+    * a *passing* run with Code Scanning findings → ``pending`` (actionable).
+
+    The tokens:
+
     * ``none``        — no watched runs for HEAD (e.g. all skipped).
-    * ``in-progress`` — at least one watched run hasn't completed.
-    * ``success``     — every watched run completed with ``success``.
-    * ``failure``     — all completed, but at least one did not succeed
-                        (``failure`` / ``cancelled`` / ``timed_out`` / …).
+    * ``in-progress`` — at least one watched run hasn't completed. This wins over
+                        ``failure`` so the handler fires only once the whole
+                        watched set for HEAD is done and a later completion
+                        finalizes it (keeps the CI-fix loop converging).
+    * ``failure``     — all completed and at least one run is actionable
+                        (``pending``).
+    * ``success``     — all completed and nothing is actionable (passed,
+                        skipped, or already handled).
+
+    Per-run classification is best-effort: a Code Scanning findings lookup needs
+    a token with ``security-events: read``, so when the gate runs without it (or
+    any read side-call errors) the run degrades to non-actionable rather than
+    crashing the gate or spinning the handler up on an unverifiable signal.
     """
     if not runs_by_workflow:
         return "none"
-    runs = list(runs_by_workflow.values())
-    if any(run.status != "completed" for run in runs):
+
+    statuses: list[str] = []
+    for run in runs_by_workflow.values():
+        try:
+            status, _ = _list_run_status(
+                app,
+                run,
+                check_findings=check_findings,
+                check_failure_kind=check_failure_kind,
+            )
+        except Exception as exc:  # noqa: BLE001 — best-effort; the gate must not crash
+            # A read side-call failed: a code-scanning lookup the gate token
+            # can't make (no security-events read → GithubException), or any
+            # transient requester/network error underneath it. Treat the run as
+            # non-actionable so the gate stays read-only, always emits a token,
+            # and never fires the handler on a signal it couldn't confirm.
+            click.echo(
+                f"warning: could not classify run {run.id} ({run.name}): {exc}; "
+                "treating as non-actionable",
+                err=True,
+            )
+            status = "skip"
+        statuses.append(status)
+
+    if "wait" in statuses:
         return "in-progress"
-    if all(run.conclusion == "success" for run in runs):
-        return "success"
-    return "failure"
+    if "pending" in statuses:
+        return "failure"
+    return "success"

src/mergai/commands/ci.py

@@ -53,7 +53,7 @@
     _run_jobs,
     build_workflow_context_for_run,
 )
-from ..ci.gate import _aggregate_state, _list_run_status, _watched_runs_for_head
+from ..ci.gate import _actionable_state, _list_run_status, _watched_runs_for_head
 from ..ci.handlers import get_handler
 from ..solution_types import CI_FIX
 from ..utils.formatters import format_ascii_table
@@ -744,15 +744,25 @@ def status(app: AppContext, state_only: bool) -> None:
 
     \b
     * in-progress — at least one watched run for HEAD hasn't completed.
-    * success     — every watched run for HEAD completed successfully.
-    * failure     — all completed, but at least one didn't succeed.
+    * failure     — all completed and at least one run is *actionable*: mergai
+                    would run the fixer on it.
+    * success     — all completed and nothing is actionable (passed, skipped,
+                    or already handled by mergai).
     * none        — no watched runs for HEAD (e.g. all skipped).
 
+    ``failure`` means actionable work exists, not merely "a run didn't return
+    success": a plain cancellation, a rejected deployment approval, an infra
+    failure with no failing step, and a run mergai already fixed all reduce to
+    ``success`` so the CI-fix handler is only spun up when there is really
+    something for it to do. A *passing* run with Code Scanning findings does
+    count as ``failure`` (needs a token with ``security-events: read`` to
+    detect; without it that run degrades to non-actionable).
+
     Exits 0 in every case; the state is communicated on stdout. With
     ``--state`` only the bare token is printed (for shell capture).
     """
     runs_by_workflow = _watched_runs_for_head(app)
-    state = _aggregate_state(runs_by_workflow)
+    state = _actionable_state(app, runs_by_workflow)
 
     if state_only:
         click.echo(state)