Python: Add query to handle SQLAlchemy TextClause Injection

instead of doing this via taint-steps. See description in code/tests.

Author: RasmusWL

python/change-notes/2021-09-02-add-SQLAlchemyTextClauseInjection-query.md

@@ -0,0 +1,2 @@
+lgtm,codescanning
+* Introduced a new query _SQLAlchemy TextClause built from user-controlled sources_ (`py/sqlalchemy-textclause-injection`) to alert if user-input is added to a TextClause from SQLAlchemy, since that can lead to SQL injection.

python/ql/lib/semmle/python/frameworks/SqlAlchemy.qll

@@ -15,12 +15,14 @@ private import semmle.python.Concepts
 private import semmle.python.frameworks.PEP249::PEP249 as PEP249
 
 /**
+ * INTERNAL: Do not use.
+ *
  * Provides models for the `SQLAlchemy` PyPI package.
  * See
  *  - https://pypi.org/project/SQLAlchemy/
  *  - https://docs.sqlalchemy.org/en/14/index.html
  */
-private module SqlAlchemy {
+module SqlAlchemy {
   /**
    * Provides models for the `sqlalchemy.engine.Engine` and `sqlalchemy.future.Engine` classes.
    *
@@ -279,80 +281,62 @@ private module SqlAlchemy {
   }
 
   /**
-   * Additional taint-steps for `sqlalchemy.text()`
+   * Provides models for the `sqlalchemy.sql.expression.TextClause` class,
+   * which represents a textual SQL string directly.
    *
-   * See https://docs.sqlalchemy.org/en/14/core/sqlelement.html#sqlalchemy.sql.expression.text
-   * See https://docs.sqlalchemy.org/en/14/core/sqlelement.html#sqlalchemy.sql.expression.TextClause
-   */
-  class SqlAlchemyTextAdditionalTaintSteps extends TaintTracking::AdditionalTaintStep {
-    override predicate step(DataFlow::Node nodeFrom, DataFlow::Node nodeTo) {
-      exists(DataFlow::CallCfgNode call |
-        (
-          call = API::moduleImport("sqlalchemy").getMember("text").getACall()
-          or
-          call = API::moduleImport("sqlalchemy").getMember("sql").getMember("text").getACall()
-          or
-          call =
-            API::moduleImport("sqlalchemy")
-                .getMember("sql")
-                .getMember("expression")
-                .getMember("text")
-                .getACall()
-          or
-          call =
-            API::moduleImport("sqlalchemy")
-                .getMember("sql")
-                .getMember("expression")
-                .getMember("TextClause")
-                .getACall()
-        ) and
-        nodeFrom in [call.getArg(0), call.getArgByName("text")] and
-        nodeTo = call
-      )
-    }
-  }
-}
-
-private module OldModeling {
-  /**
-   * Returns an instantization of a SqlAlchemy Session object.
-   * See https://docs.sqlalchemy.org/en/14/orm/session_api.html#sqlalchemy.orm.Session and
-   * https://docs.sqlalchemy.org/en/14/orm/session_api.html#sqlalchemy.orm.sessionmaker
-   */
-  private API::Node getSqlAlchemySessionInstance() {
-    result = API::moduleImport("sqlalchemy.orm").getMember("Session").getReturn() or
-    result = API::moduleImport("sqlalchemy.orm").getMember("sessionmaker").getReturn().getReturn()
-  }
-
-  /**
-   * Returns an instantization of a SqlAlchemy Query object.
-   * See https://docs.sqlalchemy.org/en/14/orm/query.html?highlight=query#sqlalchemy.orm.Query
+   * ```py
+   * session.query(For14).filter_by(description=sqlalchemy.text(f"'{user_input}'")).all()
+   * ```
+   *
+   * Initially I wanted to add lots of additional taint steps for such that the normal
+   * SQL injection query would be able to find cases as the one above where an ORM query
+   * includes a TextClause that includes user-input directly... But that presented 2
+   * problems:
+   *
+   * - which part of the query construction above should be marked as SQL to fit our
+   *   `SqlExecution` concept. Nothing really fits this well, since all the SQL
+   *   execution happens under the hood.
+   * - This would require a LOT of modeling for these additional taint steps, since
+   *   there are many many constructs we would need to have models for. (see the 2
+   *   examples below)
+   *
+   * So instead we flag user-input to a TextClause with its' own query
+   * (`py/sqlalchemy-textclause-injection`). And so we don't highlight any parts of an
+   * ORM constructed query such as these as containing SQL, and don't need the additional
+   * taint steps either.
+   *
+   * See
+   * - https://docs.sqlalchemy.org/en/14/core/sqlelement.html#sqlalchemy.sql.expression.TextClause.
+   * - https://docs.sqlalchemy.org/en/14/core/sqlelement.html#sqlalchemy.sql.expression.text
    */
-  private API::Node getSqlAlchemyQueryInstance() {
-    result = getSqlAlchemySessionInstance().getMember("query").getReturn()
-  }
+  module TextClause {
+    /**
+     * A construction of a `sqlalchemy.sql.expression.TextClause`, which represents a
+     * textual SQL string directly.
+     */
+    class TextClauseConstruction extends DataFlow::CallCfgNode {
+      TextClauseConstruction() {
+        this = API::moduleImport("sqlalchemy").getMember("text").getACall()
+        or
+        this = API::moduleImport("sqlalchemy").getMember("sql").getMember("text").getACall()
+        or
+        this =
+          API::moduleImport("sqlalchemy")
+              .getMember("sql")
+              .getMember("expression")
+              .getMember("text")
+              .getACall()
+        or
+        this =
+          API::moduleImport("sqlalchemy")
+              .getMember("sql")
+              .getMember("expression")
+              .getMember("TextClause")
+              .getACall()
+      }
 
-  /**
-   * A call on a Query object
-   * See https://docs.sqlalchemy.org/en/14/orm/query.html?highlight=query#sqlalchemy.orm.Query
-   */
-  private class SqlAlchemyQueryCall extends DataFlow::CallCfgNode, SqlExecution::Range {
-    SqlAlchemyQueryCall() {
-      this =
-        getSqlAlchemyQueryInstance()
-            .getMember(any(SqlAlchemyVulnerableMethodNames methodName))
-            .getACall()
+      /** Gets the argument that specifies the SQL text. */
+      DataFlow::Node getTextArg() { result in [this.getArg(0), this.getArgByName("text")] }
     }
-
-    override DataFlow::Node getSql() { result = this.getArg(0) }
-  }
-
-  /**
-   * This class represents a list of methods vulnerable to sql injection.
-   *
-   * See https://github.com/jty-team/codeql/pull/2#issue-611592361
-   */
-  private class SqlAlchemyVulnerableMethodNames extends string {
-    SqlAlchemyVulnerableMethodNames() { this in ["filter", "filter_by", "group_by", "order_by"] }
   }
 }

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

@@ -0,0 +1,35 @@
+/**
+ * Provides a taint-tracking configuration for detecting "SQLAlchemy TextClause injection" vulnerabilities.
+ *
+ * Note, for performance reasons: only import this file if
+ * `SQLAlchemyTextClause::Configuration` is needed, otherwise
+ * `SQLAlchemyTextClauseCustomizations` should be imported instead.
+ */
+
+private import python
+import semmle.python.dataflow.new.DataFlow
+import semmle.python.dataflow.new.TaintTracking
+
+/**
+ * Provides a taint-tracking configuration for detecting "SQLAlchemy TextClause injection" vulnerabilities.
+ */
+module SQLAlchemyTextClause {
+  import SQLAlchemyTextClauseCustomizations::SQLAlchemyTextClause
+
+  /**
+   * A taint-tracking configuration for detecting "SQLAlchemy TextClause injection" vulnerabilities.
+   */
+  class Configuration extends TaintTracking::Configuration {
+    Configuration() { this = "SQLAlchemyTextClause" }
+
+    override predicate isSource(DataFlow::Node source) { source instanceof Source }
+
+    override predicate isSink(DataFlow::Node sink) { sink instanceof Sink }
+
+    override predicate isSanitizer(DataFlow::Node node) { node instanceof Sanitizer }
+
+    override predicate isSanitizerGuard(DataFlow::BarrierGuard guard) {
+      guard instanceof SanitizerGuard
+    }
+  }
+}

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

@@ -0,0 +1,56 @@
+/**
+ * Provides default sources, sinks and sanitizers for detecting
+ * "SQLAlchemy TextClause injection"
+ * vulnerabilities, as well as extension points for adding your own.
+ */
+
+private import python
+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.SqlAlchemy
+
+/**
+ * Provides default sources, sinks and sanitizers for detecting
+ * "SQLAlchemy TextClause injection"
+ * vulnerabilities, as well as extension points for adding your own.
+ */
+module SQLAlchemyTextClause {
+  /**
+   * A data flow source for "SQLAlchemy TextClause injection" vulnerabilities.
+   */
+  abstract class Source extends DataFlow::Node { }
+
+  /**
+   * A data flow sink for "SQLAlchemy TextClause injection" vulnerabilities.
+   */
+  abstract class Sink extends DataFlow::Node { }
+
+  /**
+   * A sanitizer for "SQLAlchemy TextClause injection" vulnerabilities.
+   */
+  abstract class Sanitizer extends DataFlow::Node { }
+
+  /**
+   * A sanitizer guard for "SQLAlchemy TextClause injection" vulnerabilities.
+   */
+  abstract class SanitizerGuard extends DataFlow::BarrierGuard { }
+
+  /**
+   * A source of remote user input, considered as a flow source.
+   */
+  class RemoteFlowSourceAsSource extends Source, RemoteFlowSource { }
+
+  /**
+   * The text argument of a SQLAlchemy TextClause construction, considered as a flow sink.
+   */
+  class TextArgAsSink extends Sink {
+    TextArgAsSink() { this = any(SqlAlchemy::TextClause::TextClauseConstruction tcc).getTextArg() }
+  }
+
+  /**
+   * A comparison with a constant string, considered as a sanitizer-guard.
+   */
+  class StringConstCompareAsSanitizerGuard extends SanitizerGuard, StringConstCompare { }
+}

python/ql/src/Security/CWE-089/SQLAlchemyTextClauseInjection.qhelp

@@ -0,0 +1,52 @@
+<!DOCTYPE qhelp PUBLIC
+  "-//Semmle//qhelp//EN"
+  "qhelp.dtd">
+<qhelp>
+
+<overview>
+<p>
+The <code>TextClause</code> class in the <code>SQLAlchemy</code> PyPI package represents
+a textual SQL string directly. If user-input is added to it without sufficient
+sanitization, a user may be able to run malicious database queries, since the
+<code>TextClause</code> is inserted directly into the final SQL.
+</overview>
+
+<recommendation>
+<p>
+Don't allow user-input to be added to a <code>TextClause</code>, instead construct your
+full query with constructs from the ORM, or use query parameters for user-input.
+</p>
+</recommendation>
+
+<example>
+<p>
+In the following snippet, a user is fetched from the database using three
+different queries.
+</p>
+
+<p>
+In the first case, the final query string is built by directly including a user-supplied
+input. The parameter may include quote characters, so this code is vulnerable to a SQL
+injection attack.
+</p>
+
+<p>
+In the second case, the query is built using ORM models, but part of it is using a
+<code>TextClause</code> directly including a user-supplied input. Since the
+<code>TextClause</code> is inserted directly into the final SQL, this code is vulnerable
+to a SQL injection attack.
+</p>
+
+<p>
+In the third case, the query is built fully using the ORM models, so in the end, the
+user-supplied input will passed passed to the database using query parameters. The
+database connector library will take care of escaping and inserting quotes as needed.
+</p>
+
+<sample src="examples/sqlalchemy_textclause_injection.py" />
+</example>
+
+<references>
+<li><a href="https://docs.sqlalchemy.org/en/14/core/sqlelement.html#sqlalchemy.sql.expression.text.params.text">Official documentation of the text parameter</a>.</li>
+</references>
+</qhelp>

python/ql/src/Security/CWE-089/SQLAlchemyTextClauseInjection.ql

@@ -0,0 +1,23 @@
+/**
+ * @name SQLAlchemy TextClause built from user-controlled sources
+ * @description Building a TextClause query from user-controlled sources is vulnerable to insertion of
+ *              malicious SQL code by the user.
+ * @kind path-problem
+ * @problem.severity error
+ * @security-severity 8.8
+ * @precision high
+ * @id py/sqlalchemy-textclause-injection
+ * @tags security
+ *       external/cwe/cwe-089
+ *       external/owasp/owasp-a1
+ */
+
+import python
+import semmle.python.security.dataflow.SQLAlchemyTextClause
+import DataFlow::PathGraph
+
+from SQLAlchemyTextClause::Configuration config, DataFlow::PathNode source, DataFlow::PathNode sink
+where config.hasFlowPath(source, sink)
+select sink.getNode(), source, sink,
+  "This SQLAlchemy TextClause depends on $@, which could lead to SQL injection.", source.getNode(),
+  "a user-provided value"

python/ql/src/Security/CWE-089/examples/sqlalchemy_textclause_injection.py

@@ -0,0 +1,34 @@
+from flask import Flask, request
+import sqlalchemy
+import sqlalchemy.orm
+
+app = Flask(__name__)
+engine = sqlalchemy.create_engine(...)
+Base = sqlalchemy.orm.declarative_base()
+
+
+class User(Base):
+    __tablename__ = "users"
+
+    id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True)
+    username = sqlalchemy.Column(sqlalchemy.String)
+
+
+@app.route("/users/<username>")
+def show_user(username):
+    session = sqlalchemy.orm.Session(engine)
+
+    # BAD, normal SQL injection
+    stmt = sqlalchemy.text("SELECT * FROM users WHERE username = '{}'".format(username))
+    results = session.execute(stmt).fetchall()
+
+    # BAD, allows SQL injection
+    username_formatted_for_sql = sqlalchemy.text("'{}'".format(username))
+    stmt = sqlalchemy.select(User).where(User.username == username_formatted_for_sql)
+    results = session.execute(stmt).scalars().all()
+
+    # GOOD, does not allow for SQL injection
+    stmt = sqlalchemy.select(User).where(User.username == username)
+    results = session.execute(stmt).scalars().all()
+
+    ...

python/ql/test/library-tests/frameworks/sqlalchemy/ConceptsTest.ql

@@ -1,3 +1,2 @@
 import python
 import experimental.meta.ConceptsTest
-import experimental.semmle.python.frameworks.SqlAlchemy

python/ql/test/library-tests/frameworks/sqlalchemy/InlineTaintTest.ql

@@ -1,2 +1 @@
 import experimental.meta.InlineTaintTest
-import experimental.semmle.python.frameworks.SqlAlchemy

python/ql/test/library-tests/frameworks/sqlalchemy/SqlExecution.py

@@ -47,10 +47,10 @@ class User(Base):
 
 # Injection requiring the text() taint-step
 t = text("some sql")
-session.query(User).filter(t)  # $ getSql=t
-session.query(User).group_by(User.id).having(t)  # $ getSql=User.id MISSING: getSql=t
-session.query(User).group_by(t).first()  # $ getSql=t
-session.query(User).order_by(t).first()  # $ getSql=t
+session.query(User).filter(t)
+session.query(User).group_by(User.id).having(t)
+session.query(User).group_by(t).first()
+session.query(User).order_by(t).first()
 
 query = select(User).where(User.name == t)  # $ MISSING: getSql=t
 with engine.connect() as conn: