diff --git a/docs/usage.md b/docs/usage.md index f7e6ff2..b561dd6 100644 --- a/docs/usage.md +++ b/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`. diff --git a/src/coretrace_python/cli.py b/src/coretrace_python/cli.py index 476e3d3..bcddce6 100644 --- a/src/coretrace_python/cli.py +++ b/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) diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index d49dfbb..1b3b90f 100644 --- a/src/coretrace_python/engine.py +++ b/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, + ) diff --git a/src/coretrace_python/findings/baseline.py b/src/coretrace_python/findings/baseline.py new file mode 100644 index 0000000..0c22f0d --- /dev/null +++ b/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) diff --git a/src/coretrace_python/reporters/json_format.py b/src/coretrace_python/reporters/json_format.py index 362681d..a928458 100644 --- a/src/coretrace_python/reporters/json_format.py +++ b/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"] = { diff --git a/src/coretrace_python/reporters/report.py b/src/coretrace_python/reporters/report.py index 0c447c8..90f4a81 100644 --- a/src/coretrace_python/reporters/report.py +++ b/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.""" diff --git a/src/coretrace_python/reporters/sarif.py b/src/coretrace_python/reporters/sarif.py index 2ed4223..c752da5 100644 --- a/src/coretrace_python/reporters/sarif.py +++ b/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 + ), ], } ) diff --git a/src/coretrace_python/reporters/text.py b/src/coretrace_python/reporters/text.py index 8e57668..5c392c8 100644 --- a/src/coretrace_python/reporters/text.py +++ b/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()) diff --git a/tests/test_baseline.py b/tests/test_baseline.py new file mode 100644 index 0000000..ab94bc2 --- /dev/null +++ b/tests/test_baseline.py @@ -0,0 +1,159 @@ +"""Acceptance tests for the baseline file (issue #68). + +``--baseline PATH`` records the findings of a check the first time and, on later checks, +sets apart the findings already recorded so that only new ones fail the check. A finding +is recognised by its file relative to the root, its rule, its function and the text of +its line, not by its line number, so inserting code above it does not make it new. The +text report counts baselined findings, the JSON report lists them, the SARIF log marks +every result's ``baselineState``. + +Expected to remain red until ``coretrace_python.findings.baseline`` and ``--baseline`` exist. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from coretrace_python.cli import main +from coretrace_python.findings import Confidence, Finding, Severity +from coretrace_python.source import SourceId, SourceSpan + +try: + from coretrace_python.findings.baseline import Baseline, BaselineError, fingerprint +except ImportError as error: # pragma: no cover - red until the baseline lands + MISSING = error +else: + MISSING = None + + +@pytest.fixture(autouse=True) +def require_baseline() -> None: + if MISSING is not None: + pytest.fail(f"the baseline is not implemented yet: {MISSING}") + + +EVAL = "import hashlib\n\ndef run(code):\n eval(code)\n" +EVAL_AND_MD5 = "import hashlib\n\ndef run(code):\n eval(code)\n hashlib.md5(code)\n" +EVAL_SHIFTED = "import hashlib\n\n\ndef run(code):\n\n eval(code)\n" + + +def finding(path: str, line: int, rule: str = "dangerous-eval") -> Finding: + return Finding(rule, "m", Severity.HIGH, Confidence.HIGH, SourceSpan(SourceId(path), line, 5, line, 9), "run") + + +# --------------------------------------------------------------------------- fingerprints + + +def test_fingerprint_ignores_the_line_number_but_not_the_line_text(tmp_path: Path) -> None: + source = tmp_path / "app.py" + source.write_text(EVAL, encoding="utf-8") + before = fingerprint(finding(str(source), 4), tmp_path) + source.write_text(EVAL_SHIFTED, encoding="utf-8") + after = fingerprint(finding(str(source), 6), tmp_path) + source.write_text(EVAL.replace("eval(code)", "eval(code.strip())"), encoding="utf-8") + changed = fingerprint(finding(str(source), 4), tmp_path) + + assert before == after == ("app.py", "dangerous-eval", "run", "eval(code)") + assert changed != before + assert fingerprint(finding(str(source), 4, "weak-crypto"), tmp_path) != before + + +def test_fingerprint_falls_back_to_the_line_number_without_the_file(tmp_path: Path) -> None: + assert fingerprint(finding(str(tmp_path / "gone.py"), 7), tmp_path) == ("gone.py", "dangerous-eval", "run", "7") + + +def test_baseline_round_trips_and_partitions_with_counts(tmp_path: Path) -> None: + source = tmp_path / "app.py" + source.write_text(EVAL, encoding="utf-8") + first, second = finding(str(source), 4), finding(str(source), 4) + path = tmp_path / "baseline.json" + + Baseline.of((first,), tmp_path).save(path) + loaded = Baseline.load(path) + new, baselined = loaded.partition((first, second), tmp_path) + + assert json.loads(path.read_text(encoding="utf-8"))["schema"] == 1 + assert baselined == (first,) and new == (second,) + + +def test_corrupt_baseline_files_are_reported(tmp_path: Path) -> None: + path = tmp_path / "baseline.json" + path.write_text("{not json", encoding="utf-8") + + with pytest.raises(BaselineError): + Baseline.load(path) + + +# --------------------------------------------------------------------------- CLI + + +def project(root: Path, text: str) -> Path: + (root / "src").mkdir(exist_ok=True) + (root / "src" / "app.py").write_text(text, encoding="utf-8") + return root / "src" + + +def test_the_first_run_records_the_baseline_and_passes(tmp_path: Path, capsys) -> None: # type: ignore[no-untyped-def] + root = project(tmp_path, EVAL) + baseline = tmp_path / "baseline.json" + + assert main(["--check", str(root), "--baseline", str(baseline)]) == 0 + out = capsys.readouterr().out + + assert baseline.is_file() + assert out == "no findings, 1 baselined\ncoverage: 1/1 files, 1/1 functions\n" + + +def test_only_new_findings_fail_later_runs(tmp_path: Path, capsys) -> None: # type: ignore[no-untyped-def] + root = project(tmp_path, EVAL) + baseline = tmp_path / "baseline.json" + main(["--check", str(root), "--baseline", str(baseline)]) + capsys.readouterr() + + assert main(["--check", str(root), "--baseline", str(baseline)]) == 0 + assert capsys.readouterr().out.startswith("no findings, 1 baselined\n") + + project(tmp_path, EVAL_SHIFTED) + assert main(["--check", str(root), "--baseline", str(baseline)]) == 0 + assert capsys.readouterr().out.startswith("no findings, 1 baselined\n") + + project(tmp_path, EVAL_AND_MD5) + assert main(["--check", str(root), "--baseline", str(baseline)]) == 1 + out = capsys.readouterr().out + assert out.startswith("app.py:5:5: medium weak-crypto:") + assert "1 finding, 1 baselined\n" in out + assert "dangerous-eval" not in out + + +def test_json_and_sarif_reports_carry_the_baseline_state(tmp_path: Path, capsys) -> None: # type: ignore[no-untyped-def] + root = project(tmp_path, EVAL) + baseline = tmp_path / "baseline.json" + main(["--check", str(root), "--baseline", str(baseline)]) + capsys.readouterr() + project(tmp_path, EVAL_AND_MD5) + + assert main(["--check", str(root), "--baseline", str(baseline), "--format", "json"]) == 1 + document = json.loads(capsys.readouterr().out) + assert [f["rule_id"] for f in document["findings"]] == ["weak-crypto"] + assert [f["rule_id"] for f in document["baselined"]] == ["dangerous-eval"] + + assert main(["--check", str(root), "--baseline", str(baseline), "--format", "sarif"]) == 1 + results = json.loads(capsys.readouterr().out)["runs"][0]["results"] + assert {(r["ruleId"], r["baselineState"]) for r in results} == { + ("weak-crypto", "new"), + ("dangerous-eval", "unchanged"), + } + + +def test_baseline_requires_a_check_and_a_readable_file(tmp_path: Path, capsys) -> None: # type: ignore[no-untyped-def] + root = project(tmp_path, EVAL) + assert main(["--emit-ir", "--baseline", str(tmp_path / "b.json"), str(root / "app.py")]) == 2 + assert "--baseline" in capsys.readouterr().err + + corrupt = tmp_path / "corrupt.json" + corrupt.write_text("{not json", encoding="utf-8") + assert main(["--check", str(root), "--baseline", str(corrupt)]) == 2 + assert "corrupt.json" in capsys.readouterr().err diff --git a/tests/test_docs.py b/tests/test_docs.py index 43b3637..9a91e4d 100644 --- a/tests/test_docs.py +++ b/tests/test_docs.py @@ -45,7 +45,7 @@ def shipped_rules() -> set[str]: rules: set[str] = set() for path in (REPO / "src" / "coretrace_python").rglob("*.py"): text = path.read_text(encoding="utf-8") - rules.update(re.findall(r'rule_id[^"\n]{0,40}"([a-z-]+)"', text)) + rules.update(re.findall(r'rule_id(?:: ClassVar\[str\])?\s*=\s*"([a-z-]+)"', text)) rules.update(re.findall(r'Finding\(\s*"([a-z]+-[a-z-]+)"', text)) rules.update(re.findall(r'_note\("([a-z-]+)"', text)) return rules - {"message"}