fix base64 token validation and add tests for decode_jwt

Author: grdsdev

supabase_auth/_async/gotrue_client.py

@@ -58,7 +58,6 @@
     AuthResponse,
     ClaimsResponse,
     CodeExchangeParams,
-    DecodedJWTDict,
     IdentitiesResponse,
     JWKSet,
     MFAChallengeAndVerifyParams,
@@ -687,7 +686,7 @@ async def set_session(self, access_token: str, refresh_token: str) -> AuthRespon
         has_expired = True
         session: Optional[Session] = None
         if access_token and access_token.split(".")[1]:
-            payload = self._decode_jwt(access_token)
+            payload = decode_jwt(access_token)["payload"]
             exp = payload.get("exp")
             if exp:
                 expires_at = int(exp)
@@ -899,7 +898,7 @@ async def _get_authenticator_assurance_level(
                 next_level=None,
                 current_authentication_methods=[],
             )
-        payload = self._decode_jwt(session.access_token)
+        payload = decode_jwt(session.access_token)["payload"]
         current_level: Optional[AuthenticatorAssuranceLevels] = None
         if payload.get("aal"):
             current_level = payload.get("aal")
@@ -1137,13 +1136,6 @@ async def _get_url_for_provider(
         query = urlencode(params)
         return f"{url}?{query}", params
 
-    def _decode_jwt(self, jwt: str) -> DecodedJWTDict:
-        """
-        Decodes a JWT (without performing any validation).
-        """
-        decoded = decode_jwt(jwt)
-        return decoded["payload"]
-
     async def exchange_code_for_session(self, params: CodeExchangeParams):
         code_verifier = params.get("code_verifier") or await self._storage.get_item(
             f"{self._storage_key}-code-verifier"

supabase_auth/_sync/gotrue_client.py

@@ -58,7 +58,6 @@
     AuthResponse,
     ClaimsResponse,
     CodeExchangeParams,
-    DecodedJWTDict,
     IdentitiesResponse,
     JWKSet,
     MFAChallengeAndVerifyParams,
@@ -685,7 +684,7 @@ def set_session(self, access_token: str, refresh_token: str) -> AuthResponse:
         has_expired = True
         session: Optional[Session] = None
         if access_token and access_token.split(".")[1]:
-            payload = self._decode_jwt(access_token)
+            payload = decode_jwt(access_token)["payload"]
             exp = payload.get("exp")
             if exp:
                 expires_at = int(exp)
@@ -895,7 +894,7 @@ def _get_authenticator_assurance_level(
                 next_level=None,
                 current_authentication_methods=[],
             )
-        payload = self._decode_jwt(session.access_token)
+        payload = decode_jwt(session.access_token)["payload"]
         current_level: Optional[AuthenticatorAssuranceLevels] = None
         if payload.get("aal"):
             current_level = payload.get("aal")
@@ -1131,13 +1130,6 @@ def _get_url_for_provider(
         query = urlencode(params)
         return f"{url}?{query}", params
 
-    def _decode_jwt(self, jwt: str) -> DecodedJWTDict:
-        """
-        Decodes a JWT (without performing any validation).
-        """
-        decoded = decode_jwt(jwt)
-        return decoded["payload"]
-
     def exchange_code_for_session(self, params: CodeExchangeParams):
         code_verifier = params.get("code_verifier") or self._storage.get_item(
             f"{self._storage_key}-code-verifier"

supabase_auth/helpers.py

@@ -229,9 +229,9 @@ def decode_jwt(token: str) -> DecodedJWT:
         raise AuthInvalidJwtError("Invalid JWT structure")
 
     # regex check for base64url
-    # for part in parts:
-    #     if not re.match(BASE64URL_REGEX, part):
-    #         raise AuthInvalidJwtError("JWT not in base64url format")
+    for part in parts:
+        if not re.match(BASE64URL_REGEX, part, re.IGNORECASE):
+            raise AuthInvalidJwtError("JWT not in base64url format")
 
     return DecodedJWT(
         header=JWTHeader(**loads(str_from_base64url(parts[0]))),

tests/_async/test_gotrue.py

@@ -1,7 +1,13 @@
 import time
 import unittest
 
-from .clients import auth_client, auth_client_with_asymmetric_session
+import pytest
+from jwt import encode
+
+from supabase_auth.errors import AuthInvalidJwtError, AuthSessionMissingError
+from supabase_auth.helpers import decode_jwt
+
+from .clients import GOTRUE_JWT_SECRET, auth_client, auth_client_with_asymmetric_session
 from .utils import mock_user_credentials
 
 
@@ -71,3 +77,115 @@ async def test_jwks_ttl_cache_behavior(mocker):
     finally:
         # Restore original time function
         mocker.patch("time.time", original_time)
+
+
+async def test_set_session_with_valid_tokens():
+    client = auth_client()
+    credentials = mock_user_credentials()
+
+    # First sign up to get valid tokens
+    signup_response = await client.sign_up(
+        {
+            "email": credentials.get("email"),
+            "password": credentials.get("password"),
+        }
+    )
+    assert signup_response.session is not None
+
+    # Get the tokens from the signup response
+    access_token = signup_response.session.access_token
+    refresh_token = signup_response.session.refresh_token
+
+    # Clear the session
+    await client._remove_session()
+
+    # Set the session with the tokens
+    response = await client.set_session(access_token, refresh_token)
+
+    # Verify the response
+    assert response.session is not None
+    assert response.session.access_token == access_token
+    assert response.session.refresh_token == refresh_token
+    assert response.user is not None
+    assert response.user.email == credentials.get("email")
+
+
+async def test_set_session_with_expired_token():
+    client = auth_client()
+    credentials = mock_user_credentials()
+
+    # First sign up to get valid tokens
+    signup_response = await client.sign_up(
+        {
+            "email": credentials.get("email"),
+            "password": credentials.get("password"),
+        }
+    )
+    assert signup_response.session is not None
+
+    # Get the tokens from the signup response
+    access_token = signup_response.session.access_token
+    refresh_token = signup_response.session.refresh_token
+
+    # Clear the session
+    await client._remove_session()
+
+    # Create an expired token by modifying the JWT
+    expired_token = access_token.split(".")
+    payload = decode_jwt(access_token)["payload"]
+    payload["exp"] = int(time.time()) - 3600  # Set expiry to 1 hour ago
+    expired_token[1] = encode(payload, GOTRUE_JWT_SECRET, algorithm="HS256").split(".")[
+        1
+    ]
+    expired_access_token = ".".join(expired_token)
+
+    # Set the session with the expired token
+    response = await client.set_session(expired_access_token, refresh_token)
+
+    # Verify the response has a new access token (refreshed)
+    assert response.session is not None
+    assert response.session.access_token != expired_access_token
+    assert response.session.refresh_token != refresh_token
+    assert response.user is not None
+    assert response.user.email == credentials.get("email")
+
+
+async def test_set_session_without_refresh_token():
+    client = auth_client()
+    credentials = mock_user_credentials()
+
+    # First sign up to get valid tokens
+    signup_response = await client.sign_up(
+        {
+            "email": credentials.get("email"),
+            "password": credentials.get("password"),
+        }
+    )
+    assert signup_response.session is not None
+
+    # Get the access token from the signup response
+    access_token = signup_response.session.access_token
+
+    # Clear the session
+    await client._remove_session()
+
+    # Create an expired token
+    expired_token = access_token.split(".")
+    payload = decode_jwt(access_token)["payload"]
+    payload["exp"] = int(time.time()) - 3600  # Set expiry to 1 hour ago
+    expired_token[1] = encode(payload, GOTRUE_JWT_SECRET, algorithm="HS256").split(".")[
+        1
+    ]
+    expired_access_token = ".".join(expired_token)
+
+    # Try to set the session with an expired token but no refresh token
+    with pytest.raises(AuthSessionMissingError):
+        await client.set_session(expired_access_token, "")
+
+
+async def test_set_session_with_invalid_token():
+    client = auth_client()
+
+    # Try to set the session with invalid tokens
+    with pytest.raises(AuthInvalidJwtError):
+        await client.set_session("invalid.token.here", "invalid_refresh_token")

tests/_sync/test_gotrue.py

@@ -1,7 +1,13 @@
 import time
 import unittest
 
-from .clients import auth_client, auth_client_with_asymmetric_session
+import pytest
+from jwt import encode
+
+from supabase_auth.errors import AuthInvalidJwtError, AuthSessionMissingError
+from supabase_auth.helpers import decode_jwt
+
+from .clients import GOTRUE_JWT_SECRET, auth_client, auth_client_with_asymmetric_session
 from .utils import mock_user_credentials
 
 
@@ -71,3 +77,115 @@ def test_jwks_ttl_cache_behavior(mocker):
     finally:
         # Restore original time function
         mocker.patch("time.time", original_time)
+
+
+def test_set_session_with_valid_tokens():
+    client = auth_client()
+    credentials = mock_user_credentials()
+
+    # First sign up to get valid tokens
+    signup_response = client.sign_up(
+        {
+            "email": credentials.get("email"),
+            "password": credentials.get("password"),
+        }
+    )
+    assert signup_response.session is not None
+
+    # Get the tokens from the signup response
+    access_token = signup_response.session.access_token
+    refresh_token = signup_response.session.refresh_token
+
+    # Clear the session
+    client._remove_session()
+
+    # Set the session with the tokens
+    response = client.set_session(access_token, refresh_token)
+
+    # Verify the response
+    assert response.session is not None
+    assert response.session.access_token == access_token
+    assert response.session.refresh_token == refresh_token
+    assert response.user is not None
+    assert response.user.email == credentials.get("email")
+
+
+def test_set_session_with_expired_token():
+    client = auth_client()
+    credentials = mock_user_credentials()
+
+    # First sign up to get valid tokens
+    signup_response = client.sign_up(
+        {
+            "email": credentials.get("email"),
+            "password": credentials.get("password"),
+        }
+    )
+    assert signup_response.session is not None
+
+    # Get the tokens from the signup response
+    access_token = signup_response.session.access_token
+    refresh_token = signup_response.session.refresh_token
+
+    # Clear the session
+    client._remove_session()
+
+    # Create an expired token by modifying the JWT
+    expired_token = access_token.split(".")
+    payload = decode_jwt(access_token)["payload"]
+    payload["exp"] = int(time.time()) - 3600  # Set expiry to 1 hour ago
+    expired_token[1] = encode(payload, GOTRUE_JWT_SECRET, algorithm="HS256").split(".")[
+        1
+    ]
+    expired_access_token = ".".join(expired_token)
+
+    # Set the session with the expired token
+    response = client.set_session(expired_access_token, refresh_token)
+
+    # Verify the response has a new access token (refreshed)
+    assert response.session is not None
+    assert response.session.access_token != expired_access_token
+    assert response.session.refresh_token != refresh_token
+    assert response.user is not None
+    assert response.user.email == credentials.get("email")
+
+
+def test_set_session_without_refresh_token():
+    client = auth_client()
+    credentials = mock_user_credentials()
+
+    # First sign up to get valid tokens
+    signup_response = client.sign_up(
+        {
+            "email": credentials.get("email"),
+            "password": credentials.get("password"),
+        }
+    )
+    assert signup_response.session is not None
+
+    # Get the access token from the signup response
+    access_token = signup_response.session.access_token
+
+    # Clear the session
+    client._remove_session()
+
+    # Create an expired token
+    expired_token = access_token.split(".")
+    payload = decode_jwt(access_token)["payload"]
+    payload["exp"] = int(time.time()) - 3600  # Set expiry to 1 hour ago
+    expired_token[1] = encode(payload, GOTRUE_JWT_SECRET, algorithm="HS256").split(".")[
+        1
+    ]
+    expired_access_token = ".".join(expired_token)
+
+    # Try to set the session with an expired token but no refresh token
+    with pytest.raises(AuthSessionMissingError):
+        client.set_session(expired_access_token, "")
+
+
+def test_set_session_with_invalid_token():
+    client = auth_client()
+
+    # Try to set the session with invalid tokens
+    with pytest.raises(AuthInvalidJwtError):
+        client.set_session("invalid.token.here", "invalid_refresh_token")