From 81f81b940f6cc925a990f4a021d4de1be586fc2b Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 4 Sep 2026 16:00:43 +0200 Subject: [PATCH 1/2] test: define URL-registered views and ORM suffix sinks --- tests/test_django_routes.py | 231 ++++++++++++++++++++++++++++++++++++ 1 file changed, 231 insertions(+) create mode 100644 tests/test_django_routes.py diff --git a/tests/test_django_routes.py b/tests/test_django_routes.py new file mode 100644 index 0000000..82e6ec1 --- /dev/null +++ b/tests/test_django_routes.py @@ -0,0 +1,231 @@ +"""Acceptance tests for URL-registered views and ORM suffix sinks, motivated by +``tests-project/DjangoGoat`` (branch ``broken``). + +Django views are plain functions registered elsewhere: ``path('login/', views.log_in)`` +in ``urls.py``. A ``RouteRegistrar`` model names such a registering call and the argument +that references the view; the engine scans every module of the project for +registrations before analysing, so the registered functions, and the methods of a class +registered through ``as_view()`` or a REST framework router, are entry points wherever +they are defined. A ``SuffixSink`` matches a call by the tail of its symbol, so +``User.objects.raw(query)`` is a SQL sink whatever the model class. + +Expected to remain red until ``RouteRegistrar``, ``SuffixSink``, ``taint.routes`` and +the Django, Flask and FastAPI registrar models exist. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.cache import ProjectCache +from coretrace_python.findings import Finding +from coretrace_python.frontend import build_hir +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.source import SourceManager +from coretrace_python.taint import SecurityModelRegistry, Sink, TaintKind + +try: + from coretrace_python.taint import ( + RegisteredRoutes, + RouteRegistrar, + SuffixSink, + registered_routes, + ) +except ImportError as error: # pragma: no cover - red until routes land + MISSING: Exception | None = error +else: + MISSING = None + + +@pytest.fixture(autouse=True) +def require_routes() -> None: + if MISSING is not None: + pytest.fail(f"registered routes are not implemented yet: {MISSING}") + + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "plugins" + + +def project(root: Path, files: dict[str, str]) -> Path: + for relative, text in files.items(): + path = root / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + return root + + +def located(findings: tuple[Finding, ...]) -> list[tuple[str, str, int]]: + return sorted((Path(str(f.span.source_id)).name, f.rule_id, f.span.start_line) for f in findings) + + +def check(text: str, name: str = "views.py") -> tuple[Finding, ...]: + return engine.check(SourceManager().add_source(name, text), [PLUGINS]) + + +DJANGO_VIEWS = ( + "import os\nfrom django.views import View\n\n" + "def log_in(request):\n os.system(request.POST['cmd'])\n\n" + "def helper(request):\n os.system(request.POST['cmd'])\n\n" + "class NoteView(View):\n def get(self, request, pk):\n os.system(pk)\n" +) +DJANGO_URLS = ( + "from django.urls import path\nfrom app import views\n\n" + "urlpatterns = [\n" + " path('login/', views.log_in, name='login'),\n" + " path('note/', views.NoteView.as_view(), name='note'),\n" + "]\n" +) + + +# --------------------------------------------------------------------------- models + + +def test_registrars_and_suffix_sinks_are_indexed() -> None: + registrar = RouteRegistrar(SymbolId("python.django.urls.path"), 1, "http") + suffix = SuffixSink("objects.raw", TaintKind.SQL, ((TaintKind.SQL, (0,)),)) + registry = SecurityModelRegistry() + registry.register(registrar, suffix, Sink(SymbolId("python.os.system"), TaintKind.COMMAND)) + table = registry.freeze() + + assert table.route_registrar(SymbolId("python.django.urls.path")) == registrar + assert table.route_registrars == (registrar,) + found = table.sink(SymbolId("python.django.contrib.auth.models.User.objects.raw")) + assert found is not None and found.kinds == TaintKind.SQL and found.kinds_at(1) == TaintKind.NONE + assert table.sink(SymbolId("python.os.raw")) is None + assert table.sink(SymbolId("python.os.system")) is not None + extended = table.extended(Sink(SymbolId("python.x"), TaintKind.ADVISORY)) + assert extended.route_registrars == (registrar,) and extended.suffix_sinks == (suffix,) + assert RegisteredRoutes.name == "taint.routes" + + +def test_registrations_are_collected_from_a_module() -> None: + from coretrace_python.semantic.scopes import ScopeAnalysis + from coretrace_python.semantic.symbols import SymbolAnalysis + + module = build_hir(SourceManager().add_source("urls.py", DJANGO_URLS, module_name="app.urls")) + manager = engine.build_manager(module) + registry = SecurityModelRegistry() + registry.register(RouteRegistrar(SymbolId("python.django.urls.path"), 1, "http")) + + routes = registered_routes(module, manager.get(ScopeAnalysis), manager.get(SymbolAnalysis), registry.freeze()) + + assert {str(s): (e.label, str(e.symbol)) for s, e in routes.items()} == { + "python.app.views.log_in": ("http", "python.django.urls.path"), + "python.app.views.NoteView": ("http", "python.django.urls.path"), + } + + +# --------------------------------------------------------------------------- registered views + + +def test_registered_django_views_and_class_based_views_are_entry_points(tmp_path: Path) -> None: + root = project(tmp_path, {"app/__init__.py": "", "app/views.py": DJANGO_VIEWS, "app/urls.py": DJANGO_URLS}) + + findings = engine.analyze_project(root, [PLUGINS]).findings + + assert located(findings) == [("views.py", "command-injection", 5), ("views.py", "command-injection", 12)] + assert {f.metadata["source_label"] for f in findings} == {"http"} + + +def test_registrations_in_the_same_module_and_aliased_imports_work(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "authentication/__init__.py": "", + "authentication/views.py": "import os\n\ndef log_in(request):\n os.system(request.POST['cmd'])\n", + "site/__init__.py": "", + "site/urls.py": "from django.urls import path, re_path\nfrom authentication import views as auth_views\n\n" + "urlpatterns = [path('login/', auth_views.log_in, name='login')]\n", + "local.py": "import os\nfrom django.urls import path\n\ndef ping(request):\n os.system(request.GET['h'])\n\n" + "urlpatterns = [path('ping/', ping)]\n", + }, + ) + assert located(engine.analyze_project(root, [PLUGINS]).findings) == [ + ("local.py", "command-injection", 5), + ("views.py", "command-injection", 4), + ] + + +def test_rest_framework_routers_register_viewsets(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "api/__init__.py": "", + "api/views.py": "import os\nfrom rest_framework import viewsets\n\n" + "class Ping(viewsets.ViewSet):\n def list(self, request):\n os.system(request.query_params['h'])\n", + "api/urls.py": "from rest_framework.routers import DefaultRouter\nfrom api.views import Ping\n\n" + "router = DefaultRouter()\nrouter.register(r'ping', Ping, basename='ping')\nurlpatterns = router.urls\n", + }, + ) + assert located(engine.analyze_project(root, [PLUGINS]).findings) == [("views.py", "command-injection", 6)] + + +def test_flask_and_fastapi_programmatic_registration(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "flask_app.py": "import os\nfrom flask import Flask, request\n\napp = Flask(__name__)\n\n" + "def run():\n os.system(request.args['c'])\n\n" + "app.add_url_rule('/run', view_func=run)\n", + "fast_app.py": "import os\nfrom fastapi import FastAPI\n\napp = FastAPI()\n\n" + "def ping(host: str):\n os.system(host)\n\n" + "app.add_api_route('/ping', ping)\n", + }, + ) + assert located(engine.analyze_project(root, [PLUGINS]).findings) == [ + ("fast_app.py", "command-injection", 7), + ("flask_app.py", "command-injection", 7), + ] + + +def test_unregistered_functions_stay_clean() -> None: + assert check("import os\n\ndef helper(request):\n os.system(request.POST['cmd'])\n") == () + + +# --------------------------------------------------------------------------- suffix sinks + + +def test_model_raw_queries_are_sql_sinks(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "app/__init__.py": "", + "app/models.py": "from django.db import models\n\nclass Note(models.Model):\n body = models.TextField()\n", + "app/views.py": "from django.contrib.auth.models import User\nfrom app.models import Note\n\n" + "def log_in(request):\n" + " username = request.POST['username']\n" + " query = \"SELECT * FROM auth_user WHERE username = '%s'\" % (username,)\n" + " user = User.objects.raw(query)[0]\n" + " notes = Note.objects.raw('SELECT * FROM app_note WHERE body = %s', [username])\n" + " unsafe = Note.objects.raw('SELECT * FROM app_note WHERE body = ' + username)\n" + " return user, notes, unsafe\n", + "app/urls.py": "from django.urls import path\nfrom app import views\n\nurlpatterns = [path('login/', views.log_in)]\n", + }, + ) + assert located(engine.analyze_project(root, [PLUGINS]).findings) == [ + ("views.py", "sql-injection", 7), + ("views.py", "sql-injection", 9), + ] + + +# --------------------------------------------------------------------------- engine plumbing + + +def test_routes_are_part_of_the_cache_key_and_reach_workers(tmp_path: Path) -> None: + root = project(tmp_path, {"app/__init__.py": "", "app/views.py": DJANGO_VIEWS, "app/urls.py": "urlpatterns = []\n"}) + cache = ProjectCache(tmp_path / "cache") + before = engine.analyze_project(root, [PLUGINS], cache=cache) + (root / "app" / "urls.py").write_text(DJANGO_URLS, encoding="utf-8") + after = engine.analyze_project(root, [PLUGINS], cache=cache) + parallel = engine.analyze_project(root, [PLUGINS], jobs=2) + + # The class-based view derives from ``View``, an entry point in itself; the function + # view only becomes one once ``urls.py`` registers it. + assert located(before.findings) == [("views.py", "command-injection", 12)] + assert before.keys["app.views"] != after.keys["app.views"] + assert located(after.findings) == [("views.py", "command-injection", 5), ("views.py", "command-injection", 12)] + assert parallel.findings == after.findings From 38125f6d573ebc041929ebf9d412ab3790832410 Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 4 Sep 2026 16:00:43 +0200 Subject: [PATCH 2/2] feat: register views from URL configurations and match ORM sinks by suffix --- README.md | 8 +- plugins/models/django/django_models.py | 17 ++++ plugins/models/fastapi/fastapi_models.py | 4 +- plugins/models/flask/flask_models.py | 3 + src/coretrace_python/engine.py | 49 +++++++++++- src/coretrace_python/taint/__init__.py | 8 ++ src/coretrace_python/taint/engine.py | 14 +++- src/coretrace_python/taint/models.py | 54 ++++++++++++- src/coretrace_python/taint/routes.py | 99 ++++++++++++++++++++++++ 9 files changed, 244 insertions(+), 12 deletions(-) create mode 100644 src/coretrace_python/taint/routes.py diff --git a/README.md b/README.md index 4de12af..3c72534 100644 --- a/README.md +++ b/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 diff --git a/plugins/models/django/django_models.py b/plugins/models/django/django_models.py index b0d633c..b70c70e 100644 --- a/plugins/models/django/django_models.py +++ b/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), ) diff --git a/plugins/models/fastapi/fastapi_models.py b/plugins/models/fastapi/fastapi_models.py index bfec252..8d2b244 100644 --- a/plugins/models/fastapi/fastapi_models.py +++ b/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), diff --git a/plugins/models/flask/flask_models.py b/plugins/models/flask/flask_models.py index e954cf6..8073604 100644 --- a/plugins/models/flask/flask_models.py +++ b/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), diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index d3b41a7..0593a23 100644 --- a/src/coretrace_python/engine.py +++ b/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 {})), ) diff --git a/src/coretrace_python/taint/__init__.py b/src/coretrace_python/taint/__init__.py index 9970527..f0898bd 100644 --- a/src/coretrace_python/taint/__init__.py +++ b/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", ] diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index 81099bb..f8faf8a 100644 --- a/src/coretrace_python/taint/engine.py +++ b/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,10 +529,11 @@ 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)), @@ -538,6 +541,11 @@ def parameter_sources( ) 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), diff --git a/src/coretrace_python/taint/models.py b/src/coretrace_python/taint/models.py index f9892e1..1c212bd 100644 --- a/src/coretrace_python/taint/models.py +++ b/src/coretrace_python/taint/models.py @@ -115,6 +115,31 @@ def matches(self, name: str) -> bool: return re.search(self.pattern, name) is not None +@dataclass(frozen=True) +class RouteRegistrar: + """A call registering a handler elsewhere (``path('login/', views.log_in)``): the + function or class referenced by ``argument`` (or ``keyword``) is an entry point.""" + + symbol: SymbolId + argument: int + label: str + kinds: TaintKind = TaintKind.ALL + keyword: str | None = None + + +@dataclass(frozen=True) +class SuffixSink: + """A sink matched by the tail of a call's symbol (``objects.raw`` for any model).""" + + suffix: str + kinds: TaintKind + positions: tuple[tuple[TaintKind, tuple[int, ...]], ...] = () + + @property + def symbol(self) -> SymbolId: + return SymbolId(f"python.suffix.{self.suffix.replace('.', '_')}") + + @dataclass(frozen=True) class Validator: """A callable whose truth proves its ``argument`` safe (refutation evidence, §24).""" @@ -134,7 +159,16 @@ class AuthorizationGuard: Model = ( - Source | Sink | Sanitizer | EntryPoint | TypedParameter | Validator | AuthorizationGuard | NamedParameter + Source + | Sink + | Sanitizer + | EntryPoint + | TypedParameter + | Validator + | AuthorizationGuard + | NamedParameter + | RouteRegistrar + | SuffixSink ) @@ -148,6 +182,8 @@ class ModelTable: validators: tuple[Validator, ...] = () authorizations: tuple[AuthorizationGuard, ...] = () named_parameters: tuple[NamedParameter, ...] = () + route_registrars: tuple[RouteRegistrar, ...] = () + suffix_sinks: tuple[SuffixSink, ...] = () _by_symbol: dict[type[Model], dict[SymbolId, Model]] = field( init=False, repr=False, compare=False ) @@ -161,6 +197,7 @@ def __post_init__(self) -> None: TypedParameter: {m.symbol: m for m in self.typed_parameters}, Validator: {m.symbol: m for m in self.validators}, AuthorizationGuard: {m.symbol: m for m in self.authorizations}, + RouteRegistrar: {m.symbol: m for m in self.route_registrars}, } object.__setattr__(self, "_by_symbol", MappingProxyType(index)) @@ -197,7 +234,16 @@ def source_covering(self, symbol: SymbolId) -> Source | None: def sink(self, symbol: SymbolId) -> Sink | None: found = self._by_symbol[Sink].get(symbol) - return found if isinstance(found, Sink) else None + if isinstance(found, Sink): + return found + for suffix in self.suffix_sinks: + if symbol.canonical_name.endswith(f".{suffix.suffix}"): + return Sink(symbol, suffix.kinds, suffix.positions) + return None + + def route_registrar(self, symbol: SymbolId) -> RouteRegistrar | None: + found = self._by_symbol[RouteRegistrar].get(symbol) + return found if isinstance(found, RouteRegistrar) else None def extended(self, *sinks: Sink) -> ModelTable: """A table with extra sinks; a sink already present gains the new kinds.""" @@ -219,6 +265,8 @@ def extended(self, *sinks: Sink) -> ModelTable: self.validators, self.authorizations, self.named_parameters, + self.route_registrars, + self.suffix_sinks, ) def sanitizer(self, symbol: SymbolId) -> Sanitizer | None: @@ -252,6 +300,8 @@ def freeze(self) -> ModelTable: validators=tuple(m for m in models if isinstance(m, Validator)), authorizations=tuple(m for m in models if isinstance(m, AuthorizationGuard)), named_parameters=tuple(m for m in models if isinstance(m, NamedParameter)), + route_registrars=tuple(m for m in models if isinstance(m, RouteRegistrar)), + suffix_sinks=tuple(m for m in models if isinstance(m, SuffixSink)), ) diff --git a/src/coretrace_python/taint/routes.py b/src/coretrace_python/taint/routes.py new file mode 100644 index 0000000..e85b167 --- /dev/null +++ b/src/coretrace_python/taint/routes.py @@ -0,0 +1,99 @@ +"""Entry points registered away from their definition (architecture §16, §25). + +Django views are plain functions listed in ``urls.py``; Flask and FastAPI applications +can register handlers programmatically too. A ``RouteRegistrar`` model names the +registering call and the argument that references the handler. The engine scans every +module of a project for such calls before analysing and provides the result as the +``taint.routes`` input, so a registered function, or the methods of a registered class, +receive attacker-controlled parameters wherever they are defined. +""" + +from __future__ import annotations + +from collections.abc import Iterator, Mapping +from typing import ClassVar + +from coretrace_python.analysis import Analysis, AnalysisContext +from coretrace_python.hir import nodes +from coretrace_python.hir.visitors import Node, children +from coretrace_python.interprocedural import project_symbol +from coretrace_python.semantic.scopes import ScopeTable +from coretrace_python.semantic.symbols import SymbolId, SymbolTable +from coretrace_python.taint.models import EntryPoint, ModelTable + +Routes = Mapping[SymbolId, EntryPoint] + + +class RegisteredRoutes(Analysis[Routes]): + """Project symbols registered as handlers, provided by the engine; empty on its own.""" + + name: ClassVar[str] = "taint.routes" + + @classmethod + def compute(cls, ctx: AnalysisContext) -> Routes: + return {} + + +def registered_routes( + module: nodes.Module, scopes: ScopeTable, symbols: SymbolTable, models: ModelTable +) -> dict[SymbolId, EntryPoint]: + """The handlers this module registers, as project symbols, with the entry point the + registrar grants them.""" + + scope = scopes.module_scope.id + defined = { + s.name for s in module.body if isinstance(s, nodes.Function | nodes.Class) + } + found: dict[SymbolId, EntryPoint] = {} + for call in _calls(module.body): + callee = symbols.resolve_expression(scope, call.callee) + registrar = models.route_registrar(callee) if callee is not None else None + if registrar is None: + continue + handler: nodes.Expression | None = None + if registrar.argument < len(call.arguments): + handler = call.arguments[registrar.argument] + elif registrar.keyword is not None: + handler = next((k.value for k in call.keywords if k.name == registrar.keyword), None) + if handler is None: + continue + target = _handler_symbol(handler, module, symbols, scope, defined) + if target is not None: + found.setdefault(target, EntryPoint(registrar.symbol, registrar.label, registrar.kinds)) + return found + + +def _handler_symbol( + handler: nodes.Expression, + module: nodes.Module, + symbols: SymbolTable, + scope: object, + defined: set[str], +) -> SymbolId | None: + if isinstance(handler, nodes.Call): + # ``NoteView.as_view()``: the class is the handler. + handler = handler.callee + if isinstance(handler, nodes.Attribute) and handler.name == "as_view": + handler = handler.value + symbol = symbols.resolve_expression(scope, handler) # type: ignore[arg-type] + if symbol is not None: + return symbol + if isinstance(handler, nodes.Name) and handler.identifier in defined: + return project_symbol(module.name, handler.identifier) + return None + + +def _calls(body: tuple[nodes.Statement, ...]) -> Iterator[nodes.Call]: + for statement in body: + if isinstance(statement, nodes.Function | nodes.Class): + continue + yield from _calls_in(statement) + + +def _calls_in(node: Node) -> Iterator[nodes.Call]: + if isinstance(node, nodes.Function | nodes.Class | nodes.Lambda | nodes.Comprehension): + return + if isinstance(node, nodes.Call): + yield node + for child in children(node): + yield from _calls_in(child)