WIP: Python: CORS Bypass

python/ql/src/experimental/Security/CWE-346/CorsBad.py

@@ -0,0 +1,9 @@
+import cherrypy
+
+def bad():
+    request = cherrypy.request
+    validCors = "domain.com"
+    if request.method in ['POST', 'PUT', 'PATCH', 'DELETE']:
+        origin = request.headers.get('Origin', None)
+        if origin.startswith(validCors):
+            print("Origin Valid")

python/ql/src/experimental/Security/CWE-346/CorsBypass.qhelp

@@ -0,0 +1,28 @@
+<!DOCTYPE qhelp PUBLIC "-//Semmle//qhelp//EN" "qhelp.dtd">
+<qhelp>
+  <overview>
+    <p>Cross-origin resource sharing policy may be bypassed due to incorrect checks like the <code>string.startswith</code> call.</p>
+  </overview>
+  <recommendation>
+    <p>Use a more stronger check to test for CORS policy bypass.</p>
+  </recommendation>
+
+  <example>
+    <p>Most Python frameworks provide a mechanism for testing origins and performing CORS checks.
+      For example, consider the code snippet below, <code>origin</code> is compared using a <code>
+      startswith</code> call against a list of whitelisted origins. This check can be bypassed
+      easily by origin like <code>domain.com.baddomain.com</code>
+    </p>
+    <sample src="CorsBad.py" />
+    <p>This can be prevented by comparing the origin in a manner shown below.
+    </p>
+    <sample src="CorsGood.py" />
+
+  </example>
+
+  <references>
+    <li>PortsSwigger : <a href="https://portswigger.net/web-security/cors"></a>Cross-origin resource
+      sharing (CORS)</li>
+    <li>Related CVE: <a href="https://github.com/advisories/GHSA-824x-jcxf-hpfg">CVE-2022-3457</a>.</li>
+  </references>
+</qhelp>

python/ql/src/experimental/Security/CWE-346/CorsBypass.ql

@@ -0,0 +1,97 @@
+/**
+ * @name Cross Origin Resource Sharing(CORS) Policy Bypass
+ * @description Checking user supplied origin headers using weak comparators like 'string.startswith' may lead to CORS policy bypass.
+ * @kind path-problem
+ * @problem.severity warning
+ * @id py/cors-bypass
+ * @tags security
+ *       externa/cwe/CWE-346
+ */
+
+import python
+import semmle.python.ApiGraphs
+import semmle.python.dataflow.new.TaintTracking
+import semmle.python.Flow
+import semmle.python.dataflow.new.RemoteFlowSources
+
+/**
+ * Returns true if the control flow node may be useful in the current context.
+ *
+ * Ideally for more completeness, we should alert on every `startswith` call and every remote flow source which gets partailly checked. But, as this can lead to lots of FPs, we apply heuristics to filter some calls. This predicate provides logic for this filteration.
+ */
+private predicate maybeInteresting(ControlFlowNode c) {
+  // Check if the name of the variable which calls the function matches the heuristic.
+  // This would typically occur at the sink.
+  // This should deal with cases like
+  // `origin.startswith("bla")`
+  heuristics(c.(CallNode).getFunction().(AttrNode).getObject().(NameNode).getId())
+  or
+  // Check if the name of the variable passed as an argument to the functions matches the heuristic. This would typically occur at the sink.
+  // This should deal with cases like
+  // `bla.startswith(origin)`
+  heuristics(c.(CallNode).getArg(0).(NameNode).getId())
+  or
+  // Check if the value gets written to any interesting variable. This would typically occur at the source.
+  // This should deal with cases like
+  // `origin = request.headers.get('My-custom-header')`
+  exists(Variable v | heuristics(v.getId()) | c.getASuccessor*().getNode() = v.getAStore())
+}
+
+private class StringStartswithCall extends ControlFlowNode {
+  StringStartswithCall() { this.(CallNode).getFunction().(AttrNode).getName() = "startswith" }
+}
+
+bindingset[s]
+predicate heuristics(string s) { s.matches(["%origin%", "%cors%"]) }
+
+/**
+ * A member of the `cherrypy.request` class taken as a `RemoteFlowSource`.
+ */
+class CherryPyRequest extends RemoteFlowSource::Range {
+  CherryPyRequest() {
+    this =
+      API::moduleImport("cherrypy")
+          .getMember("request")
+          .getMember([
+              "charset", "content_type", "filename", "fp", "name", "params", "headers", "length",
+            ])
+          .asSource()
+  }
+
+  override string getSourceType() { result = "cherrypy.request" }
+}
+
+module CorsBypassConfig implements DataFlow::ConfigSig {
+  predicate isSource(DataFlow::Node node) { node instanceof RemoteFlowSource }
+
+  predicate isSink(DataFlow::Node node) {
+    exists(StringStartswithCall s |
+      node.asCfgNode() = s.(CallNode).getArg(0) or
+      node.asCfgNode() = s.(CallNode).getFunction().(AttrNode).getObject()
+    )
+  }
+
+  predicate isAdditionalFlowStep(DataFlow::Node node1, DataFlow::Node node2) {
+    exists(API::CallNode c, API::Node n |
+      n = API::moduleImport("cherrypy").getMember("request").getMember("headers") and
+      c = n.getMember("get").getACall()
+    |
+      c.getReturn().asSource() = node2 and n.asSource() = node1
+    )
+  }
+}
+
+module CorsFlow = TaintTracking::Global<CorsBypassConfig>;
+
+import CorsFlow::PathGraph
+
+from CorsFlow::PathNode source, CorsFlow::PathNode sink
+where
+  CorsFlow::flowPath(source, sink) and
+  (
+    maybeInteresting(source.getNode().asCfgNode())
+    or
+    maybeInteresting(sink.getNode().asCfgNode())
+  )
+select sink, source, sink,
+  "Potentially incorrect string comparison which could lead to a CORS bypass."

python/ql/src/experimental/Security/CWE-346/CorsGood.py

@@ -0,0 +1,9 @@
+import cherrypy
+
+def good():
+    request = cherrypy.request
+    validOrigin = "domain.com"
+    if request.method in ['POST', 'PUT', 'PATCH', 'DELETE']:
+        origin = request.headers.get('Origin', None)
+        if origin == validOrigin:
+            print("Origin Valid")

python/ql/test/experimental/query-tests/Security/CWE-346/Cors.py

@@ -0,0 +1,17 @@
+import cherrypy
+
+def bad():
+    request = cherrypy.request
+    validCors = "domain.com"
+    if request.method in ['POST', 'PUT', 'PATCH', 'DELETE']:
+        origin = request.headers.get('Origin', None)
+        if origin.startswith(validCors):
+            print("Origin Valid")
+
+def good():
+    request = cherrypy.request
+    validOrigin = "domain.com"
+    if request.method in ['POST', 'PUT', 'PATCH', 'DELETE']:
+        origin = request.headers.get('Origin', None)
+        if origin == validOrigin:
+            print("Origin Valid")

python/ql/test/experimental/query-tests/Security/CWE-346/CorsBypass.expected

@@ -0,0 +1,13 @@
+edges
+| Cors.py:7:9:7:14 | ControlFlowNode for origin | Cors.py:8:12:8:17 | ControlFlowNode for origin | provenance |  |
+| Cors.py:7:18:7:32 | ControlFlowNode for Attribute | Cors.py:7:18:7:52 | ControlFlowNode for Attribute() | provenance | Config |
+| Cors.py:7:18:7:32 | ControlFlowNode for Attribute | Cors.py:7:18:7:52 | ControlFlowNode for Attribute() | provenance | dict.get |
+| Cors.py:7:18:7:52 | ControlFlowNode for Attribute() | Cors.py:7:9:7:14 | ControlFlowNode for origin | provenance |  |
+nodes
+| Cors.py:7:9:7:14 | ControlFlowNode for origin | semmle.label | ControlFlowNode for origin |
+| Cors.py:7:18:7:32 | ControlFlowNode for Attribute | semmle.label | ControlFlowNode for Attribute |
+| Cors.py:7:18:7:52 | ControlFlowNode for Attribute() | semmle.label | ControlFlowNode for Attribute() |
+| Cors.py:8:12:8:17 | ControlFlowNode for origin | semmle.label | ControlFlowNode for origin |
+subpaths
+#select
+| Cors.py:8:12:8:17 | ControlFlowNode for origin | Cors.py:7:18:7:32 | ControlFlowNode for Attribute | Cors.py:8:12:8:17 | ControlFlowNode for origin | Potentially incorrect string comparison which could lead to a CORS bypass. |

python/ql/test/experimental/query-tests/Security/CWE-346/CorsBypass.qlref

@@ -0,0 +1 @@
+experimental/Security/CWE-346/CorsBypass.ql