Finish query

python/ql/src/experimental/Security/CWE-347/JWTEmptyKeyOrAlgorithm.ql

@@ -10,15 +10,15 @@
 // determine precision above
 import python
 import experimental.semmle.python.Concepts
-import experimental.semmle.python.security.JWT
+import experimental.semmle.python.frameworks.JWT
 
 from JWTEncoding jwtEncoding, string affectedComponent
 where
   exists( |
     affectedComponent = "algorithm" and
-    isEmptyOrNone(jwtEncoding.getAlgorithmNode())
+    isEmptyOrNone(jwtEncoding.getAlgorithm())
     or
     affectedComponent = "key" and
-    isEmptyOrNone(jwtEncoding.getKeyNode())
+    isEmptyOrNone(jwtEncoding.getKey())
   )
 select jwtEncoding, affectedComponent, "is empty."

python/ql/src/experimental/Security/CWE-347/JWTMissingSecretOrPublicKeyVerification.ql

@@ -13,5 +13,5 @@ import python
 import experimental.semmle.python.Concepts
 
 from JWTDecoding jwtDecoding
-where not jwtDecoding.verifiesSignature()
+where jwtDecoding.verifiesSignature() = false
 select jwtDecoding, "does not verify the JWT payload with a cryptographic secret or public key."

python/ql/src/experimental/Security/CWE-347/tests/JWTEmptyKeyOrAlgorithm.qlref

@@ -0,0 +1 @@
+experimental/Security/CWE-347/JWTEmptyKeyOrAlgorithm.ql

python/ql/src/experimental/Security/CWE-347/tests/JWTMissingSecretOrPublicKeyVerification.qlref

@@ -0,0 +1 @@
+experimental/Security/CWE-347/JWTMissingSecretOrPublicKeyVerification.ql

python/ql/src/experimental/Security/CWE-347/tests/pyjwt.py

@@ -2,7 +2,7 @@ import jwt
 
 # Encoding
 
-# good - key and algorithm
+# good - key and algorithm supplied
 jwt.encode({"foo": "bar"}, "key", "HS256")
 jwt.encode({"foo": "bar"}, key="key", algorithm="HS256")
 
@@ -21,3 +21,7 @@ jwt.decode(token, "key", "HS256")
 # bad - unverified decoding
 jwt.decode(token, verify=False)
 jwt.decode(token, key, options={"verify_signature": False})
+
+# good - verified decoding
+jwt.decode(token, verify=True)
+jwt.decode(token, key, options={"verify_signature": True})

python/ql/src/experimental/semmle/python/Concepts.qll

@@ -23,20 +23,25 @@ module JWTEncoding {
    * extend `JWTEncoding` instead.
    */
   abstract class Range extends DataFlow::Node {
+    /**
+     * Gets the argument containing the encoding payload.
+     */
+    abstract DataFlow::Node getPayload();
+
     /**
      * Gets the argument containing the encoding key.
      */
-    abstract DataFlow::Node getKeyNode();
+    abstract DataFlow::Node getKey();
 
     /**
      * Gets the algorithm Node used in the encoding.
      */
-    abstract DataFlow::Node getAlgorithmNode();
+    abstract DataFlow::Node getAlgorithm();
 
     /**
      * Tries to get the algorithm used in the encoding.
      */
-    abstract string getAlgorithm();
+    abstract string getAlgorithmString();
   }
 }
 
@@ -51,11 +56,25 @@ class JWTEncoding extends DataFlow::Node {
 
   JWTEncoding() { this = range }
 
-  DataFlow::Node getKeyNode() { result = range.getKeyNode() }
+  /**
+   * Gets the argument containing the payload.
+   */
+  DataFlow::Node getPayload() { result = range.getPayload() }
 
-  DataFlow::Node getAlgorithmNode() { result = range.getAlgorithmNode() }
+  /**
+   * Gets the argument containing the encoding key.
+   */
+  DataFlow::Node getKey() { result = range.getKey() }
 
-  string getAlgorithm() { result = range.getAlgorithm() }
+  /**
+   * Gets the algorithm Node used in the encoding.
+   */
+  DataFlow::Node getAlgorithm() { result = range.getAlgorithm() }
+
+  /**
+   * Tries to get the algorithm used in the encoding.
+   */
+  string getAlgorithmString() { result = range.getAlgorithmString() }
 }
 
 /** Provides classes for modeling JWT decoding-related APIs. */
@@ -67,7 +86,35 @@ module JWTDecoding {
    * extend `JWTDecoding` instead.
    */
   abstract class Range extends DataFlow::Node {
-    abstract predicate verifiesSignature();
+    /**
+     * Gets the argument containing the encoding payload.
+     */
+    abstract DataFlow::Node getPayload();
+
+    /**
+     * Gets the argument containing the encoding key.
+     */
+    abstract DataFlow::Node getKey();
+
+    /**
+     * Gets the algorithm Node used in the encoding.
+     */
+    abstract DataFlow::Node getAlgorithm();
+
+    /**
+     * Tries to get the algorithm used in the encoding.
+     */
+    abstract string getAlgorithmString();
+
+    /**
+     * Gets the options Node used in the encoding.
+     */
+    abstract DataFlow::Node getOptions();
+
+    /**
+     * Checks if the signature gets verified while decoding.
+     */
+    abstract boolean verifiesSignature();
   }
 }
 
@@ -82,5 +129,33 @@ class JWTDecoding extends DataFlow::Node {
 
   JWTDecoding() { this = range }
 
-  predicate verifiesSignature() { range.verifiesSignature() }
+  /**
+   * Gets the argument containing the payload.
+   */
+  DataFlow::Node getPayload() { result = range.getPayload() }
+
+  /**
+   * Gets the argument containing the encoding key.
+   */
+  DataFlow::Node getKey() { result = range.getKey() }
+
+  /**
+   * Gets the algorithm Node used in the encoding.
+   */
+  DataFlow::Node getAlgorithm() { result = range.getAlgorithm() }
+
+  /**
+   * Tries to get the algorithm used in the encoding.
+   */
+  string getAlgorithmString() { result = range.getAlgorithmString() }
+
+  /**
+   * Gets the options Node used in the encoding.
+   */
+  DataFlow::Node getOptions() { result = range.getOptions() }
+
+  /**
+   * Checks if the signature gets verified while decoding.
+   */
+  boolean verifiesSignature() { result = range.verifiesSignature() }
 }

python/ql/src/experimental/semmle/python/frameworks/JWT.qll

@@ -0,0 +1,94 @@
+private import python
+private import experimental.semmle.python.Concepts
+private import semmle.python.ApiGraphs
+
+predicate isEmptyOrNone(DataFlow::Node arg) { isEmpty(arg) or isNone(arg) }
+
+predicate isEmpty(DataFlow::Node arg) {
+  exists(StrConst emptyString |
+    emptyString.getText() = "" and
+    DataFlow::exprNode(emptyString).(DataFlow::LocalSourceNode).flowsTo(arg)
+  )
+}
+
+predicate isNone(DataFlow::Node arg) {
+  exists( | DataFlow::exprNode(any(None no)).(DataFlow::LocalSourceNode).flowsTo(arg))
+}
+
+predicate isFalse(DataFlow::Node arg) {
+  exists( | DataFlow::exprNode(any(False falseExpr)).(DataFlow::LocalSourceNode).flowsTo(arg))
+}
+
+private module JWT {
+  /** Gets a reference to `jwt` */
+  private API::Node pyjwt() { result = API::moduleImport("jwt") }
+
+  /** Gets a reference to `jwt.encode` */
+  private API::Node pyjwt_encode() { result = pyjwt().getMember("encode") }
+
+  /** Gets a reference to `jwt.decode` */
+  private API::Node pyjwt_decode() { result = pyjwt().getMember("decode") }
+
+  private class PyJWTEncodeCall extends DataFlow::CallCfgNode, JWTEncoding::Range {
+    PyJWTEncodeCall() { this = pyjwt_encode().getACall() }
+
+    override DataFlow::Node getPayload() {
+      result in [this.getArg(0), this.getArgByName("payload")]
+    }
+
+    override DataFlow::Node getKey() { result in [this.getArg(1), this.getArgByName("key")] }
+
+    override DataFlow::Node getAlgorithm() {
+      result in [this.getArg(2), this.getArgByName("algorithm")]
+    }
+
+    override string getAlgorithmString() {
+      exists(StrConst str |
+        DataFlow::exprNode(str).(DataFlow::LocalSourceNode).flowsTo(getAlgorithm()) and
+        result = str.getText()
+      )
+    }
+  }
+
+  private class PyJWTDecodeCall extends DataFlow::CallCfgNode, JWTDecoding::Range {
+    PyJWTDecodeCall() { this = pyjwt_decode().getACall() }
+
+    override DataFlow::Node getPayload() { result in [this.getArg(0), this.getArgByName("jwt")] }
+
+    override DataFlow::Node getKey() { result in [this.getArg(1), this.getArgByName("key")] }
+
+    override DataFlow::Node getAlgorithm() {
+      result in [this.getArg(2), this.getArgByName("algorithms")]
+    }
+
+    override string getAlgorithmString() {
+      exists(StrConst str |
+        DataFlow::exprNode(str).(DataFlow::LocalSourceNode).flowsTo(getAlgorithm()) and
+        result = str.getText()
+      )
+    }
+
+    override DataFlow::Node getOptions() {
+      result in [this.getArg(3), this.getArgByName("options")]
+    }
+
+    override boolean verifiesSignature() {
+      // jwt.decode(token, "key", "HS256")
+      not exists(this.getArgByName("verify")) and not exists(this.getOptions()) and result = true
+      or
+      (
+        // not -> jwt.decode(token, verify=False)
+        isFalse(this.getArgByName("verify"))
+        or
+        // not -> jwt.decode(token, key, options={"verify_signature": False})
+        exists(KeyValuePair optionsDict, NameConstant falseName |
+          falseName.getId() = "False" and
+          optionsDict = this.getArgByName("options").asExpr().(Dict).getItems().getAnItem() and
+          optionsDict.getKey().(Str_).getS().matches("%verify%") and
+          falseName = optionsDict.getValue()
+        )
+      ) and
+      result = false
+    }
+  }
+}

python/ql/src/experimental/semmle/python/frameworks/Stdlib.qll

@@ -9,40 +9,3 @@ private import semmle.python.dataflow.new.TaintTracking
 private import semmle.python.dataflow.new.RemoteFlowSources
 private import experimental.semmle.python.Concepts
 private import semmle.python.ApiGraphs
-private import experimental.semmle.python.security.JWT
-
-private module JWT {
-  private class PyJWTEncodeCall extends DataFlow::CallCfgNode, JWTEncoding::Range {
-    PyJWTEncodeCall() { this = API::moduleImport("jwt").getMember("encode").getACall() }
-
-    override DataFlow::Node getKeyNode() {
-      result = this.getArg(1) or result = this.getArgByName("key")
-    }
-
-    override DataFlow::Node getAlgorithmNode() {
-      result = this.getArg(2) or
-      result = this.getArgByName("algorithm")
-    }
-
-    override string getAlgorithm() {
-      exists(StrConst str |
-        DataFlow::exprNode(str).(DataFlow::LocalSourceNode).flowsTo(getAlgorithmNode()) and
-        result = str.getText()
-      )
-    }
-  }
-
-  private class PyJWTDecodeCall extends DataFlow::CallCfgNode, JWTDecoding::Range {
-    PyJWTDecodeCall() { this = API::moduleImport("jwt").getMember("decode").getACall() }
-
-    override predicate verifiesSignature() {
-      not isFalse(this.getArgByName("verify")) and
-      not exists(KeyValuePair optionsDict, NameConstant falseName |
-        falseName.getId() = "False" and
-        optionsDict = this.getArgByName("options").asExpr().(Dict).getItems().getAnItem() and
-        optionsDict.getKey().(Str_).getS().matches("verify_signature") and
-        falseName = optionsDict.getValue()
-      )
-    }
-  }
-}

python/ql/src/experimental/semmle/python/security/JWT.qll

@@ -1,20 +0,0 @@
-import python
-import experimental.semmle.python.Concepts
-import semmle.python.dataflow.new.DataFlow
-
-predicate isEmptyOrNone(DataFlow::Node arg) { isEmpty(arg) or isNone(arg) }
-
-predicate isEmpty(DataFlow::Node arg) {
-  exists(StrConst emptyString |
-    emptyString.getText() = "" and
-    DataFlow::exprNode(emptyString).(DataFlow::LocalSourceNode).flowsTo(arg)
-  )
-}
-
-predicate isNone(DataFlow::Node arg) {
-  exists( | DataFlow::exprNode(any(None no)).(DataFlow::LocalSourceNode).flowsTo(arg))
-}
-
-predicate isFalse(DataFlow::Node arg) {
-  exists( | DataFlow::exprNode(any(False falseExpr)).(DataFlow::LocalSourceNode).flowsTo(arg))
-}