chore: use pydantic to parse jwt, instead of json.loads

Author: o-santi

src/auth/Makefile

@@ -1,32 +1,57 @@
-tests: pytest
+help::
+	@echo "Available commands"
+	@echo "  help           -- (default) print this message"
+
+tests: mypy pytest
+help::
+	@echo "  tests          -- run all tests for supabase_auth"
 
 pytest: start-infra
 	uv run --package supabase_auth pytest --cov=./ --cov-report=xml --cov-report=html -vv
 
+mypy:
+	uv run --package supabase_auth mypy src/supabase_auth tests
+help::
+	@echo "  mypy           -- run mypy on supabase_auth"
+
 start-infra:
 	cd infra &&\
 	docker compose down &&\
 	docker compose up -d
 	sleep 2
+help::
+	@echo "  start-infra    -- start containers for tests"
 
 clean-infra:
 	cd infra &&\
 	docker compose down --remove-orphans &&\
 	docker system prune -a --volumes -f
+help::
+	@echo "  clean-infra    -- delete all stored information about the containers"
 
 stop-infra:
 	cd infra &&\
 	docker compose down --remove-orphans
+help::
+	@echo "  stop-infra     -- stop containers for tests"
 
 sync-infra:
 	uv run --package supabase_auth scripts/gh-download.py --repo=supabase/gotrue-js --branch=master --folder=infra
+help::
+	@echo "  sync-infra     -- update locked versions for test containers"
 
 build-sync:
 	uv run --package supabase_auth scripts/run-unasync.py
+help::
+	@echo "  build-sync     -- generate _sync from _async code"
 
 clean:
 	rm -rf htmlcov .pytest_cache .mypy_cache .ruff_cache
 	rm -f .coverage coverage.xml
+help::
+	@echo "  clean          -- clean intermediary files"
 
 build:
 	uv build --package supabase_auth
+help::
+	@echo "  build          -- invoke uv build on supabase_auth package"

src/auth/pyproject.toml

