Python: Reduce log injection false positives

Add `SimpleTypeSanitizer` in a new `semmle.python.security.Sanitizers`, the
Python counterpart of the class the Java and C# queries already use. A routed
parameter annotated with a simple type such as `int` or `uuid.UUID` is validated
by the web framework before the request handler runs, so it cannot contain a
line break and cannot be used to forge a log entry. The annotation is only
trusted on routed parameters, since Python does not enforce annotations at run
time. `Annotated[T, ...]`, `Optional[T]` and `T | None` are unwrapped.

Recognize more ways of neutralizing a log message: `str.translate`,
`str.encode("unicode_escape")`, a `re.sub` or `re.compile(...).sub` whose
pattern matches control characters, and escaping conversions such as `repr` and
`json.dumps`.

Recognize a `logging.Formatter` subclass that strips control characters from the
records it renders. Such a formatter sanitizes when the record is written rather
than where it is created, so no sanitizing call appears between the source and
the logging call. Callees are resolved through module-level definitions, which
local flow does not reach from a nested scope.

Author: Ali-Razmjoo

python/ql/lib/change-notes/2026-07-24-log-injection-false-positives.md

@@ -0,0 +1,11 @@
+---
+category: minorAnalysis
+---
+* The `py/log-injection` query no longer reports a routed parameter whose type annotation a
+  web framework validates to be a simple type, such as `int` or `uuid.UUID`, since such a
+  value cannot contain a line break. The query also recognizes more ways of neutralizing a
+  log message, including `str.translate`, `re.sub` with a pattern matching control
+  characters, escaping conversions such as `repr` and `json.dumps`, and a
+  `logging.Formatter` subclass that strips control characters from every record it renders.
+  This reduces false positives in applications that use typed path parameters or sanitize
+  when the record is written rather than where it is created.

python/ql/lib/change-notes/2026-07-24-simple-type-sanitizer.md

@@ -0,0 +1,8 @@
+---
+category: feature
+---
+* Added a `SimpleTypeSanitizer` class to the new `semmle.python.security.Sanitizers` module. It
+  identifies values of a type that cannot carry an injection payload, such as numbers, UUIDs,
+  dates, and enum members, either as the result of a conversion or as a routed parameter whose
+  type annotation a web framework validates. This is the Python counterpart of the simple-type
+  sanitizers used by the Java and C# queries.

python/ql/lib/semmle/python/security/Sanitizers.qll

