fix(reporting): re-anchor code-location line numbers to the verbatim block

Agents report start_line/end_line from memory and routinely land tens of
lines off while copying the vulnerable code verbatim. Since fix_before
(falling back to snippet) is a byte-faithful copy, locate that block in
the checked-out file under /workspace and correct the range before the
locations are validated and stored. Locations whose anchor is not found
keep their reported numbers. Applies on both create_vulnerability_report
and update_vulnerability_report.
This commit is contained in:
yoni 2026-09-25 02:34:45 +00:00
parent ae38fe70cd
commit 303f5dec83
2 changed files with 208 additions and 2 deletions

View file

@ -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,108 @@ def _validate_code_locations(locations: list[dict[str, Any]]) -> list[str]:
return errors
def _repo_workspace_roots() -> list[Path]:
"""Workspace directories that hold checked-out repo / source targets.
Sandbox targets land at ``/workspace/<workspace_subdir>``; ``/workspace``
itself covers root-level mounts. Best-effort: any lookup failure yields the
bare ``/workspace`` root.
"""
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}")
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 to multi-line anchors or single lines found
exactly once (a lone generic line like ``}`` matches too many places to be
trusted otherwise).
"""
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()
if not anchor or not 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")
best: tuple[int, int] | None = None
for root in roots:
try:
text = (root / rel).read_text(encoding="utf-8", errors="replace")
except OSError:
continue
found = _find_anchor(text.splitlines(), anchor_lines, reported)
if found is not None and (
best is None or abs(found - reported) < abs(best[0] - reported)
):
best = (found, found + len(anchor_lines) - 1)
if best is None or 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 +546,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 +944,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)

View file

@ -1169,6 +1169,108 @@ 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 = ["x = 1"] * 20
anchor = ["x = 1"]
assert reporting_tool._find_anchor(file_lines, anchor, 12) == 12
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) == 1
file_lines = [" }", " }"]
assert reporting_tool._find_anchor(file_lines, ["}"], 1) is None
def test_vuln_tool_exposes_fix_verification() -> None:
assert "fix_verification" in create_vulnerability_report.params_json_schema["properties"]