Enhance JWT validation in SessionService by adding pre-flight checks for algorithm restrictions. Update tests to validate handling of disallowed algorithms, including "none".

Author: Dopeamin

src/corbado_python_sdk/services/implementation/session_service.py

@@ -1,5 +1,6 @@
 import jwt
 from jwt import (
+    get_unverified_header,          # ← added
     ExpiredSignatureError,
     ImmatureSignatureError,
     InvalidAlgorithmError,
@@ -80,6 +81,23 @@ def validate_token(self, session_token: StrictStr) -> UserEntity:
                 message=ValidationErrorType.CODE_JWT_EMPTY_SESSION_TOKEN.name,
             )
 
+        # ---- pre-flight alg rejection ----
+        try:
+            header = get_unverified_header(session_token)
+        except Exception as err:
+            raise TokenValidationException(
+                error_type=ValidationErrorType.CODE_JWT_GENERAL,
+                message=f"Error parsing JWT header: {session_token}",
+                original_exception=err,
+            )
+        if header.get("alg") not in ALLOWED_ALGS:
+            raise TokenValidationException(
+                error_type=ValidationErrorType.CODE_JWT_INVALID_SIGNATURE,
+                message="Algorithm not allowed",
+                original_exception=InvalidAlgorithmError("Algorithm not allowed"),
+            )
+        # -----------------------------------------
+
         # retrieve signing key
         try:
             signing_key: jwt.PyJWK = self._jwk_client.get_signing_key_from_jwt(token=session_token)

tests/unit/test_session_service.py

@@ -10,7 +10,6 @@
     ImmatureSignatureError,
     InvalidAlgorithmError,
     InvalidSignatureError,
-    PyJWKClientError,
     encode,
 )
 from pydantic import ValidationError
@@ -129,8 +128,8 @@ def _provide_jwts(self):
                 False,
                 """eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6
                 IkpvaG4gRG9lIiwiYWRtaW4iOnRydWV9.dyt0CoTl4WoVjAHI9Q_CwSKhl6d_9rhM3NrXuJttkao""",
-                PyJWKClientError,
-                'Unable to find a signing key that matches: "None"',
+                InvalidAlgorithmError,
+                "Algorithm not allowed",
             ),
             # Not before (nfb) in future
             (
@@ -180,6 +179,13 @@ def _provide_jwts(self):
                 None,
                 None,
             ),
+            # Disallowed algorithm "none"
+            (
+                False,
+                self._generate_jwt(iss="https://auth.acme.com", exp=int(time()) + 100, nbf=int(time()) - 100, algorithm="none"),
+                InvalidAlgorithmError,
+                "Algorithm not allowed",
+            ),
             # Success with old Frontend API URL in config (2)
             (
                 True,
@@ -194,20 +200,17 @@ def _provide_jwts(self):
                 None,
                 None,
             ),
-            # Disallowed algorithm "none"
-            (
-                False,
-                "eyJhbGciOiJub25lIiwidHlwIjoiSldUIn0."
-                "eyJpc3MiOiJodHRwczovL2F1dGguYWNtZS5jb20iLCJzdWIiOiIxMjM0NSIsImlhdCI6"
-                + str(int(time()))
-                + "f.",
-                InvalidAlgorithmError,
-                "Algorithm not allowed",
-            ),
         ]
 
     @classmethod
-    def _generate_jwt(cls, iss: str, exp: int, nbf: int, valid_key: bool = True) -> str:
+    def _generate_jwt(
+        cls,
+        iss: str,
+        exp: int,
+        nbf: int,
+        valid_key: bool = True,
+        algorithm: str = "RS256",
+    ) -> str:
         payload = {
             "iss": iss,
             "iat": int(time()),
@@ -217,9 +220,24 @@ def _generate_jwt(cls, iss: str, exp: int, nbf: int, valid_key: bool = True) ->
             "name": TEST_NAME,
         }
 
-        if valid_key:
-            return encode(payload, key=cls.private_key, algorithm="RS256", headers={"kid": "kid123"})
-        return encode(payload, key=cls.invalid_private_key, algorithm="RS256", headers={"kid": "kid123"})
+        key_to_use = cls.private_key if valid_key else cls.invalid_private_key
+
+        # unsecured JWT (“none”)
+        if algorithm.lower() == "none":
+            # key must be None for alg=none
+            return encode(
+                payload,
+                key=None,
+                headers={"alg": "none", "typ": "JWT"},
+            )
+
+        # signed JWT (RS256 by default)
+        return encode(
+            payload,
+            key=key_to_use,
+            algorithm=algorithm,
+            headers={"kid": "kid123"},
+        )
 
 
 class TestSessionService(TestBase):