From b3c1a0ee74222152a0216b508aeb513588a459be Mon Sep 17 00:00:00 2001 From: qiaobochi040726-source Date: Thu, 6 Aug 2026 05:29:38 +0800 Subject: [PATCH] Add --diff-only mode for gating on newly introduced findings vs a baseline report --- stellargate/cli.py | 79 +++++++++++++- stellargate/report.py | 46 +++++++- tests/test_aggregator_and_report.py | 163 +++++++++++++++++++++++++++- 3 files changed, 279 insertions(+), 9 deletions(-) diff --git a/stellargate/cli.py b/stellargate/cli.py index 1a6c532..e1d6389 100644 --- a/stellargate/cli.py +++ b/stellargate/cli.py @@ -4,10 +4,11 @@ import argparse import json import sys +from pathlib import Path -from stellargate.aggregator import all_findings, run_all +from stellargate.aggregator import ToolRunResult, all_findings, run_all from stellargate.config import VALID_SEVERITIES, Config, ConfigError -from stellargate.report import gate_passed, to_json, to_markdown +from stellargate.report import diff_findings, finding_key, gate_passed, to_json, to_markdown def main(argv: list[str] | None = None) -> int: @@ -19,6 +20,14 @@ def main(argv: list[str] | None = None) -> int: run_parser.add_argument("--json-report", default=None, help="Write JSON report to this path") run_parser.add_argument("--md-report", default=None, help="Write Markdown report to this path") run_parser.add_argument("--fail-on", default=None, help="Override fail_on threshold from config") + run_parser.add_argument( + "--diff-only", + default=None, + metavar="BASELINE.json", + help=("Gate only on findings newly introduced since a baseline StellarGate " + "JSON report. Loads the baseline, keeps only findings not present in it, " + "and bases both the gate and the report on those new findings."), + ) args = parser.parse_args(argv) @@ -44,20 +53,80 @@ def _run(args: argparse.Namespace) -> int: results = run_all(config) findings = all_findings(results) + diff_mode = False + + if args.diff_only: + baseline = _load_baseline(args.diff_only) + if baseline is None: + return 2 + diff_mode = True + findings = diff_findings(findings, baseline) + keep = {finding_key(f) for f in findings} + results = [ + ToolRunResult( + r.tool, + [f for f in r.findings if finding_key(f) in keep], + r.error, + ) + for r in results + ] + passed = gate_passed(findings, fail_on) - print(to_markdown(results, fail_on, passed)) + print(to_markdown(results, fail_on, passed, diff_mode)) if args.json_report: with open(args.json_report, "w") as f: - json.dump(to_json(results, fail_on, passed), f, indent=2) + json.dump(to_json(results, fail_on, passed, diff_mode), f, indent=2) if args.md_report: with open(args.md_report, "w") as f: - f.write(to_markdown(results, fail_on, passed)) + f.write(to_markdown(results, fail_on, passed, diff_mode)) return 0 if passed else 1 +def _load_baseline(path: str) -> dict | None: + """Load and validate a baseline StellarGate JSON report. + + Returns the parsed report on success, or None after printing a clear + error to stderr. None signals the caller to exit non-zero — a missing + or corrupt baseline is never silently treated as an empty baseline. + """ + p = Path(path) + if not p.exists(): + print(f"diff-only: baseline report not found: {path}", file=sys.stderr) + return None + try: + with open(p) as f: + report = json.load(f) + except json.JSONDecodeError as e: + print(f"diff-only: baseline {path} is not valid JSON: {e}", file=sys.stderr) + return None + except OSError as e: + print(f"diff-only: cannot read baseline {path}: {e}", file=sys.stderr) + return None + + if not isinstance(report, dict) or "findings" not in report: + print( + f"diff-only: {path} is not a StellarGate report " + "(expected an object with a 'findings' list)", + file=sys.stderr, + ) + return None + if not isinstance(report["findings"], list): + print(f"diff-only: baseline {path} has a non-list 'findings' field", file=sys.stderr) + return None + for i, item in enumerate(report["findings"]): + if not isinstance(item, dict) or "tool" not in item or "rule_id" not in item: + print( + f"diff-only: baseline {path} finding[{i}] is malformed " + "(each finding needs 'tool' and 'rule_id')", + file=sys.stderr, + ) + return None + return report + + if __name__ == "__main__": sys.exit(main()) diff --git a/stellargate/report.py b/stellargate/report.py index e32127a..8e72df6 100644 --- a/stellargate/report.py +++ b/stellargate/report.py @@ -2,12 +2,45 @@ from __future__ import annotations from datetime import datetime, timezone +from typing import Any from stellargate.aggregator import ToolRunResult from stellargate.schema import SEVERITY_ORDER, Finding -def to_json(results: list[ToolRunResult], fail_on: str, gate_passed: bool) -> dict: +def finding_key(finding: "Finding | dict[str, Any]") -> tuple: + """Uniqueness key for a finding: (tool, rule_id, location). + + Accepts either a live Finding object or a dict as produced by + Finding.to_dict() (i.e. entries inside the JSON report's "findings" + list), so the same key works for diffing current results against a + previously serialized baseline report. + """ + if isinstance(finding, Finding): + return (finding.tool, finding.rule_id, finding.location) + return (finding["tool"], finding["rule_id"], finding.get("location")) + + +def diff_findings( + current: list[Finding], baseline_report: dict[str, Any] +) -> list[Finding]: + """Return only findings newly introduced since a baseline report. + + A finding is considered pre-existing if its (tool, rule_id, location) + key already appears in the baseline report's "findings" list. Only + brand-new findings are returned — this is what makes gating on + regressions possible without penalizing every historical finding. + """ + baseline_keys = {finding_key(f) for f in baseline_report.get("findings", [])} + return [f for f in current if finding_key(f) not in baseline_keys] + + +def to_json( + results: list[ToolRunResult], + fail_on: str, + gate_passed: bool, + diff_mode: bool = False, +) -> dict: findings = [f for r in results for f in r.findings] counts = {sev: 0 for sev in SEVERITY_ORDER} for f in findings: @@ -16,6 +49,7 @@ def to_json(results: list[ToolRunResult], fail_on: str, gate_passed: bool) -> di return { "generated_at": datetime.now(timezone.utc).isoformat(), "fail_on": fail_on, + "diff": bool(diff_mode), "passed": gate_passed, "summary": counts, "tools": [ @@ -30,12 +64,18 @@ def to_json(results: list[ToolRunResult], fail_on: str, gate_passed: bool) -> di } -def to_markdown(results: list[ToolRunResult], fail_on: str, gate_passed: bool) -> str: +def to_markdown( + results: list[ToolRunResult], + fail_on: str, + gate_passed: bool, + diff_mode: bool = False, +) -> str: findings = [f for r in results for f in r.findings] lines: list[str] = [] status = "✅ PASSED" if gate_passed else "❌ FAILED" - lines.append(f"# StellarGate Compliance Report — {status}\n") + mode_label = " — diff mode (only new findings vs baseline)" if diff_mode else "" + lines.append(f"# StellarGate Compliance Report — {status}{mode_label}\n") lines.append(f"Threshold: fail on **{fail_on}** or above.\n") lines.append("## Per-tool summary\n") diff --git a/tests/test_aggregator_and_report.py b/tests/test_aggregator_and_report.py index 842c5cf..42e87a0 100644 --- a/tests/test_aggregator_and_report.py +++ b/tests/test_aggregator_and_report.py @@ -1,8 +1,15 @@ +import json from unittest.mock import patch from stellargate.aggregator import ToolRunResult, all_findings, run_all from stellargate.config import Config, ToolConfig -from stellargate.report import gate_passed, to_json, to_markdown +from stellargate.report import ( + diff_findings, + finding_key, + gate_passed, + to_json, + to_markdown, +) from stellargate.schema import Finding @@ -82,3 +89,157 @@ def test_run_all_survives_an_unexpected_adapter_exception(): assert results[0].error is not None assert "unexpected adapter error" in results[0].error # crucially: run_all itself did not raise + + +def test_finding_key_matches_live_objects_and_json_dicts(): + f = Finding("rytscan", "AUTH-001", "high", "no auth check", "vault.rs:42") + assert finding_key(f) == ("rytscan", "AUTH-001", "vault.rs:42") + assert finding_key(f.to_dict()) == finding_key(f) + + +def test_finding_key_is_location_sensitive(): + a = Finding("rytscan", "AUTH-001", "high", "x", "vault.rs:42") + b = Finding("rytscan", "AUTH-001", "high", "x", "vault.rs:99") + assert finding_key(a) != finding_key(b) + + +def test_diff_findings_drops_baseline_findings_keeps_new(): + current = [ + Finding("rytscan", "AUTH-001", "high", "no auth check", "vault.rs:42"), + Finding("rytscan", "AUTH-002", "high", "new rule hit", "vault.rs:99"), + ] + baseline = { + "findings": [ + {"tool": "rytscan", "rule_id": "AUTH-001", "location": "vault.rs:42"}, + ] + } + new = diff_findings(current, baseline) + assert [f.rule_id for f in new] == ["AUTH-002"] + + +def test_diff_findings_ignores_severity_message_changes_on_same_key(): + """Same (tool, rule_id, location) with a reworded message is not 'new'.""" + current = [Finding("rytscan", "AUTH-001", "high", "rewritten message", "vault.rs:42")] + baseline = { + "findings": [ + {"tool": "rytscan", "rule_id": "AUTH-001", "location": "vault.rs:42", + "severity": "high", "message": "old wording"}, + ] + } + assert diff_findings(current, baseline) == [] + + +def test_diff_gate_ignores_pre_existing_findings(): + """The whole point of --diff-only: gating on regressions, not history.""" + results = make_results() # AUTH-001 (high) + STELLAR-001 (critical) + baseline = { + "findings": [ + {"tool": "rytscan", "rule_id": "AUTH-001", "location": "vault.rs:42"}, + {"tool": "vaultsweep", "rule_id": "STELLAR-001", "location": ".env:3"}, + ] + } + findings = diff_findings(all_findings(results), baseline) + assert findings == [] + assert gate_passed(findings, "high") is True + + +def test_diff_gate_fails_on_newly_introduced_critical(): + current = [ + Finding("rytscan", "AUTH-001", "high", "no auth check", "vault.rs:42"), + Finding("vaultsweep", "STELLAR-999", "critical", "new leak", ".env:9"), + ] + baseline = { + "findings": [ + {"tool": "rytscan", "rule_id": "AUTH-001", "location": "vault.rs:42"}, + ] + } + new = diff_findings(current, baseline) + assert [f.rule_id for f in new] == ["STELLAR-999"] + assert gate_passed(new, "high") is False + + +def test_to_json_marks_diff_mode(): + results = make_results() + normal = to_json(results, "high", False) + diff = to_json(results, "high", False, diff_mode=True) + assert normal["diff"] is False + assert diff["diff"] is True + + +def test_to_markdown_annotates_diff_mode_in_title(): + results = make_results() + md = to_markdown(results, "high", False, diff_mode=True) + assert "diff mode" in md + assert "AUTH-001" in md + assert "STELLAR-001" in md + + +def _dummy_config(): + return Config(target=".", fail_on="high", tools={}) + + +def _run_cli(argv, capsys): + from stellargate import cli + + with patch("stellargate.cli.Config.load", return_value=_dummy_config()), patch( + "stellargate.cli.run_all", return_value=make_results() + ): + rc = cli.main(argv) + return rc, capsys.readouterr() + + +def test_cli_diff_only_passes_when_only_baseline_findings(tmp_path, capsys): + baseline = tmp_path / "baseline.json" + baseline.write_text( + json.dumps( + { + "findings": [ + {"tool": "rytscan", "rule_id": "AUTH-001", "location": "vault.rs:42"}, + {"tool": "vaultsweep", "rule_id": "STELLAR-001", "location": ".env:3"}, + ] + } + ) + ) + rc, captured = _run_cli( + ["run", "--config", "x.yaml", "--diff-only", str(baseline)], capsys + ) + assert rc == 0 # no new findings -> pass + assert "diff mode" in captured.out + assert "AUTH-001" not in captured.out + assert "STELLAR-001" not in captured.out + + +def test_cli_diff_only_fails_on_new_finding(tmp_path, capsys): + baseline = tmp_path / "baseline.json" + baseline.write_text( + json.dumps({"findings": [{"tool": "rytscan", "rule_id": "AUTH-001", "location": "vault.rs:42"}]}) + ) + rc, captured = _run_cli( + ["run", "--config", "x.yaml", "--diff-only", str(baseline)], capsys + ) + assert rc == 1 # STELLAR-001 is newly introduced -> gate fails + assert "STELLAR-001" in captured.out + assert "diff mode" in captured.out + + +def test_cli_diff_only_rejects_missing_baseline(tmp_path, capsys): + missing = tmp_path / "nope.json" + rc, captured = _run_cli(["run", "--config", "x.yaml", "--diff-only", str(missing)], capsys) + assert rc == 2 + assert "baseline report not found" in captured.err + + +def test_cli_diff_only_rejects_corrupt_baseline(tmp_path, capsys): + baseline = tmp_path / "baseline.json" + baseline.write_text("{not json") + rc, captured = _run_cli(["run", "--config", "x.yaml", "--diff-only", str(baseline)], capsys) + assert rc == 2 + assert "not valid JSON" in captured.err + + +def test_cli_diff_only_rejects_non_report_baseline(tmp_path, capsys): + baseline = tmp_path / "baseline.json" + baseline.write_text(json.dumps({"unrelated": True})) + rc, captured = _run_cli(["run", "--config", "x.yaml", "--diff-only", str(baseline)], capsys) + assert rc == 2 + assert "not a StellarGate report" in captured.err