@@ -44,6 +44,9 @@ tests = [
 lints = [
   "ruff >=0.12.1",
   "unasync >= 0.6.0",
+  "python-lsp-server (>=1.12.2,<2.0.0)",
+  "pylsp-mypy (>=0.7.0,<0.8.0)",
+  "python-lsp-ruff (>=2.2.2,<3.0.0)",
 ]
 dev = [{ include-group = "lints" }, {include-group = "tests" }]
 
@@ -76,3 +79,15 @@ asyncio_mode = "auto"
 [build-system]
 requires = ["uv_build>=0.8.3,<0.9.0"]
 build-backend = "uv_build"
+
+[tool.mypy]
+python_version = "3.9"
+check_untyped_defs = true
+allow_redefinition = true
+follow_untyped_imports = true # for deprecation module that does not have stubs
+
+no_warn_no_return = true
+warn_return_any = true
+warn_unused_configs = true
+warn_redundant_casts = true
+warn_unused_ignores = true

src/auth/scripts/run-unasync.py

@@ -2,7 +2,7 @@
 
 import unasync
 
-paths = Path("../src/supabase").glob("**/*.py")
+paths = Path("src/supabase_auth").glob("**/*.py")
 tests = Path("tests").glob("**/*.py")
 
 rules = (unasync._DEFAULT_RULE,)

src/auth/src/supabase_auth/_async/gotrue_base_api.py

@@ -2,11 +2,11 @@
 
 from typing import Any, Callable, Dict, Optional, TypeVar, overload
 
-from httpx import Response
+from httpx import HTTPStatusError, Response
 from pydantic import BaseModel
 from typing_extensions import Literal, Self
 
-from ..constants import API_VERSION_HEADER_NAME, API_VERSIONS
+from ..constants import API_VERSION_HEADER_NAME, API_VERSIONS_2024_01_01_NAME
 from ..helpers import handle_exception, model_dump
 from ..http_clients import AsyncClient
 
@@ -96,12 +96,12 @@ async def _request(
         query: Optional[Dict[str, str]] = None,
         body: Optional[Any] = None,
         no_resolve_json: bool = False,
-        xform: Optional[Callable[[Any], T]] = None,
+        xform: Optional[Callable[[Response], T]] = None,
     ) -> Optional[T]:
         url = f"{self._url}/{path}"
         headers = {**self._headers, **(headers or {})}
         if API_VERSION_HEADER_NAME not in headers:
-            headers[API_VERSION_HEADER_NAME] = API_VERSIONS["2024-01-01"].get("name")
+            headers[API_VERSION_HEADER_NAME] = API_VERSIONS_2024_01_01_NAME
         if "Content-Type" not in headers:
             headers["Content-Type"] = "application/json;charset=UTF-8"
         if jwt:
@@ -121,5 +121,6 @@ async def _request(
             result = response if no_resolve_json else response.json()
             if xform:
                 return xform(result)
-        except Exception as e:
+            return None
+        except (HTTPStatusError, RuntimeError) as e:
             raise handle_exception(e)

src/auth/src/supabase_auth/_sync/gotrue_base_api.py

@@ -2,11 +2,11 @@
 
 from typing import Any, Callable, Dict, Optional, TypeVar, overload
 
-from httpx import Response
+from httpx import HTTPStatusError, Response
 from pydantic import BaseModel
 from typing_extensions import Literal, Self
 
-from ..constants import API_VERSION_HEADER_NAME, API_VERSIONS
+from ..constants import API_VERSION_HEADER_NAME, API_VERSIONS_2024_01_01_NAME
 from ..helpers import handle_exception, model_dump
 from ..http_clients import SyncClient
 
@@ -96,12 +96,12 @@ def _request(
         query: Optional[Dict[str, str]] = None,
         body: Optional[Any] = None,
         no_resolve_json: bool = False,
-        xform: Optional[Callable[[Any], T]] = None,
+        xform: Optional[Callable[[Response], T]] = None,
     ) -> Optional[T]:
         url = f"{self._url}/{path}"
         headers = {**self._headers, **(headers or {})}
         if API_VERSION_HEADER_NAME not in headers:
-            headers[API_VERSION_HEADER_NAME] = API_VERSIONS["2024-01-01"].get("name")
+            headers[API_VERSION_HEADER_NAME] = API_VERSIONS_2024_01_01_NAME
         if "Content-Type" not in headers:
             headers["Content-Type"] = "application/json;charset=UTF-8"
         if jwt:
@@ -121,5 +121,6 @@ def _request(
             result = response if no_resolve_json else response.json()
             if xform:
                 return xform(result)
-        except Exception as e:
+            return None
+        except (HTTPStatusError, RuntimeError) as e:
             raise handle_exception(e)

src/auth/src/supabase_auth/constants.py

@@ -15,10 +15,8 @@
 STORAGE_KEY = "supabase.auth.token"
 
 API_VERSION_HEADER_NAME = "X-Supabase-Api-Version"
-API_VERSIONS = {
-    "2024-01-01": {
-        "timestamp": datetime.timestamp(datetime.strptime("2024-01-01", "%Y-%m-%d")),
-        "name": "2024-01-01",
-    },
-}
+API_VERSIONS_2024_01_01_TIMESTAMP = datetime.timestamp(
+    datetime.strptime("2024-01-01", "%Y-%m-%d")
+)
+API_VERSIONS_2024_01_01_NAME = "2024-01-01"
 BASE64URL_REGEX = r"^([a-z0-9_-]{4})*($|[a-z0-9_-]{3}$|[a-z0-9_-]{2}$)$"

src/auth/src/supabase_auth/errors.py

@@ -86,11 +86,12 @@
     "invalid_credentials",
     "email_address_not_authorized",
     "email_address_invalid",
+    "invalid_jwt",
 ]
 
 
 class AuthError(Exception):
-    def __init__(self, message: str, code: ErrorCode) -> None:
+    def __init__(self, message: str, code: ErrorCode | None) -> None:
         Exception.__init__(self, message)
         self.message = message
         self.name = "AuthError"
@@ -101,11 +102,11 @@ class AuthApiErrorDict(TypedDict):
     name: str
     message: str
     status: int
-    code: ErrorCode
+    code: ErrorCode | None
 
 
 class AuthApiError(AuthError):
-    def __init__(self, message: str, status: int, code: ErrorCode) -> None:
+    def __init__(self, message: str, status: int, code: Optional[ErrorCode]) -> None:
         AuthError.__init__(self, message, code)
         self.name = "AuthApiError"
         self.status = status
@@ -128,7 +129,9 @@ def __init__(self, message: str, original_error: Exception) -> None:
 
 
 class CustomAuthError(AuthError):
-    def __init__(self, message: str, name: str, status: int, code: ErrorCode) -> None:
+    def __init__(
+        self, message: str, name: str, status: int, code: Optional[ErrorCode]
+    ) -> None:
         AuthError.__init__(self, message, code)
         self.name = name
         self.status = status
@@ -138,6 +141,7 @@ def to_dict(self) -> AuthApiErrorDict:
             "name": self.name,
             "message": self.message,
             "status": self.status,
+            "code": self.code,
         }
 
 
@@ -193,6 +197,7 @@ def to_dict(self) -> AuthImplicitGrantRedirectErrorDict:
             "message": self.message,
             "status": self.status,
             "details": self.details,
+            "code": self.code,
         }
 
 
@@ -207,6 +212,10 @@ def __init__(self, message: str, status: int) -> None:
         )
 
 
+class AuthApiErrorWithReasonsDict(AuthApiErrorDict):
+    reasons: List[str]
+
+
 class AuthWeakPasswordError(CustomAuthError):
     def __init__(self, message: str, status: int, reasons: List[str]) -> None:
         CustomAuthError.__init__(
@@ -218,12 +227,13 @@ def __init__(self, message: str, status: int, reasons: List[str]) -> None:
         )
         self.reasons = reasons
 
