diff --git a/README.md b/README.md index 31c37b2..9b02a58 100644 --- a/README.md +++ b/README.md @@ -52,6 +52,15 @@ serve the project plugins and its findings are reported as they were, so editing re-analyses that file and the modules that import it only. Entries are plain data, never code; an unreadable entry is simply recomputed. +Modules are analysed by strongly connected components of the module graph, imports +first, so a module starts with the final summaries of everything it imports and mutually +importing modules are iterated together. `--jobs N` analyses the components of one wave +in `N` processes; each worker rebuilds the configuration from the project root and the +plugin roots and hands back summaries, call sites and findings as plain data, so the +result is the same whatever `N`. Once a module's results are extracted, its intermediate +representation and derived results are dropped and only its semantic tables stay in +memory. + `--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 diff --git a/src/coretrace_python/cache.py b/src/coretrace_python/cache.py index f62bcf5..c7350e8 100644 --- a/src/coretrace_python/cache.py +++ b/src/coretrace_python/cache.py @@ -28,6 +28,7 @@ FunctionSummary, KnownFunction, ModuleGraph, + SummaryIndex, Target, UnknownTarget, ) @@ -140,6 +141,16 @@ def decode(data: Mapping[str, Any]) -> CachedModule: ) +def encode_index(index: SummaryIndex) -> dict[str, Any]: + """A summary index as plain data, to hand a worker process what it imports.""" + + return {str(symbol): _encode_summary(index.summaries[symbol]) for symbol in index.symbols} + + +def decode_index(data: Mapping[str, Any]) -> SummaryIndex: + return SummaryIndex({SymbolId(_string(k)): _decode_summary(v) for k, v in data.items()}) + + def _string(value: Any) -> str: if not isinstance(value, str): raise TypeError(f"expected a string, got {value!r}") diff --git a/src/coretrace_python/cli.py b/src/coretrace_python/cli.py index 7f33d7e..cc1466e 100644 --- a/src/coretrace_python/cli.py +++ b/src/coretrace_python/cli.py @@ -79,6 +79,13 @@ def build_parser() -> argparse.ArgumentParser: help="with --check on a directory, keep per-module results under DIR and reuse " "them for modules unchanged since the previous run", ) + parser.add_argument( + "--jobs", + type=int, + default=None, + metavar="N", + help="with --check on a directory, analyse independent modules in N processes", + ) return parser @@ -93,6 +100,12 @@ def main(argv: list[str] | None = None) -> int: if args.cache is not None and not (args.check and args.path.is_dir()): print("error: --cache only applies to --check on a directory", file=sys.stderr) return EXIT_ERROR + if args.jobs is not None and not (args.check and args.path.is_dir()): + print("error: --jobs only applies to --check on a directory", file=sys.stderr) + return EXIT_ERROR + if args.jobs is not None and args.jobs < 1: + print("error: --jobs must be at least 1", 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) @@ -105,7 +118,9 @@ def main(argv: list[str] | None = None) -> int: if args.check: if args.path.is_dir(): cache = None if args.cache is None else ProjectCache(args.cache) - findings = engine.analyze_project(args.path, args.plugins, cache=cache).findings + findings = engine.analyze_project( + args.path, args.plugins, cache=cache, jobs=args.jobs or 1 + ).findings else: findings = engine.check(SourceManager().load_file(args.path), args.plugins) print(render(args.format or "text", engine.report(findings)), end="") diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index 26be07c..ed5378c 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -6,11 +6,13 @@ from __future__ import annotations +import concurrent.futures +import multiprocessing from collections.abc import Iterable, Mapping, Sequence from dataclasses import dataclass, field from pathlib import Path from types import MappingProxyType -from typing import ClassVar +from typing import Any, ClassVar from coretrace_python import __version__ from coretrace_python.abstract import ConstantPropagation @@ -23,7 +25,11 @@ from coretrace_python.cache import ( CachedModule, ProjectCache, + decode, + decode_index, directory_fingerprint, + encode, + encode_index, fingerprint, module_keys, ) @@ -128,6 +134,20 @@ def run(cls, ctx: AnalysisContext) -> None: pass +class ResultsEvicted(TransformationPass): + """Drops a module's PyIR and every derived result once its summaries, call sites and + findings are extracted (§30); the semantic tables and the engine inputs stay.""" + + name: ClassVar[str] = "project.results-evicted" + preserves: ClassVar[frozenset[AnyAnalysis]] = frozenset( + {*SEMANTIC_ANALYSES, SecurityModelAnalysis, ProjectSummaries, DependencyAnalysis} + ) + + @classmethod + def run(cls, ctx: AnalysisContext) -> None: + pass + + def _no_keys() -> Mapping[str, str]: return MappingProxyType({}) @@ -194,12 +214,17 @@ def analyze_project( plugin_roots: Sequence[Path] = (), plugins: Sequence[Plugin] = (), cache: ProjectCache | None = None, + jobs: int = 1, ) -> ProjectAnalysis: """Analyse every Python file under ``root`` with a shared summary index (§21) and the dependency graph of its manifests (§26). ``plugins`` adds plugin instances to the ones discovered under ``plugin_roots``. With a ``cache``, modules whose key is - unchanged since a previous run are served from it (§11).""" + unchanged since a previous run are served from it (§11). Modules are scheduled by + strongly connected components of the module graph, imports first; with ``jobs`` + above one the components of a wave are analysed in that many processes (§29).""" + if jobs < 1: + raise ValueError("jobs must be at least 1") sources = SourceManager() findings: list[Finding] = [] dependencies = resolve_dependencies(root, sources) @@ -244,29 +269,112 @@ def analyze_project( graph, {name: fingerprint(configuration, str(files[name].source_id), name, files[name].text) for name in analysable}, ) - cached: dict[str, CachedModule] = {} + results: dict[str, CachedModule] = {} if cache is not None: for name in analysable: entry = cache.load(keys[name]) if entry is not None: - cached[name] = entry - fresh = {name: manager for name, manager in analysable.items() if name not in cached} - - index = SummaryIndex() - cached_summaries = { - project_symbol(name, function): summary - for name, entry in cached.items() - for function, summary in entry.summaries.items() - } - for manager in fresh.values(): + results[name] = entry + reused = tuple(sorted(results)) + + module_plugins = tuple(p for p in all_plugins if not isinstance(p, ProjectPlugin)) + pool = None + if jobs > 1: + pool = concurrent.futures.ProcessPoolExecutor( + max_workers=jobs, mp_context=multiprocessing.get_context("spawn") + ) + try: + for wave in graph.schedule(): + pending = [c for c in wave if any(m not in results for m in c)] + computed: dict[str, CachedModule] = {} + if pool is None: + for component in pending: + batch = {name: analysable[name] for name in sorted(component)} + computed.update(_analyse_managers(batch, _seed(results, graph, component), module_plugins, affected)) + for manager in batch.values(): + manager.run(ResultsEvicted) + else: + futures = [ + pool.submit( + _analyse_batch, + _Batch( + root, + tuple(plugin_roots), + tuple(plugins), + {name: _path_of(files[name]) for name in sorted(component)}, + encode_index(_seed(results, graph, component)), + ), + ) + for component in pending + ] + for future in futures: + computed.update({name: decode(data) for name, data in future.result().items()}) + for name, entry in computed.items(): + if cache is not None: + cache.store(keys[name], entry) + results.update(computed) + finally: + if pool is not None: + pool.shutdown() + + index = _seed(results, graph, frozenset()) + call_graphs: dict[str, CallGraph] = {} + for name in sorted(analysable): + entry = results[name] + findings.extend(entry.findings) + sites: dict[str, list[CallSite]] = {function: [] for function in entry.functions} + for site in entry.sites: + sites.setdefault(site.caller, []).append(site) + call_graphs[name] = CallGraph({}, {f: tuple(s) for f, s in sites.items()}, frozenset()) + context = ProjectContext(graph, dependencies, advisories, analysable, call_graphs) + for plugin in all_plugins: + if isinstance(plugin, ProjectPlugin): + findings.extend(plugin.analyze_project(context)) + return ProjectAnalysis(graph, index, tuple(findings), dependencies, MappingProxyType(keys), reused) + + +def _seed(results: Mapping[str, CachedModule], graph: ModuleGraph, component: frozenset[str]) -> SummaryIndex: + """The summaries a component starts from: those of every module it imports, + transitively, all final by the time its wave runs; the whole index for no component.""" + + if component: + wanted: set[str] = set() + pending = list(component) + while pending: + for imported in graph.imports(pending.pop()): + if imported in results and imported not in wanted: + wanted.add(imported) + pending.append(imported) + else: + wanted = set(results) + return SummaryIndex( + { + project_symbol(name, function): summary + for name in sorted(wanted) + for function, summary in results[name].summaries.items() + } + ) + + +def _analyse_managers( + managers: Mapping[str, AnalysisManager], + seed: SummaryIndex, + plugins: tuple[Plugin, ...], + affected: Mapping[SymbolId, Advisory], +) -> dict[str, CachedModule]: + """Analyse one component: iterate its summaries to a fixpoint over ``seed`` (§21), + then extract what the rest of the run needs from each module.""" + + index = seed + for manager in managers.values(): manager.provide(ProjectSummaries, index) for _ in range(MAX_PROJECT_ITERATIONS): updated = SummaryIndex( { - **cached_summaries, + **seed.summaries, **{ project_symbol(name, function): summary - for name, manager in fresh.items() + for name, manager in managers.items() for function, summary in _summaries_of(manager).items() }, } @@ -274,31 +382,44 @@ def analyze_project( if updated == index: break index = updated - for manager in fresh.values(): + for manager in managers.values(): manager.run(ProjectSummariesUpdated) manager.provide(ProjectSummaries, index) + return {name: _analyse_module(manager, plugins, affected) for name, manager in managers.items()} + +@dataclass(frozen=True) +class _Batch: + """One component handed to a worker process: it rebuilds the configuration from the + project root and the plugin roots, so only paths and the imported summaries travel.""" + + root: Path + plugin_roots: tuple[Path, ...] + plugins: tuple[Plugin, ...] + paths: Mapping[str, Path] + seed: Mapping[str, Any] + + +def _analyse_batch(batch: _Batch) -> dict[str, dict[str, Any]]: + sources = SourceManager() + managers = {name: _register_all(build_hir(sources.load_file(path))) for name, path in batch.paths.items()} + registry = load_plugins(batch.plugin_roots, next(iter(managers.values()))) + all_plugins: tuple[Plugin, ...] = (*(loaded.plugin for loaded in registry), *batch.plugins) + dependencies = resolve_dependencies(batch.root, sources) + advisories = tuple(a for plugin in all_plugins for a in plugin.advisories) + affected = affected_symbols(dependencies, advisories) + models = plugin_models(all_plugins).extended(*advisory_sinks(affected)) + for manager in managers.values(): + manager.provide(SecurityModelAnalysis, models) + manager.provide(DependencyAnalysis, dependencies) module_plugins = tuple(p for p in all_plugins if not isinstance(p, ProjectPlugin)) - call_graphs: dict[str, CallGraph] = {} - for name in sorted(analysable): - entry = cached.get(name) - if entry is None: - entry = _analyse_module(analysable[name], module_plugins, affected) - if cache is not None: - cache.store(keys[name], entry) - else: - sites: dict[str, list[CallSite]] = {function: [] for function in entry.functions} - for site in entry.sites: - sites.setdefault(site.caller, []).append(site) - call_graphs[name] = CallGraph({}, {f: tuple(s) for f, s in sites.items()}, frozenset()) - findings.extend(entry.findings) - context = ProjectContext(graph, dependencies, advisories, analysable, call_graphs) - for plugin in all_plugins: - if isinstance(plugin, ProjectPlugin): - findings.extend(plugin.analyze_project(context)) - return ProjectAnalysis( - graph, index, tuple(findings), dependencies, MappingProxyType(keys), tuple(sorted(cached)) - ) + results = _analyse_managers(managers, decode_index(batch.seed), module_plugins, affected) + return {name: encode(entry) for name, entry in results.items()} + + +def _path_of(source: SourceFile) -> Path: + assert source.path is not None, "project sources are loaded from files" + return source.path def _configuration_key( diff --git a/src/coretrace_python/interprocedural/modulegraph.py b/src/coretrace_python/interprocedural/modulegraph.py index 0a49801..553f57d 100644 --- a/src/coretrace_python/interprocedural/modulegraph.py +++ b/src/coretrace_python/interprocedural/modulegraph.py @@ -58,6 +58,63 @@ def imports(self, name: str) -> frozenset[str]: def importers(self, name: str) -> frozenset[str]: return frozenset(m for m, imported in self._imports.items() if name in imported) + def schedule(self) -> tuple[tuple[frozenset[str], ...], ...]: + """Strongly connected components in waves: every component of a wave imports, + outside itself, only components of earlier waves, so one wave can be analysed + in parallel and a component's imports are final when it starts (§29).""" + + names = frozenset(self._sources) + edges = {name: sorted(m for m in self.imports(name) if m in names) for name in names} + component_of: dict[str, int] = {} + components: list[frozenset[str]] = [] + index: dict[str, int] = {} + low: dict[str, int] = {} + stack: list[str] = [] + + def visit(name: str) -> None: + index[name] = low[name] = len(index) + stack.append(name) + for imported in edges[name]: + if imported not in index: + visit(imported) + low[name] = min(low[name], low[imported]) + elif imported in stack: + low[name] = min(low[name], index[imported]) + if low[name] == index[name]: + members: list[str] = [] + while True: + member = stack.pop() + members.append(member) + if member == name: + break + for member in members: + component_of[member] = len(components) + components.append(frozenset(members)) + + for name in sorted(names): + if name not in index: + visit(name) + + depth: dict[int, int] = {} + + def level(component: int) -> int: + if component not in depth: + below = { + component_of[imported] + for member in components[component] + for imported in edges[member] + if component_of[imported] != component + } + depth[component] = 1 + max((level(c) for c in below), default=-1) + return depth[component] + + waves: dict[int, list[frozenset[str]]] = {} + for number in range(len(components)): + waves.setdefault(level(number), []).append(components[number]) + return tuple( + tuple(sorted(waves[wave], key=min)) for wave in sorted(waves) + ) + def build_module_graph( sources: Mapping[str, SourceFile], diff --git a/tests/test_parallel.py b/tests/test_parallel.py new file mode 100644 index 0000000..56d75ce --- /dev/null +++ b/tests/test_parallel.py @@ -0,0 +1,253 @@ +"""Acceptance tests for parallel analysis and memory eviction (``docs/architecture.md`` +§29, §30, §38 Phase 10). + +The module graph is scheduled by strongly connected components in topological order: +components of one wave depend only on earlier waves, so they can be analysed at the same +time, in separate processes when ``jobs`` is above one. Each component reads the +summaries of the modules it imports, which are final by then, and hands back its +summaries, call sites and findings; the engine then drops its intermediate results, +keeping the semantic tables only. The result is the same whatever the number of jobs. + +Expected to remain red until ``ModuleGraph.schedule``, ``analyze_project(jobs=)``, +``engine.ResultsEvicted`` and the ``--jobs`` option exist. +""" + +from __future__ import annotations + +import inspect +from pathlib import Path +from types import MappingProxyType + +import pytest + +from coretrace_python import engine +from coretrace_python.cache import ProjectCache +from coretrace_python.cfg import CFGAnalysis +from coretrace_python.cli import main +from coretrace_python.findings import Finding +from coretrace_python.frontend import build_hir +from coretrace_python.interprocedural import ModuleGraph, ProjectSummaries +from coretrace_python.ir.ssa import SSAAnalysis +from coretrace_python.semantic.imports import ImportAnalysis +from coretrace_python.semantic.scopes import ScopeAnalysis +from coretrace_python.source import SourceManager +from coretrace_python.taint import SecurityModelAnalysis + +try: + from coretrace_python.engine import ResultsEvicted +except ImportError as error: # pragma: no cover - red until parallel analysis lands + MISSING: Exception | None = error +else: + MISSING = None + if "jobs" not in inspect.signature(engine.analyze_project).parameters: + MISSING = AttributeError("analyze_project has no jobs parameter") + elif not hasattr(ModuleGraph, "schedule"): + MISSING = AttributeError("ModuleGraph has no schedule") + + +@pytest.fixture(autouse=True) +def require_parallel() -> None: + if MISSING is not None: + pytest.fail(f"parallel analysis is not implemented yet: {MISSING}") + + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "plugins" + +HELPERS = "import os\n\ndef execute(command):\n os.system(command)\n" +MAIN = "from app.helpers import execute\n\ndef run():\n execute(input())\n" +CLEAN = "def ok():\n return 1\n" + + +def project(root: Path, files: dict[str, str]) -> Path: + for relative, text in files.items(): + path = root / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + return root + + +def standard(root: Path) -> Path: + return project( + root / "src", + {"app/__init__.py": "", "app/helpers.py": HELPERS, "app/main.py": MAIN, "app/clean.py": CLEAN}, + ) + + +def rules(findings: tuple[Finding, ...]) -> list[tuple[str, str, int]]: + return sorted((Path(str(f.span.source_id)).name, f.rule_id, f.span.start_line) for f in findings) + + +def graph_of(imports: dict[str, set[str]]) -> ModuleGraph: + sources = SourceManager() + return ModuleGraph( + MappingProxyType({name: sources.add_source(f"{name}.py", "", name) for name in imports}), + MappingProxyType({name: frozenset(found) for name, found in imports.items()}), + ) + + +# --------------------------------------------------------------------------- schedule + + +def test_schedule_groups_cycles_and_orders_waves_topologically() -> None: + graph = graph_of({"a": {"b", "c"}, "b": {"d"}, "c": {"d"}, "d": set(), "e": {"f"}, "f": {"e"}, "g": set()}) + + waves = graph.schedule() + + assert waves == ( + (frozenset({"d"}), frozenset({"e", "f"}), frozenset({"g"})), + (frozenset({"b"}), frozenset({"c"})), + (frozenset({"a"}),), + ) + + +def test_schedule_of_a_project_puts_importers_after_their_imports(tmp_path: Path) -> None: + graph = engine.analyze_project(standard(tmp_path), [PLUGINS]).graph + + assert graph.schedule() == ( + (frozenset({"app"}), frozenset({"app.clean"}), frozenset({"app.helpers"})), + (frozenset({"app.main"}),), + ) + + +def test_schedule_ignores_imports_outside_the_graph() -> None: + assert graph_of({"a": {"missing"}}).schedule() == ((frozenset({"a"}),),) + + +# --------------------------------------------------------------------------- parallel + + +def test_parallel_analysis_matches_the_sequential_result(tmp_path: Path) -> None: + root = standard(tmp_path) + + sequential = engine.analyze_project(root, [PLUGINS]) + parallel = engine.analyze_project(root, [PLUGINS], jobs=3) + + assert parallel.findings == sequential.findings + assert rules(parallel.findings) == [("main.py", "command-injection", 4)] + assert parallel.index == sequential.index + assert parallel.keys == sequential.keys + assert parallel.reused == () + + +def test_mutually_importing_modules_are_analysed_together(tmp_path: Path) -> None: + root = project( + tmp_path / "src", + { + "a.py": "from b import g\n\ndef f():\n g(input())\n", + "b.py": "import os\nfrom a import f\n\ndef g(x):\n os.system(x)\n\ndef h():\n return f()\n", + }, + ) + + result = engine.analyze_project(root, [PLUGINS], jobs=2) + + assert result.graph.schedule() == ((frozenset({"a", "b"}),),) + assert rules(result.findings) == [("a.py", "command-injection", 4)] + + +def test_parallel_analysis_serves_project_plugins_and_correlation(tmp_path: Path) -> None: + root = project( + tmp_path / "src", + { + "requirements.txt": "pyyaml==5.3.1\n", + "app/__init__.py": "", + "app/config.py": "import yaml\n\ndef parse(text):\n return yaml.load(text)\n", + "app/main.py": "from app.config import parse\n\ndef run():\n parse(input())\n", + }, + ) + + findings = engine.analyze_project(root, [PLUGINS], jobs=2).findings + + assert rules(findings) == [ + ("config.py", "reachable-vulnerability", 4), + ("main.py", "exploitable-vulnerability", 4), + ("requirements.txt", "vulnerable-dependency", 1), + ] + + +def test_parallel_analysis_keeps_syntax_and_unsupported_notes(tmp_path: Path) -> None: + root = project( + tmp_path / "src", + { + "broken.py": "def (:\n", + "nested.py": "def outer():\n def inner():\n pass\n return inner\n", + "fine.py": CLEAN, + }, + ) + + sequential = engine.analyze_project(root, [PLUGINS]) + parallel = engine.analyze_project(root, [PLUGINS], jobs=2) + + assert parallel.findings == sequential.findings + assert [f.rule_id for f in parallel.findings] == ["syntax-error", "unsupported-syntax"] + + +def test_parallel_analysis_fills_and_reuses_the_cache(tmp_path: Path) -> None: + root = standard(tmp_path) + cache = ProjectCache(tmp_path / "cache") + + first = engine.analyze_project(root, [PLUGINS], cache=cache, jobs=2) + second = engine.analyze_project(root, [PLUGINS], cache=cache, jobs=2) + third = engine.analyze_project(root, [PLUGINS], cache=cache) + + assert first.reused == () + assert set(second.reused) == set(third.reused) == {"app", "app.helpers", "app.main", "app.clean"} + assert second.findings == first.findings == third.findings + + +def test_one_job_never_spawns_workers(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + import concurrent.futures + + def forbidden(*args: object, **kwargs: object) -> None: + raise AssertionError("a process pool was created for jobs=1") + + monkeypatch.setattr(concurrent.futures, "ProcessPoolExecutor", forbidden) + + assert rules(engine.analyze_project(standard(tmp_path), [PLUGINS], jobs=1).findings) == [ + ("main.py", "command-injection", 4) + ] + + +# --------------------------------------------------------------------------- eviction + + +def test_eviction_keeps_semantic_tables_and_engine_inputs_only() -> None: + source = SourceManager().add_source("m.py", "import os\n\ndef f(x):\n os.system(x)\n") + manager = engine.build_manager(build_hir(source)) + function = manager.module.body[-1] + manager.get(SSAAnalysis, function) # type: ignore[arg-type] + manager.get(ImportAnalysis) + + manager.run(ResultsEvicted) + + assert not manager.is_cached(SSAAnalysis, function) # type: ignore[arg-type] + assert not manager.is_cached(CFGAnalysis, function) # type: ignore[arg-type] + assert manager.is_cached(ImportAnalysis) + assert manager.is_cached(ScopeAnalysis) + assert manager.is_cached(ProjectSummaries) + assert manager.is_cached(SecurityModelAnalysis) + + +# --------------------------------------------------------------------------- CLI + + +def test_check_accepts_a_job_count(tmp_path: Path, capsys) -> None: # type: ignore[no-untyped-def] + root = standard(tmp_path) + + sequential = main(["--check", str(root), "--plugins", str(PLUGINS)]) + sequential_output = capsys.readouterr().out + parallel = main(["--check", str(root), "--plugins", str(PLUGINS), "--jobs", "2"]) + parallel_output = capsys.readouterr().out + + assert sequential == parallel == 1 + assert parallel_output == sequential_output + + +def test_jobs_option_requires_a_directory_check(tmp_path: Path, capsys) -> None: # type: ignore[no-untyped-def] + source = tmp_path / "x.py" + source.write_text("", encoding="utf-8") + + assert main(["--emit-ir", "--jobs", "2", str(source)]) == 2 + assert "--jobs" in capsys.readouterr().err + assert main(["--check", "--jobs", "0", str(source)]) == 2 + assert "--jobs" in capsys.readouterr().err