From 0bc13be06c6d67bb40b1976472a5d7a66d506658 Mon Sep 17 00:00:00 2001 From: Hugo Date: Mon, 28 Sep 2026 13:04:36 +0200 Subject: [PATCH 1/2] test: specify entry points by a pattern on the base class A gRPC servicer subclasses a class generated per project, so no fixed symbol names it. The tests specify a BasePattern model: the methods of a class whose base's canonical symbol matches the pattern are entry points whatever the module's prefix and however the base is imported, their parameters typed by their position (the message is input of every kind, the context denotes a class whose invocation_metadata() is a text source and whose peer() is not), only the methods the base defines when the project holds it, and every method with the listed arity otherwise. Async methods, client streams, other processes, a single-file check and a base preceded by another are covered. --- tests/test_base_patterns.py | 326 ++++++++++++++++++++++++++++++++++++ 1 file changed, 326 insertions(+) create mode 100644 tests/test_base_patterns.py diff --git a/tests/test_base_patterns.py b/tests/test_base_patterns.py new file mode 100644 index 0000000..3f0ba6d --- /dev/null +++ b/tests/test_base_patterns.py @@ -0,0 +1,326 @@ +"""Entry points by a pattern on the base class, with the parameters typed by their role. + +A gRPC servicer subclasses a class generated per project, ``greeter_pb2_grpc.GreeterServicer``, +so no fixed symbol names it, and its RPC methods receive the client's message and a +``ServicerContext`` without a decorator or an annotation saying so. A ``BasePattern`` +model matches the canonical symbol of a class's base against a regular expression, as +``NamedParameter`` matches parameter names, whatever the module's prefix and however +the base is imported, and says what the methods receive, by position after ``self``: a +parameter left untyped is attacker input of the model's kinds; one typed with a class +denotes that class, so a ``Source`` on one of its methods applies without the whole +object being tainted. + +Only the RPC methods are entry points: those overriding a method the base defines when +the project holds the generated module, otherwise those taking exactly the parameters +the model lists. A utility method of the servicer is not an RPC. Inheritance identifies +a potential service; it does not prove the servicer is registered on a server. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.findings import Finding +from coretrace_python.plugins import ProjectContext, ProjectPlugin +from coretrace_python.source import SourceManager + +MANIFEST = ( + 'name = "grpc-test-models"\nversion = "1.0.0"\nplugin_api = ">=1,<2"\nrequires = []\n' + 'provides = ["model.grpc-test"]\n\n[entrypoint]\nmodule = "grpc_models"\nclass = "Grpc"\n' +) +PLUGIN = ''' +from typing import ClassVar + +from coretrace_python.plugins import ModelPlugin +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.taint import TEXT_KINDS, BasePattern, Model, Sink, Source, TaintKind + +CONTEXT = SymbolId("python.grpc.ServicerContext") + + +class Grpc(ModelPlugin): + name: ClassVar[str] = "grpc-test-models" + models: ClassVar[tuple[Model, ...]] = ( + BasePattern(r"(?:^|\\.)\\w+_pb2_grpc\\.\\w+Servicer$", "grpc", parameters=(None, CONTEXT)), + Source(CONTEXT.attribute("invocation_metadata"), "grpc", TEXT_KINDS), + Sink(SymbolId("python.store.find"), TaintKind.NOSQL), + ) +''' +GENERATED = ( + "class GreeterServicer(object):\n" + " def SayHello(self, request, context):\n" + " raise NotImplementedError\n\n" + " def Stream(self, request_iterator, context):\n" + " raise NotImplementedError\n\n" + "def add_GreeterServicer_to_server(servicer, server):\n" + " pass\n" +) +HEADER = "import os\nimport store\nfrom google.protobuf.json_format import MessageToDict\n" +IN_PROJECT = ("from app.proto import greeter_pb2_grpc", "greeter_pb2_grpc.GreeterServicer") +ABSENT = ("import greeter_pb2_grpc", "greeter_pb2_grpc.GreeterServicer") +SAY_HELLO = " def SayHello(self, request, context):\n os.system(request.name)\n" +LINE = 8 + + +def plugins(tmp_path: Path) -> Path: + plugin = tmp_path / "plugins" / "grpc_models" + plugin.mkdir(parents=True, exist_ok=True) + (plugin / "plugin.toml").write_text(MANIFEST, encoding="utf-8") + (plugin / "grpc_models.py").write_text(PLUGIN, encoding="utf-8") + return tmp_path / "plugins" + + +def project(tmp_path: Path, files: dict[str, str]) -> Path: + root = tmp_path / "project" + for relative, text in {"app/__init__.py": "", "app/proto/__init__.py": "", **files}.items(): + (root / relative).parent.mkdir(parents=True, exist_ok=True) + (root / relative).write_text(text, encoding="utf-8") + return root + + +def servicer(base: tuple[str, str], body: str) -> str: + base_import, name = base + return f"{HEADER}{base_import}\n\nclass Greeter({name}):\n{body}" + + +def reported(findings: tuple[Finding, ...]) -> list[tuple[str, int, str]]: + return sorted( + (f.rule_id, f.span.start_line, f.metadata["source_label"]) + for f in findings + if f.rule_id in ("command-injection", "nosql-injection") + ) + + +def analyze(tmp_path: Path, files: dict[str, str], **options: object) -> list[tuple[str, int, str]]: + root = project(tmp_path, files) + roots = [engine.BUNDLED_PLUGINS, plugins(tmp_path)] + return reported(engine.analyze_project(root, roots, **options).findings) # type: ignore[arg-type] + + +# --------------------------------------------------------------------------- recognition + + +@pytest.mark.parametrize( + "base", + [ + IN_PROJECT, + ("from app.proto import greeter_pb2_grpc as pb", "pb.GreeterServicer"), + ("import app.proto.greeter_pb2_grpc as pb", "pb.GreeterServicer"), + ("from app.proto.greeter_pb2_grpc import GreeterServicer", "GreeterServicer"), + ("from app.proto.greeter_pb2_grpc import GreeterServicer as Base", "Base"), + ], + ids=["module", "module-alias", "import-as", "name", "name-alias"], +) +def test_a_class_whose_base_matches_the_pattern_receives_attacker_input(tmp_path: Path, base: tuple[str, str]) -> None: + files = {"app/proto/greeter_pb2_grpc.py": GENERATED, "app/servicer.py": servicer(base, SAY_HELLO)} + + assert analyze(tmp_path, files) == [("command-injection", LINE, "grpc")] + + +@pytest.mark.parametrize( + "first", + ["logging.Handler", "LoggingMixin"], + ids=["an-installed-class", "a-class-of-the-module"], +) +def test_a_servicer_with_another_base_before_the_generated_one_is_typed_all_the_same(tmp_path: Path, first: str) -> None: + body = SAY_HELLO + " os.system(dict(context.invocation_metadata())['x-user'])\n" + module = ( + f"{HEADER}import logging\nfrom app.proto import greeter_pb2_grpc\n\n" + "class LoggingMixin:\n" + " def log(self, message):\n" + " pass\n\n" + f"class Greeter({first}, greeter_pb2_grpc.GreeterServicer):\n{body}" + ) + files = {"app/proto/greeter_pb2_grpc.py": GENERATED, "app/servicer.py": module} + + assert analyze(tmp_path, files) == [("command-injection", LINE + 5, "grpc"), ("command-injection", LINE + 6, "grpc")] + + +def test_a_generated_module_absent_from_the_project_still_matches(tmp_path: Path) -> None: + assert analyze(tmp_path, {"app/servicer.py": servicer(ABSENT, SAY_HELLO)}) == [("command-injection", LINE, "grpc")] + + +@pytest.mark.parametrize( + "base", + [ + ("from app.proto import greeter_pb2", "greeter_pb2.GreeterServicer"), + ("from app.proto import greeter_pb2_grpc", "greeter_pb2_grpc.GreeterStub"), + ("from app import services", "services.GreeterServicer"), + ("import greeter_pb2_grpc", "greeter_pb2_grpc.GreeterServicerMixin"), + ], + ids=["not-a-grpc-module", "a-stub", "a-project-class-named-servicer", "a-name-going-on-after-servicer"], +) +def test_a_base_outside_the_pattern_is_no_entry_point(tmp_path: Path, base: tuple[str, str]) -> None: + files = { + "app/proto/greeter_pb2.py": "class GreeterServicer:\n pass\n", + "app/proto/greeter_pb2_grpc.py": GENERATED + "\nclass GreeterStub:\n pass\n\nclass GreeterServicerMixin:\n pass\n", + "app/services.py": "class GreeterServicer:\n def SayHello(self, request, context):\n pass\n", + "app/servicer.py": servicer(base, SAY_HELLO), + } + + assert analyze(tmp_path, files) == [] + + +def test_async_methods_and_client_streams_are_covered(tmp_path: Path) -> None: + body = ( + " async def SayHello(self, request, context):\n" + " os.system(request.name)\n\n" + " async def Stream(self, request_iterator, context):\n" + " async for message in request_iterator:\n" + " os.system(message.name)\n" + ) + files = {"app/proto/greeter_pb2_grpc.py": GENERATED, "app/servicer.py": servicer(IN_PROJECT, body)} + + assert analyze(tmp_path, files) == [("command-injection", LINE, "grpc"), ("command-injection", LINE + 4, "grpc")] + + +def test_a_sync_client_stream_is_covered(tmp_path: Path) -> None: + body = ( + " def Stream(self, request_iterator, context):\n" + " for message in request_iterator:\n" + " os.system(message.name)\n" + ) + files = {"app/proto/greeter_pb2_grpc.py": GENERATED, "app/servicer.py": servicer(IN_PROJECT, body)} + + assert analyze(tmp_path, files) == [("command-injection", LINE + 1, "grpc")] + + +@pytest.mark.parametrize("jobs", [1, 2]) +def test_modules_analysed_in_other_processes_recognise_the_servicer(tmp_path: Path, jobs: int) -> None: + body = SAY_HELLO + " os.system(dict(context.invocation_metadata())['x-user'])\n" + files = { + "app/proto/greeter_pb2_grpc.py": GENERATED, + "app/servicer.py": servicer(IN_PROJECT, body), + "app/other.py": "X = 1\n", + } + + expected = [("command-injection", LINE, "grpc"), ("command-injection", LINE + 1, "grpc")] + assert analyze(tmp_path, files, jobs=jobs) == expected + + +def test_a_decorator_matching_the_pattern_is_no_base(tmp_path: Path) -> None: + module = ( + f"{HEADER}import greeter_pb2_grpc\n\n" + "@greeter_pb2_grpc.GreeterServicer\n" + "def run(request, context, extra):\n" + " os.system(request)\n" + ) + + assert analyze(tmp_path, {"app/run.py": module}) == [] + + +def test_a_single_file_check_recognises_the_servicer_too(tmp_path: Path) -> None: + source = SourceManager().add_source("servicer.py", servicer(ABSENT, SAY_HELLO)) + + findings = engine.analyze_file(source, [engine.BUNDLED_PLUGINS, plugins(tmp_path)]).findings + + assert reported(findings) == [("command-injection", LINE, "grpc")] + + +# --------------------------------------------------------------------------- which methods + + +def test_only_the_methods_the_base_defines_are_rpcs_when_the_project_holds_it(tmp_path: Path) -> None: + body = ( + " def SayHello(self, request, context):\n" + " return self._greet(request.name, context)\n\n" + " def _greet(self, name, context):\n" + " os.system(name)\n\n" + " def render(self, request, context):\n" + " os.system(request.name)\n" + ) + files = {"app/proto/greeter_pb2_grpc.py": GENERATED, "app/servicer.py": servicer(IN_PROJECT, body)} + + # ``_greet`` and ``render`` are no RPCs: the only flow is the one ``SayHello`` passes + # to ``_greet``, reported at the call. + assert analyze(tmp_path, files) == [("command-injection", LINE, "grpc")] + + +def test_a_utility_method_of_another_arity_gets_no_typed_parameter(tmp_path: Path) -> None: + body = ( + " def SayHello(self, request, context):\n" + " return None\n\n" + " def _audit(self, name, ctx, extra):\n" + " os.system(dict(ctx.invocation_metadata())['x-user'])\n" + ) + files = {"app/proto/greeter_pb2_grpc.py": GENERATED, "app/servicer.py": servicer(IN_PROJECT, body)} + + # ``ctx`` is not a context the signature types: ``_audit`` takes three parameters. + assert analyze(tmp_path, files) == [] + + +def test_without_the_generated_module_the_rpc_signature_decides(tmp_path: Path) -> None: + body = ( + " def SayHello(self, request, context):\n" + " os.system(request.name)\n\n" + " def _load(self, name):\n" + " os.system(name)\n\n" + " def _log(self, message, level):\n" + " os.system(message)\n" + ) + files = {"app/servicer.py": servicer(ABSENT, body)} + + # Two parameters after ``self`` is the RPC signature; ``_log`` has it too, and + # without the generated module the engine cannot tell it from an RPC. + assert analyze(tmp_path, files) == [("command-injection", LINE, "grpc"), ("command-injection", LINE + 6, "grpc")] + + +def test_a_project_plugin_sees_the_rpc_methods_as_grpc_entry_points(tmp_path: Path) -> None: + seen: dict[str, str | None] = {} + + class Peek(ProjectPlugin): + name = "peek" + + def analyze_project(self, ctx: ProjectContext) -> tuple[Finding, ...]: + for module in ctx.modules: + for function in ctx.functions(module): + seen[f"{module}:{function.name}"] = function.entry_point + return () + + body = SAY_HELLO + "\n def _greet(self, name, context):\n pass\n" + files = {"app/proto/greeter_pb2_grpc.py": GENERATED, "app/servicer.py": servicer(IN_PROJECT, body)} + root = project(tmp_path, files) + + engine.analyze_project(root, [engine.BUNDLED_PLUGINS, plugins(tmp_path)], plugins=[Peek()]) + + assert seen["app.servicer:Greeter.SayHello"] == "grpc" + assert seen["app.servicer:Greeter._greet"] is None + assert seen["app.proto.greeter_pb2_grpc:GreeterServicer.SayHello"] is None + + +# --------------------------------------------------------------------------- what they receive + + +def test_the_message_carries_every_kind_including_its_dict_conversion(tmp_path: Path) -> None: + body = ( + " def SayHello(self, request, context):\n" + " store.find(MessageToDict(request))\n" + " store.find(request.filters)\n" + " store.find({'name': request.name})\n" + ) + files = {"app/proto/greeter_pb2_grpc.py": GENERATED, "app/servicer.py": servicer(IN_PROJECT, body)} + + assert analyze(tmp_path, files) == [ + ("nosql-injection", LINE, "grpc"), + ("nosql-injection", LINE + 1, "grpc"), + ("nosql-injection", LINE + 2, "grpc"), + ] + + +def test_the_context_gives_client_metadata_as_text_and_nothing_else(tmp_path: Path) -> None: + body = ( + " def SayHello(self, request, context):\n" + " for key, value in context.invocation_metadata():\n" + " os.system(value)\n" + " store.find({'key': dict(context.invocation_metadata())})\n" + " os.system(context.peer())\n" + " os.system(str(context))\n" + ) + files = {"app/proto/greeter_pb2_grpc.py": GENERATED, "app/servicer.py": servicer(IN_PROJECT, body)} + + # The metadata is text, so no ``nosql-injection``; ``peer()`` is transport + # information, and the context itself is not attacker input. + assert analyze(tmp_path, files) == [("command-injection", LINE + 1, "grpc")] From 04f07022cd1c6ffed311cc3f6fafebfd10a1e99b Mon Sep 17 00:00:00 2001 From: Hugo Date: Mon, 28 Sep 2026 13:04:36 +0200 Subject: [PATCH 2/2] feat: recognise entry points by a pattern on the base class BasePattern(pattern, label, parameters, kinds) matches the canonical symbol of a class's base against a regular expression, for bases generated per project such as a gRPC servicer's. Each parameter after self is, by position, 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. 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. The engine derives Signatures from the models and provides them to the call graph, which types the parameters by their position for every resolvable base of the class, since the inheritance map keeps the first base only. The taint engine decides the entry point with the project index, which says what the base defines. --- CHANGELOG.md | 6 ++ docs/plugins.md | 1 + src/coretrace_python/engine.py | 7 +++ .../interprocedural/__init__.py | 4 ++ .../interprocedural/callgraph.py | 59 ++++++++++++++++--- .../interprocedural/summaries.py | 6 ++ src/coretrace_python/taint/__init__.py | 2 + src/coretrace_python/taint/engine.py | 41 ++++++++++--- src/coretrace_python/taint/models.py | 44 +++++++++++++- 9 files changed, 155 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1a13d4d..9c4b555 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/plugins.md b/docs/plugins.md index da938cc..9f39592 100644 --- a/docs/plugins.md +++ b/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. | diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index 774515c..b877096 100644 --- a/src/coretrace_python/engine.py +++ b/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) diff --git a/src/coretrace_python/interprocedural/__init__.py b/src/coretrace_python/interprocedural/__init__.py index edc8ba2..a770a8c 100644 --- a/src/coretrace_python/interprocedural/__init__.py +++ b/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", diff --git a/src/coretrace_python/interprocedural/callgraph.py b/src/coretrace_python/interprocedural/callgraph.py index de41d84..2970f91 100644 --- a/src/coretrace_python/interprocedural/callgraph.py +++ b/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, diff --git a/src/coretrace_python/interprocedural/summaries.py b/src/coretrace_python/interprocedural/summaries.py index 819ecbf..0e2b770 100644 --- a/src/coretrace_python/interprocedural/summaries.py +++ b/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) diff --git a/src/coretrace_python/taint/__init__.py b/src/coretrace_python/taint/__init__.py index 7d76731..6114195 100644 --- a/src/coretrace_python/taint/__init__.py +++ b/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", diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index 046a885..5faffca 100644 --- a/src/coretrace_python/taint/engine.py +++ b/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) diff --git a/src/coretrace_python/taint/models.py b/src/coretrace_python/taint/models.py index 1cd08ed..e951556 100644 --- a/src/coretrace_python/taint/models.py +++ b/src/coretrace_python/taint/models.py @@ -17,7 +17,7 @@ from typing import ClassVar from coretrace_python.analysis import Analysis, AnalysisContext, MissingInputError -from coretrace_python.interprocedural import Clearing, TemplateCalls, TemplateFilter +from coretrace_python.interprocedural import Clearing, Signature, TemplateCalls, TemplateFilter from coretrace_python.semantic.symbols import Members, SymbolId @@ -145,6 +145,32 @@ def matches(self, name: str) -> bool: return re.search(self.pattern, name) is not None +@dataclass(frozen=True) +class BasePattern: + """Methods of a class whose base's canonical symbol matches ``pattern`` are entry + points labelled ``label``, when they take the parameters the model lists, by + position after ``self``: a position left None is attacker input carrying ``kinds``; + one holding a class symbol denotes an object of that class, which the sources on + the class taint. A gRPC servicer's RPCs take the request, or the iterator of the + client's messages, and a ``ServicerContext``. When the project holds the base, + only the methods it defines are entry points; otherwise every method with that + number of parameters. Inheritance identifies a potential service, not its + registration on a server.""" + + pattern: str + label: str + parameters: tuple[SymbolId | None, ...] = (None,) + kinds: TaintKind = TaintKind.ALL + + @property + def symbol(self) -> SymbolId: + digest = hashlib.sha1(self.pattern.encode("utf-8")).hexdigest()[:12] + return SymbolId(f"python.base.p{digest}") + + def matches(self, base: SymbolId) -> bool: + return re.search(self.pattern, base.canonical_name) is not None + + @dataclass(frozen=True) class RouteRegistrar: """A call registering a handler elsewhere (``path('login/', views.log_in)``): the @@ -251,6 +277,7 @@ class AuthorizationGuard: | TemplateRender | Members | RequestObject + | BasePattern ) @@ -270,6 +297,7 @@ class ModelTable: template_renders: tuple[TemplateRender, ...] = () members: tuple[Members, ...] = () request_objects: tuple[RequestObject, ...] = () + base_patterns: tuple[BasePattern, ...] = () _by_symbol: dict[type[Model], dict[SymbolId, Model]] = field( init=False, repr=False, compare=False ) @@ -287,6 +315,7 @@ def __post_init__(self) -> None: SafeArgument: {m.symbol: m for m in self.safe_arguments}, TemplateRender: {m.symbol: m for m in self.template_renders}, RequestObject: {m.symbol: m for m in self.request_objects}, + BasePattern: {m.symbol: m for m in self.base_patterns}, } object.__setattr__(self, "_by_symbol", MappingProxyType(index)) @@ -356,8 +385,20 @@ def extended(self, *sinks: Sink) -> ModelTable: self.template_renders, self.members, self.request_objects, + self.base_patterns, ) + def base_pattern(self, base: SymbolId) -> BasePattern | None: + """The pattern model the canonical symbol of a class's ``base`` matches, if any.""" + + return next((m for m in self.base_patterns if m.matches(base)), None) + + def signatures(self) -> tuple[Signature, ...]: + """What the methods of the classes the base patterns match receive, by position, + for the call graph to type them.""" + + return tuple(Signature(m.pattern, m.parameters) for m in self.base_patterns) + def sanitizer(self, symbol: SymbolId) -> Sanitizer | None: found = self._by_symbol[Sanitizer].get(symbol) return found if isinstance(found, Sanitizer) else None @@ -451,6 +492,7 @@ def freeze(self) -> ModelTable: template_renders=tuple(m for m in models if isinstance(m, TemplateRender)), members=tuple(m for m in models if isinstance(m, Members)), request_objects=tuple(m for m in models if isinstance(m, RequestObject)), + base_patterns=tuple(m for m in models if isinstance(m, BasePattern)), )