feat: findings reporters and --check

README.md

@@ -24,6 +24,17 @@ python -m pytest
 python -m ruff check .
 ```
 
+## Check a file with plugins
+
+```bash
+coretrace-python-analyzer --check app.py --plugins plugins/
+coretrace-python-analyzer --check app.py --plugins plugins/ --format sarif > report.sarif
+```
+
+`--plugins` is a directory searched recursively for `plugin.toml` manifests and may be
+repeated. `--format` is `text` (default), `json` or `sarif`. Exit status is 0 when nothing
+was found, 1 when findings were reported, and 2 on a usage or analysis error.
+
 ## Emit PyIR
 
 ```bash

src/coretrace_python/cli.py

@@ -4,17 +4,40 @@
 import sys
 from pathlib import Path
 
+from coretrace_python import engine
+from coretrace_python.analysis import AnalysisError
+from coretrace_python.cfg import CFGError
 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
+from coretrace_python.plugins import IncompatiblePluginError, ManifestError
+from coretrace_python.reporters import FORMATS, render
 from coretrace_python.semantic.imports import ImportResolutionError
 from coretrace_python.semantic.scopes import ScopeError
 from coretrace_python.source import SourceManager
 
+EXIT_CLEAN = 0
+EXIT_FINDINGS = 1
+EXIT_ERROR = 2
+
+_ANALYSIS_ERRORS = (
+    OSError,
+    UnicodeError,
+    ParseError,
+    HIRBuildError,
+    ImportResolutionError,
+    ScopeError,
+    CFGError,
+    LoweringError,
+    AnalysisError,
+    ManifestError,
+    IncompatiblePluginError,
+)
+
 
 def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(
-        prog="coretrace-python-analyzer",
+        prog=engine.TOOL_NAME,
         description="Analyze Python source with CoreTrace.",
     )
     parser.add_argument("path", type=Path, help="Python source file to analyze")
@@ -28,32 +51,49 @@ def build_parser() -> argparse.ArgumentParser:
         action="store_true",
         help="with --emit-ir, print the static single assignment form",
     )
+    parser.add_argument(
+        "--check",
+        action="store_true",
+        help="run the loaded plugins and report their findings",
+    )
+    parser.add_argument(
+        "--plugins",
+        action="append",
+        type=Path,
+        default=[],
+        metavar="DIR",
+        help="with --check, a directory searched recursively for plugin.toml (repeatable)",
+    )
+    parser.add_argument(
+        "--format",
+        choices=sorted(FORMATS),
+        default=None,
+        help="with --check, the report format (default: text)",
+    )
     return parser
 
 
 def main(argv: list[str] | None = None) -> int:
     args = build_parser().parse_args(argv)
-    if not args.emit_ir:
-        print("error: no action selected; pass --emit-ir", file=sys.stderr)
-        return 2
+    if not (args.emit_ir or args.check):
+        print("error: no action selected; pass --emit-ir or --check", file=sys.stderr)
+        return EXIT_ERROR
+    if args.format is not None and not args.check:
+        print("error: --format only applies to --check", file=sys.stderr)
+        return EXIT_ERROR
 
     try:
         source = SourceManager().load_file(args.path)
-        module = lower_module(build_hir(source), ssa=args.ssa)
-    except (
-        OSError,
-        UnicodeError,
-        ParseError,
-        HIRBuildError,
-        ImportResolutionError,
-        ScopeError,
-        LoweringError,
-    ) as error:
+        if args.emit_ir:
+            print(format_module(lower_module(build_hir(source), ssa=args.ssa)))
+        if args.check:
+            findings = engine.check(source, args.plugins)
+            print(render(args.format or "text", engine.report(findings)), end="")
+            return EXIT_FINDINGS if findings else EXIT_CLEAN
+    except _ANALYSIS_ERRORS as error:
         print(f"error: {error}", file=sys.stderr)
-        return 1
-
-    print(format_module(module))
-    return 0
+        return EXIT_ERROR
+    return EXIT_CLEAN
 
 
 if __name__ == "__main__":

src/coretrace_python/engine.py

