feat: multi-file analysis with a module graph and project summaries

README.md

@@ -28,9 +28,14 @@ python -m ruff check .
 
 ```bash
 coretrace-python-analyzer --check app.py --plugins plugins/
-coretrace-python-analyzer --check app.py --plugins plugins/ --format sarif > report.sarif
+coretrace-python-analyzer --check src/ --plugins plugins/ --format sarif > report.sarif
 ```
 
+Given a directory, every Python file below it is analysed as one project: modules are
+named after their packages, and taint follows calls into functions defined in other files
+through a project-wide index of function summaries iterated to a fixpoint. Hidden and
+tooling directories (`.venv`, `node_modules`, `__pycache__`, `build`, `dist`) are skipped.
+
 `--plugins` is a directory searched recursively for `plugin.toml` manifests and may be
 repeated. The repository ships a first set under `plugins/`: security models for the
 standard library, Flask, FastAPI and SQLAlchemy (`models/`), taint detectors for SQL

src/coretrace_python/cli.py

@@ -40,7 +40,7 @@ def build_parser() -> argparse.ArgumentParser:
         prog=engine.TOOL_NAME,
         description="Analyze Python source with CoreTrace.",
     )
-    parser.add_argument("path", type=Path, help="Python source file to analyze")
+    parser.add_argument("path", type=Path, help="Python source file, or a directory for --check")
     parser.add_argument(
         "--emit-ir",
         action="store_true",
@@ -82,12 +82,19 @@ def main(argv: list[str] | None = None) -> int:
         print("error: --format only applies to --check", file=sys.stderr)
         return EXIT_ERROR
 
+    if args.path.is_dir() and args.emit_ir:
+        print(f"error: {args.path} is a directory; --emit-ir needs a file", file=sys.stderr)
+        return EXIT_ERROR
+
     try:
-        source = SourceManager().load_file(args.path)
         if args.emit_ir:
+            source = SourceManager().load_file(args.path)
             print(format_module(lower_module(build_hir(source), ssa=args.ssa)))
         if args.check:
-            findings = engine.check(source, args.plugins)
+            if args.path.is_dir():
+                findings = engine.analyze_project(args.path, args.plugins).findings
+            else:
+                findings = engine.check(SourceManager().load_file(args.path), 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:

src/coretrace_python/engine.py

@@ -7,17 +7,34 @@
 from __future__ import annotations
 
 from collections.abc import Sequence
+from dataclasses import dataclass
 from pathlib import Path
+from typing import ClassVar
 
 from coretrace_python import __version__
 from coretrace_python.abstract import ConstantPropagation
-from coretrace_python.analysis import AnalysisManager, AnyAnalysis
+from coretrace_python.analysis import (
+    AnalysisContext,
+    AnalysisManager,
+    AnyAnalysis,
+    TransformationPass,
+)
 from coretrace_python.cfg import CFGAnalysis, CFGError, DominanceAnalysis, PostDominanceAnalysis
 from coretrace_python.findings import Confidence, Finding, Severity
 from coretrace_python.findings.refutation import RefutationAnalysis
-from coretrace_python.frontend import build_hir
+from coretrace_python.frontend import HIRBuildError, ParseError, build_hir
 from coretrace_python.hir import nodes
-from coretrace_python.interprocedural import CallGraphAnalysis, SummaryAnalysis
+from coretrace_python.interprocedural import (
+    CallGraphAnalysis,
+    FunctionSummary,
+    ModuleGraph,
+    ProjectSummaries,
+    SummaryAnalysis,
+    SummaryIndex,
+    build_module_graph,
+    discover_sources,
+    project_symbol,
+)
 from coretrace_python.ir.defuse import DefUseAnalysis
 from coretrace_python.ir.lowering import (
     LoweringError,
@@ -26,11 +43,18 @@
     analyzable_functions,
 )
 from coretrace_python.ir.ssa import SSAAnalysis
-from coretrace_python.plugins import PluginRegistry, discover_plugins, run_plugins
+from coretrace_python.plugins import LoadedPlugin, 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
+from coretrace_python.semantic.imports import ImportAnalysis, ImportResolutionError, ImportTable
+from coretrace_python.semantic.scopes import ScopeError
+from coretrace_python.source import SourceFile, SourceManager, SourceSpan
+from coretrace_python.taint import (
+    ModelTable,
+    SecurityModelAnalysis,
+    SecurityModelRegistry,
+    TaintAnalysis,
+)
 
 TOOL_NAME = "coretrace-python-analyzer"
 
@@ -46,39 +70,143 @@
     ConstantPropagation,
     CallGraphAnalysis,
     SummaryAnalysis,
+    ProjectSummaries,
     SecurityModelAnalysis,
     TaintAnalysis,
     RefutationAnalysis,
 )
 
 
+# Named so the set stays right whichever analyses a build registers.
+_PROJECT_DEPENDANT_NAMES = frozenset(
+    {"interprocedural.project", "interprocedural.summaries", "taint.flows", "findings.refutation"}
+)
+_PROJECT_DEPENDANTS: frozenset[AnyAnalysis] = frozenset(
+    a for a in ALL_ANALYSES if a.name in _PROJECT_DEPENDANT_NAMES
+)
+MAX_PROJECT_ITERATIONS = 20
+
+
+class ProjectSummariesUpdated(TransformationPass):
+    """Invalidates every result that depends on the project-wide summary index."""
+
+    name: ClassVar[str] = "project.summaries-updated"
+    preserves: ClassVar[frozenset[AnyAnalysis]] = frozenset(
+        a for a in ALL_ANALYSES if a not in _PROJECT_DEPENDANTS
+    )
+
+    @classmethod
+    def run(cls, ctx: AnalysisContext) -> None:
+        pass
+
+
+@dataclass(frozen=True)
+class ProjectAnalysis:
+    graph: ModuleGraph
+    index: SummaryIndex
+    findings: tuple[Finding, ...]
+
+
 def build_manager(
     module: nodes.Module, models: SecurityModelRegistry | None = None
 ) -> AnalysisManager:
-    """A manager with every engine analysis registered and the security models provided."""
+    """A manager with every engine analysis registered and the engine inputs provided."""
 
     manager = _register_all(module)
     manager.provide(SecurityModelAnalysis, (models or SecurityModelRegistry()).freeze())
+    manager.provide(ProjectSummaries, SummaryIndex())
     return manager
 
 
-def check(source: SourceFile, plugin_roots: Sequence[Path]) -> tuple[Finding, ...]:
-    """Run every plugin found under ``plugin_roots`` against ``source``.
-
-    Plugins are loaded first so the models they contribute are all registered before the
-    table is provided; framework models and detectors compose without knowing each other.
-    """
-
-    manager = _register_all(build_hir(source))
+def load_plugins(plugin_roots: Sequence[Path], manager: AnalysisManager) -> PluginRegistry:
     registry = PluginRegistry()
     for root in plugin_roots:
         for loaded in discover_plugins(root, manager):
             registry.add(loaded)
+    return registry
+
+
+def plugin_models(registry: PluginRegistry) -> ModelTable:
     models = SecurityModelRegistry()
     for loaded in registry:
         models.register(*loaded.plugin.models)
-    manager.provide(SecurityModelAnalysis, models.freeze())
+    return models.freeze()
+
+
+def check(source: SourceFile, plugin_roots: Sequence[Path]) -> tuple[Finding, ...]:
+    """Run every plugin found under ``plugin_roots`` against one file."""
 
+    manager = _register_all(build_hir(source))
+    registry = load_plugins(plugin_roots, manager)
+    manager.provide(SecurityModelAnalysis, plugin_models(registry))
+    manager.provide(ProjectSummaries, SummaryIndex())
+    return _check_module(manager, tuple(registry))
+
+
+def analyze_project(root: Path, plugin_roots: Sequence[Path] = ()) -> ProjectAnalysis:
+    """Analyse every Python file under ``root`` with a shared summary index (§21)."""
+
+    sources = SourceManager()
+    findings: list[Finding] = []
+    modules: dict[str, nodes.Module] = {}
+    files: dict[str, SourceFile] = {}
+    for source in discover_sources(root, sources):
+        try:
+            modules[source.module_name] = build_hir(source)
+        except (ParseError, HIRBuildError) as error:
+            findings.append(_note("syntax-error", str(error), source, 1))
+            continue
+        files[source.module_name] = source
+
+    managers = {name: _register_all(module) for name, module in modules.items()}
+    probe = next(iter(managers.values()), None) or _register_all(build_hir(sources.add_source("<empty>", "")))
+    registry = load_plugins(plugin_roots, probe)
+    models = plugin_models(registry)
+    index = SummaryIndex()
+    for manager in managers.values():
+        manager.provide(SecurityModelAnalysis, models)
+        manager.provide(ProjectSummaries, index)
+
+    imports: dict[str, ImportTable] = {}
+    analysable: dict[str, AnalysisManager] = {}
+    for name, manager in managers.items():
+        try:
+            imports[name] = manager.get(ImportAnalysis)
+        except (ImportResolutionError, ScopeError) as error:
+            findings.append(_note("syntax-error", str(error), files[name], 1))
+            continue
+        analysable[name] = manager
+    graph = build_module_graph(
+        {name: files[name] for name in analysable}, {name: modules[name] for name in analysable}, imports
+    )
+
+    for _ in range(MAX_PROJECT_ITERATIONS):
+        updated = SummaryIndex(
+            {
+                project_symbol(name, function): summary
+                for name, manager in analysable.items()
+                for function, summary in _summaries_of(manager).items()
+            }
+        )
+        if updated == index:
+            break
+        index = updated
+        for manager in analysable.values():
+            manager.run(ProjectSummariesUpdated)
+            manager.provide(ProjectSummaries, index)
+
+    plugins = tuple(registry)
+    for name in sorted(analysable):
+        findings.extend(_check_module(analysable[name], plugins))
+    return ProjectAnalysis(graph, index, tuple(findings))
+
+
+def _summaries_of(manager: AnalysisManager) -> dict[str, FunctionSummary]:
+    table = manager.get(SummaryAnalysis)
+    return {name: table.summary(name) for name in table.names}
+
+
+def _check_module(manager: AnalysisManager, plugins: tuple[LoadedPlugin, ...]) -> tuple[Finding, ...]:
     supported: list[nodes.Function] = []
     notes: list[Finding] = []
     for function in analyzable_functions(manager.module):
@@ -97,10 +225,20 @@ def check(source: SourceFile, plugin_roots: Sequence[Path]) -> tuple[Finding, ..
             )
         else:
             supported.append(function)
-    findings = run_plugins(manager, [loaded.plugin for loaded in registry], tuple(supported))
+    findings = run_plugins(manager, [loaded.plugin for loaded in plugins], tuple(supported))
     return (*findings, *notes)
 
 
+def _note(rule_id: str, message: str, source: SourceFile, line: int) -> Finding:
+    return Finding(
+        rule_id=rule_id,
+        message=message,
+        severity=Severity.INFO,
+        confidence=Confidence.HIGH,
+        span=SourceSpan(source.source_id, line, 1),
+    )
+
+
 def _register_all(module: nodes.Module) -> AnalysisManager:
     manager = AnalysisManager(module)
     manager.register(*ALL_ANALYSES)

src/coretrace_python/frontend/ast_adapter.py

@@ -313,7 +313,7 @@ def module(self, tree: ast.Module) -> nodes.Module:
             )
         else:
             span = SourceSpan(self._source.source_id, 1, 1, 1, 1)
