Python: use SqlConstruction in SqlAlchemy and

`SqlInjection`

Author: yoff

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

@@ -313,9 +313,11 @@ module SqlAlchemy {
      * A construction of a `sqlalchemy.sql.expression.TextClause`, which represents a
      * textual SQL string directly.
      */
-    abstract class TextClauseConstruction extends DataFlow::CallCfgNode {
+    abstract class TextClauseConstruction extends SqlConstruction::Range, DataFlow::CallCfgNode {
       /** Gets the argument that specifies the SQL text. */
-      DataFlow::Node getTextArg() { result in [this.getArg(0), this.getArgByName("text")] }
+      final override DataFlow::Node getSql() {
+        result in [this.getArg(0), this.getArgByName("text")]
+      }
     }
 
     /** `TextClause` constructions from the `sqlalchemy` package. */

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

@@ -43,17 +43,17 @@ module SqlInjection {
   class RemoteFlowSourceAsSource extends Source, RemoteFlowSource { }
 
   /**
-   * A SQL statement of a SQL execution, considered as a flow sink.
+   * A SQL statement of a SQL construction, considered as a flow sink.
    */
-  class SqlExecutionAsSink extends Sink {
-    SqlExecutionAsSink() { this = any(SqlExecution e).getSql() }
+  class SqlConstructionAsSink extends Sink {
+    SqlConstructionAsSink() { this = any(SqlConstruction c).getSql() }
   }
 
   /**
-   * The text argument of a SQLAlchemy TextClause construction, considered as a flow sink.
+   * A SQL statement of a SQL execution, considered as a flow sink.
    */
-  class TextArgAsSink extends Sink {
-    TextArgAsSink() { this = any(SqlAlchemy::TextClause::TextClauseConstruction tcc).getTextArg() }
+  class SqlExecutionAsSink extends Sink {
+    SqlExecutionAsSink() { this = any(SqlExecution e).getSql() }
   }
 
   /**

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

@@ -12,7 +12,7 @@
 # - https://github.com/pallets/flask-sqlalchemy/blob/931ec00d1e27f51508e05706eef41cc4419a0b32/src/flask_sqlalchemy/__init__.py#L765
 # - https://github.com/pallets/flask-sqlalchemy/blob/931ec00d1e27f51508e05706eef41cc4419a0b32/src/flask_sqlalchemy/__init__.py#L99-L109
 
-assert str(type(db.text("Foo"))) == "<class 'sqlalchemy.sql.elements.TextClause'>"
+assert str(type(db.text("Foo"))) == "<class 'sqlalchemy.sql.elements.TextClause'>"  # $ constructedSql="Foo"
 
 # also has engine/session instantiated
 
@@ -44,8 +44,8 @@
 
 
 # text
-t = db.text("foo")
+t = db.text("foo")  # $ constructedSql="foo"
 assert isinstance(t, sqlalchemy.sql.expression.TextClause)
 
-t = db.text(text="foo")
+t = db.text(text="foo")  # $ constructedSql="foo"
 assert isinstance(t, sqlalchemy.sql.expression.TextClause)

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

@@ -46,7 +46,7 @@ class User(Base):
     connection.execute("some sql")  # $ getSql="some sql"
 
 # Injection requiring the text() taint-step
-t = text("some sql")
+t = text("some sql")  # $ constructedSql="some sql"
 session.query(User).filter(t)
 session.query(User).group_by(User.id).having(t)
 session.query(User).group_by(t).first()

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

@@ -6,7 +6,7 @@
 # either v1.4 or v2.0, such that we cover both.
 
 raw_sql = "select 'FOO'"
-text_sql = sqlalchemy.text(raw_sql)
+text_sql = sqlalchemy.text(raw_sql)  # $ constructedSql=raw_sql
 
 Base = sqlalchemy.orm.declarative_base()
 
@@ -169,7 +169,7 @@ class For14(Base):
 
 # and now we can do the actual querying
 
-text_foo = sqlalchemy.text("'FOO'")
+text_foo = sqlalchemy.text("'FOO'")  # $ constructedSql="'FOO'"
 
 # filter_by is only vulnerable to injection if sqlalchemy.text is used, which is evident
 # from the logs produced if this file is run
@@ -298,7 +298,7 @@ class For14(Base):
     assert scalar_result == "FOO"
 
     # This is a contrived example
-    select = sqlalchemy.select(sqlalchemy.text("'BAR'"))
+    select = sqlalchemy.select(sqlalchemy.text("'BAR'"))  # $ constructedSql="'BAR'"
     result = conn.execute(select) # $ getSql=select
     assert result.fetchall() == [("BAR",)]
 

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

@@ -8,14 +8,14 @@ def test_taint():
 
     ensure_tainted(ts) # $ tainted
 
-    t1 = sqlalchemy.text(ts)
-    t2 = sqlalchemy.text(text=ts)
-    t3 = sqlalchemy.sql.text(ts)
-    t4 = sqlalchemy.sql.text(text=ts)
-    t5 = sqlalchemy.sql.expression.text(ts)
-    t6 = sqlalchemy.sql.expression.text(text=ts)
-    t7 = sqlalchemy.sql.expression.TextClause(ts)
-    t8 = sqlalchemy.sql.expression.TextClause(text=ts)
+    t1 = sqlalchemy.text(ts)  # $ constructedSql=ts
+    t2 = sqlalchemy.text(text=ts)  # $ constructedSql=ts
+    t3 = sqlalchemy.sql.text(ts)  # $ constructedSql=ts
+    t4 = sqlalchemy.sql.text(text=ts)  # $ constructedSql=ts
+    t5 = sqlalchemy.sql.expression.text(ts)  # $ constructedSql=ts
+    t6 = sqlalchemy.sql.expression.text(text=ts)  # $ constructedSql=ts
+    t7 = sqlalchemy.sql.expression.TextClause(ts)  # $ constructedSql=ts
+    t8 = sqlalchemy.sql.expression.TextClause(text=ts)  # $ constructedSql=ts
 
     # Since we flag user-input to a TextClause with its' own query, we don't want to
     # have a taint-step for it as that would lead to us also giving an alert for normal