Make local blame guidance reliable in live reporting

strix/tools/reporting/tool.py

@@ -927,6 +927,29 @@ async def create_vulnerability_report(
 ) -> str:
     """File a vulnerability report — one report per fully-verified finding.
 
+    **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 -c safe.directory=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. Convert the Unix timestamp
+    using a tool (e.g. Python ``datetime`` with ``timezone.utc``), never mentally;
+    if conversion fails, keep the observed Unix timestamp. Use only observed
+    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.
+
     **When to file**: you have a concrete vulnerability with a working
     proof-of-concept and you're 100% sure it's a real issue.
 
@@ -1123,7 +1146,13 @@ 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. Before filing a finding
+            with code locations, use ``exec_command`` to attempt local ``git blame``
+            on the primary vulnerable line (otherwise ``start_line``) in the correct
+            existing checkout, using the short timeout described above. Include the
+            observed author name/email, full commit SHA, UTC commit timestamp, and
+            summary under **Last modified by** here. If history is unavailable or
+            the command fails, omit attribution and file normally; never invent it.
         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).
@@ -1441,6 +1470,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

tests/test_report_history.py

@@ -0,0 +1,299 @@
+"""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
+from typing import TYPE_CHECKING, Any
+
+import pytest
+from agents.tool_context import ToolContext
+
+from strix.interface.scan_setup import build_targets_info, prepare_run
+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
+
+
+_ANALYSIS = "The query interpolates attacker-controlled input."
+_FIRST_AUTHOR = "Alice Original"
+_LATEST_AUTHOR = "Bea Reviewer"
+_LOCAL_AUTHOR = "Carol Local"
+_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(autouse=True)
+def _isolate_git_config(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("GIT_CONFIG_GLOBAL", os.devnull)
+    monkeypatch.setenv("GIT_CONFIG_NOSYSTEM", "1")
+
+
+def _seed_repository(origin: Path) -> dict[str, str]:
+    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)
+    return {"first": first_sha, "latest": _git(origin, "rev-parse", "HEAD")}
+
+
+@pytest.fixture
+def scan_setup_run(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> Iterator[tuple[ReportState, dict[str, str], dict[str, str]]]:
+    """Build the run the way the CLI does: target inference, cloning, local sources."""
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setattr(tempfile, "tempdir", str(tmp_path / "tmp"))
+    commits = _seed_repository(tmp_path / "origin.git")
+    repo_target = (tmp_path / "origin.git").as_uri()
+    local = tmp_path / "service"
+    local.mkdir()
+    _git(local, "init", "--quiet")
+    (local / "app.py").write_text("token = request.args['token']\n", encoding="utf-8")
+    _git(local, "add", "app.py")
+    _git(local, "commit", "--quiet", "-m", "Read the token", author=_LOCAL_AUTHOR)
+
+    args = argparse.Namespace(
+        target=[str(local), repo_target],
+        target_list=None,
+        resume=None,
+        scan_mode="quick",
+        scope_mode="full",
+        diff_base=None,
+        non_interactive=True,
+        instruction="",
+        user_instruction=None,
+        workspace_mount=None,
+        workspace_files=[],
+    )
+    build_targets_info(args)
+    prepare_run(args)
+    clone = Path(args.targets_info[1]["details"]["cloned_repo_path"])
+    assert clone.is_relative_to(tmp_path / "tmp")
+
+    state = ReportState(args.run_name)
+    state.hydrate_from_run_dir()
+    state.set_scan_config(
+        {
+            "targets": args.targets_info,
+            "local_sources": args.local_sources,
+            "run_name": args.run_name,
+            "scope_mode": args.scope_mode,
+            "non_interactive": True,
+        }
+    )
+    set_global_report_state(state)
+    try:
+        yield (
+            state,
+            {"local": str(local.resolve()), "repository": repo_target},
+            {**commits, "local": _git(local, "rev-parse", "HEAD")},
+        )
+    finally:
+        set_global_report_state(None)
+
+
+async def _create(
+    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": technical_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
+
+
+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",
+        "using a tool (e.g. Python",
+        "timezone.utc",
+        "summary",
+        "all-zero SHAs",
+        "errors/timeouts",
+        "never block",
+    ):
+        assert instruction in description
+    assert "refresh or remove" in update_vulnerability_report.description
+    assert "Before filing a finding" in description
+    assert description.index("Local Git attribution") < description.index("When to file")
+
+
+@pytest.mark.parametrize(
+    "case", ["primary", "start", "local", "missing", "out_of_range", "uncommitted"]
+)
+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
+    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")
+
+    # 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 <author@example.test>" 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
+
+    callbacks: list[dict[str, Any]] = []
+    state.vulnerability_found_callback = lambda report: callbacks.append(dict(report))
+    result = await _create(
+        {"file": file_path, "start_line": 1, "end_line": 2},
+        target=target,
+        technical_analysis=analysis,
+    )
+    assert result["success"] is True
+    report = state.vulnerability_reports[0]
+    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.split("## Technical Analysis", 1)[1].split("## Proof of Concept", 1)[0]
+    )
+
+    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