fix(reporting): harden code-location re-anchoring for host runs and ambiguity

The engine runs on the host for OSS Docker scans, where /workspace does
not exist: include cloned_repo_path, local-code target_path and the
workspace mount as roots so re-anchoring works outside the sandbox.

Skip locations whose file fails path validation before any read so
absolute paths (e.g. /dev/zero) can never hang reporting, resolve
candidates and require they stay inside the root (symlink escapes), and
cap buffered files at 4 MiB.

Require the anchor to resolve to one position across every root instead
of picking the globally-nearest hit, which could store another repo's
range in a multi-target scan; apply the single-line uniqueness rule to
exact matches too so a lone repeated line keeps the reported range.

strix/tools/reporting/tool.py

@@ -11,8 +11,8 @@ import asyncio
 import json
 import logging
 import re
-from pathlib import PurePosixPath
-from typing import TYPE_CHECKING, Any
+from pathlib import Path, PurePosixPath
+from typing import TYPE_CHECKING, Any, cast
 
 from agents import RunContextWrapper, function_tool
 
@@ -110,6 +110,145 @@ def _validate_code_locations(locations: list[dict[str, Any]]) -> list[str]:
     return errors
 
 
+# Source files a location would point at are at most a few hundred KB; never
+# buffer anything larger (e.g. ``/dev/zero``-like reads or binary blobs).
+_MAX_ANCHOR_FILE_BYTES = 4 * 1024 * 1024
+
+
+def _repo_workspace_roots() -> list[Path]:
+    """Directories under which a location's repo-relative ``file`` may resolve.
+
+    The agent sees checkouts at ``/workspace/<workspace_subdir>``, but this
+    tool runs wherever the engine does: inside the sandbox for embedded runs,
+    on the host for normal OSS Docker scans. ``/workspace`` only exists in
+    the former, so host-side checkouts are added too — ``cloned_repo_path``
+    for cloned repositories, ``target_path`` for live local-code mounts, and
+    the workspace mount. Best-effort: roots that do not exist on this
+    filesystem simply fail to read.
+    """
+    roots = ["/workspace"]
+    try:
+        from strix.report.state import get_global_report_state
+
+        state = get_global_report_state()
+        config: dict[str, Any] = (state.scan_config if state else None) or {}
+        for target in cast("list[dict[str, Any]]", config.get("targets") or []):
+            details: dict[str, Any] = target.get("details") or {}
+            subdir = str(details.get("workspace_subdir") or "").strip("/")
+            if subdir:
+                roots.append(f"/workspace/{subdir}")
+            for key in ("cloned_repo_path", "target_path"):
+                host_path = str(details.get(key) or "").strip()
+                if host_path:
+                    roots.append(host_path)
+        mount = str(config.get("workspace_mount") or "").strip()
+        if mount:
+            roots.append(mount)
+    except Exception:  # noqa: BLE001 - opportunistic; never block a report on it
+        logger.debug("Could not resolve workspace roots for code-location re-anchoring")
+    return [Path(root) for root in dict.fromkeys(roots)]
+
+
+def _find_anchor(file_lines: list[str], anchor_lines: list[str], reported: int) -> int | None:
+    """1-based line where ``anchor_lines`` sits in ``file_lines``, nearest ``reported``.
+
+    Verbatim blocks (``fix_before``/``snippet``) are far more reliable than the
+    agent's reported line numbers, so locating the block in the real file is
+    the better source of truth. Trailing-whitespace differences are tolerated;
+    a dedented fallback applies when the exact block is not found (agents
+    routinely shift leading whitespace when quoting code). A lone generic
+    line (``}``, ``import os``) can match many places, so a single-line
+    anchor only re-anchors when it appears exactly once in the file; a
+    repeated multi-line block keeps the match nearest the reported line and
+    unmatched blocks keep the reported range.
+    """
+    count = len(anchor_lines)
+    if count == 0 or count > len(file_lines):
+        return None
+    wanted = [ln.rstrip() for ln in anchor_lines]
+    hits = [
+        i
+        for i in range(len(file_lines) - count + 1)
+        if [fl.rstrip() for fl in file_lines[i : i + count]] == wanted
+    ]
+    if not hits:
+        dedented = [ln.lstrip() for ln in anchor_lines]
+        hits = [
+            i
+            for i in range(len(file_lines) - count + 1)
+            if [fl.lstrip() for fl in file_lines[i : i + count]] == dedented
+        ]
+    if count == 1 and len(hits) != 1:
+        return None
+    if not hits:
+        return None
+    return min(hits, key=lambda i: (abs(i + 1 - reported), i)) + 1
+
+
+def _reanchor_code_locations(locations: list[dict[str, Any]] | None) -> None:
+    """Correct ``start_line``/``end_line`` against the checked-out files.
+
+    Agents report line numbers from memory and routinely get them wrong while
+    copying the vulnerable code verbatim. When a location carries such a block
+    (``fix_before``, else ``snippet``), locating it in the real file yields a
+    trustworthy range; locations whose anchor is not found keep their reported
+    numbers.
+    """
+    if not locations:
+        return
+    roots: list[Path] | None = None
+    for loc in locations:
+        anchor = loc.get("fix_before") or loc.get("snippet")
+        rel = str(loc.get("file") or "").strip()
+        # Locations failing path validation are rejected downstream anyway;
+        # skip here so absolute paths and traversals never reach a read.
+        if not anchor or not rel or _validate_file_path(rel):
+            continue
+        if roots is None:
+            roots = _repo_workspace_roots()
+        reported = loc.get("start_line")
+        reported = reported if isinstance(reported, int) and reported > 0 else 1
+        anchor_lines = anchor.split("\n")
+        found_lines: set[int] = set()
+        for root in roots:
+            try:
+                resolved_root = root.resolve()
+                candidate = (resolved_root / rel).resolve()
+                # A symlink inside a checkout must not escape the root, and
+                # oversized files are not source worth buffering.
+                if (
+                    not candidate.is_relative_to(resolved_root)
+                    or candidate.stat().st_size > _MAX_ANCHOR_FILE_BYTES
+                ):
+                    continue
+                text = candidate.read_text(encoding="utf-8", errors="replace")
+            except OSError:
+                continue
+            found = _find_anchor(text.splitlines(), anchor_lines, reported)
+            if found is not None:
+                found_lines.add(found)
+        # The anchor must resolve to ONE position across every root. In a
+        # multi-target scan two repos can hold the same relative path and
+        # anchor text; picking the "nearest" hit there could store the other
+        # repo's range, so an ambiguous or absent match keeps the reported
+        # lines instead.
+        if len(found_lines) != 1:
+            continue
+        start = found_lines.pop()
+        best = (start, start + len(anchor_lines) - 1)
+        if best == (loc.get("start_line"), loc.get("end_line")):
+            continue
+        logger.info(
+            "Re-anchored %s lines %s-%s -> %s-%s",
+            rel,
+            loc.get("start_line"),
+            loc.get("end_line"),
+            best[0],
+            best[1],
+        )
+        loc["start_line"], loc["end_line"] = best
+
+
 def _extract_cve(cve: str) -> str:
     match = re.search(r"CVE-\d{4}-\d{4,}", cve)
     return match.group(0) if match else cve.strip()
