Accept known findings through a baseline file

docs/usage.md

@@ -52,6 +52,7 @@ coretrace-python-analyzer [--check | --emit-ir [--ssa]] [options] [path]
 | `--advisories FILE` | Read a local advisory file in addition to `advisories.json` at the root. Repeatable. |
 | `--policy FILE` | Apply this dependency policy instead of `coretrace-policy.toml` at the root. |
 | `--import-advisories SRC OUT` | Convert an OSV dump into the local advisory file `OUT`, then exit. |
+| `--baseline PATH` | The accepted findings: written on the first run, then only findings not recorded there fail the check. |
 | `--emit-ir` | Print the intermediate representation of `path` instead of checking it. |
 | `--ssa` | With `--emit-ir`, print the static single assignment form. |
 | `--help` | Show the options and exit. |
@@ -154,6 +155,23 @@ finding. Suppressed findings are kept apart: the text report counts them, the JS
 lists them under `suppressed`, the SARIF log marks them as suppressed in source, and
 they never affect the exit status.
 
+## Adopting the analyzer on an existing code base
+
+A repository with known findings starts from a baseline:
+
+```bash
+coretrace-python-analyzer --check src/ --baseline coretrace-baseline.json
+```
+
+The first run writes the file with every current finding and passes. Later runs set the
+recorded findings apart and fail only on new ones. A finding is recognised by its file,
+its rule, its function and the text of its line, not by its line number, so code inserted
+above it does not make it new; a change to the line itself does. Commit the file and
+shrink it as findings are fixed: an entry without a matching finding is simply unused.
+Baselined findings are counted in the text report, listed under `baselined` in the JSON
+report and marked `baselineState: unchanged` in the SARIF log, where new results are
+marked `new`.
+
 ## Reports
 
 `--format` accepts `text`, `json` and `sarif`.

src/coretrace_python/cli.py

@@ -10,6 +10,7 @@
 from coretrace_python.cache import ProjectCache
 from coretrace_python.cfg import CFGError
 from coretrace_python.dependency import dump_advisories, import_osv, read_osv, render_sbom
+from coretrace_python.findings.baseline import Baseline, BaselineError
 from coretrace_python.frontend import HIRBuildError, ParseError, build_hir
 from coretrace_python.ir.lowering import LoweringError, lower_module
 from coretrace_python.ir.printer import format_module
@@ -24,6 +25,7 @@
 EXIT_ERROR = 2
 
 _ANALYSIS_ERRORS = (
+    BaselineError,
     OSError,
     UnicodeError,
     ParseError,
@@ -120,6 +122,14 @@ def build_parser() -> argparse.ArgumentParser:
         help="with --check on a directory, the dependency policy to apply instead of "
         "coretrace-policy.toml at the root",
     )
+    parser.add_argument(
+        "--baseline",
+        type=Path,
+        default=None,
+        metavar="PATH",
+        help="with --check, the accepted findings: written on the first run, then only "
+        "findings not recorded there fail the check",
+    )
     parser.add_argument(
         "--import-advisories",
         nargs=2,
@@ -170,6 +180,9 @@ def main(argv: list[str] | None = None) -> int:
         print("error: --policy only applies to --check on a directory", file=sys.stderr)
         return EXIT_ERROR
 
+    if args.baseline is not None and not args.check:
+        print("error: --baseline only applies to --check", file=sys.stderr)
+        return EXIT_ERROR
     if args.no_bundled_plugins and not args.check:
         print("error: --no-bundled-plugins only applies to --check", file=sys.stderr)
         return EXIT_ERROR
@@ -207,8 +220,15 @@ def main(argv: list[str] | None = None) -> int:
                 findings, coverage = file_analysis.findings, file_analysis.coverage
                 suppressed = file_analysis.suppressed
             root = args.path.resolve() if args.path.is_dir() else args.path.resolve().parent
-            rendered = render(args.format or "text", engine.report(findings, coverage, root, suppressed))
-            print(rendered, end="")
+            baselined = None
+            if args.baseline is not None:
+                if args.baseline.is_file():
+                    findings, baselined = Baseline.load(args.baseline).partition(findings, root)
+                else:
+                    Baseline.of(findings, root).save(args.baseline)
+                    findings, baselined = (), findings
+            report = engine.report(findings, coverage, root, suppressed, baselined)
+            print(render(args.format or "text", report), end="")
             return EXIT_FINDINGS if findings else EXIT_CLEAN
     except _ANALYSIS_ERRORS as error:
         print(f"error: {error}", file=sys.stderr)

src/coretrace_python/engine.py

@@ -713,7 +713,18 @@ def report(
     coverage: Coverage | None = None,
     root: Path | None = None,
     suppressed: Sequence[Finding] = (),
+    baselined: Sequence[Finding] | None = None,
 ) -> Report:
-    """The report of a check; paths under ``root`` are rendered relative to it."""
+    """The report of a check; paths under ``root`` are rendered relative to it.
+    ``baselined`` is the findings a baseline accounts for, ``None`` when none applied."""
 
-    return Report(tuple(findings), TOOL_NAME, __version__, coverage, root, tuple(suppressed))
+    return Report(
+        tuple(findings),
+        TOOL_NAME,
+        __version__,
+        coverage,
+        root,
+        tuple(suppressed),
+        tuple(baselined or ()),
+        baselined is not None,
+    )

src/coretrace_python/findings/baseline.py

@@ -0,0 +1,99 @@
+"""The baseline: findings accepted at a point in time, so only new ones fail a check.
+
+A finding is recognised by its file relative to the root, its rule, its function and the
+text of its line, never by its line number: code inserted above it does not make it
+new, a change to the line itself does. Entries are counted, so two identical findings on
+identical lines need two entries.
+"""
+
+from __future__ import annotations
+
+import json
+from collections import Counter
+from collections.abc import Iterable
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from coretrace_python.findings.model import Finding
+from coretrace_python.source import decode_text
+
+BASELINE_SCHEMA = 1
+
+Fingerprint = tuple[str, str, str, str]
+
+
+class BaselineError(Exception):
+    """The baseline file cannot be read."""
+
+
+def fingerprint(finding: Finding, root: Path) -> Fingerprint:
+    path = Path(str(finding.span.source_id))
+    try:
+        relative = path.relative_to(root).as_posix()
+    except ValueError:
+        relative = path.as_posix()
+    return (relative, finding.rule_id, finding.function or "", _line_text(path, finding.span.start_line))
+
+
+def _line_text(path: Path, line: int) -> str:
+    try:
+        lines = decode_text(path.read_bytes()).splitlines()
+    except (OSError, UnicodeDecodeError):
+        return str(line)
+    if not 1 <= line <= len(lines):
+        return str(line)
+    return lines[line - 1].strip()
+
+
+@dataclass(frozen=True)
+class Baseline:
+    entries: Counter[Fingerprint] = field(default_factory=Counter)
+
+    @classmethod
+    def of(cls, findings: Iterable[Finding], root: Path) -> Baseline:
+        return cls(Counter(fingerprint(finding, root) for finding in findings))
+
+    @classmethod
+    def load(cls, path: Path) -> Baseline:
+        try:
+            document = json.loads(path.read_text(encoding="utf-8"))
+            if document.get("schema") != BASELINE_SCHEMA:
+                raise BaselineError(f"{path}: unsupported baseline schema {document.get('schema')!r}")
+            entries = Counter(
+                {
+                    (str(e["path"]), str(e["rule"]), str(e["function"]), str(e["line"])): int(e["count"])
+                    for e in document["findings"]
+                }
+            )
+        except BaselineError:
+            raise
+        except (OSError, ValueError, KeyError, TypeError, AttributeError) as error:
+            raise BaselineError(f"{path}: {error}") from error
+        return cls(entries)
+
+    def save(self, path: Path) -> None:
+        document = {
+            "schema": BASELINE_SCHEMA,
+            "findings": [
+                {"path": p, "rule": r, "function": f, "line": line, "count": count}
+                for (p, r, f, line), count in sorted(self.entries.items())
+            ],
+        }
+        path.write_text(json.dumps(document, indent=2) + "\n", encoding="utf-8")
+
+    def partition(
+        self, findings: Iterable[Finding], root: Path
+    ) -> tuple[tuple[Finding, ...], tuple[Finding, ...]]:
+        """The findings not in the baseline, and those it accounts for."""
+
+        remaining = Counter(self.entries)
+        new: list[Finding] = []
+        baselined: list[Finding] = []
+        for finding in findings:
+            key = fingerprint(finding, root)
+            if remaining[key] > 0:
+                remaining[key] -= 1
+                baselined.append(finding)
+            else:
+                new.append(finding)
+        return tuple(new), tuple(baselined)

src/coretrace_python/reporters/json_format.py

@@ -38,6 +38,8 @@ def render_json(report: Report) -> str:
     document["findings"] = [finding_record(finding, report) for finding in report.findings]
     if report.suppressed:
         document["suppressed"] = [finding_record(finding, report) for finding in report.suppressed]
+    if report.baselined:
+        document["baselined"] = [finding_record(finding, report) for finding in report.baselined]
     if report.coverage is not None:
         coverage = report.coverage
         document["coverage"] = {

src/coretrace_python/reporters/report.py

@@ -23,10 +23,14 @@ class Report:
     root: Path | None = None
     # Findings silenced by an inline suppression; reported apart, never counted.
     suppressed: tuple[Finding, ...] = ()
+    # Findings the baseline accounts for; reported apart, never counted.
+    baselined: tuple[Finding, ...] = ()
+    # Whether a baseline was applied, so new findings can be marked as such.
+    with_baseline: bool = False
 
     def __post_init__(self) -> None:
-        object.__setattr__(self, "findings", tuple(sorted(self.findings, key=_order)))
-        object.__setattr__(self, "suppressed", tuple(sorted(self.suppressed, key=_order)))
+        for name in ("findings", "suppressed", "baselined"):
+            object.__setattr__(self, name, tuple(sorted(getattr(self, name), key=_order)))
 
     def locate(self, path: str) -> str:
         """``path`` relative to the root, POSIX style, when it lies under the root."""

src/coretrace_python/reporters/sarif.py

@@ -27,7 +27,11 @@ def _artifact(report: Report, path: str) -> dict[str, str]:
 
 
 def _result(
-    report: Report, finding: Finding, rule_index: int, suppressed: bool = False
+    report: Report,
+    finding: Finding,
+    rule_index: int,
+    suppressed: bool = False,
+    baseline_state: str | None = None,
 ) -> dict[str, object]:
     span = finding.span
     region: dict[str, int] = {"startLine": span.start_line, "startColumn": span.start_column}
@@ -50,14 +54,17 @@ def _result(
     }
     if suppressed:
         result["suppressions"] = [{"kind": "inSource"}]
+    if baseline_state is not None:
+        result["baselineState"] = baseline_state
     return result
 
 
 def render_sarif(report: Report) -> str:
     rule_ids: list[str] = []
-    for finding in (*report.findings, *report.suppressed):
+    for finding in (*report.findings, *report.suppressed, *report.baselined):
         if finding.rule_id not in rule_ids:
             rule_ids.append(finding.rule_id)
+    new = "new" if report.with_baseline else None
     run: dict[str, object] = {}
     if report.root is not None:
         run["originalUriBaseIds"] = {SRCROOT: {"uri": report.root.as_uri() + "/"}}
@@ -74,8 +81,12 @@ def render_sarif(report: Report) -> str:
                     }
                 },
             "results": [
-                *(_result(report, f, rule_ids.index(f.rule_id)) for f in report.findings),
+                *(_result(report, f, rule_ids.index(f.rule_id), False, new) for f in report.findings),
                 *(_result(report, f, rule_ids.index(f.rule_id), True) for f in report.suppressed),
+                *(
+                    _result(report, f, rule_ids.index(f.rule_id), False, "unchanged")
+                    for f in report.baselined
+                ),
             ],
         }
     )

src/coretrace_python/reporters/text.py

@@ -20,6 +20,8 @@ def render_text(report: Report) -> str:
     summary = "no findings" if count == 0 else f"{count} finding{'s' if count > 1 else ''}"
     if report.suppressed:
         summary += f", {len(report.suppressed)} suppressed"
+    if report.baselined:
+        summary += f", {len(report.baselined)} baselined"
     lines.append(summary)
     if report.coverage is not None:
         lines.append(report.coverage.summary())