@@ -0,0 +1,66 @@
+"""Composition root: the full analysis DAG, plugin loading and one check run.
+
+This is the only module that knows every layer. The CLI and future hosts call it;
+plugins and analyses never import it.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from pathlib import Path
+
+from coretrace_python import __version__
+from coretrace_python.abstract import ConstantPropagation
+from coretrace_python.analysis import AnalysisManager, AnyAnalysis
+from coretrace_python.cfg import CFGAnalysis, DominanceAnalysis, PostDominanceAnalysis
+from coretrace_python.findings import Finding
+from coretrace_python.frontend import build_hir
+from coretrace_python.hir import nodes
+from coretrace_python.ir.defuse import DefUseAnalysis
+from coretrace_python.ir.lowering import ModuleIRAnalysis, PyIRAnalysis
+from coretrace_python.ir.ssa import SSAAnalysis
+from coretrace_python.plugins import PluginRegistry, discover_plugins, run_plugins
+from coretrace_python.reporters import Report
+from coretrace_python.semantic import SEMANTIC_ANALYSES
+from coretrace_python.source import SourceFile
+from coretrace_python.taint import SecurityModelAnalysis, SecurityModelRegistry, TaintAnalysis
+
+TOOL_NAME = "coretrace-python-analyzer"
+
+ALL_ANALYSES: tuple[AnyAnalysis, ...] = (
+    *SEMANTIC_ANALYSES,
+    CFGAnalysis,
+    DominanceAnalysis,
+    PostDominanceAnalysis,
+    PyIRAnalysis,
+    ModuleIRAnalysis,
+    SSAAnalysis,
+    DefUseAnalysis,
+    ConstantPropagation,
+    SecurityModelAnalysis,
+    TaintAnalysis,
+)
+
+
+def build_manager(module: nodes.Module, models: SecurityModelRegistry | None = None) -> AnalysisManager:
+    """A manager with every engine analysis registered and the security models provided."""
+
+    manager = AnalysisManager(module)
+    manager.register(*ALL_ANALYSES)
+    manager.provide(SecurityModelAnalysis, (models or SecurityModelRegistry()).freeze())
+    return manager
+
+
+def check(source: SourceFile, plugin_roots: Sequence[Path]) -> tuple[Finding, ...]:
+    """Run every plugin found under ``plugin_roots`` against ``source``."""
+
+    manager = build_manager(build_hir(source))
+    registry = PluginRegistry()
+    for root in plugin_roots:
+        for loaded in discover_plugins(root, manager):
+            registry.add(loaded)
+    return run_plugins(manager, [loaded.plugin for loaded in registry])
+
+
+def report(findings: Sequence[Finding]) -> Report:
+    return Report(tuple(findings), TOOL_NAME, __version__)

src/coretrace_python/reporters/__init__.py

@@ -0,0 +1,22 @@
+"""Reporters render normalized findings and never run an analysis (architecture §28)."""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Mapping
+from types import MappingProxyType
+
+from coretrace_python.reporters.json_format import render_json
+from coretrace_python.reporters.report import Report
+from coretrace_python.reporters.sarif import render_sarif
+from coretrace_python.reporters.text import render_text
+
+FORMATS: Mapping[str, Callable[[Report], str]] = MappingProxyType(
+    {"text": render_text, "json": render_json, "sarif": render_sarif}
+)
+
+
+def render(format_name: str, report: Report) -> str:
+    return FORMATS[format_name](report)
+
+
+__all__ = ["FORMATS", "Report", "render", "render_json", "render_sarif", "render_text"]

src/coretrace_python/reporters/json_format.py

@@ -0,0 +1,36 @@
+"""JSON reporter: versioned normalized finding records."""
+
+from __future__ import annotations
+
+import json
+
+from coretrace_python.findings import FINDING_SCHEMA_VERSION, Finding
+from coretrace_python.reporters.report import Report
+
+
+def finding_record(finding: Finding) -> dict[str, object]:
+    span = finding.span
+    return {
+        "rule_id": finding.rule_id,
+        "message": finding.message,
+        "severity": finding.severity.value,
+        "confidence": finding.confidence.value,
+        "location": {
+            "path": str(span.source_id),
+            "line": span.start_line,
+            "column": span.start_column,
+            "end_line": span.end_line,
+            "end_column": span.end_column,
+        },
+        "function": finding.function,
+        "metadata": dict(finding.metadata),
+    }
+
+
+def render_json(report: Report) -> str:
+    document = {
+        "schema_version": FINDING_SCHEMA_VERSION,
+        "tool": {"name": report.tool_name, "version": report.tool_version},
+        "findings": [finding_record(finding) for finding in report.findings],
+    }
+    return json.dumps(document, indent=2) + "\n"

src/coretrace_python/reporters/report.py

@@ -0,0 +1,22 @@
+"""The normalized report every reporter renders (architecture §23, §28)."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from coretrace_python.findings import Finding
+
+
+def _order(finding: Finding) -> tuple[str, int, int, str]:
+    span = finding.span
+    return (str(span.source_id), span.start_line, span.start_column, finding.rule_id)
+
+
+@dataclass(frozen=True)
+class Report:
+    findings: tuple[Finding, ...]
+    tool_name: str
+    tool_version: str
+
+    def __post_init__(self) -> None:
+        object.__setattr__(self, "findings", tuple(sorted(self.findings, key=_order)))

