diff --git a/docs/advanced/configuration.mdx b/docs/advanced/configuration.mdx index 5d56c9a16..3cf580dea 100644 --- a/docs/advanced/configuration.mdx +++ b/docs/advanced/configuration.mdx @@ -45,6 +45,10 @@ Configure Strix using environment variables or a config file. Telemetry toggle. Set to `0`, `false`, `no`, or `off` to disable telemetry (PostHog, Scarf, OTEL). + + Enable retraction of fixed findings on a resumed scan. Off by default, so `--resume` is append-only (findings only carry forward). When set to `1`/`true`, a resumed whitebox scan gains the `retract_vulnerability_report` tool: if the agent determines a rehydrated finding has been fixed, it is dropped from the cumulative report and SARIF so the resolved finding stops re-appearing. See [Resuming scans](/usage/cli#resuming-a-scan). + + OTLP/Traceloop base URL for remote OpenTelemetry export. If unset, Strix keeps traces local only. diff --git a/docs/usage/cli.mdx b/docs/usage/cli.mdx index 68f7d5edd..562c4105b 100644 --- a/docs/usage/cli.mdx +++ b/docs/usage/cli.mdx @@ -84,6 +84,15 @@ strix (--target | --target-list | --mount ) [options] counts. + + Resume a prior scan by its run name (the directory under `./strix_runs/`). + Restores the root agent plus every non-terminal subagent's full LLM history + and topology, and rehydrates prior findings so the report and SARIF are + cumulative across runs. Cannot be combined with `--target`/`--target-list`/`--mount` + (the original targets are restored from the prior run). See + [Resuming a scan](#resuming-a-scan). + + ## Examples ```bash @@ -112,6 +121,48 @@ strix --target-list ./targets.txt strix --mount ./huge-monorepo ``` +## Resuming a scan + +`--resume ` continues a prior scan: it restores the full agent history +and rehydrates the findings from the previous run, so the report and SARIF are +**cumulative** across runs. This suits re-scanning a target as it changes — for +example re-running against each push of a pull request. + +```bash +# First scan of the code +strix --run-name pr-42 -t ./repo + +# Later, after new commits land — pick up where it left off +strix --resume pr-42 +``` + +By default a resumed scan is **append-only**: prior findings carry forward, and +new issues are added, but nothing is removed. If the target has changed such that +a previously-reported finding is now **fixed**, it still re-appears in the +cumulative SARIF — which a code-scanning integration cannot auto-resolve. + +### Retracting fixed findings + +Set `STRIX_RESUME_RETRACT=1` to let a resumed **whitebox** scan drop findings it +determines are fixed. This exposes a `retract_vulnerability_report` tool: when +the agent judges a rehydrated finding resolved (the vulnerable code was fixed or +removed), it is removed from the cumulative report and SARIF, so a code-scanning +alert can auto-resolve. + +```bash +STRIX_RESUME_RETRACT=1 strix --resume pr-42 +``` + + + Retraction trusts the agent's judgement — the same judgement used to find the + vulnerability in the first place. Each retraction is re-checked against the + current source where possible and the result is recorded (`grounded` on the + tool result) for auditability, but the agent's decision is not overridden. + Because findings are re-derived on every scan, a mistaken retraction is + self-correcting: it re-appears on the next scan of the same code. The feature + is off by default; `--resume` stays append-only unless you opt in. + + ## Exit Codes | Code | Meaning | diff --git a/strix/agents/factory.py b/strix/agents/factory.py index 977d01f19..e89237844 100644 --- a/strix/agents/factory.py +++ b/strix/agents/factory.py @@ -411,6 +411,7 @@ def build_strix_agent( system_prompt_context: dict[str, Any] | None = None, extra_tools: Sequence[Tool] | None = None, instructions_override: str | None = None, + is_resume: bool = False, ) -> SandboxAgent[Any]: """Build a SandboxAgent for either root or child use. @@ -421,6 +422,11 @@ def build_strix_agent( registered via ``register_agent_tools``. instructions_override: Use this verbatim as the system prompt instead of rendering the built-in scan prompt. + is_resume: On a resumed root run, and only when opted in via + ``RuntimeSettings.resume_retract_enabled`` (``STRIX_RESUME_RETRACT``), + expose ``retract_vulnerability_report`` so the agent can drop a + rehydrated finding it re-verifies as fixed. Off by default → resume + stays append-only. A fresh scan never needs it. """ if instructions_override is not None: instructions = instructions_override @@ -437,6 +443,16 @@ def build_strix_agent( agent_tools = [*_EXTRA_TOOLS, *(extra_tools or [])] if is_root: tools: list[Tool] = [*_BASE_TOOLS, *agent_tools, finish_scan] + # retract_vulnerability_report is the inverse of create — meaningful + # only on a RESUMED root run (fresh scans have nothing to retract) and + # only when explicitly opted in. Guarded again at runtime: the tool + # refuses unless a source-tree reader is wired (whitebox resume). + from strix.config import load_settings + + if is_resume and load_settings().runtime.resume_retract_enabled: + from strix.tools.reporting.retract_tool import retract_vulnerability_report + + tools.append(retract_vulnerability_report) else: tools = [*_BASE_TOOLS, *agent_tools, agent_finish] _ensure_unique_tool_names(tools) diff --git a/strix/config/settings.py b/strix/config/settings.py index 9d16e0836..ba7a55f68 100644 --- a/strix/config/settings.py +++ b/strix/config/settings.py @@ -56,6 +56,16 @@ class RuntimeSettings(BaseSettings): # on large repos). Above this, the user must bind-mount via ``--mount``. # Set to 0 (or less) to disable the pre-flight check entirely. max_local_copy_mb: int = Field(default=1024, alias="STRIX_MAX_LOCAL_COPY_MB") + # Opt-in: on a resumed scan, expose the retract_vulnerability_report tool so + # the agent can DROP a rehydrated finding it re-verifies as fixed. Default + # OFF — resume stays append-only (today's behaviour) unless enabled. Only has + # any effect on a resumed whitebox run (the guard needs a source tree to + # re-verify against); a no-op otherwise. Useful when driving resume as a + # PR-lifecycle loop where cumulative findings must shrink as fixes land. + resume_retract_enabled: bool = Field( + default=False, + alias="STRIX_RESUME_RETRACT", + ) class TelemetrySettings(BaseSettings): diff --git a/strix/core/runner.py b/strix/core/runner.py index 0ebcc48b4..c74664714 100644 --- a/strix/core/runner.py +++ b/strix/core/runner.py @@ -98,6 +98,62 @@ def _compose_root_instructions_override( ) +def _make_workspace_file_reader( + session: Any, + ws_paths: list[str], +) -> Callable[[str], Any]: + """Build the target-file reader the retract guard uses to re-verify a fix. + + Fail-honest contract (the guard scores the result, see retract_tool): + - returns the file text when a candidate path is read; + - returns ``None`` ONLY when the file is POSITIVELY confirmed absent under + EVERY workspace root (``cat`` -> "No such file"); the sink is gone; + - RAISES when a read neither returns content nor confirms absence (exec + error, or an ambiguous nonzero like permission-denied). "Couldn't read" + is scored inconclusive by the guard, never mistaken for "confirmed gone" + — so a misdirected read can't masquerade as a verified fix. + """ + + async def _read_target_file(rel_path: str) -> str | None: + rel = rel_path.lstrip("/") + confirmed_absent_everywhere = True + errors: list[str] = [] + for base in ws_paths: + path = f"{base}/{rel}" + try: + result = await session.exec("cat", path, timeout=15) + except Exception as exc: # noqa: BLE001 — read failure ≠ proof of absence + confirmed_absent_everywhere = False + errors.append(f"{path}: exec error {exc!r}") + continue + if getattr(result, "exit_code", 1) == 0: + # ExecResult.stdout is bytes; the guard does a str substring + # check, so decode here. + stdout = result.stdout + return ( + stdout.decode("utf-8", errors="replace") + if isinstance(stdout, bytes | bytearray) + else stdout + ) + stderr = getattr(result, "stderr", b"") or b"" + stderr_txt = ( + stderr.decode("utf-8", errors="replace") + if isinstance(stderr, bytes | bytearray) + else str(stderr) + ) + # Only "No such file" is a POSITIVE absence signal. Any other nonzero + # (permission denied, is-a-directory) is NOT proof of a fix. + if "No such file" not in stderr_txt: + confirmed_absent_everywhere = False + errors.append(f"{path}: exit {result.exit_code}: {stderr_txt.strip()[:120]}") + if confirmed_absent_everywhere: + return None # genuinely absent under every root → sink is gone + msg = f"could not read {rel} for retract verification: {'; '.join(errors)}" + raise OSError(msg) + + return _read_target_file + + async def run_strix_scan( *, scan_config: dict[str, Any], @@ -236,6 +292,24 @@ async def run_strix_scan( system_prompt_context=root_context, ) + # Opt-in grounded retraction on resume: wire a sandbox-backed reader so + # the retract tool's guard can re-verify a "fixed" claim against the live + # /workspace tree before dropping a finding. Whitebox only — the guard + # checks source sinks; without a tree it stays fail-safe (refuse). Gated + # behind resume_retract_enabled so this is a no-op unless opted in. + if is_resume and is_whitebox and settings.runtime.resume_retract_enabled: + from strix.tools.reporting.retract_tool import set_target_file_reader + + ws_paths = [ + f"/workspace/{sub}" if sub else "/workspace" + for sub in ( + (t.get("details") or {}).get("workspace_subdir") + for t in targets + if t.get("type") == "local_code" + ) + ] or ["/workspace"] + set_target_file_reader(_make_workspace_file_reader(bundle["session"], ws_paths)) + root_agent = build_strix_agent( name="strix", skills=skills, @@ -246,6 +320,7 @@ async def run_strix_scan( chat_completions_tools=chat_completions_tools, system_prompt_context=root_context, instructions_override=root_instructions, + is_resume=is_resume, ) if not is_resume: @@ -397,6 +472,13 @@ async def spawn_child_agent(**kwargs: Any) -> dict[str, Any]: await coordinator.set_status(root_id, "failed") raise finally: + # Clear the retract reader so it can't bind to a torn-down session and + # verify a later scan against the wrong (or dead) sandbox. The guard + # then fails safe (no reader → refuse) until the next resume re-wires it. + with contextlib.suppress(Exception): + from strix.tools.reporting.retract_tool import set_target_file_reader + + set_target_file_reader(None) for s in sessions_to_close: with contextlib.suppress(Exception): s.close() diff --git a/strix/report/state.py b/strix/report/state.py index 178c4047c..a31f9c788 100644 --- a/strix/report/state.py +++ b/strix/report/state.py @@ -1,3 +1,4 @@ +import contextlib import json import logging import subprocess @@ -205,6 +206,43 @@ def hydrate_from_run_dir(self) -> None: len(self.vulnerability_reports), ) + @staticmethod + def _vuln_ordinal(report_id: object) -> int: + if isinstance(report_id, str) and report_id.startswith("vuln-"): + try: + return int(report_id.removeprefix("vuln-")) + except ValueError: + return 0 + return 0 + + def _next_vuln_ordinal(self) -> int: + """Next ``vuln-NNNN`` ordinal, derived from the max ordinal EVER seen + rather than the current list length. + + List-length allocation (``len(reports) + 1``) is only correct while the + report set is append-only. Once a report can be *removed* + (:meth:`retract_vulnerability_report` on resume), length-based ids + recycle: retract ``vuln-0002`` from ``[0001,0002,0003]`` and the next + add computes ``len+1 = 3`` -> ``vuln-0003``, colliding with a live id and + overwriting its on-disk MD, and letting a retracted id later re-emit + under a different finding (SARIF fingerprint collision). + + The high-water mark is persisted in ``run_record["max_vuln_ordinal"]`` + and restored by :meth:`hydrate_from_run_dir`, so it survives across the + process boundary of a resume. Without that, a NEW process rebuilds + ``_saved_vuln_ids`` only from the live ``vulnerabilities.json`` — the + retracted id is gone from disk, so the next allocation would reuse it. + We take the max of the persisted mark and every ordinal currently in + memory (live reports + retained ``_saved_vuln_ids``), so it stays correct + even if the record is missing/stale. + """ + candidates = [ + int(self.run_record.get("max_vuln_ordinal") or 0), + *(self._vuln_ordinal(r.get("id")) for r in self.vulnerability_reports), + *(self._vuln_ordinal(rid) for rid in self._saved_vuln_ids), + ] + return max(candidates) + 1 + def add_vulnerability_report( self, title: str, @@ -232,7 +270,11 @@ def add_vulnerability_report( agent_id: str | None = None, agent_name: str | None = None, ) -> str: - report_id = f"vuln-{len(self.vulnerability_reports) + 1:04d}" + ordinal = self._next_vuln_ordinal() + report_id = f"vuln-{ordinal:04d}" + # Persist the high-water mark so a retracted top id is never reused after + # a resume in a fresh process (run_record survives via run.json). + self.run_record["max_vuln_ordinal"] = ordinal report: dict[str, Any] = { "id": report_id, @@ -296,6 +338,72 @@ def add_vulnerability_report( self.save_run_data() return report_id + def retract_vulnerability_report(self, report_id: str, reason: str) -> dict[str, Any]: + """Remove a previously-reported finding from the cumulative set. + + On resume, prior findings are rehydrated from ``vulnerabilities.json``. + When a later push FIXES one, the agent re-validates and must be able to + DROP it — otherwise rehydration is append-only and the fixed finding + re-emits into every subsequent SARIF forever, producing a scanner alert + that can never auto-resolve (fail-CLOSED). This is the inverse of + :meth:`add_vulnerability_report`. + + Reverses add's side effects: + 1. drop the report from the in-memory ``vulnerability_reports`` list + 2. delete the stale per-finding ``{id}.md`` (the one append-only + artefact; CSV/JSON/SARIF are rewritten wholesale by + :meth:`save_run_data` below from the reduced set) + + The id is retained in ``_saved_vuln_ids`` (NOT recycled): combined with + :meth:`_next_vuln_ordinal` deriving from the max ordinal ever seen, a + physical removal can never cause a later finding to reuse a retracted + id. Idempotent: retracting an unknown/already-gone id is a no-op success. + + NOTE: the caller (the retract tool) applies the groundedness guard — + re-verifying against the current tree that the vulnerable sink is + actually gone — BEFORE invoking this. This method performs the state + mutation only; it does not decide whether the retraction is justified. + """ + reason = (reason or "").strip() + if not reason: + raise ValueError( + "retract_vulnerability_report requires a non-empty reason " + "(cite what was verified fixed, and where)", + ) + retracted = next( + (r for r in self.vulnerability_reports if r.get("id") == report_id), + None, + ) + if retracted is None: + logger.info("retract: %s not present (no-op)", report_id) + return {"success": True, "retracted": False, "reason": "id not present"} + + before = len(self.vulnerability_reports) + self.vulnerability_reports = [ + r for r in self.vulnerability_reports if r.get("id") != report_id + ] + # Deliberately keep report_id in _saved_vuln_ids so the ordinal counter + # never re-allocates it; delete the stale MD so a future rehydration + # doesn't reload the retracted finding from disk. + with contextlib.suppress(OSError): + md_path = self.get_run_dir() / "vulnerabilities" / f"{report_id}.md" + md_path.unlink(missing_ok=True) + + logger.info( + "Retracted vulnerability report %s (%s) — reason: %s", + report_id, + retracted.get("title", ""), + reason, + ) + self.save_run_data() + return { + "success": True, + "retracted": True, + "report_id": report_id, + "title": retracted.get("title"), + "remaining": before - 1, + } + def get_existing_vulnerabilities(self) -> list[dict[str, Any]]: return list(self.vulnerability_reports) diff --git a/strix/tools/reporting/retract_tool.py b/strix/tools/reporting/retract_tool.py new file mode 100644 index 000000000..837dff38a --- /dev/null +++ b/strix/tools/reporting/retract_tool.py @@ -0,0 +1,205 @@ +"""``retract_vulnerability_report`` — the inverse of create, for resumed scans. + +On ``--resume`` the agent rehydrates every prior finding. When a later push +FIXES one, the agent must be able to DROP it, or rehydration is append-only and +the fixed finding re-emits into every subsequent SARIF forever — an alert that +can never auto-resolve. This tool closes that gap. + +TRUST MODEL — the agent decides, the guard only annotates. Strix trusts the LLM +to FIND vulnerabilities, so it trusts the LLM's judgement that one is FIXED. The +retraction therefore ALWAYS proceeds on the agent's cited reason; the guard +never vetoes it. What the guard adds is an audit signal: it re-reads the +finding's fix-site (``fix_before``) sinks against the current tree and records a +verdict on the result, so a drop is explainable and recoverable — not a +decision-maker that can stubbornly block a genuine fix it merely can't +re-locate (moved/renamed file, dependency finding with no sink, transient read +failure). Recorded on the result: + +- ``grounded=True`` (``guard_verdict="gone"``): the sink was read and the + vulnerable line is gone (or the file is confirmed absent). The fix is + independently corroborated. +- ``grounded=False`` (``guard_verdict="inconclusive"``): couldn't corroborate — + no reader wired, no ``fix_before`` sink on the finding, or the sink file + couldn't be read. Retract still honoured on the agent's word. +- ``grounded=False`` (``guard_verdict="present"``): the ``fix_before`` line is + STILL present verbatim, which contradicts the "fixed" claim — logged as a loud + warning, but the agent may have legitimately re-reasoned (line now unreachable, + guarded, or the fixed form contains the substring), so the retract is honoured + and flagged rather than blocked. + +A wrong retraction is self-correcting: findings are re-derived on every scan of +the code, so a mistakenly-dropped-but-still-live finding re-appears next run. +That recoverability is what makes trusting the agent safe here. + +The reader is injected rather than hard-wired: the target lives inside the +sandbox in v1, so the concrete sandbox-backed reader is supplied by the caller +(the agent factory, which holds the session). Its contract: return file text if +read, ``None`` ONLY when the file is positively confirmed absent, and RAISE when +a read neither returns content nor confirms absence — so "couldn't read" is +scored ``inconclusive``, never mistaken for "confirmed gone". + +Only exposed to the agent on resumed runs, and only when +``STRIX_RESUME_RETRACT`` is set (see factory gating) — a fresh scan can't need +it, and resume stays append-only by default. +""" + +from __future__ import annotations + +import logging +from collections.abc import Awaitable, Callable +from typing import Any + +from agents import RunContextWrapper, function_tool + + +logger = logging.getLogger(__name__) + +# read_target_file(repo_relative_path) -> awaitable file contents, or None if +# absent/unreadable. Injected by the caller (holds the sandbox session), so the +# concrete reader can `session.exec("cat", ...)` against the live /workspace +# tree. None = no reader wired → the guard fails safe (refuse). +ReadTargetFileFn = Callable[[str], Awaitable["str | None"]] + +# Single-element list rather than a module global + ``global`` statement, so the +# reader can be swapped in place without PLW0603. +_reader: list[ReadTargetFileFn | None] = [None] + + +def set_target_file_reader(reader: ReadTargetFileFn | None) -> None: + """Wire the sandbox-backed target-file reader (called at agent build time on + resume). Passing None disables source grounding — retracts still proceed on + the agent's reason, scored ``inconclusive`` (ungrounded) rather than + ``gone``.""" + _reader[0] = reader + + +def _sink_locations(report: dict[str, Any]) -> list[dict[str, Any]]: + """Fix-site (sink) locations to re-verify: only those carrying a + ``fix_before`` (the vulnerable line to change). A plain context ``snippet`` + frequently survives a genuine fix unchanged, so matching on it would + false-refuse a real fix (real-canary lesson from the 0.8 impl).""" + return [ + loc for loc in (report.get("code_locations") or []) + if isinstance(loc, dict) and loc.get("fix_before") and loc.get("file") + ] + + +# Three-way guard verdict. The point is to separate "the agent is provably +# WRONG" (hard block) from "we can't independently confirm" (defer to the +# agent's judgement but flag it) — so the tool neither silently drops live +# findings NOR stubbornly refuses a genuine fix it just can't re-locate (moved +# file, no structured sink, transient read failure). +GUARD_PRESENT = "present" # sink confirmed still in source -> hard REFUSE +GUARD_GONE = "gone" # sink/file confirmed absent -> grounded ALLOW +GUARD_INCONCLUSIVE = "inconclusive" # couldn't verify -> allow on agent's word, flagged + + +async def _guard(report: dict[str, Any]) -> tuple[str, str]: + """Re-verify a retract against current source. Returns ``(verdict, detail)``. + + - ``GUARD_PRESENT``: a ``fix_before`` sink is still present verbatim — the + "it's fixed" claim is provably false. Hard refuse. + - ``GUARD_GONE``: every verifiable sink's file was read and the vulnerable + line is gone (or the file is confirmed absent). Grounded — allow. + - ``GUARD_INCONCLUSIVE``: nothing could be positively checked — no reader, + no ``fix_before`` sink (older/dependency findings), or the file couldn't + be read (moved/renamed/permission). NOT proof either way; the tool allows + on the agent's cited reason but records it as ungrounded so a wrong drop + is auditable and a later re-scan re-adds it. + """ + reader = _reader[0] + if reader is None: + return GUARD_INCONCLUSIVE, "no target-file reader wired — cannot ground the retract" + sinks = _sink_locations(report) + if not sinks: + return GUARD_INCONCLUSIVE, "no fix_before sink on this finding — nothing to verify" + + verified_gone = False + unread: list[str] = [] + for loc in sinks: + try: + result = await reader(str(loc["file"])) + except Exception as exc: # noqa: BLE001 — a failed read is not proof either way + unread.append(f"{loc['file']} ({exc!r})") + continue + needle = str(loc["fix_before"]).strip() + if result is None: + verified_gone = True # file confirmed absent → sink can't be present + continue + if needle and needle in result: + return GUARD_PRESENT, (f"vulnerable code still present at {loc['file']}: " + f"{needle[:80]!r}") + verified_gone = True # read the file, vulnerable line is gone + + if unread and not verified_gone: + # Every sink was unreadable (e.g. the file moved) — can't confirm a fix, + # but can't confirm it's still vulnerable either. + return GUARD_INCONCLUSIVE, f"could not read any sink to verify: {'; '.join(unread)}" + return GUARD_GONE, "vulnerable code no longer present at any readable sink" + + +async def _do_retract(report_id: str, reason: str) -> dict[str, Any]: + reason = (reason or "").strip() + if not reason: + return {"success": False, + "error": "reason is required — cite what was fixed and where"} + + from strix.report.state import get_global_report_state + state = get_global_report_state() + if state is None: + return {"success": False, "error": "no active report state"} + + report = next( + (r for r in state.get_existing_vulnerabilities() if r.get("id") == report_id), + None, + ) + if report is None: + return {"success": True, "retracted": False, "reason": "id not present"} + + # The guard NEVER overrides the agent: Strix trusts the LLM to find the + # vulnerability, so it trusts the LLM's judgement that it's fixed. The guard + # is an audit signal, not a veto — it records whether the drop could be + # grounded in source and warns loudly when the agent's "fixed" claim looks + # contradicted, but the retract always proceeds on the agent's cited reason. + # A wrong drop re-appears on the next scan of the same code (findings are + # re-derived every run), so this is recoverable, not silent data loss. + verdict, detail = await _guard(report) + grounded = verdict == GUARD_GONE + if verdict == GUARD_PRESENT: + logger.warning( + "retract for %s: guard says vulnerable sink still present (%s), but " + "honouring agent reason: %s", + report_id, detail, reason, + ) + elif not grounded: + logger.info("retract for %s ungrounded (%s) — honouring agent reason", report_id, detail) + + try: + result = state.retract_vulnerability_report(report_id, reason) + except (ValueError, AttributeError) as e: + return {"success": False, "error": f"retract failed: {e!s}"} + logger.info("retract honoured for %s (grounded=%s, guard: %s)", report_id, grounded, detail) + result["guard"] = detail + result["grounded"] = grounded + result["guard_verdict"] = verdict + return result + + +@function_tool(timeout=120, strict_mode=False) +async def retract_vulnerability_report( + ctx: RunContextWrapper, + report_id: str, + reason: str, +) -> dict[str, Any]: + """Retract a prior finding you have determined is FIXED in the current code. + + Use on a resumed scan when a finding rehydrated from a previous run is no + longer valid — the vulnerable code has been fixed or removed since it was + reported. It is dropped from the cumulative report and SARIF so the resolved + finding stops re-appearing. Cite what changed and where in ``reason``. + + Args: + report_id: the finding id to retract (e.g. ``vuln-0003``). + reason: what was fixed, and where. + """ + return await _do_retract(report_id, reason) diff --git a/tests/test_retract_vulnerability.py b/tests/test_retract_vulnerability.py new file mode 100644 index 000000000..19413172c --- /dev/null +++ b/tests/test_retract_vulnerability.py @@ -0,0 +1,190 @@ +"""Tests for resume-time vulnerability retraction + id stability under removal.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest + +from strix.report.state import ReportState, set_global_report_state +from strix.tools.reporting import retract_tool + + +if TYPE_CHECKING: + from pathlib import Path + + +@pytest.fixture +def report_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> ReportState: + monkeypatch.chdir(tmp_path) + state = ReportState(run_name="test-run") + set_global_report_state(state) + return state + + +def _add(state: ReportState, title: str) -> str: + return state.add_vulnerability_report(title=title, severity="high") + + +# --- ReportState.retract_vulnerability_report + id stability ----------------- + + +def test_retract_removes_from_cumulative_set(report_state: ReportState) -> None: + a, b, c = (_add(report_state, t) for t in ("SQLi", "CmdInj", "PathTrav")) + result = report_state.retract_vulnerability_report(b, reason="parameterised, verified fixed") + assert result["retracted"] is True + assert result["remaining"] == 2 + ids = [r["id"] for r in report_state.vulnerability_reports] + assert ids == [a, c] + assert b not in ids + + +def test_retract_is_idempotent_on_missing_id(report_state: ReportState) -> None: + _add(report_state, "SQLi") + result = report_state.retract_vulnerability_report("vuln-9999", reason="already gone") + assert result["success"] is True + assert result["retracted"] is False + + +def test_retract_requires_reason(report_state: ReportState) -> None: + vid = _add(report_state, "SQLi") + with pytest.raises(ValueError, match="non-empty reason"): + report_state.retract_vulnerability_report(vid, reason=" ") + + +def test_ids_are_stable_under_removal(report_state: ReportState) -> None: + """A retracted ordinal must never be recycled — otherwise a later finding + reuses a retracted id and collides with its on-disk MD / SARIF fingerprint. + """ + _add(report_state, "SQLi") # vuln-0001 + b = _add(report_state, "CmdInj") # vuln-0002 + c = _add(report_state, "PathTrav") # vuln-0003 + assert (b, c) == ("vuln-0002", "vuln-0003") + + report_state.retract_vulnerability_report(b, reason="fixed") + + # Next add must be vuln-0004, NOT a recycled 0002/0003. + assert _add(report_state, "SSRF") == "vuln-0004" + + +def test_retracted_md_is_deleted(report_state: ReportState) -> None: + vid = _add(report_state, "SQLi") + md = report_state.get_run_dir() / "vulnerabilities" / f"{vid}.md" + assert md.exists() + report_state.retract_vulnerability_report(vid, reason="fixed") + assert not md.exists() + + +# --- retract tool groundedness guard ----------------------------------------- +# NOTE: the ``os.system(...)`` / ``subprocess`` strings below are INERT test +# data — literal file-content and fix_before needles fed to the guard's text +# substring check. Nothing is executed; no shell is invoked. + + +async def _guard(report: dict[str, object]) -> tuple[str, str]: + return await retract_tool._guard(report) + + +_SINK = {"code_locations": [{"file": "a.py", "fix_before": "os.system(x)"}]} + + +async def test_guard_inconclusive_without_reader() -> None: + retract_tool.set_target_file_reader(None) + verdict, detail = await _guard(_SINK) + assert verdict == retract_tool.GUARD_INCONCLUSIVE + assert "no target-file reader" in detail + + +async def test_guard_present_when_sink_still_there() -> None: + async def reader(_path: str) -> str: + return "def f(x):\n os.system(x)\n" # vulnerable line still present + + retract_tool.set_target_file_reader(reader) + verdict, detail = await _guard(_SINK) + assert verdict == retract_tool.GUARD_PRESENT + assert "still present" in detail + + +async def test_guard_gone_when_sink_removed() -> None: + async def reader(_path: str) -> str: + return "def f(x):\n subprocess.run(['echo', x])\n" # fixed + + retract_tool.set_target_file_reader(reader) + verdict, _ = await _guard(_SINK) + assert verdict == retract_tool.GUARD_GONE + + +async def test_guard_gone_when_file_confirmed_absent() -> None: + async def reader(_path: str) -> None: + return None # reader confirms the file is gone + + retract_tool.set_target_file_reader(reader) + verdict, _ = await _guard(_SINK) + assert verdict == retract_tool.GUARD_GONE + + +async def test_guard_inconclusive_when_no_fix_before() -> None: + async def reader(_path: str) -> str: + return "irrelevant" + + retract_tool.set_target_file_reader(reader) + verdict, _ = await _guard({"code_locations": [{"file": "a.py", "snippet": "context only"}]}) + assert verdict == retract_tool.GUARD_INCONCLUSIVE + + +async def test_guard_inconclusive_when_read_raises() -> None: + async def reader(_path: str) -> str: + raise OSError("could not read — file moved") + + retract_tool.set_target_file_reader(reader) + # A failed read (e.g. moved/renamed file) is NOT proof of a fix, but also + # not proof it's still vulnerable → inconclusive, not present. + verdict, _ = await _guard(_SINK) + assert verdict == retract_tool.GUARD_INCONCLUSIVE + + +# --- tool always retracts (trust model) + records grounding ------------------ + + +async def test_tool_retracts_and_marks_grounded_when_gone(report_state: ReportState) -> None: + vid = report_state.add_vulnerability_report( + title="CmdInj", severity="high", + code_locations=[{"file": "a.py", "fix_before": "os.system(x)"}], + ) + + async def reader(_path: str) -> str: + return "subprocess.run(['echo', x])\n" # fixed + + retract_tool.set_target_file_reader(reader) + res = await retract_tool._do_retract(vid, reason="rewrote to subprocess argv") + assert res["retracted"] is True + assert res["grounded"] is True + assert vid not in [r["id"] for r in report_state.vulnerability_reports] + + +async def test_tool_still_retracts_when_sink_present_but_flags_ungrounded( + report_state: ReportState, +) -> None: + # Guard NEVER vetoes: even when the fix_before line is still present, the + # agent's judgement wins — the drop proceeds but is flagged grounded=False. + vid = report_state.add_vulnerability_report( + title="CmdInj", severity="high", + code_locations=[{"file": "a.py", "fix_before": "os.system(x)"}], + ) + + async def reader(_path: str) -> str: + return "os.system(x) # now behind an unreachable guard\n" + + retract_tool.set_target_file_reader(reader) + res = await retract_tool._do_retract(vid, reason="line now unreachable — dead branch") + assert res["retracted"] is True + assert res["grounded"] is False + assert res["guard_verdict"] == retract_tool.GUARD_PRESENT + assert vid not in [r["id"] for r in report_state.vulnerability_reports] + + +async def test_tool_requires_reason(report_state: ReportState) -> None: + vid = report_state.add_vulnerability_report(title="X", severity="high") + res = await retract_tool._do_retract(vid, reason=" ") + assert res["success"] is False + assert vid in [r["id"] for r in report_state.vulnerability_reports]