feat: registered routes and suffix sinks

README.md

@@ -113,9 +113,11 @@ in a database without passing through a hashing function it is a
 `plaintext-credential-storage` finding at medium confidence, since a name is a hint.
 Django views are undecorated, so a parameter annotated with a request class
 (`request: HttpRequest`), a method of a class-based view and a view decorated by one of
-Django's or Django REST framework's view decorators receive HTTP input; a bare
-`request` parameter without any of these is not recognised, and URL configurations are
-not read. Requests and httpx calls are SSRF sinks whose responses are untrusted
+Django's or Django REST framework's view decorators receive HTTP input, and so does a
+view registered in a URL configuration (`path('login/', views.log_in)`, `as_view()`,
+REST framework routers) or programmatically (`add_url_rule`, `add_api_route`): the
+engine reads every module's registrations before analysing. `Model.objects.raw` and
+`extra` are SQL sinks for any model class. Requests and httpx calls are SSRF sinks whose responses are untrusted
 `http-response` sources. Model plugins contribute sources, sinks and sanitizers;
 detectors consume the shared taint result, so adding a framework model makes every
 detector aware of it. Taint follows calls between functions of the same module through

plugins/models/django/django_models.py

@@ -17,8 +17,10 @@
     AuthorizationGuard,
     EntryPoint,
     Model,
+    RouteRegistrar,
     Sanitizer,
     Sink,
+    SuffixSink,
     TaintKind,
     TypedParameter,
 )
@@ -67,6 +69,14 @@
     "rest_framework.decorators.action",
 )
 
+_ROUTE_REGISTRARS = (
+    "django.urls.path",
+    "django.urls.re_path",
+    "django.conf.urls.url",
+    "rest_framework.routers.DefaultRouter.register",
+    "rest_framework.routers.SimpleRouter.register",
+)
+
 _AUTHORIZATION_DECORATORS = (
     ("django.contrib.auth.decorators.login_required", "login"),
     ("django.contrib.auth.decorators.permission_required", "permission"),
@@ -103,4 +113,11 @@ class DjangoModels(ModelPlugin):
         Sanitizer(_sym("django.utils.html.escape"), TaintKind.HTML),
         Sanitizer(_sym("django.utils.html.conditional_escape"), TaintKind.HTML),
         *(AuthorizationGuard(_sym(decorator), label) for decorator, label in _AUTHORIZATION_DECORATORS),
+        # ``urlpatterns = [path('login/', views.log_in)]``: the referenced view is an
+        # entry point wherever it is defined; routers register viewsets.
+        *(RouteRegistrar(_sym(registrar), 1, "http") for registrar in _ROUTE_REGISTRARS),
+        # ``Model.objects.raw(sql)`` for any model class: the statement is the first
+        # argument, its parameters are not.
+        SuffixSink("objects.raw", TaintKind.SQL, ((TaintKind.SQL, (0,)),)),
+        SuffixSink("objects.extra", TaintKind.SQL),
     )

plugins/models/fastapi/fastapi_models.py

@@ -6,7 +6,7 @@
 
 from coretrace_python.plugins import ModelPlugin
 from coretrace_python.semantic.symbols import SymbolId
-from coretrace_python.taint import EntryPoint, Model, Sink, TaintKind
+from coretrace_python.taint import EntryPoint, Model, RouteRegistrar, Sink, TaintKind
 
 _METHODS = ("get", "post", "put", "delete", "patch", "options", "head", "api_route", "websocket")
 
@@ -20,6 +20,8 @@ class FastApiModels(ModelPlugin):
     models: ClassVar[tuple[Model, ...]] = (
         *(EntryPoint(_sym(f"fastapi.FastAPI.{method}"), "http") for method in _METHODS),
         *(EntryPoint(_sym(f"fastapi.APIRouter.{method}"), "http") for method in _METHODS),
+        RouteRegistrar(_sym("fastapi.FastAPI.add_api_route"), 1, "http"),
+        RouteRegistrar(_sym("fastapi.APIRouter.add_api_route"), 1, "http"),
         Sink(_sym("fastapi.responses.HTMLResponse"), TaintKind.HTML),
         Sink(_sym("fastapi.responses.FileResponse"), TaintKind.PATH),
         Sink(_sym("fastapi.responses.RedirectResponse"), TaintKind.REDIRECT),

plugins/models/flask/flask_models.py

@@ -10,6 +10,7 @@
     AuthorizationGuard,
     EntryPoint,
     Model,
+    RouteRegistrar,
     Sanitizer,
     Sink,
     Source,
@@ -32,6 +33,8 @@ class FlaskModels(ModelPlugin):
         *(Source(_sym(f"flask.request.{attribute}"), "http") for attribute in _REQUEST_ATTRIBUTES),
         EntryPoint(_sym("flask.Flask.route"), "http"),
         EntryPoint(_sym("flask.Blueprint.route"), "http"),
+        RouteRegistrar(_sym("flask.Flask.add_url_rule"), 2, "http", keyword="view_func"),
+        RouteRegistrar(_sym("flask.Blueprint.add_url_rule"), 2, "http", keyword="view_func"),
         Sink(_sym("flask.render_template_string"), TaintKind.HTML),
         Sink(_sym("flask.make_response"), TaintKind.HTML),
         Sink(_sym("flask.Response"), TaintKind.HTML),

src/coretrace_python/engine.py

@@ -94,14 +94,19 @@
 from coretrace_python.reporters import Report
 from coretrace_python.semantic import SEMANTIC_ANALYSES
 from coretrace_python.semantic.imports import ImportAnalysis, ImportResolutionError, ImportTable
-from coretrace_python.semantic.scopes import ScopeError
-from coretrace_python.semantic.symbols import SymbolId
+from coretrace_python.semantic.scopes import ScopeAnalysis, ScopeError
+from coretrace_python.semantic.symbols import SymbolAnalysis, SymbolId
 from coretrace_python.source import SourceFile, SourceManager, SourceSpan, decode_text
 from coretrace_python.taint import (
+    EntryPoint,
     ModelTable,
+    RegisteredRoutes,
+    Routes,
     SecurityModelAnalysis,
     SecurityModelRegistry,
     TaintAnalysis,
+    TaintKind,
+    registered_routes,
 )
 
 TOOL_NAME = "coretrace-python-analyzer"
@@ -125,6 +130,7 @@
     TaintAnalysis,
     RefutationAnalysis,
     DependencyAnalysis,
+    RegisteredRoutes,
 )
 
 
