feat: recognise entry points by a pattern on the base class

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## Unreleased
+
+### Plugins
+
+- `BasePattern(pattern, label, parameters, kinds)` makes the methods of a class whose base's canonical symbol matches a regular expression entry points, for bases generated per project such as a gRPC servicer's, whatever the module's prefix and however the base is imported. Each parameter after `self` is, by position, either input of the model's kinds or an object of a class the model names, which the sources on that class taint without the object being input: a `Source` on `ServicerContext.invocation_metadata` gives the client's metadata, and `peer()` gives nothing. When the project holds the base, only the methods the base defines are entry points, so a servicer's utility methods are not; otherwise every method with the listed number of parameters is. `async` methods and the messages of an iterated parameter are covered (#176).
+
 ## 0.16.0 (2026-09-27)
 
 ### Advisories

docs/plugins.md

@@ -154,6 +154,7 @@ class BottleModels(ModelPlugin):
 | `EntryPoint(symbol, label, kinds=ALL)` | A decorator or a class base whose functions receive their parameters as `label` input, such as `flask.Flask.route`. |
 | `TypedParameter(symbol, label, kinds=ALL)` | A class whose annotated parameters carry `label` input, such as Django's `HttpRequest`. |
 | `NamedParameter(pattern, label, kinds)` | A regular expression on parameter names; matching parameters carry `kinds`, as `password` carries `CREDENTIAL`. |
+| `BasePattern(pattern, label, parameters=(None,), kinds=ALL)` | A regular expression on the canonical symbol of a class's base, for bases generated per project such as a gRPC servicer's (`(?:^|\.)\w+_pb2_grpc\.\w+Servicer$`), whatever the module's prefix and however the base is imported. The methods taking the parameters listed, by position after `self`, are entry points: a position left `None` is `label` input carrying `kinds`; one holding a class symbol denotes an object of that class, which the sources on the class taint, as `invocation_metadata()` on a `ServicerContext`, without the object itself being input. When the project holds the base, only the methods the base defines count; otherwise every method with that number of parameters. Inheritance identifies a potential service, not its registration on a server. |
 | `RouteRegistrar(symbol, argument, label, kinds=ALL, keyword=None)` | A call registering a view at argument `argument` (or `keyword`), such as Django's `path`; the view receives `label` input. |
 | `SuffixSink(suffix, kinds, positions=())` | A sink matched by the end of the symbol, for methods of any class such as `objects.raw`. |
 | `SafeArgument(symbol, argument, values, position=None, kinds=ALL)` | A call to the sink `symbol` whose `argument` (by keyword, or at `position`) denotes one of `values` is not a sink for `kinds`, such as `yaml.load` with `Loader=yaml.SafeLoader`. Values are written as the analyzer records arguments: a symbol by its canonical name, a constant as Python writes it (`True`). Only a value given explicitly makes the call safe. |

src/coretrace_python/engine.py

@@ -80,6 +80,7 @@
     ModuleFunction,
     ModuleGraph,
     ProjectSummaries,
+    SignaturesAnalysis,
     SummaryAnalysis,
     SummaryIndex,
     SymbolRead,
@@ -164,6 +165,7 @@
     EscapedTemplates,
     ClearingAnalysis,
     TemplateCallsAnalysis,
+    SignaturesAnalysis,
 )
 
 
@@ -205,6 +207,7 @@ class ResultsEvicted(TransformationPass):
             EscapedTemplates,
             ClearingAnalysis,
             TemplateCallsAnalysis,
+            SignaturesAnalysis,
         }
     )
 
@@ -268,6 +271,7 @@ def build_manager(
     table = (models or SecurityModelRegistry()).freeze()
     manager.provide(SecurityModelAnalysis, table)
     manager.provide(MembersAnalysis, table.members_by_class())
+    manager.provide(SignaturesAnalysis, table.signatures())
     manager.provide(ClearingAnalysis, table.clearing())
     manager.provide(ProjectSummaries, SummaryIndex())
     manager.provide(RegisteredRoutes, _routes_of(manager))
@@ -323,6 +327,7 @@ def analyze_file(source: SourceFile, plugin_roots: Sequence[Path]) -> FileAnalys
     table = plugin_models(loaded.plugin for loaded in registry)
     manager.provide(SecurityModelAnalysis, table)
     manager.provide(MembersAnalysis, table.members_by_class())
+    manager.provide(SignaturesAnalysis, table.signatures())
     manager.provide(ClearingAnalysis, table.clearing())
     manager.provide(ProjectSummaries, SummaryIndex())
     manager.provide(RegisteredRoutes, _routes_of(manager))
@@ -418,6 +423,7 @@ def analyze_project(
     for manager in managers.values():
         manager.provide(SecurityModelAnalysis, models)
         manager.provide(MembersAnalysis, members)
+        manager.provide(SignaturesAnalysis, models.signatures())
         manager.provide(DependencyAnalysis, dependencies)
 
     imports: dict[str, ImportTable] = {}
@@ -730,6 +736,7 @@ def _analyse_batch(batch: _Batch) -> dict[str, dict[str, Any]]:
     for manager in managers.values():
         manager.provide(SecurityModelAnalysis, models)
         manager.provide(MembersAnalysis, members)
+        manager.provide(SignaturesAnalysis, models.signatures())
         manager.provide(DependencyAnalysis, dependencies)
         manager.provide(RegisteredRoutes, routes)
         manager.provide(EscapedTemplates, escaped)

src/coretrace_python/interprocedural/__init__.py

@@ -8,6 +8,8 @@
     ExternalSymbol,
     KnownFunction,
     ModuleFunction,
+    Signature,
+    SignaturesAnalysis,
     SymbolRead,
     Target,
     UnknownTarget,
@@ -56,6 +58,8 @@
     "Mutation",
     "NonlocalWrite",
     "ProjectSummaries",
+    "Signature",
+    "SignaturesAnalysis",
     "SummaryAnalysis",
     "SummaryIndex",
     "SummaryTable",

src/coretrace_python/interprocedural/callgraph.py

@@ -9,6 +9,7 @@
 
 from __future__ import annotations
 
+import re
 from collections.abc import Mapping
 from dataclasses import dataclass, field
 from types import MappingProxyType
@@ -104,6 +105,30 @@ class SymbolRead:
     symbol: SymbolId
 
 
+@dataclass(frozen=True)
+class Signature:
+    """What the parameters of the methods of a class denote when the canonical symbol
+    of its base matches ``pattern``, by position after ``self``: a class symbol, or None
+    for a parameter denoting none. A method taking another number of parameters has
+    none. The engine derives signatures from the security models (§16)."""
+
+    pattern: str
+    parameters: tuple[SymbolId | None, ...]
+
+    def matches(self, base: SymbolId) -> bool:
+        return re.search(self.pattern, base.canonical_name) is not None
+
+
+class SignaturesAnalysis(Analysis[tuple[Signature, ...]]):
+    """The signatures the models declare, provided by the engine; none on its own."""
+
+    name: ClassVar[str] = "interprocedural.signatures"
+
+    @classmethod
+    def compute(cls, ctx: AnalysisContext) -> tuple[Signature, ...]:
+        return ()
+
+
 @dataclass(frozen=True)
 class ModuleFunction:
     """One function of a module as a project plugin sees it: the name the call graph
@@ -357,6 +382,20 @@ def _annotated(
     return found
 
 
+def _declared(
+    function: nodes.Function, ssa: FunctionIR, bases: tuple[SymbolId, ...], signatures: tuple[Signature, ...]
+) -> dict[Value, SymbolId]:
+    """Parameters a signature types by their position, for a method of a class one of
+    whose ``bases`` matches the signature and which takes the parameters it lists."""
+
+    given = function.parameters[1:]
+    for signature in signatures:
+        if any(signature.matches(base) for base in bases) and len(given) == len(signature.parameters):
+            typed = zip(ssa.parameters[1:], signature.parameters, strict=False)
+            return {value: symbol for value, symbol in typed if symbol is not None}
+    return {}
+
+
 def _typed_with_module_classes(
     function: nodes.Function, ssa: FunctionIR, classes: Mapping[str, frozenset[str]]
 ) -> dict[Value, str]:
@@ -372,12 +411,15 @@ def _typed_with_module_classes(
 
 class CallGraphAnalysis(Analysis[CallGraph]):
     name: ClassVar[str] = "interprocedural.callgraph"
-    requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({SSAAnalysis, ScopeAnalysis, SymbolAnalysis})
+    requires: ClassVar[frozenset[AnyAnalysis]] = frozenset(
+        {SSAAnalysis, ScopeAnalysis, SymbolAnalysis, SignaturesAnalysis}
+    )
 
     @classmethod
     def compute(cls, ctx: AnalysisContext) -> CallGraph:
         scopes = ctx.get(ScopeAnalysis)
         table = ctx.get(SymbolAnalysis)
+        signatures = ctx.get(SignaturesAnalysis)
         definitions: dict[str, nodes.Function] = {}
         for function in analyzable_functions(ctx.module):
             # A property and its setter, or a redefinition, share a qualified name; each
@@ -396,14 +438,15 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph:
             for statement in ctx.module.body
             if isinstance(statement, nodes.Class)
         }
+        # Inheritance follows the first base that resolves; a signature may match any.
         bases: dict[str, SymbolId] = {}
+        ancestry: dict[str, tuple[SymbolId, ...]] = {}
         for statement in ctx.module.body:
             if isinstance(statement, nodes.Class):
-                for base_expression in statement.bases:
-                    symbol = table.resolve_expression(scopes.module_scope.id, base_expression)
-                    if symbol is not None:
-                        bases[statement.name] = symbol
-                        break
+                resolved = [table.resolve_expression(scopes.module_scope.id, base) for base in statement.bases]
+                ancestry[statement.name] = tuple(symbol for symbol in resolved if symbol is not None)
+                if ancestry[statement.name]:
+                    bases[statement.name] = ancestry[statement.name][0]
         sites: dict[str, tuple[CallSite, ...]] = {}
         symbols: dict[str, Mapping[Value, SymbolId]] = {}
         reads: dict[str, tuple[SymbolRead, ...]] = {}
@@ -418,11 +461,13 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph:
             owner = name.rsplit(".", 1)[0] if "." in name and name.rsplit(".", 1)[0] in classes else None
             if owner is not None and _is_staticmethod(function):
                 owner = None  # a static method has no receiver
+            # A signature types parameters by their position; an annotation says more.
+            declared = _declared(function, ssa, ancestry.get(owner, ()) if owner is not None else (), signatures)
             targets, symbols[name] = resolve_targets(
                 ssa,
                 scopes,
                 known,
-                _annotated(function, ssa, scopes, table),
+                {**declared, **_annotated(function, ssa, scopes, table)},
                 frozenset(definitions),
                 classes,
                 owner,

src/coretrace_python/interprocedural/summaries.py

@@ -363,6 +363,12 @@ def __init__(self, summaries: Mapping[SymbolId, FunctionSummary] | None = None)
     def summary(self, symbol: SymbolId) -> FunctionSummary | None:
         return self.summaries.get(symbol)
 
+    def defines(self, owner: SymbolId) -> bool:
+        """Whether the project defines a function of ``owner``, a class or a module."""
+
+        prefix = owner.canonical_name + "."
+        return any(s.canonical_name.startswith(prefix) for s in self.symbols)
+
     def __eq__(self, other: object) -> bool:
         return isinstance(other, SummaryIndex) and dict(self.summaries) == dict(other.summaries)
 

src/coretrace_python/taint/__init__.py

@@ -12,6 +12,7 @@
 from coretrace_python.taint.models import (
     TEXT_KINDS,
     AuthorizationGuard,
+    BasePattern,
     EntryPoint,
     Model,
     ModelError,
@@ -47,6 +48,7 @@
     "FILTER_FUNCTIONS",
     "TEXT_KINDS",
     "AuthorizationGuard",
+    "BasePattern",
     "EntryPoint",
     "EntryPointAnalysis",
     "EscapedTemplates",

src/coretrace_python/taint/engine.py

@@ -80,6 +80,7 @@
 from coretrace_python.source import SourceSpan
 from coretrace_python.taint.models import (
     TEXT_KINDS,
+    BasePattern,
     EntryPoint,
     ModelTable,
     SecurityModelAnalysis,
@@ -780,9 +781,11 @@ def entry_point_of(
     symbols: SymbolTable,
     owner: nodes.Class | None = None,
     instances: Instances | None = None,
+    project: SummaryIndex | None = None,
 ) -> EntryPoint | None:
     """The entry-point model matching one of the function's decorators or, for a
-    method, one of the bases of ``owner``, if any."""
+    method, one of the bases of ``owner``, if any: by symbol, or by a base pattern when
+    the method is one the pattern covers (``project`` says what the base defines)."""
 
     scope = scopes.scope_for(function)
     enclosing = scope.parent if scope.parent is not None else scope.id
@@ -803,9 +806,24 @@ def entry_point_of(
             entry = models.entry_point(candidate)
             if entry is not None:
                 return entry
+            pattern = models.base_pattern(candidate) if id(expression) in enclosing_of else None
+            if pattern is not None and _covered(function, candidate, pattern, project):
+                return EntryPoint(candidate, pattern.label, pattern.kinds)
     return None
 
 
+def _covered(function: nodes.Function, base: SymbolId, pattern: BasePattern, project: SummaryIndex | None) -> bool:
+    """Whether a method of a class deriving from ``base`` is one ``pattern`` covers: it
+    takes the parameters the pattern lists and, when the project holds the base, the
+    base defines it; a utility method of the class is no entry point."""
+
+    if len(function.parameters) - 1 != len(pattern.parameters):
+        return False
+    if project is None or not project.defines(base):
+        return True
+    return project.summary(base.attribute(function.name)) is not None
+
+
 def function_entry_point(
     function: nodes.Function,
     module: nodes.Module,
@@ -814,6 +832,7 @@ def function_entry_point(
     symbols: SymbolTable,
     instances: Instances | None = None,
     routes: Routes | None = None,
+    project: SummaryIndex | None = None,
 ) -> 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``)."""
@@ -823,7 +842,7 @@ def function_entry_point(
     if enclosing_function is not None:
         # ``app = Flask(__name__)`` inside ``create_app``: routes defined there resolve.
         instances = {**(instances or {}), **local_instances(enclosing_function, scopes, symbols)}
-    entry = entry_point_of(function, models, scopes, symbols, owner, instances)
+    entry = entry_point_of(function, models, scopes, symbols, owner, instances, project)
     if entry is None and routes:
         qualified = function.name if owner is None else f"{owner.name}.{function.name}"
         entry = routes.get(project_symbol(module.name, qualified))
@@ -849,23 +868,29 @@ def parameter_sources(
     symbols: SymbolTable,
     instances: Instances | None = None,
     routes: Routes | None = None,
+    project: SummaryIndex | 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. An entry
     point's parameter annotated with scalars is text: FastAPI validates it before the
     handler runs, and a scalar cannot hold a query operator. So are the URL parameters
-    a view receives after a request object."""
+    a view receives after a request object. A parameter a base pattern types with a
+    class is not input: the sources on that class say what it gives."""
 
     owner = _owner(module, function)
     sources: dict[int, Source] = {}
     scope = scopes.scope_for(function)
     enclosing = scope.parent if scope.parent is not None else scope.id
-    entry = function_entry_point(function, module, models, scopes, symbols, instances, routes)
+    entry = function_entry_point(function, module, models, scopes, symbols, instances, routes, project)
     if entry is not None:
         first = 1 if owner is not None else 0
         request = models.request_object(entry.symbol)
+        pattern = models.base_pattern(entry.symbol) if models.entry_point(entry.symbol) is None else None
+        roles = pattern.parameters if pattern is not None else ()
         for index in range(first, len(function.parameters)):
+            if index - first < len(roles) and roles[index - first] is not None:
+                continue
             annotation = function.parameters[index].annotation
             text = (request is not None and index > first) or (
                 annotation is not None and _scalar(annotation, symbols, enclosing)
@@ -1010,11 +1035,12 @@ def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> TaintFacts:
         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))
+        project = ctx.get(ProjectSummaries)
+        instances = factory_instances(ctx.module, scopes, symbols, ctx.get(SummaryAnalysis), project)
         routes = ctx.get(RegisteredRoutes)
 
         def sources_of(member: nodes.Function) -> Mapping[int, Source]:
-            return parameter_sources(member, ctx.module, models, scopes, symbols, instances, routes)
+            return parameter_sources(member, ctx.module, models, scopes, symbols, instances, routes, project)
 
         ssa = ctx.get(SSAAnalysis, function)
         heap = ctx.get(HeapAnalysis, function)
@@ -1068,9 +1094,10 @@ def compute(cls, ctx: AnalysisContext) -> tuple[ModuleFunction, ...]:
         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)
+        project = ctx.get(ProjectSummaries)
         functions: list[ModuleFunction] = []
         for name, function in graph.definitions.items():
-            entry = function_entry_point(function, ctx.module, models, scopes, symbols, instances, routes)
+            entry = function_entry_point(function, ctx.module, models, scopes, symbols, instances, routes, project)
             functions.append(ModuleFunction(name, function.span, entry.label if entry is not None else None))
         return tuple(functions)