-    def to_dict(self) -> AuthApiErrorDict:
+    def to_dict(self) -> AuthApiErrorWithReasonsDict:
         return {
             "name": self.name,
             "message": self.message,
             "status": self.status,
             "reasons": self.reasons,
+            "code": self.code,
         }
 
 

src/auth/src/supabase_auth/helpers.py

@@ -1,21 +1,25 @@
 from __future__ import annotations
 
 import base64
+import binascii
 import hashlib
 import re
 import secrets
 import string
 import uuid
 from base64 import urlsafe_b64decode
 from datetime import datetime
-from json import loads
 from typing import Any, Dict, Optional, Type, TypedDict, TypeVar, cast
 from urllib.parse import urlparse
 
 from httpx import HTTPStatusError, Response
-from pydantic import BaseModel
+from pydantic import BaseModel, TypeAdapter
 
-from .constants import API_VERSION_HEADER_NAME, API_VERSIONS, BASE64URL_REGEX
+from .constants import (
+    API_VERSION_HEADER_NAME,
+    API_VERSIONS_2024_01_01_TIMESTAMP,
+    BASE64URL_REGEX,
+)
 from .errors import (
     AuthApiError,
     AuthError,
@@ -136,18 +140,9 @@ def get_error_message(error: Any) -> str:
     return next((error[prop] for prop in props if filter(prop)), str(error))
 
 
-def get_error_code(error: Any) -> str:
-    return error.get("error_code", None) if isinstance(error, dict) else None
-
-
-def looks_like_http_status_error(exception: Exception) -> bool:
-    return isinstance(exception, HTTPStatusError)
-
-
-def handle_exception(exception: Exception) -> AuthError:
-    if not looks_like_http_status_error(exception):
-        return AuthRetryableError(get_error_message(exception), 0)
-    error = cast(HTTPStatusError, exception)
+def handle_exception(error: HTTPStatusError | RuntimeError) -> AuthError:
+    if not isinstance(error, HTTPStatusError):
+        return AuthRetryableError(get_error_message(error), 0)
     try:
         network_error_codes = [502, 503, 504]
         if error.response.status_code in network_error_codes:
@@ -161,8 +156,10 @@ def handle_exception(exception: Exception) -> AuthError:
 
         if (
             response_api_version
-            and datetime.timestamp(response_api_version)
-            >= API_VERSIONS.get("2024-01-01").get("timestamp")
+            and (
+                datetime.timestamp(response_api_version)
+                >= API_VERSIONS_2024_01_01_TIMESTAMP
+            )
             and isinstance(data, dict)
             and data
             and isinstance(data.get("code"), str)
@@ -180,18 +177,18 @@ def handle_exception(exception: Exception) -> AuthError:
                 and isinstance(data.get("weak_password"), dict)
                 and data.get("weak_password")
                 and isinstance(data.get("weak_password"), list)
-                and len(data.get("weak_password"))
+                and len(data["weak_password"])
             ):
                 return AuthWeakPasswordError(
                     get_error_message(data),
                     error.response.status_code,
-                    data.get("weak_password").get("reasons"),
+                    data["weak_password"].get("reasons"),
                 )
         elif error_code == "weak_password":
             return AuthWeakPasswordError(
                 get_error_message(data),
                 error.response.status_code,
-                data.get("weak_password", {}).get("reasons", {}),
+                data["weak_password"].get("reasons", {}),
             )
 
         return AuthApiError(
@@ -224,20 +221,26 @@ class DecodedJWT(TypedDict):
     raw: Dict[str, str]
 
 
+JWTHeaderParser = TypeAdapter(JWTHeader)
+JWTPayloadParser = TypeAdapter(JWTPayload)
+
+
 def decode_jwt(token: str) -> DecodedJWT:
     parts = token.split(".")
     if len(parts) != 3:
         raise AuthInvalidJwtError("Invalid JWT structure")
 
-    # regex check for base64url
-    for part in parts:
-        if not re.match(BASE64URL_REGEX, part, re.IGNORECASE):
-            raise AuthInvalidJwtError("JWT not in base64url format")
+    try:
+        header = base64url_to_bytes(parts[0])
+        payload = base64url_to_bytes(parts[1])
+        signature = base64url_to_bytes(parts[2])
+    except binascii.Error:
+        raise AuthInvalidJwtError("Invalid JWT structure")
 
     return DecodedJWT(
-        header=JWTHeader(**loads(str_from_base64url(parts[0]))),
-        payload=JWTPayload(**loads(str_from_base64url(parts[1]))),
-        signature=base64url_to_bytes(parts[2]),
+        header=JWTHeaderParser.validate_json(header),
+        payload=JWTPayloadParser.validate_json(payload),
+        signature=signature,
         raw={
             "header": parts[0],
             "payload": parts[1],