Python: Model FastAPI requests

python/ql/lib/change-notes/2024-12-18-fastapi-request-modeling.md

@@ -0,0 +1,4 @@
+---
+category: minorAnalysis
+---
+* Added modeling of `fastapi.Request`, which will improve taint-flow for users of FastAPI.

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

@@ -245,4 +245,60 @@
 
     override DataFlow::Node getAPathArgument() { result = this.getParameter(0, "path").asSink() }
   }
+
+  /**
+   * Provides models for the `starlette.requests.Request` class
+   *
+   * See https://www.starlette.io/requests/.
+   */
+  module Request {
+    /** Gets a reference to the `starlette.requests.Request` class. */
+    API::Node classRef() {
+      result = API::moduleImport("starlette").getMember("requests").getMember("Request")
+      or
+      result = API::moduleImport("fastapi").getMember("Request")
+    }
+
+    /**
+     * A source of instances of `starlette.requests.Request`, extend this class to model new instances.
+     *
+     * This can include instantiations of the class, return values from function
+     * calls, or a special parameter that will be set when functions are called by an external
+     * library.
+     *
+     * Use the predicate `Request::instance()` to get references to instances of `starlette.requests.Request`.
+     */
+    abstract class InstanceSource extends DataFlow::LocalSourceNode { }
+
+    /** A direct instantiation of `starlette.requests.Request`. */
+    private class ClassInstantiation extends InstanceSource {
+      ClassInstantiation() { this = classRef().getAnInstance().asSource() }
+    }
+
+    /** Gets a reference to an instance of `starlette.requests.Request`. */
+    private DataFlow::TypeTrackingNode instance(DataFlow::TypeTracker t) {
+      t.start() and
+      result instanceof InstanceSource
+      or
+      exists(DataFlow::TypeTracker t2 | result = instance(t2).track(t2, t))
+    }
+
+    /** Gets a reference to an instance of `starlette.requests.Request`. */
+    DataFlow::Node instance() { instance(DataFlow::TypeTracker::end()).flowsTo(result) }
+
+    /**
+     * Taint propagation for `starlette.requests.Request`.
+     */
+    private class InstanceTaintSteps extends InstanceTaintStepsHelper {
+      InstanceTaintSteps() { this = "starlette.requests.Request" }
+
+      override DataFlow::Node getInstance() { result = instance() }
+
+      override string getAttributeName() { result in ["cookies"] }
+
+      override string getMethodName() { none() }
+
+      override string getAsyncMethodName() { result in ["body", "json", "form", "stream"] }
+    }
+  }
 }

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

@@ -187,3 +187,38 @@ async def websocket_test(websocket: WebSocket): # $ requestHandler routedParamet
 
     async for data in  websocket.iter_json():
         ensure_tainted(data) # $ tainted
+
+
+# --- Request ---
+
+import starlette.requests
+from fastapi import Request
+
+
+assert Request == starlette.requests.Request
+
+@app.websocket("/req") # $ routeSetup="/req"
+async def request_test(request: Request): # $ requestHandler routedParameter=request
+    ensure_tainted(
+        request, # $ tainted
+
+        await request.body(), # $ tainted
+
+        await request.json(), # $ tainted
+        await request.json()["key"], # $ tainted
+
+        # form() returns a FormData (which is a starlette ImmutableMultiDict)
+        await request.form(), # $ tainted
+        await request.form()["key"], # $ tainted
+        await request.form().getlist("key"), # $ MISSING: tainted
+        await request.form().getlist("key")[0], # $ MISSING: tainted
+        # data in the form could be an starlette.datastructures.UploadFile
+        await request.form()["file"].filename, # $ MISSING: tainted
+        await request.form().getlist("file")[0].filename, # $ MISSING: tainted
+
+        request.cookies, # $ tainted
+        request.cookies["key"], # $ tainted
+    )
+
+    async for chunk in request.stream():
+        ensure_tainted(chunk) # $ tainted