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.
This commit is contained in:
yoni 2026-09-25 05:34:26 +00:00
parent 303f5dec83
commit c6f39fb835
2 changed files with 187 additions and 22 deletions

View file

@ -110,12 +110,21 @@ 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.
# 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
Sandbox targets land at ``/workspace/<workspace_subdir>``; ``/workspace``
itself covers root-level mounts. Best-effort: any lookup failure yields the
bare ``/workspace`` root.
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:
@ -128,6 +137,13 @@ def _repo_workspace_roots() -> list[Path]:
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)]
@ -139,9 +155,12 @@ def _find_anchor(file_lines: list[str], anchor_lines: list[str], reported: int)
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).
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):
@ -159,8 +178,8 @@ def _find_anchor(file_lines: list[str], anchor_lines: list[str], reported: int)
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 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
@ -181,25 +200,43 @@ def _reanchor_code_locations(locations: list[dict[str, Any]] | 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:
# 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")
best: tuple[int, int] | None = None
found_lines: set[int] = set()
for root in roots:
try:
text = (root / rel).read_text(encoding="utf-8", errors="replace")
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 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")):
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",

View file

@ -1253,9 +1253,9 @@ async def test_code_locations_reanchored_on_update(
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
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:
@ -1266,9 +1266,137 @@ def test_find_anchor_dedented_multiline() -> None:
def test_find_anchor_single_line_ambiguous_dedented() -> None:
file_lines = ["}", " }", "}"]
assert reporting_tool._find_anchor(file_lines, ["}"], 1) == 1
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: