diff --git a/CHANGELOG.md b/CHANGELOG.md index eb5bfee..5d1f88c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## Unreleased + +### Plugins + +- `ProjectContext.functions(module)` gives project plugins every function of a module as its call graph names it, with its span and the label of the entry point it is (`http` for a route, `argv` for a command), decided by the taint engine's own rule; modules served from the cache keep them, and the cache format is bumped (#116). + ## 0.4.0 (2026-09-24) ### Analysis diff --git a/docs/plugins.md b/docs/plugins.md index dba0bab..a6c7348 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -222,9 +222,12 @@ rule. Findings carry a redacted preview only. Runs once per project after every module is analysed, with a `ProjectContext`: the module graph, the dependency graph, every advisory the plugins and advisory files -contributed, the policy, the project root, and each module's imports and call graph. The -shipped `vulnerable-dependency`, `reachable-vulnerability` and `dependency-policy` plugins -are of this kind. +contributed, the policy, the project root, and each module's imports, call graph and +functions. `ctx.functions(module)` gives every function as the call graph names it +(`Admin.get`, `outer.inner`, ``), with its span, so a finding can be placed in +one, and the label of the entry point it is (`http` for a route, `argv` for a command) +when the models make it one. The shipped `vulnerable-dependency`, +`reachable-vulnerability` and `dependency-policy` plugins are of this kind. ```python from collections.abc import Sequence @@ -238,7 +241,7 @@ class RequirementsPresent(ProjectPlugin): name: ClassVar[str] = "requirements-present" def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]: - return () # inspect ctx.dependencies, ctx.modules(), ctx.call_graph(module) + return () # inspect ctx.dependencies, ctx.modules, ctx.call_graph(module), ctx.functions(module) ``` ### Anything else: `Plugin` diff --git a/src/coretrace_python/cache.py b/src/coretrace_python/cache.py index 0f43b4f..68ecbe2 100644 --- a/src/coretrace_python/cache.py +++ b/src/coretrace_python/cache.py @@ -27,6 +27,7 @@ ExternalSymbol, FunctionSummary, KnownFunction, + ModuleFunction, ModuleGraph, Mutation, NonlocalWrite, @@ -37,14 +38,14 @@ from coretrace_python.semantic.symbols import SymbolId from coretrace_python.source import SourceId, SourceSpan -CACHE_FORMAT = 5 +CACHE_FORMAT = 6 @dataclass(frozen=True) class CachedModule: """Everything a later run needs from one module without lowering it again.""" - functions: tuple[str, ...] + functions: tuple[ModuleFunction, ...] summaries: Mapping[str, FunctionSummary] sites: tuple[CallSite, ...] findings: tuple[Finding, ...] @@ -128,7 +129,7 @@ def store(self, key: str, module: CachedModule) -> None: def encode(module: CachedModule) -> dict[str, Any]: return { "format": CACHE_FORMAT, - "functions": list(module.functions), + "functions": [[f.name, _encode_span(f.span), f.entry_point] for f in module.functions], "summaries": {name: _encode_summary(s) for name, s in module.summaries.items()}, "sites": [_encode_site(site) for site in module.sites], "findings": [_encode_finding(finding) for finding in module.findings], @@ -139,7 +140,7 @@ def decode(data: Mapping[str, Any]) -> CachedModule: if data["format"] != CACHE_FORMAT: raise ValueError(f"unsupported cache format {data['format']!r}") return CachedModule( - tuple(_string(name) for name in data["functions"]), + tuple(_decode_function(function) for function in data["functions"]), {_string(name): _decode_summary(s) for name, s in data["summaries"].items()}, tuple(_decode_site(site) for site in data["sites"]), tuple(_decode_finding(finding) for finding in data["findings"]), @@ -187,6 +188,11 @@ def _decode_span(data: Any) -> SourceSpan: ) +def _decode_function(data: Any) -> ModuleFunction: + name, span, entry_point = data + return ModuleFunction(_string(name), _decode_span(span), None if entry_point is None else _string(entry_point)) + + def _encode_finding(finding: Finding) -> dict[str, Any]: return { "rule": finding.rule_id, diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index e62cbb6..b574597 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -71,6 +71,7 @@ CallGraphAnalysis, CallSite, FunctionSummary, + ModuleFunction, ModuleGraph, ProjectSummaries, SummaryAnalysis, @@ -106,6 +107,7 @@ from coretrace_python.source import SourceFile, SourceId, SourceManager, SourceSpan, decode_text from coretrace_python.taint import ( EntryPoint, + EntryPointAnalysis, ModelTable, RegisteredRoutes, Routes, @@ -139,6 +141,7 @@ ProjectSummaries, SecurityModelAnalysis, TaintAnalysis, + EntryPointAnalysis, RefutationAnalysis, DependencyAnalysis, RegisteredRoutes, @@ -147,7 +150,7 @@ # Named so the set stays right whichever analyses a build registers. _PROJECT_DEPENDANT_NAMES = frozenset( - {"interprocedural.project", "interprocedural.summaries", "taint.flows", "findings.refutation"} + {"interprocedural.project", "interprocedural.summaries", "taint.flows", "taint.entry_points", "findings.refutation"} ) _PROJECT_DEPENDANTS: frozenset[AnyAnalysis] = frozenset( a for a in ALL_ANALYSES if a.name in _PROJECT_DEPENDANT_NAMES @@ -451,6 +454,7 @@ def analyze_project( index = _seed(results, graph, frozenset()) call_graphs: dict[str, CallGraph] = {} + functions: dict[str, tuple[ModuleFunction, ...]] = {} for name in sorted(analysable): entry = results[name] findings.extend(entry.findings) @@ -458,11 +462,12 @@ def analyze_project( coverage.append( FileCoverage(str(files[name].source_id), "analysed", len(entry.functions), len(entry.functions) - unsupported) ) - sites: dict[str, list[CallSite]] = {function: [] for function in entry.functions} + functions[name] = entry.functions + sites: dict[str, list[CallSite]] = {function.name: [] 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, policy, root) + context = ProjectContext(graph, dependencies, advisories, analysable, call_graphs, policy, root, functions) for plugin in all_plugins: if isinstance(plugin, ProjectPlugin): findings.extend(plugin.analyze_project(context)) @@ -662,7 +667,7 @@ def _analyse_module( ) ) return CachedModule( - graph.functions, + manager.get(EntryPointAnalysis), _summaries_of(manager), tuple(site for function in graph.functions for site in graph.sites(function)), (*findings, *correlated), diff --git a/src/coretrace_python/interprocedural/__init__.py b/src/coretrace_python/interprocedural/__init__.py index d9ef053..379d9fb 100644 --- a/src/coretrace_python/interprocedural/__init__.py +++ b/src/coretrace_python/interprocedural/__init__.py @@ -6,6 +6,7 @@ CallSite, ExternalSymbol, KnownFunction, + ModuleFunction, Target, UnknownTarget, ) @@ -35,6 +36,7 @@ "ExternalSymbol", "FunctionSummary", "KnownFunction", + "ModuleFunction", "ModuleGraph", "Mutation", "NonlocalWrite", diff --git a/src/coretrace_python/interprocedural/callgraph.py b/src/coretrace_python/interprocedural/callgraph.py index f39ced5..f589756 100644 --- a/src/coretrace_python/interprocedural/callgraph.py +++ b/src/coretrace_python/interprocedural/callgraph.py @@ -67,6 +67,17 @@ class CallSite: keywords: int +@dataclass(frozen=True) +class ModuleFunction: + """One function of a module as a project plugin sees it: the name the call graph + gives it (``Class.method``, ``outer.inner``, ````), where it is, and the label + of the entry point it is (``http`` for a route, ``argv`` for a command), if any.""" + + name: str + span: SourceSpan + entry_point: str | None = None + + class CallGraph: def __init__( self, diff --git a/src/coretrace_python/plugins/api.py b/src/coretrace_python/plugins/api.py index 22078f4..6f052fa 100644 --- a/src/coretrace_python/plugins/api.py +++ b/src/coretrace_python/plugins/api.py @@ -25,10 +25,15 @@ from coretrace_python.dependency import Advisory, DependencyGraph, Policy from coretrace_python.findings import Finding from coretrace_python.hir import nodes -from coretrace_python.interprocedural import CallGraph, CallGraphAnalysis, ModuleGraph +from coretrace_python.interprocedural import ( + CallGraph, + CallGraphAnalysis, + ModuleFunction, + ModuleGraph, +) from coretrace_python.ir.lowering import analyzable_functions from coretrace_python.semantic.imports import ImportAnalysis, ImportTable -from coretrace_python.taint import Model +from coretrace_python.taint import EntryPointAnalysis, Model PLUGIN_API_VERSION = 1 @@ -57,8 +62,8 @@ def analyze(self, ctx: PluginContext) -> Sequence[Finding]: class ProjectContext: """What a project-scoped plugin sees: the module graph, the dependency graph, the advisories every plugin and advisory file contributed, the dependency policy, and - each module's imports and call graph. The engine passes the call graphs of modules - it served from its cache.""" + each module's imports, call graph and functions. The engine passes the call graphs + and functions of modules it served from its cache.""" def __init__( self, @@ -69,6 +74,7 @@ def __init__( call_graphs: Mapping[str, CallGraph] | None = None, policy: Policy | None = None, root: Path | None = None, + functions: Mapping[str, tuple[ModuleFunction, ...]] | None = None, ) -> None: self.root = root self.graph = graph @@ -77,6 +83,7 @@ def __init__( self.policy = policy or Policy() self._managers = managers self._call_graphs = dict(call_graphs or {}) + self._functions = dict(functions or {}) @property def modules(self) -> tuple[str, ...]: @@ -89,6 +96,13 @@ def call_graph(self, module: str) -> CallGraph: cached = self._call_graphs.get(module) return cached if cached is not None else self._managers[module].get(CallGraphAnalysis) + def functions(self, module: str) -> tuple[ModuleFunction, ...]: + """The module's functions as its call graph names them, each with its span, so a + finding can be placed in one, and the label of the entry point it is, if any.""" + + cached = self._functions.get(module) + return cached if cached is not None else self._managers[module].get(EntryPointAnalysis) + class ProjectPlugin(Plugin): """A plugin that runs once per project rather than once per module (§26).""" diff --git a/src/coretrace_python/taint/__init__.py b/src/coretrace_python/taint/__init__.py index f0898bd..769d2cc 100644 --- a/src/coretrace_python/taint/__init__.py +++ b/src/coretrace_python/taint/__init__.py @@ -1,6 +1,7 @@ """Security models and the global multi-kind taint engine (architecture §16, §17).""" from coretrace_python.taint.engine import ( + EntryPointAnalysis, Taint, TaintAnalysis, TaintFacts, @@ -30,6 +31,7 @@ __all__ = [ "AuthorizationGuard", "EntryPoint", + "EntryPointAnalysis", "Model", "ModelError", "ModelTable", diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index caeb6c9..30d657e 100644 --- a/src/coretrace_python/taint/engine.py +++ b/src/coretrace_python/taint/engine.py @@ -22,7 +22,7 @@ HeapLocation, mutated_by, ) -from coretrace_python.analysis import AnalysisContext, AnyAnalysis, FunctionAnalysis +from coretrace_python.analysis import Analysis, AnalysisContext, AnyAnalysis, FunctionAnalysis from coretrace_python.cfg import CFG, BlockId, CFGAnalysis from coretrace_python.dataflow import DataflowProblem, Direction, solve from coretrace_python.hir import nodes @@ -32,6 +32,7 @@ ExternalSymbol, FunctionSummary, KnownFunction, + ModuleFunction, ProjectSummaries, SummaryAnalysis, SummaryIndex, @@ -684,7 +685,7 @@ def entry_point_of( return None -def parameter_sources( +def function_entry_point( function: nodes.Function, module: nodes.Module, models: ModelTable, @@ -692,16 +693,11 @@ def parameter_sources( symbols: SymbolTable, instances: Instances | None = None, routes: Routes | None = None, -) -> Mapping[int, Source]: - """The attacker-controlled parameters of ``function``, by index: every parameter of an - entry point (``self`` excepted for a method), including one registered elsewhere - (``routes``), and every parameter annotated with a typed-parameter symbol.""" +) -> EntryPoint | None: + """The entry point ``function`` is: one of its decorators or, for a method, one of its + class's bases matching an entry-point model, or a registration elsewhere (``routes``).""" - owner = next( - (s for s in module.body if isinstance(s, nodes.Class) and any(f is function for f in s.body)), - None, - ) - sources: dict[int, Source] = {} + owner = _owner(module, function) enclosing_function = _enclosing(module, function) if enclosing_function is not None: # ``app = Flask(__name__)`` inside ``create_app``: routes defined there resolve. @@ -712,6 +708,34 @@ def parameter_sources( entry = routes.get(project_symbol(module.name, qualified)) if entry is None and owner is not None: entry = routes.get(project_symbol(module.name, owner.name)) + return entry + + +def _owner(module: nodes.Module, function: nodes.Function) -> nodes.Class | None: + """The top-level class ``function`` is a method of, if any.""" + + return next( + (s for s in module.body if isinstance(s, nodes.Class) and any(f is function for f in s.body)), + None, + ) + + +def parameter_sources( + function: nodes.Function, + module: nodes.Module, + models: ModelTable, + scopes: ScopeTable, + symbols: SymbolTable, + instances: Instances | None = None, + routes: Routes | None = None, +) -> Mapping[int, Source]: + """The attacker-controlled parameters of ``function``, by index: every parameter of an + entry point (``self`` excepted for a method), including one registered elsewhere + (``routes``), and every parameter annotated with a typed-parameter symbol.""" + + owner = _owner(module, function) + sources: dict[int, Source] = {} + entry = function_entry_point(function, module, models, scopes, symbols, instances, routes) if entry is not None: first = 1 if owner is not None else 0 for index in range(first, len(function.parameters)): @@ -817,6 +841,38 @@ def sources_of(member: nodes.Function) -> Mapping[int, Source]: ) +class EntryPointAnalysis(Analysis[tuple[ModuleFunction, ...]]): + """The module's functions as the call graph names them, each with its span and the + label of the entry point it is, decided as the taint engine decides which parameters + are attacker-controlled. What a project plugin sees of a module, cached with it.""" + + name: ClassVar[str] = "taint.entry_points" + requires: ClassVar[frozenset[AnyAnalysis]] = frozenset( + { + CallGraphAnalysis, + SecurityModelAnalysis, + ScopeAnalysis, + SymbolAnalysis, + SummaryAnalysis, + ProjectSummaries, + RegisteredRoutes, + } + ) + + @classmethod + def compute(cls, ctx: AnalysisContext) -> tuple[ModuleFunction, ...]: + graph = ctx.get(CallGraphAnalysis) + models = ctx.get(SecurityModelAnalysis) + scopes, symbols = ctx.get(ScopeAnalysis), ctx.get(SymbolAnalysis) + instances = factory_instances(ctx.module, scopes, symbols, ctx.get(SummaryAnalysis), ctx.get(ProjectSummaries)) + routes = ctx.get(RegisteredRoutes) + functions: list[ModuleFunction] = [] + for name, function in graph.definitions.items(): + entry = function_entry_point(function, ctx.module, models, scopes, symbols, instances, routes) + functions.append(ModuleFunction(name, function.span, entry.label if entry is not None else None)) + return tuple(functions) + + def self_seeds( function: nodes.Function, module: nodes.Module, diff --git a/tests/test_cache.py b/tests/test_cache.py index 011396e..419a7ae 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -68,7 +68,7 @@ def rules(findings: tuple[Finding, ...]) -> list[tuple[str, str, int]]: def test_findings_and_summaries_round_trip_through_json() -> None: - from coretrace_python.interprocedural import ExternalCall, FunctionSummary + from coretrace_python.interprocedural import ExternalCall, FunctionSummary, ModuleFunction from coretrace_python.semantic.symbols import SymbolId span = SourceSpan(SourceId("/p/a.py"), 3, 5, 3, 9) @@ -82,12 +82,13 @@ def test_findings_and_summaries_round_trip_through_json() -> None: frozenset({SymbolId("python.builtins.input")}), ) - text = json.dumps(encode(CachedModule(("f",), {"f": summary}, (), (finding,)))) + functions = (ModuleFunction("f", span, "http"), ModuleFunction("g", span)) + text = json.dumps(encode(CachedModule(functions, {"f": summary}, (), (finding,)))) restored = decode(json.loads(text)) assert restored.findings == (finding,) assert restored.summaries["f"] == summary - assert restored.functions == ("f",) + assert restored.functions == functions # --------------------------------------------------------------------------- keys diff --git a/tests/test_heap.py b/tests/test_heap.py index 8dfe789..b360ff4 100644 --- a/tests/test_heap.py +++ b/tests/test_heap.py @@ -278,7 +278,7 @@ def test_mutations_round_trip_through_the_cache_codec() -> None: summary = summary_of("def fill(items):\n items.append(input())\n", "fill") - restored = decode(json.loads(json.dumps(encode(CachedModule(("fill",), {"fill": summary}, (), ()))))) + restored = decode(json.loads(json.dumps(encode(CachedModule((), {"fill": summary}, (), ()))))) assert restored.summaries["fill"] == summary diff --git a/tests/test_project.py b/tests/test_project.py index 4134392..5708fd3 100644 --- a/tests/test_project.py +++ b/tests/test_project.py @@ -356,3 +356,102 @@ def refine(self, ctx: ProjectContext, findings: tuple[Finding, ...]) -> tuple[Fi with pytest.raises(RefinementError, match="tampering"): engine.analyze_project(root, [PLUGINS], plugins=[Tampering()]) + + +# --------------------------------------------------------------------------- functions and entry points + + +WEB = ( + "from flask import Flask\n" + "\n" + "app = Flask(__name__)\n" + "\n" + "@app.route('/')\n" + "def index():\n" + " return helper()\n" + "\n" + "def helper():\n" + " return 'x'\n" + "\n" + "class Admin:\n" + " def get(self):\n" + " return None\n" +) + + +def functions_seen(root: Path, **options: object) -> tuple[dict[str, dict[str, object]], object]: + """What a project plugin sees of each module's functions, by module then name, and the + analysis of the run.""" + + from coretrace_python.plugins import ProjectContext, ProjectPlugin + + seen: dict[str, dict[str, object]] = {} + + class Recorder(ProjectPlugin): + name = "recorder" + + def analyze_project(self, ctx: ProjectContext) -> tuple[Finding, ...]: + for module in ctx.modules: + seen[module] = {f.name: f for f in ctx.functions(module)} + return () + + analysis = engine.analyze_project(root, [PLUGINS], plugins=[Recorder()], **options) # type: ignore[arg-type] + return seen, analysis + + +def test_project_plugins_see_every_function_with_its_span_and_entry_point(tmp_path: Path) -> None: + root = project(tmp_path, {"app/__init__.py": "", "app/web.py": WEB}) + + web = functions_seen(root)[0]["app.web"] + + assert list(web) == ["", "index", "helper", "Admin.get"] + assert {name: f.entry_point for name, f in web.items()} == { # type: ignore[attr-defined] + "": None, + "index": "http", + "helper": None, + "Admin.get": None, + } + assert [(f.span.start_line, f.span.end_line) for f in web.values()] == [(1, 14), (6, 7), (9, 10), (13, 14)] # type: ignore[attr-defined] + + +def test_a_view_registered_in_another_module_is_an_entry_point(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "shop/__init__.py": "", + "shop/views.py": "def index(request):\n return 'x'\n\ndef internal():\n return 'y'\n", + "shop/urls.py": "from django.urls import path\n\nfrom shop import views\n\nurlpatterns = [path('', views.index)]\n", + }, + ) + + views = functions_seen(root)[0]["shop.views"] + + assert views["index"].entry_point == "http" # type: ignore[attr-defined] + assert views["internal"].entry_point is None # type: ignore[attr-defined] + + +def test_the_routes_of_an_app_made_by_a_factory_of_another_module_are_entry_points(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "app/__init__.py": "", + "app/factory.py": "from flask import Flask\n\ndef create_app():\n return Flask(__name__)\n", + "app/web.py": "from app.factory import create_app\n\napp = create_app()\n\n@app.route('/')\ndef index():\n return 'x'\n", + }, + ) + + assert functions_seen(root)[0]["app.web"]["index"].entry_point == "http" # type: ignore[attr-defined] + + +def test_modules_served_from_the_cache_keep_their_functions_and_entry_points(tmp_path: Path) -> None: + from coretrace_python.cache import ProjectCache + + root = project(tmp_path / "src", {"app/__init__.py": "", "app/web.py": WEB}) + cache = ProjectCache(tmp_path / "cache") + + first, _ = functions_seen(root, cache=cache) + second, analysis = functions_seen(root, cache=cache) + + assert "app.web" in analysis.reused # type: ignore[attr-defined] + assert second == first + assert second["app.web"]["index"].entry_point == "http" # type: ignore[attr-defined]