Merge pull request #5485 from RasmusWL/django-queryset-chains

Approved by tausbn

Author: codeql-ci

python/change-notes/2021-03-22-django-queryset-chains.md

@@ -0,0 +1,2 @@
+lgtm,codescanning
+* Improved modeling of `django` to recognize QuerySet chains such as `User.objects.using("db-name").exclude(username="admin").extra("some sql")`. This can lead to new results for `py/sql-injection`.

python/ql/src/semmle/python/frameworks/Django.qll

@@ -105,9 +105,6 @@ private module Django {
       // -------------------------------------------------------------------------
       // django.db.models
       // -------------------------------------------------------------------------
-      // NOTE: The modelling of django models is currently fairly incomplete.
-      // It does not fully take `Model`s, `Manager`s, `and QuerySet`s into account.
-      // It simply identifies some common dangerous cases.
       /** Gets a reference to the `django.db.models` module. */
       private DataFlow::Node models(DataFlow::TypeTracker t) {
         t.start() and
@@ -124,71 +121,51 @@ private module Django {
 
       /** Provides models for the `django.db.models` module. */
       module models {
-        /** Provides models for the `django.db.models.Model` class. */
+        /**
+         * Provides models for the `django.db.models.Model` class and subclasses.
+         *
+         * See https://docs.djangoproject.com/en/3.1/topics/db/models/.
+         */
         module Model {
-          /** Gets a reference to the `django.db.models.Model` class. */
-          private DataFlow::Node classRef(DataFlow::TypeTracker t) {
-            t.start() and
-            result = DataFlow::importNode("django.db.models.Model")
-            or
-            t.startInAttr("Model") and
-            result = models()
-            or
-            // subclass
-            result.asExpr().(ClassExpr).getABase() = classRef(t.continue()).asExpr()
-            or
-            exists(DataFlow::TypeTracker t2 | result = classRef(t2).track(t2, t))
+          /** Gets a reference to the `flask.views.View` class or any subclass. */
+          API::Node subclassRef() {
+            result =
+              API::moduleImport("django")
+                  .getMember("db")
+                  .getMember("models")
+                  .getMember("Model")
+                  .getASubclass*()
           }
-
-          /** Gets a reference to the `django.db.models.Model` class. */
-          DataFlow::Node classRef() { result = classRef(DataFlow::TypeTracker::end()) }
-        }
-
-        /** Gets a reference to the `objects` object of a django model. */
-        private DataFlow::Node objects(DataFlow::TypeTracker t) {
-          t.startInAttr("objects") and
-          result = Model::classRef()
-          or
-          exists(DataFlow::TypeTracker t2 | result = objects(t2).track(t2, t))
         }
 
-        /** Gets a reference to the `objects` object of a model. */
-        DataFlow::Node objects() { result = objects(DataFlow::TypeTracker::end()) }
-
         /**
-         * Gets a reference to the attribute `attr_name` of an `objects` object.
-         * WARNING: Only holds for a few predefined attributes.
+         * Gets a reference to the Manager (django.db.models.Manager) for a django Model,
+         * accessed by `<ModelName>.objects`.
          */
-        private DataFlow::Node objects_attr(DataFlow::TypeTracker t, string attr_name) {
-          attr_name in ["annotate", "extra", "raw"] and
-          t.startInAttr(attr_name) and
-          result = objects()
-          or
-          // Due to bad performance when using normal setup with `objects_attr(t2, attr_name).track(t2, t)`
-          // we have inlined that code and forced a join
-          exists(DataFlow::TypeTracker t2 |
-            exists(DataFlow::StepSummary summary |
-              objects_attr_first_join(t2, attr_name, result, summary) and
-              t = t2.append(summary)
-            )
-          )
-        }
+        API::Node manager() { result = Model::subclassRef().getMember("objects") }
 
-        pragma[nomagic]
-        private predicate objects_attr_first_join(
-          DataFlow::TypeTracker t2, string attr_name, DataFlow::Node res,
-          DataFlow::StepSummary summary
-        ) {
-          DataFlow::StepSummary::step(objects_attr(t2, attr_name), res, summary)
+        /**
+         * Gets a method with `name` that returns a QuerySet.
+         * This method can originate on a QuerySet or a Manager.
+         *
+         * See https://docs.djangoproject.com/en/3.1/ref/models/querysets/
+         */
+        API::Node querySetReturningMethod(string name) {
+          name in [
+              "none", "all", "filter", "exclude", "complex_filter", "union", "intersection",
+              "difference", "select_for_update", "select_related", "prefetch_related", "order_by",
+              "distinct", "reverse", "defer", "only", "using", "annotate", "extra", "raw",
+              "datetimes", "dates", "values", "values_list"
+            ] and
+          result = [manager(), querySet()].getMember(name)
         }
 
         /**
-         * Gets a reference to the attribute `attr_name` of an `objects` object.
-         * WARNING: Only holds for a few predefined attributes.
+         * Gets a reference to a QuerySet (django.db.models.query.QuerySet).
+         *
+         * See https://docs.djangoproject.com/en/3.1/ref/models/querysets/
          */
-        DataFlow::Node objects_attr(string attr_name) {
-          result = objects_attr(DataFlow::TypeTracker::end(), attr_name)
-        }
+        API::Node querySet() { result = querySetReturningMethod(_).getReturn() }
 
         /** Gets a reference to the `django.db.models.expressions` module. */
         private DataFlow::Node expressions(DataFlow::TypeTracker t) {
@@ -254,14 +231,13 @@ private module Django {
      *
      * See https://docs.djangoproject.com/en/3.1/ref/models/querysets/#annotate
      */
-    private class ObjectsAnnotate extends SqlExecution::Range, DataFlow::CfgNode {
-      override CallNode node;
+    private class ObjectsAnnotate extends SqlExecution::Range, DataFlow::CallCfgNode {
       ControlFlowNode sql;
 
       ObjectsAnnotate() {
-        node.getFunction() = django::db::models::objects_attr("annotate").asCfgNode() and
-        django::db::models::expressions::RawSQL::instance(sql).asCfgNode() in [
-            node.getArg(_), node.getArgByName(_)
+        this = django::db::models::querySetReturningMethod("annotate").getACall() and
+        django::db::models::expressions::RawSQL::instance(sql) in [
+            this.getArg(_), this.getArgByName(_)
           ]
       }
 
@@ -275,27 +251,24 @@ private module Django {
      * - https://docs.djangoproject.com/en/3.1/topics/db/sql/#django.db.models.Manager.raw
      * - https://docs.djangoproject.com/en/3.1/ref/models/querysets/#raw
      */
-    private class ObjectsRaw extends SqlExecution::Range, DataFlow::CfgNode {
-      override CallNode node;
-
-      ObjectsRaw() { node.getFunction() = django::db::models::objects_attr("raw").asCfgNode() }
+    private class ObjectsRaw extends SqlExecution::Range, DataFlow::CallCfgNode {
+      ObjectsRaw() { this = django::db::models::querySetReturningMethod("raw").getACall() }
 
-      override DataFlow::Node getSql() { result.asCfgNode() = node.getArg(0) }
+      override DataFlow::Node getSql() { result = this.getArg(0) }
     }
 
     /**
      * A call to the `extra` function on a model.
      *
      * See https://docs.djangoproject.com/en/3.1/ref/models/querysets/#extra
      */
-    private class ObjectsExtra extends SqlExecution::Range, DataFlow::CfgNode {
-      override CallNode node;
-
-      ObjectsExtra() { node.getFunction() = django::db::models::objects_attr("extra").asCfgNode() }
+    private class ObjectsExtra extends SqlExecution::Range, DataFlow::CallCfgNode {
+      ObjectsExtra() { this = django::db::models::querySetReturningMethod("extra").getACall() }
 
       override DataFlow::Node getSql() {
-        result.asCfgNode() =
-          [node.getArg([0, 1, 3, 4]), node.getArgByName(["select", "where", "tables", "order_by"])]
+        result in [
+            this.getArg([0, 1, 3, 4]), this.getArgByName(["select", "where", "tables", "order_by"])
+          ]
       }
     }
 

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

@@ -27,3 +27,6 @@ def test_model():
 
     raw = RawSQL("so raw")
     User.objects.annotate(val=raw)  # $getSql="so raw"
+
+    # chaining QuerySet calls
+    User.objects.using("db-name").exclude(username="admin").extra("some sql") # $ getSql="some sql"