diff --git a/docs/usage.md b/docs/usage.md index 12255d5..b46a7bd 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -79,8 +79,9 @@ report's per-file detail tell "no findings" from "nothing analysed". Taint follows values between functions and across files through function summaries, into objects (containers, attributes, instances of the project's own classes) and through -closures. Flask, FastAPI, Django and aiohttp route handlers, class-based views and registered -URL patterns receive HTTP input; click and Typer commands receive `argv` input; `input()`, +closures. Flask, FastAPI, Django, aiohttp, Tornado and Bottle route handlers, class-based views +and registered URL patterns receive HTTP input, and a method a handler class inherits +from its framework base, `self.get_argument` or `self.request`, denotes the base's; click and Typer commands receive `argv` input; `input()`, `sys.argv`, environment variables, the output of local processes and the responses of HTTP clients are further sources. The first three are operator-controlled: a command-line tool opens the paths it is given, a command or a path built from the environment is the @@ -309,9 +310,9 @@ real 45 000-line project is exercised on every change. ## Plugins -The package ships 28 plugins, loaded by default. Security models for the standard library +The package ships 30 plugins, loaded by default. Security models for the standard library and the supported frameworks: `python-stdlib-models`, `flask-models`, `fastapi-models`, -`django-models`, `aiohttp-models`, `sqlalchemy-models`, `db-driver-models` (aiopg, +`django-models`, `aiohttp-models`, `tornado-models`, `bottle-models`, `sqlalchemy-models`, `db-driver-models` (aiopg, asyncpg, psycopg2, PyMySQL), `http-client-models`, `credential-models` and `cli-models`. Taint detectors: `command-injection`, `sql-injection`, `path-traversal`, `ssrf`, `xss`, `insecure-deserialization`, `open-redirect` and `plaintext-credentials`. Dangerous API detectors: `dangerous-eval`, `weak-crypto`, `flask-debug` and diff --git a/src/coretrace_python/bundled/models/bottle/bottle_models.py b/src/coretrace_python/bundled/models/bottle/bottle_models.py new file mode 100644 index 0000000..2936074 --- /dev/null +++ b/src/coretrace_python/bundled/models/bottle/bottle_models.py @@ -0,0 +1,39 @@ +"""Bottle security models: ``@route``, ``@get`` and the methods of a ``Bottle`` application +are entry points, ``bottle.request`` is the HTTP source, ``redirect``, ``static_file``, +``template`` and ``HTTPResponse`` are sinks, ``html_escape`` a sanitizer.""" + +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, Source, TaintKind + +_METHODS = ("route", "get", "post", "put", "delete", "patch") +_REQUEST_ATTRIBUTES = ( + "query", "forms", "params", "json", "body", "headers", "cookies", "GET", "POST", "files", + "url", "path", "fullpath", "query_string", "get_cookie", "get_header", +) +_FIRST_ONLY = lambda kind: ((kind, (0,)),) + + +def _sym(path: str) -> SymbolId: + return SymbolId(f"python.{path}") + + +class BottleModels(ModelPlugin): + name: ClassVar[str] = "bottle-models" + models: ClassVar[tuple[Model, ...]] = ( + *(EntryPoint(_sym(f"bottle.{method}"), "http") for method in _METHODS), + *(EntryPoint(_sym(f"bottle.Bottle.{method}"), "http") for method in _METHODS), + *(Source(_sym(f"bottle.request.{attribute}"), "http") for attribute in _REQUEST_ATTRIBUTES), + Sink(_sym("bottle.redirect"), TaintKind.REDIRECT, _FIRST_ONLY(TaintKind.REDIRECT)), + # ``static_file(filename, root)``: the root is the application's own. + Sink(_sym("bottle.static_file"), TaintKind.PATH, _FIRST_ONLY(TaintKind.PATH)), + # ``template(source_or_name, **variables)``: a tainted template is injection, the + # variables are escaped by the engine. + Sink(_sym("bottle.template"), TaintKind.HTML, _FIRST_ONLY(TaintKind.HTML)), + Sink(_sym("bottle.HTTPResponse"), TaintKind.HTML), + Sanitizer(_sym("bottle.html_escape"), TaintKind.HTML), + ) diff --git a/src/coretrace_python/bundled/models/bottle/plugin.toml b/src/coretrace_python/bundled/models/bottle/plugin.toml new file mode 100644 index 0000000..5220b67 --- /dev/null +++ b/src/coretrace_python/bundled/models/bottle/plugin.toml @@ -0,0 +1,9 @@ +name = "bottle-models" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = [] +provides = ["model.http-sources", "model.bottle-routes"] + +[entrypoint] +module = "bottle_models" +class = "BottleModels" diff --git a/src/coretrace_python/bundled/models/django/django_models.py b/src/coretrace_python/bundled/models/django/django_models.py index e0a55cc..060b346 100644 --- a/src/coretrace_python/bundled/models/django/django_models.py +++ b/src/coretrace_python/bundled/models/django/django_models.py @@ -20,6 +20,7 @@ RouteRegistrar, Sanitizer, Sink, + Source, SuffixSink, TaintKind, TypedParameter, @@ -98,6 +99,8 @@ class DjangoModels(ModelPlugin): models: ClassVar[tuple[Model, ...]] = ( *(TypedParameter(_sym(cls), "http") for cls in _REQUEST_CLASSES), *(EntryPoint(_sym(base), "http") for base in _VIEW_BASES), + # ``self.request`` in a class-based view: the attribute is inherited from the base. + *(Source(_sym(f"{base}.request"), "http") for base in _VIEW_BASES), *(EntryPoint(_sym(decorator), "http") for decorator in _VIEW_DECORATORS), Sink(_sym("django.db.connection.cursor.execute"), TaintKind.SQL | TaintKind.CREDENTIAL, ((TaintKind.SQL, (0,)),)), Sink(_sym("django.db.connection.cursor.executemany"), TaintKind.SQL | TaintKind.CREDENTIAL, ((TaintKind.SQL, (0,)),)), diff --git a/src/coretrace_python/bundled/models/tornado/plugin.toml b/src/coretrace_python/bundled/models/tornado/plugin.toml new file mode 100644 index 0000000..1b21444 --- /dev/null +++ b/src/coretrace_python/bundled/models/tornado/plugin.toml @@ -0,0 +1,9 @@ +name = "tornado-models" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = [] +provides = ["model.http-sources", "model.tornado-handlers", "model.ssrf-sinks"] + +[entrypoint] +module = "tornado_models" +class = "TornadoModels" diff --git a/src/coretrace_python/bundled/models/tornado/tornado_models.py b/src/coretrace_python/bundled/models/tornado/tornado_models.py new file mode 100644 index 0000000..6960d14 --- /dev/null +++ b/src/coretrace_python/bundled/models/tornado/tornado_models.py @@ -0,0 +1,37 @@ +"""Tornado security models: ``RequestHandler`` subclasses are entry points whose inherited +request methods are HTTP sources and whose ``write``, ``finish`` and ``redirect`` are +sinks; the HTTP clients are SSRF sinks with untrusted responses.""" + +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, Source, TaintKind + +_HANDLERS = ("tornado.web.RequestHandler", "tornado.websocket.WebSocketHandler") +_REQUEST_METHODS = ( + "get_argument", "get_arguments", "get_query_argument", "get_query_arguments", + "get_body_argument", "get_body_arguments", "get_cookie", "request", "path_args", "path_kwargs", +) +_CLIENTS = ("tornado.httpclient.AsyncHTTPClient.fetch", "tornado.httpclient.HTTPClient.fetch") +_TARGET_ONLY = ((TaintKind.REDIRECT, (0,)),) + + +def _sym(path: str) -> SymbolId: + return SymbolId(f"python.{path}") + + +class TornadoModels(ModelPlugin): + name: ClassVar[str] = "tornado-models" + models: ClassVar[tuple[Model, ...]] = ( + *(EntryPoint(_sym(handler), "http") for handler in _HANDLERS), + *(Source(_sym(f"{handler}.{method}"), "http") for handler in _HANDLERS for method in _REQUEST_METHODS), + *(Sink(_sym(f"{handler}.write"), TaintKind.HTML) for handler in _HANDLERS), + *(Sink(_sym(f"{handler}.finish"), TaintKind.HTML) for handler in _HANDLERS), + *(Sink(_sym(f"{handler}.redirect"), TaintKind.REDIRECT, _TARGET_ONLY) for handler in _HANDLERS), + *(Sink(_sym(client), TaintKind.SSRF) for client in _CLIENTS), + *(Source(_sym(client), "http-response") for client in _CLIENTS), + Sanitizer(_sym("tornado.escape.xhtml_escape"), TaintKind.HTML), + ) diff --git a/src/coretrace_python/cache.py b/src/coretrace_python/cache.py index 51b2344..a0331c7 100644 --- a/src/coretrace_python/cache.py +++ b/src/coretrace_python/cache.py @@ -37,7 +37,7 @@ from coretrace_python.semantic.symbols import SymbolId from coretrace_python.source import SourceId, SourceSpan -CACHE_FORMAT = 3 +CACHE_FORMAT = 4 @dataclass(frozen=True) @@ -252,6 +252,7 @@ def _encode_summary(summary: FunctionSummary) -> dict[str, Any]: {"name": w.name, "dependencies": sorted(w.dependencies), "externals": sorted(str(s) for s in w.externals)} for w in summary.nonlocal_writes ], + "static": summary.static, } @@ -279,6 +280,7 @@ def _decode_summary(data: Mapping[str, Any]) -> FunctionSummary: ) for w in data["nonlocal_writes"] ), + bool(data["static"]), ) diff --git a/src/coretrace_python/interprocedural/callgraph.py b/src/coretrace_python/interprocedural/callgraph.py index 8a56880..4460fb9 100644 --- a/src/coretrace_python/interprocedural/callgraph.py +++ b/src/coretrace_python/interprocedural/callgraph.py @@ -158,23 +158,43 @@ def resolve_targets( classes: Mapping[str, frozenset[str]] | None = None, owner: str | None = None, typed: Mapping[Value, str] | None = None, + bases: Mapping[str, SymbolId] | None = None, ) -> tuple[dict[Value, Target], dict[Value, SymbolId]]: """Map every callee value of ``function`` to its target, and every symbol value. ``nested`` names the functions defined inside this one (``outer.inner``); ``classes`` maps the module's classes to their methods, ``owner`` is the class of a method and ``typed`` the parameters annotated with a module class. ``App(x)`` is a - call to ``App.__init__``, ``app.run()`` and ``self.run()`` calls to ``App.run``.""" + call to ``App.__init__``, ``app.run()`` and ``self.run()`` calls to ``App.run``. + ``bases`` maps a module class to the symbol of its base: an attribute the class does + not define is inherited, so ``self.get_argument`` in a ``RequestHandler`` subclass + denotes ``tornado.web.RequestHandler.get_argument``.""" module = scopes.module_scope - symbols = derive_symbols(function, parameters) - targets: dict[Value, Target] = { - value: ExternalSymbol(symbol) for value, symbol in symbols.items() - } classes = classes or {} instance_of: dict[Value, str] = dict(typed or {}) if owner is not None and function.parameters: instance_of[function.parameters[0]] = owner defs = {i.result: i for block in function.blocks for i in block.instructions if i.result} + for block in function.blocks: + for instruction in block.instructions: + if isinstance(instruction, Call): + callee = defs.get(instruction.callee) + if isinstance(callee, Global) and callee.name in classes: + instance_of[instruction.result] = callee.name + inherited: dict[Value, SymbolId] = dict(parameters or {}) + for block in function.blocks: + for instruction in block.instructions: + if isinstance(instruction, GetAttr) and instruction.object in instance_of: + if instruction.object in inherited: + continue # an annotated parameter denotes its own class + class_name = instance_of[instruction.object] + base = (bases or {}).get(class_name) + if base is not None and instruction.attribute not in classes.get(class_name, ()): + inherited[instruction.result] = base.attribute(instruction.attribute) + symbols = derive_symbols(function, inherited) + targets: dict[Value, Target] = { + value: ExternalSymbol(symbol) for value, symbol in symbols.items() + } for block in function.blocks: for instruction in block.instructions: if isinstance(instruction, MakeFunction): @@ -189,17 +209,29 @@ def resolve_targets( targets[instruction.result] = KnownFunction(instruction.name) elif binding.kind is BindingKind.CLASS and "__init__" in classes.get(instruction.name, ()): targets[instruction.result] = KnownFunction(f"{instruction.name}.__init__") - elif isinstance(instruction, Call): - callee = defs.get(instruction.callee) - if isinstance(callee, Global) and callee.name in classes: - instance_of[instruction.result] = callee.name elif isinstance(instruction, GetAttr) and instruction.object in instance_of: method = f"{instance_of[instruction.object]}.{instruction.attribute}" if method in nested: targets[instruction.result] = KnownFunction(method) + elif isinstance(instruction, GetAttr): + # ``Student.create(...)``: a static or class method called on the class. + origin = defs.get(instruction.object) + if isinstance(origin, Global) and origin.name in classes: + method = f"{origin.name}.{instruction.attribute}" + if method in nested: + targets[instruction.result] = KnownFunction(method) return targets, symbols +def _is_staticmethod(function: nodes.Function) -> bool: + for decorator in function.decorators: + if isinstance(decorator, nodes.Name) and decorator.identifier == "staticmethod": + return True + if isinstance(decorator, nodes.Attribute) and decorator.name == "staticmethod": + return True + return False + + def _annotated( function: nodes.Function, ssa: FunctionIR, scopes: ScopeTable, table: SymbolTable ) -> dict[Value, SymbolId]: @@ -257,6 +289,14 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph: for statement in ctx.module.body if isinstance(statement, nodes.Class) } + bases: dict[str, 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 sites: dict[str, tuple[CallSite, ...]] = {} symbols: dict[str, Mapping[Value, SymbolId]] = {} unsupported: set[str] = set() @@ -268,6 +308,8 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph: sites[name] = () continue 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 targets, symbols[name] = resolve_targets( ssa, scopes, @@ -277,6 +319,7 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph: classes, owner, _typed_with_module_classes(function, ssa, classes), + bases, ) found: list[CallSite] = [] for block in ssa.blocks: diff --git a/src/coretrace_python/interprocedural/summaries.py b/src/coretrace_python/interprocedural/summaries.py index f2d556c..a16e4df 100644 --- a/src/coretrace_python/interprocedural/summaries.py +++ b/src/coretrace_python/interprocedural/summaries.py @@ -31,6 +31,7 @@ ExternalSymbol, KnownFunction, UnknownTarget, + _is_staticmethod, ) from coretrace_python.ir.model import ( BasicBlock, @@ -125,6 +126,8 @@ class FunctionSummary: mutations: tuple[Mutation, ...] = () side_effects: frozenset[str] = frozenset() nonlocal_writes: tuple[NonlocalWrite, ...] = () + # A ``@staticmethod``: ``Class.method(...)`` passes no receiver. + static: bool = False class SummaryIndex: @@ -322,7 +325,7 @@ def call(self, call: Call, state: dict[Key, Dep]) -> Dep: if project is None: project = self.project.get(target.symbol.attribute("__init__")) if project is not None: - bound = self.receiver(call, project.name) + bound = () if project.static else self.receiver(call, project.name) arguments = (*(self.deep(r, state) for r in bound), *arguments) return self.known(project, arguments, keywords, everything, call, state, bound) self.record( @@ -337,9 +340,10 @@ def call(self, call: Call, state: dict[Key, Dep]) -> Dep: made = self.defs.get(call.callee) if isinstance(made, MakeFunction): arguments = (*arguments, *(self.deep(v, state) for v in made.captured)) - bound = self.receiver(call, target.name) + callee_summary = self.table[target.name] + bound = () if callee_summary.static else self.receiver(call, target.name) arguments = (*(self.deep(r, state) for r in bound), *arguments) - return self.known(self.table[target.name], arguments, keywords, everything, call, state, bound) + return self.known(callee_summary, arguments, keywords, everything, call, state, bound) assert isinstance(target, UnknownTarget) return everything | state.get(call.callee, EMPTY) @@ -357,7 +361,7 @@ def nonlocal_result(self, instruction: NonlocalResult, state: dict[Key, Dep]) -> made = self.defs.get(call.callee) if isinstance(made, MakeFunction): arguments = (*arguments, *(self.deep(v, state) for v in made.captured)) - bound = self.receiver(call, target.name) + bound = () if callee.static else self.receiver(call, target.name) arguments = (*(self.deep(r, state) for r in bound), *arguments) keywords = EMPTY for value in (*call.starred, *(v for _, v in call.keywords)): @@ -447,6 +451,7 @@ def summarize( table: Mapping[str, FunctionSummary], project: Mapping[SymbolId, FunctionSummary] | None = None, heap: HeapFacts | None = None, + static: bool = False, ) -> FunctionSummary: problem = _DependenceProblem(name, function, graph, table, project, heap) solution = solve(problem, cfg) @@ -479,6 +484,7 @@ def summarize( for name, deps in sorted(problem.nonlocal_stores.items()) if deps.parameters or deps.externals ), + static=static, ) @@ -509,7 +515,9 @@ def compute(cls, ctx: AnalysisContext) -> SummaryTable: while changed: changed = False for name, (ssa, cfg, heap) in supported.items(): - updated = summarize(name, ssa, cfg, graph, table, project, heap) + updated = summarize( + name, ssa, cfg, graph, table, project, heap, _is_staticmethod(graph.definitions[name]) + ) if updated != table[name]: table[name] = updated changed = True diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index b32ad22..54dc35d 100644 --- a/src/coretrace_python/taint/engine.py +++ b/src/coretrace_python/taint/engine.py @@ -292,7 +292,7 @@ def call(self, call: Call, state: dict[Key, Taint], flows: list[TaintFlow]) -> T project = self.project.summary(target.symbol.attribute("__init__")) if project is not None: through = target.symbol.canonical_name.removeprefix("python.") - bound = self.receiver(call, project.name) + bound = () if project.static else self.receiver(call, project.name) arguments = (*(self.deep(r, state) for r in bound), *arguments) return self.known( project, through, arguments, keywords, everything, call, flows, state, (), bound @@ -306,7 +306,7 @@ def call(self, call: Call, state: dict[Key, Taint], flows: list[TaintFlow]) -> T if isinstance(target, KnownFunction): summary = self.summaries.summary(target.name) captured = self.captured(call) - bound = self.receiver(call, target.name) + bound = () if summary.static else self.receiver(call, target.name) arguments = ( *(self.deep(r, state) for r in bound), *arguments, @@ -331,7 +331,7 @@ def nonlocal_result(self, instruction: NonlocalResult, state: dict[Key, Taint]) return Taint.none() summary = self.summaries.summary(target.name) arguments = ( - *(self.deep(r, state) for r in self.receiver(call, target.name)), + *(self.deep(r, state) for r in (() if summary.static else self.receiver(call, target.name))), *(self.deep(a, state) for a in call.arguments), *(self.deep(v, state) for v in self.captured(call)), ) @@ -384,9 +384,9 @@ def external( sink = self.models.sink(symbol) if sink is not None: for position, argument in enumerate(call.arguments): - self.report(flows, sink, self.deep(argument, state), argument, call, None, None, position) + self.report(flows, sink, self.rendered(argument, state), argument, call, None, None, position) for argument in (*call.starred, *(value for _, value in call.keywords)): - self.report(flows, sink, self.deep(argument, state), argument, call, None, None, None) + self.report(flows, sink, self.rendered(argument, state), argument, call, None, None, None) sanitizer = self.models.sanitizer(symbol) if sanitizer is not None: return everything.without(sanitizer.kinds) @@ -397,6 +397,16 @@ def external( return everything.join(Taint(source.kinds, frozenset({source}))) return everything + def rendered(self, argument: Value, state: Mapping[Key, Taint]) -> Taint: + """The taint an argument brings to a sink. A container literal is serialised by + every framework (``self.write({...})`` answers JSON), never rendered as markup, so + it carries no ``HTML``.""" + + taint = self.deep(argument, state) + if isinstance(self.defs.get(argument), BuildDict | BuildList | BuildSet | BuildTuple): + return taint.without(TaintKind.HTML) + return taint + def returned_symbol(self, call: Call) -> SymbolId | None: """``get_db().execute(...)``: the method of what a known function returns, when its summary says the return value is one external symbol.""" diff --git a/tests/regression/expected/full-stack-fastapi-template.json b/tests/regression/expected/full-stack-fastapi-template.json index 7ec4134..f3e9c81 100644 --- a/tests/regression/expected/full-stack-fastapi-template.json +++ b/tests/regression/expected/full-stack-fastapi-template.json @@ -42,12 +42,6 @@ "rule": "xss", "function": "recover_password_html_content" }, - { - "file": "backend/app/api/routes/login.py", - "line": 121, - "rule": "xss", - "function": "recover_password_html_content" - }, { "file": "backend/tests/api/routes/test_login.py", "line": 31, diff --git a/tests/test_detectors.py b/tests/test_detectors.py index baaae6e..8fc756f 100644 --- a/tests/test_detectors.py +++ b/tests/test_detectors.py @@ -163,6 +163,7 @@ def test_shipped_plugins_load_with_their_manifests() -> None: assert set(by_name) == { "aiohttp-models", + "bottle-models", "cli-models", "command-injection", "config-secrets", @@ -187,6 +188,7 @@ def test_shipped_plugins_load_with_their_manifests() -> None: "sql-injection", "sqlalchemy-models", "ssrf", + "tornado-models", "vulnerable-dependency", "weak-crypto", "xss", diff --git a/tests/test_docs.py b/tests/test_docs.py index 6a67c88..266309d 100644 --- a/tests/test_docs.py +++ b/tests/test_docs.py @@ -90,7 +90,7 @@ def test_every_bundled_plugin_is_named_in_the_guides() -> None: re.search(r'^name = "([a-z-]+)"', manifest.read_text(encoding="utf-8"), re.MULTILINE).group(1) # type: ignore[union-attr] for manifest in engine.BUNDLED_PLUGINS.rglob("plugin.toml") ] - assert len(names) == 28 + assert len(names) == 30 assert all(f"`{name}`" in text for name in names), [n for n in names if f"`{n}`" not in text] diff --git a/tests/test_packaging.py b/tests/test_packaging.py index 527f973..b55e2db 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -64,7 +64,7 @@ def test_bundled_plugins_live_inside_the_package() -> None: assert bundled == Path(coretrace_python.__file__).resolve().parent / "bundled" assert bundled.is_dir() manifests = sorted(p.relative_to(bundled).as_posix() for p in bundled.rglob("plugin.toml")) - assert len(manifests) == 28 + assert len(manifests) == 30 assert "syntax/dangerous_eval/plugin.toml" in manifests assert not (REPO / "plugins").exists() @@ -108,7 +108,7 @@ def test_the_wheel_carries_the_bundled_plugins(tmp_path: Path) -> None: assert "coretrace_python/bundled/syntax/dangerous_eval/plugin.toml" in names assert "coretrace_python/bundled/syntax/dangerous_eval/dangerous_eval.py" in names - assert sum(1 for n in names if n.endswith("/plugin.toml")) == 28 + assert sum(1 for n in names if n.endswith("/plugin.toml")) == 30 assert not any(n.startswith("tests/") or "tests-project" in n for n in names) diff --git a/tests/test_tornado_bottle_models.py b/tests/test_tornado_bottle_models.py new file mode 100644 index 0000000..69280c4 --- /dev/null +++ b/tests/test_tornado_bottle_models.py @@ -0,0 +1,128 @@ +"""Acceptance tests for the Tornado and Bottle models (issue #70). + +Tornado handlers are classes: ``class Main(tornado.web.RequestHandler)`` reads its input +through inherited methods, ``self.get_argument(...)`` and ``self.request``, and answers +through ``self.write`` and ``self.redirect``. The call graph resolves an attribute a +module class does not define through the class's base symbol, so those calls denote +``tornado.web.RequestHandler.get_argument`` and the models apply; Django class-based +views reading ``self.request`` benefit the same way. Bottle is decorator-based like Flask: +``@route``, ``@get`` and ``Bottle().route`` are entry points, ``bottle.request`` the +source, ``redirect``, ``static_file``, ``template`` and ``HTTPResponse`` the sinks. + +Expected to remain red until both 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.source import SourceManager + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "src" / "coretrace_python" / "bundled" +MISSING = [n for n in ("tornado", "bottle") if not (PLUGINS / "models" / n / "plugin.toml").is_file()] or None + + +@pytest.fixture(autouse=True) +def require_models() -> None: + if MISSING is not None: + pytest.fail(f"model plugins are not written yet: {MISSING}") + + +def check(text: str) -> tuple[Finding, ...]: + return engine.check(SourceManager().add_source("app.py", text), [PLUGINS]) + + +def rules(findings: tuple[Finding, ...]) -> list[str]: + return sorted(f.rule_id for f in findings) + + +TORNADO = "import os\nimport tornado.web\nfrom tornado import escape\n\n" + + +def test_inherited_request_methods_of_tornado_handlers_are_sources() -> None: + text = TORNADO + "class Run(tornado.web.RequestHandler):\n def get(self):\n os.system(self.get_argument('cmd'))\n" + assert rules(check(text)) == ["command-injection"] + text = TORNADO + "class Run(tornado.web.RequestHandler):\n def post(self):\n os.system(self.request.body)\n" + assert rules(check(text)) == ["command-injection"] + + +def test_handler_method_parameters_are_http_input() -> None: + text = TORNADO + "class File(tornado.web.RequestHandler):\n def get(self, name):\n open(name).read()\n" + assert rules(check(text)) == ["path-traversal"] + + +def test_tornado_responses_and_redirects_are_sinks_and_escape_sanitises() -> None: + text = TORNADO + "class Page(tornado.web.RequestHandler):\n def get(self):\n self.write(self.get_argument('name'))\n" + assert rules(check(text)) == ["xss"] + text = TORNADO + "class Go(tornado.web.RequestHandler):\n def get(self):\n self.redirect(self.get_argument('next'))\n" + assert rules(check(text)) == ["open-redirect"] + text = TORNADO + "class Page(tornado.web.RequestHandler):\n def get(self):\n self.write(escape.xhtml_escape(self.get_argument('name')))\n" + assert check(text) == () + + +def test_methods_the_class_defines_itself_are_not_inherited() -> None: + text = TORNADO + ( + "class Run(tornado.web.RequestHandler):\n" + " def get_argument(self, name):\n return 'ls'\n" + " def get(self):\n os.system(self.get_argument('cmd'))\n" + ) + assert check(text) == () + + +def test_container_literals_are_serialised_not_rendered() -> None: + text = TORNADO + "class Api(tornado.web.RequestHandler):\n def get(self):\n self.write({'name': self.get_argument('name')})\n" + assert check(text) == () + + +def test_static_methods_have_no_receiver() -> None: + text = ( + "from typing import NamedTuple\nfrom aiopg.connection import Connection\nfrom aiohttp.web import Request\n\n" + "class Student(NamedTuple):\n id: int\n\n @staticmethod\n" + " async def create(conn: Connection, name):\n async with conn.cursor() as cur:\n" + " await cur.execute('INSERT INTO s VALUES (%s)' % name)\n\n" + "async def view(request: Request):\n await Student.create(request.app['db'], (await request.post())['name'])\n" + ) + assert rules(check(text)) == ["sql-injection"] + + +def test_tornado_http_client_is_an_ssrf_sink() -> None: + text = ( + "import tornado.web\nfrom tornado.httpclient import AsyncHTTPClient\n\n" + "class Fetch(tornado.web.RequestHandler):\n async def get(self):\n" + " await AsyncHTTPClient().fetch(self.get_argument('url'))\n" + ) + assert rules(check(text)) == ["ssrf"] + + +def test_django_class_based_views_read_self_request_through_the_base() -> None: + text = ( + "import os\nfrom django.views import View\n\n" + "class Run(View):\n def get(self, request):\n os.system(self.request.GET['cmd'])\n" + ) + assert rules(check(text)) == ["command-injection"] + + +BOTTLE = "import os\nfrom bottle import Bottle, route, get, request, redirect, static_file, template, html_escape\n\n" + + +def test_bottle_routes_are_entry_points_and_request_is_the_source() -> None: + assert rules(check(BOTTLE + "@route('/run')\ndef run():\n os.system(request.query.cmd)\n")) == ["command-injection"] + assert rules(check(BOTTLE + "@get('/run')\ndef run():\n os.system(request.forms.get('cmd'))\n")) == ["command-injection"] + assert rules(check(BOTTLE + "app = Bottle()\n\n@app.post('/run')\ndef run():\n os.system(request.json['cmd'])\n")) == [ + "command-injection" + ] + assert check(BOTTLE + "def helper(value):\n os.system(value)\n") == () + + +def test_bottle_sinks_and_sanitizer() -> None: + assert rules(check(BOTTLE + "@route('/go')\ndef go():\n redirect(request.query.next)\n")) == ["open-redirect"] + assert rules(check(BOTTLE + "@route('/f')\ndef f():\n return static_file(request.query.name, root='/srv/files')\n")) == [ + "path-traversal" + ] + assert rules(check(BOTTLE + "@route('/t')\ndef t():\n return template(request.query.tpl)\n")) == ["xss"] + assert check(BOTTLE + "@route('/t')\ndef t():\n return template('page', name=html_escape(request.query.name))\n") == ()