diff --git a/strix/core/repository_history.py b/strix/core/repository_history.py new file mode 100644 index 00000000..a721c951 --- /dev/null +++ b/strix/core/repository_history.py @@ -0,0 +1,95 @@ +"""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) -> 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. + """ + if type(line) is not int or line < 1 or not isinstance(file_path, str) or not file_path: + 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=3, + 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 new file mode 100644 index 00000000..7104d799 --- /dev/null +++ b/strix/report/history.py @@ -0,0 +1,112 @@ +"""Optional repository history inside the existing technical analysis text.""" + +from __future__ import annotations + +import html +import logging +import re +from pathlib import Path +from typing import Any + +from strix.core.repository_history import blame_line + + +logger = logging.getLogger(__name__) +_START = "" +_END = "" + + +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) + return sorted(matched or roots) + + +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. Ambiguous paths across multiple repositories are + omitted unless the finding's target identifies its repository. Markers let + revisions and resumed scans replace only the generated part of the analysis. + """ + try: + 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")) + entries: list[str] = [] + seen: set[tuple[str, int]] = set() + for location in locations: + 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 + info = blame_line(candidates[0], file_path, line) + 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 ac15db89..7feac413 100644 --- a/strix/report/state.py +++ b/strix/report/state.py @@ -14,6 +14,7 @@ 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 ( @@ -407,6 +408,7 @@ 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) @@ -496,6 +498,9 @@ 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 2d245b69..c0c1cc6e 100644 --- a/strix/tools/reporting/tool.py +++ b/strix/tools/reporting/tool.py @@ -43,6 +43,7 @@ _CODE_LOCATION_FIELDS = ( "file", "start_line", "end_line", + "primary_line", "snippet", "label", "fix_before", @@ -73,7 +74,9 @@ 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"): + if field in ("start_line", "end_line", "primary_line"): + if isinstance(value, (bool, float)): + continue try: normalized[field] = int(value) except (TypeError, ValueError): @@ -1123,7 +1126,9 @@ 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. + 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. 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). @@ -1221,6 +1226,9 @@ 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. diff --git a/tests/test_report_history.py b/tests/test_report_history.py new file mode 100644 index 00000000..9183fbbb --- /dev/null +++ b/tests/test_report_history.py @@ -0,0 +1,284 @@ +"""Repository history through the reporting tool, callbacks, and persisted artifacts.""" + +from __future__ import annotations + +import json +import os +import subprocess +from typing import TYPE_CHECKING, Any + +import pytest +from agents.tool_context import ToolContext + +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 + + +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + + +_ANALYSIS = "The query interpolates attacker-controlled input." +_TARGET = "https://example.test/team/application.git" +_FIRST_AUTHOR = "Alice Original" +_LATEST_AUTHOR = "Bea Reviewer" +_CVSS = { + "attack_vector": "N", + "attack_complexity": "L", + "privileges_required": "N", + "user_interaction": "N", + "scope": "U", + "confidentiality": "H", + "integrity": "H", + "availability": "H", +} + + +def _git(path: Path, *args: str, author: str = _FIRST_AUTHOR) -> str: + result = subprocess.run( # noqa: S603 + ["git", "-c", "commit.gpgsign=false", "-C", str(path), *args], # noqa: S607 + env={ + **os.environ, + "GIT_AUTHOR_NAME": author, + "GIT_AUTHOR_EMAIL": "author@example.test", + "GIT_COMMITTER_NAME": author, + "GIT_COMMITTER_EMAIL": "committer@example.test", + "GIT_AUTHOR_DATE": "2024-01-02T03:04:05+00:00", + "GIT_COMMITTER_DATE": "2024-01-02T03:04:05+00:00", + }, + check=True, + capture_output=True, + text=True, + ) + return result.stdout.strip() + + +@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" + origin.mkdir() + _git(origin, "init", "--quiet") + (origin / "app.py").write_text("query = 'initial'\nresult = query\n", encoding="utf-8") + _git(origin, "add", "app.py") + _git(origin, "commit", "--quiet", "-m", "Add query handler") + first_sha = _git(origin, "rev-parse", "HEAD") + (origin / "app.py").write_text("query = 'initial'\nresult = execute(query)\n", encoding="utf-8") + _git(origin, "commit", "--quiet", "-am", "Execute the query", author=_LATEST_AUTHOR) + latest_sha = _git(origin, "rev-parse", "HEAD") + 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, {"first": first_sha, "latest": latest_sha} + finally: + set_global_report_state(None) + + +async def _create(location: dict[str, Any]) -> 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, + "poc_description": "Submit a quote in the query parameter.", + "poc_script_code": "GET /query?q='", + "remediation_steps": "Use parameterized queries.", + "evidence": "The response includes another user's record.", + "assumptions": "The observed database role is available in production.", + "counterevidence": "No input validation runs before the query.", + "confidence": "high", + "severity_change_conditions": "A read-only database role would limit the impact.", + "fix_effort": "low", + "cvss_breakdown": _CVSS, + "code_locations": [location], + } + context = ToolContext( + context={"agent_id": "root"}, + tool_name="create_vulnerability_report", + tool_call_id="create-1", + tool_arguments=json.dumps(arguments), + ) + result: dict[str, Any] = json.loads( + await create_vulnerability_report.on_invoke_tool(context, json.dumps(arguments)) + ) + return result + + +async def _update(report_id: str, **fields: Any) -> dict[str, Any]: + arguments = { + "report_id": report_id, + "update_reason": "Refined the vulnerable location.", + **fields, + } + context = ToolContext( + context={"agent_id": "root"}, + tool_name="update_vulnerability_report", + tool_call_id="update-1", + tool_arguments=json.dumps(arguments), + ) + result: dict[str, Any] = json.loads( + await update_vulnerability_report.on_invoke_tool(context, json.dumps(arguments)) + ) + 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 + + +@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}, + ], +) +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) -> 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 diff --git a/tests/test_repository_history.py b/tests/test_repository_history.py new file mode 100644 index 00000000..5962b38d --- /dev/null +++ b/tests/test_repository_history.py @@ -0,0 +1,198 @@ +"""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]