@@ -444,6 +583,7 @@ def _collect_update_changes(  # noqa: PLR0912, PLR0915
     raw_locations = fields.get("code_locations")
     locations = _normalize_code_locations(raw_locations)
     if locations:
+        _reanchor_code_locations(locations)
         errors.extend(_validate_code_locations(locations))
         errors.extend(_validate_fix_verification(locations, changes.get("fix_verification")))
         changes["code_locations"] = locations
@@ -841,6 +981,7 @@ async def _do_create(
 
     parsed_locations = _normalize_code_locations(code_locations)
     if parsed_locations:
+        _reanchor_code_locations(parsed_locations)
         errors.extend(_validate_code_locations(parsed_locations))
     errors.extend(_validate_fix_verification(parsed_locations, fix_verification))
     cve, cwe, identifier_errors = _validate_identifiers(cve, cwe)

tests/test_reporting_fields.py

@@ -1169,6 +1169,236 @@ async def test_informational_location_needs_no_verification(report_state: Report
     assert "fix_verification" not in report_state.vulnerability_reports[0]
 
 
+async def test_code_locations_reanchored_to_verbatim_block(
+    report_state: ReportState, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Agents report line numbers from memory; the verbatim block is the
+    trustworthy anchor, so mis-reported ranges are corrected against the
+    checked-out file."""
+    source = tmp_path / "server" / "routes" / "index.js"
+    source.parent.mkdir(parents=True)
+    source.write_text(
+        "const a = 1\n"
+        "const b = 2\n"
+        "            let order = {\n"
+        "                residentId: req.body.residentId,\n"
+        "                numCredits: req.body.numCredits,\n"
+        "                price: req.body.price\n"
+        "            }\n"
+    )
+    monkeypatch.setattr(reporting_tool, "_repo_workspace_roots", lambda: [tmp_path])
+    result = await _create_with(
+        report_state,
+        code_locations=[
+            {
+                "file": "server/routes/index.js",
+                "start_line": 40,
+                "end_line": 46,
+                "fix_before": (
+                    "            let order = {\n"
+                    "                residentId: req.body.residentId,\n"
+                    "                numCredits: req.body.numCredits,\n"
+                    "                price: req.body.price\n"
+                    "            }"
+                ),
+            }
+        ],
+    )
+    assert result["success"] is True
+    loc = report_state.vulnerability_reports[0]["code_locations"][0]
+    assert loc["start_line"] == 3
+    assert loc["end_line"] == 7
+
+
+async def test_code_locations_kept_when_anchor_not_found(
+    report_state: ReportState, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    source = tmp_path / "app" / "views.py"
+    source.parent.mkdir(parents=True)
+    source.write_text("def show():\n    return 1\n")
+    monkeypatch.setattr(reporting_tool, "_repo_workspace_roots", lambda: [tmp_path])
+    result = await _create_with(report_state, code_locations=[_INFO_LOCATION])
+    assert result["success"] is True
+    loc = report_state.vulnerability_reports[0]["code_locations"][0]
+    assert loc["start_line"] == _INFO_LOCATION["start_line"]
+    assert loc["end_line"] == _INFO_LOCATION["end_line"]
+
+
+async def test_code_locations_reanchored_on_update(
+    report_state: ReportState, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    result = await _create_with(report_state)
+    report_id = result["report_id"]
+    source = tmp_path / "app" / "views.py"
+    source.parent.mkdir(parents=True)
+    source.write_text("alpha\nbeta\ngamma\ndelta\n")
+    monkeypatch.setattr(reporting_tool, "_repo_workspace_roots", lambda: [tmp_path])
+    update = _do_update(
+        report_id=report_id,
+        update_reason="Pinpoint the vulnerable call.",
+        fields={
+            "code_locations": [
+                {
+                    "file": "app/views.py",
+                    "start_line": 99,
+                    "snippet": "beta",
+                }
+            ]
+        },
+    )
+    assert update["success"] is True
+    loc = report_state.vulnerability_reports[0]["code_locations"][0]
+    assert loc["start_line"] == 2
+    assert loc["end_line"] == 2
+
+
+def test_find_anchor_picks_nearest_match() -> None:
+    file_lines = ["a = 1", "b = 2"] * 20
+    anchor = ["a = 1", "b = 2"]
+    assert reporting_tool._find_anchor(file_lines, anchor, 23) == 23
+
+
+def test_find_anchor_dedented_multiline() -> None:
+    file_lines = ["def f():", "    inner()", "    other()", "end"]
+    anchor = ["inner()", "other()"]
+    assert reporting_tool._find_anchor(file_lines, anchor, 99) == 2
+
+
+def test_find_anchor_single_line_ambiguous_dedented() -> None:
+    file_lines = ["}", "    }", "}"]
+    assert reporting_tool._find_anchor(file_lines, ["}"], 1) is None
+    file_lines = ["    }", "    }"]
+    assert reporting_tool._find_anchor(file_lines, ["}"], 1) is None
+    file_lines = ["x()", "    }", "end"]
+    assert reporting_tool._find_anchor(file_lines, ["}"], 1) == 2
+
+
+def test_code_locations_absolute_path_never_read(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """An absolute ``file`` fails path validation downstream; re-anchoring
+    must skip it before any read so ``/dev/zero``-style paths can't hang
+    report creation."""
+    monkeypatch.setattr(reporting_tool, "_repo_workspace_roots", lambda: [tmp_path])
+    location = {
+        "file": "/dev/zero",
+        "start_line": 40,
+        "end_line": 42,
+        "snippet": "const x = 1",
+    }
+    reporting_tool._reanchor_code_locations([location])
+    assert location["start_line"] == 40
+    assert location["end_line"] == 42
+
+
+def test_code_locations_symlink_escape_skipped(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A symlink inside a checkout pointing outside the root is not followed."""
+    outside = tmp_path / "outside.txt"
+    outside.write_text("secret\nconst x = 1\n")
+    root = tmp_path / "root"
+    root.mkdir()
+    (root / "link.ts").symlink_to(outside)
+    monkeypatch.setattr(reporting_tool, "_repo_workspace_roots", lambda: [root])
+    location = {
+        "file": "link.ts",
+        "start_line": 40,
+        "end_line": 40,
+        "snippet": "const x = 1",
+    }
+    reporting_tool._reanchor_code_locations([location])
+    assert location["start_line"] == 40
+
+
+def test_code_locations_ambiguous_across_roots_kept(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Multi-target scan: the same relative path plus anchor in two repos is
+    ambiguous — the reported lines stay rather than picking the wrong repo."""
+    cases = (("repo-a", "const x = 1\nalpha\nbeta\n"), ("repo-b", "alpha\nbeta\nconst x = 1\n"))
+    for name, lines in cases:
+        source = tmp_path / name / "src" / "index.ts"
+        source.parent.mkdir(parents=True)
+        source.write_text(lines)
+    roots = [tmp_path / "repo-a", tmp_path / "repo-b"]
+    monkeypatch.setattr(reporting_tool, "_repo_workspace_roots", lambda: roots)
+    location = {
+        "file": "src/index.ts",
+        "start_line": 40,
+        "end_line": 40,
+        "snippet": "const x = 1",
+    }
+    reporting_tool._reanchor_code_locations([location])
+    assert location["start_line"] == 40
+
+
+def test_code_locations_single_root_match_reanchors(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The anchor existing in exactly one root disambiguates the repo."""
+    present = tmp_path / "repo-b" / "src" / "index.ts"
+    present.parent.mkdir(parents=True)
+    present.write_text("alpha\nbeta\nconst x = 1\n")
+    roots = [tmp_path / "repo-a", tmp_path / "repo-b"]
+    monkeypatch.setattr(reporting_tool, "_repo_workspace_roots", lambda: roots)
+    location = {
+        "file": "src/index.ts",
+        "start_line": 40,
+        "end_line": 40,
+        "snippet": "const x = 1",
+    }
+    reporting_tool._reanchor_code_locations([location])
+    assert location["start_line"] == 3
+    assert location["end_line"] == 3
+
+
+def test_code_locations_oversized_file_skipped(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    source = tmp_path / "big.ts"
+    source.write_text("x" * (reporting_tool._MAX_ANCHOR_FILE_BYTES + 1))
+    monkeypatch.setattr(reporting_tool, "_repo_workspace_roots", lambda: [tmp_path])
+    location = {
+        "file": "big.ts",
+        "start_line": 40,
+        "end_line": 40,
+        "snippet": "const x = 1",
+    }
+    reporting_tool._reanchor_code_locations([location])
+    assert location["start_line"] == 40
+
+
+def test_repo_workspace_roots_include_host_checkouts(report_state: ReportState) -> None:
+    """OSS scans run the engine host-side where ``/workspace`` does not
+    exist; cloned and live-mounted repos resolve via their host paths."""
+    report_state.scan_config = {
+        "targets": [
+            {
+                "type": "repository",
+                "details": {
+                    "workspace_subdir": "repo-a",
+                    "cloned_repo_path": "/host/runs/x/repo-a",
+                },
+            },
+            {
+                "type": "local_code",
+                "details": {"workspace_subdir": "repo-b", "target_path": "/home/user/repo-b"},
+            },
+        ],
+        "workspace_mount": "/home/user/work",
+    }
+    roots = [str(p) for p in reporting_tool._repo_workspace_roots()]
+    assert roots == [
+        "/workspace",
+        "/workspace/repo-a",
+        "/host/runs/x/repo-a",
+        "/workspace/repo-b",
+        "/home/user/repo-b",
+        "/home/user/work",
+    ]
+
+
 def test_vuln_tool_exposes_fix_verification() -> None:
     assert "fix_verification" in create_vulnerability_report.params_json_schema["properties"]