Add controls for verify_sub option in PyJWT

docs/options.rst

@@ -29,6 +29,7 @@ Overview:
   * `JWT_REFRESH_TOKEN_EXPIRES`_
   * `JWT_SECRET_KEY`_
   * `JWT_TOKEN_LOCATION`_
+  * `JWT_VERIFY_SUB`_
 
 - `Header Options:`_
 
@@ -255,6 +256,18 @@ General Options:
 
     Default: ``"headers"``
 
+.. _JWT_VERIFY_SUB:
+.. py:data:: JWT_VERIFY_SUB
+
+    Control whether the ``sub`` claim is verified during JWT decoding.
+
+    The ``sub`` claim MUST be a ``str`` according the the JWT spec. Setting this option
+    to ``True`` (the default) will enforce this requirement, and consider non-compliant
+    JWTs invalid. Setting the option to ``False`` will skip this validation of the type
+    of the ``sub`` claim, allowing any type for the ``sub`` claim to be considered valid.
+
+    Default: ``True``
+
 
 Header Options:
 ~~~~~~~~~~~~~~~

flask_jwt_extended/config.py

@@ -307,6 +307,10 @@ def decode_issuer(self) -> str:
     def leeway(self) -> int:
         return current_app.config["JWT_DECODE_LEEWAY"]
 
+    @property
+    def verify_sub(self) -> bool:
+        return current_app.config["JWT_VERIFY_SUB"]
+
     @property
     def encode_nbf(self) -> bool:
         return current_app.config["JWT_ENCODE_NBF"]

flask_jwt_extended/jwt_manager.py

@@ -228,6 +228,7 @@ def _set_default_configuration_options(app: Flask) -> None:
         app.config.setdefault("JWT_SECRET_KEY", None)
         app.config.setdefault("JWT_SESSION_COOKIE", True)
         app.config.setdefault("JWT_TOKEN_LOCATION", ("headers",))
+        app.config.setdefault("JWT_VERIFY_SUB", True)
         app.config.setdefault("JWT_ENCODE_NBF", True)
 
     def additional_claims_loader(self, callback: Callable) -> Callable:
@@ -549,6 +550,7 @@ def _decode_jwt_from_config(
             "leeway": config.leeway,
             "secret": secret,
             "verify_aud": config.decode_audience is not None,
+            "verify_sub": config.verify_sub,
         }
 
         try:

flask_jwt_extended/tokens.py

@@ -85,8 +85,9 @@ def _decode_jwt(
     leeway: int,
     secret: str,
     verify_aud: bool,
+    verify_sub: bool,
 ) -> dict:
-    options = {"verify_aud": verify_aud}
+    options = {"verify_aud": verify_aud, "verify_sub": verify_sub}
     if allow_expired:
         options["verify_exp"] = False
 

tests/test_decode_tokens.py

@@ -12,6 +12,7 @@
 from jwt import InvalidIssuerError
 from jwt import InvalidSignatureError
 from jwt import MissingRequiredClaimError
+from jwt.exceptions import InvalidSubjectError
 
 from flask_jwt_extended import create_access_token
 from flask_jwt_extended import create_refresh_token
@@ -279,6 +280,35 @@ def test_verify_no_aud(app, default_access_token, token_aud):
         assert decoded["aud"] == token_aud
 
 
+@pytest.mark.parametrize("token_sub", [123, {}, [], False])
+def test_invalid_sub_values(app, default_access_token, token_sub):
+    """Verifies that invalid values for the sub claim fail decoding, the
+    default behavior of JWT_VERIFY_SUB = True
+    """
+
+    default_access_token["sub"] = token_sub
+    invalid_token = encode_token(app, default_access_token)
+    with pytest.raises(InvalidSubjectError):
+        with app.test_request_context():
+            decode_token(invalid_token)
+
+
+@pytest.mark.parametrize("token_sub", [123, {}, [], False])
+def test_invalid_sub_values_allowed_with_no_verify(
+    app, default_access_token, token_sub
+):
+    """Verifies that invalid values for the sub claim succeed at decoding, if
+    the user configures JWT_VERIFY_SUB = False
+    """
+
+    app.config["JWT_VERIFY_SUB"] = False
+    default_access_token["sub"] = token_sub
+    valid_token = encode_token(app, default_access_token)
+    with app.test_request_context():
+        decoded = decode_token(valid_token)
+        assert decoded["sub"] == token_sub
+
+
 def test_encode_iss(app, default_access_token):
     app.config["JWT_ENCODE_ISSUER"] = "foobar"