fix(runner): read-only launcher and explicit ungraded-result failures

README.md

@@ -168,8 +168,15 @@ findings against last week's.
 ### Cost control
 
 `--budget-usd` is a runaway-loop ceiling per pass, not a spending target. A pass that trips
-it returns nothing and its section of the report comes out empty, so setting it low does
-not save money — it silently produces a worse report. The default is deliberately generous.
+it returns nothing, so setting it low does not save money — it produces a worse report.
+The default is deliberately generous.
+
+A pass that fails, times out, or returns no `structured_output`, or one whose shape does
+not match its schema (a missing required field, a wrong type), is not treated as "no
+findings". It is listed under **Coverage gaps** with the reason, and that section is what
+tells a clean result apart from an ungraded one. A well-formed empty findings list is a
+real answer and is not a gap. The command still exits 0 when some passes fail, because a
+partial report is written; read the gaps section before trusting a report as clean.
 
 `--model` is passed straight through to the CLI, so you can run the cheap passes on a small
 model.

SECURITY.md

@@ -30,6 +30,15 @@ Useful context when judging severity:
   open the files themselves, and everything untrusted is wrapped in an explicit data frame
   saying it is data rather than instructions. This is a mitigation for prompt injection
   from an audited file, not a guarantee against it.
+- **Each pass runs with a read-only tool set.** The launcher starts `claude` with
+  `--tools=Read,Grep,Glob`, an explicit `--disallowedTools` list (shell, edit, write,
+  notebook, web fetch and search) and `--strict-mcp-config` (no MCP servers). Before
+  launching, it refuses any command that lacks these flags or adds a widening one
+  (`--allowedTools`, `--permission-mode`, `--dangerously-skip-permissions`, `--mcp-config`,
+  `--add-dir`), so a code change cannot silently drop the boundary. This limits what an
+  injected instruction can *do*; it does not stop a pass from *reading* any file your user
+  account can read, because audited files are read in place, not copied to an isolated
+  directory.
 
 ## Known limits, not vulnerabilities
 

src/agent_audit/runner.py

@@ -32,6 +32,36 @@
 # prompt dominates cache-creation, so the default is deliberately generous.
 DEFAULT_BUDGET_USD = 5.00
 
+#: The only built-in tools an analysis pass may use: read files, search them.
+#: The files being audited are untrusted input — they can carry instructions
+#: aimed at the auditor — so the boundary is enforced by the launcher, below
+#: the prompt, not by asking the model to behave.
+READ_ONLY_TOOLS: tuple[str, ...] = ("Read", "Grep", "Glob")
+
+#: Tools named in ``--disallowedTools`` as a second line of defence: anything
+#: that writes, executes, or reaches the network.
+DENIED_TOOLS: tuple[str, ...] = (
+    "Bash",
+    "Edit",
+    "Write",
+    "MultiEdit",
+    "NotebookEdit",
+    "WebFetch",
+    "WebSearch",
+)
+
+#: Flags that would widen the capability set. :func:`assert_read_only` refuses a
+#: command carrying any of them, whoever added it.
+_WIDENING_FLAGS = (
+    "--allowedTools",
+    "--allowed-tools",
+    "--dangerously-skip-permissions",
+    "--allow-dangerously-skip-permissions",
+    "--permission-mode",
+    "--mcp-config",
+    "--add-dir",
+)
+
 
 class AgentError(RuntimeError):
     """An analysis pass failed, timed out, or returned unparseable output."""
@@ -51,6 +81,89 @@ def as_object_schema(schema: dict[str, Any]) -> dict[str, Any]:
     return schema
 
 
+#: Validation problems reported per pass before the rest are summarised as a count.
+MAX_SCHEMA_PROBLEMS = 5
+
+
+def _json_type(value: Any) -> str:
+    """JSON-schema name of a decoded JSON value."""
+    if value is None:
+        return "null"
+    if isinstance(value, bool):
+        return "boolean"
+    if isinstance(value, int):
+        return "integer"
+    if isinstance(value, float):
+        return "number"
+    if isinstance(value, str):
+        return "string"
+    if isinstance(value, list):
+        return "array"
+    return "object"
+
+
+def _type_matches(value: Any, expected: str) -> bool:
+    actual = _json_type(value)
+    return actual == expected or (expected == "number" and actual == "integer")
+
+
+def schema_problems(value: Any, schema: dict[str, Any], path: str) -> list[str]:
+    """Where ``value`` departs from ``schema``, as ``"<path>: <problem>"`` lines.
+
+    Checks the subset of JSON Schema this package's own schemas use — ``type``
+    (including a list of types), ``required``, ``properties`` and ``items``.
+    Enums and formats are not enforced: a value the model chose badly is still
+    a graded result, whereas a missing or wrong-shaped one is not.
+    """
+    expected = schema.get("type")
+    if expected is not None:
+        allowed = [expected] if isinstance(expected, str) else list(expected)
+        if not any(_type_matches(value, t) for t in allowed):
+            return [f"{path}: expected {' or '.join(allowed)}, got {_json_type(value)}"]
+    problems: list[str] = []
+    if isinstance(value, dict):
+        for name in schema.get("required", []):
+            if name not in value:
+                problems.append(f"{path}.{name}: required field missing")
+        for name, sub in (schema.get("properties") or {}).items():
+            if name in value and isinstance(sub, dict):
+                problems += schema_problems(value[name], sub, f"{path}.{name}")
+    elif isinstance(value, list) and isinstance(schema.get("items"), dict):
+        for index, item in enumerate(value):
+            problems += schema_problems(item, schema["items"], f"{path}[{index}]")
+    return problems
+
+
+def validate_result(envelope: dict[str, Any], schema: dict[str, Any] | None) -> None:
+    """Raise :class:`AgentError` unless ``envelope`` carries a well-shaped result.
+
+    An empty findings list that matches the schema is a real answer ("nothing
+    found") and passes. A missing, null or wrong-shaped ``structured_output`` is
+    not an answer at all: the pass was never graded, and reading it as "no
+    findings" would print a clean report over a hole.
+    """
+    if schema is None:
+        return
+    value = envelope.get("structured_output")
+    if value is None:
+        raise AgentError(
+            "pass returned no structured_output "
+            f"(envelope keys: {', '.join(sorted(envelope)) or 'none'}); it was not graded"
+        )
+    # :func:`rows` also accepts a bare list for a boxed array schema; so does this.
+    if schema.get("type") == "array" and isinstance(value, list):
+        target = schema
+    else:
+        target = as_object_schema(schema)
+    problems = schema_problems(value, target, "structured_output")
+    if problems:
+        shown = problems[:MAX_SCHEMA_PROBLEMS]
+        extra = len(problems) - len(shown)
+        if extra:
+            shown.append(f"(+{extra} more)")
+        raise AgentError("pass returned malformed structured_output: " + "; ".join(shown))
+
+
 def failure_detail(proc: subprocess.CompletedProcess[str]) -> str | None:
     """Describe a failed run, or ``None`` when it succeeded.
 
@@ -99,6 +212,11 @@ def build_command(
         "--max-budget-usd",
         str(budget_usd),
         "--no-session-persistence",
+        # ``=`` form: these flags take variadic values, and a bare value list
+        # would swallow whatever argument follows it.
+        f"--tools={','.join(READ_ONLY_TOOLS)}",
+        f"--disallowedTools={','.join(DENIED_TOOLS)}",
+        "--strict-mcp-config",
     ]
     if model:
         cmd += ["--model", model]
@@ -107,6 +225,42 @@ def build_command(
     return cmd
 
 
+def _flag_values(cmd: list[str], flag: str) -> list[str] | None:
+    """Comma-split values of ``--flag=a,b`` / ``--flag a,b`` in ``cmd``, or ``None``."""
+    for i, token in enumerate(cmd):
+        if token == flag and i + 1 < len(cmd):
+            return [v for v in cmd[i + 1].split(",") if v]
+        if token.startswith(flag + "="):
+            return [v for v in token.split("=", 1)[1].split(",") if v]
+    return None
+
+
+def assert_read_only(cmd: list[str]) -> None:
+    """Refuse to launch a command that does not carry the read-only restrictions.
+
+    Raises :class:`AgentError` rather than running unrestricted, so a future
+    change to :func:`build_command` that drops or widens the boundary fails
+    loudly instead of silently handing untrusted input a writable agent.
+    """
+    tools = _flag_values(cmd, "--tools")
+    denied = _flag_values(cmd, "--disallowedTools")
+    problems = []
+    if tools is None or not set(tools) <= set(READ_ONLY_TOOLS):
+        problems.append(f"--tools must be a subset of {','.join(READ_ONLY_TOOLS)}")
+    if denied is None or not set(DENIED_TOOLS) <= set(denied):
+        problems.append("--disallowedTools must name every write/execute/network tool")
+    if "--strict-mcp-config" not in cmd:
+        problems.append("--strict-mcp-config is missing")
+    for token in cmd:
+        flag = token.split("=", 1)[0]
+        if flag in _WIDENING_FLAGS:
+            problems.append(f"{flag} widens the capability set")
+    if problems:
+        raise AgentError(
+            "refusing to launch: command is not read-only (" + "; ".join(problems) + ")"
+        )
+
+
 def call_claude(
     prompt: str,
     schema: dict[str, Any] | None,
@@ -117,9 +271,11 @@ def call_claude(
 ) -> dict[str, Any]:
     """Run one analysis pass and return the parsed result envelope.
 
-    Raises :class:`AgentError` on timeout, failure, or unparseable output.
+    Raises :class:`AgentError` on timeout, failure, unparseable output, or — when
+    a ``schema`` was requested — a missing or wrong-shaped ``structured_output``.
     """
     cmd = build_command(prompt, schema, budget_usd=budget_usd, model=model)
+    assert_read_only(cmd)
     try:
         proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout_s)
     except FileNotFoundError as exc:
@@ -133,9 +289,13 @@ def call_claude(
     if detail is not None:
         raise AgentError(f"pass failed: {detail}")
     try:
-        return json.loads(proc.stdout)
+        envelope = json.loads(proc.stdout)
     except json.JSONDecodeError as exc:
         raise AgentError(f"pass returned non-JSON: {exc}") from exc
+    if not isinstance(envelope, dict):
+        raise AgentError(f"pass result is not a JSON object (got {_json_type(envelope)})")
+    validate_result(envelope, schema)
+    return envelope
 
 
 def output(envelope: dict[str, Any]) -> dict[str, Any]: