Add TarSlip Improv query

Author: Sim4n6

python/ql/src/experimental/Security/CWE-022bis/TarSlip.qhelp

@@ -0,0 +1,60 @@
+<!DOCTYPE qhelp PUBLIC
+  "-//Semmle//qhelp//EN"
+  "qhelp.dtd">
+<qhelp>
+
+<overview>
+<p>Extracting files from a malicious tarball without validating that the destination file path
+is within the destination directory can cause files outside the destination directory to be
+overwritten, due to the possible presence of directory traversal elements (<code>..</code>) in
+archive path names.</p>
+
+<p>Tarball contain archive entries representing each file in the archive. These entries
+include a file path for the entry, but these file paths are not restricted and may contain
+unexpected special elements such as the directory traversal element (<code>..</code>). If these
+file paths are used to determine an output file to write the contents of the archive item to, then
+the file may be written to an unexpected location. This can result in sensitive information being
+revealed or deleted, or an attacker being able to influence behavior by modifying unexpected
+files.</p>
+
+<p>For example, if a tarball contains a file entry <code>../sneaky-file</code>, and the tarball
+is extracted to the directory <code>/tmp/tmp123</code>, then naively combining the paths would result
+in an output file path of <code>/tmp/tmp123/../sneaky-file</code>, which would cause the file to be
+written to <code>/tmp/</code>.</p>
+
+</overview>
+<recommendation>
+
+<p>Ensure that output paths constructed from tarball entries are validated
+to prevent writing files to unexpected locations.</p>
+
+<p>The recommended way of writing an output file from a tarball entry is to call <code>extract()</code> or <code>extractall()</code>.
+</p>
+
+</recommendation>
+
+<example>
+<p>
+In this example an archive is extracted without validating file paths.
+</p>
+
+<sample src="examples/TarSlip_1.py" />
+
+<p>To fix this vulnerability, we need to call the function <code>extractall()</code>.
+</p>
+
+<sample src="examples/NoHIT_TarSlip_1.py" />
+
+</example>
+<references>
+<li>
+Snyk:
+<a href="https://snyk.io/research/zip-slip-vulnerability">Zip Slip Vulnerability</a>.
+</li>
+
+<li>
+  Tarfile documentation
+  <a href="https://docs.python.org/3/library/tarfile.html#tarfile.TarFile.extractall">extractall() warning</a>
+  </li>
+</references>
+</qhelp>

python/ql/src/experimental/Security/CWE-022bis/TarSlipImprov.ql

@@ -0,0 +1,108 @@
+/**
+ * @name Arbitrary file write during tarfile extraction
+ * @description Extracting files from a malicious tar archive without validating that the
+ *              destination file path is within the destination directory can cause files outside
+ *              the destination directory to be overwritten.
+ * @kind path-problem
+ * @id py/tarslip
+ * @problem.severity error
+ * @security-severity 7.5
+ * @precision high
+ * @tags security
+ *       external/cwe/cwe-022
+ */
+
+import python
+import semmle.python.dataflow.new.DataFlow
+import semmle.python.dataflow.new.TaintTracking
+import DataFlow::PathGraph
+import semmle.python.ApiGraphs
+import semmle.python.dataflow.new.internal.Attributes
+import semmle.python.dataflow.new.BarrierGuards
+import semmle.python.dataflow.new.RemoteFlowSources
+
+/**
+ * Handle those three cases of Tarfile opens:
+ *  - `tarfile.open()`
+ *  - `tarfile.TarFile()`
+ *  - `MKtarfile.Tarfile.open()`
+ */
+API::Node tarfileOpen() {
+  result in [
+      API::moduleImport("tarfile").getMember(["open", "TarFile"]),
+      API::moduleImport("tarfile").getMember("TarFile").getASubclass().getMember("open")
+    ]
+}
+
+/**
+ * Handle the previous three cases, plus the use of `closing` in the previous cases
+ */
+class AllTarfileOpens extends API::CallNode {
+  AllTarfileOpens() {
+    this = tarfileOpen().getACall()
+    or
+    exists(API::Node closing, Node arg |
+      closing = API::moduleImport("contextlib").getMember("closing") and
+      this = closing.getACall() and
+      arg = this.getArg(0) and
+      arg = tarfileOpen().getACall()
+    )
+  }
+}
+/**
+ * A taint-tracking configuration for detecting more "TarSlip" vulnerabilities.
+ */
+class Configuration extends TaintTracking::Configuration {
+  Configuration() { this = "TarSlip" }
+
+  override predicate isSource(DataFlow::Node source) { source instanceof AllTarfileOpens }
+
+  override predicate isSink(DataFlow::Node sink) {
+    // A sink capturing method calls to `extractall` without `members` argument.
+    // For a call to `file.extractall` without `members` argument, `file` is considered a sink.
+    exists(MethodCallNode call , AllTarfileOpens atfo|
+      call = atfo.getReturn().getMember("extractall").getACall() and
+      not exists(Node arg | arg = call.getArgByName("members")) and
+      sink = call.getObject()
+    )
+    or
+    // A sink capturing method calls to `extractall` with `members` argument.
+    // For a call to `file.extractall` with `members` argument, `file` is considered a sink if not
+    // a the `members` argument contains a NameConstant as None, a List or call to the method `getmembers`.
+    // Otherwise, the argument of `members` is considered a sink.
+    exists(MethodCallNode call, Node arg, AllTarfileOpens atfo|
+      call = atfo.getReturn().getMember("extractall").getACall() and
+      arg = call.getArgByName("members") and
+      if
+        arg.asCfgNode() instanceof NameConstantNode or
+        arg.asCfgNode() instanceof ListNode
+      then sink = call.getObject()
+      else
+        if arg.(MethodCallNode).getMethodName() = "getmembers"
+        then sink = arg.(MethodCallNode).getObject()
+        else sink = call.getArgByName("members")
+    )
+    or
+    // An argument to `extract` is considered a sink.
+    exists(AllTarfileOpens atfo | sink = atfo.getReturn().getMember("extract").getACall().getArg(0))
+    or
+    //An argument to `_extract_member` is considered a sink.
+    exists(MethodCallNode call, AllTarfileOpens atfo |
+      call = atfo.getReturn().getMember("_extract_member").getACall() and
+      call.getArg(1).(AttrRead).accesses(sink, "name")
+    )
+  }
+
+  override predicate isAdditionalTaintStep(DataFlow::Node nodeFrom, DataFlow::Node nodeTo) {
+    exists(AttrRead attr, MethodCallNode call |
+      attr.accesses(nodeFrom, "getmembers") and
+      nodeFrom = call.getObject() and
+      nodeTo = call
+    )
+  }
+}
+
+from Configuration config, DataFlow::PathNode source, DataFlow::PathNode sink
+where config.hasFlowPath(source, sink)
+select sink, source, sink, "Extraction of tarfile from $@ to a potentially untrusted source $@.",
+  source.getNode(), source.getNode().toString(), sink.getNode(), sink.getNode().toString()

python/ql/src/experimental/Security/CWE-022bis/examples/NoHIT_TarSlip_1.py

@@ -0,0 +1,19 @@
+import sys
+import tarfile
+
+def managed_members_archive_handler(filename):
+    tar = tarfile.open(filename)
+    result = []
+    for member in tar:
+        if ".." in member.name:
+            raise ValueError("Path in member name !!!")    
+        result.append(member)
+    path = sys.argv[2]
+    #print("files are extracted to: ", path)
+    tar.extractall(path=path, members=result)
+    tar.close()
+
+if __name__ == "__main__":
+    if len(sys.argv) > 1:
+        filename = sys.argv[1]
+        managed_members_archive_handler(filename)

python/ql/src/experimental/Security/CWE-022bis/examples/NoHIT_TarSlip_2.py

@@ -0,0 +1,27 @@
+import sys
+import tarfile
+import tempfile
+
+def managed_members_archive_handler(filename):
+    tar = tarfile.open(filename)
+    tar.extractall(path=tempfile.mkdtemp(), members=members_filter(tar))
+    tar.close()
+
+
+def members_filter(tarfile):
+    result = []
+    for member in tarfile:
+        if '../' in member.name:
+            print('Member name container directory traversal sequence')
+            continue
+        elif member.issym() or member.islnk():
+            print('Symlink to external resource')
+            continue    
+        result.append(member)
+    return result
+
+
+if __name__ == "__main__":
+    if len(sys.argv) > 1:
+        filename = sys.argv[1]
+        managed_members_archive_handler(filename)

python/ql/src/experimental/Security/CWE-022bis/examples/NoHIT_TarSlip_3.py

@@ -0,0 +1,8 @@
+import tarfile
+import sys
+
+with tarfile.open(sys.argv[1]) as tar:
+    for entry in tar:
+        if ".." in entry.name:
+            raise ValueError("Illegal tar archive entry")
+        tar.extract(entry, "/tmp/unpack/")

python/ql/src/experimental/Security/CWE-022bis/examples/NoHIT_TarSlip_4.py

@@ -0,0 +1,13 @@
+import tarfile
+import sys
+import os
+
+def _validate_archive_name(name, target):
+    if not os.path.abspath(os.path.join(target, name)).startswith(target + os.path.sep):
+        raise ValueError(f"Provided language pack contains invalid name {name}")
+
+with tarfile.open(sys.argv[1]) as tar:
+    target = "/tmp/unpack"
+    for entry in tar:
+        _validate_archive_name(entry.name, target)
+        tar.extract(entry, target)

python/ql/src/experimental/Security/CWE-022bis/examples/NoHIT_TarSlip_5.py

@@ -0,0 +1,29 @@
+# https://github.com/PyCQA/bandit
+
+import sys
+import tarfile
+import tempfile
+
+def managed_members_archive_handler(filename):
+    tar = tarfile.open(filename)
+    tar.extractall(path=tempfile.mkdtemp(), members=members_filter(tar))
+    tar.close()
+
+
+def members_filter(tarfile):
+    result = []
+    for member in tarfile.getmembers():
+        if '../' in member.name:
+            print('Member name container directory traversal sequence')
+            continue
+        elif (member.issym() or member.islnk()) and ('../' in member.linkname):
+            print('Symlink to external resource')
+            continue    
+        result.append(member)
+    return result
+
+
+if __name__ == "__main__":
+    if len(sys.argv) > 1:
+        filename = sys.argv[1]
+        managed_members_archive_handler(filename)

python/ql/src/experimental/Security/CWE-022bis/examples/NoHIT_TarSlip_6.py

@@ -0,0 +1,26 @@
+# https://github.com/OctoPrint/OctoPrint/
+
+import sys
+import tarfile
+import os
+
+def _validate_tar_info(info, target):
+    _validate_archive_name(info.name, target)
+    if not (info.isfile() or info.isdir()):
+        raise ValueError("Provided language pack contains invalid file type")
+
+def _validate_archive_name(name, target):
+    if not os.path.abspath(os.path.join(target, name)).startswith(target + os.path.sep):
+        raise ValueError(f"Provided language pack contains invalid name {name}")
+
+target = "/tmp/unpack"
+
+with tarfile.open(sys.argv[1], "r") as tar:
+
+    # sanity check
+    for info in tar.getmembers():
+        _validate_tar_info(info, target)
+
+    # unpack everything
+    tar.extractall(target)
+

python/ql/src/experimental/Security/CWE-022bis/examples/NoHIT_TarSlip_7.py

@@ -0,0 +1,29 @@
+# https://github.com/PyCQA/bandit
+
+import sys
+import tarfile
+import tempfile
+
+def managed_members_archive_handler(filename):
+    tar = tarfile.open(filename)
+    tar.extractall(path=tempfile.mkdtemp(), members=members_filter(tar))
+    tar.close()
+
+
+def members_filter(tarfile):
+    result = []
+    for member in tarfile.getmembers():
+        if '../' in member.name:
+            print('Member name container directory traversal sequence')
+            continue
+        elif member.issym() or member.islnk():
+            print('Symlink to external resource')
+            continue    
+        result.append(member)
+    return result
+
+
+if __name__ == "__main__":
+    if len(sys.argv) > 1:
+        filename = sys.argv[1]
+        managed_members_archive_handler(filename)

python/ql/src/experimental/Security/CWE-022bis/examples/TarSlip_1.py

@@ -0,0 +1,18 @@
+# https://github.com/PyCQA/bandit
+
+import sys
+import tarfile
+import tempfile
+
+def managed_members_archive_handler(filename):
+    tar = tarfile.open(filename)
+    tarf = tar.getmembers()
+    for f in tarf:
+        if not f.issym():
+            tar.extractall(path=tempfile.mkdtemp(), members=[f])
+    tar.close()
+
+if __name__ == "__main__":
+    if len(sys.argv) > 1:
+        filename = sys.argv[1]
+        managed_members_archive_handler(filename)