@@ -0,0 +1,108 @@
+/**
+ * Provides sanitizers that are applicable to several security queries.
+ */
+
+private import python
+private import semmle.python.ApiGraphs
+private import semmle.python.Concepts
+private import semmle.python.dataflow.new.DataFlow
+
+/** Gets a reference to `name` from the `typing` or `typing_extensions` module. */
+private API::Node typingRef(string name) {
+  result = API::moduleImport(["typing", "typing_extensions"]).getMember(name)
+}
+
+/**
+ * Gets a reference to a type whose instances cannot carry an injection payload.
+ *
+ * These are types with a closed, machine-generated string representation, such as
+ * numbers, UUIDs, dates and enum members. A value of such a type can never contain
+ * a line break, a quote, or a control character, and so cannot be used to forge a
+ * log entry, break out of a query string, and so on.
+ */
+private API::Node simpleTypeRef() {
+  result = API::builtin(["int", "float", "bool", "complex"])
+  or
+  result = API::moduleImport("uuid").getMember("UUID")
+  or
+  result = API::moduleImport("datetime").getMember(["date", "datetime", "time", "timedelta"])
+  or
+  result = API::moduleImport("decimal").getMember("Decimal")
+  or
+  result = API::moduleImport("ipaddress").getMember(["IPv4Address", "IPv6Address"])
+  or
+  // Pydantic aliases that validate to the corresponding stdlib type.
+  // See https://docs.pydantic.dev/latest/api/types/
+  result =
+    API::moduleImport("pydantic")
+        .getMember([
+            "UUID1", "UUID3", "UUID4", "UUID5", "StrictBool", "StrictInt", "StrictFloat",
+            "PositiveInt", "NegativeInt", "NonNegativeInt", "NonPositiveInt", "PositiveFloat",
+            "NegativeFloat", "NonNegativeFloat", "NonPositiveFloat"
+          ])
+  or
+  // The members of an enum are fixed when the class is defined, so an attacker can
+  // at most select between values that already occur in the source code.
+  result =
+    API::moduleImport("enum")
+        .getMember(["Enum", "IntEnum", "StrEnum", "Flag", "IntFlag"])
+        .getASubclass+()
+}
+
+/**
+ * Gets the type denoted by the parameter annotation `annotation`, looking through the
+ * `Annotated[T, ...]`, `Optional[T]` and `T | None` wrappers that web frameworks accept.
+ */
+private Expr unwrapTypeAnnotation(Expr annotation) {
+  annotation = any(Parameter p).getAnnotation() and
+  result = annotation
+  or
+  exists(Expr inner | inner = unwrapTypeAnnotation(annotation) |
+    // `Annotated[T, ...]`, where only the first element denotes the type.
+    inner.(Subscript).getObject() =
+      typingRef("Annotated").getAValueReachableFromSource().asExpr() and
+    result = inner.(Subscript).getIndex().(Tuple).getElt(0)
+    or
+    // `Optional[T]`
+    inner.(Subscript).getObject() =
+      typingRef("Optional").getAValueReachableFromSource().asExpr() and
+    result = inner.(Subscript).getIndex()
+    or
+    // `T | None`
+    exists(BinaryExpr union | union = inner and union.getOp() instanceof BitOr |
+      result = union.getLeft() and union.getRight() instanceof None
+      or
+      result = union.getRight() and union.getLeft() instanceof None
+    )
+  )
+}
+
+/**
+ * A node whose value is of a simple type unlikely to carry taint, such as a number,
+ * a `uuid.UUID`, a `datetime.datetime`, or an enum member.
+ *
+ * This is the Python counterpart of the `SimpleTypeSanitizer` classes used by the
+ * Java and C# queries.
+ *
+ * Unlike in a statically typed language, a Python type annotation is not enforced at
+ * run time, so an annotation is not trusted on its own. The annotation of a routed
+ * parameter is trusted, however, since a web framework validates an incoming request
+ * against it -- and rejects the request outright -- before the request handler body
+ * runs.
+ */
+class SimpleTypeSanitizer extends DataFlow::Node {
+  SimpleTypeSanitizer() {
+    // A routed parameter whose annotation the web framework validates.
+    exists(Parameter p |
+      p = this.(DataFlow::ParameterNode).getParameter() and
+      p = any(Http::Server::RequestHandler handler).getARoutedParameter() and
+      unwrapTypeAnnotation(p.getAnnotation()) =
+        simpleTypeRef().getAValueReachableFromSource().asExpr()
+    )
+    or
+    // The result of converting a value to a simple type.
+    this = simpleTypeRef().getACall()
+    or
+    this = API::builtin(["len", "hash", "id", "ord"]).getACall()
+  }
+}

python/ql/lib/semmle/python/security/dataflow/LogInjectionCustomizations.qll

@@ -5,11 +5,13 @@
  */
 
 private import python
+private import semmle.python.ApiGraphs
 private import semmle.python.dataflow.new.DataFlow
 private import semmle.python.Concepts
 private import semmle.python.dataflow.new.RemoteFlowSources
 private import semmle.python.dataflow.new.BarrierGuards
 private import semmle.python.frameworks.data.ModelsAsData
+private import semmle.python.security.Sanitizers
 
 /**
  * Provides default sources, sinks and sanitizers for detecting
@@ -81,6 +83,16 @@ module LogInjection {
     SinkFromModel() { ModelOutput::sinkNode(this, "log-injection") }
   }
 
+  /**
+   * A value of a simple type, such as an `int` or a `uuid.UUID`, considered as a
+   * sanitizer.
+   *
+   * Such a value has a machine-generated string representation that can never
+   * contain a line break, so it cannot be used to forge a log entry. The Java and
+   * C# log injection queries use the same sanitizer.
+   */
+  class SimpleTypeAsSanitizer extends Sanitizer instanceof SimpleTypeSanitizer { }
+
   /**
    * A comparison with a constant, considered as a sanitizer-guard.
    */
@@ -107,6 +119,160 @@ module LogInjection {
     }
   }
 
+  /**
+   * Gets the pattern of a call to `re.sub` or `re.subn`, whether the pattern is
+   * passed directly or was compiled with `re.compile`.
+   */
+  private string getARegexSubstitutionPattern(DataFlow::CallCfgNode call) {
+    call = API::moduleImport("re").getMember(["sub", "subn"]).getACall() and
+    result = call.getArg(0).asExpr().(StringLiteral).getText()
+    or
+    exists(API::CallNode compile |
+      compile = API::moduleImport("re").getMember("compile").getACall() and
+      call = compile.getReturn().getMember(["sub", "subn"]).getACall() and
+      result = compile.getArg(0).asExpr().(StringLiteral).getText()
+    )
+  }
+
+  /**
+   * Holds if `call` removes or escapes the characters that could be used to forge a
+   * log entry.
+   *
+   * Like `ReplaceLineBreaksSanitizer`, this is deliberately generous: we do not
+   * verify that every kind of line break is handled, only that the value has been
+   * put through an operation whose purpose is to neutralize them.
+   */
+  private predicate stripsControlCharacters(DataFlow::CallCfgNode call) {
+    exists(string method | method = call.getFunction().(DataFlow::AttrRead).getAttributeName() |
+      // `x.replace("\n", "")`
+      method = "replace" and
+      call.getArg(0).asExpr().(StringLiteral).getText() in ["\r\n", "\n", "\r"]
+      or
+      // `x.translate(table)`. The table is assumed to escape control characters, as
+      // `str.translate` has no other common use on a log message.
+      method = "translate"
+      or
+      // `x.encode("unicode_escape")`
+      method = "encode" and
+      call.getArg(0).asExpr().(StringLiteral).getText() = "unicode_escape"
+    )
+    or
+    // A substitution whose pattern mentions a line break or a control character class.
+    getARegexSubstitutionPattern(call)
+        .matches(["%\\n%", "%\\r%", "%\\x0%", "%\\s%", "%cntrl%", "%\n%"])
+    or
+    // Conversions that render a control character inert by escaping it.
+    call = API::builtin(["repr", "ascii"]).getACall()
+    or
+    call = API::moduleImport("json").getMember("dumps").getACall()
+    or
+    call = API::moduleImport("urllib.parse").getMember(["quote", "quote_plus"]).getACall()
+  }
+
+  /**
+   * A call that strips control characters from its input, considered as a sanitizer.
+   *
+   * This subsumes `ReplaceLineBreaksSanitizer`, which is retained because it is part
+   * of the public API of this module.
+   */
+  class StripControlCharactersSanitizer extends Sanitizer, DataFlow::CallCfgNode {
+    StripControlCharactersSanitizer() { stripsControlCharacters(this) }
+  }
+
+  /**
+   * Gets a callable that `call` may invoke, where the callable is defined in the
+   * module being analyzed.
+   *
+   * Local flow does not cross a scope boundary, so a module-level definition that is
+   * referenced from inside a function or a method has to be resolved by name.
+   */
+  private Scope getALocallyResolvedCallee(DataFlow::CallCfgNode call) {
+    exists(DataFlow::LocalSourceNode callee | callee.flowsTo(call.getFunction()) |
+      callee.asExpr() = result.(Function).getDefinition()
+      or
+      callee.asExpr().(ClassExpr).getInnerScope() = result
+    )
+    or
+    exists(DataFlow::ModuleVariableNode var |
+      var.getARead() = call.getFunction() and
+      result.getName() = var.getVariable().getId() and
+      result.getEnclosingScope() = var.getModule()
+    )
+  }
+
+  /** Gets a method of `cls` that renders a log record. */
+  private Function getARecordRenderingMethod(Class cls) {
+    result = cls.getAMethod() and
+    result.getName() in ["format", "formatMessage"]
+  }
+
+  /**
+   * Gets a function called, directly or indirectly, from the record-rendering method
+   * `f`.
+   *
+   * Only calls that resolve to a definition in the module being analyzed are
+   * followed, which covers the common case of a formatter delegating to a helper
+   * defined alongside it.
+   */
+  private Function getATransitiveCallee(Function f) {
+    f = getARecordRenderingMethod(_) and
+    result = f
+    or
+    exists(DataFlow::CallCfgNode call |
+      call.getScope() = getATransitiveCallee(f) and
+      result = getALocallyResolvedCallee(call)
+    )
+  }
+
+  /**
+   * Holds if `cls` is a `logging.Formatter` subclass that strips control characters
+   * from the records it renders.
+   *
+   * Such a formatter sanitizes when the record is written rather than where it is
+   * created, so no sanitizing call appears between the source and the logging call.
+   */
+  private predicate isSanitizingFormatter(Class cls) {
+    cls.getABase() =
+      API::moduleImport("logging").getMember("Formatter").getAValueReachableFromSource().asExpr() and
+    exists(DataFlow::CallCfgNode strip |
+      stripsControlCharacters(strip) and
+      strip.getScope() = getATransitiveCallee(getARecordRenderingMethod(cls))
+    )
+  }
+
+  /**
+   * Holds if a `logging.Formatter` that strips control characters is installed on a
+   * handler.
+   *
+   * This is a property of the application as a whole rather than of an individual
+   * logging call. We do not determine which loggers the formatter ends up attached
+   * to, as that would require resolving handler registration, which is typically
+   * spread across module-level configuration code and often iterates over
+   * `logging.getLogger().handlers`. Such a formatter is nearly always installed on
+   * the root logger, where it applies to every record that propagates to it, so its
+   * presence is treated as sanitizing all logging calls. The trade-off is a false
+   * negative for an application that also logs through a deliberately unsanitized
+   * handler.
+   */
+  private predicate hasSanitizingFormatter() {
+    exists(DataFlow::MethodCallNode setFormatter, DataFlow::CallCfgNode instantiation |
+      setFormatter.getMethodName() = "setFormatter" and
+      instantiation.flowsTo(setFormatter.getArg(0)) and
+      isSanitizingFormatter(getALocallyResolvedCallee(instantiation))
+    )
+  }
+
+  /**
+   * A logging operation in an application that installs a `logging.Formatter`
+   * stripping control characters, considered as a sanitizer.
+   */
+  class LoggingSanitizedByFormatter extends Sanitizer {
+    LoggingSanitizedByFormatter() {
+      hasSanitizingFormatter() and
+      this instanceof LoggingAsSink
+    }
+  }
+
   /**
    * A sanitizer defined via models-as-data with kind "log-injection".
    */

