diff --git a/README.md b/README.md index 9938820..81bb055 100644 --- a/README.md +++ b/README.md @@ -97,7 +97,8 @@ memory. 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 and XSS (`security/`), and syntactic -detectors for `eval`/`exec` and weak hashes (`syntax/`). Route handlers of Flask and +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 diff --git a/plugins/dependency/sample_advisories/sample_advisories.py b/plugins/dependency/sample_advisories/sample_advisories.py index cd9757f..6530cae 100644 --- a/plugins/dependency/sample_advisories/sample_advisories.py +++ b/plugins/dependency/sample_advisories/sample_advisories.py @@ -61,4 +61,49 @@ class SampleAdvisories(ModelPlugin): Severity.HIGH, (_sym("flask.Flask"),), ), + Advisory( + "CVE-2024-1135", + "gunicorn", + "<22.0.0", + "HTTP request smuggling through improper Transfer-Encoding validation", + Severity.HIGH, + (), + ("GHSA-w3h3-4rj7-4ph4",), + ), + Advisory( + "CVE-2023-4863", + "opencv-python", + "<4.8.1.78", + "heap buffer overflow in the bundled libwebp when decoding a crafted WebP image", + Severity.CRITICAL, + (_sym("cv2.imread"), _sym("cv2.imdecode"), _sym("cv2.imreadmulti")), + ("GHSA-qr4w-53vh-m672",), + ), + Advisory( + "CVE-2024-34069", + "werkzeug", + "<3.0.3", + "the debugger can be tricked into executing code through a crafted domain", + Severity.HIGH, + (_sym("werkzeug.debug.DebuggedApplication"),), + ("GHSA-2g68-c3qc-8985",), + ), + Advisory( + "CVE-2023-46136", + "werkzeug", + "<3.0.1", + "a crafted multipart body exhausts CPU and memory in the form parser", + Severity.HIGH, + (_sym("werkzeug.formparser.MultiPartParser"),), + ("GHSA-hrfv-mqp8-q5rw",), + ), + Advisory( + "CVE-2026-25990", + "pillow", + "<=11.2.1", + "out-of-bounds write when loading a crafted PSD image", + Severity.HIGH, + (_sym("PIL.Image.open"),), + ("GHSA-cfh3-3jmp-rvhc",), + ), ) diff --git a/plugins/models/python_stdlib/python_stdlib.py b/plugins/models/python_stdlib/python_stdlib.py index 2dd9018..2352c0f 100644 --- a/plugins/models/python_stdlib/python_stdlib.py +++ b/plugins/models/python_stdlib/python_stdlib.py @@ -20,6 +20,10 @@ class PythonStdlibModels(ModelPlugin): Source(_sym("sys.stdin"), "stdin"), Source(_sym("sys.argv"), "argv"), Source(_sym("os.environ"), "environment"), + Source(_sym("subprocess.run.stdout"), "process-output"), + Source(_sym("subprocess.check_output"), "process-output"), + Source(_sym("subprocess.getoutput"), "process-output"), + Source(_sym("subprocess.Popen.communicate"), "process-output"), Sink(_sym("os.system"), TaintKind.COMMAND), Sink(_sym("os.popen"), TaintKind.COMMAND), Sink(_sym("subprocess.run"), TaintKind.COMMAND), diff --git a/plugins/secrets/hardcoded_secrets/hardcoded_secrets.py b/plugins/secrets/hardcoded_secrets/hardcoded_secrets.py index 4d9e064..ae87981 100644 --- a/plugins/secrets/hardcoded_secrets/hardcoded_secrets.py +++ b/plugins/secrets/hardcoded_secrets/hardcoded_secrets.py @@ -20,6 +20,11 @@ class HardcodedSecrets(SecretDetector): SecretPattern("jwt", r"\beyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\b"), SecretPattern("sendgrid", r"\bSG\.[A-Za-z0-9_-]{22}\.[A-Za-z0-9_-]{43}\b"), SecretPattern("twilio", r"\bSK[0-9a-fA-F]{32}\b"), + SecretPattern( + "url", + r"://[^/\s:@'\"]+:[^/\s@'\"]+@" + r"|[?&](?:password|passwd|pwd|token|api_key|apikey|secret|access_key)=[^&\s'\"]{3,}", + ), ) credential_names: ClassVar[tuple[str, ...]] = ( "password", diff --git a/plugins/syntax/flask_debug/flask_debug.py b/plugins/syntax/flask_debug/flask_debug.py new file mode 100644 index 0000000..137d2ff --- /dev/null +++ b/plugins/syntax/flask_debug/flask_debug.py @@ -0,0 +1,63 @@ +"""``app.run(debug=True)``: the Werkzeug debugger executes code on any exception.""" + +from __future__ import annotations + +from collections.abc import Iterator, Sequence +from typing import ClassVar + +from coretrace_python.analysis import AnyAnalysis +from coretrace_python.findings import Confidence, Finding, Severity +from coretrace_python.hir import nodes +from coretrace_python.hir.visitors import Node, children +from coretrace_python.plugins import Plugin, PluginContext +from coretrace_python.semantic.scopes import ScopeAnalysis, ScopeId, ScopeTable +from coretrace_python.semantic.symbols import SymbolAnalysis, SymbolId + +RUN = SymbolId("python.flask.Flask.run") + + +class FlaskDebugPlugin(Plugin): + name: ClassVar[str] = "flask-debug" + requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({ScopeAnalysis, SymbolAnalysis}) + + def analyze(self, ctx: PluginContext) -> Sequence[Finding]: + scopes = ctx.get(ScopeAnalysis) + symbols = ctx.get(SymbolAnalysis) + findings: list[Finding] = [] + for call, scope, function in _calls(ctx.module.body, scopes, scopes.module_scope.id, None): + if symbols.resolve_expression(scope, call.callee) != RUN: + continue + if not any( + k.name == "debug" and isinstance(k.value, nodes.Constant) and k.value.value is True + for k in call.keywords + ): + continue + findings.append( + Finding( + "debug-enabled", + "Flask debugger enabled: app.run(debug=True) exposes an interactive " + "debugger able to execute Python on any exception", + Severity.HIGH, + Confidence.HIGH, + call.span, + function, + {"symbol": str(RUN)}, + ) + ) + return findings + + +def _calls( + body: Sequence[Node], scopes: ScopeTable, scope: ScopeId, function: str | None +) -> Iterator[tuple[nodes.Call, ScopeId, str | None]]: + for node in body: + if isinstance(node, nodes.Function): + yield from _calls(node.body, scopes, scopes.scope_for(node).id, node.name) + elif isinstance(node, nodes.Class): + yield from _calls(node.body, scopes, scopes.scope_for(node).id, None) + elif isinstance(node, nodes.Lambda | nodes.Comprehension): + continue + else: + if isinstance(node, nodes.Call): + yield node, scope, function + yield from _calls(list(children(node)), scopes, scope, function) diff --git a/plugins/syntax/flask_debug/plugin.toml b/plugins/syntax/flask_debug/plugin.toml new file mode 100644 index 0000000..1bd4a38 --- /dev/null +++ b/plugins/syntax/flask_debug/plugin.toml @@ -0,0 +1,9 @@ +name = "flask-debug" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = ["semantic.scopes", "semantic.symbols"] +provides = ["syntax.flask-debug"] + +[entrypoint] +module = "flask_debug" +class = "FlaskDebugPlugin" diff --git a/plugins/syntax/missing_timeout/missing_timeout.py b/plugins/syntax/missing_timeout/missing_timeout.py new file mode 100644 index 0000000..f1a585a --- /dev/null +++ b/plugins/syntax/missing_timeout/missing_timeout.py @@ -0,0 +1,51 @@ +"""HTTP client calls without a ``timeout``: a slow or silent peer blocks the caller.""" + +from __future__ import annotations + +from collections.abc import Sequence +from typing import ClassVar + +from coretrace_python.analysis import AnyAnalysis +from coretrace_python.findings import Confidence, Finding, Severity +from coretrace_python.interprocedural import CallGraphAnalysis +from coretrace_python.ir.model import Call +from coretrace_python.ir.ssa import SSAAnalysis +from coretrace_python.plugins import Plugin, PluginContext +from coretrace_python.semantic.symbols import SymbolId + +_METHODS = ("get", "post", "put", "patch", "delete", "head", "options", "request") +_CALLERS = ("requests", "requests.Session", "requests.api", "httpx", "httpx.Client", "httpx.AsyncClient") +HTTP_CALLS = frozenset(SymbolId(f"python.{caller}.{method}") for caller in _CALLERS for method in _METHODS) + + +class MissingTimeoutPlugin(Plugin): + name: ClassVar[str] = "missing-timeout" + requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({SSAAnalysis, CallGraphAnalysis}) + + def analyze(self, ctx: PluginContext) -> Sequence[Finding]: + graph = ctx.get(CallGraphAnalysis) + findings: list[Finding] = [] + for function in ctx.functions(): + symbols = graph.symbols(graph.name_of(function)) + for block in ctx.get(SSAAnalysis, function).blocks: + for instruction in block.instructions: + if not isinstance(instruction, Call): + continue + symbol = symbols.get(instruction.callee) + if symbol is None or symbol not in HTTP_CALLS: + continue + if any(name == "timeout" for name, _ in instruction.keywords): + continue + findings.append( + Finding( + "missing-timeout", + f"{symbol} is called without a timeout; a slow or silent peer " + "blocks this code indefinitely", + Severity.LOW, + Confidence.HIGH, + instruction.location, + function.name, + {"symbol": str(symbol)}, + ) + ) + return findings diff --git a/plugins/syntax/missing_timeout/plugin.toml b/plugins/syntax/missing_timeout/plugin.toml new file mode 100644 index 0000000..1db8e8e --- /dev/null +++ b/plugins/syntax/missing_timeout/plugin.toml @@ -0,0 +1,9 @@ +name = "missing-timeout" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = ["ir.ssa", "interprocedural.callgraph"] +provides = ["syntax.missing-timeout"] + +[entrypoint] +module = "missing_timeout" +class = "MissingTimeoutPlugin" diff --git a/src/coretrace_python/plugins/secrets.py b/src/coretrace_python/plugins/secrets.py index 675a554..084c856 100644 --- a/src/coretrace_python/plugins/secrets.py +++ b/src/coretrace_python/plugins/secrets.py @@ -71,6 +71,19 @@ def _walk(node: Node, name: str | None, function: str | None) -> Iterator[Litera if isinstance(node, nodes.Keyword): yield from _walk(node.value, node.name, function) return + if isinstance(node, nodes.Parameter): + if node.default is not None: + yield from _walk(node.default, node.name, function) + return + if isinstance(node, nodes.Call) and _is_environment_lookup(node): + # ``os.getenv("APP_TOKEN", "fallback")``: the fallback is bound to the variable name. + variable, fallback = node.arguments[0], node.arguments[1] + assert isinstance(variable, nodes.Constant) and isinstance(variable.value, str) + yield from _walk(variable, None, function) + yield from _walk(fallback, variable.value, function) + for keyword in node.keywords: + yield from _walk(keyword, None, function) + return if isinstance(node, nodes.Dict): for key, value in node.items: # A constant key names the value; it is not a value itself. @@ -86,6 +99,17 @@ def _walk(node: Node, name: str | None, function: str | None) -> Iterator[Litera yield from _walk(child, None, function) +def _is_environment_lookup(call: nodes.Call) -> bool: + callee = call.callee + return ( + isinstance(callee, nodes.Attribute) + and callee.name in ("getenv", "get") + and len(call.arguments) >= 2 + and isinstance(call.arguments[0], nodes.Constant) + and isinstance(call.arguments[0].value, str) + ) + + def _bound_name(target: nodes.Target) -> str | None: """The name an assignment binds: ``name``, ``obj.attr`` or ``mapping['key']``.""" diff --git a/tests/test_audit_detectors.py b/tests/test_audit_detectors.py new file mode 100644 index 0000000..06a5f65 --- /dev/null +++ b/tests/test_audit_detectors.py @@ -0,0 +1,217 @@ +"""Acceptance tests for the detectors the ``tests-project/SECURITY_AUDIT.md`` review asked +for, each traced to a line of one of the twelve repositories. + +- Advisories for Gunicorn, OpenCV, Werkzeug and Pillow with their affected APIs, so the + vulnerable requirements and the reachable calls (``cv2.imread``, ``PIL.Image.open``) + are reported (uploadbox, Flask-Image-Editing-Website, MiniBookApiServer). +- ``app.run(debug=True)``, at module level too (Flask-Image-Editing-Website, Spam-Detection). +- HTTP client calls without a ``timeout`` (Chatbot). +- The output of a subprocess is an untrusted source: ``sh`` fed with ``curl``'s stdout + is a command injection (Chatbot ``setup.py``). +- Credentials inside a URL literal, in a parameter default too (Pylinkit). +- The fallback value of ``os.getenv("..._TOKEN", "literal")`` is a hardcoded credential + (tag-collector-server). + +Expected to remain red until the plugins ``syntax/flask_debug`` and +``syntax/missing_timeout`` and the model and pattern additions exist. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.findings import Confidence, Finding, Severity +from coretrace_python.frontend import build_hir +from coretrace_python.plugins import discover_plugins +from coretrace_python.plugins.secrets import literals +from coretrace_python.source import SourceManager + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "plugins" + + +@pytest.fixture(autouse=True) +def require_plugins() -> None: + module = build_hir(SourceManager().add_source("empty.py", "")) + names = {p.manifest.name for p in discover_plugins(PLUGINS, engine.build_manager(module))} + missing = {"flask-debug", "missing-timeout"} - names + if missing: + pytest.fail(f"audit detectors are not implemented yet: {sorted(missing)}") + + +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 + + +def project_rules(root: Path) -> list[tuple[str, str, int]]: + findings = engine.analyze_project(root, [PLUGINS]).findings + return sorted((f.rule_id, Path(str(f.span.source_id)).name, f.span.start_line) for f in findings) + + +# --------------------------------------------------------------------------- advisories + + +def test_audit_cited_advisories_are_shipped_with_affected_apis() -> None: + module = build_hir(SourceManager().add_source("empty.py", "")) + loaded = {p.manifest.name: p for p in discover_plugins(PLUGINS, engine.build_manager(module))} + advisories = {a.id: a for a in loaded["sample-advisories"].plugin.advisories} + + assert advisories["CVE-2024-1135"].package == "gunicorn" and advisories["CVE-2024-1135"].vulnerable == "<22.0.0" + assert advisories["CVE-2023-4863"].package == "opencv-python" + assert "python.cv2.imread" in {str(s) for s in advisories["CVE-2023-4863"].affected_symbols} + assert advisories["CVE-2024-34069"].package == "werkzeug" and advisories["CVE-2024-34069"].vulnerable == "<3.0.3" + assert advisories["CVE-2023-46136"].package == "werkzeug" + assert advisories["CVE-2026-25990"].package == "pillow" + assert "python.PIL.Image.open" in {str(s) for s in advisories["CVE-2026-25990"].affected_symbols} + + +def test_image_processing_with_a_vulnerable_opencv_is_reachable(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "requirements.txt": "Flask==2.3.2\nopencv-python==4.8.0.74\nWerkzeug==2.3.6\n", + "main.py": "import cv2\n\ndef process(filename):\n return cv2.imread(filename)\n", + }, + ) + assert project_rules(root) == [ + ("reachable-vulnerability", "main.py", 4), + ("vulnerable-dependency", "requirements.txt", 2), + ("vulnerable-dependency", "requirements.txt", 3), + ("vulnerable-dependency", "requirements.txt", 3), + ] + + +def test_pillow_and_gunicorn_requirements_are_reported(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "requirements.txt": "pillow==11.2.1\ngunicorn==20.0.3\n", + "res.py": "from PIL import Image\n\ndef load(data):\n return Image.open(data)\n", + }, + ) + assert project_rules(root) == [ + ("reachable-vulnerability", "res.py", 4), + ("vulnerable-dependency", "requirements.txt", 1), + ("vulnerable-dependency", "requirements.txt", 2), + ] + + +# --------------------------------------------------------------------------- flask debug + + +def test_flask_debug_is_reported_at_module_level_and_in_functions() -> None: + findings = check( + "from flask import Flask\n\napp = Flask(__name__)\n\n" + "def main():\n app.run(host='0.0.0.0', debug=True)\n\n" + "app.run(debug=True)\n" + "app.run(debug=False)\n" + "app.run()\n" + ) + debug = [f for f in findings if f.rule_id == "debug-enabled"] + assert [f.span.start_line for f in debug] == [6, 8] + assert debug[0].severity is Severity.HIGH and debug[0].confidence is Confidence.HIGH + assert debug[0].function == "main" and debug[1].function is None + assert "debugger" in debug[0].message + + +def test_debug_on_other_objects_is_ignored() -> None: + assert check("class App:\n def run(self, debug):\n pass\n\napp = App()\napp.run(debug=True)\n") == () + + +# --------------------------------------------------------------------------- timeouts + + +def test_http_calls_without_a_timeout_are_reported() -> None: + findings = check( + "import requests\nimport httpx\n\nsession = requests.Session()\n\n" + "def fetch(url):\n" + " requests.get(url)\n" + " requests.post(url, timeout=5)\n" + " session.get(url)\n" + " httpx.get(url, timeout=None)\n" + " client = httpx.Client()\n" + " client.get(url)\n" + ) + timeouts = [f for f in findings if f.rule_id == "missing-timeout"] + assert [(f.span.start_line, f.metadata["symbol"]) for f in timeouts] == [ + (7, "python.requests.get"), + (9, "python.requests.Session.get"), + (12, "python.httpx.Client.get"), + ] + assert timeouts[0].severity is Severity.LOW + + +# --------------------------------------------------------------------------- process output + + +def test_subprocess_output_is_an_untrusted_source() -> None: + findings = check( + "import subprocess\n\n" + "def install():\n" + " result = subprocess.run(['curl', '-fsSL', 'https://example.com/install.sh'], capture_output=True)\n" + " subprocess.run(['sh'], input=result.stdout, check=True)\n" + ) + (finding,) = [f for f in findings if f.rule_id == "command-injection"] + assert finding.span.start_line == 5 + assert finding.metadata["source_label"] == "process-output" + + +def test_check_output_is_a_source_too() -> None: + findings = check("import os\nimport subprocess\n\ndef f():\n os.system(subprocess.check_output(['ls']))\n") + assert rules(findings) == [("command-injection", 5)] + + +# --------------------------------------------------------------------------- secrets + + +def test_credentials_in_urls_are_secrets_including_parameter_defaults() -> None: + findings = check( + "import requests\n\n" + "def sync(url='http://host.example/api?login=LIAM&password=LIAM&application=x'):\n" + " return requests.get(url, timeout=3)\n\n" + "DB = 'postgres://admin:hunter2@db.example/app'\n" + "SAFE = 'https://host.example/api?page=2'\n" + ) + secrets = [f for f in findings if f.rule_id == "hardcoded-secret"] + assert [(f.span.start_line, f.metadata["provider"], f.metadata["name"]) for f in secrets] == [ + (3, "url", "url"), + (6, "url", "DB"), + ] + assert all("hunter2" not in f.message and "LIAM" not in f.message for f in secrets) + + +def test_environment_fallbacks_are_credentials() -> None: + module = build_hir( + SourceManager().add_source( + "conf.py", + "import os\n" + "TOKEN = os.getenv('INFLUXDB_INIT_ADMIN_TOKEN', 'mon-token-secret')\n" + "URL = os.environ.get('INFLUXDB_INIT_URL', 'http://localhost:8086')\n" + "NAME = os.getenv('APP_NAME')\n", + ) + ) + found = [(value, name) for value, name, _, _ in literals(module)] + assert found == [ + ("INFLUXDB_INIT_ADMIN_TOKEN", None), + ("mon-token-secret", "INFLUXDB_INIT_ADMIN_TOKEN"), + ("INFLUXDB_INIT_URL", None), + ("http://localhost:8086", "INFLUXDB_INIT_URL"), + ("APP_NAME", None), + ] + findings = check("import os\nTOKEN = os.getenv('INFLUXDB_INIT_ADMIN_TOKEN', 'mon-token-secret')\n") + assert [(f.rule_id, f.metadata["name"]) for f in findings] == [("hardcoded-credential", "INFLUXDB_INIT_ADMIN_TOKEN")] diff --git a/tests/test_detectors.py b/tests/test_detectors.py index 92bee2c..aa41e9d 100644 --- a/tests/test_detectors.py +++ b/tests/test_detectors.py @@ -167,9 +167,11 @@ def test_shipped_plugins_load_with_their_manifests() -> None: "dependency-policy", "django-models", "fastapi-models", + "flask-debug", "flask-models", "hardcoded-secrets", "http-client-models", + "missing-timeout", "path-traversal", "python-stdlib-models", "reachable-vulnerability", diff --git a/tests/test_django_http_clients.py b/tests/test_django_http_clients.py index eee857a..cf483a5 100644 --- a/tests/test_django_http_clients.py +++ b/tests/test_django_http_clients.py @@ -266,7 +266,7 @@ def test_requests_with_user_url_is_an_ssrf_and_its_response_is_input() -> None: "import os\nimport requests\nfrom flask import Flask, request\n\napp = Flask(__name__)\n\n" "@app.route('/fetch')\n" "def fetch():\n" - " response = requests.get(request.args['url'])\n" + " response = requests.get(request.args['url'], timeout=5)\n" " os.system(response.text)\n" ) # The response carries two provenances: the remote server's answer, and the URL the @@ -282,9 +282,9 @@ def test_httpx_client_instances_and_sessions_are_ssrf_sinks() -> None: "import httpx\nimport requests\n\n" "client = httpx.Client()\nsession = requests.Session()\n\n" "def fetch():\n" - " client.post(input())\n" - " session.get(input())\n" - " httpx.get('https://example.com')\n" + " client.post(input(), timeout=5)\n" + " session.get(input(), timeout=5)\n" + " httpx.get('https://example.com', timeout=5)\n" ) assert rules(findings) == ["ssrf", "ssrf"] assert sorted(f.span.start_line for f in findings) == [8, 9]