diff --git a/README.md b/README.md index 9b02a58..3a2a598 100644 --- a/README.md +++ b/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 diff --git a/plugins/models/django/django_models.py b/plugins/models/django/django_models.py new file mode 100644 index 0000000..5284883 --- /dev/null +++ b/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), + ) diff --git a/plugins/models/django/plugin.toml b/plugins/models/django/plugin.toml new file mode 100644 index 0000000..98d1dad --- /dev/null +++ b/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" diff --git a/plugins/models/http_clients/http_client_models.py b/plugins/models/http_clients/http_client_models.py new file mode 100644 index 0000000..f66f168 --- /dev/null +++ b/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), + ) diff --git a/plugins/models/http_clients/plugin.toml b/plugins/models/http_clients/plugin.toml new file mode 100644 index 0000000..67e7b4d --- /dev/null +++ b/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" diff --git a/src/coretrace_python/frontend/ast_adapter.py b/src/coretrace_python/frontend/ast_adapter.py index 6a9e08b..8a5479a 100644 --- a/src/coretrace_python/frontend/ast_adapter.py +++ b/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) diff --git a/src/coretrace_python/hir/nodes.py b/src/coretrace_python/hir/nodes.py index 9c7911c..f1aac2b 100644 --- a/src/coretrace_python/hir/nodes.py +++ b/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) diff --git a/src/coretrace_python/taint/__init__.py b/src/coretrace_python/taint/__init__.py index 8428d1a..23242a9 100644 --- a/src/coretrace_python/taint/__init__.py +++ b/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", ] diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index 4bb6fbe..b00dacb 100644 --- a/src/coretrace_python/taint/engine.py +++ b/src/coretrace_python/taint/engine.py @@ -112,7 +112,7 @@ 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 @@ -120,17 +120,19 @@ def __init__( 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,14 +294,27 @@ 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: @@ -307,6 +322,39 @@ def entry_point_of( 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, @@ -314,10 +362,10 @@ def propagate_taint( 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), ) diff --git a/src/coretrace_python/taint/models.py b/src/coretrace_python/taint/models.py index 41a9d79..538c118 100644 --- a/src/coretrace_python/taint/models.py +++ b/src/coretrace_python/taint/models.py @@ -60,14 +60,24 @@ class Sanitizer: @dataclass(frozen=True) class EntryPoint: - """Functions decorated by ``symbol`` receive attacker-controlled parameters.""" + """Functions decorated by ``symbol``, and methods of classes deriving from it, + receive attacker-controlled parameters.""" symbol: SymbolId label: str kinds: TaintKind = TaintKind.ALL -Model = Source | Sink | Sanitizer | EntryPoint +@dataclass(frozen=True) +class TypedParameter: + """A parameter annotated with ``symbol`` is attacker-controlled.""" + + symbol: SymbolId + label: str + kinds: TaintKind = TaintKind.ALL + + +Model = Source | Sink | Sanitizer | EntryPoint | TypedParameter @dataclass(frozen=True) @@ -76,6 +86,7 @@ class ModelTable: sinks: tuple[Sink, ...] sanitizers: tuple[Sanitizer, ...] entry_points: tuple[EntryPoint, ...] = () + typed_parameters: tuple[TypedParameter, ...] = () _by_symbol: dict[type[Model], dict[SymbolId, Model]] = field( init=False, repr=False, compare=False ) @@ -86,6 +97,7 @@ def __post_init__(self) -> None: Sink: {m.symbol: m for m in self.sinks}, Sanitizer: {m.symbol: m for m in self.sanitizers}, EntryPoint: {m.symbol: m for m in self.entry_points}, + TypedParameter: {m.symbol: m for m in self.typed_parameters}, } object.__setattr__(self, "_by_symbol", MappingProxyType(index)) @@ -93,6 +105,10 @@ def entry_point(self, symbol: SymbolId) -> EntryPoint | None: found = self._by_symbol[EntryPoint].get(symbol) return found if isinstance(found, EntryPoint) else None + def typed_parameter(self, symbol: SymbolId) -> TypedParameter | None: + found = self._by_symbol[TypedParameter].get(symbol) + return found if isinstance(found, TypedParameter) else None + def source(self, symbol: SymbolId) -> Source | None: found = self._by_symbol[Source].get(symbol) return found if isinstance(found, Source) else None @@ -121,7 +137,13 @@ def extended(self, *sinks: Sink) -> ModelTable: merged[sink.symbol] = ( Sink(sink.symbol, current.kinds | sink.kinds) if current is not None else sink ) - return ModelTable(self.sources, tuple(merged.values()), self.sanitizers, self.entry_points) + return ModelTable( + self.sources, + tuple(merged.values()), + self.sanitizers, + self.entry_points, + self.typed_parameters, + ) def sanitizer(self, symbol: SymbolId) -> Sanitizer | None: found = self._by_symbol[Sanitizer].get(symbol) @@ -150,6 +172,7 @@ def freeze(self) -> ModelTable: sinks=tuple(m for m in models if isinstance(m, Sink)), sanitizers=tuple(m for m in models if isinstance(m, Sanitizer)), entry_points=tuple(m for m in models if isinstance(m, EntryPoint)), + typed_parameters=tuple(m for m in models if isinstance(m, TypedParameter)), ) diff --git a/tests/test_detectors.py b/tests/test_detectors.py index 9995f1f..3f7bc45 100644 --- a/tests/test_detectors.py +++ b/tests/test_detectors.py @@ -164,8 +164,10 @@ def test_shipped_plugins_load_with_their_manifests() -> None: assert set(by_name) == { "command-injection", "dangerous-eval", + "django-models", "fastapi-models", "flask-models", + "http-client-models", "path-traversal", "python-stdlib-models", "reachable-vulnerability", diff --git a/tests/test_django_http_clients.py b/tests/test_django_http_clients.py new file mode 100644 index 0000000..c4c706f --- /dev/null +++ b/tests/test_django_http_clients.py @@ -0,0 +1,300 @@ +"""Acceptance tests for the Django and Requests/httpx models (``docs/architecture.md`` +§15, §25, §36; roadmap issue #30). + +Django views are undecorated functions taking a request, so two engine mechanisms are +added for them, both generic: + +- a ``TypedParameter`` model: a parameter annotated with the model's class symbol is + attacker-controlled, which is how ``request: HttpRequest`` becomes HTTP input; +- an ``EntryPoint`` also applies to the methods of a class deriving from its symbol, + which is how class-based views receive HTTP input; ``self`` stays clean. + +Requests and httpx are SSRF sinks whose results are ``http-response`` sources. + +Expected to remain red until ``Parameter.annotation``, ``TypedParameter``, class-based +entry points and the ``models/django`` and ``models/http_clients`` plugins exist. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.findings import Finding +from coretrace_python.frontend import build_hir +from coretrace_python.hir import nodes +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.source import SourceManager +from coretrace_python.taint import EntryPoint, SecurityModelRegistry, Sink, TaintAnalysis, TaintKind + +try: + from coretrace_python.taint import TypedParameter +except ImportError as error: # pragma: no cover - red until typed parameters land + MISSING = error +else: + MISSING = None + + +@pytest.fixture(autouse=True) +def require_models() -> None: + if MISSING is not None: + pytest.fail(f"Django and HTTP client models are not implemented yet: {MISSING}") + + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "plugins" + +OS_SYSTEM = Sink(SymbolId("python.os.system"), TaintKind.COMMAND) + + +def module_for(text: str) -> nodes.Module: + return build_hir(SourceManager().add_source("web.py", text)) + + +def flows(text: str, *models: object): # type: ignore[no-untyped-def] + registry = SecurityModelRegistry() + registry.register(*models) # type: ignore[arg-type] + manager = engine.build_manager(module_for(text), registry) + found = [] + for statement in manager.module.body: + functions = [statement] if isinstance(statement, nodes.Function) else [] + if isinstance(statement, nodes.Class): + functions = [s for s in statement.body if isinstance(s, nodes.Function)] + for function in functions: + found.extend(manager.get(TaintAnalysis, function).flows) + return tuple(found) + + +def check(text: str) -> tuple[Finding, ...]: + return engine.check(SourceManager().add_source("web.py", text), [PLUGINS]) + + +def rules(findings: tuple[Finding, ...]) -> list[str]: + return sorted(f.rule_id for f in findings) + + +# --------------------------------------------------------------------------- HIR + + +def test_parameter_annotations_are_kept() -> None: + module = module_for("from django.http import HttpRequest\n\ndef view(request: HttpRequest, n: int = 1, *rest):\n pass\n") + function = module.body[-1] + assert isinstance(function, nodes.Function) + + request, n, rest = function.parameters + assert isinstance(request.annotation, nodes.Name) and request.annotation.identifier == "HttpRequest" + assert isinstance(n.annotation, nodes.Name) and n.annotation.identifier == "int" + assert isinstance(n.default, nodes.Constant) and n.default.value == 1 + assert rest.annotation is None + + +def test_unsupported_annotations_do_not_break_the_build() -> None: + module = module_for("def f(x: (lambda: 1), y: 'Request'):\n return x\n") + function = module.body[0] + assert isinstance(function, nodes.Function) + assert function.parameters[0].annotation is None + + +# --------------------------------------------------------------------------- typed parameters + + +REQUEST = SymbolId("python.django.http.HttpRequest") + + +def test_typed_parameters_are_indexed_in_the_model_table() -> None: + typed = TypedParameter(REQUEST, "http") + registry = SecurityModelRegistry() + registry.register(typed, Sink(REQUEST, TaintKind.HTML)) + table = registry.freeze() + + assert table.typed_parameter(REQUEST) == typed + assert table.typed_parameter(SymbolId("python.other")) is None + assert table.typed_parameters == (typed,) + assert typed.kinds == TaintKind.ALL + assert table.extended(Sink(SymbolId("python.x"), TaintKind.ADVISORY)).typed_parameters == (typed,) + + +def test_a_parameter_annotated_with_a_typed_symbol_is_a_source() -> None: + found = flows( + "from django.http import HttpRequest\nimport os\n\n" + "def view(request: HttpRequest, name):\n" + " os.system(request.GET['cmd'])\n" + " os.system(name)\n", + TypedParameter(REQUEST, "http"), + OS_SYSTEM, + ) + + (flow,) = found + assert flow.source.symbol == REQUEST + assert flow.source.label == "http" + assert flow.location.start_line == 5 + + +def test_annotations_resolve_through_module_aliases() -> None: + found = flows( + "import django.http as http\nimport os\n\n" + "def view(request: http.HttpRequest):\n" + " os.system(request.body)\n", + TypedParameter(REQUEST, "http"), + OS_SYSTEM, + ) + assert len(found) == 1 + + +def test_untyped_parameters_stay_clean() -> None: + found = flows( + "import os\n\ndef view(request):\n os.system(request.GET['cmd'])\n", + TypedParameter(REQUEST, "http"), + OS_SYSTEM, + ) + assert found == () + + +# --------------------------------------------------------------------------- class-based views + + +VIEW = EntryPoint(SymbolId("python.django.views.View"), "http") + + +def test_methods_of_classes_deriving_from_an_entry_point_receive_input() -> None: + found = flows( + "from django.views import View\nimport os\n\n" + "class Run(View):\n" + " def get(self, request, cmd):\n" + " os.system(cmd)\n", + VIEW, + OS_SYSTEM, + ) + + (flow,) = found + assert flow.source.symbol == VIEW.symbol + assert flow.source.label == "http" + + +def test_self_is_not_input_in_class_based_views() -> None: + found = flows( + "from django.views import View\nimport os\n\n" + "class Run(View):\n" + " def get(self, request):\n" + " os.system(self.command)\n", + VIEW, + OS_SYSTEM, + ) + assert found == () + + +def test_classes_without_an_entry_point_base_stay_clean() -> None: + found = flows( + "import os\n\nclass Run:\n def get(self, request, cmd):\n os.system(cmd)\n", + VIEW, + OS_SYSTEM, + ) + assert found == () + + +# --------------------------------------------------------------------------- shipped plugins + + +def test_shipped_django_and_http_client_plugins_load() -> None: + from coretrace_python.plugins import discover_plugins + + loaded = {p.manifest.name: p.manifest for p in discover_plugins(PLUGINS, engine.build_manager(module_for("")))} + + assert loaded["django-models"].provides == ("model.http-sources", "model.django-views", "model.sql-sinks") + assert loaded["http-client-models"].provides == ("model.ssrf-sinks", "model.http-response-sources") + + +def test_django_decorated_view_with_raw_cursor_is_a_sql_injection() -> None: + findings = check( + "from django.db import connection\n" + "from django.views.decorators.csrf import csrf_exempt\n\n" + "@csrf_exempt\n" + "def search(request):\n" + " cursor = connection.cursor()\n" + " cursor.execute(\"SELECT * FROM t WHERE n = '\" + request.POST['n'] + \"'\")\n" + ) + assert rules(findings) == ["sql-injection"] + assert findings[0].span.start_line == 7 + + +def test_django_typed_view_with_mark_safe_is_an_xss_and_escape_refutes_it() -> None: + assert rules( + check( + "from django.http import HttpRequest, HttpResponse\n" + "from django.utils.safestring import mark_safe\n\n" + "def hello(request: HttpRequest):\n" + " return HttpResponse(mark_safe('' + request.GET['name'] + ''))\n" + ) + ) == ["xss", "xss"] + assert ( + check( + "from django.http import HttpRequest, HttpResponse\n" + "from django.utils.html import escape\n\n" + "def hello(request: HttpRequest):\n" + " return HttpResponse('' + escape(request.GET['name']) + '')\n" + ) + == () + ) + + +def test_django_class_based_view_parameter_to_os_system_is_a_command_injection() -> None: + findings = check( + "import os\nfrom django.views.generic import View\n\n" + "class Ping(View):\n" + " def get(self, request, host):\n" + " os.system('ping ' + host)\n" + ) + assert rules(findings) == ["command-injection"] + assert findings[0].function == "get" + assert findings[0].span.start_line == 6 + + +def test_django_rest_framework_api_view_is_an_entry_point() -> None: + findings = check( + "import os\nfrom rest_framework.decorators import api_view\n\n" + "@api_view(['GET'])\n" + "def ping(request):\n" + " os.system(request.query_params['host'])\n" + ) + assert rules(findings) == ["command-injection"] + + +def test_requests_with_user_url_is_an_ssrf_and_its_response_is_input() -> None: + findings = check( + "import os\nimport requests\nfrom flask import Flask, request\n\napp = Flask(__name__)\n\n" + "@app.route('/fetch')\n" + "def fetch():\n" + " response = requests.get(request.args['url'])\n" + " os.system(response.text)\n" + ) + # The response carries two provenances: the remote server's answer, and the URL the + # request was built from, which the engine propagates through the call. + assert rules(findings) == ["command-injection", "command-injection", "ssrf"] + commands = [f for f in findings if f.rule_id == "command-injection"] + assert {f.span.start_line for f in commands} == {10} + assert sorted(f.metadata["source_label"] for f in commands) == ["http", "http-response"] + + +def test_httpx_client_instances_and_sessions_are_ssrf_sinks() -> None: + findings = check( + "import httpx\nimport requests\n\n" + "client = httpx.Client()\nsession = requests.Session()\n\n" + "def fetch():\n" + " client.post(input())\n" + " session.get(input())\n" + " httpx.get('https://example.com')\n" + ) + assert rules(findings) == ["ssrf", "ssrf"] + assert sorted(f.span.start_line for f in findings) == [8, 9] + + +def test_detectors_stay_generic() -> None: + assert sorted(p.name for p in (PLUGINS / "security").iterdir() if p.is_dir()) == [ + "command_injection", + "path_traversal", + "sql_injection", + "ssrf", + "xss", + ]