diff --git a/README.md b/README.md index 7db42fe..f5da2c5 100644 --- a/README.md +++ b/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 diff --git a/src/coretrace_python/cli.py b/src/coretrace_python/cli.py index d71cdd5..5ff2487 100644 --- a/src/coretrace_python/cli.py +++ b/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: diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index cd1fd1e..45d5b3f 100644 --- a/src/coretrace_python/engine.py +++ b/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("", ""))) + 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) diff --git a/src/coretrace_python/frontend/ast_adapter.py b/src/coretrace_python/frontend/ast_adapter.py index 93140e7..66c3577 100644 --- a/src/coretrace_python/frontend/ast_adapter.py +++ b/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: diff --git a/src/coretrace_python/hir/nodes.py b/src/coretrace_python/hir/nodes.py index 0814694..651d15f 100644 --- a/src/coretrace_python/hir/nodes.py +++ b/src/coretrace_python/hir/nodes.py @@ -344,3 +344,4 @@ class Module: name: str body: tuple[Statement, ...] span: SourceSpan + is_package: bool = False diff --git a/src/coretrace_python/interprocedural/__init__.py b/src/coretrace_python/interprocedural/__init__.py index c259c85..b5d0533 100644 --- a/src/coretrace_python/interprocedural/__init__.py +++ b/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", ] diff --git a/src/coretrace_python/interprocedural/modulegraph.py b/src/coretrace_python/interprocedural/modulegraph.py new file mode 100644 index 0000000..3eb7a6a --- /dev/null +++ b/src/coretrace_python/interprocedural/modulegraph.py @@ -0,0 +1,115 @@ +"""Module graph and project-wide summary index (architecture §21). + +A project is a directory of Python files. Each file is one module named after its +packages (``app/helpers.py`` is ``app.helpers``); the graph records which project modules +each module imports. Functions defined in the project get project symbols +(``python.app.helpers.run``) whose summaries live in a ``SummaryIndex`` that the engine +provides to every module's manager, so calls into other files are analysed through +summaries rather than by retaining every module's PyIR. +""" + +from __future__ import annotations + +from collections.abc import Iterable, Mapping +from dataclasses import dataclass +from pathlib import Path +from types import MappingProxyType + +from coretrace_python.hir import nodes +from coretrace_python.interprocedural.summaries import ProjectSummaries, SummaryIndex +from coretrace_python.semantic.imports import ImportTable +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.source import SourceFile, SourceManager + +IGNORED_DIRECTORIES = frozenset({"__pycache__", "node_modules", "venv", "build", "dist", "site-packages"}) + + +def discover_sources(root: Path, manager: SourceManager) -> tuple[SourceFile, ...]: + """Load every ``.py`` file under ``root``, skipping hidden and tooling directories.""" + + found: list[SourceFile] = [] + for path in sorted(root.rglob("*.py")): + relative = path.relative_to(root) + if any(p.startswith(".") or p in IGNORED_DIRECTORIES for p in relative.parts[:-1]): + continue + found.append(manager.load_file(path)) + return tuple(found) + + +def project_symbol(module_name: str, qualified_name: str) -> SymbolId: + return SymbolId(f"python.{module_name}.{qualified_name}") + + +@dataclass(frozen=True) +class ModuleGraph: + _sources: Mapping[str, SourceFile] + _imports: Mapping[str, frozenset[str]] + + @property + def modules(self) -> tuple[str, ...]: + return tuple(sorted(self._sources)) + + def source(self, name: str) -> SourceFile: + return self._sources[name] + + def imports(self, name: str) -> frozenset[str]: + return self._imports.get(name, frozenset()) + + def importers(self, name: str) -> frozenset[str]: + return frozenset(m for m, imported in self._imports.items() if name in imported) + + +def build_module_graph( + sources: Mapping[str, SourceFile], + modules: Mapping[str, nodes.Module], + imports: Mapping[str, ImportTable], +) -> ModuleGraph: + """Edges from each module to the project modules it imports, in any scope.""" + + names = frozenset(sources) + edges: dict[str, frozenset[str]] = {} + for name, module in modules.items(): + candidates: set[str] = set() + dotted_tops: set[str] = set() + for statement in _statements(module.body): + if isinstance(statement, nodes.Import): + for alias in statement.names: + candidates.add(alias.name) + if "." in alias.name: + dotted_tops.add(alias.name.partition(".")[0]) + for symbol in imports[name].all_symbols(): + path = symbol.canonical_name.removeprefix("python.") + # ``import app.config`` binds ``app``; the statement already named the module. + if path not in dotted_tops: + candidates.add(path) + # The most specific project module each import names; ``app.helpers`` implies + # the ``app`` package, which is not an edge worth recording. + found = { + next((m for m in _prefixes(candidate) if m in names), None) for candidate in candidates + } + edges[name] = frozenset(m for m in found if m is not None and m != name) + return ModuleGraph(MappingProxyType(dict(sources)), MappingProxyType(edges)) + + +def _prefixes(dotted: str) -> list[str]: + parts = dotted.split(".") + return [".".join(parts[:length]) for length in range(len(parts), 0, -1)] + + +def _statements(body: Iterable[nodes.Statement]) -> Iterable[nodes.Statement]: + for statement in body: + yield statement + for attribute in ("body", "orelse"): + nested = getattr(statement, attribute, None) + if isinstance(nested, tuple): + yield from _statements(nested) + +__all__ = [ + "IGNORED_DIRECTORIES", + "ModuleGraph", + "ProjectSummaries", + "SummaryIndex", + "build_module_graph", + "discover_sources", + "project_symbol", +] diff --git a/src/coretrace_python/interprocedural/summaries.py b/src/coretrace_python/interprocedural/summaries.py index 17139cb..05bc978 100644 --- a/src/coretrace_python/interprocedural/summaries.py +++ b/src/coretrace_python/interprocedural/summaries.py @@ -83,6 +83,36 @@ class FunctionSummary: return_externals: frozenset[SymbolId] = frozenset() +class SummaryIndex: + """Summaries of every project function, keyed by project symbol (§21).""" + + def __init__(self, summaries: Mapping[SymbolId, FunctionSummary] | None = None) -> None: + self.summaries: Mapping[SymbolId, FunctionSummary] = MappingProxyType( + dict(summaries or {}) + ) + self.symbols = tuple(sorted(self.summaries, key=str)) + + def summary(self, symbol: SymbolId) -> FunctionSummary | None: + return self.summaries.get(symbol) + + def __eq__(self, other: object) -> bool: + return isinstance(other, SummaryIndex) and dict(self.summaries) == dict(other.summaries) + + def __hash__(self) -> int: + return hash(self.symbols) + + +class ProjectSummaries(Analysis[SummaryIndex]): + """The project-wide summary index. The engine provides it for multi-file analysis; + on its own a module sees an empty index, which is the single-file behaviour.""" + + name: ClassVar[str] = "interprocedural.project" + + @classmethod + def compute(cls, ctx: AnalysisContext) -> SummaryIndex: + return SummaryIndex() + + class SummaryTable: def __init__(self, summaries: Mapping[str, FunctionSummary]) -> None: self._summaries = MappingProxyType(dict(summaries)) @@ -100,12 +130,18 @@ class _DependenceProblem(DataflowProblem[State]): direction: ClassVar[Direction] = Direction.FORWARD def __init__( - self, name: str, function: FunctionIR, graph: CallGraph, table: Mapping[str, FunctionSummary] + self, + name: str, + function: FunctionIR, + graph: CallGraph, + table: Mapping[str, FunctionSummary], + project: Mapping[SymbolId, FunctionSummary] | None = None, ) -> None: self.name = name self.function = function self.graph = graph self.table = table + self.project = project or {} self.blocks = {block.id: block for block in function.blocks} self.external: dict[_CallKey, ExternalCall] = {} self.returns: Dep = EMPTY @@ -180,6 +216,9 @@ def call(self, call: Call, state: Mapping[Value, Dep]) -> Dep: target = self.graph.target_at(self.name, call.location) if isinstance(target, ExternalSymbol): + project = self.project.get(target.symbol) + if project is not None: + return self.known(project, arguments, keywords, everything, call) self.record( target.symbol, tuple(a.parameters for a in arguments), @@ -189,28 +228,39 @@ def call(self, call: Call, state: Mapping[Value, Dep]) -> Dep: ) return everything | Dep(externals=frozenset({target.symbol})) if isinstance(target, KnownFunction): - callee = self.table[target.name] - if callee.unsupported: - return everything - - def mapped(indices: Dependencies) -> Dep: - result = EMPTY - for index in indices: - result |= arguments[index] if index < len(arguments) else keywords - return result - - for reached in callee.external_calls: - self.record( - reached.symbol, - tuple(mapped(d).parameters for d in reached.argument_dependencies), - mapped(reached.keyword_dependencies).parameters, - reached.location, - call.location, - ) - return mapped(callee.return_dependencies) | Dep(externals=callee.return_externals) + return self.known(self.table[target.name], arguments, keywords, everything, call) assert isinstance(target, UnknownTarget) return everything | state.get(call.callee, EMPTY) + def known( + self, + callee: FunctionSummary, + arguments: tuple[Dep, ...], + keywords: Dep, + everything: Dep, + call: Call, + ) -> Dep: + """Dependencies of a call to a function whose summary is known.""" + + if callee.unsupported: + return everything + + def mapped(indices: Dependencies) -> Dep: + result = EMPTY + for index in indices: + result |= arguments[index] if index < len(arguments) else keywords + return result + + for reached in callee.external_calls: + self.record( + reached.symbol, + tuple(mapped(d).parameters for d in reached.argument_dependencies), + mapped(reached.keyword_dependencies).parameters, + reached.location, + call.location, + ) + return mapped(callee.return_dependencies) | Dep(externals=callee.return_externals) + def record( self, symbol: SymbolId, @@ -230,9 +280,14 @@ def record( def summarize( - name: str, function: FunctionIR, cfg: CFG, graph: CallGraph, table: Mapping[str, FunctionSummary] + name: str, + function: FunctionIR, + cfg: CFG, + graph: CallGraph, + table: Mapping[str, FunctionSummary], + project: Mapping[SymbolId, FunctionSummary] | None = None, ) -> FunctionSummary: - problem = _DependenceProblem(name, function, graph, table) + problem = _DependenceProblem(name, function, graph, table, project) solution = solve(problem, cfg) problem.external = {} problem.returns = EMPTY @@ -251,12 +306,13 @@ def summarize( class SummaryAnalysis(Analysis[SummaryTable]): name: ClassVar[str] = "interprocedural.summaries" requires: ClassVar[frozenset[AnyAnalysis]] = frozenset( - {CallGraphAnalysis, SSAAnalysis, CFGAnalysis} + {CallGraphAnalysis, SSAAnalysis, CFGAnalysis, ProjectSummaries} ) @classmethod def compute(cls, ctx: AnalysisContext) -> SummaryTable: graph = ctx.get(CallGraphAnalysis) + project = ctx.get(ProjectSummaries).summaries table: dict[str, FunctionSummary] = {} supported: dict[str, tuple[FunctionIR, CFG]] = {} for name, function in graph.definitions.items(): @@ -270,7 +326,7 @@ def compute(cls, ctx: AnalysisContext) -> SummaryTable: while changed: changed = False for name, (ssa, cfg) in supported.items(): - updated = summarize(name, ssa, cfg, graph, table) + updated = summarize(name, ssa, cfg, graph, table, project) if updated != table[name]: table[name] = updated changed = True diff --git a/src/coretrace_python/semantic/imports.py b/src/coretrace_python/semantic/imports.py index c251fbe..5d725e7 100644 --- a/src/coretrace_python/semantic/imports.py +++ b/src/coretrace_python/semantic/imports.py @@ -42,10 +42,20 @@ def bindings(self, scope_id: ScopeId) -> Mapping[str, SymbolId]: def wildcards(self, scope_id: ScopeId) -> tuple[SymbolId, ...]: return self._wildcards.get(scope_id, ()) + def all_symbols(self) -> tuple[SymbolId, ...]: + """Every symbol any import in the module refers to, in every scope.""" + + found: list[SymbolId] = [] + for bindings in self._bindings.values(): + found.extend(bindings.values()) + for wildcards in self._wildcards.values(): + found.extend(wildcards) + return tuple(found) + class _Collector: def __init__(self, module: nodes.Module, scopes: ScopeTable) -> None: - self.package = module.name.rpartition(".")[0] + self.package = module.name if module.is_package else module.name.rpartition(".")[0] self.scopes = scopes self.bindings: dict[ScopeId, dict[str, SymbolId]] = {} self.wildcards: dict[ScopeId, list[SymbolId]] = {} diff --git a/src/coretrace_python/source/manager.py b/src/coretrace_python/source/manager.py index 0da4fb1..038b9e9 100644 --- a/src/coretrace_python/source/manager.py +++ b/src/coretrace_python/source/manager.py @@ -8,14 +8,16 @@ def module_name_for(path: Path) -> str: - """Dotted module name of ``path``, walking up through ``__init__.py`` packages.""" + """Dotted module name of ``path``, walking up through ``__init__.py`` packages. - parts = [path.stem] + A package's ``__init__.py`` is named after the package itself.""" + + parts = [] if path.name == "__init__.py" else [path.stem] directory = path.parent while (directory / "__init__.py").is_file(): parts.append(directory.name) directory = directory.parent - return ".".join(reversed(parts)) + return ".".join(reversed(parts)) or path.stem class SourceManager: @@ -54,6 +56,7 @@ def load_file(self, path: str | Path) -> SourceFile: text=text, module_name=module_name_for(resolved_path), path=resolved_path, + is_package=resolved_path.name == "__init__.py", ) self._sources[source_id] = source return source diff --git a/src/coretrace_python/source/model.py b/src/coretrace_python/source/model.py index 2b0dc72..8836fd0 100644 --- a/src/coretrace_python/source/model.py +++ b/src/coretrace_python/source/model.py @@ -53,4 +53,5 @@ class SourceFile: text: str module_name: str path: Path | None = None + is_package: bool = False diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index b0086cf..9b525c1 100644 --- a/src/coretrace_python/taint/engine.py +++ b/src/coretrace_python/taint/engine.py @@ -22,8 +22,11 @@ CallGraph, CallGraphAnalysis, ExternalSymbol, + FunctionSummary, KnownFunction, + ProjectSummaries, SummaryAnalysis, + SummaryIndex, SummaryTable, ) from coretrace_python.ir.model import ( @@ -110,6 +113,7 @@ def __init__( graph: CallGraph, summaries: SummaryTable, entry: EntryPoint | None = None, + project: SummaryIndex | None = None, ) -> None: self.name = name self.function = function @@ -117,6 +121,7 @@ def __init__( self.graph = graph self.summaries = summaries self.entry = entry + self.project = project or SummaryIndex() self.blocks = {block.id: block for block in function.blocks} self.symbols = graph.symbols(name) @@ -197,6 +202,10 @@ def call(self, call: Call, state: Mapping[Value, Taint], flows: list[TaintFlow]) target = self.graph.target_at(self.name, call.location) if isinstance(target, ExternalSymbol): + project = self.project.summary(target.symbol) + if project is not None: + through = target.symbol.canonical_name.removeprefix("python.") + return self.known(project, through, arguments, keywords, everything, call, flows) sink = self.models.sink(target.symbol) if sink is not None: for argument in call.argument_values(): @@ -212,36 +221,48 @@ def call(self, call: Call, state: Mapping[Value, Taint], flows: list[TaintFlow]) return everything if isinstance(target, KnownFunction): summary = self.summaries.summary(target.name) - if summary.unsupported: - return everything - - def mapped(deps: frozenset[int]) -> tuple[Taint, Value | None]: - taint, witness = Taint.none(), None - for index in sorted(deps): - part = arguments[index] if index < len(arguments) else keywords - if part and witness is None: - witness = ( - call.arguments[index] if index < len(arguments) else call.keywords[0][1] - ) - taint = taint.join(part) - return taint, witness - - for reached in summary.external_calls: - sink = self.models.sink(reached.symbol) - if sink is None: - continue - for deps in (*reached.argument_dependencies, reached.keyword_dependencies): - taint, witness = mapped(deps) - if witness is not None: - self.report(flows, sink, taint, witness, call, target.name, reached.location) - result = mapped(summary.return_dependencies)[0] - for symbol in sorted(summary.return_externals, key=str): - returned_source = self.models.source(symbol) - if returned_source is not None: - result = result.join(Taint(returned_source.kinds, frozenset({returned_source}))) - return result + return self.known(summary, target.name, arguments, keywords, everything, call, flows) return everything.join(state.get(call.callee, Taint.none())) + def known( + self, + summary: FunctionSummary, + through: str, + arguments: tuple[Taint, ...], + keywords: Taint, + everything: Taint, + call: Call, + flows: list[TaintFlow], + ) -> Taint: + """Flows and result taint of a call to a function whose summary is known.""" + + if summary.unsupported: + return everything + + def mapped(deps: frozenset[int]) -> tuple[Taint, Value | None]: + taint, witness = Taint.none(), None + for index in sorted(deps): + part = arguments[index] if index < len(arguments) else keywords + if part and witness is None: + witness = call.arguments[index] if index < len(arguments) else call.keywords[0][1] + taint = taint.join(part) + return taint, witness + + for reached in summary.external_calls: + sink = self.models.sink(reached.symbol) + if sink is None: + continue + for deps in (*reached.argument_dependencies, reached.keyword_dependencies): + taint, witness = mapped(deps) + if witness is not None: + self.report(flows, sink, taint, witness, call, through, reached.location) + result = mapped(summary.return_dependencies)[0] + for symbol in sorted(summary.return_externals, key=str): + returned_source = self.models.source(symbol) + if returned_source is not None: + result = result.join(Taint(returned_source.kinds, frozenset({returned_source}))) + return result + @staticmethod def report( flows: list[TaintFlow], @@ -293,8 +314,9 @@ def propagate_taint( graph: CallGraph, summaries: SummaryTable, entry: EntryPoint | None = None, + project: SummaryIndex | None = None, ) -> TaintFacts: - problem = _TaintProblem(name, function, models, graph, summaries, entry) + problem = _TaintProblem(name, function, models, graph, summaries, entry, project) solution = solve(problem, cfg) taints: dict[Value, Taint] = {} flows: list[TaintFlow] = [] @@ -319,6 +341,7 @@ class TaintAnalysis(FunctionAnalysis[TaintFacts]): SummaryAnalysis, ScopeAnalysis, SymbolAnalysis, + ProjectSummaries, } ) @@ -334,4 +357,5 @@ def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> TaintFacts: graph, ctx.get(SummaryAnalysis), entry_point_of(function, models, ctx.get(ScopeAnalysis), ctx.get(SymbolAnalysis)), + ctx.get(ProjectSummaries), ) diff --git a/tests/test_project.py b/tests/test_project.py new file mode 100644 index 0000000..742de43 --- /dev/null +++ b/tests/test_project.py @@ -0,0 +1,268 @@ +"""Acceptance tests for multi-file analysis (``docs/architecture.md`` §21). + +``engine.analyze_project`` discovers the Python files under a directory, builds a +``ModuleGraph`` of the imports between project modules, analyses every module with its +own manager, and shares a project-wide ``SummaryIndex`` so taint follows calls into +functions defined in other files. Summaries are iterated to a fixpoint across modules +without retaining more than one module's PyIR at a time in a given manager. + +Expected to remain red until ``coretrace_python.interprocedural.modulegraph`` and +``engine.analyze_project`` exist. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.cli import main +from coretrace_python.findings import Finding +from coretrace_python.semantic.symbols import SymbolId + +try: + from coretrace_python.interprocedural.modulegraph import ( + ModuleGraph, + ProjectSummaries, + SummaryIndex, + ) +except ImportError as error: # pragma: no cover - red until the module graph lands + MISSING = error +else: + MISSING = None + + +@pytest.fixture(autouse=True) +def require_module_graph() -> None: + if MISSING is not None: + pytest.fail(f"multi-file analysis is not implemented yet: {MISSING}") + if not hasattr(engine, "analyze_project"): + pytest.fail("engine.analyze_project is not implemented yet") + + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "plugins" + + +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 rules(findings: tuple[Finding, ...]) -> list[tuple[str, str, int]]: + return [(Path(str(f.span.source_id)).name, f.rule_id, f.span.start_line) for f in findings] + + +# --------------------------------------------------------------------------- discovery + + +def test_discovers_modules_and_their_dotted_names(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "app/__init__.py": "", + "app/main.py": "import os\n", + "app/helpers.py": "", + "scripts/tool.py": "", + ".venv/lib/site.py": "", + "node_modules/x.py": "", + "app/__pycache__/main.cpython-313.py": "", + }, + ) + + graph = engine.analyze_project(root).graph + + assert isinstance(graph, ModuleGraph) + assert graph.modules == ("app", "app.helpers", "app.main", "tool") + assert graph.source("app.main").path == (root / "app" / "main.py").resolve() + + +def test_module_graph_records_imports_between_project_modules(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "app/__init__.py": "", + "app/main.py": "import os\nfrom app.helpers import run\nfrom . import config\n", + "app/helpers.py": "import app.config\n", + "app/config.py": "", + }, + ) + + graph = engine.analyze_project(root).graph + + assert graph.imports("app.main") == frozenset({"app.helpers", "app.config"}) + assert graph.imports("app.helpers") == frozenset({"app.config"}) + assert graph.imports("app.config") == frozenset() + assert graph.importers("app.config") == frozenset({"app.main", "app.helpers"}) + + +def test_project_functions_have_project_symbols(tmp_path: Path) -> None: + root = project( + tmp_path, + {"app/__init__.py": "", "app/helpers.py": "def run(x):\n return x\n\nclass K:\n def m(self):\n pass\n"}, + ) + + index = engine.analyze_project(root).index + + assert isinstance(index, SummaryIndex) + summary = index.summary(SymbolId("python.app.helpers.run")) + assert summary is not None and summary.return_dependencies == frozenset({0}) + assert index.summary(SymbolId("python.app.helpers.K.m")) is not None + assert index.summary(SymbolId("python.os.system")) is None + assert SymbolId("python.app.helpers.run") in index.symbols + + +# --------------------------------------------------------------------------- cross-module taint + + +def test_taint_flows_into_a_function_defined_in_another_file(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "app/__init__.py": "", + "app/helpers.py": "import os\n\ndef execute(command):\n os.system(command)\n", + "app/main.py": "from app.helpers import execute\n\ndef run():\n execute(input())\n", + }, + ) + + findings = engine.analyze_project(root, [PLUGINS]).findings + + assert rules(findings) == [("main.py", "command-injection", 4)] + assert findings[0].metadata["through"] == "app.helpers.execute" + assert findings[0].metadata["sink_line"] == "4" + + +def test_module_attribute_calls_resolve_too(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "app/__init__.py": "", + "app/helpers.py": "import os\n\ndef execute(command):\n os.system(command)\n", + "app/main.py": "import app.helpers\n\ndef run():\n app.helpers.execute(input())\n", + }, + ) + + assert rules(engine.analyze_project(root, [PLUGINS]).findings) == [("main.py", "command-injection", 4)] + + +def test_tainted_results_cross_files(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "app/__init__.py": "", + "app/io.py": "def read():\n return input()\n", + "app/main.py": "import os\nfrom app.io import read\n\ndef run():\n os.system(read())\n", + }, + ) + + assert rules(engine.analyze_project(root, [PLUGINS]).findings) == [("main.py", "command-injection", 5)] + + +def test_relative_imports_inside_packages(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "app/__init__.py": "", + "app/helpers.py": "import os\n\ndef execute(command):\n os.system(command)\n", + "app/main.py": "from .helpers import execute\n\ndef run():\n execute(input())\n", + }, + ) + + assert rules(engine.analyze_project(root, [PLUGINS]).findings) == [("main.py", "command-injection", 4)] + + +def test_relative_imports_inside_a_package_init(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "app/__init__.py": "from .helpers import execute\n\ndef run():\n execute(input())\n", + "app/helpers.py": "import os\n\ndef execute(command):\n os.system(command)\n", + }, + ) + + result = engine.analyze_project(root, [PLUGINS]) + + assert result.graph.imports("app") == frozenset({"app.helpers"}) + assert rules(result.findings) == [("__init__.py", "command-injection", 4)] + + +def test_mutual_recursion_across_files_converges(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "a.py": "import os\nfrom b import g\n\ndef f(x):\n return g(x)\n\ndef run():\n os.system(f(input()))\n", + "b.py": "from a import f\n\ndef g(x):\n if x:\n return f(x)\n return x\n", + }, + ) + + result = engine.analyze_project(root, [PLUGINS]) + + assert result.index.summary(SymbolId("python.a.f")).return_dependencies == frozenset({0}) # type: ignore[union-attr] + assert rules(result.findings) == [("a.py", "command-injection", 8)] + + +def test_single_file_projects_behave_like_check(tmp_path: Path) -> None: + root = project(tmp_path, {"one.py": "import os\n\ndef run():\n os.system(input())\n"}) + assert rules(engine.analyze_project(root, [PLUGINS]).findings) == [("one.py", "command-injection", 4)] + + +def test_unsupported_functions_and_syntax_errors_are_reported_per_file(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "good.py": "def f():\n pass\n", + "nested.py": "def outer():\n def inner():\n pass\n return inner\n", + "broken.py": "def broken(:\n", + }, + ) + + findings = engine.analyze_project(root, [PLUGINS]).findings + + assert rules(findings) == [("broken.py", "syntax-error", 1), ("nested.py", "unsupported-syntax", 1)] + assert "broken.py:1:12" in findings[0].message + + +def test_project_summaries_default_to_an_empty_index_and_can_be_provided() -> None: + assert ProjectSummaries.name == "interprocedural.project" + manager = engine.build_manager(engine.build_hir(engine.SourceManager().add_source("x.py", ""))) + assert manager.get(ProjectSummaries).symbols == () + + provided = SummaryIndex() + bare = engine.AnalysisManager(engine.build_hir(engine.SourceManager().add_source("y.py", ""))) + bare.register(ProjectSummaries) + bare.provide(ProjectSummaries, provided) + assert bare.get(ProjectSummaries) is provided + + +# --------------------------------------------------------------------------- CLI + + +def test_check_accepts_a_directory(tmp_path: Path, capsys) -> None: # type: ignore[no-untyped-def] + root = project( + tmp_path, + { + "app/__init__.py": "", + "app/helpers.py": "import os\n\ndef execute(command):\n os.system(command)\n", + "app/main.py": "from app.helpers import execute\n\ndef run():\n execute(input())\n", + "app/clean.py": "def ok():\n return 1\n", + }, + ) + + exit_code = main(["--check", str(root), "--plugins", str(PLUGINS)]) + output = capsys.readouterr().out + + assert exit_code == 1 + assert output == ( + f"{(root / 'app' / 'main.py').resolve()}:4:5: high command-injection: Command injection:" + " stdin input reaches python.os.system through app.helpers.execute [run]\n" + "1 finding\n" + ) + + +def test_emit_ir_rejects_directories(tmp_path: Path, capsys) -> None: # type: ignore[no-untyped-def] + assert main(["--emit-ir", str(tmp_path)]) == 2 + assert "directory" in capsys.readouterr().err