diff --git a/stellargate/cli.py b/stellargate/cli.py index ca49bfe..87422a5 100644 --- a/stellargate/cli.py +++ b/stellargate/cli.py @@ -5,10 +5,11 @@ import json import logging 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 logger = logging.getLogger("stellargate") @@ -58,6 +59,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) @@ -95,30 +104,80 @@ def _run(args: argparse.Namespace) -> int: results = run_all(config) findings = all_findings(results) - passed = gate_passed(findings, fail_on) - - for r in results: - if r.error: - logger.warning("Tool %s errored: %s", r.tool, r.error) - else: - logger.info("Tool %s produced %d finding(s)", r.tool, len(r.findings)) + 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 + ] - logger.info("Gate result: %s (fail_on=%s)", "passed" if passed else "failed", fail_on) + passed = gate_passed(findings, fail_on) - # The Markdown report is the tool's primary output: it must go to stdout - # verbatim so CI/scripts can capture it. Diagnostics use logging (stderr). - 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 46c194b..0d8a430 100644 --- a/tests/test_aggregator_and_report.py +++ b/tests/test_aggregator_and_report.py @@ -1,3 +1,4 @@ +import json from unittest.mock import patch from hypothesis import given, settings @@ -5,8 +6,14 @@ 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.schema import SEVERITY_ORDER, Finding +from stellargate.report import ( + diff_findings, + finding_key, + gate_passed, + to_json, + to_markdown, +) +from stellargate.schema import Finding def make_results():