fix some qldocs, change Sink extenstion model, deduct some not necessarily checks :)

Author: am0o0

python/ql/src/experimental/Security/CWE-409/DecompressionBombs.ql

@@ -18,6 +18,7 @@ import semmle.python.ApiGraphs
 import semmle.python.dataflow.new.RemoteFlowSources
 import semmle.python.dataflow.new.internal.DataFlowPublic
 import experimental.semmle.python.security.DecompressionBomb
+import FileAndFormRemoteFlowSource::FileAndFormRemoteFlowSource
 
 /**
  * `io.TextIOWrapper(ip, encoding='utf-8')` like following:
@@ -39,59 +40,12 @@ predicate isAdditionalTaintStepTextIOWrapper(DataFlow::Node nodeFrom, DataFlow::
   )
 }
 
-module FileAndFormRemoteFlowSource {
-  class FastAPI extends RemoteFlowSource::Range {
-    FastAPI() {
-      exists(API::Node fastApiParam, Expr fastApiUploadFile |
-        fastApiParam =
-          API::moduleImport("fastapi")
-              .getMember("FastAPI")
-              .getReturn()
-              .getMember("post")
-              .getReturn()
-              .getParameter(0)
-              .getKeywordParameter(_) and
-        fastApiUploadFile =
-          API::moduleImport("fastapi")
-              .getMember("UploadFile")
-              .getASubclass*()
-              .getAValueReachableFromSource()
-              .asExpr()
-      |
-        fastApiUploadFile =
-          fastApiParam.asSource().asExpr().(Parameter).getAnnotation().getASubExpression*() and
-        // Multiple Uploaded files as list of fastapi.UploadFile
-        exists(For f, Attribute attr |
-          fastApiParam.getAValueReachableFromSource().asExpr() = f.getIter().getASubExpression*()
-        |
-          TaintTracking::localExprTaint(f.getIter(), attr.getObject()) and
-          attr.getName() = ["filename", "content_type", "headers", "file", "read"] and
-          this.asExpr() = attr
-        )
-        or
-        // one Uploaded file as fastapi.UploadFile
-        this =
-          [
-            fastApiParam.getMember(["filename", "content_type", "headers"]).asSource(),
-            fastApiParam
-                .getMember("file")
-                .getMember(["readlines", "readline", "read"])
-                .getReturn()
-                .asSource(), fastApiParam.getMember("read").getReturn().asSource()
-          ]
-      )
-    }
-
-    override string getSourceType() { result = "fastapi HTTP FORM files" }
-  }
-}
-
 module BombsConfig implements DataFlow::ConfigSig {
   predicate isSource(DataFlow::Node source) {
     (
       source instanceof RemoteFlowSource
       or
-      source instanceof FileAndFormRemoteFlowSource::FastAPI
+      source instanceof FastAPI
     ) and
     not source.getLocation().getFile().inStdlib() and
     not source.getLocation().getFile().getRelativePath().matches("%venv%")
@@ -113,11 +67,11 @@ module BombsConfig implements DataFlow::ConfigSig {
   }
 }
 
-module Bombs = TaintTracking::Global<BombsConfig>;
+module BombsFlow = TaintTracking::Global<BombsConfig>;
 
-import Bombs::PathGraph
+import BombsFlow::PathGraph
 
-from Bombs::PathNode source, Bombs::PathNode sink
-where Bombs::flowPath(source, sink)
+from BombsFlow::PathNode source, BombsFlow::PathNode sink
+where BombsFlow::flowPath(source, sink)
 select sink.getNode(), source, sink, "This uncontrolled file extraction is $@.", source.getNode(),
   "depends on this user controlled data"

python/ql/src/experimental/Security/CWE-409/FileAndFormRemoteFlowSource.qll

@@ -0,0 +1,63 @@
+import python
+import semmle.python.dataflow.new.DataFlow
+import semmle.python.dataflow.new.TaintTracking
+import semmle.python.ApiGraphs
+
+/**
+ * Provides user-controllable Remote sources for file(s) upload and Multipart-Form
+ */
+module FileAndFormRemoteFlowSource {
+  /**
+   * A
+   */
+  class FastAPI extends DataFlow::Node {
+    FastAPI() {
+      exists(API::Node fastApiParam, Expr fastApiUploadFile |
+        fastApiParam =
+          API::moduleImport("fastapi")
+              .getMember("FastAPI")
+              .getReturn()
+              .getMember("post")
+              .getReturn()
+              .getParameter(0)
+              .getKeywordParameter(_) and
+        fastApiUploadFile =
+          API::moduleImport("fastapi")
+              .getMember("UploadFile")
+              .getASubclass*()
+              .getAValueReachableFromSource()
+              .asExpr()
+      |
+        // Multiple uploaded files as list of fastapi.UploadFile
+        // @app.post("/")
+        //    def upload(files: List[UploadFile] = File(...)):
+        //        for file in files:
+        fastApiUploadFile =
+          fastApiParam.asSource().asExpr().(Parameter).getAnnotation().getASubExpression*() and
+        exists(For f, Attribute attr |
+          fastApiParam.getAValueReachableFromSource().asExpr() = f.getIter().getASubExpression*()
+        |
+          TaintTracking::localExprTaint(f.getIter(), attr.getObject()) and
+          attr.getName() = ["filename", "content_type", "headers", "file", "read"] and
+          this.asExpr() = attr
+        )
+        or
+        // One uploaded file as fastapi.UploadFile
+        // @app.post("/zipbomb2")
+        //    async def zipbomb2(file: UploadFile):
+        //        print(file.filename)
+        this =
+          [
+            fastApiParam.getMember(["filename", "content_type", "headers"]).asSource(),
+            fastApiParam
+                .getMember("file")
+                .getMember(["readlines", "readline", "read"])
+                .getReturn()
+                .asSource(), fastApiParam.getMember("read").getReturn().asSource()
+          ]
+      )
+    }
+
+    string getSourceType() { result = "fastapi HTTP FORM files" }
+  }
+}

python/ql/src/experimental/Security/CWE-409/example_good.py

@@ -2,20 +2,33 @@
 
 
 def safeUnzip(zipFileName):
-    buffer_size = 2 ** 10 * 8
+    '''
+    safeUnzip reads each file inside the zipfile 1 MB by 1 MB
+    and during extraction or reading of these files it checks the total decompressed size
+    doesn't exceed the SIZE_THRESHOLD
+    '''
+    buffer_size = 1024 * 1024 * 1  # 1 MB
     total_size = 0
-    SIZE_THRESHOLD = 2 ** 10 * 100
+    SIZE_THRESHOLD = 1024 * 1024 * 10  # 10 MB
     with zipfile.ZipFile(zipFileName) as myzip:
         for fileinfo in myzip.infolist():
-            content = b''
             with myzip.open(fileinfo.filename, mode="r") as myfile:
+                content = b''
                 chunk = myfile.read(buffer_size)
+                total_size += buffer_size
+                if total_size > SIZE_THRESHOLD:
+                    print("Bomb detected")
+                    return False  # it isn't a successful extract or read
+                content += chunk
+                # reading next bytes of uncompressed data
                 while chunk:
                     chunk = myfile.read(buffer_size)
                     total_size += buffer_size
                     if total_size > SIZE_THRESHOLD:
                         print("Bomb detected")
-                        return
+                        return False  # it isn't a successful extract or read
                     content += chunk
+
+                # An example of extracting or reading each decompressed file here
                 print(bytes.decode(content, 'utf-8'))
-    return "No Bombs!"
+    return True  # it is a successful extract or read