Add finding-identity research harness for studying scheme effectiveness

research/__init__.py

@@ -0,0 +1 @@
+"""Research experiments built on CodeGuard (not part of the released package)."""

research/finding_identity/.gitignore

@@ -0,0 +1,2 @@
+.repos/
+results/

research/finding_identity/README.md

@@ -0,0 +1,91 @@
+# Finding identity: replay harness
+
+Research code for the study *How well do static-analysis findings keep their identity as
+code changes?* It is not part of the released `codeguard` package.
+
+Baselines, suppressions, "fail only on new findings" CI checks and alert de-duplication all
+rely on recognising that a finding in version *n+1* of a file is the same finding as one in
+version *n*. This harness replays real git histories, scans every changed file before and
+after each commit with CodeGuard, and scores several **identity schemes** against a
+**ground truth** built from the diff.
+
+## Schemes
+
+Every key includes the rule ID and the file path.
+
+| Scheme | Identity is built from |
+|---|---|
+| `line` | the line number |
+| `line_text` | the exact text of the flagged line |
+| `github` | GitHub code scanning's `primaryLocationLineHash` (exact port of `fingerprints.ts` from github/codeql-action, verified against the original) |
+| `codeguard` | CodeGuard's `codeguard/v1` fingerprint: enclosing function/class plus the normalized statement (literals masked, comments and whitespace removed) |
+| `cg_no_scope` | ablation: `codeguard` without the enclosing scope |
+| `cg_no_mask` | ablation: `codeguard` without masking literals |
+| `cg_raw` | ablation: `codeguard` without any normalization (scope plus raw line) |
+
+## Ground truth
+
+For each changed file, old lines are mapped to new lines with a line diff
+(`linemap.py`). Unchanged lines map exactly ("moved"). Inside a replaced block, lines are
+paired when their similarity is at least 0.6 ("edited"). An old and a new finding are the
+same finding when they have the same rule and the old line maps to the new line. Each new
+finding is then truly **known** (it existed before) or truly **new**.
+
+## Outcomes and metrics
+
+For each new finding, a scheme either matches it to an old finding or not:
+
+| Outcome | Truth | Scheme | Meaning |
+|---|---|---|---|
+| `known_ok` | known | matched to the right finding | correct |
+| `mispaired` | known | matched to a different old finding | wrong history carried over |
+| `false_new` | known | not matched | old problem reported as new: noise, broken baselines |
+| `false_known` | new | matched to an old finding | **new problem hidden** |
+| `new_ok` | new | not matched | correct |
+
+The summary reports, per scheme:
+
+- **false-new rate** = `false_new / known`, also split by moved and edited findings
+- **hidden rate** = `false_known / introduced`
+- **mispair rate** = `mispaired / known`
+- **collision rate** = findings sharing a key with another finding in the same file version,
+  over all findings in all scanned versions
+
+Only files that changed in a commit are compared. Every scheme trivially keeps the findings
+in untouched files, so including them would only inflate the counts.
+
+## Usage
+
+From the repository root, with CodeGuard installed (`pip install -e .`):
+
+```bash
+# Replay the last 200 first-parent commits of a repository (clones shallowly if a URL)
+python -m research.finding_identity replay https://github.com/pallets/flask --commits 200
+
+# Branch, label, rule categories, and one record per finding for manual labelling
+python -m research.finding_identity replay https://github.com/DefectDojo/django-DefectDojo \
+    --ref dev --commits 200 --label defectdojo --categories security,performance,quality --details
+
+# Rate tables for one or more runs; pools all runs into an "ALL" table
+python -m research.finding_identity summarize research/finding_identity/results/* --csv rates.csv
+```
+
+Clones go to `.repos/` and results to `results/<label>/`; both are git-ignored. Each result
+directory holds `summary.json`, `pairs.jsonl` (one line per commit pair) and, with
+`--details`, `findings.jsonl` (one line per compared finding with its true status and each
+scheme's outcome, for sampling and manual checks).
+
+Tests:
+
+```bash
+python -m pytest research/finding_identity/tests
+```
+
+## Known limitations
+
+- The ground truth is a line diff. A finding whose anchor line is rewritten beyond the
+  similarity threshold counts as removed plus introduced; check a sample by hand.
+- CodeGuard's fingerprint takes the statement from `str.splitlines()`, which also splits on
+  form feeds and other rare separators, so in files containing them the `codeguard` key can
+  hash a neighbouring line. The harness measures the tool as it is.
+- Taint findings depend on several lines; the ground truth tracks only the reported line.

research/finding_identity/__init__.py

@@ -0,0 +1,2 @@
+"""Replay harness for the finding-identity study: how well do static-analysis findings
+keep their identity as code changes?  See README.md."""

research/finding_identity/__main__.py

@@ -0,0 +1,3 @@
+from .cli import main
+
+main()

research/finding_identity/cli.py

@@ -0,0 +1,64 @@
+"""Command line: ``python -m research.finding_identity {replay,summarize} ...``."""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+from .gitrepo import GitRepo
+from .replay import replay
+from .scan import Scanner, rule_ids_for
+from .schemes import ALL_SCHEMES
+from .summarize import summarize
+
+HERE = Path(__file__).parent
+
+
+def main(argv: list[str] | None = None) -> None:
+    parser = argparse.ArgumentParser(prog="python -m research.finding_identity")
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    rp = sub.add_parser("replay", help="replay a repository's history")
+    rp.add_argument("repo", help="local repository path or a git URL to clone")
+    rp.add_argument("--ref", default="HEAD", help="branch or commit to walk back from")
+    rp.add_argument("--commits", type=int, default=200, help="number of commit pairs")
+    rp.add_argument("--label", help="name for the results (default: repository name)")
+    rp.add_argument("--out", type=Path, help="results directory")
+    rp.add_argument("--workdir", type=Path, default=HERE / ".repos", help="where to clone")
+    rp.add_argument(
+        "--categories",
+        default="security,performance,quality",
+        help="comma-separated rule categories to run",
+    )
+    rp.add_argument("--details", action="store_true", help="write one record per finding")
+
+    sp = sub.add_parser("summarize", help="print rate tables for one or more result dirs")
+    sp.add_argument("results", nargs="+", type=Path)
+    sp.add_argument("--csv", type=Path, help="also write the rows to this CSV file")
+
+    args = parser.parse_args(argv)
+    if args.command == "summarize":
+        summarize(args.results, args.csv)
+        return
+
+    ref = None if args.ref == "HEAD" else args.ref
+    repo = GitRepo.open_or_clone(args.repo, args.workdir, depth=args.commits + 2, ref=ref)
+    label = args.label or repo.path.name
+    out = args.out or HERE / "results" / label
+    categories = [c.strip() for c in args.categories.split(",") if c.strip()]
+    scanner = Scanner(rule_ids_for(categories), ALL_SCHEMES)
+    try:
+        replay(
+            repo,
+            ref=args.ref,
+            commits=args.commits,
+            scanner=scanner,
+            schemes=ALL_SCHEMES,
+            out_dir=out,
+            label=label,
+            categories=categories,
+            details=args.details,
+        )
+    finally:
+        repo.close()
+    summarize([out])

research/finding_identity/gitrepo.py

@@ -0,0 +1,119 @@
+"""Minimal git access for the replay: history, per-commit changes, and blob contents."""
+
+from __future__ import annotations
+
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+from typing import IO
+
+_NULL_SHA = "0" * 40
+_SKIP_MODES = {"160000", "120000"}  # submodules and symlinks
+
+
+@dataclass(frozen=True)
+class Change:
+    """One file changed between two commits. A missing side is ``None``."""
+
+    status: str  # A, D, M, R (rename), C (copy), T (type change)
+    old_path: str | None
+    new_path: str | None
+    old_sha: str | None
+    new_sha: str | None
+
+
+class GitRepo:
+    def __init__(self, path: Path) -> None:
+        self.path = path
+        self._batch: subprocess.Popen[bytes] | None = None
+
+    # -- construction -------------------------------------------------------
+
+    @classmethod
+    def open_or_clone(cls, spec: str, workdir: Path, *, depth: int, ref: str | None) -> GitRepo:
+        """Use *spec* as a local repository, or clone it (shallow) into *workdir*."""
+        local = Path(spec)
+        if local.is_dir() and (local / ".git").exists():
+            return cls(local)
+        name = spec.rstrip("/").removesuffix(".git").rsplit("/", 1)[-1]
+        target = workdir / name
+        if not (target / ".git").exists():
+            workdir.mkdir(parents=True, exist_ok=True)
+            args = ["git", "clone", "--quiet", "--no-tags", "--single-branch", f"--depth={depth}"]
+            if ref:
+                args += ["--branch", ref]
+            subprocess.run([*args, spec, str(target)], check=True)
+        return cls(target)
+
+    # -- queries ------------------------------------------------------------
+
+    def _git(self, *args: str) -> bytes:
+        return subprocess.run(
+            ["git", "-C", str(self.path), *args],
+            check=True,
+            capture_output=True,
+        ).stdout
+
+    def resolve(self, ref: str) -> str:
+        return self._git("rev-parse", "--verify", f"{ref}^{{commit}}").decode().strip()
+
+    def first_parent_history(self, ref: str, count: int) -> list[str]:
+        """The last *count* commits on *ref*'s first-parent line, oldest first."""
+        out = self._git("rev-list", "--first-parent", f"--max-count={count}", ref)
+        return list(reversed(out.decode().split()))
+
+    def changes(self, old: str, new: str) -> list[Change]:
+        raw = self._git("diff", "--raw", "-z", "-M", "--no-abbrev", old, new)
+        fields = raw.decode("utf-8", "surrogateescape").split("\0")
+        out: list[Change] = []
+        i = 0
+        while i < len(fields) and fields[i]:
+            meta = fields[i].lstrip(":").split()
+            old_mode, new_mode, old_sha, new_sha, status = meta[:5]
+            kind = status[0]
+            if kind in ("R", "C"):
+                old_path, new_path = fields[i + 1], fields[i + 2]
+                i += 3
+            else:
+                old_path = new_path = fields[i + 1]
+                i += 2
+            if old_mode in _SKIP_MODES or new_mode in _SKIP_MODES:
+                continue
+            out.append(
+                Change(
+                    status=kind,
+                    old_path=None if kind == "A" else old_path,
+                    new_path=None if kind == "D" else new_path,
+                    old_sha=None if old_sha == _NULL_SHA else old_sha,
+                    new_sha=None if new_sha == _NULL_SHA else new_sha,
+                )
+            )
+        return out
+
+    def blob(self, sha: str) -> bytes:
+        """Contents of blob *sha*, through one long-running ``git cat-file --batch``."""
+        if self._batch is None:
+            self._batch = subprocess.Popen(
+                ["git", "-C", str(self.path), "cat-file", "--batch"],
+                stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE,
+            )
+        stdin: IO[bytes] = self._batch.stdin  # type: ignore[assignment]
+        stdout: IO[bytes] = self._batch.stdout  # type: ignore[assignment]
+        stdin.write(sha.encode() + b"\n")
+        stdin.flush()
+        header = stdout.readline().decode().split()
+        if len(header) < 3 or header[1] != "blob":
+            raise KeyError(f"blob {sha} not found")
+        data = stdout.read(int(header[2]))
+        stdout.read(1)  # trailing newline
+        return data
+
+    def close(self) -> None:
+        if self._batch is not None:
+            if self._batch.stdin:
+                self._batch.stdin.close()
+            self._batch.wait()
+            if self._batch.stdout:
+                self._batch.stdout.close()
+            self._batch = None

research/finding_identity/linemap.py

@@ -0,0 +1,64 @@
+"""Map lines of an old file version to lines of the new version (the ground truth).
+
+Lines the diff keeps unchanged map exactly ("equal"). Inside a replaced block, old and
+new lines are paired when they are similar enough ("edited"): an order-preserving
+alignment that maximizes total similarity, counting only pairs at or above the
+threshold. Deleted lines and lines with no similar partner are not mapped.
+"""
+
+from __future__ import annotations
+
+import difflib
+from collections.abc import Sequence
+
+EDIT_THRESHOLD = 0.6
+#: Replaced blocks larger than this (old lines x new lines) are aligned position by
+#: position instead of with the full alignment, to keep the cost bounded.
+MAX_ALIGN_CELLS = 4000
+
+
+def similarity(a: str, b: str) -> float:
+    a, b = a.strip(), b.strip()
+    if not a and not b:
+        return 1.0
+    matcher = difflib.SequenceMatcher(None, a, b, autojunk=False)
+    if matcher.real_quick_ratio() < EDIT_THRESHOLD or matcher.quick_ratio() < EDIT_THRESHOLD:
+        return 0.0
+    return matcher.ratio()
+
+
+def _align(old: Sequence[str], new: Sequence[str]) -> list[tuple[int, int]]:
+    n, m = len(old), len(new)
+    if n * m > MAX_ALIGN_CELLS:
+        return [(k, k) for k in range(min(n, m)) if similarity(old[k], new[k]) >= EDIT_THRESHOLD]
+    sim = [[similarity(old[i], new[j]) for j in range(m)] for i in range(n)]
+    best = [[0.0] * (m + 1) for _ in range(n + 1)]
+    for i in range(n - 1, -1, -1):
+        for j in range(m - 1, -1, -1):
+            take = sim[i][j] + best[i + 1][j + 1] if sim[i][j] >= EDIT_THRESHOLD else -1.0
+            best[i][j] = max(take, best[i + 1][j], best[i][j + 1])
+    pairs: list[tuple[int, int]] = []
+    i = j = 0
+    while i < n and j < m:
+        if sim[i][j] >= EDIT_THRESHOLD and best[i][j] == sim[i][j] + best[i + 1][j + 1]:
+            pairs.append((i, j))
+            i, j = i + 1, j + 1
+        elif best[i][j] == best[i + 1][j]:
+            i += 1
+        else:
+            j += 1
+    return pairs
+
+
+def map_lines(old: Sequence[str], new: Sequence[str]) -> dict[int, tuple[int, str]]:
+    """``{old line: (new line, "equal" | "edited")}`` with 1-based line numbers."""
+    out: dict[int, tuple[int, str]] = {}
+    matcher = difflib.SequenceMatcher(None, old, new, autojunk=False)
+    for tag, i1, i2, j1, j2 in matcher.get_opcodes():
+        if tag == "equal":
+            for k in range(i2 - i1):
+                out[i1 + k + 1] = (j1 + k + 1, "equal")
+        elif tag == "replace":
+            for i, j in _align(old[i1:i2], new[j1:j2]):
+                out[i1 + i + 1] = (j1 + j + 1, "edited")
+    return out