From 2eaadc1cc9808450c3f4e1d6f0f6085f69b4aee6 Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 4 Sep 2026 01:24:14 +0200 Subject: [PATCH 1/2] test: define framework model behavior --- tests/test_detectors.py | 3 + tests/test_frameworks.py | 276 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 279 insertions(+) create mode 100644 tests/test_frameworks.py diff --git a/tests/test_detectors.py b/tests/test_detectors.py index 2e8a0eb..31434df 100644 --- a/tests/test_detectors.py +++ b/tests/test_detectors.py @@ -161,9 +161,12 @@ def test_shipped_plugins_load_with_their_manifests() -> None: assert set(by_name) == { "command-injection", "dangerous-eval", + "fastapi-models", + "flask-models", "path-traversal", "python-stdlib-models", "sql-injection", + "sqlalchemy-models", "ssrf", "weak-crypto", "xss", diff --git a/tests/test_frameworks.py b/tests/test_frameworks.py new file mode 100644 index 0000000..a62f862 --- /dev/null +++ b/tests/test_frameworks.py @@ -0,0 +1,276 @@ +"""Acceptance tests for framework models (``docs/architecture.md`` §15, §25, §36). + +Framework plugins enrich the security model; detectors stay generic. Three engine +mechanisms make that possible: + +- a module-level ``name = Symbol(...)`` binds ``name`` to that symbol, so ``app.route`` + resolves to ``python.flask.Flask.route`` when ``app = Flask(__name__)``; +- inside functions, call results and ``with`` contexts carry their callee's symbol, so + ``conn.cursor().execute`` resolves to ``python.sqlite3.connect.cursor.execute``; +- an ``EntryPoint`` model taints every parameter of a function decorated by its symbol, + which is how route handler arguments become HTTP input. + +Shipped: ``models/flask``, ``models/fastapi``, ``models/sqlalchemy`` and sqlite3 sinks in +``models/python_stdlib``. Expected to remain red until they exist. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.analysis import AnalysisManager +from coretrace_python.findings import Finding +from coretrace_python.frontend import build_hir +from coretrace_python.hir import nodes +from coretrace_python.interprocedural import CallGraphAnalysis, ExternalSymbol +from coretrace_python.ir.lowering import lower_module +from coretrace_python.ir.printer import format_module +from coretrace_python.semantic.scopes import ScopeAnalysis +from coretrace_python.semantic.symbols import SymbolAnalysis, SymbolId +from coretrace_python.source import SourceManager +from coretrace_python.taint import ( + SecurityModelAnalysis, + SecurityModelRegistry, + Sink, + TaintAnalysis, + TaintKind, +) + +try: + from coretrace_python.taint import EntryPoint +except ImportError as error: # pragma: no cover - red until entry points land + MISSING = error +else: + MISSING = None + + +@pytest.fixture(autouse=True) +def require_entry_points() -> None: + if MISSING is not None: + pytest.fail(f"framework support is not implemented yet: {MISSING}") + + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "plugins" + + +def module_for(source_text: str) -> nodes.Module: + return build_hir(SourceManager().add_source("web.py", source_text)) + + +def manager_for(source_text: str, *models: object) -> AnalysisManager: + registry = SecurityModelRegistry() + registry.register(*models) # type: ignore[arg-type] + return engine.build_manager(module_for(source_text), registry) + + +def check(source_text: str) -> tuple[Finding, ...]: + return engine.check(SourceManager().add_source("web.py", source_text), [PLUGINS]) + + +def rules(findings: tuple[Finding, ...]) -> list[str]: + return [f.rule_id for f in findings] + + +# --------------------------------------------------------------------------- symbol derivation + + +def test_module_level_instances_resolve_to_their_class_symbol() -> None: + manager = manager_for("from flask import Flask\napp = Flask(__name__)\nlimit = 3\n\ndef f():\n return app\n") + symbols = manager.get(SymbolAnalysis) + scopes = manager.get(ScopeAnalysis) + f = next(s for s in scopes.children(scopes.module_scope.id) if s.name == "f") + + assert symbols.resolve(scopes.module_scope.id, "app") == SymbolId("python.flask.Flask") + assert symbols.resolve(f.id, "app") == SymbolId("python.flask.Flask") + assert symbols.resolve(f.id, "limit") is None + + +def test_attributes_of_module_level_instances_lower_to_symbols() -> None: + module = module_for("from flask import Flask\napp = Flask(__name__)\n\ndef f():\n return app.route\n") + output = format_module(lower_module(module)) + + assert "symbol @python.flask.Flask.route" in output + + +def test_call_results_and_with_contexts_carry_their_symbol_in_the_call_graph() -> None: + manager = manager_for( + "import sqlite3\n\n" + "def query(path, sql):\n" + " conn = sqlite3.connect(path)\n" + " cur = conn.cursor()\n" + " cur.execute(sql)\n" + " with sqlite3.connect(path) as other:\n" + " other.execute(sql)\n" + ) + graph = manager.get(CallGraphAnalysis) + + assert [site.target for site in graph.sites("query")] == [ + ExternalSymbol(SymbolId("python.sqlite3.connect")), + ExternalSymbol(SymbolId("python.sqlite3.connect.cursor")), + ExternalSymbol(SymbolId("python.sqlite3.connect.cursor.execute")), + ExternalSymbol(SymbolId("python.sqlite3.connect")), + ExternalSymbol(SymbolId("python.sqlite3.connect.execute")), + ] + + +def test_symbols_are_not_derived_through_known_functions_or_parameters() -> None: + manager = manager_for("def make():\n return 1\n\ndef use(x):\n make().run()\n x.run()\n") + graph = manager.get(CallGraphAnalysis) + + targets = [site.target for site in graph.sites("use")] + assert not any(isinstance(t, ExternalSymbol) for t in targets[1:]) + + +# --------------------------------------------------------------------------- entry points + + +ROUTE = EntryPoint(SymbolId("python.flask.Flask.route"), "http") if MISSING is None else None + + +def flows(source_text: str, *models: object): # type: ignore[no-untyped-def] + manager = manager_for(source_text, *models) + module = manager.module + function = next(s for s in module.body if isinstance(s, nodes.Function)) + return manager.get(TaintAnalysis, function).flows + + +def test_entry_point_parameters_are_sources() -> None: + found = flows( + "from flask import Flask\nimport os\napp = Flask(__name__)\n\n" + "@app.route('/ping/')\n" + "def ping(host):\n" + " os.system('ping ' + host)\n", + ROUTE, + Sink(SymbolId("python.os.system"), TaintKind.COMMAND), + ) + + (flow,) = found + assert flow.source.label == "http" + assert flow.source.symbol == SymbolId("python.flask.Flask.route") + assert flow.location.start_line == 7 + + +def test_only_decorated_functions_are_entry_points() -> None: + found = flows( + "from flask import Flask\nimport os\napp = Flask(__name__)\n\n" + "def helper(host):\n" + " os.system(host)\n", + ROUTE, + Sink(SymbolId("python.os.system"), TaintKind.COMMAND), + ) + assert found == () + + +def test_bare_decorators_also_mark_entry_points() -> None: + found = flows( + "from auth import protected\nimport os\n\n" + "@protected\n" + "def run(cmd):\n" + " os.system(cmd)\n", + EntryPoint(SymbolId("python.auth.protected"), "rpc"), + Sink(SymbolId("python.os.system"), TaintKind.COMMAND), + ) + (flow,) = found + assert flow.source.label == "rpc" + + +def test_entry_points_are_indexed_in_the_model_table() -> None: + registry = SecurityModelRegistry() + registry.register(ROUTE) + table = registry.freeze() + + assert table.entry_point(SymbolId("python.flask.Flask.route")) == ROUTE + assert table.entry_point(SymbolId("python.other")) is None + assert table.entry_points == (ROUTE,) + manager = AnalysisManager(module_for("")) + manager.register(SecurityModelAnalysis) + manager.provide(SecurityModelAnalysis, table) + assert manager.get(SecurityModelAnalysis).entry_points == (ROUTE,) + + +# --------------------------------------------------------------------------- shipped plugins + + +def test_shipped_framework_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["flask-models"].provides == ("model.http-sources", "model.flask-routes") + assert loaded["fastapi-models"].provides == ("model.http-sources", "model.fastapi-routes") + assert loaded["sqlalchemy-models"].provides == ("model.sql-sinks",) + + +def test_flask_request_to_sqlite_is_a_sql_injection() -> None: + findings = check( + "import sqlite3\nfrom flask import Flask, request\n\napp = Flask(__name__)\n\n" + "@app.route('/user')\n" + "def user():\n" + " conn = sqlite3.connect('app.db')\n" + " conn.execute(\"SELECT * FROM users WHERE name = '\" + request.args['name'] + \"'\")\n" + ) + + assert rules(findings) == ["sql-injection"] + assert findings[0].span.start_line == 9 + assert findings[0].metadata["source_label"] == "http" + + +def test_flask_route_parameter_to_os_system_is_a_command_injection() -> None: + findings = check( + "import os\nfrom flask import Flask\n\napp = Flask(__name__)\n\n" + "@app.route('/ping/')\n" + "def ping(host):\n" + " os.system('ping ' + host)\n" + ) + assert rules(findings) == ["command-injection"] + + +def test_flask_template_string_is_an_xss_sink_and_escape_a_sanitizer() -> None: + findings = check( + "from flask import Flask, request, render_template_string\nfrom markupsafe import escape\n\n" + "app = Flask(__name__)\n\n" + "@app.route('/hello')\n" + "def hello():\n" + " name = request.args.get('name')\n" + " render_template_string('

' + name + '

')\n" + " return render_template_string('

' + escape(name) + '

')\n" + ) + assert rules(findings) == ["xss"] + assert findings[0].span.start_line == 9 + + +def test_fastapi_route_parameters_are_http_input() -> None: + findings = check( + "import os\nfrom fastapi import FastAPI\n\napp = FastAPI()\n\n" + "@app.get('/run')\n" + "def run(cmd):\n" + " os.system(cmd)\n" + ) + assert rules(findings) == ["command-injection"] + + +def test_sqlalchemy_text_and_connection_execute_are_sql_sinks() -> None: + findings = check( + "from flask import Flask, request\nfrom sqlalchemy import create_engine, text\n\n" + "app = Flask(__name__)\nengine = create_engine('sqlite://')\n\n" + "@app.route('/find')\n" + "def find():\n" + " name = request.args['name']\n" + " with engine.connect() as conn:\n" + " conn.execute(text('SELECT * FROM t WHERE n = ' + name))\n" + " conn.execute('SELECT * FROM t WHERE n = ' + name)\n" + ) + assert rules(findings) == ["sql-injection", "sql-injection"] + assert [f.span.start_line for f in findings] == [11, 12] + + +def test_detectors_stay_generic_across_frameworks() -> None: + from coretrace_python.plugins import discover_plugins + + names = {p.manifest.name for p in discover_plugins(PLUGINS, engine.build_manager(module_for("")))} + assert not any("flask" in name and "injection" in name for name in names) + assert {"sql-injection", "command-injection", "xss"} <= names From a9391160d11973104a4374216d96c398ba9ffdd5 Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 4 Sep 2026 01:24:15 +0200 Subject: [PATCH 2/2] feat: add Flask, FastAPI and SQLAlchemy models with entry points and symbol derivation --- README.md | 13 +++- plugins/models/fastapi/fastapi_models.py | 25 +++++++ plugins/models/fastapi/plugin.toml | 9 +++ plugins/models/flask/flask_models.py | 34 +++++++++ plugins/models/flask/plugin.toml | 9 +++ plugins/models/python_stdlib/python_stdlib.py | 5 ++ plugins/models/sqlalchemy/plugin.toml | 9 +++ .../models/sqlalchemy/sqlalchemy_models.py | 38 ++++++++++ .../interprocedural/callgraph.py | 58 +++++++++++--- src/coretrace_python/semantic/symbols.py | 55 ++++++++++++-- src/coretrace_python/taint/__init__.py | 2 + src/coretrace_python/taint/engine.py | 75 ++++++++++++++----- src/coretrace_python/taint/models.py | 29 ++++++- 13 files changed, 322 insertions(+), 39 deletions(-) create mode 100644 plugins/models/fastapi/fastapi_models.py create mode 100644 plugins/models/fastapi/plugin.toml create mode 100644 plugins/models/flask/flask_models.py create mode 100644 plugins/models/flask/plugin.toml create mode 100644 plugins/models/sqlalchemy/plugin.toml create mode 100644 plugins/models/sqlalchemy/sqlalchemy_models.py diff --git a/README.md b/README.md index f9a4e06..2a794f3 100644 --- a/README.md +++ b/README.md @@ -32,10 +32,12 @@ coretrace-python-analyzer --check app.py --plugins plugins/ --format sarif > rep ``` `--plugins` is a directory searched recursively for `plugin.toml` manifests and may be -repeated. The repository ships a first set under `plugins/`: standard-library security -models (`models/python_stdlib`), taint detectors for SQL injection, command injection, -path traversal, SSRF and XSS (`security/`), and syntactic detectors for `eval`/`exec` -and weak hashes (`syntax/`). Model plugins contribute sources, sinks and sanitizers; +repeated. The repository ships a first set under `plugins/`: security models for the +standard library, Flask, FastAPI and SQLAlchemy (`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; 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 @@ -64,6 +66,9 @@ instructions and unassigned paths read an explicit `undefined` value. Unsupported syntax produces a source-located diagnostic and a non-zero exit status. Names are resolved to canonical symbols through lexical scopes, imports and builtins. +Calling a symbol denotes that symbol, so `app = Flask(__name__)` makes `app.route` resolve +to `python.flask.Flask.route` and `sqlite3.connect(p).cursor().execute` resolves to +`python.sqlite3.connect.cursor.execute`; model plugins register those chains. Import aliases do not change an API's identity, so all of the following resolve to `python.os.system`: diff --git a/plugins/models/fastapi/fastapi_models.py b/plugins/models/fastapi/fastapi_models.py new file mode 100644 index 0000000..cf5ef79 --- /dev/null +++ b/plugins/models/fastapi/fastapi_models.py @@ -0,0 +1,25 @@ +"""FastAPI security models: route handlers and response sinks.""" + +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, Sink, TaintKind + +_METHODS = ("get", "post", "put", "delete", "patch", "options", "head", "api_route", "websocket") + + +def _sym(path: str) -> SymbolId: + return SymbolId(f"python.{path}") + + +class FastApiModels(ModelPlugin): + name: ClassVar[str] = "fastapi-models" + 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), + Sink(_sym("fastapi.responses.HTMLResponse"), TaintKind.HTML), + Sink(_sym("fastapi.responses.FileResponse"), TaintKind.PATH), + ) diff --git a/plugins/models/fastapi/plugin.toml b/plugins/models/fastapi/plugin.toml new file mode 100644 index 0000000..384188f --- /dev/null +++ b/plugins/models/fastapi/plugin.toml @@ -0,0 +1,9 @@ +name = "fastapi-models" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = [] +provides = ["model.http-sources", "model.fastapi-routes"] + +[entrypoint] +module = "fastapi_models" +class = "FastApiModels" diff --git a/plugins/models/flask/flask_models.py b/plugins/models/flask/flask_models.py new file mode 100644 index 0000000..d2faba3 --- /dev/null +++ b/plugins/models/flask/flask_models.py @@ -0,0 +1,34 @@ +"""Flask security models: HTTP sources, route handlers, HTML sinks and escaping.""" + +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 + +_REQUEST_ATTRIBUTES = ( + "args", "form", "values", "json", "data", "cookies", "headers", "files", + "get_json", "get_data", "url", "full_path", "path", "query_string", "stream", +) + + +def _sym(path: str) -> SymbolId: + return SymbolId(f"python.{path}") + + +class FlaskModels(ModelPlugin): + name: ClassVar[str] = "flask-models" + models: ClassVar[tuple[Model, ...]] = ( + *(Source(_sym(f"flask.request.{attribute}"), "http") for attribute in _REQUEST_ATTRIBUTES), + EntryPoint(_sym("flask.Flask.route"), "http"), + EntryPoint(_sym("flask.Blueprint.route"), "http"), + Sink(_sym("flask.render_template_string"), TaintKind.HTML), + Sink(_sym("flask.make_response"), TaintKind.HTML), + Sink(_sym("flask.Response"), TaintKind.HTML), + Sink(_sym("flask.Markup"), TaintKind.HTML), + Sink(_sym("flask.send_file"), TaintKind.PATH), + Sanitizer(_sym("flask.escape"), TaintKind.HTML), + Sanitizer(_sym("markupsafe.escape"), TaintKind.HTML), + ) diff --git a/plugins/models/flask/plugin.toml b/plugins/models/flask/plugin.toml new file mode 100644 index 0000000..da1b5c7 --- /dev/null +++ b/plugins/models/flask/plugin.toml @@ -0,0 +1,9 @@ +name = "flask-models" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = [] +provides = ["model.http-sources", "model.flask-routes"] + +[entrypoint] +module = "flask_models" +class = "FlaskModels" diff --git a/plugins/models/python_stdlib/python_stdlib.py b/plugins/models/python_stdlib/python_stdlib.py index 9266bac..82a0447 100644 --- a/plugins/models/python_stdlib/python_stdlib.py +++ b/plugins/models/python_stdlib/python_stdlib.py @@ -35,6 +35,11 @@ class PythonStdlibModels(ModelPlugin): Sink(_sym("os.rmdir"), TaintKind.PATH), Sink(_sym("shutil.rmtree"), TaintKind.PATH), Sink(_sym("urllib.request.urlopen"), TaintKind.SSRF), + *( + Sink(_sym(f"sqlite3.connect{cursor}.{method}"), TaintKind.SQL) + for cursor in ("", ".cursor") + for method in ("execute", "executemany", "executescript") + ), Sanitizer(_sym("shlex.quote"), TaintKind.COMMAND), Sanitizer(_sym("html.escape"), TaintKind.HTML), Sanitizer(_sym("os.path.basename"), TaintKind.PATH), diff --git a/plugins/models/sqlalchemy/plugin.toml b/plugins/models/sqlalchemy/plugin.toml new file mode 100644 index 0000000..9c2215f --- /dev/null +++ b/plugins/models/sqlalchemy/plugin.toml @@ -0,0 +1,9 @@ +name = "sqlalchemy-models" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = [] +provides = ["model.sql-sinks"] + +[entrypoint] +module = "sqlalchemy_models" +class = "SqlAlchemyModels" diff --git a/plugins/models/sqlalchemy/sqlalchemy_models.py b/plugins/models/sqlalchemy/sqlalchemy_models.py new file mode 100644 index 0000000..07a426a --- /dev/null +++ b/plugins/models/sqlalchemy/sqlalchemy_models.py @@ -0,0 +1,38 @@ +"""SQLAlchemy security models: raw SQL execution sinks. + +Symbols follow the engine's call-chain rule: the result of calling a symbol denotes +that symbol, so ``create_engine(url).connect().execute`` is +``python.sqlalchemy.create_engine.connect.execute``. +""" + +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, TaintKind + +_EXECUTORS = ( + "sqlalchemy.create_engine", + "sqlalchemy.create_engine.connect", + "sqlalchemy.create_engine.begin", + "sqlalchemy.engine.create_engine", + "sqlalchemy.engine.create_engine.connect", + "sqlalchemy.orm.Session", + "sqlalchemy.orm.sessionmaker", + "sqlalchemy.orm.scoped_session", +) + + +def _sym(path: str) -> SymbolId: + return SymbolId(f"python.{path}") + + +class SqlAlchemyModels(ModelPlugin): + name: ClassVar[str] = "sqlalchemy-models" + models: ClassVar[tuple[Model, ...]] = tuple( + Sink(_sym(f"{executor}.{method}"), TaintKind.SQL) + for executor in _EXECUTORS + for method in ("execute", "exec_driver_sql", "scalar", "scalars") + ) diff --git a/src/coretrace_python/interprocedural/callgraph.py b/src/coretrace_python/interprocedural/callgraph.py index 888ddbd..db8b386 100644 --- a/src/coretrace_python/interprocedural/callgraph.py +++ b/src/coretrace_python/interprocedural/callgraph.py @@ -17,7 +17,7 @@ from coretrace_python.cfg import CFGError from coretrace_python.hir import nodes from coretrace_python.ir.lowering import LoweringError, analyzable_functions, qualified_name -from coretrace_python.ir.model import Call, FunctionIR, Global, Symbol, Value +from coretrace_python.ir.model import Call, FunctionIR, GetAttr, Global, Symbol, Value, WithEnter from coretrace_python.ir.ssa import SSAAnalysis from coretrace_python.semantic.scopes import BindingKind, ScopeAnalysis, ScopeTable from coretrace_python.semantic.symbols import SymbolId @@ -57,7 +57,9 @@ def __init__( definitions: Mapping[str, nodes.Function], sites: Mapping[str, tuple[CallSite, ...]], unsupported: frozenset[str], + symbols: Mapping[str, Mapping[Value, SymbolId]] | None = None, ) -> None: + self._symbols = {name: MappingProxyType(dict(found)) for name, found in (symbols or {}).items()} self.definitions: Mapping[str, nodes.Function] = MappingProxyType(dict(definitions)) self.functions = tuple(definitions) self._names = {function.span: name for name, function in definitions.items()} @@ -76,6 +78,11 @@ def __init__( def name_of(self, function: nodes.Function) -> str: return self._names[function.span] + def symbols(self, name: str) -> Mapping[Value, SymbolId]: + """Values of ``name`` that denote a symbol, including derived call-chain symbols.""" + + return self._symbols.get(name, MappingProxyType({})) + def sites(self, caller: str) -> tuple[CallSite, ...]: return self._sites.get(caller, ()) @@ -91,18 +98,48 @@ def callers(self, name: str) -> frozenset[str]: return self._callers.get(name, frozenset()) +def derive_symbols(function: FunctionIR) -> dict[Value, SymbolId]: + """Symbols of values: ``Symbol`` results, attributes of symbol values, results of + calling a symbol (``sqlite3.connect(p)`` denotes ``python.sqlite3.connect``) and the + values a ``with`` on such a result binds. Known functions and parameters derive nothing.""" + + symbols: dict[Value, SymbolId] = {} + changed = True + while changed: + changed = False + for block in function.blocks: + for instruction in block.instructions: + if instruction.result is None or instruction.result in symbols: + continue + symbol: SymbolId | None = None + if isinstance(instruction, Symbol): + symbol = instruction.symbol_id + elif isinstance(instruction, GetAttr) and instruction.object in symbols: + symbol = symbols[instruction.object].attribute(instruction.attribute) + elif isinstance(instruction, Call | WithEnter): + origin = ( + instruction.callee if isinstance(instruction, Call) else instruction.context + ) + symbol = symbols.get(origin) + if symbol is not None: + symbols[instruction.result] = symbol + changed = True + return symbols + + def resolve_targets( function: FunctionIR, scopes: ScopeTable, known: frozenset[str] -) -> dict[Value, Target]: - """Map every callee value of ``function`` to its target.""" +) -> tuple[dict[Value, Target], dict[Value, SymbolId]]: + """Map every callee value of ``function`` to its target, and every symbol value.""" module = scopes.module_scope - targets: dict[Value, Target] = {} + symbols = derive_symbols(function) + 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, Symbol): - targets[instruction.result] = ExternalSymbol(instruction.symbol_id) - elif isinstance(instruction, Global): + if isinstance(instruction, Global): binding = module.bindings.get(instruction.name) if ( binding is not None @@ -110,7 +147,7 @@ def resolve_targets( and instruction.name in known ): targets[instruction.result] = KnownFunction(instruction.name) - return targets + return targets, symbols class CallGraphAnalysis(Analysis[CallGraph]): @@ -128,6 +165,7 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph: name for name, function in definitions.items() if function in ctx.module.body ) sites: dict[str, tuple[CallSite, ...]] = {} + symbols: dict[str, Mapping[Value, SymbolId]] = {} unsupported: set[str] = set() for name, function in definitions.items(): try: @@ -136,7 +174,7 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph: unsupported.add(name) sites[name] = () continue - targets = resolve_targets(ssa, scopes, known) + targets, symbols[name] = resolve_targets(ssa, scopes, known) found: list[CallSite] = [] for block in ssa.blocks: for instruction in block.instructions: @@ -151,4 +189,4 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph: ) ) sites[name] = tuple(found) - return CallGraph(definitions, sites, frozenset(unsupported)) + return CallGraph(definitions, sites, frozenset(unsupported), symbols) diff --git a/src/coretrace_python/semantic/symbols.py b/src/coretrace_python/semantic/symbols.py index 47dadc1..20d3de4 100644 --- a/src/coretrace_python/semantic/symbols.py +++ b/src/coretrace_python/semantic/symbols.py @@ -3,9 +3,12 @@ from __future__ import annotations import builtins +from collections.abc import Mapping +from types import MappingProxyType from typing import ClassVar from coretrace_python.analysis import Analysis, AnalysisContext, AnyAnalysis +from coretrace_python.hir import nodes from coretrace_python.semantic.identity import SymbolId from coretrace_python.semantic.imports import ImportAnalysis, ImportTable from coretrace_python.semantic.scopes import ResolutionKind, ScopeAnalysis, ScopeId, ScopeTable @@ -14,22 +17,62 @@ class SymbolTable: - """Resolve names to canonical symbols through scopes, imports and builtins.""" + """Resolve names to canonical symbols through scopes, imports, builtins and the + module-level instances created by calling a symbol (``app = Flask(__name__)``).""" - def __init__(self, scopes: ScopeTable, imports: ImportTable) -> None: + def __init__( + self, + scopes: ScopeTable, + imports: ImportTable, + instances: Mapping[str, SymbolId] | None = None, + ) -> None: self._scopes = scopes self._imports = imports + self._instances = MappingProxyType(dict(instances or {})) def resolve(self, scope_id: ScopeId, name: str) -> SymbolId | None: resolution = self._scopes.resolve(scope_id, name) if resolution.kind is ResolutionKind.UNBOUND: return SymbolId(f"python.builtins.{name}") if name in BUILTIN_NAMES else None assert resolution.scope is not None - return self._imports.bindings(resolution.scope).get(name) + imported = self._imports.bindings(resolution.scope).get(name) + if imported is not None: + return imported + if resolution.scope == self._scopes.module_scope.id: + return self._instances.get(name) + return None + def resolve_expression(self, scope_id: ScopeId, node: nodes.Expression) -> SymbolId | None: + """The symbol of a name or attribute chain, or of the callee of a call.""" -def analyze_symbols(scopes: ScopeTable, imports: ImportTable) -> SymbolTable: - return SymbolTable(scopes, imports) + if isinstance(node, nodes.Name): + return self.resolve(scope_id, node.identifier) + if isinstance(node, nodes.Attribute): + parent = self.resolve_expression(scope_id, node.value) + return parent.attribute(node.name) if parent is not None else None + if isinstance(node, nodes.Call): + return self.resolve_expression(scope_id, node.callee) + return None + + +def analyze_symbols( + scopes: ScopeTable, imports: ImportTable, module: nodes.Module | None = None +) -> SymbolTable: + table = SymbolTable(scopes, imports) + if module is None: + return table + instances: dict[str, SymbolId] = {} + module_scope = scopes.module_scope.id + for statement in module.body: + if ( + isinstance(statement, nodes.Assign) + and isinstance(statement.target, nodes.Name) + and isinstance(statement.value, nodes.Call) + ): + symbol = table.resolve_expression(module_scope, statement.value.callee) + if symbol is not None: + instances[statement.target.identifier] = symbol + return SymbolTable(scopes, imports, instances) class SymbolAnalysis(Analysis[SymbolTable]): @@ -38,7 +81,7 @@ class SymbolAnalysis(Analysis[SymbolTable]): @classmethod def compute(cls, ctx: AnalysisContext) -> SymbolTable: - return analyze_symbols(ctx.get(ScopeAnalysis), ctx.get(ImportAnalysis)) + return analyze_symbols(ctx.get(ScopeAnalysis), ctx.get(ImportAnalysis), ctx.module) __all__ = ["BUILTIN_NAMES", "SymbolAnalysis", "SymbolId", "SymbolTable", "analyze_symbols"] diff --git a/src/coretrace_python/taint/__init__.py b/src/coretrace_python/taint/__init__.py index ba559e1..8428d1a 100644 --- a/src/coretrace_python/taint/__init__.py +++ b/src/coretrace_python/taint/__init__.py @@ -8,6 +8,7 @@ propagate_taint, ) from coretrace_python.taint.models import ( + EntryPoint, Model, ModelError, ModelTable, @@ -20,6 +21,7 @@ ) __all__ = [ + "EntryPoint", "Model", "ModelError", "ModelTable", diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index 74ce95e..b0086cf 100644 --- a/src/coretrace_python/taint/engine.py +++ b/src/coretrace_python/taint/engine.py @@ -40,9 +40,11 @@ Value, ) from coretrace_python.ir.ssa import SSAAnalysis -from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.semantic.scopes import ScopeAnalysis, ScopeTable +from coretrace_python.semantic.symbols import SymbolAnalysis, SymbolTable from coretrace_python.source import SourceSpan from coretrace_python.taint.models import ( + EntryPoint, ModelTable, SecurityModelAnalysis, Sink, @@ -107,22 +109,23 @@ def __init__( models: ModelTable, graph: CallGraph, summaries: SummaryTable, + entry: EntryPoint | None = None, ) -> None: self.name = name self.function = function self.models = models self.graph = graph self.summaries = summaries + self.entry = entry self.blocks = {block.id: block for block in function.blocks} - self.symbols: dict[Value, SymbolId] = { - i.result: i.symbol_id - for block in function.blocks - for i in block.instructions - if isinstance(i, Symbol) - } + self.symbols = graph.symbols(name) def initial(self) -> State: - return MappingProxyType({}) + 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}) def join(self, a: State, b: State) -> State: merged = dict(a) @@ -171,12 +174,14 @@ def flow(self, cfg: CFG, block_id: BlockId, incoming: Mapping[BlockId, State]) - # ------------------------------------------------------------------ transfer def instruction(self, instruction: Instruction, state: Mapping[Value, Taint]) -> Taint: - if isinstance(instruction, Symbol): - source = self.models.source(instruction.symbol_id) - return Taint(source.kinds, frozenset({source})) if source else Taint.none() - if isinstance(instruction, Compare): - return Taint.none() taint = Taint.none() + symbol = self.symbols.get(instruction.result) if instruction.result else None + if symbol is not None: + source = self.models.source_covering(symbol) + if source is not None: + taint = Taint(source.kinds, frozenset({source})) + if isinstance(instruction, Symbol | Compare): + return taint for operand in instruction.operands(): taint = taint.join(state.get(operand, Taint.none())) return taint @@ -199,7 +204,9 @@ def call(self, call: Call, state: Mapping[Value, Taint], flows: list[TaintFlow]) sanitizer = self.models.sanitizer(target.symbol) if sanitizer is not None: return everything.without(sanitizer.kinds) - source = self.models.source(target.symbol) + # A method on a tainted object returns tainted data (``request.args.get``). + everything = everything.join(state.get(call.callee, Taint.none())) + source = self.models.source_covering(target.symbol) if source is not None: return everything.join(Taint(source.kinds, frozenset({source}))) return everything @@ -262,10 +269,32 @@ def report( ) +def entry_point_of( + function: nodes.Function, models: ModelTable, scopes: ScopeTable, symbols: SymbolTable +) -> EntryPoint | None: + """The entry-point model matching one of the function's decorators, 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) + if symbol is not None: + entry = models.entry_point(symbol) + if entry is not None: + return entry + return None + + def propagate_taint( - name: str, function: FunctionIR, cfg: CFG, models: ModelTable, graph: CallGraph, summaries: SummaryTable + name: str, + function: FunctionIR, + cfg: CFG, + models: ModelTable, + graph: CallGraph, + summaries: SummaryTable, + entry: EntryPoint | None = None, ) -> TaintFacts: - problem = _TaintProblem(name, function, models, graph, summaries) + problem = _TaintProblem(name, function, models, graph, summaries, entry) solution = solve(problem, cfg) taints: dict[Value, Taint] = {} flows: list[TaintFlow] = [] @@ -282,17 +311,27 @@ class TaintAnalysis(FunctionAnalysis[TaintFacts]): name: ClassVar[str] = "taint.flows" requires: ClassVar[frozenset[AnyAnalysis]] = frozenset( - {SSAAnalysis, CFGAnalysis, SecurityModelAnalysis, CallGraphAnalysis, SummaryAnalysis} + { + SSAAnalysis, + CFGAnalysis, + SecurityModelAnalysis, + CallGraphAnalysis, + SummaryAnalysis, + ScopeAnalysis, + SymbolAnalysis, + } ) @classmethod def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> TaintFacts: graph = ctx.get(CallGraphAnalysis) + models = ctx.get(SecurityModelAnalysis) return propagate_taint( graph.name_of(function), ctx.get(SSAAnalysis, function), ctx.get(CFGAnalysis, function), - ctx.get(SecurityModelAnalysis), + models, graph, ctx.get(SummaryAnalysis), + entry_point_of(function, models, ctx.get(ScopeAnalysis), ctx.get(SymbolAnalysis)), ) diff --git a/src/coretrace_python/taint/models.py b/src/coretrace_python/taint/models.py index ab737ba..05042a0 100644 --- a/src/coretrace_python/taint/models.py +++ b/src/coretrace_python/taint/models.py @@ -57,7 +57,16 @@ class Sanitizer: kinds: TaintKind -Model = Source | Sink | Sanitizer +@dataclass(frozen=True) +class EntryPoint: + """Functions decorated by ``symbol`` receive attacker-controlled parameters.""" + + symbol: SymbolId + label: str + kinds: TaintKind = TaintKind.ALL + + +Model = Source | Sink | Sanitizer | EntryPoint @dataclass(frozen=True) @@ -65,6 +74,7 @@ class ModelTable: sources: tuple[Source, ...] sinks: tuple[Sink, ...] sanitizers: tuple[Sanitizer, ...] + entry_points: tuple[EntryPoint, ...] = () _by_symbol: dict[type[Model], dict[SymbolId, Model]] = field( init=False, repr=False, compare=False ) @@ -74,13 +84,29 @@ def __post_init__(self) -> None: Source: {m.symbol: m for m in self.sources}, 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}, } object.__setattr__(self, "_by_symbol", MappingProxyType(index)) + def entry_point(self, symbol: SymbolId) -> EntryPoint | None: + found = self._by_symbol[EntryPoint].get(symbol) + return found if isinstance(found, EntryPoint) else None + def source(self, symbol: SymbolId) -> Source | None: found = self._by_symbol[Source].get(symbol) return found if isinstance(found, Source) else None + def source_covering(self, symbol: SymbolId) -> Source | None: + """The source registered for ``symbol`` or for the closest symbol above it, so a + source on ``flask.request.args`` also covers ``flask.request.args.get``.""" + + parts = symbol.canonical_name.split(".") + for length in range(len(parts), 1, -1): + found = self.source(SymbolId(".".join(parts[:length]))) + if found is not None: + return found + return None + def sink(self, symbol: SymbolId) -> Sink | None: found = self._by_symbol[Sink].get(symbol) return found if isinstance(found, Sink) else None @@ -111,6 +137,7 @@ def freeze(self) -> ModelTable: sources=tuple(m for m in models if isinstance(m, Source)), 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)), )