src/coretrace_python/reporters/sarif.py

@@ -0,0 +1,70 @@
+"""SARIF 2.1.0 reporter."""
+
+from __future__ import annotations
+
+import json
+
+from coretrace_python.findings import Finding, Severity
+from coretrace_python.reporters.report import Report
+
+SARIF_VERSION = "2.1.0"
+SARIF_SCHEMA = "https://json.schemastore.org/sarif-2.1.0.json"
+
+_LEVELS = {
+    Severity.CRITICAL: "error",
+    Severity.HIGH: "error",
+    Severity.MEDIUM: "warning",
+    Severity.LOW: "note",
+    Severity.INFO: "note",
+}
+
+
+def _result(finding: Finding, rule_index: int) -> dict[str, object]:
+    span = finding.span
+    region: dict[str, int] = {"startLine": span.start_line, "startColumn": span.start_column}
+    if span.end_line is not None and span.end_column is not None:
+        region["endLine"] = span.end_line
+        region["endColumn"] = span.end_column
+    return {
+        "ruleId": finding.rule_id,
+        "ruleIndex": rule_index,
+        "level": _LEVELS[finding.severity],
+        "message": {"text": finding.message},
+        "locations": [
+            {
+                "physicalLocation": {
+                    "artifactLocation": {"uri": str(span.source_id)},
+                    "region": region,
+                }
+            }
+        ],
+    }
+
+
+def render_sarif(report: Report) -> str:
+    rule_ids: list[str] = []
+    for finding in report.findings:
+        if finding.rule_id not in rule_ids:
+            rule_ids.append(finding.rule_id)
+    document = {
+        "$schema": SARIF_SCHEMA,
+        "version": SARIF_VERSION,
+        "runs": [
+            {
+                "tool": {
+                    "driver": {
+                        "name": report.tool_name,
+                        "version": report.tool_version,
+                        "rules": [
+                            {"id": rule_id, "shortDescription": {"text": rule_id}}
+                            for rule_id in rule_ids
+                        ],
+                    }
+                },
+                "results": [
+                    _result(finding, rule_ids.index(finding.rule_id)) for finding in report.findings
+                ],
+            }
+        ],
+    }
+    return json.dumps(document, indent=2) + "\n"

src/coretrace_python/reporters/text.py

@@ -0,0 +1,22 @@
+"""Plain-text reporter: one ``path:line:column`` line per finding and a summary."""
+
+from __future__ import annotations
+
+from coretrace_python.reporters.report import Report
+
+
+def render_text(report: Report) -> str:
+    lines = []
+    for finding in report.findings:
+        span = finding.span
+        line = (
+            f"{span.source_id}:{span.start_line}:{span.start_column}: "
+            f"{finding.severity.value} {finding.rule_id}: {finding.message}"
+        )
+        if finding.function is not None:
+            line += f" [{finding.function}]"
+        lines.append(line)
+    count = len(report.findings)
+    summary = "no findings" if count == 0 else f"{count} finding{'s' if count > 1 else ''}"
+    lines.append(summary)
+    return "\n".join(lines) + "\n"

tests/test_architecture.py

@@ -41,7 +41,7 @@
 }
 
 # Top-level modules that wire the pipeline together and may import any layer.
-ROOT_MODULES = {"__init__", "__main__", "cli"}
+ROOT_MODULES = {"__init__", "__main__", "cli", "engine"}
 
 
 def package_files() -> Iterator[tuple[Path, str, str]]: