v1

python/ql/lib/semmle/python/Frameworks.qll

@@ -21,6 +21,7 @@ private import semmle.python.frameworks.Dill
 private import semmle.python.frameworks.Django
 private import semmle.python.frameworks.Fabric
 private import semmle.python.frameworks.FastApi
+private import semmle.python.frameworks.FileSystemAccess
 private import semmle.python.frameworks.Flask
 private import semmle.python.frameworks.FlaskAdmin
 private import semmle.python.frameworks.FlaskSqlAlchemy
@@ -51,6 +52,7 @@ private import semmle.python.frameworks.Requests
 private import semmle.python.frameworks.RestFramework
 private import semmle.python.frameworks.Rsa
 private import semmle.python.frameworks.RuamelYaml
+private import semmle.python.frameworks.Sanic
 private import semmle.python.frameworks.ServerLess
 private import semmle.python.frameworks.Simplejson
 private import semmle.python.frameworks.SqlAlchemy

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

@@ -0,0 +1,95 @@
+/**
+ * Provides classes modeling security-relevant aspects of the I/O write or read operations
+ */
+
+private import python
+private import semmle.python.dataflow.new.DataFlow
+private import semmle.python.dataflow.new.RemoteFlowSources
+private import semmle.python.dataflow.new.TaintTracking
+private import semmle.python.Concepts
+private import semmle.python.ApiGraphs
+
+/**
+ * Provides models for the `aiofile` PyPI package.
+ * See https://github.com/agronholm/anyio.
+ */
+private module Aiofile {
+  /**
+   * A call to the `async_open` function or `AIOFile` constructor from `aiofile` as a sink for Filesystem access.
+   */
+  class FileResponseCall extends FileSystemAccess::Range, API::CallNode {
+    string methodName;
+
+    FileResponseCall() {
+      this = API::moduleImport("aiofile").getMember("async_open").getACall() and
+      methodName = "async_open"
+      or
+      this = API::moduleImport("aiofile").getMember("AIOFile").getACall() and
+      methodName = "AIOFile"
+    }
+
+    override DataFlow::Node getAPathArgument() {
+      result = this.getParameter(0, "file_specifier").asSink() and
+      methodName = "async_open"
+      or
+      result = this.getParameter(0, "filename").asSink() and
+      methodName = "AIOFile"
+    }
+  }
+}
+
+/**
+ * Provides models for the `aiofiles` PyPI package.
+ * See https://github.com/Tinche/aiofiles.
+ */
+private module Aiofiles {
+  /**
+   * A call to the `open` function from `aiofiles` as a sink for Filesystem access.
+   */
+  class FileResponseCall extends FileSystemAccess::Range, API::CallNode {
+    FileResponseCall() { this = API::moduleImport("aiofiles").getMember("open").getACall() }
+
+    override DataFlow::Node getAPathArgument() { result = this.getParameter(0, "file").asSink() }
+  }
+}
+
+/**
+ * Provides models for the `anyio` PyPI package.
+ * See https://github.com/agronholm/anyio.
+ */
+private module Anyio {
+  /**
+   * A call to the `from_path` function from `FileReadStream` or `FileWriteStream` constructors of `anyio.streams.file` as a sink for Filesystem access.
+   */
+  class FileStreamCall extends FileSystemAccess::Range, API::CallNode {
+    FileStreamCall() {
+      this =
+        API::moduleImport("anyio")
+            .getMember("streams")
+            .getMember("file")
+            .getMember(["FileReadStream", "FileWriteStream"])
+            .getMember("from_path")
+            .getACall()
+    }
+
+    override DataFlow::Node getAPathArgument() { result = this.getParameter(0, "path").asSink() }
+  }
+
+  /**
+   * A call to the `Path` constructor from `anyio` as a sink for Filesystem access.
+   */
+  class PathCall extends FileSystemAccess::Range, API::CallNode {
+    PathCall() { this = API::moduleImport("anyio").getMember("Path").getACall() }
+
+    override DataFlow::Node getAPathArgument() { result = this.getParameter(0).asSink() }
+  }
+
+  /**
+   * A call to the `open_file` function from `anyio` as a sink for Filesystem access.
+   */
+  class OpenFileCall extends FileSystemAccess::Range, API::CallNode {
+    OpenFileCall() { this = API::moduleImport("anyio").getMember("open_file").getACall() }
+
+    override DataFlow::Node getAPathArgument() { result = this.getParameter(0, "file").asSink() }
+  }
+}

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

@@ -0,0 +1,42 @@
+/**
+ * Provides classes modeling security-relevant aspects of the `sanic` PyPI package.
+ * See https://sanic.dev/.
+ */
+
+private import python
+private import semmle.python.dataflow.new.DataFlow
+private import semmle.python.dataflow.new.RemoteFlowSources
+private import semmle.python.dataflow.new.TaintTracking
+private import semmle.python.Concepts
+private import semmle.python.ApiGraphs
+
+/**
+ * Provides models for the `sanic` PyPI package.
+ * See https://sanic.dev/.
+ */
+private module Sanic {
+  /**
+   * Provides models for Sanic applications (an instance of `sanic.Sanic`).
+   */
+  module App {
+    /** Gets a reference to a Sanic application (an instance of `sanic.Sanic`). */
+    API::Node instance() { result = API::moduleImport("sanic").getMember("Sanic").getReturn() }
+  }
+
+  /**
+   * A call to the `file` or `file_stream` functions of `sanic.response` as a sink for Filesystem access.
+   */
+  class FileResponseCall extends FileSystemAccess::Range, API::CallNode {
+    FileResponseCall() {
+      this =
+        API::moduleImport("sanic")
+            .getMember("response")
+            .getMember(["file", "file_stream"])
+            .getACall()
+    }
+
+    override DataFlow::Node getAPathArgument() {
+      result = this.getParameter(0, "location").asSink()
+    }
+  }
+}

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

@@ -163,4 +163,31 @@ module Starlette {
 
   /** DEPRECATED: Alias for Url */
   deprecated module URL = Url;
+
+  /**
+   * A call to the `starlette.responses.FileResponse` constructor as a sink for Filesystem access.
+   */
+  class FileResponseCall extends FileSystemAccess::Range, API::CallNode {
+    FileResponseCall() {
+      this =
+        API::moduleImport("starlette").getMember("responses").getMember("FileResponse").getACall()
+    }
+
+    override DataFlow::Node getAPathArgument() { result = this.getParameter(0, "path").asSink() }
+  }
+
+  /**
+   * A call to the `baize.asgi.FileResponse` constructor as a sink for Filesystem access.
+   *
+   * it is not contained to Starlette source code but it is mentioned as an alternative to Starlette FileResponse
+   */
+  class BaizeFileResponseCall extends FileSystemAccess::Range, API::CallNode {
+    BaizeFileResponseCall() {
+      this = API::moduleImport("baize").getMember("asgi").getMember("FileResponse").getACall()
+    }
+
+    override DataFlow::Node getAPathArgument() {
+      result = this.getParameter(0, "filepath").asSink()
+    }
+  }
 }

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

@@ -1479,6 +1479,26 @@ private module StdlibPrivate {
     }
   }
 
+  /**
+   * A call to the `io.FileIO` constructor.
+   * See https://docs.python.org/3/library/io.html#io.FileIO
+   */
+  private class FileIOCall extends FileSystemAccess::Range, API::CallNode {
+    FileIOCall() { this = API::moduleImport("io").getMember("FileIO").getACall() }
+
+    override DataFlow::Node getAPathArgument() { result = this.getParameter(0, "file").asSink() }
+  }
+
+  /**
+   * A call to the `io.open_code` function.
+   * See https://docs.python.org/3/library/io.html#io.FileIO
+   */
+  private class OpenCodeCall extends FileSystemAccess::Range, API::CallNode {
+    OpenCodeCall() { this = API::moduleImport("io").getMember("open_code").getACall() }
+
+    override DataFlow::Node getAPathArgument() { result = this.getParameter(0, "path").asSink() }
+  }
+
   /** Gets a reference to an open file. */
   private DataFlow::TypeTrackingNode openFile(DataFlow::TypeTracker t, FileSystemAccess openCall) {
     t.start() and

python/ql/test/library-tests/frameworks/FileSystemAccess/ConceptsTest.expected

@@ -0,0 +1,2 @@
+testFailures
+failures

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

@@ -0,0 +1,2 @@
+import python
+import experimental.meta.ConceptsTest

python/ql/test/library-tests/frameworks/FileSystemAccess/Query.ql

@@ -0,0 +1,20 @@
+import python
+import semmle.python.dataflow.new.DataFlow
+import semmle.python.Concepts
+import TestUtilities.InlineExpectationsTest
+private import semmle.python.dataflow.new.internal.PrintNode
+
+module FileSystemAccessTest implements TestSig {
+  string getARelevantTag() { result = "getAPathArgument" }
+
+  predicate hasActualResult(Location location, string element, string tag, string value) {
+    exists(location.getFile().getRelativePath()) and
+    exists(FileSystemAccess a, DataFlow::Node path |
+      path = a.getAPathArgument() and
+      location = a.getLocation() and
+      element = path.toString() and
+      value = prettyNodeForInlineTest(path) and
+      tag = "getAPathArgument"
+    )
+  }
+}

python/ql/test/library-tests/frameworks/FileSystemAccess/aiofile.py

@@ -0,0 +1,4 @@
+from aiofile import async_open, AIOFile
+
+AIOFile("file", 'r') # $ getAPathArgument="file"
+async_open("file", "r") # $ getAPathArgument="file"

python/ql/test/library-tests/frameworks/FileSystemAccess/aiofiles.py

@@ -0,0 +1,3 @@
+import aiofiles
+
+aiofiles.open("file", mode='r')  # $ getAPathArgument="file"

python/ql/test/library-tests/frameworks/FileSystemAccess/anyio.py

@@ -0,0 +1,8 @@
+import anyio
+from anyio.streams.file import FileReadStream, FileWriteStream
+from anyio import Path
+
+anyio.open_file("file", 'r') # $ getAPathArgument="file"
+FileReadStream.from_path("file") # $ getAPathArgument="file"
+FileWriteStream.from_path("file") # $ getAPathArgument="file"
+Path("file") # $ getAPathArgument="file"

python/ql/test/library-tests/frameworks/FileSystemAccess/sanic.py

@@ -0,0 +1,4 @@
+from sanic import response
+
+response.file("file")  # $ getAPathArgument="file"
+response.file_stream("file")  # $ getAPathArgument="file"

python/ql/test/library-tests/frameworks/FileSystemAccess/starlette.py

@@ -0,0 +1,5 @@
+from starlette.responses import FileResponse
+from baize.asgi import FileResponse as baizeFileResponse
+
+baizeFileResponse("file")  # $ getAPathArgument="file"
+FileResponse("file")  # $ getAPathArgument="file"

python/ql/test/library-tests/frameworks/stdlib/ConceptsTest.expected

@@ -1,2 +1,2 @@
-failures
 testFailures
+failures

python/ql/test/library-tests/frameworks/stdlib/FileSystemAccess.py

@@ -20,6 +20,8 @@ builtins.open(file="file")  # $ getAPathArgument="file"
 
 io.open("file")  # $ getAPathArgument="file"
 io.open(file="file")  # $ getAPathArgument="file"
+io.open_code("file")  # $ getAPathArgument="file"
+io.FileIO("file")  # $ getAPathArgument="file"
 
 f = open("path") # $ getAPathArgument="path"
 f.write("foo") # $ getAPathArgument="path" fileWriteData="foo"