python/ql/test/query-tests/Security/CWE-117-LogInjection-formatter/LogInjection.expected

@@ -0,0 +1,4 @@
+#select
+edges
+nodes
+subpaths

python/ql/test/query-tests/Security/CWE-117-LogInjection-formatter/LogInjection.qlref

@@ -0,0 +1,2 @@
+query: Security/CWE-117/LogInjection.ql
+postprocess: utils/test/InlineExpectationsTestQuery.ql

python/ql/test/query-tests/Security/CWE-117-LogInjection-formatter/LogInjectionFormatter.py

@@ -0,0 +1,59 @@
+#!/usr/bin/env python
+# -*- coding: UTF-8 -*-
+"""
+@Desc   ：Log Injection prevented by a sanitizing `logging.Formatter`
+
+A formatter that strips control characters sanitizes when the record is written,
+so the individual logging calls are not vulnerable.
+"""
+import logging
+
+from flask import Flask
+from flask import request
+
+app = Flask(__name__)
+
+ESCAPE_TABLE = {ord("\n"): "\\n", ord("\r"): "\\r", ord("\t"): "\\t"}
+
+
+def escape_controls(text):
+    """Escape the control characters in a log message."""
+    return text.translate(ESCAPE_TABLE)
+
+
+class EscapingFormatter(logging.Formatter):
+    """A formatter that escapes control characters in the rendered message."""
+
+    def format(self, record):
+        record.msg = escape_controls(record.getMessage())
+        record.args = None
+        return super().format(record)
+
+
+logging.basicConfig(level=logging.INFO, handlers=[logging.StreamHandler()])
+
+for handler in logging.getLogger().handlers:
+    handler.setFormatter(EscapingFormatter())
+
+logger = logging.getLogger("test")
+
+
+@app.route("/f-string")
+def f_string():
+    name = request.args.get("name")
+    logger.warning(f"User name: {name}")  # Good
+    return "ok"
+
+
+@app.route("/percent-args")
+def percent_args():
+    name = request.args.get("name")
+    logger.info("User name: %s", name)  # Good
+    return "ok"
+
+
+@app.route("/escaped-at-call-site")
+def escaped_at_call_site():
+    name = request.args.get("name")
+    logger.warning("User name: " + escape_controls(name))  # Good
+    return "ok"