feat: give project plugins each module's functions with their span and entry point

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

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`, `<module>`), 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`

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,

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,18 +454,20 @@ 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)
         unsupported = sum(1 for f in entry.findings if f.rule_id == "unsupported-syntax")
         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),

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",

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``, ``<module>``), 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,

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)."""

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",

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,24 +685,19 @@ def entry_point_of(
     return None
 
 
-def parameter_sources(
+def function_entry_point(
     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."""
+) -> 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,