From c95a4b9abc021aaf4b3ae3fd88cc1d5c5290df89 Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 4 Sep 2026 15:53:15 +0200 Subject: [PATCH 1/2] test: define FastAPI, SQL position, deserialization, redirect and CLI behaviour --- tests/test_advisory_feeds.py | 4 +- tests/test_correlation.py | 6 +- tests/test_detectors.py | 9 +- tests/test_django_http_clients.py | 2 + tests/test_fastapi_cli_precision.py | 254 ++++++++++++++++++++++++++++ tests/test_parallel.py | 1 + 6 files changed, 271 insertions(+), 5 deletions(-) create mode 100644 tests/test_fastapi_cli_precision.py diff --git a/tests/test_advisory_feeds.py b/tests/test_advisory_feeds.py index 535c938..461660c 100644 --- a/tests/test_advisory_feeds.py +++ b/tests/test_advisory_feeds.py @@ -234,6 +234,7 @@ def test_advisory_files_are_passed_explicitly_and_feed_the_correlation(tmp_path: assert rules(findings) == [ ("app.py", "exploitable-vulnerability", 4), + ("app.py", "insecure-deserialization", 4), ("app.py", "reachable-vulnerability", 4), ("requirements.txt", "vulnerable-dependency", 1), ] @@ -326,7 +327,8 @@ def test_policies_silence_accepted_advisories_in_a_project(tmp_path: Path) -> No }, ) - assert engine.analyze_project(root, [PLUGINS]).findings == () + # The accepted advisory is silenced; the deserialization itself is not an advisory. + assert [f.rule_id for f in engine.analyze_project(root, [PLUGINS]).findings] == ["insecure-deserialization"] def test_policy_plugin_is_shipped_and_sees_the_policy() -> None: diff --git a/tests/test_correlation.py b/tests/test_correlation.py index 5616279..67848f8 100644 --- a/tests/test_correlation.py +++ b/tests/test_correlation.py @@ -110,6 +110,7 @@ def test_tainted_call_to_an_affected_api_is_exploitable(tmp_path: Path) -> None: assert rules(findings) == [ ("app.py", "exploitable-vulnerability", 4), + ("app.py", "insecure-deserialization", 4), ("app.py", "reachable-vulnerability", 4), ("requirements.txt", "vulnerable-dependency", 1), ] @@ -141,7 +142,8 @@ def test_safe_versions_produce_nothing(tmp_path: Path) -> None: tmp_path, {"requirements.txt": "pyyaml==6.0\n", "app.py": "import yaml\n\ndef load():\n return yaml.load(input())\n"}, ) - assert engine.analyze_project(root, [PLUGINS]).findings == () + # No advisory applies; untrusted ``yaml.load`` stays an insecure deserialization. + assert [f.rule_id for f in engine.analyze_project(root, [PLUGINS]).findings] == ["insecure-deserialization"] def test_correlation_crosses_functions_and_files(tmp_path: Path) -> None: @@ -211,7 +213,7 @@ def test_single_files_have_no_dependency_context() -> None: SourceManager().add_source("one.py", "import yaml\n\ndef load():\n return yaml.load(input())\n"), [PLUGINS], ) - assert findings == () + assert [f.rule_id for f in findings] == ["insecure-deserialization"] def test_correlate_is_a_pure_function() -> None: diff --git a/tests/test_detectors.py b/tests/test_detectors.py index 8f041c1..cd1c126 100644 --- a/tests/test_detectors.py +++ b/tests/test_detectors.py @@ -162,6 +162,7 @@ def test_shipped_plugins_load_with_their_manifests() -> None: by_name = {plugin.manifest.name: plugin.manifest for plugin in loaded} assert set(by_name) == { + "cli-models", "command-injection", "config-secrets", "credential-models", @@ -173,7 +174,9 @@ def test_shipped_plugins_load_with_their_manifests() -> None: "flask-models", "hardcoded-secrets", "http-client-models", + "insecure-deserialization", "missing-timeout", + "open-redirect", "path-traversal", "plaintext-credentials", "python-stdlib-models", @@ -209,9 +212,11 @@ def test_shlex_quote_sanitizes_command_arguments() -> None: assert findings == () -def test_argv_to_open_is_a_path_traversal() -> None: - findings = check("import sys\n\ndef read():\n return open(sys.argv[1])\n") +def test_stdin_to_open_is_a_path_traversal_but_argv_is_not() -> None: + findings = check("def read():\n return open(input())\n") assert rules(findings) == ["path-traversal"] + # A command-line tool is expected to open the paths it is given. + assert check("import sys\n\ndef read():\n return open(sys.argv[1])\n") == () def test_environment_to_urlopen_is_ssrf() -> None: diff --git a/tests/test_django_http_clients.py b/tests/test_django_http_clients.py index a172499..fbb1227 100644 --- a/tests/test_django_http_clients.py +++ b/tests/test_django_http_clients.py @@ -293,6 +293,8 @@ def test_httpx_client_instances_and_sessions_are_ssrf_sinks() -> None: def test_detectors_stay_generic() -> None: assert sorted(p.name for p in (PLUGINS / "security").iterdir() if p.is_dir()) == [ "command_injection", + "insecure_deserialization", + "open_redirect", "path_traversal", "plaintext_credentials", "sql_injection", diff --git a/tests/test_fastapi_cli_precision.py b/tests/test_fastapi_cli_precision.py new file mode 100644 index 0000000..271ba74 --- /dev/null +++ b/tests/test_fastapi_cli_precision.py @@ -0,0 +1,254 @@ +"""Acceptance tests for what the FastAPI and command-line repositories under +``tests-project/`` showed: ``Vulnerable_fastapi``, ``InsecureGram`` and ``yt-fts``. + +- A method on what a project function returns resolves through that function's summary + even when the call graph already derived a project symbol for it + (``get_conn().execute`` in another file was ``app.database.get_conn.execute``). +- SQL sinks read the statement only: a tainted value in the parameter tuple of a + parameterised query is not an injection. ``Sink.positions`` restricts kinds to + argument positions; unrestricted kinds keep every argument. +- An annotated parameter denotes its class, so ``db: Session = Depends(get_db)`` gives + ``db.execute`` and ``sqlalchemy.text`` their sinks. +- Two new kinds with their sinks and detectors: ``DESERIALIZATION`` (``pickle``, + ``marshal``, ``yaml``) and ``REDIRECT`` (Flask, Django, FastAPI redirects). +- Command-line entry points: ``click.command``, ``click.group`` and the commands of a + group or a Typer application, whose parameters are ``argv`` input; a function decorated + by a symbol is an instance of that symbol, so ``@cli.command()`` resolves. + +Expected to remain red until ``Sink.positions``, the two kinds, the ``models/cli`` plugin +and the two detectors 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.interprocedural import CallGraphAnalysis +from coretrace_python.plugins import discover_plugins +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.source import SourceManager +from coretrace_python.taint import SecurityModelRegistry, Sink, TaintKind + +MISSING: Exception | None = None +if "positions" not in Sink.__dataclass_fields__: + MISSING = AttributeError("Sink has no positions") +elif not hasattr(TaintKind, "DESERIALIZATION") or not hasattr(TaintKind, "REDIRECT"): + MISSING = AttributeError("TaintKind lacks DESERIALIZATION or REDIRECT") + + +@pytest.fixture(autouse=True) +def require_pass() -> None: + if MISSING is not None: + pytest.fail(f"FastAPI and CLI precision pass is not implemented yet: {MISSING}") + + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "plugins" + + +def check(text: str, name: str = "app.py") -> tuple[Finding, ...]: + return engine.check(SourceManager().add_source(name, text), [PLUGINS]) + + +def rules(findings: tuple[Finding, ...]) -> list[tuple[str, int]]: + return sorted((f.rule_id, f.span.start_line) for f in findings) + + +def project(root: Path, files: dict[str, str]) -> Path: + for relative, text in files.items(): + path = root / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + return root + + +FLASK = "import os\nimport sqlite3\nfrom flask import Flask, request, redirect\n\napp = Flask(__name__)\n\n" +FASTAPI = "from fastapi import APIRouter, Depends, Query\n\nrouter = APIRouter()\n\n" + + +# --------------------------------------------------------------------------- returned symbols + + +def test_methods_on_project_function_results_resolve_across_files(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "database.py": "import sqlite3\n\ndef get_conn():\n conn = sqlite3.connect('x.db')\n conn.row_factory = sqlite3.Row\n return conn\n", + "auth.py": ( + "from fastapi import APIRouter, Form\nfrom database import get_conn\n\nrouter = APIRouter()\n\n" + "@router.post('/login')\n" + "def login(username: str = Form(...)):\n" + " conn = get_conn()\n" + " query = f\"SELECT * FROM users WHERE username = '{username}'\"\n" + " return conn.execute(query).fetchone()\n" + ), + }, + ) + findings = engine.analyze_project(root, [PLUGINS]).findings + assert [(f.rule_id, Path(str(f.span.source_id)).name, f.span.start_line) for f in findings] == [ + ("sql-injection", "auth.py", 10) + ] + + +# --------------------------------------------------------------------------- sink positions + + +def test_sinks_restrict_kinds_to_argument_positions() -> None: + sink = Sink(SymbolId("python.db.execute"), TaintKind.SQL | TaintKind.CREDENTIAL, positions=((TaintKind.SQL, (0,)),)) + + assert sink.kinds_at(0) == TaintKind.SQL | TaintKind.CREDENTIAL + assert sink.kinds_at(1) == TaintKind.CREDENTIAL + assert sink.kinds_at(None) == TaintKind.CREDENTIAL + assert Sink(SymbolId("python.os.system"), TaintKind.COMMAND).kinds_at(3) == TaintKind.COMMAND + registry = SecurityModelRegistry() + registry.register(sink) + extended = registry.freeze().extended(Sink(SymbolId("python.db.execute"), TaintKind.ADVISORY)) + merged = extended.sink(SymbolId("python.db.execute")) + assert merged is not None and merged.kinds_at(1) == TaintKind.CREDENTIAL | TaintKind.ADVISORY + + +def test_parameterised_queries_are_not_injections() -> None: + findings = check( + FLASK + "@app.route('/u')\n" + "def user():\n" + " conn = sqlite3.connect('app.db')\n" + " conn.execute('SELECT * FROM users WHERE name = ?', (request.args['name'],))\n" + " conn.execute('SELECT * FROM users WHERE name = ' + request.args['name'])\n" + " conn.execute('SELECT 1', parameters=(request.args['name'],))\n" + ) + assert rules(findings) == [("sql-injection", 11)] + + +def test_credentials_in_the_parameter_tuple_are_still_seen() -> None: + findings = check( + "import sqlite3\n\ndef register(email, password):\n conn = sqlite3.connect('app.db')\n" + " conn.execute('INSERT INTO users VALUES (?, ?)', (email, password))\n" + ) + assert rules(findings) == [("plaintext-credential-storage", 5)] + + +# --------------------------------------------------------------------------- annotated parameters + + +def test_annotated_parameters_denote_their_class() -> None: + module = build_hir( + SourceManager().add_source( + "m.py", "from sqlalchemy.orm import Session\n\ndef f(db: Session, other):\n return db.execute('x')\n" + ) + ) + manager = engine.build_manager(module) + graph = manager.get(CallGraphAnalysis) + function = module.body[1] + assert isinstance(function, nodes.Function) + from coretrace_python.ir.ssa import SSAAnalysis + + ssa = manager.get(SSAAnalysis, function) + symbols = graph.symbols("f") + + assert symbols[ssa.parameters[0]] == SymbolId("python.sqlalchemy.orm.Session") + assert ssa.parameters[1] not in symbols + (site,) = graph.sites("f") + from coretrace_python.interprocedural import ExternalSymbol + + assert isinstance(site.target, ExternalSymbol) and site.target.symbol == SymbolId("python.sqlalchemy.orm.Session.execute") + + +def test_fastapi_dependency_sessions_and_text_are_sql_sinks() -> None: + findings = check( + FASTAPI + "from sqlalchemy.orm import Session\nfrom sqlalchemy import text\nfrom database import get_db\n\n" + "@router.get('/search')\n" + "def search(field: str = 'username', keyword: str = '', db: Session = Depends(get_db)):\n" + " db.query(text(f'LOWER({field}) LIKE :kw')).params(kw=f'%{keyword}%')\n" + " raw_sql = f\"SELECT username FROM users WHERE {field} = '{keyword}'\"\n" + " return db.execute(text(raw_sql)).fetchall()\n" + ) + assert rules(findings) == [("sql-injection", 11), ("sql-injection", 13)] + + +# --------------------------------------------------------------------------- deserialization and redirects + + +def test_new_kinds_are_part_of_all() -> None: + assert TaintKind.DESERIALIZATION & TaintKind.ALL + assert TaintKind.REDIRECT & TaintKind.ALL + + +def test_untrusted_pickle_and_yaml_are_insecure_deserialization() -> None: + findings = check( + FASTAPI + "import pickle\nimport json\nimport yaml\nfrom pydantic import BaseModel\n\n" + "class Payload(BaseModel):\n payload: str\n\n" + "@router.post('/deserialize')\n" + "async def deserialize(data: Payload, raw: str = Query('')):\n" + " obj = pickle.loads(bytes.fromhex(data.payload))\n" + " doc = yaml.load(raw)\n" + " safe = json.loads(raw)\n" + " return obj, doc, safe\n" + ) + assert rules(findings) == [("insecure-deserialization", 15), ("insecure-deserialization", 16)] + assert all(f.metadata["source_label"] == "http" for f in findings) + + +def test_user_controlled_redirects_are_open_redirects() -> None: + findings = check( + FLASK + "@app.route('/go')\n" + "def go():\n" + " return redirect(request.args['next'])\n\n" + "@app.route('/home')\n" + "def home():\n" + " return redirect('/dashboard')\n" + ) + assert rules(findings) == [("open-redirect", 9)] + findings = check( + FASTAPI + "from fastapi.responses import RedirectResponse\n\n" + "@router.get('')\n" + "def redirect_to(target: str = Query(...)):\n" + " return RedirectResponse(url=target)\n" + ) + assert rules(findings) == [("open-redirect", 9)] + + +# --------------------------------------------------------------------------- command-line entry points + + +def test_click_commands_receive_argv_input() -> None: + findings = check( + "import os\nimport click\n\n" + "@click.command()\n@click.argument('url')\n" + "def download(url):\n os.system('yt-dlp ' + url)\n\n" + "def helper(url):\n os.system(url)\n" + ) + (finding,) = findings + assert (finding.rule_id, finding.span.start_line, finding.metadata["source_label"]) == ("command-injection", 7, "argv") + + +def test_command_line_tools_may_open_the_paths_they_are_given() -> None: + assert check("import sys\nimport click\n\n@click.command()\n@click.argument('path')\ndef export(path):\n open(path).read()\n open(sys.argv[1]).read()\n") == () + + +def test_group_commands_and_typer_apps_are_entry_points() -> None: + assert rules( + check( + "import os\nimport click\n\n" + "@click.group()\ndef cli():\n pass\n\n" + "@cli.command()\n@click.argument('name')\n" + "def run(name):\n os.system(name)\n" + ) + ) == [("command-injection", 11)] + assert rules( + check("import os\nimport typer\n\napp = typer.Typer()\n\n@app.command()\ndef run(name: str):\n os.system(name)\n") + ) == [("command-injection", 8)] + + +def test_shipped_cli_models_and_detectors_load() -> None: + module = build_hir(SourceManager().add_source("empty.py", "")) + loaded = {p.manifest.name: p.manifest for p in discover_plugins(PLUGINS, engine.build_manager(module))} + + assert loaded["cli-models"].provides == ("model.cli-sources",) + assert loaded["insecure-deserialization"].provides == ("vulnerability.insecure-deserialization",) + assert loaded["open-redirect"].provides == ("vulnerability.open-redirect",) diff --git a/tests/test_parallel.py b/tests/test_parallel.py index b3fe025..f5b6d4c 100644 --- a/tests/test_parallel.py +++ b/tests/test_parallel.py @@ -161,6 +161,7 @@ def test_parallel_analysis_serves_project_plugins_and_correlation(tmp_path: Path assert rules(findings) == [ ("config.py", "reachable-vulnerability", 4), ("main.py", "exploitable-vulnerability", 4), + ("main.py", "insecure-deserialization", 4), ("requirements.txt", "vulnerable-dependency", 1), ] From 2e2321eaedb74ae024f1e6c39c427422cbbffca8 Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 4 Sep 2026 15:53:15 +0200 Subject: [PATCH 2/2] feat: resolve project-returned objects and annotated parameters, add deserialization, redirect and CLI models --- README.md | 12 +++-- plugins/models/cli/cli_models.py | 35 +++++++++++++++ plugins/models/cli/plugin.toml | 9 ++++ plugins/models/django/django_models.py | 7 ++- plugins/models/fastapi/fastapi_models.py | 2 + plugins/models/flask/flask_models.py | 2 + plugins/models/python_stdlib/python_stdlib.py | 20 ++++++++- .../models/sqlalchemy/sqlalchemy_models.py | 19 +++++--- .../insecure_deserialization.py | 17 +++++++ .../insecure_deserialization/plugin.toml | 9 ++++ .../security/open_redirect/open_redirect.py | 17 +++++++ plugins/security/open_redirect/plugin.toml | 9 ++++ .../interprocedural/callgraph.py | 41 +++++++++++++---- src/coretrace_python/taint/engine.py | 45 ++++++++++++++----- src/coretrace_python/taint/models.py | 23 ++++++++-- 15 files changed, 231 insertions(+), 36 deletions(-) create mode 100644 plugins/models/cli/cli_models.py create mode 100644 plugins/models/cli/plugin.toml create mode 100644 plugins/security/insecure_deserialization/insecure_deserialization.py create mode 100644 plugins/security/insecure_deserialization/plugin.toml create mode 100644 plugins/security/open_redirect/open_redirect.py create mode 100644 plugins/security/open_redirect/plugin.toml diff --git a/README.md b/README.md index 0833c5e..4de12af 100644 --- a/README.md +++ b/README.md @@ -96,15 +96,19 @@ 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, Django, SQLAlchemy and the Requests and httpx clients (`models/`), taint detectors for SQL -injection, command injection, path traversal, SSRF, XSS and plaintext credential storage -(`security/`), and syntactic +standard library, Flask, FastAPI, Django, SQLAlchemy, the Requests and httpx clients, and the +click and Typer command-line frameworks (`models/`), taint detectors for SQL +injection, command injection, path traversal, SSRF, XSS, insecure deserialization, open +redirects and plaintext credential storage (`security/`), and syntactic detectors for `eval`/`exec`, weak hashes, `app.run(debug=True)` and HTTP client calls without a timeout (`syntax/`). Route handlers of Flask and FastAPI receive their parameters as HTTP input, whether the application is created directly (`app = Flask(__name__)`) or by a project factory whose summary returns it (`app = create_app()`); `request.args` and its siblings are HTTP -sources. A parameter named `password` or the like is a credential wherever it appears: stored +sources; the parameters of a click or Typer command are `argv` input. SQL sinks read the +statement only, so a tainted value in the parameter tuple of a parameterised query is not an +injection, and an annotated parameter denotes its class, so `db: Session = Depends(get_db)` +gives `db.execute` its sink. A parameter named `password` or the like is a credential wherever it appears: stored in a database without passing through a hashing function it is a `plaintext-credential-storage` finding at medium confidence, since a name is a hint. Django views are undecorated, so a parameter annotated with a request class diff --git a/plugins/models/cli/cli_models.py b/plugins/models/cli/cli_models.py new file mode 100644 index 0000000..72ac98f --- /dev/null +++ b/plugins/models/cli/cli_models.py @@ -0,0 +1,35 @@ +"""Command-line entry points: the parameters of click and Typer commands are argv input.""" + +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, TaintKind + +# A command-line tool is expected to open the paths it is given: argv input carries every +# kind but PATH. +ARGV_KINDS = TaintKind.ALL & ~TaintKind.PATH + +_COMMANDS = ( + "click.command", + "click.group", + "click.command.command", + "click.group.command", + "click.group.group", + "click.Group.command", + "typer.Typer.command", + "typer.Typer.callback", +) + + +def _sym(path: str) -> SymbolId: + return SymbolId(f"python.{path}") + + +class CliModels(ModelPlugin): + name: ClassVar[str] = "cli-models" + models: ClassVar[tuple[Model, ...]] = tuple( + EntryPoint(_sym(command), "argv", ARGV_KINDS) for command in _COMMANDS + ) diff --git a/plugins/models/cli/plugin.toml b/plugins/models/cli/plugin.toml new file mode 100644 index 0000000..413a0f4 --- /dev/null +++ b/plugins/models/cli/plugin.toml @@ -0,0 +1,9 @@ +name = "cli-models" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = [] +provides = ["model.cli-sources"] + +[entrypoint] +module = "cli_models" +class = "CliModels" diff --git a/plugins/models/django/django_models.py b/plugins/models/django/django_models.py index b073eae..b0d633c 100644 --- a/plugins/models/django/django_models.py +++ b/plugins/models/django/django_models.py @@ -86,8 +86,8 @@ class DjangoModels(ModelPlugin): *(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 | TaintKind.CREDENTIAL), - Sink(_sym("django.db.connection.cursor.executemany"), TaintKind.SQL | TaintKind.CREDENTIAL), + 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,)),)), Sink(_sym("django.db.models.expressions.RawSQL"), TaintKind.SQL | TaintKind.CREDENTIAL), Sink(_sym("django.db.models.Manager.raw"), TaintKind.SQL | TaintKind.CREDENTIAL), Sink(_sym("django.db.models.QuerySet.raw"), TaintKind.SQL | TaintKind.CREDENTIAL), @@ -97,6 +97,9 @@ class DjangoModels(ModelPlugin): Sink(_sym("django.http.response.HttpResponse"), TaintKind.HTML), Sink(_sym("django.template.Template"), TaintKind.HTML), Sink(_sym("django.http.FileResponse"), TaintKind.PATH), + Sink(_sym("django.shortcuts.redirect"), TaintKind.REDIRECT), + Sink(_sym("django.http.HttpResponseRedirect"), TaintKind.REDIRECT), + Sink(_sym("django.http.HttpResponsePermanentRedirect"), TaintKind.REDIRECT), Sanitizer(_sym("django.utils.html.escape"), TaintKind.HTML), Sanitizer(_sym("django.utils.html.conditional_escape"), TaintKind.HTML), *(AuthorizationGuard(_sym(decorator), label) for decorator, label in _AUTHORIZATION_DECORATORS), diff --git a/plugins/models/fastapi/fastapi_models.py b/plugins/models/fastapi/fastapi_models.py index cf5ef79..bfec252 100644 --- a/plugins/models/fastapi/fastapi_models.py +++ b/plugins/models/fastapi/fastapi_models.py @@ -22,4 +22,6 @@ class FastApiModels(ModelPlugin): *(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), + Sink(_sym("fastapi.responses.RedirectResponse"), TaintKind.REDIRECT), + Sink(_sym("starlette.responses.RedirectResponse"), TaintKind.REDIRECT), ) diff --git a/plugins/models/flask/flask_models.py b/plugins/models/flask/flask_models.py index 761456e..e954cf6 100644 --- a/plugins/models/flask/flask_models.py +++ b/plugins/models/flask/flask_models.py @@ -37,6 +37,8 @@ class FlaskModels(ModelPlugin): Sink(_sym("flask.Response"), TaintKind.HTML), Sink(_sym("flask.Markup"), TaintKind.HTML), Sink(_sym("flask.send_file"), TaintKind.PATH), + Sink(_sym("flask.redirect"), TaintKind.REDIRECT), + Sink(_sym("werkzeug.utils.redirect"), TaintKind.REDIRECT), Sink(_sym("flask.request.files.save"), TaintKind.PATH), Sanitizer(_sym("werkzeug.utils.secure_filename"), TaintKind.PATH), Sanitizer(_sym("flask.escape"), TaintKind.HTML), diff --git a/plugins/models/python_stdlib/python_stdlib.py b/plugins/models/python_stdlib/python_stdlib.py index 92f682e..8661b47 100644 --- a/plugins/models/python_stdlib/python_stdlib.py +++ b/plugins/models/python_stdlib/python_stdlib.py @@ -18,7 +18,8 @@ class PythonStdlibModels(ModelPlugin): models: ClassVar[tuple[Model, ...]] = ( Source(_sym("builtins.input"), "stdin"), Source(_sym("sys.stdin"), "stdin"), - Source(_sym("sys.argv"), "argv"), + # A command-line tool is expected to open the paths it is given. + Source(_sym("sys.argv"), "argv", TaintKind.ALL & ~TaintKind.PATH), Source(_sym("os.environ"), "environment"), Source(_sym("subprocess.run.stdout"), "process-output"), Source(_sym("subprocess.check_output"), "process-output"), @@ -40,12 +41,27 @@ class PythonStdlibModels(ModelPlugin): Sink(_sym("shutil.rmtree"), TaintKind.PATH), Sink(_sym("urllib.request.urlopen"), TaintKind.SSRF), *( - Sink(_sym(f"sqlite3.connect{cursor}.{method}"), TaintKind.SQL | TaintKind.CREDENTIAL) + Sink( + _sym(f"sqlite3.connect{cursor}.{method}"), + TaintKind.SQL | TaintKind.CREDENTIAL, + ((TaintKind.SQL, (0,)),), + ) for cursor in ("", ".cursor") for method in ("execute", "executemany", "executescript") ), Sanitizer(_sym("shlex.quote"), TaintKind.COMMAND), Sanitizer(_sym("html.escape"), TaintKind.HTML), + Sink(_sym("pickle.loads"), TaintKind.DESERIALIZATION), + Sink(_sym("pickle.load"), TaintKind.DESERIALIZATION), + Sink(_sym("pickle.Unpickler"), TaintKind.DESERIALIZATION), + Sink(_sym("marshal.loads"), TaintKind.DESERIALIZATION), + Sink(_sym("marshal.load"), TaintKind.DESERIALIZATION), + Sink(_sym("shelve.open"), TaintKind.DESERIALIZATION), + Sink(_sym("dill.loads"), TaintKind.DESERIALIZATION), + Sink(_sym("jsonpickle.decode"), TaintKind.DESERIALIZATION), + Sink(_sym("yaml.load"), TaintKind.DESERIALIZATION), + Sink(_sym("yaml.unsafe_load"), TaintKind.DESERIALIZATION), + Sink(_sym("yaml.full_load"), TaintKind.DESERIALIZATION), Sanitizer(_sym("os.path.basename"), TaintKind.PATH), Validator(_sym("re.fullmatch"), argument=1), Validator(_sym("re.compile.fullmatch")), diff --git a/plugins/models/sqlalchemy/sqlalchemy_models.py b/plugins/models/sqlalchemy/sqlalchemy_models.py index 2a3d8d0..79e0d70 100644 --- a/plugins/models/sqlalchemy/sqlalchemy_models.py +++ b/plugins/models/sqlalchemy/sqlalchemy_models.py @@ -11,7 +11,7 @@ from coretrace_python.plugins import ModelPlugin from coretrace_python.semantic.symbols import SymbolId -from coretrace_python.taint import Model, Sink, TaintKind +from coretrace_python.taint import Model, Sanitizer, Sink, TaintKind _EXECUTORS = ( "sqlalchemy.create_engine", @@ -25,14 +25,23 @@ ) +_TEXT_MODULES = ("sqlalchemy", "sqlalchemy.sql", "sqlalchemy.sql.expression") + + 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 | TaintKind.CREDENTIAL) - for executor in _EXECUTORS - for method in ("execute", "exec_driver_sql", "scalar", "scalars") + models: ClassVar[tuple[Model, ...]] = ( + *( + Sink(_sym(f"{executor}.{method}"), TaintKind.SQL | TaintKind.CREDENTIAL, ((TaintKind.SQL, (0,)),)) + for executor in _EXECUTORS + for method in ("execute", "exec_driver_sql", "scalar", "scalars") + ), + # ``text()`` is where SQL text is committed: a sink, and its result carries no + # further SQL taint so the ``execute`` that follows does not report it twice. + *(Sink(_sym(f"{module}.text"), TaintKind.SQL) for module in _TEXT_MODULES), + *(Sanitizer(_sym(f"{module}.text"), TaintKind.SQL) for module in _TEXT_MODULES), ) diff --git a/plugins/security/insecure_deserialization/insecure_deserialization.py b/plugins/security/insecure_deserialization/insecure_deserialization.py new file mode 100644 index 0000000..6700bfb --- /dev/null +++ b/plugins/security/insecure_deserialization/insecure_deserialization.py @@ -0,0 +1,17 @@ +"""Insecure deserialization: attacker-controlled input reaching a insecure deserialization sink.""" + +from __future__ import annotations + +from typing import ClassVar + +from coretrace_python.findings import Severity +from coretrace_python.plugins import TaintDetector +from coretrace_python.taint import TaintKind + + +class InsecureDeserializationPlugin(TaintDetector): + name: ClassVar[str] = "insecure-deserialization" + rule_id: ClassVar[str] = "insecure-deserialization" + kind: ClassVar[TaintKind] = TaintKind.DESERIALIZATION + severity: ClassVar[Severity] = Severity.CRITICAL + title: ClassVar[str] = "Insecure deserialization" diff --git a/plugins/security/insecure_deserialization/plugin.toml b/plugins/security/insecure_deserialization/plugin.toml new file mode 100644 index 0000000..c90e8d1 --- /dev/null +++ b/plugins/security/insecure_deserialization/plugin.toml @@ -0,0 +1,9 @@ +name = "insecure-deserialization" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = ["taint.flows", "findings.refutation"] +provides = ["vulnerability.insecure-deserialization"] + +[entrypoint] +module = "insecure_deserialization" +class = "InsecureDeserializationPlugin" diff --git a/plugins/security/open_redirect/open_redirect.py b/plugins/security/open_redirect/open_redirect.py new file mode 100644 index 0000000..9a2c6a9 --- /dev/null +++ b/plugins/security/open_redirect/open_redirect.py @@ -0,0 +1,17 @@ +"""Open redirect: attacker-controlled input reaching a open redirect sink.""" + +from __future__ import annotations + +from typing import ClassVar + +from coretrace_python.findings import Severity +from coretrace_python.plugins import TaintDetector +from coretrace_python.taint import TaintKind + + +class OpenRedirectPlugin(TaintDetector): + name: ClassVar[str] = "open-redirect" + rule_id: ClassVar[str] = "open-redirect" + kind: ClassVar[TaintKind] = TaintKind.REDIRECT + severity: ClassVar[Severity] = Severity.MEDIUM + title: ClassVar[str] = "Open redirect" diff --git a/plugins/security/open_redirect/plugin.toml b/plugins/security/open_redirect/plugin.toml new file mode 100644 index 0000000..db3564c --- /dev/null +++ b/plugins/security/open_redirect/plugin.toml @@ -0,0 +1,9 @@ +name = "open-redirect" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = ["taint.flows", "findings.refutation"] +provides = ["vulnerability.open-redirect"] + +[entrypoint] +module = "open_redirect" +class = "OpenRedirectPlugin" diff --git a/src/coretrace_python/interprocedural/callgraph.py b/src/coretrace_python/interprocedural/callgraph.py index c26ccab..b0fe63f 100644 --- a/src/coretrace_python/interprocedural/callgraph.py +++ b/src/coretrace_python/interprocedural/callgraph.py @@ -29,7 +29,7 @@ ) from coretrace_python.ir.ssa import SSAAnalysis from coretrace_python.semantic.scopes import BindingKind, ScopeAnalysis, ScopeTable -from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.semantic.symbols import SymbolAnalysis, SymbolId, SymbolTable from coretrace_python.source import SourceSpan @@ -108,13 +108,15 @@ def callers(self, name: str) -> frozenset[str]: return self._callers.get(name, frozenset()) -def derive_symbols(function: FunctionIR) -> dict[Value, SymbolId]: +def derive_symbols( + function: FunctionIR, initial: Mapping[Value, SymbolId] | None = None +) -> dict[Value, SymbolId]: """Symbols of values: ``Symbol`` results, attributes and items 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.""" + and the values a ``with`` on such a result binds. Known functions derive nothing; + parameters only through ``initial``, their annotated classes.""" - symbols: dict[Value, SymbolId] = {} + symbols: dict[Value, SymbolId] = dict(initial or {}) changed = True while changed: changed = False @@ -143,12 +145,15 @@ def derive_symbols(function: FunctionIR) -> dict[Value, SymbolId]: def resolve_targets( - function: FunctionIR, scopes: ScopeTable, known: frozenset[str] + function: FunctionIR, + scopes: ScopeTable, + known: frozenset[str], + parameters: Mapping[Value, SymbolId] | None = None, ) -> tuple[dict[Value, Target], dict[Value, SymbolId]]: """Map every callee value of ``function`` to its target, and every symbol value.""" module = scopes.module_scope - symbols = derive_symbols(function) + symbols = derive_symbols(function, parameters) targets: dict[Value, Target] = { value: ExternalSymbol(symbol) for value, symbol in symbols.items() } @@ -165,13 +170,31 @@ def resolve_targets( return targets, symbols +def _annotated( + function: nodes.Function, ssa: FunctionIR, scopes: ScopeTable, table: SymbolTable +) -> dict[Value, SymbolId]: + """Parameters annotated with a resolvable class denote that class (``db: Session``).""" + + scope = scopes.scope_for(function) + enclosing = scope.parent if scope.parent is not None else scope.id + found: dict[Value, SymbolId] = {} + for value, parameter in zip(ssa.parameters, function.parameters, strict=True): + if parameter.annotation is None: + continue + symbol = table.resolve_expression(enclosing, parameter.annotation) + if symbol is not None: + found[value] = symbol + return found + + class CallGraphAnalysis(Analysis[CallGraph]): name: ClassVar[str] = "interprocedural.callgraph" - requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({SSAAnalysis, ScopeAnalysis}) + requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({SSAAnalysis, ScopeAnalysis, SymbolAnalysis}) @classmethod def compute(cls, ctx: AnalysisContext) -> CallGraph: scopes = ctx.get(ScopeAnalysis) + table = ctx.get(SymbolAnalysis) definitions: dict[str, nodes.Function] = {} for function in analyzable_functions(ctx.module): # A property and its setter, or a redefinition, share a qualified name; each @@ -195,7 +218,7 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph: unsupported.add(name) sites[name] = () continue - targets, symbols[name] = resolve_targets(ssa, scopes, known) + targets, symbols[name] = resolve_targets(ssa, scopes, known, _annotated(function, ssa, scopes, table)) found: list[CallSite] = [] for block in ssa.blocks: for instruction in block.instructions: diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index 076b189..81099bb 100644 --- a/src/coretrace_python/taint/engine.py +++ b/src/coretrace_python/taint/engine.py @@ -279,7 +279,12 @@ def call(self, call: Call, state: dict[Key, Taint], flows: list[TaintFlow]) -> T if project is not None: through = target.symbol.canonical_name.removeprefix("python.") return self.known(project, through, arguments, keywords, everything, call, flows, state) - return self.external(target.symbol, everything, call, state, flows) + symbol = target.symbol + if not self.modelled(symbol): + # ``get_conn().execute`` derived ``app.database.get_conn.execute``; what + # the project function returns says what ``execute`` really is. + symbol = self.returned_symbol(call) or symbol + return self.external(symbol, everything, call, state, flows) if isinstance(target, KnownFunction): summary = self.summaries.summary(target.name) return self.known(summary, target.name, arguments, keywords, everything, call, flows, state) @@ -288,6 +293,13 @@ def call(self, call: Call, state: dict[Key, Taint], flows: list[TaintFlow]) -> T return self.external(returned, everything, call, state, flows) return everything.join(state.get(call.callee, Taint.none())) + def modelled(self, symbol: SymbolId) -> bool: + return ( + self.models.sink(symbol) is not None + or self.models.sanitizer(symbol) is not None + or self.models.source_covering(symbol) is not None + ) + def external( self, symbol: SymbolId, everything: Taint, call: Call, state: Mapping[Key, Taint], flows: list[TaintFlow] ) -> Taint: @@ -295,8 +307,10 @@ def external( sink = self.models.sink(symbol) if sink is not None: - for argument in call.argument_values(): - self.report(flows, sink, self.deep(argument, state), argument, call, None, None) + for position, argument in enumerate(call.arguments): + self.report(flows, sink, self.deep(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) sanitizer = self.models.sanitizer(symbol) if sanitizer is not None: return everything.without(sanitizer.kinds) @@ -329,11 +343,7 @@ def returned_symbol(self, call: Call) -> SymbolId | None: # function may return, the one the models know about is the one that matters. for returned in sorted(summary.return_externals, key=str): candidate = returned.attribute(callee.attribute) - if ( - self.models.sink(candidate) is not None - or self.models.sanitizer(candidate) is not None - or self.models.source_covering(candidate) is not None - ): + if self.modelled(candidate): return candidate return None @@ -369,10 +379,11 @@ def mapped(deps: frozenset[int]) -> tuple[Taint, Value | None]: sink = self.models.sink(reached.symbol) if sink is None: continue - for deps in (*reached.argument_dependencies, reached.keyword_dependencies): + positions: list[int | None] = [*range(len(reached.argument_dependencies)), None] + for position, deps in zip(positions, (*reached.argument_dependencies, reached.keyword_dependencies), strict=True): taint, witness = mapped(deps) if witness is not None: - self.report(flows, sink, taint, witness, call, through, reached.location) + self.report(flows, sink, taint, witness, call, through, reached.location, position) for mutation in summary.mutations: if mutation.parameter < len(call.arguments): stored = mapped(mutation.dependencies)[0] @@ -397,8 +408,9 @@ def report( call: Call, through: str | None, sink_location: SourceSpan | None, + position: int | None = None, ) -> None: - reaching = taint.kinds & sink.kinds + reaching = taint.kinds & sink.kinds_at(position) if not reaching: return for source in sorted(taint.sources, key=lambda s: str(s.symbol)): @@ -432,6 +444,17 @@ def factory_instances( found: dict[str, tuple[SymbolId, ...]] = {} module_scope = scopes.module_scope.id for statement in module.body: + if isinstance(statement, nodes.Function) and statement.decorators: + # ``@click.group() def cli``: the function is what its decorator returns, so + # ``@cli.command()`` resolves to ``click.group.command``. + decorated = tuple( + s + for d in statement.decorators + if (s := symbols.resolve_expression(module_scope, d)) is not None + ) + if decorated: + found[statement.name] = decorated + continue if not ( isinstance(statement, nodes.Assign) and isinstance(statement.target, nodes.Name) diff --git a/src/coretrace_python/taint/models.py b/src/coretrace_python/taint/models.py index 35e8533..f9892e1 100644 --- a/src/coretrace_python/taint/models.py +++ b/src/coretrace_python/taint/models.py @@ -28,7 +28,9 @@ class TaintKind(Flag): SSRF = auto() CODE = auto() ADVISORY = auto() - ALL = SQL | COMMAND | HTML | PATH | SSRF | CODE | ADVISORY + DESERIALIZATION = auto() + REDIRECT = auto() + ALL = SQL | COMMAND | HTML | PATH | SSRF | CODE | ADVISORY | DESERIALIZATION | REDIRECT # Outside ALL on purpose: only credential-named parameters carry it, so a database # write reached by ordinary input is not a plaintext credential. CREDENTIAL = auto() @@ -49,10 +51,23 @@ class Source: @dataclass(frozen=True) class Sink: - """A callable whose arguments must not carry the given taint kinds.""" + """A callable whose arguments must not carry the given taint kinds. ``positions`` + restricts some kinds to argument positions: a SQL statement is the first argument + of ``execute``, its parameter tuple is not a statement.""" symbol: SymbolId kinds: TaintKind + positions: tuple[tuple[TaintKind, tuple[int, ...]], ...] = () + + def kinds_at(self, position: int | None) -> TaintKind: + """The kinds that must not reach the argument at ``position`` (``None`` for a + keyword or starred argument).""" + + kinds = self.kinds + for restricted, allowed in self.positions: + if position is None or position not in allowed: + kinds &= ~restricted + return kinds @dataclass(frozen=True) @@ -191,7 +206,9 @@ def extended(self, *sinks: Sink) -> ModelTable: for sink in sinks: current = merged.get(sink.symbol) merged[sink.symbol] = ( - Sink(sink.symbol, current.kinds | sink.kinds) if current is not None else sink + Sink(sink.symbol, current.kinds | sink.kinds, current.positions + sink.positions) + if current is not None + else sink ) return ModelTable( self.sources,