From 84e83884d90a9c0a76648ac2fd66c879e2582829 Mon Sep 17 00:00:00 2001 From: bearsyankees Date: Thu, 17 Sep 2026 14:43:29 -0400 Subject: [PATCH] Bound report history enrichment and require unambiguous repository identity --- strix/core/repository_history.py | 15 +++- strix/report/history.py | 21 ++++- tests/test_report_history.py | 140 ++++++++++++++++++++++++++++++- tests/test_repository_history.py | 19 +++++ 4 files changed, 185 insertions(+), 10 deletions(-) diff --git a/strix/core/repository_history.py b/strix/core/repository_history.py index a721c951..45855097 100644 --- a/strix/core/repository_history.py +++ b/strix/core/repository_history.py @@ -49,14 +49,23 @@ def _parse_blame(output: str) -> BlameInfo | None: return None -def blame_line(repository: Path, file_path: object, line: int) -> BlameInfo | None: +def blame_line( + repository: Path, file_path: object, line: int, *, timeout: float = 3.0 +) -> BlameInfo | None: """Attribute one working-tree line, or return None when history is unavailable. Paths are relative to ``repository``. Git follows committed renames itself; missing paths, binary files and uncommitted lines are intentionally omitted. Lookups never fetch history and are bounded so enrichment stays optional. + Callers may supply a shorter timeout to share a budget across lookups. """ - if type(line) is not int or line < 1 or not isinstance(file_path, str) or not file_path: + if ( + type(line) is not int + or line < 1 + or not isinstance(file_path, str) + or not file_path + or timeout <= 0 + ): return None try: relative = Path(file_path) @@ -87,7 +96,7 @@ def blame_line(repository: Path, file_path: object, line: int) -> BlameInfo | No encoding="utf-8", errors="replace", check=False, - timeout=3, + timeout=min(timeout, 3.0), env={**os.environ, "GIT_NO_LAZY_FETCH": "1"}, ) except (OSError, ValueError, RuntimeError, subprocess.SubprocessError): diff --git a/strix/report/history.py b/strix/report/history.py index 7104d799..2c3a398e 100644 --- a/strix/report/history.py +++ b/strix/report/history.py @@ -6,6 +6,7 @@ import html import logging import re from pathlib import Path +from time import monotonic from typing import Any from strix.core.repository_history import blame_line @@ -14,6 +15,7 @@ from strix.core.repository_history import blame_line logger = logging.getLogger(__name__) _START = "" _END = "" +_ENRICHMENT_TIMEOUT = 3.0 def _text(value: str) -> str: @@ -48,7 +50,10 @@ def _repositories(run_record: dict[str, Any], target: str | None) -> list[Path]: f"/workspace/{details.get('workspace_subdir')}", ): matched.add(root) - return sorted(matched or roots) + # File existence cannot identify the repository for an unmatched target. + # Fall back only when the scan itself has exactly one possible checkout. + selected = matched or roots + return sorted(selected) if len(selected) == 1 else [] def _location_line(location: dict[str, Any]) -> int | None: @@ -65,11 +70,12 @@ def _location_line(location: dict[str, Any]) -> int | None: def enrich_report(report: dict[str, Any], run_record: dict[str, Any]) -> None: """Refresh blame for the current locations, never failing a report operation. - No new checkout is made. Ambiguous paths across multiple repositories are - omitted unless the finding's target identifies its repository. Markers let + No new checkout is made. Multiple repositories require a uniquely matching + target. A shared time budget bounds all lookups in a report. Markers let revisions and resumed scans replace only the generated part of the analysis. """ try: + deadline = monotonic() + _ENRICHMENT_TIMEOUT analysis = re.sub( re.escape(_START) + r".*?" + re.escape(_END), "", @@ -82,9 +88,13 @@ def enrich_report(report: dict[str, Any], run_record: dict[str, Any]) -> None: if not locations: return roots = _repositories(run_record, report.get("target")) + if not roots: + return entries: list[str] = [] seen: set[tuple[str, int]] = set() for location in locations: + if monotonic() >= deadline: + break file_path = location.get("file") line = _location_line(location) if not isinstance(file_path, str) or line is None or (file_path, line) in seen: @@ -93,7 +103,10 @@ def enrich_report(report: dict[str, Any], run_record: dict[str, Any]) -> None: candidates = [root for root in roots if (root / file_path).is_file()] if len(candidates) != 1: continue - info = blame_line(candidates[0], file_path, line) + remaining = deadline - monotonic() + if remaining <= 0: + break + info = blame_line(candidates[0], file_path, line, timeout=remaining) if info is None: continue entries.append( diff --git a/tests/test_report_history.py b/tests/test_report_history.py index 9183fbbb..194358ca 100644 --- a/tests/test_report_history.py +++ b/tests/test_report_history.py @@ -10,6 +10,8 @@ from typing import TYPE_CHECKING, Any import pytest from agents.tool_context import ToolContext +from strix.core.repository_history import BlameInfo +from strix.report import history from strix.report.history import enrich_report from strix.report.state import ReportState, set_global_report_state from strix.tools.reporting.tool import create_vulnerability_report, update_vulnerability_report @@ -99,7 +101,9 @@ def history_run( set_global_report_state(None) -async def _create(location: dict[str, Any]) -> dict[str, Any]: +async def _create( + location: dict[str, Any], *additional_locations: dict[str, Any] +) -> dict[str, Any]: arguments = { "title": "SQL injection in the query handler", "description": "The query handler executes unsanitized input.", @@ -116,7 +120,7 @@ async def _create(location: dict[str, Any]) -> dict[str, Any]: "severity_change_conditions": "A read-only database role would limit the impact.", "fix_effort": "low", "cvss_breakdown": _CVSS, - "code_locations": [location], + "code_locations": [location, *additional_locations], } context = ToolContext( context={"agent_id": "root"}, @@ -246,7 +250,7 @@ async def test_unexpected_history_failure_does_not_block_reporting( ) -> None: state, _clone, _commits = history_run - def broken_blame(*_args: Any) -> None: + def broken_blame(*_args: Any, **_kwargs: Any) -> None: raise RuntimeError("history lookup failed") monkeypatch.setattr("strix.report.history.blame_line", broken_blame) @@ -282,3 +286,133 @@ def test_ambiguous_repository_requires_matching_target_and_target_updates_refres assert "Other Author" in report["technical_analysis"] assert commits["first"] not in report["technical_analysis"] assert report["technical_analysis"].count("Last modified by") == 1 + + +@pytest.mark.parametrize("target", [None, "https://example.test/query", "query handler"]) +def test_unmatched_target_cannot_select_repository_by_unique_file( + history_run: tuple[ReportState, Path, dict[str, str]], tmp_path: Path, target: str | None +) -> None: + state, _clone, _commits = history_run + other = tmp_path / "other" + other.mkdir() + state.run_record["local_sources"].append( + {"source_path": str(other), "workspace_subdir": "other"} + ) + report = { + "target": target, + "technical_analysis": _ANALYSIS, + "code_locations": [{"file": "app.py", "start_line": 1, "end_line": 1}], + } + + enrich_report(report, state.run_record) + + assert report["technical_analysis"] == _ANALYSIS + + +@pytest.mark.parametrize("target", ["application", "/workspace/application"]) +def test_colliding_workspace_alias_cannot_select_repository_by_unique_file( + history_run: tuple[ReportState, Path, dict[str, str]], tmp_path: Path, target: str +) -> None: + state, _clone, _commits = history_run + other = tmp_path / "other" + other.mkdir() + state.run_record["local_sources"].append( + {"source_path": str(other), "workspace_subdir": "application"} + ) + report = { + "target": target, + "technical_analysis": _ANALYSIS, + "code_locations": [{"file": "app.py", "start_line": 1, "end_line": 1}], + } + + enrich_report(report, state.run_record) + + assert report["technical_analysis"] == _ANALYSIS + + +def test_unmatched_target_can_use_only_configured_repository( + history_run: tuple[ReportState, Path, dict[str, str]], +) -> None: + state, _clone, commits = history_run + report = { + "target": "https://example.test/query", + "technical_analysis": _ANALYSIS, + "code_locations": [{"file": "app.py", "start_line": 1, "end_line": 1}], + } + + enrich_report(report, state.run_record) + + assert commits["first"] in report["technical_analysis"] + + +@pytest.mark.parametrize("first_lookup_succeeds", [True, False]) +async def test_history_budget_is_shared_and_partial_results_are_persisted( + history_run: tuple[ReportState, Path, dict[str, str]], + monkeypatch: pytest.MonkeyPatch, + first_lookup_succeeds: bool, +) -> None: + state, _clone, commits = history_run + clock = [0.0] + timeouts: list[float] = [] + info = BlameInfo( + author_name=_FIRST_AUTHOR, + author_email="author@example.test", + commit_sha=commits["first"], + commit_timestamp="2024-01-02T03:04:05+00:00", + commit_summary="Add query handler", + ) + + def slow_blame(*_args: Any, timeout: float) -> BlameInfo | None: + timeouts.append(timeout) + clock[0] += min(1.0 if len(timeouts) == 1 else 3.0, timeout) + return info if first_lookup_succeeds and len(timeouts) == 1 else None + + monkeypatch.setattr(history, "monotonic", lambda: clock[0]) + monkeypatch.setattr(history, "blame_line", slow_blame) + + result = await _create( + {"file": "app.py", "start_line": 1, "end_line": 1}, + {"file": "app.py", "start_line": 2, "end_line": 2}, + {"file": "app.py", "start_line": 999, "end_line": 999}, + ) + + assert result["success"] is True + assert timeouts == pytest.approx([3.0, 2.0]) + assert clock[0] == pytest.approx(3.0) + report = state.vulnerability_reports[0] + analysis = report["technical_analysis"] + if first_lookup_succeeds: + assert commits["first"] in analysis + assert "**app.py:1**" in analysis + assert "**app.py:2**" not in analysis + else: + assert analysis == _ANALYSIS + saved = json.loads((state.get_run_dir() / "vulnerabilities.json").read_text()) + assert saved == [report] + markdown = (state.get_run_dir() / "vulnerabilities" / f"{report['id']}.md").read_text() + assert analysis in markdown + + +def test_invalid_locations_still_consume_shared_history_budget( + history_run: tuple[ReportState, Path, dict[str, str]], monkeypatch: pytest.MonkeyPatch +) -> None: + state, _clone, _commits = history_run + clock = [0.0] + location_line = history._location_line + + def slow_location_line(location: dict[str, Any]) -> int | None: + clock[0] += 0.5 + return location_line(location) + + monkeypatch.setattr(history, "monotonic", lambda: clock[0]) + monkeypatch.setattr(history, "_location_line", slow_location_line) + report = { + "target": _TARGET, + "technical_analysis": _ANALYSIS, + "code_locations": [{"file": "app.py"} for _ in range(20)], + } + + enrich_report(report, state.run_record) + + assert clock[0] <= 3.0 + assert report["technical_analysis"] == _ANALYSIS diff --git a/tests/test_repository_history.py b/tests/test_repository_history.py index 5962b38d..bc4c8eb0 100644 --- a/tests/test_repository_history.py +++ b/tests/test_repository_history.py @@ -196,3 +196,22 @@ def test_nonzero_exit_is_skipped_and_git_is_local_bounded(repository: Path) -> N assert kwargs["timeout"] <= 3 assert kwargs["env"]["GIT_NO_LAZY_FETCH"] == "1" assert "--no-textconv" in args[0] + + +@pytest.mark.parametrize(("timeout", "expected"), [(0.25, 0.25), (1.5, 1.5), (30, 3)]) +def test_blame_timeout_respects_remaining_budget_and_per_lookup_limit( + repository: Path, timeout: float, expected: float +) -> None: + result = subprocess.CompletedProcess(["git"], 128, stdout="", stderr="fatal") + with patch("strix.core.repository_history.subprocess.run", return_value=result) as run: + assert blame_line(repository, "vulnerable.py", 2, timeout=timeout) is None + + assert run.call_args.kwargs["timeout"] == pytest.approx(expected) + + +@pytest.mark.parametrize("timeout", [0, -1]) +def test_expired_blame_budget_does_not_start_git(repository: Path, timeout: float) -> None: + with patch("strix.core.repository_history.subprocess.run") as run: + assert blame_line(repository, "vulnerable.py", 2, timeout=timeout) is None + + run.assert_not_called()