feat: Django and HTTP client models

README.md

@@ -63,11 +63,16 @@ memory.
 
 `--plugins` is a directory searched recursively for `plugin.toml` manifests and may be
 repeated. The repository ships a first set under `plugins/`: security models for the
-standard library, Flask, FastAPI and SQLAlchemy (`models/`), taint detectors for SQL
+standard library, Flask, FastAPI, Django, SQLAlchemy and the Requests and httpx clients (`models/`), taint detectors for SQL
 injection, command injection, path traversal, SSRF and XSS (`security/`), and syntactic
 detectors for `eval`/`exec` and weak hashes (`syntax/`). Route handlers of Flask and
 FastAPI receive their parameters as HTTP input; `request.args` and its siblings are HTTP
-sources. Model plugins contribute sources, sinks and sanitizers;
+sources. 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
+`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
 function summaries: a tainted argument passed to a helper that reaches a sink is reported

plugins/models/django/django_models.py

@@ -0,0 +1,86 @@
+"""Django security models: typed requests, views, decorators, ORM and HTML sinks.
+
+Django views are undecorated, so HTTP input comes from three generic mechanisms: a
+parameter annotated with a request class, a method of a class-based view, or a view
+decorated by one of Django's or Django REST framework's view decorators. A view with a
+bare ``request`` parameter and none of these is not recognised; URL configurations are
+not read.
+"""
+
+from __future__ import annotations
+
+from typing import ClassVar
+
+from coretrace_python.plugins import ModelPlugin
+from coretrace_python.semantic.symbols import SymbolId
+from coretrace_python.taint import EntryPoint, Model, Sanitizer, Sink, TaintKind, TypedParameter
+
+_REQUEST_CLASSES = (
+    "django.http.HttpRequest",
+    "django.http.request.HttpRequest",
+    "django.core.handlers.wsgi.WSGIRequest",
+    "django.core.handlers.asgi.ASGIRequest",
+    "rest_framework.request.Request",
+)
+
+_VIEW_BASES = (
+    "django.views.View",
+    "django.views.generic.View",
+    "django.views.generic.base.View",
+    "django.views.generic.TemplateView",
+    "django.views.generic.RedirectView",
+    "django.views.generic.ListView",
+    "django.views.generic.DetailView",
+    "django.views.generic.FormView",
+    "django.views.generic.CreateView",
+    "django.views.generic.UpdateView",
+    "django.views.generic.DeleteView",
+    "rest_framework.views.APIView",
+    "rest_framework.generics.GenericAPIView",
+    "rest_framework.viewsets.ViewSet",
+    "rest_framework.viewsets.GenericViewSet",
+    "rest_framework.viewsets.ModelViewSet",
+    "rest_framework.viewsets.ReadOnlyModelViewSet",
+)
+
+_VIEW_DECORATORS = (
+    "django.views.decorators.csrf.csrf_exempt",
+    "django.views.decorators.csrf.csrf_protect",
+    "django.views.decorators.http.require_http_methods",
+    "django.views.decorators.http.require_GET",
+    "django.views.decorators.http.require_POST",
+    "django.views.decorators.http.require_safe",
+    "django.views.decorators.cache.never_cache",
+    "django.views.decorators.cache.cache_page",
+    "django.contrib.auth.decorators.login_required",
+    "django.contrib.auth.decorators.permission_required",
+    "django.contrib.auth.decorators.user_passes_test",
+    "rest_framework.decorators.api_view",
+    "rest_framework.decorators.action",
+)
+
+
+def _sym(path: str) -> SymbolId:
+    return SymbolId(f"python.{path}")
+
+
+class DjangoModels(ModelPlugin):
+    name: ClassVar[str] = "django-models"
+    models: ClassVar[tuple[Model, ...]] = (
+        *(TypedParameter(_sym(cls), "http") for cls in _REQUEST_CLASSES),
+        *(EntryPoint(_sym(base), "http") for base in _VIEW_BASES),
+        *(EntryPoint(_sym(decorator), "http") for decorator in _VIEW_DECORATORS),
+        Sink(_sym("django.db.connection.cursor.execute"), TaintKind.SQL),
+        Sink(_sym("django.db.connection.cursor.executemany"), TaintKind.SQL),
+        Sink(_sym("django.db.models.expressions.RawSQL"), TaintKind.SQL),
+        Sink(_sym("django.db.models.Manager.raw"), TaintKind.SQL),
+        Sink(_sym("django.db.models.QuerySet.raw"), TaintKind.SQL),
+        Sink(_sym("django.db.models.QuerySet.extra"), TaintKind.SQL),
+        Sink(_sym("django.utils.safestring.mark_safe"), TaintKind.HTML),
+        Sink(_sym("django.http.HttpResponse"), TaintKind.HTML),
+        Sink(_sym("django.http.response.HttpResponse"), TaintKind.HTML),
+        Sink(_sym("django.template.Template"), TaintKind.HTML),
+        Sink(_sym("django.http.FileResponse"), TaintKind.PATH),
+        Sanitizer(_sym("django.utils.html.escape"), TaintKind.HTML),
+        Sanitizer(_sym("django.utils.html.conditional_escape"), TaintKind.HTML),
+    )

plugins/models/django/plugin.toml

@@ -0,0 +1,9 @@
+name = "django-models"
+version = "1.0.0"
+plugin_api = ">=1,<2"
+requires = []
+provides = ["model.http-sources", "model.django-views", "model.sql-sinks"]
+
+[entrypoint]
+module = "django_models"
+class = "DjangoModels"

plugins/models/http_clients/http_client_models.py

@@ -0,0 +1,43 @@
+"""Requests and httpx models: every request function is a SSRF sink, and what it returns
+is an ``http-response`` source, so data fetched from a remote server is untrusted."""
+
+from __future__ import annotations
+
+from typing import ClassVar
+
+from coretrace_python.plugins import ModelPlugin
+from coretrace_python.semantic.symbols import SymbolId
+from coretrace_python.taint import Model, Sink, Source, TaintKind
+
+_METHODS = ("get", "post", "put", "patch", "delete", "head", "options", "request")
+
+_CALLERS = (
+    "requests",
+    "requests.Session",
+    "requests.api",
+    "httpx",
+    "httpx.Client",
+    "httpx.AsyncClient",
+)
+
+_REQUEST_FUNCTIONS = (
+    *(f"{caller}.{method}" for caller in _CALLERS for method in _METHODS),
+    "requests.Session.send",
+    "httpx.stream",
+    "httpx.Client.stream",
+    "httpx.AsyncClient.stream",
+    "httpx.Client.send",
+    "httpx.AsyncClient.send",
+)
+
+
+def _sym(path: str) -> SymbolId:
+    return SymbolId(f"python.{path}")
+
+
+class HttpClientModels(ModelPlugin):
+    name: ClassVar[str] = "http-client-models"
+    models: ClassVar[tuple[Model, ...]] = (
+        *(Sink(_sym(function), TaintKind.SSRF) for function in _REQUEST_FUNCTIONS),
+        *(Source(_sym(function), "http-response") for function in _REQUEST_FUNCTIONS),
+    )

plugins/models/http_clients/plugin.toml

@@ -0,0 +1,9 @@
+name = "http-client-models"
+version = "1.0.0"
+plugin_api = ">=1,<2"
+requires = []
+provides = ["model.ssrf-sinks", "model.http-response-sources"]
+
+[entrypoint]
+module = "http_client_models"
+class = "HttpClientModels"

src/coretrace_python/frontend/ast_adapter.py

@@ -322,7 +322,14 @@ def parameters(self, arguments: ast.arguments) -> tuple[nodes.Parameter, ...]:
 
     def parameter(self, argument: ast.arg, default: ast.expr | None, kind: str) -> nodes.Parameter:
         value = self.expression(default) if default is not None else None
-        return nodes.Parameter(argument.arg, self.span(argument), value, kind)
+        annotation: nodes.Expression | None = None
+        if argument.annotation is not None:
+            # Annotations never affect behaviour; one the HIR cannot represent is dropped.
+            try:
+                annotation = self.expression(argument.annotation)
+            except HIRBuildError:
+                annotation = None
+        return nodes.Parameter(argument.arg, self.span(argument), value, kind, annotation)
 
     def module(self, tree: ast.Module) -> nodes.Module:
         body = tuple(self.statement(statement) for statement in tree.body)

src/coretrace_python/hir/nodes.py

@@ -158,12 +158,14 @@ class Comprehension:
 
 @dataclass(frozen=True, slots=True)
 class Parameter:
-    """``kind`` is ``positional``, ``keyword``, ``var_positional`` or ``var_keyword``."""
+    """``kind`` is ``positional``, ``keyword``, ``var_positional`` or ``var_keyword``.
+    ``annotation`` is kept when it is an expression the HIR can represent."""
 
     name: str
     span: SourceSpan
     default: Expression | None = None
     kind: str = "positional"
+    annotation: Expression | None = None
 
 
 @dataclass(frozen=True, slots=True)

src/coretrace_python/taint/__init__.py

@@ -18,6 +18,7 @@
     Sink,
     Source,
     TaintKind,
+    TypedParameter,
 )
 
 __all__ = [
@@ -35,5 +36,6 @@
     "TaintFacts",
     "TaintFlow",
     "TaintKind",
+    "TypedParameter",
     "propagate_taint",
 ]

src/coretrace_python/taint/engine.py

@@ -112,25 +112,27 @@ def __init__(
         models: ModelTable,
         graph: CallGraph,
         summaries: SummaryTable,
-        entry: EntryPoint | None = None,
+        parameters: Mapping[int, Source] | None = None,
         project: SummaryIndex | None = None,
     ) -> None:
         self.name = name
         self.function = function
         self.models = models
         self.graph = graph
         self.summaries = summaries
-        self.entry = entry
+        self.parameters = parameters or {}
         self.project = project or SummaryIndex()
         self.blocks = {block.id: block for block in function.blocks}
         self.symbols = graph.symbols(name)
 
     def initial(self) -> State:
-        if self.entry is None:
-            return MappingProxyType({})
-        source = Source(self.entry.symbol, self.entry.label, self.entry.kinds)
-        seed = Taint(self.entry.kinds, frozenset({source}))
-        return MappingProxyType({p: seed for p in self.function.parameters})
+        return MappingProxyType(
+            {
+                self.function.parameters[index]: Taint(source.kinds, frozenset({source}))
+                for index, source in self.parameters.items()
+                if index < len(self.function.parameters)
+            }
+        )
 
     def join(self, a: State, b: State) -> State:
         merged = dict(a)
@@ -292,32 +294,78 @@ def report(
 
 
 def entry_point_of(
-    function: nodes.Function, models: ModelTable, scopes: ScopeTable, symbols: SymbolTable
+    function: nodes.Function,
+    models: ModelTable,
+    scopes: ScopeTable,
+    symbols: SymbolTable,
+    owner: nodes.Class | None = None,
 ) -> EntryPoint | None:
-    """The entry-point model matching one of the function's decorators, if any."""
+    """The entry-point model matching one of the function's decorators or, for a
+    method, one of the bases of ``owner``, if any."""
 
     scope = scopes.scope_for(function)
     enclosing = scope.parent if scope.parent is not None else scope.id
-    for decorator in function.decorators:
-        symbol = symbols.resolve_expression(enclosing, decorator)
+    candidates = list(function.decorators)
+    if owner is not None:
+        class_scope = scopes.scope_for(owner)
+        outside = class_scope.parent if class_scope.parent is not None else class_scope.id
+        candidates.extend(owner.bases)
+        enclosing_of = {id(base): outside for base in owner.bases}
+    else:
+        enclosing_of = {}
+    for expression in candidates:
+        symbol = symbols.resolve_expression(enclosing_of.get(id(expression), enclosing), expression)
         if symbol is not None:
             entry = models.entry_point(symbol)
             if entry is not None:
                 return entry
     return None
 
 
+def parameter_sources(
+    function: nodes.Function,
+    module: nodes.Module,
+    models: ModelTable,
+    scopes: ScopeTable,
+    symbols: SymbolTable,
+) -> 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."""
+
+    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)
+    if entry is not None:
+        first = 1 if owner is not None else 0
+        for index in range(first, len(function.parameters)):
+            sources[index] = Source(entry.symbol, entry.label, entry.kinds)
+    scope = scopes.scope_for(function)
+    enclosing = scope.parent if scope.parent is not None else scope.id
+    for index, parameter in enumerate(function.parameters):
+        if parameter.annotation is None:
+            continue
+        symbol = symbols.resolve_expression(enclosing, parameter.annotation)
+        typed = models.typed_parameter(symbol) if symbol is not None else None
+        if typed is not None:
+            sources[index] = Source(typed.symbol, typed.label, typed.kinds)
+    return sources
+
+
 def propagate_taint(
     name: str,
     function: FunctionIR,
     cfg: CFG,
     models: ModelTable,
     graph: CallGraph,
     summaries: SummaryTable,
-    entry: EntryPoint | None = None,
+    parameters: Mapping[int, Source] | None = None,
     project: SummaryIndex | None = None,
 ) -> TaintFacts:
-    problem = _TaintProblem(name, function, models, graph, summaries, entry, project)
+    problem = _TaintProblem(name, function, models, graph, summaries, parameters, project)
     solution = solve(problem, cfg)
     taints: dict[Value, Taint] = {}
     flows: list[TaintFlow] = []
@@ -357,6 +405,8 @@ def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> TaintFacts:
             models,
             graph,
             ctx.get(SummaryAnalysis),
-            entry_point_of(function, models, ctx.get(ScopeAnalysis), ctx.get(SymbolAnalysis)),
+            parameter_sources(
+                function, ctx.module, models, ctx.get(ScopeAnalysis), ctx.get(SymbolAnalysis)
+            ),
             ctx.get(ProjectSummaries),
         )