@@ -157,7 +163,7 @@ class ResultsEvicted(TransformationPass):
 
     name: ClassVar[str] = "project.results-evicted"
     preserves: ClassVar[frozenset[AnyAnalysis]] = frozenset(
-        {*SEMANTIC_ANALYSES, SecurityModelAnalysis, ProjectSummaries, DependencyAnalysis}
+        {*SEMANTIC_ANALYSES, SecurityModelAnalysis, ProjectSummaries, DependencyAnalysis, RegisteredRoutes}
     )
 
     @classmethod
@@ -195,9 +201,21 @@ def build_manager(
     manager = _register_all(module)
     manager.provide(SecurityModelAnalysis, (models or SecurityModelRegistry()).freeze())
     manager.provide(ProjectSummaries, SummaryIndex())
+    manager.provide(RegisteredRoutes, _routes_of(manager))
     return manager
 
 
+def _routes_of(manager: AnalysisManager) -> dict[SymbolId, EntryPoint]:
+    """The handlers one module registers, given the models already provided."""
+
+    return registered_routes(
+        manager.module,
+        manager.get(ScopeAnalysis),
+        manager.get(SymbolAnalysis),
+        manager.get(SecurityModelAnalysis),
+    )
+
+
 def load_plugins(plugin_roots: Sequence[Path], manager: AnalysisManager) -> PluginRegistry:
     registry = PluginRegistry()
     for root in plugin_roots:
@@ -226,6 +244,7 @@ def analyze_file(source: SourceFile, plugin_roots: Sequence[Path]) -> FileAnalys
     registry = load_plugins(plugin_roots, manager)
     manager.provide(SecurityModelAnalysis, plugin_models(loaded.plugin for loaded in registry))
     manager.provide(ProjectSummaries, SummaryIndex())
+    manager.provide(RegisteredRoutes, _routes_of(manager))
     findings, supported = _check_module(manager, tuple(loaded.plugin for loaded in registry))
     functions = len(analyzable_functions(manager.module))
     coverage = Coverage((FileCoverage(str(source.source_id), "analysed", functions, len(supported)),))
@@ -331,8 +350,14 @@ def analyze_project(
     graph = build_module_graph(
         {name: files[name] for name in analysable}, {name: modules[name] for name in analysable}, imports
     )
+    routes: dict[SymbolId, EntryPoint] = {}
+    for name in sorted(analysable):
+        for symbol, registered in _routes_of(analysable[name]).items():
+            routes.setdefault(symbol, registered)
+    for manager in analysable.values():
+        manager.provide(RegisteredRoutes, routes)
 
-    configuration = _configuration_key(registry, plugins, models, advisories, dependencies)
+    configuration = _configuration_key(registry, plugins, models, advisories, dependencies, routes)
     keys = module_keys(
         graph,
         {name: fingerprint(configuration, str(files[name].source_id), name, files[name].text) for name in analysable},
@@ -372,6 +397,7 @@ def analyze_project(
                             {name: _path_of(files[name]) for name in sorted(component)},
                             encode_index(_seed(results, graph, component)),
                             advisory_paths,
+                            _encode_routes(routes),
                         ),
                     )
                     for component in pending
@@ -500,6 +526,7 @@ class _Batch:
     paths: Mapping[str, Path]
     seed: Mapping[str, Any]
     advisory_paths: tuple[Path, ...] = ()
+    routes: tuple[tuple[str, str, str, int], ...] = ()
 
 
 def _analyse_batch(batch: _Batch) -> dict[str, dict[str, Any]]:
@@ -519,9 +546,11 @@ def _analyse_batch(batch: _Batch) -> dict[str, dict[str, Any]]:
     )
     affected = affected_symbols(dependencies, advisories)
     models = plugin_models(all_plugins).extended(*advisory_sinks(affected))
+    routes = _decode_routes(batch.routes)
     for manager in managers.values():
         manager.provide(SecurityModelAnalysis, models)
         manager.provide(DependencyAnalysis, dependencies)
+        manager.provide(RegisteredRoutes, routes)
     module_plugins = tuple(p for p in all_plugins if not isinstance(p, ProjectPlugin))
     results = _analyse_managers(managers, decode_index(batch.seed), module_plugins, affected)
     return {name: encode(entry) for name, entry in results.items()}
@@ -532,12 +561,23 @@ def _path_of(source: SourceFile) -> Path:
     return source.path
 
 
+def _encode_routes(routes: Routes) -> tuple[tuple[str, str, str, int], ...]:
+    return tuple(
+        sorted((str(symbol), str(entry.symbol), entry.label, entry.kinds.value) for symbol, entry in routes.items())
+    )
+
+
+def _decode_routes(data: tuple[tuple[str, str, str, int], ...]) -> dict[SymbolId, EntryPoint]:
+    return {SymbolId(symbol): EntryPoint(SymbolId(registrar), label, TaintKind(kinds)) for symbol, registrar, label, kinds in data}
+
+
 def _configuration_key(
     registry: PluginRegistry,
     plugins: Sequence[Plugin],
     models: ModelTable,
     advisories: tuple[Advisory, ...],
     dependencies: DependencyGraph,
+    routes: Routes | None = None,
 ) -> str:
     """Everything a module's results depend on besides the project sources (§11)."""
 
@@ -556,6 +596,7 @@ def _configuration_key(
         repr(advisories),
         repr(dependencies.requirements),
         repr(dependencies.errors),
+        repr(_encode_routes(routes or {})),
     )
 
 

src/coretrace_python/taint/__init__.py

@@ -14,15 +14,18 @@
     ModelError,
     ModelTable,
     NamedParameter,
+    RouteRegistrar,
     Sanitizer,
     SecurityModelAnalysis,
     SecurityModelRegistry,
     Sink,
     Source,
+    SuffixSink,
     TaintKind,
     TypedParameter,
     Validator,
 )
+from coretrace_python.taint.routes import RegisteredRoutes, Routes, registered_routes
 
 __all__ = [
     "AuthorizationGuard",
@@ -31,11 +34,15 @@
     "ModelError",
     "ModelTable",
     "NamedParameter",
+    "RegisteredRoutes",
+    "RouteRegistrar",
+    "Routes",
     "Sanitizer",
     "SecurityModelAnalysis",
     "SecurityModelRegistry",
     "Sink",
     "Source",
+    "SuffixSink",
     "Taint",
     "TaintAnalysis",
     "TaintFacts",
@@ -44,4 +51,5 @@
     "TypedParameter",
     "Validator",
     "propagate_taint",
+    "registered_routes",
 ]

src/coretrace_python/taint/engine.py

@@ -36,6 +36,7 @@
     SummaryAnalysis,
     SummaryIndex,
     SummaryTable,
+    project_symbol,
 )
 from coretrace_python.ir.model import (
     BasicBlock,
@@ -71,6 +72,7 @@
     Source,
     TaintKind,
 )
+from coretrace_python.taint.routes import RegisteredRoutes, Routes
 
 
 @dataclass(frozen=True)
@@ -527,17 +529,23 @@ def parameter_sources(
     scopes: ScopeTable,
     symbols: SymbolTable,
     instances: Instances | None = None,
+    routes: Routes | None = None,
 ) -> Mapping[int, Source]:
     """The attacker-controlled parameters of ``function``, by index: every parameter of an
-    entry point (``self`` excepted for a method) and every parameter annotated with a
-    typed-parameter symbol."""
+    entry point (``self`` excepted for a method), including one registered elsewhere
+    (``routes``), and every parameter annotated with a typed-parameter symbol."""
 
     owner = next(
         (s for s in module.body if isinstance(s, nodes.Class) and any(f is function for f in s.body)),
         None,
     )
     sources: dict[int, Source] = {}
     entry = entry_point_of(function, models, scopes, symbols, owner, instances)
+    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))
+        if entry is None and owner is not None:
+            entry = routes.get(project_symbol(module.name, owner.name))
     if entry is not None:
         first = 1 if owner is not None else 0
         for index in range(first, len(function.parameters)):
@@ -601,6 +609,7 @@ class TaintAnalysis(FunctionAnalysis[TaintFacts]):
             SymbolAnalysis,
             ProjectSummaries,
             HeapAnalysis,
+            RegisteredRoutes,
         }
     )
 
@@ -628,6 +637,7 @@ def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> TaintFacts:
                     ctx.get(SummaryAnalysis),
                     ctx.get(ProjectSummaries),
                 ),
+                ctx.get(RegisteredRoutes),
             ),
             ctx.get(ProjectSummaries),
             ctx.get(HeapAnalysis, function),