diff --git a/strix/core/repository_history.py b/strix/core/repository_history.py deleted file mode 100644 index 45855097..00000000 --- a/strix/core/repository_history.py +++ /dev/null @@ -1,104 +0,0 @@ -"""Best-effort history lookups against an existing local Git checkout.""" - -from __future__ import annotations - -import os -import re -import subprocess -from dataclasses import dataclass -from datetime import UTC, datetime -from pathlib import Path - - -@dataclass(frozen=True) -class BlameInfo: - author_name: str - author_email: str - commit_sha: str - commit_timestamp: str - commit_summary: str - - -def _parse_blame(output: str) -> BlameInfo | None: - lines = output.splitlines() - if not lines or "\x00" in output: - return None - header = lines[0].split() - if not header or not re.fullmatch(r"[0-9a-f]{40}|[0-9a-f]{64}", header[0]): - return None - commit_sha = header[0] - if not commit_sha.strip("0"): - # Git uses an all-zero object ID for a line with uncommitted changes. - return None - metadata: dict[str, str] = {} - for raw_line in lines[1:]: - if raw_line.startswith("\t"): - break - key, _, value = raw_line.partition(" ") - metadata[key] = value - try: - timestamp = datetime.fromtimestamp(int(metadata["committer-time"]), tz=UTC).isoformat() - return BlameInfo( - author_name=metadata["author"], - author_email=metadata["author-mail"].removeprefix("<").removesuffix(">"), - commit_sha=commit_sha, - commit_timestamp=timestamp, - commit_summary=metadata["summary"], - ) - except (KeyError, ValueError, OverflowError, OSError): - return 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 - or timeout <= 0 - ): - return None - try: - relative = Path(file_path) - if relative.is_absolute() or ".." in relative.parts: - return None - root = repository.resolve(strict=True) - path = (root / relative).resolve(strict=True) - if not path.is_relative_to(root) or not path.is_file(): - return None - with path.open("rb") as source: - if b"\x00" in source.read(8192): - return None - result = subprocess.run( # noqa: S603 - [ # noqa: S607 - "git", - "-C", - str(root), - "blame", - "--line-porcelain", - "--no-textconv", - "-L", - f"{line},{line}", - "--", - file_path, - ], - capture_output=True, - text=True, - encoding="utf-8", - errors="replace", - check=False, - timeout=min(timeout, 3.0), - env={**os.environ, "GIT_NO_LAZY_FETCH": "1"}, - ) - except (OSError, ValueError, RuntimeError, subprocess.SubprocessError): - return None - return _parse_blame(result.stdout) if result.returncode == 0 else None diff --git a/strix/report/history.py b/strix/report/history.py deleted file mode 100644 index 2c3a398e..00000000 --- a/strix/report/history.py +++ /dev/null @@ -1,125 +0,0 @@ -"""Optional repository history inside the existing technical analysis text.""" - -from __future__ import annotations - -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 - - -logger = logging.getLogger(__name__) -_START = "" -_END = "" -_ENRICHMENT_TIMEOUT = 3.0 - - -def _text(value: str) -> str: - """Keep repository-authored metadata inert in Markdown and HTML renderers.""" - return re.sub(r"([\\`*_\[\]])", r"\\\1", html.escape(" ".join(value.split()))) - - -def _repositories(run_record: dict[str, Any], target: str | None) -> list[Path]: - sources: list[dict[str, Any]] = run_record.get("local_sources") or [] - roots = {Path(s["source_path"]).resolve() for s in sources if s.get("source_path")} - matched: set[Path] = set() - for source in sources: - if target and target in ( - source.get("source_path"), - source.get("workspace_subdir"), - f"/workspace/{source.get('workspace_subdir')}", - ): - matched.add(Path(source["source_path"]).resolve()) - targets: list[dict[str, Any]] = run_record.get("targets_info") or [] - for entry in targets: - details: dict[str, Any] = entry.get("details") or {} - path = details.get("cloned_repo_path") or details.get("target_path") - if entry.get("type") not in {"repository", "local_code"} or not path: - continue - root = Path(path).resolve() - roots.add(root) - if target and target in ( - entry.get("original"), - details.get("target_repo"), - details.get("target_path"), - details.get("workspace_subdir"), - f"/workspace/{details.get('workspace_subdir')}", - ): - matched.add(root) - # 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: - start = location.get("start_line") - if type(start) is not int or start < 1: - return None - primary = location.get("primary_line") - end = location.get("end_line", start) - if type(primary) is int and type(end) is int and start <= primary <= end: - return primary - return start - - -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. 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), - "", - report.get("technical_analysis") or "", - flags=re.DOTALL, - ).rstrip() - if "technical_analysis" in report: - report["technical_analysis"] = analysis - locations: list[dict[str, Any]] = report.get("code_locations") or [] - 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: - continue - seen.add((file_path, line)) - candidates = [root for root in roots if (root / file_path).is_file()] - if len(candidates) != 1: - continue - remaining = deadline - monotonic() - if remaining <= 0: - break - info = blame_line(candidates[0], file_path, line, timeout=remaining) - if info is None: - continue - entries.append( - f"**{_text(file_path)}:{line}**\n\n" - f"- Author: {_text(info.author_name)} ({_text(info.author_email)})\n" - f"- Commit: {_text(info.commit_sha)}\n" - f"- Commit timestamp (UTC): {_text(info.commit_timestamp)}\n" - f"- Commit summary: {_text(info.commit_summary)}" - ) - if entries: - block = "\n\n".join(entries) - report["technical_analysis"] = ( - f"{analysis}\n\n{_START}\n### Last modified by\n\n{block}\n{_END}" - ).lstrip() - except Exception: # noqa: BLE001 - history must never prevent issue creation or reporting. - logger.debug("Repository history enrichment failed (non-fatal)", exc_info=True) diff --git a/strix/report/state.py b/strix/report/state.py index 7feac413..ac15db89 100644 --- a/strix/report/state.py +++ b/strix/report/state.py @@ -14,7 +14,6 @@ from strix.config import codex from strix.config.loader import load_settings from strix.core.paths import run_dir_for, runtime_state_dir from strix.report.coverage import write_coverage -from strix.report.history import enrich_report from strix.report.pricing import resolve_litellm_model from strix.report.sarif import write_sarif from strix.report.writer import ( @@ -408,7 +407,6 @@ class ReportState: if agent_name: report["agent_name"] = agent_name - enrich_report(report, self.run_record) if self.vulnerability_found_callback: self.vulnerability_found_callback(report) @@ -498,9 +496,6 @@ class ReportState: revised["update_history"] = history revised["updated_at"] = entry["timestamp"] - if {"technical_analysis", "code_locations", "target"} & changed.keys(): - enrich_report(revised, self.run_record) - # Persistence must accept the revision before local state changes. A # failed callback leaves the old evidence intact and the update retryable. if self.vulnerability_updated_callback: diff --git a/strix/tools/reporting/tool.py b/strix/tools/reporting/tool.py index c0c1cc6e..3e671de7 100644 --- a/strix/tools/reporting/tool.py +++ b/strix/tools/reporting/tool.py @@ -43,7 +43,6 @@ _CODE_LOCATION_FIELDS = ( "file", "start_line", "end_line", - "primary_line", "snippet", "label", "fix_before", @@ -74,9 +73,7 @@ def _normalize_code_locations( if field not in loc or loc[field] is None: continue value = loc[field] - if field in ("start_line", "end_line", "primary_line"): - if isinstance(value, (bool, float)): - continue + if field in ("start_line", "end_line"): try: normalized[field] = int(value) except (TypeError, ValueError): @@ -1020,6 +1017,27 @@ async def create_vulnerability_report( for the full rules around ``fix_before`` / ``fix_after``, multi-part fixes, and informational-vs-actionable entries. + **Local Git attribution (best effort)**: for a finding with a repository + file and valid line number, use ``exec_command`` in the existing full-clone + checkout. Identify the affected repository first; if ambiguous, skip + attribution. Blame the primary vulnerable line when known, otherwise the + ``start_line`` of the primary code location. Quote paths and use a short + timeout, for example:: + + GIT_NO_LAZY_FETCH=1 timeout 3s git -C REPO blame --line-porcelain \\ + --no-textconv -L LINE,LINE -- FILE + + Add a **Last modified by** subsection to ``technical_analysis`` with the + file/line, author name (``author``), author email (``author-mail``), commit + SHA (first field), commit timestamp (``committer-time``, rendered in UTC), + and commit summary (``summary``), where available. Use only observed Git + output; never invent attribution or imply the author introduced the flaw. + Skip all-zero SHAs (uncommitted lines), missing/invalid line numbers, + missing or renamed paths you cannot resolve, binary/generated files without + useful history, unavailable Git metadata, and command errors/timeouts. + Do not clone, fetch, or retry just for attribution; omit it and file the + finding normally when unavailable. It must never block scanning or reporting. + **CVSS breakdown** is an object with all 8 metrics (each a single uppercase letter): @@ -1126,9 +1144,8 @@ async def create_vulnerability_report( but unverified follow-on risks separate; do not use them to set CVSS metrics. target: Affected URL / domain / repository. - technical_analysis: The mechanism and root cause. When local repository - history is available, the report automatically appends "Last modified - by" details for the code locations. Do not invent attribution. + technical_analysis: The mechanism and root cause, including "Last modified + by" details when verified using local Git attribution as described above. poc_description: Step-by-step reproduction (steps only, no code). poc_script_code: Working PoC (Python preferred). remediation_steps: Specific, actionable fix (prose, no code). @@ -1226,9 +1243,6 @@ async def create_vulnerability_report( - ``end_line`` (REQUIRED): 1-based; ``>= start_line``. Only equal to ``start_line`` when the block truly is one line. - - ``primary_line`` (optional): the primary vulnerable line within - this range. Local Git blame uses this line, or ``start_line`` - when omitted or outside the range. History enrichment is best-effort. - ``snippet`` (optional): verbatim source at this range. - ``label`` (optional): short role description; especially important for multi-part fixes. @@ -1449,6 +1463,8 @@ async def update_vulnerability_report( ``severity_change_conditions`` with a new ``cvss_breakdown``. - ``code_locations`` replaces the whole list. A location carrying ``fix_after`` needs ``fix_verification``. + - When changing a target or code location, refresh or remove any "Last modified + by" attribution in ``technical_analysis``; do not retain stale Git details. The report keeps its id, its original author, and its filing time. The revision is recorded in the report as update history, so state the diff --git a/tests/test_report_history.py b/tests/test_report_history.py index c7001078..45d72bb8 100644 --- a/tests/test_report_history.py +++ b/tests/test_report_history.py @@ -1,10 +1,12 @@ -"""Repository history through the reporting tool, callbacks, and persisted artifacts.""" +"""Smoke-test the prompted Git command and agent-supplied report details.""" from __future__ import annotations import argparse import json import os +import re +import shlex import subprocess import tempfile from pathlib import Path @@ -13,10 +15,7 @@ from typing import TYPE_CHECKING, Any import pytest from agents.tool_context import ToolContext -from strix.core.repository_history import BlameInfo from strix.interface.scan_setup import build_targets_info, prepare_run -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 @@ -26,7 +25,6 @@ if TYPE_CHECKING: _ANALYSIS = "The query interpolates attacker-controlled input." -_TARGET = "https://example.test/team/application.git" _FIRST_AUTHOR = "Alice Original" _LATEST_AUTHOR = "Bea Reviewer" _LOCAL_AUTHOR = "Carol Local" @@ -79,42 +77,6 @@ def _seed_repository(origin: Path) -> dict[str, str]: return {"first": first_sha, "latest": _git(origin, "rev-parse", "HEAD")} -@pytest.fixture -def history_run( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> Iterator[tuple[ReportState, Path, dict[str, str]]]: - """Use an existing full clone, including an older author for the range start.""" - monkeypatch.chdir(tmp_path) - origin = tmp_path / "origin" - commits = _seed_repository(origin) - clone = tmp_path / "application" - _git(tmp_path, "clone", "--quiet", "--no-hardlinks", str(origin), str(clone)) - assert _git(clone, "rev-parse", "--is-shallow-repository") == "false" - - state = ReportState(run_name="history-run") - state.set_scan_config( - { - "targets": [ - { - "type": "repository", - "original": _TARGET, - "details": { - "target_repo": _TARGET, - "cloned_repo_path": str(clone), - "workspace_subdir": "application", - }, - } - ], - "local_sources": [{"source_path": str(clone), "workspace_subdir": "application"}], - } - ) - set_global_report_state(state) - try: - yield state, clone, commits - finally: - set_global_report_state(None) - - @pytest.fixture def scan_setup_run( tmp_path: Path, monkeypatch: pytest.MonkeyPatch @@ -172,14 +134,14 @@ def scan_setup_run( async def _create( - location: dict[str, Any], *additional_locations: dict[str, Any], target: str = _TARGET + location: dict[str, Any], *, target: str, technical_analysis: str ) -> dict[str, Any]: arguments = { "title": "SQL injection in the query handler", "description": "The query handler executes unsanitized input.", "impact": "An anonymous caller can access other users' records.", "target": target, - "technical_analysis": _ANALYSIS, + "technical_analysis": technical_analysis, "poc_description": "Submit a quote in the query parameter.", "poc_script_code": "GET /query?q='", "remediation_steps": "Use parameterized queries.", @@ -190,7 +152,7 @@ async def _create( "severity_change_conditions": "A read-only database role would limit the impact.", "fix_effort": "low", "cvss_breakdown": _CVSS, - "code_locations": [location, *additional_locations], + "code_locations": [location], } context = ToolContext( context={"agent_id": "root"}, @@ -222,296 +184,112 @@ async def _update(report_id: str, **fields: Any) -> dict[str, Any]: return result -@pytest.mark.parametrize("primary_line", [2, None, -1, 3, "invalid", True]) -async def test_tool_enriches_callbacks_and_artifacts_from_full_clone( - history_run: tuple[ReportState, Path, dict[str, str]], primary_line: Any -) -> None: - state, _clone, commits = history_run - callbacks: list[dict[str, Any]] = [] - state.vulnerability_found_callback = lambda report: callbacks.append(dict(report)) - location = {"file": "app.py", "start_line": 1, "end_line": 2, "primary_line": primary_line} - - result = await _create(location) - - assert result["success"] is True - report = state.vulnerability_reports[0] - analysis = report["technical_analysis"] - expected_author = _LATEST_AUTHOR if primary_line == 2 else _FIRST_AUTHOR - expected_sha = commits["latest"] if primary_line == 2 else commits["first"] - expected_summary = "Execute the query" if primary_line == 2 else "Add query handler" - assert analysis.startswith(_ANALYSIS) - assert analysis.count("Last modified by") == 1 - for value in (expected_author, "author@example.test", expected_sha, expected_summary): - assert value in analysis - assert "2024-01-02T03:04:05+00:00" in analysis - assert callbacks == [report] - assert "blame" not in report - assert "author_name" not in report - 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() - technical_section = markdown.split("## Technical Analysis", 1)[1].split( - "## Proof of Concept", 1 - )[0] - assert analysis in technical_section - - -async def test_resume_and_updates_refresh_history_without_stale_or_duplicate_details( - history_run: tuple[ReportState, Path, dict[str, str]], -) -> None: - state, _clone, commits = history_run - assert (await _create({"file": "app.py", "start_line": 1, "end_line": 2}))["success"] - resumed = ReportState(run_name=state.run_name) - resumed.hydrate_from_run_dir() - set_global_report_state(resumed) - callbacks: list[dict[str, Any]] = [] - resumed.vulnerability_updated_callback = lambda report: callbacks.append(dict(report)) - report_id = resumed.vulnerability_reports[0]["id"] - assert commits["first"] in resumed.vulnerability_reports[0]["technical_analysis"] - - result = await _update( - report_id, - code_locations=[{"file": "app.py", "start_line": 1, "end_line": 2, "primary_line": 2}], - ) - assert result["success"] is True - analysis = resumed.vulnerability_reports[0]["technical_analysis"] - assert commits["latest"] in analysis - assert commits["first"] not in analysis - result = await _update(report_id, technical_analysis="Revised root cause.\n\n" + analysis) - assert result["success"] is True - analysis = resumed.vulnerability_reports[0]["technical_analysis"] - assert analysis.count("Last modified by") == 1 - assert analysis.startswith("Revised root cause.") - - result = await _update( - report_id, code_locations=[{"file": "deleted.py", "start_line": 1, "end_line": 2}] - ) - assert result["success"] is True - final_report = resumed.vulnerability_reports[0] - assert "Last modified by" not in final_report["technical_analysis"] - assert callbacks[-1] == final_report - saved = json.loads((resumed.get_run_dir() / "vulnerabilities.json").read_text()) - assert saved == [final_report] - markdown = (resumed.get_run_dir() / "vulnerabilities" / f"{report_id}.md").read_text() - assert "Last modified by" not in markdown +def test_reporting_tools_expose_optional_git_attribution_guidance() -> None: + description = create_vulnerability_report.description + for instruction in ( + "exec_command", + "existing full-clone", + "primary vulnerable line", + "start_line", + "Last modified by", + "technical_analysis", + "author-mail", + "committer-time", + "summary", + "all-zero SHAs", + "errors/timeouts", + "never block", + ): + assert instruction in description + assert "refresh or remove" in update_vulnerability_report.description @pytest.mark.parametrize( - "location", - [ - {"file": "deleted.py", "start_line": 1, "end_line": 1}, - {"file": "app.py", "start_line": 999, "end_line": 999}, - {"file": "app.py", "start_line": True, "end_line": 1}, - {"file": "app.py", "start_line": 1.5, "end_line": 2}, - {"file": "app.py", "end_line": 1}, - ], + "case", ["primary", "start", "local", "missing", "out_of_range", "uncommitted"] ) -async def test_unavailable_history_does_not_block_reporting( - history_run: tuple[ReportState, Path, dict[str, str]], location: dict[str, Any] -) -> None: - state, _clone, _commits = history_run - assert (await _create(location))["success"] is True - assert state.vulnerability_reports[0]["technical_analysis"] == _ANALYSIS - assert (state.get_run_dir() / "vulnerabilities.json").exists() - - -async def test_unexpected_history_failure_does_not_block_reporting( - history_run: tuple[ReportState, Path, dict[str, str]], monkeypatch: pytest.MonkeyPatch -) -> None: - state, _clone, _commits = history_run - - def broken_blame(*_args: Any, **_kwargs: Any) -> None: - raise RuntimeError("history lookup failed") - - monkeypatch.setattr("strix.report.history.blame_line", broken_blame) - assert (await _create({"file": "app.py", "start_line": 1, "end_line": 1}))["success"] is True - assert state.vulnerability_reports[0]["technical_analysis"] == _ANALYSIS - assert (state.get_run_dir() / "vulnerabilities" / "vuln-0001.md").exists() - - -@pytest.mark.parametrize("target_alias", [_TARGET, "application", "/workspace/application"]) -def test_ambiguous_repository_requires_matching_target_and_target_updates_refresh_history( - history_run: tuple[ReportState, Path, dict[str, str]], tmp_path: Path, target_alias: str -) -> None: - state, clone, commits = history_run - other = tmp_path / "other" - _git(tmp_path, "clone", "--quiet", str(clone), str(other)) - (other / "app.py").write_text("query = 'other'\nresult = execute(query)\n", encoding="utf-8") - _git(other, "commit", "--quiet", "-am", "Other repository change", author="Other Author") - state.run_record["local_sources"].append( - {"source_path": str(other), "workspace_subdir": "other"} - ) - report = { - "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 - - report["target"] = target_alias - enrich_report(report, state.run_record) - assert commits["first"] in report["technical_analysis"] - report["target"] = "other" - enrich_report(report, state.run_record) - 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("alias", ["local", "repository", "workspace", "unmatched"]) -async def test_cli_scan_setup_wires_local_and_cloned_targets_into_history( - scan_setup_run: tuple[ReportState, dict[str, str], dict[str, str]], alias: str +async def test_prompted_blame_command_to_report_artifacts( + scan_setup_run: tuple[ReportState, dict[str, str], dict[str, str]], case: str ) -> None: + # Exercise real CLI setup and Git, then supply the analysis as an agent would. + # This is a tool/report smoke test, not an LLM compliance test. state, targets, commits = scan_setup_run - target = { - "local": targets["local"], - "repository": targets["repository"], - "workspace": "/workspace/origin", - "unmatched": "https://example.test/service", - }[alias] + local = case == "local" + target = targets["local" if local else "repository"] + checkout = Path(state.run_record["local_sources"][0 if local else 1]["source_path"]) + assert _git(checkout, "rev-parse", "--is-shallow-repository") == "false" + file_path = "missing.py" if case == "missing" else "app.py" + line = 999 if case == "out_of_range" else 2 if case == "primary" else 1 + if case == "uncommitted": + (checkout / "app.py").write_text("uncommitted change\nresult = query\n", encoding="utf-8") - result = await _create({"file": "app.py", "start_line": 1, "end_line": 1}, target=target) - - assert result["success"] is True - analysis = state.vulnerability_reports[0]["technical_analysis"] - if alias == "local": - assert _LOCAL_AUTHOR in analysis - assert commits["local"] in analysis - elif alias in {"repository", "workspace"}: - assert _FIRST_AUTHOR in analysis - assert commits["first"] in analysis - assert commits["local"] not in analysis - else: - assert analysis == _ANALYSIS - saved = json.loads((state.get_run_dir() / "vulnerabilities.json").read_text()) - assert saved == state.vulnerability_reports - - -@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", + # Execute the exact example exposed to the model, with quoted real paths. + match = re.search( + r"GIT_NO_LAZY_FETCH=1 timeout 3s git.*?-- FILE", + create_vulnerability_report.description, + re.DOTALL, ) + assert match is not None + command = ( + match.group() + .replace("REPO", shlex.quote(str(checkout))) + .replace("LINE", str(line)) + .replace("FILE", shlex.quote(file_path)) + ) + output = subprocess.run( # noqa: S603 + ["bash", "-c", command], # noqa: S607 + capture_output=True, + text=True, + timeout=5, + check=False, + ) + analysis = _ANALYSIS + has_history = case in {"primary", "start", "local"} + if has_history: + assert output.returncode == 0 + sha = commits["local" if local else "latest" if case == "primary" else "first"] + author = _LOCAL_AUTHOR if local else _LATEST_AUTHOR if case == "primary" else _FIRST_AUTHOR + assert output.stdout.split()[0] == sha + assert f"author {author}\n" in output.stdout + assert "author-mail " in output.stdout + assert "committer-time 1704164645" in output.stdout + summary = next( + s.removeprefix("summary ") + for s in output.stdout.splitlines() + if s.startswith("summary ") + ) + analysis += ( + f"\n\n### Last modified by\n\n{file_path}:{line} — {author} " + f"(author@example.test); commit {sha}; " + f"2024-01-02T03:04:05+00:00; {summary}" + ) + elif case == "uncommitted": + assert output.stdout.split()[0] == "0" * 40 + else: + assert output.returncode != 0 - 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) - + callbacks: list[dict[str, Any]] = [] + state.vulnerability_found_callback = lambda report: callbacks.append(dict(report)) 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}, + {"file": file_path, "start_line": 1, "end_line": 2}, + target=target, + technical_analysis=analysis, ) - 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 + assert callbacks == [report] + assert report["technical_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 + assert ( + analysis in markdown.split("## Technical Analysis", 1)[1].split("## Proof of Concept", 1)[0] + ) - -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 + if has_history: + revised = await _update( + report["id"], + target="https://example.test/new-target", + technical_analysis=_ANALYSIS, + ) + assert revised["success"] is True + assert state.vulnerability_reports[0]["technical_analysis"] == _ANALYSIS diff --git a/tests/test_repository_history.py b/tests/test_repository_history.py deleted file mode 100644 index bc4c8eb0..00000000 --- a/tests/test_repository_history.py +++ /dev/null @@ -1,217 +0,0 @@ -"""Local history lookup coverage using real Git history and failure injection.""" - -from __future__ import annotations - -import os -import subprocess -from typing import TYPE_CHECKING, Any -from unittest.mock import patch - -import pytest - -from strix.core.repository_history import blame_line - - -if TYPE_CHECKING: - from pathlib import Path - - -def _git(repository: Path, *args: str, env: dict[str, str] | None = None) -> str: - return subprocess.run( # noqa: S603 - ["git", "-C", str(repository), *args], # noqa: S607 - env={**os.environ, **(env or {})}, - check=True, - capture_output=True, - text=True, - ).stdout.strip() - - -@pytest.fixture -def repository(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: - monkeypatch.setenv("GIT_CONFIG_GLOBAL", os.devnull) - monkeypatch.setenv("GIT_CONFIG_NOSYSTEM", "1") - root = tmp_path / "repository" - root.mkdir() - _git(root, "init") - _git(root, "config", "user.name", "First Author") - _git(root, "config", "user.email", "first@example.test") - (root / "vulnerable.py").write_text("first line\nvulnerable line\nlast line\n") - _git(root, "add", "vulnerable.py") - _git( - root, - "commit", - "-m", - "Original source", - env={ - "GIT_AUTHOR_DATE": "2020-01-01T00:00:00Z", - "GIT_COMMITTER_DATE": "2020-01-02T00:00:00Z", - }, - ) - (root / "vulnerable.py").write_text("first line\nchanged vulnerable line\nlast line\n") - _git(root, "add", "vulnerable.py") - _git( - root, - "commit", - "-m", - "Change vulnerable line\n\nCommit body.", - env={ - "GIT_AUTHOR_NAME": "Second Author", - "GIT_AUTHOR_EMAIL": "second@example.test", - "GIT_AUTHOR_DATE": "2021-02-03T04:05:06Z", - "GIT_COMMITTER_DATE": "2021-02-04T05:06:07Z", - }, - ) - return root - - -def test_blame_attributes_requested_line_and_commit_timestamp(repository: Path) -> None: - result = blame_line(repository, "vulnerable.py", 2) - - assert result is not None - assert result.author_name == "Second Author" - assert result.author_email == "second@example.test" - assert result.commit_sha == _git(repository, "rev-parse", "HEAD") - assert result.commit_timestamp == "2021-02-04T05:06:07+00:00" - assert result.commit_summary == "Change vulnerable line" - unchanged = blame_line(repository, "vulnerable.py", 1) - assert unchanged is not None - assert unchanged.author_name == "First Author" - assert unchanged.commit_sha == _git(repository, "rev-parse", "HEAD~1") - - -def test_committed_rename_follows_history_and_missing_old_path_is_skipped(repository: Path) -> None: - original_sha = _git(repository, "rev-parse", "HEAD") - _git(repository, "mv", "vulnerable.py", "renamed.py") - _git(repository, "commit", "-m", "Rename source") - - result = blame_line(repository, "renamed.py", 2) - assert result is not None - assert result.commit_sha == original_sha - assert blame_line(repository, "vulnerable.py", 2) is None - - -def test_uncommitted_line_is_skipped_but_unchanged_line_retains_history(repository: Path) -> None: - (repository / "vulnerable.py").write_text("first line\nuncommitted change\nlast line\n") - - assert blame_line(repository, "vulnerable.py", 2) is None - assert blame_line(repository, "vulnerable.py", 1) is not None - _git(repository, "add", "vulnerable.py") - assert blame_line(repository, "vulnerable.py", 2) is None - - -@pytest.mark.parametrize("line", [None, 0, -1, True, "2", 1.5, 500]) -def test_invalid_missing_or_out_of_bounds_lines_are_skipped(repository: Path, line: Any) -> None: - assert blame_line(repository, "vulnerable.py", line) is None - - -@pytest.mark.parametrize( - "file_path", ["", "missing.py", "../vulnerable.py", "/etc/passwd", "bad\x00"] -) -def test_invalid_or_missing_paths_are_skipped(repository: Path, file_path: str) -> None: - assert blame_line(repository, file_path, 1) is None - - -def test_binary_untracked_and_non_repository_files_are_skipped( - repository: Path, tmp_path: Path -) -> None: - (repository / "binary.dat").write_bytes(b"binary\x00data\n") - _git(repository, "add", "binary.dat") - _git(repository, "commit", "-m", "Add binary") - (repository / "generated.py").write_text("generated source\n") - (tmp_path / "plain.py").write_text("not a repository\n") - - assert blame_line(repository, "binary.dat", 1) is None - assert blame_line(repository, "generated.py", 1) is None - assert blame_line(tmp_path, "plain.py", 1) is None - assert blame_line(tmp_path / "absent", "plain.py", 1) is None - - -def test_tracked_generated_text_can_be_attributed(repository: Path) -> None: - (repository / "generated.py").write_text("# Generated file\nvalue = 1\n") - _git(repository, "add", "generated.py") - _git(repository, "commit", "-m", "Generate source") - - result = blame_line(repository, "generated.py", 2) - assert result is not None - assert result.commit_summary == "Generate source" - - -def test_existing_linked_worktree_uses_its_local_history(repository: Path, tmp_path: Path) -> None: - worktree = tmp_path / "worktree" - _git(repository, "worktree", "add", "--detach", str(worktree), "HEAD") - - result = blame_line(worktree, "vulnerable.py", 2) - assert result is not None - assert result.commit_sha == _git(repository, "rev-parse", "HEAD") - - -def test_symlink_cannot_escape_repository(repository: Path, tmp_path: Path) -> None: - outside = tmp_path / "outside.py" - outside.write_text("outside source\n") - (repository / "escape.py").symlink_to(outside) - - with patch("strix.core.repository_history.subprocess.run") as run: - assert blame_line(repository, "escape.py", 1) is None - run.assert_not_called() - - -@pytest.mark.parametrize( - "failure", - [ - FileNotFoundError("git missing"), - PermissionError("denied"), - subprocess.TimeoutExpired("git", 3), - ], -) -def test_git_failures_do_not_escape(repository: Path, failure: Exception) -> None: - with patch("strix.core.repository_history.subprocess.run", side_effect=failure): - assert blame_line(repository, "vulnerable.py", 2) is None - - -@pytest.mark.parametrize("output", ["", "malformed", "a" * 40 + " 1 1 1\nauthor Someone\n"]) -def test_malformed_output_is_skipped(repository: Path, output: str) -> None: - result = subprocess.CompletedProcess(["git"], 0, stdout=output) - with patch("strix.core.repository_history.subprocess.run", return_value=result): - assert blame_line(repository, "vulnerable.py", 2) is None - - -@pytest.mark.parametrize("timestamp", ["not-a-time", "9999999999999999999999999999999"]) -def test_invalid_commit_timestamp_is_skipped(repository: Path, timestamp: str) -> None: - output = ( - f"{'a' * 40} 2 2 1\nauthor Name\nauthor-mail \n" - f"committer-time {timestamp}\nsummary Message\n\tcode\n" - ) - result = subprocess.CompletedProcess(["git"], 0, stdout=output) - with patch("strix.core.repository_history.subprocess.run", return_value=result): - assert blame_line(repository, "vulnerable.py", 2) is None - - -def test_nonzero_exit_is_skipped_and_git_is_local_bounded(repository: Path) -> 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) is None - - args, kwargs = run.call_args - assert args[0][-4:] == ["-L", "2,2", "--", "vulnerable.py"] - 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()