feat: audit-driven detectors

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

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

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

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",

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)

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"

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

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"

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']``."""