Add insecure TLS, unsafe XML, archive extraction, mktemp and weak random rules

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`.
 

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"
+    )

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"

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

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"

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"

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)

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"

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"
+    )

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"

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

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)},

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,