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