Python: Model exec_driver_sql

Author: RasmusWL

python/ql/src/experimental/semmle/python/frameworks/SqlAlchemy.qll

@@ -195,6 +195,18 @@ private module SqlAlchemy {
     override DataFlow::Node getSql() { result in [this.getArg(0), this.getArgByName("statement")] }
   }
 
+  /**
+   * A call to `exec_driver_sql` on a SQLAlchemy Connection.
+   * See
+   *  - https://docs.sqlalchemy.org/en/14/core/connections.html#sqlalchemy.engine.Connection.exec_driver_sql
+   *  - https://docs.sqlalchemy.org/en/14/core/future.html#sqlalchemy.future.Connection.exec_driver_sql
+   */
+  private class SqlAlchemyExecDriverSqlCall extends DataFlow::MethodCallNode, SqlExecution::Range {
+    SqlAlchemyExecDriverSqlCall() { this.calls(Connection::instance(), "exec_driver_sql") }
+
+    override DataFlow::Node getSql() { result in [this.getArg(0), this.getArgByName("statement")] }
+  }
+
   /**
    * A call to `scalar` on a SQLAlchemy Engine, Connection, or Session.
    * See

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

@@ -69,7 +69,7 @@
 
 
 # exec_driver_sql
-result = conn.exec_driver_sql(raw_sql) # $ MISSING: getSql=raw_sql
+result = conn.exec_driver_sql(raw_sql) # $ getSql=raw_sql
 assert result.fetchall() == [("FOO",)]
 
 # construction by object
@@ -284,7 +284,7 @@ class For14(Base):
     result = conn.execute(statement=text_sql) # $ getSql=text_sql
     assert result.fetchall() == [("FOO",)]
 
-    result = conn.exec_driver_sql(raw_sql) # $ MISSING: getSql=raw_sql
+    result = conn.exec_driver_sql(raw_sql) # $ getSql=raw_sql
     assert result.fetchall() == [("FOO",)]
 
     raw_conn = conn.connection