-        return nodes.Module(self._source.module_name, body, span)
+        return nodes.Module(self._source.module_name, body, span, self._source.is_package)
 
 
 def build_module(source: SourceFile, tree: ast.Module) -> nodes.Module:

src/coretrace_python/hir/nodes.py

@@ -344,3 +344,4 @@ class Module:
     name: str
     body: tuple[Statement, ...]
     span: SourceSpan
+    is_package: bool = False

src/coretrace_python/interprocedural/__init__.py

@@ -9,10 +9,18 @@
     Target,
     UnknownTarget,
 )
+from coretrace_python.interprocedural.modulegraph import (
+    ModuleGraph,
+    build_module_graph,
+    discover_sources,
+    project_symbol,
+)
 from coretrace_python.interprocedural.summaries import (
     ExternalCall,
     FunctionSummary,
+    ProjectSummaries,
     SummaryAnalysis,
+    SummaryIndex,
     SummaryTable,
 )
 
@@ -24,8 +32,14 @@
     "ExternalSymbol",
     "FunctionSummary",
     "KnownFunction",
+    "ModuleGraph",
+    "ProjectSummaries",
     "SummaryAnalysis",
+    "SummaryIndex",
     "SummaryTable",
     "Target",
     "UnknownTarget",
+    "build_module_graph",
+    "discover_sources",
+    "project_symbol",
 ]