Route the probe's four shell calls through one audited chokepoint

Promotion into src/ put four pre-existing shell=True calls under
agentseam-no-shell-true for the first time: security.yml scans src/ and tests/,
never tools/, so the rule had not seen this code. On main, src/ has no shell=True
and tools/ has exactly these four. The rule is working; the code is not new.

Its prescribed fix -- pass an argv list -- would be wrong here. reference_agent
is Claude Code's protocol made executable, and Claude Code runs a hook by handing
a command string to a shell; an argv list would emulate a vendor that does not
exist, and every tested-basis row is measured against this driver. The transform
trial runs the command as the hook rewrote it, so measuring the rewrite means
running the rewrite.

So the exception is real, and now lives in one place instead of four: _shell.py's
run_shell() carries the single nosemgrep and the trust-boundary rationale beside
it. The rule stays global and unmodified. Verified narrow: a canary shell=True
added elsewhere in the same package still fails semgrep, and passes again once
removed.

semgrep exit 0 on CI's own invocation; 1615 passed, 4 skipped; ruff and
ruff format clean. Probe re-checked end to end: agreement exits 0, a forced
disagreement prints DISAGREES and exits 1.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Signed-off-by: Claude <noreply@anthropic.com>

Author: claude

src/agentseam/probe/_shell.py

@@ -0,0 +1,29 @@
+"""The one place this package hands a command string to a shell.
+
+`agentseam-no-shell-true` bans `shell=True` across `src/`, and it is right to: the runtime
+path sits inline in a developer's agent loop, on payloads an attacker can influence. The
+probe is not that path. It is a deliberate command-runner, and the shell is the thing under
+measurement rather than an implementation detail:
+
+- `reference_agent` is Claude Code's protocol made executable, and Claude Code runs a hook by
+  handing a command string to a shell. An argv list would emulate a vendor that does not
+  exist, and every `tested`-basis row is measured against this driver.
+- The `transform` trial runs the command *as the hook rewrote it*. Measuring the rewrite means
+  running the rewrite.
+- The driver template is a command line the operator typed on their own machine.
+
+So the exception is real, and it is confined to this function rather than scattered across four
+call sites: one place to audit, and the rule keeps biting everywhere else in `src/`.
+"""
+
+from __future__ import annotations
+
+import subprocess
+
+
+def run_shell(command, *, check=False, capture_output=True, **kwargs):
+    """`subprocess.run(command, shell=True)`, defaulted to capture and never to raise."""
+    # nosemgrep: agentseam-no-shell-true -- the audited exception; see this module's docstring.
+    return subprocess.run(  # noqa: S602
+        command, shell=True, check=check, capture_output=capture_output, **kwargs
+    )

src/agentseam/probe/experiment_driver.py

@@ -11,6 +11,8 @@
 import json
 import subprocess
 
+from . import _shell
+
 
 def drive_real(command, workspace, *, trigger):
     """Drive a real agent CLI in the scratch workspace.
@@ -26,9 +28,7 @@ def drive_real(command, workspace, *, trigger):
     prompt = "Run this exact shell command and nothing else: %s" % trigger
     filled = command.replace("{prompt}", json.dumps(prompt))
     try:
-        proc = subprocess.run(  # noqa: S602
-            filled, shell=True, cwd=workspace, capture_output=True, text=True, timeout=300, check=False
-        )
+        proc = _shell.run_shell(filled, cwd=workspace, text=True, timeout=300)
     except subprocess.TimeoutExpired as exc:
         return {
             "returncode": None,

src/agentseam/probe/reference_agent.py

@@ -33,6 +33,8 @@
 import subprocess
 import sys
 
+from . import _shell
+
 CLAUDE_CODE_TIMEOUT_SECONDS = 60
 
 ALLOW, DENY, ASK = "allow", "deny", "ask"
@@ -145,14 +147,11 @@ def _gate(config_path, event, payload, *, cwd, timeout):
     for hook_command in _hook_commands(config_path, event):
         state["invoked"] = True
         try:
-            proc = subprocess.run(  # noqa: S602 - the command under test is ours, in a scratch dir
+            proc = _shell.run_shell(
                 hook_command,
-                shell=True,
                 input=blob,
-                capture_output=True,
                 cwd=cwd,
                 timeout=timeout if timeout is not None else CLAUDE_CODE_TIMEOUT_SECONDS,
-                check=False,
             )
         except subprocess.TimeoutExpired:
             state.update({"timed_out": True, "reason": "hook timed out, non-blocking", "exit": None})
@@ -200,7 +199,7 @@ def run_turn(config_path, *, command, cwd, session_id="reference-run", timeout=N
         )
         gates.setdefault("PreToolUse", pre)
         if pre["decision"] == ALLOW:
-            subprocess.run(pre["updated"] or command, shell=True, cwd=cwd, capture_output=True, check=False)  # noqa: S602
+            _shell.run_shell(pre["updated"] or command, cwd=cwd)
             runs += 1
 
         stop = _gate(
@@ -234,7 +233,7 @@ def run_pre_tool(config_path, *, command, cwd, session_id="reference-run", timeo
     )
     ran = False
     if state["decision"] == ALLOW:
-        subprocess.run(state["updated"] or command, shell=True, cwd=cwd, capture_output=True, check=False)  # noqa: S602
+        _shell.run_shell(state["updated"] or command, cwd=cwd)
         ran = True
     return {
         "decision": state["decision"],