From b40c16078a1ccfaa1ab59c8e71649bad2d71b3be Mon Sep 17 00:00:00 2001 From: Hugo Date: Sun, 6 Sep 2026 13:26:30 +0200 Subject: [PATCH 1/2] test: define five syntactic rules --- tests/test_syntactic_rules.py | 80 +++++++++++++++++++++++++++++++++++ 1 file changed, 80 insertions(+) create mode 100644 tests/test_syntactic_rules.py diff --git a/tests/test_syntactic_rules.py b/tests/test_syntactic_rules.py new file mode 100644 index 0000000..ebe0ffa --- /dev/null +++ b/tests/test_syntactic_rules.py @@ -0,0 +1,80 @@ +"""Acceptance tests for five syntactic rules (issue #70). + +Cheap checks in the spirit of ``dangerous-eval``: disabled certificate verification, +XML parsers that expand entities, archive extraction without member checks, ``mktemp``, +and ``random`` used to build secrets. Each is a call to a known symbol, resolved through +imports, aliases and call chains (``tarfile.open(p).extractall()``), with a keyword check +where the same call has a safe form (``verify=True``, ``filter="data"``). + +Expected to remain red until the five plugins 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.source import SourceManager + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "src" / "coretrace_python" / "bundled" +NAMES = ("insecure_tls", "unsafe_xml", "unsafe_archive", "insecure_tempfile", "weak_random") +MISSING = [n for n in NAMES if not (PLUGINS / "syntax" / n / "plugin.toml").is_file()] or None + + +@pytest.fixture(autouse=True) +def require_rules() -> None: + if MISSING is not None: + pytest.fail(f"syntactic rule plugins are not written yet: {MISSING}") + + +def check(text: str) -> tuple[Finding, ...]: + return engine.check(SourceManager().add_source("m.py", text), [PLUGINS]) + + +def rules(findings: tuple[Finding, ...]) -> list[str]: + return sorted(f.rule_id for f in findings if f.rule_id != "missing-timeout") + + +def test_disabled_certificate_verification() -> None: + (finding,) = check("import requests\n\ndef f(u):\n return requests.get(u, verify=False, timeout=3)\n") + assert finding.rule_id == "insecure-tls" and finding.severity is Severity.HIGH + assert rules(check("import httpx\n\ndef f(u):\n with httpx.Client(verify=False) as c:\n return c.get(u, timeout=3)\n")) == ["insecure-tls"] + assert rules(check("import ssl\n\ndef f():\n return ssl._create_unverified_context()\n")) == ["insecure-tls"] + assert check("import requests\n\ndef f(u, v):\n requests.get(u, verify=True, timeout=3)\n requests.get(u, verify=v, timeout=3)\n") == () + + +def test_xml_parsers_that_expand_entities() -> None: + text = ( + "import xml.etree.ElementTree as ET\nfrom xml.dom import minidom\nimport xml.sax\nfrom lxml import etree\n\n" + "def f(data):\n ET.fromstring(data)\n minidom.parseString(data)\n xml.sax.parseString(data, None)\n etree.fromstring(data)\n" + ) + assert rules(check(text)) == ["unsafe-xml"] * 4 + assert check("import defusedxml.ElementTree as ET\n\ndef f(data):\n ET.fromstring(data)\n") == () + + +def test_archive_extraction_without_member_checks() -> None: + text = "import tarfile\nimport zipfile\nimport shutil\n\ndef f(p):\n tarfile.open(p).extractall('/tmp/x')\n zipfile.ZipFile(p).extractall('/tmp/x')\n shutil.unpack_archive(p, '/tmp/x')\n" + assert rules(check(text)) == ["unsafe-archive-extraction"] * 3 + text = "import tarfile\n\ndef f(p):\n with tarfile.open(p) as tar:\n tar.extractall('/tmp/x', filter='data')\n" + assert check(text) == () + text = "import tarfile\n\ndef f(p):\n with tarfile.open(p) as tar:\n tar.extractall('/tmp/x')\n" + assert rules(check(text)) == ["unsafe-archive-extraction"] + + +def test_insecure_temporary_files() -> None: + assert rules(check("import tempfile\n\ndef f():\n return tempfile.mktemp()\n")) == ["insecure-temp-file"] + assert check("import tempfile\n\ndef f():\n return tempfile.mkstemp()\n") == () + + +def test_random_used_for_secrets() -> None: + text = "import random\nimport string\n\ndef make():\n token = ''.join(random.choice(string.ascii_letters) for _ in range(32))\n return token\n" + (finding,) = check(text) + assert finding.rule_id == "weak-random" and finding.confidence is Confidence.MEDIUM + assert rules(check("import random\n\ndef generate_api_key():\n return random.getrandbits(128)\n")) == ["weak-random"] + assert rules(check("import random\n\nclass S:\n def reset(self):\n self.session_id = random.randint(0, 10**9)\n")) == ["weak-random"] + assert check("import random\n\ndef shuffle(cards):\n random.shuffle(cards)\n delay = random.uniform(1, 3)\n return delay\n") == () + assert check("import secrets\n\ndef make():\n token = secrets.token_hex(16)\n return token\n") == () From c0f275a620559bd1d55ce825d968b7df3615f6fb Mon Sep 17 00:00:00 2001 From: Hugo Date: Sun, 6 Sep 2026 13:26:30 +0200 Subject: [PATCH 2/2] feat: add insecure TLS, unsafe XML, archive extraction, mktemp and weak random rules --- docs/usage.md | 12 ++- .../insecure_tempfile/insecure_tempfile.py | 25 ++++++ .../syntax/insecure_tempfile/plugin.toml | 9 ++ .../syntax/insecure_tls/insecure_tls.py | 40 +++++++++ .../bundled/syntax/insecure_tls/plugin.toml | 9 ++ .../bundled/syntax/unsafe_archive/plugin.toml | 9 ++ .../syntax/unsafe_archive/unsafe_archive.py | 36 ++++++++ .../bundled/syntax/unsafe_xml/plugin.toml | 9 ++ .../bundled/syntax/unsafe_xml/unsafe_xml.py | 35 ++++++++ .../bundled/syntax/weak_random/plugin.toml | 9 ++ .../bundled/syntax/weak_random/weak_random.py | 86 +++++++++++++++++++ src/coretrace_python/plugins/detectors.py | 33 ++++--- tests/regression/expected/httpie.json | 6 ++ tests/regression/expected/luigi.json | 42 +++++++++ tests/regression/expected/pygoat.json | 24 ++++++ tests/regression/expected/yt-fts.json | 12 +++ tests/test_detectors.py | 5 ++ tests/test_docs.py | 2 +- tests/test_packaging.py | 4 +- 19 files changed, 391 insertions(+), 16 deletions(-) create mode 100644 src/coretrace_python/bundled/syntax/insecure_tempfile/insecure_tempfile.py create mode 100644 src/coretrace_python/bundled/syntax/insecure_tempfile/plugin.toml create mode 100644 src/coretrace_python/bundled/syntax/insecure_tls/insecure_tls.py create mode 100644 src/coretrace_python/bundled/syntax/insecure_tls/plugin.toml create mode 100644 src/coretrace_python/bundled/syntax/unsafe_archive/plugin.toml create mode 100644 src/coretrace_python/bundled/syntax/unsafe_archive/unsafe_archive.py create mode 100644 src/coretrace_python/bundled/syntax/unsafe_xml/plugin.toml create mode 100644 src/coretrace_python/bundled/syntax/unsafe_xml/unsafe_xml.py create mode 100644 src/coretrace_python/bundled/syntax/weak_random/plugin.toml create mode 100644 src/coretrace_python/bundled/syntax/weak_random/weak_random.py diff --git a/docs/usage.md b/docs/usage.md index b46a7bd..c95b1c7 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -126,6 +126,11 @@ medium confidence. The verdict and its evidence are in the finding's metadata. | `weak-crypto` | `hashlib.md5`, `hashlib.sha1` and other broken algorithms. | | `debug-enabled` | `app.run(debug=True)` on a Flask application. | | `missing-timeout` | A Requests or httpx call without a `timeout`. | +| `insecure-tls` | `verify=False` on a Requests or httpx call, or `ssl._create_unverified_context()`. | +| `unsafe-xml` | A standard-library or lxml XML parser, which expands entities; `defusedxml` is the safe replacement. | +| `unsafe-archive-extraction` | `tarfile` or `zipfile` extraction, or `shutil.unpack_archive`, without a member `filter`. | +| `insecure-temp-file` | `tempfile.mktemp`, which names a file without creating it. | +| `weak-random` | A `random` function whose result is assigned to a credential-like name, or produced by a function named like one. Medium confidence. | ### Secrets @@ -310,13 +315,14 @@ real 45 000-line project is exercised on every change. ## Plugins -The package ships 30 plugins, loaded by default. Security models for the standard library +The package ships 35 plugins, loaded by default. Security models for the standard library and the supported frameworks: `python-stdlib-models`, `flask-models`, `fastapi-models`, `django-models`, `aiohttp-models`, `tornado-models`, `bottle-models`, `sqlalchemy-models`, `db-driver-models` (aiopg, asyncpg, psycopg2, PyMySQL), `http-client-models`, `credential-models` and `cli-models`. Taint detectors: `command-injection`, `sql-injection`, `path-traversal`, `ssrf`, `xss`, `insecure-deserialization`, `open-redirect` and `plaintext-credentials`. -Dangerous API detectors: `dangerous-eval`, `weak-crypto`, `flask-debug` and -`missing-timeout`. Secret scanners: `hardcoded-secrets` for Python sources and +Dangerous API detectors: `dangerous-eval`, `weak-crypto`, `flask-debug`, +`missing-timeout`, `insecure-tls`, `unsafe-xml`, `unsafe-archive-extraction`, +`insecure-temp-file` and `weak-random`. Secret scanners: `hardcoded-secrets` for Python sources and `config-secrets` for configuration files. Dependency checks: `sample-advisories`, `vulnerable-dependency`, `reachable-vulnerability` and `dependency-policy`. diff --git a/src/coretrace_python/bundled/syntax/insecure_tempfile/insecure_tempfile.py b/src/coretrace_python/bundled/syntax/insecure_tempfile/insecure_tempfile.py new file mode 100644 index 0000000..217155d --- /dev/null +++ b/src/coretrace_python/bundled/syntax/insecure_tempfile/insecure_tempfile.py @@ -0,0 +1,25 @@ +"""``tempfile.mktemp`` returns a name without creating the file: another process can +create it first. ``mkstemp``, ``NamedTemporaryFile`` and ``TemporaryDirectory`` are safe.""" + +from __future__ import annotations + +from typing import ClassVar + +from coretrace_python.analysis import AnyAnalysis +from coretrace_python.findings import Severity +from coretrace_python.interprocedural import CallGraphAnalysis +from coretrace_python.ir.ssa import SSAAnalysis +from coretrace_python.plugins import SymbolCallDetector +from coretrace_python.semantic.symbols import SymbolId + + +class InsecureTempFilePlugin(SymbolCallDetector): + name: ClassVar[str] = "insecure-temp-file" + rule_id: ClassVar[str] = "insecure-temp-file" + requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({SSAAnalysis, CallGraphAnalysis}) + symbols: ClassVar[frozenset[SymbolId]] = frozenset({SymbolId("python.tempfile.mktemp")}) + severity: ClassVar[Severity] = Severity.MEDIUM + message_template: ClassVar[str] = ( + "call to {symbol} names a temporary file without creating it; use mkstemp or " + "NamedTemporaryFile" + ) diff --git a/src/coretrace_python/bundled/syntax/insecure_tempfile/plugin.toml b/src/coretrace_python/bundled/syntax/insecure_tempfile/plugin.toml new file mode 100644 index 0000000..3908f60 --- /dev/null +++ b/src/coretrace_python/bundled/syntax/insecure_tempfile/plugin.toml @@ -0,0 +1,9 @@ +name = "insecure-temp-file" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = ["interprocedural.callgraph", "ir.ssa"] +provides = ["vulnerability.insecure-temp-file"] + +[entrypoint] +module = "insecure_tempfile" +class = "InsecureTempFilePlugin" diff --git a/src/coretrace_python/bundled/syntax/insecure_tls/insecure_tls.py b/src/coretrace_python/bundled/syntax/insecure_tls/insecure_tls.py new file mode 100644 index 0000000..d602203 --- /dev/null +++ b/src/coretrace_python/bundled/syntax/insecure_tls/insecure_tls.py @@ -0,0 +1,40 @@ +"""Disabled certificate verification: ``verify=False`` on an HTTP client call, or an +``ssl`` context built without verification, lets any peer impersonate the server.""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import ClassVar + +from coretrace_python.analysis import AnyAnalysis +from coretrace_python.findings import Severity +from coretrace_python.interprocedural import CallGraphAnalysis +from coretrace_python.ir.model import Call, Constant, Instruction, Value +from coretrace_python.ir.ssa import SSAAnalysis +from coretrace_python.plugins import SymbolCallDetector +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 +) | frozenset(SymbolId(f"python.{client}") for client in ("httpx.Client", "httpx.AsyncClient")) +UNVERIFIED = frozenset({SymbolId("python.ssl._create_unverified_context")}) + + +class InsecureTlsPlugin(SymbolCallDetector): + name: ClassVar[str] = "insecure-tls" + rule_id: ClassVar[str] = "insecure-tls" + requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({SSAAnalysis, CallGraphAnalysis}) + symbols: ClassVar[frozenset[SymbolId]] = HTTP_CALLS | UNVERIFIED + severity: ClassVar[Severity] = Severity.HIGH + message_template: ClassVar[str] = "call to {symbol} disables certificate verification" + + def accepts(self, call: Call, symbol: SymbolId, defs: Mapping[Value, Instruction]) -> bool: + if symbol in UNVERIFIED: + return True + for name, value in call.keywords: + if name == "verify": + defined = defs.get(value) + return isinstance(defined, Constant) and defined.value is False + return False diff --git a/src/coretrace_python/bundled/syntax/insecure_tls/plugin.toml b/src/coretrace_python/bundled/syntax/insecure_tls/plugin.toml new file mode 100644 index 0000000..eb337ad --- /dev/null +++ b/src/coretrace_python/bundled/syntax/insecure_tls/plugin.toml @@ -0,0 +1,9 @@ +name = "insecure-tls" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = ["interprocedural.callgraph", "ir.ssa"] +provides = ["vulnerability.insecure-tls"] + +[entrypoint] +module = "insecure_tls" +class = "InsecureTlsPlugin" diff --git a/src/coretrace_python/bundled/syntax/unsafe_archive/plugin.toml b/src/coretrace_python/bundled/syntax/unsafe_archive/plugin.toml new file mode 100644 index 0000000..0f83289 --- /dev/null +++ b/src/coretrace_python/bundled/syntax/unsafe_archive/plugin.toml @@ -0,0 +1,9 @@ +name = "unsafe-archive-extraction" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = ["interprocedural.callgraph", "ir.ssa"] +provides = ["vulnerability.unsafe-archive-extraction"] + +[entrypoint] +module = "unsafe_archive" +class = "UnsafeArchivePlugin" diff --git a/src/coretrace_python/bundled/syntax/unsafe_archive/unsafe_archive.py b/src/coretrace_python/bundled/syntax/unsafe_archive/unsafe_archive.py new file mode 100644 index 0000000..097e598 --- /dev/null +++ b/src/coretrace_python/bundled/syntax/unsafe_archive/unsafe_archive.py @@ -0,0 +1,36 @@ +"""Archive extraction without member checks: a crafted archive with ``../`` members or +absolute paths writes outside the target directory. ``tarfile`` accepts a ``filter``.""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import ClassVar + +from coretrace_python.analysis import AnyAnalysis +from coretrace_python.findings import Severity +from coretrace_python.interprocedural import CallGraphAnalysis +from coretrace_python.ir.model import Call, Instruction, Value +from coretrace_python.ir.ssa import SSAAnalysis +from coretrace_python.plugins import SymbolCallDetector +from coretrace_python.semantic.symbols import SymbolId + +_OPENERS = ("tarfile.open", "tarfile.TarFile", "tarfile.TarFile.open", "zipfile.ZipFile", "zipfile.Path") +_EXTRACTORS = tuple(f"{opener}.{method}" for opener in _OPENERS for method in ("extractall", "extract")) + + +class UnsafeArchivePlugin(SymbolCallDetector): + name: ClassVar[str] = "unsafe-archive-extraction" + rule_id: ClassVar[str] = "unsafe-archive-extraction" + requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({SSAAnalysis, CallGraphAnalysis}) + symbols: ClassVar[frozenset[SymbolId]] = frozenset( + SymbolId(f"python.{p}") for p in (*_EXTRACTORS, "shutil.unpack_archive") + ) + severity: ClassVar[Severity] = Severity.MEDIUM + message_template: ClassVar[str] = ( + "call to {symbol} extracts an archive without checking its member paths; a " + "crafted archive writes outside the target directory" + ) + + def accepts(self, call: Call, symbol: SymbolId, defs: Mapping[Value, Instruction]) -> bool: + # ``extractall(path, filter="data")`` rejects unsafe members (Python 3.12). + return not any(name == "filter" for name, _ in call.keywords) diff --git a/src/coretrace_python/bundled/syntax/unsafe_xml/plugin.toml b/src/coretrace_python/bundled/syntax/unsafe_xml/plugin.toml new file mode 100644 index 0000000..311b5ee --- /dev/null +++ b/src/coretrace_python/bundled/syntax/unsafe_xml/plugin.toml @@ -0,0 +1,9 @@ +name = "unsafe-xml" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = ["interprocedural.callgraph", "ir.ssa"] +provides = ["vulnerability.unsafe-xml"] + +[entrypoint] +module = "unsafe_xml" +class = "UnsafeXmlPlugin" diff --git a/src/coretrace_python/bundled/syntax/unsafe_xml/unsafe_xml.py b/src/coretrace_python/bundled/syntax/unsafe_xml/unsafe_xml.py new file mode 100644 index 0000000..87adfff --- /dev/null +++ b/src/coretrace_python/bundled/syntax/unsafe_xml/unsafe_xml.py @@ -0,0 +1,35 @@ +"""XML parsers of the standard library and lxml expand entities: a crafted document reads +local files or exhausts memory. ``defusedxml`` offers the same functions safely.""" + +from __future__ import annotations + +from typing import ClassVar + +from coretrace_python.analysis import AnyAnalysis +from coretrace_python.findings import Severity +from coretrace_python.interprocedural import CallGraphAnalysis +from coretrace_python.ir.ssa import SSAAnalysis +from coretrace_python.plugins import SymbolCallDetector +from coretrace_python.semantic.symbols import SymbolId + +_PARSERS = ( + "xml.etree.ElementTree.parse", "xml.etree.ElementTree.fromstring", "xml.etree.ElementTree.iterparse", + "xml.etree.ElementTree.XMLParser", "xml.etree.ElementTree.fromstringlist", + "xml.dom.minidom.parse", "xml.dom.minidom.parseString", + "xml.dom.pulldom.parse", "xml.dom.pulldom.parseString", + "xml.sax.parse", "xml.sax.parseString", "xml.sax.make_parser", + "xml.dom.expatbuilder.parse", "xml.dom.expatbuilder.parseString", + "lxml.etree.parse", "lxml.etree.fromstring", "lxml.etree.XMLParser", "lxml.etree.iterparse", + "lxml.etree.XML", +) + + +class UnsafeXmlPlugin(SymbolCallDetector): + name: ClassVar[str] = "unsafe-xml" + rule_id: ClassVar[str] = "unsafe-xml" + requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({SSAAnalysis, CallGraphAnalysis}) + symbols: ClassVar[frozenset[SymbolId]] = frozenset(SymbolId(f"python.{p}") for p in _PARSERS) + severity: ClassVar[Severity] = Severity.MEDIUM + message_template: ClassVar[str] = ( + "call to {symbol} parses XML with entity expansion enabled; use defusedxml" + ) diff --git a/src/coretrace_python/bundled/syntax/weak_random/plugin.toml b/src/coretrace_python/bundled/syntax/weak_random/plugin.toml new file mode 100644 index 0000000..29f383f --- /dev/null +++ b/src/coretrace_python/bundled/syntax/weak_random/plugin.toml @@ -0,0 +1,9 @@ +name = "weak-random" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = ["semantic.scopes", "semantic.symbols"] +provides = ["vulnerability.weak-random"] + +[entrypoint] +module = "weak_random" +class = "WeakRandomPlugin" diff --git a/src/coretrace_python/bundled/syntax/weak_random/weak_random.py b/src/coretrace_python/bundled/syntax/weak_random/weak_random.py new file mode 100644 index 0000000..d69dff7 --- /dev/null +++ b/src/coretrace_python/bundled/syntax/weak_random/weak_random.py @@ -0,0 +1,86 @@ +"""``random`` builds a secret: the Mersenne Twister is predictable from a few outputs, so +a token, key or session identifier drawn from it can be guessed. ``secrets`` and +``random.SystemRandom`` are the alternatives. A call is reported when its result is +assigned to a credential-like name, or when the enclosing function's name says it +produces one.""" + +from __future__ import annotations + +import re +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 +from coretrace_python.semantic.symbols import SymbolAnalysis, SymbolId + +_FUNCTIONS = ("random", "randint", "choice", "choices", "randrange", "getrandbits", "uniform", "sample", "randbytes") +RANDOM = frozenset(SymbolId(f"python.random.{f}") for f in _FUNCTIONS) +CREDENTIAL = re.compile(r"(?i)(token|secret|password|passwd|api_?key|session|nonce|salt|otp|csrf|auth)") + + +class WeakRandomPlugin(Plugin): + name: ClassVar[str] = "weak-random" + rule_id: ClassVar[str] = "weak-random" + 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 function in ctx.functions(): + scope = scopes.scope_for(function).id + for target, call in _random_calls(function, scope, symbols): + purpose = target if target is not None and CREDENTIAL.search(target) else None + if purpose is None and CREDENTIAL.search(function.name): + purpose = function.name + if purpose is None: + continue + symbol = symbols.resolve_expression(scope, call.callee) + findings.append( + Finding( + self.rule_id, + f"{symbol} is not a secure random source; {purpose!r} looks like a secret", + Severity.MEDIUM, + Confidence.MEDIUM, + call.span, + function.name, + {"symbol": str(symbol), "purpose": purpose}, + ) + ) + return findings + + +def _random_calls( + function: nodes.Function, scope: object, symbols: object +) -> Iterator[tuple[str | None, nodes.Call]]: + """Calls to ``random`` functions with the name of the target they are assigned to.""" + + def walk(node: Node, target: str | None) -> Iterator[tuple[str | None, nodes.Call]]: + if isinstance(node, nodes.Function | nodes.Class | nodes.Lambda): + return + if isinstance(node, nodes.Assign | nodes.AugAssign): + target = _target_name(node.target) + if isinstance(node, nodes.Call): + symbol = symbols.resolve_expression(scope, node.callee) # type: ignore[attr-defined] + if symbol in RANDOM: + yield target, node + for child in children(node): + yield from walk(child, target) + + for statement in function.body: + yield from walk(statement, None) + + +def _target_name(target: nodes.Target) -> str | None: + if isinstance(target, nodes.Name): + return target.identifier + if isinstance(target, nodes.Attribute): + return target.name + if isinstance(target, nodes.Subscript) and isinstance(target.key, nodes.Constant): + return str(target.key.value) + return None diff --git a/src/coretrace_python/plugins/detectors.py b/src/coretrace_python/plugins/detectors.py index 987a17b..846183a 100644 --- a/src/coretrace_python/plugins/detectors.py +++ b/src/coretrace_python/plugins/detectors.py @@ -8,13 +8,14 @@ from __future__ import annotations -from collections.abc import Sequence +from collections.abc import Mapping, Sequence from typing import ClassVar from coretrace_python.analysis import AnyAnalysis from coretrace_python.findings import Confidence, Finding, Severity from coretrace_python.findings.refutation import RefutationAnalysis, Status -from coretrace_python.ir.model import Call, Symbol +from coretrace_python.interprocedural import CallGraphAnalysis +from coretrace_python.ir.model import Call, Instruction, Symbol, Value from coretrace_python.ir.ssa import SSAAnalysis from coretrace_python.plugins.api import Plugin, PluginContext from coretrace_python.semantic.symbols import SymbolId @@ -75,37 +76,49 @@ def _lower(confidence: Confidence) -> Confidence: class SymbolCallDetector(Plugin): - """Report calls whose callee is one of ``symbols``, whatever name the file uses.""" + """Report calls whose callee is one of ``symbols``, whatever name the file uses. + A subclass that also requires ``CallGraphAnalysis`` sees derived call-chain symbols + too (``tarfile.open(p).extractall`` is ``tarfile.open.extractall``); ``accepts`` + lets it keep only some calls, such as those with a given keyword.""" requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({SSAAnalysis}) rule_id: ClassVar[str] symbols: ClassVar[frozenset[SymbolId]] severity: ClassVar[Severity] message_template: ClassVar[str] + confidence: ClassVar[Confidence] = Confidence.HIGH + + def accepts(self, call: Call, symbol: SymbolId, defs: Mapping[Value, Instruction]) -> bool: + return True def analyze(self, ctx: PluginContext) -> Sequence[Finding]: findings: list[Finding] = [] + graph = ctx.get(CallGraphAnalysis) if CallGraphAnalysis in self.requires else None for function in ctx.functions(): ssa = ctx.get(SSAAnalysis, function) - defined = { - i.result: i.symbol_id - for block in ssa.blocks - for i in block.instructions - if isinstance(i, Symbol) + defs: dict[Value, Instruction] = { + i.result: i for block in ssa.blocks for i in block.instructions if i.result is not None } + symbols: Mapping[Value, SymbolId] + if graph is not None: + symbols = graph.symbols(graph.name_of(function)) + else: + symbols = {v: i.symbol_id for v, i in defs.items() if isinstance(i, Symbol)} for block in ssa.blocks: for instruction in block.instructions: if not isinstance(instruction, Call): continue - symbol = defined.get(instruction.callee) + symbol = symbols.get(instruction.callee) if symbol is None or symbol not in self.symbols: continue + if not self.accepts(instruction, symbol, defs): + continue findings.append( Finding( rule_id=self.rule_id, message=self.message_template.format(symbol=symbol), severity=self.severity, - confidence=Confidence.HIGH, + confidence=self.confidence, span=instruction.location, function=function.name, metadata={"symbol": str(symbol)}, diff --git a/tests/regression/expected/httpie.json b/tests/regression/expected/httpie.json index 53a5688..0488bcb 100644 --- a/tests/regression/expected/httpie.json +++ b/tests/regression/expected/httpie.json @@ -24,6 +24,12 @@ "rule": "ssrf", "function": "collect_messages" }, + { + "file": "httpie/internal/update_warnings.py", + "line": 44, + "rule": "insecure-tls", + "function": "_fetch_updates" + }, { "file": "httpie/internal/update_warnings.py", "line": 44, diff --git a/tests/regression/expected/luigi.json b/tests/regression/expected/luigi.json index 7839d6f..fbc7ff0 100644 --- a/tests/regression/expected/luigi.json +++ b/tests/regression/expected/luigi.json @@ -30,6 +30,12 @@ "rule": "command-injection", "function": "_run_job" }, + { + "file": "luigi/contrib/mrrunner.py", + "line": 64, + "rule": "unsafe-archive-extraction", + "function": "extract_packages_archive" + }, { "file": "luigi/contrib/pai.py", "line": 229, @@ -54,6 +60,12 @@ "rule": "missing-timeout", "function": "start_session" }, + { + "file": "luigi/contrib/salesforce.py", + "line": 261, + "rule": "unsafe-xml", + "function": "start_session" + }, { "file": "luigi/contrib/salesforce.py", "line": 286, @@ -78,6 +90,12 @@ "rule": "missing-timeout", "function": "create_operation_job" }, + { + "file": "luigi/contrib/salesforce.py", + "line": 401, + "rule": "unsafe-xml", + "function": "create_operation_job" + }, { "file": "luigi/contrib/salesforce.py", "line": 412, @@ -102,12 +120,24 @@ "rule": "missing-timeout", "function": "create_batch" }, + { + "file": "luigi/contrib/salesforce.py", + "line": 468, + "rule": "unsafe-xml", + "function": "create_batch" + }, { "file": "luigi/contrib/salesforce.py", "line": 512, "rule": "missing-timeout", "function": "get_batch_result_ids" }, + { + "file": "luigi/contrib/salesforce.py", + "line": 515, + "rule": "unsafe-xml", + "function": "get_batch_result_ids" + }, { "file": "luigi/contrib/salesforce.py", "line": 528, @@ -120,12 +150,24 @@ "rule": "missing-timeout", "function": "_get_batch_info" }, + { + "file": "luigi/contrib/salesforce.py", + "line": 537, + "rule": "unsafe-xml", + "function": "_get_batch_info" + }, { "file": "luigi/contrib/sge.py", "line": 296, "rule": "command-injection", "function": "_run_job" }, + { + "file": "luigi/safe_extractor.py", + "line": 97, + "rule": "unsafe-archive-extraction", + "function": "safe_extract" + }, { "file": "test/_test_ftp.py", "line": 43, diff --git a/tests/regression/expected/pygoat.json b/tests/regression/expected/pygoat.json index fc226e9..80d31f2 100644 --- a/tests/regression/expected/pygoat.json +++ b/tests/regression/expected/pygoat.json @@ -48,6 +48,12 @@ "rule": "debug-enabled", "function": null }, + { + "file": "dockerized_labs/sensitive_data_exposure/dataexposure/views.py", + "line": 42, + "rule": "weak-random", + "function": "generate_api_key" + }, { "file": "dockerized_labs/sensitive_data_exposure/sensitive_data_lab/settings.py", "line": 8, @@ -144,6 +150,18 @@ "rule": "insecure-deserialization", "function": "insec_des_lab" }, + { + "file": "introduction/views.py", + "line": 258, + "rule": "unsafe-xml", + "function": "xxe_parse" + }, + { + "file": "introduction/views.py", + "line": 260, + "rule": "unsafe-xml", + "function": "xxe_parse" + }, { "file": "introduction/views.py", "line": 290, @@ -174,6 +192,12 @@ "rule": "dangerous-eval", "function": "cmd_lab2" }, + { + "file": "introduction/views.py", + "line": 496, + "rule": "weak-random", + "function": "Otp" + }, { "file": "introduction/views.py", "line": 538, diff --git a/tests/regression/expected/yt-fts.json b/tests/regression/expected/yt-fts.json index f331bfc..dc86579 100644 --- a/tests/regression/expected/yt-fts.json +++ b/tests/regression/expected/yt-fts.json @@ -6,6 +6,12 @@ "functions_analysed": 147 }, "findings": [ + { + "file": "src/yt_fts/download/download_handler.py", + "line": 190, + "rule": "unsafe-xml", + "function": "get_channel_name" + }, { "file": "tests/test_export.py", "line": 48, @@ -17,6 +23,12 @@ "line": 17, "rule": "missing-timeout", "function": "fetch_and_unzip_test_db" + }, + { + "file": "tests/testing_utils.py", + "line": 26, + "rule": "unsafe-archive-extraction", + "function": "fetch_and_unzip_test_db" } ] } diff --git a/tests/test_detectors.py b/tests/test_detectors.py index 8fc756f..c5a8b9d 100644 --- a/tests/test_detectors.py +++ b/tests/test_detectors.py @@ -178,6 +178,8 @@ def test_shipped_plugins_load_with_their_manifests() -> None: "hardcoded-secrets", "http-client-models", "insecure-deserialization", + "insecure-temp-file", + "insecure-tls", "missing-timeout", "open-redirect", "path-traversal", @@ -189,7 +191,10 @@ def test_shipped_plugins_load_with_their_manifests() -> None: "sqlalchemy-models", "ssrf", "tornado-models", + "unsafe-archive-extraction", + "unsafe-xml", "vulnerable-dependency", + "weak-random", "weak-crypto", "xss", } diff --git a/tests/test_docs.py b/tests/test_docs.py index 266309d..d604eac 100644 --- a/tests/test_docs.py +++ b/tests/test_docs.py @@ -90,7 +90,7 @@ def test_every_bundled_plugin_is_named_in_the_guides() -> None: re.search(r'^name = "([a-z-]+)"', manifest.read_text(encoding="utf-8"), re.MULTILINE).group(1) # type: ignore[union-attr] for manifest in engine.BUNDLED_PLUGINS.rglob("plugin.toml") ] - assert len(names) == 30 + assert len(names) == 35 assert all(f"`{name}`" in text for name in names), [n for n in names if f"`{n}`" not in text] diff --git a/tests/test_packaging.py b/tests/test_packaging.py index b55e2db..0fce742 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -64,7 +64,7 @@ def test_bundled_plugins_live_inside_the_package() -> None: assert bundled == Path(coretrace_python.__file__).resolve().parent / "bundled" assert bundled.is_dir() manifests = sorted(p.relative_to(bundled).as_posix() for p in bundled.rglob("plugin.toml")) - assert len(manifests) == 30 + assert len(manifests) == 35 assert "syntax/dangerous_eval/plugin.toml" in manifests assert not (REPO / "plugins").exists() @@ -108,7 +108,7 @@ def test_the_wheel_carries_the_bundled_plugins(tmp_path: Path) -> None: assert "coretrace_python/bundled/syntax/dangerous_eval/plugin.toml" in names assert "coretrace_python/bundled/syntax/dangerous_eval/dangerous_eval.py" in names - assert sum(1 for n in names if n.endswith("/plugin.toml")) == 30 + assert sum(1 for n in names if n.endswith("/plugin.toml")) == 35 assert not any(n.startswith("tests/") or "tests-project" in n for n in names)