feat: do not send non-JWTs in Authorization header

supabase/_async/client.py

@@ -16,6 +16,8 @@
 from storage3.constants import DEFAULT_TIMEOUT as DEFAULT_STORAGE_CLIENT_TIMEOUT
 from supafunc import AsyncFunctionsClient
 
+from supabase.lib.helpers import is_valid_jwt
+
 from ..lib.client_options import AsyncClientOptions as ClientOptions
 from .auth_client import AsyncSupabaseAuthClient
 
@@ -278,9 +280,10 @@ def _create_auth_header(self, token: str):
 
     def _get_auth_headers(self, authorization: Optional[str] = None) -> Dict[str, str]:
         if authorization is None:
-            authorization = self.options.headers.get(
-                "Authorization", self._create_auth_header(self.supabase_key)
-            )
+            if is_valid_jwt(self.supabase_key):
+                authorization = self.options.headers.get(
+                    "Authorization", self._create_auth_header(self.supabase_key)
+                )
 
         """Helper method to get auth headers."""
         return {
@@ -291,13 +294,16 @@ def _get_auth_headers(self, authorization: Optional[str] = None) -> Dict[str, st
     def _listen_to_auth_events(
         self, event: AuthChangeEvent, session: Optional[Session]
     ):
-        access_token = self.supabase_key
+        default_access_token = (
+            self.supabase_key if is_valid_jwt(self.supabase_key) else None
+        )
+        access_token = default_access_token
         if event in ["SIGNED_IN", "TOKEN_REFRESHED", "SIGNED_OUT"]:
             # reset postgrest and storage instance on event change
             self._postgrest = None
             self._storage = None
             self._functions = None
-            access_token = session.access_token if session else self.supabase_key
+            access_token = session.access_token if session else default_access_token
 
         self.options.headers["Authorization"] = self._create_auth_header(access_token)
         asyncio.create_task(self.realtime.set_auth(access_token))

supabase/_sync/client.py

@@ -15,6 +15,8 @@
 from storage3.constants import DEFAULT_TIMEOUT as DEFAULT_STORAGE_CLIENT_TIMEOUT
 from supafunc import SyncFunctionsClient
 
+from supabase.lib.helpers import is_valid_jwt
+
 from ..lib.client_options import SyncClientOptions as ClientOptions
 from .auth_client import SyncSupabaseAuthClient
 
@@ -277,9 +279,10 @@ def _create_auth_header(self, token: str):
 
     def _get_auth_headers(self, authorization: Optional[str] = None) -> Dict[str, str]:
         if authorization is None:
-            authorization = self.options.headers.get(
-                "Authorization", self._create_auth_header(self.supabase_key)
-            )
+            if is_valid_jwt(self.supabase_key):
+                authorization = self.options.headers.get(
+                    "Authorization", self._create_auth_header(self.supabase_key)
+                )
 
         """Helper method to get auth headers."""
         return {
@@ -290,13 +293,16 @@ def _get_auth_headers(self, authorization: Optional[str] = None) -> Dict[str, st
     def _listen_to_auth_events(
         self, event: AuthChangeEvent, session: Optional[Session]
     ):
-        access_token = self.supabase_key
+        default_access_token = (
+            self.supabase_key if is_valid_jwt(self.supabase_key) else None
+        )
+        access_token = default_access_token
         if event in ["SIGNED_IN", "TOKEN_REFRESHED", "SIGNED_OUT"]:
             # reset postgrest and storage instance on event change
             self._postgrest = None
             self._storage = None
             self._functions = None
-            access_token = session.access_token if session else self.supabase_key
+            access_token = session.access_token if session else default_access_token
 
         self.options.headers["Authorization"] = self._create_auth_header(access_token)
 

supabase/lib/helpers.py

@@ -0,0 +1,41 @@
+import re
+from typing import Dict
+
+BASE64URL_REGEX = r"^([a-z0-9_-]{4})*($|[a-z0-9_-]{3}$|[a-z0-9_-]{2}$)$"
+
+
+def is_valid_jwt(value: str) -> bool:
+    """Checks if value looks like a JWT, does not do any extra parsing."""
+    if not isinstance(value, str):
+        return False
+
+    # Remove trailing whitespaces if any.
+    value = value.strip()
+
+    # Remove "Bearer " prefix if any.
+    if value.startswith("Bearer "):
+        value = value[7:]
+
+    # Valid JWT must have 2 dots (Header.Paylod.Signature)
+    if value.count(".") != 2:
+        return False
+
+    for part in value.split("."):
+        if not re.search(BASE64URL_REGEX, part, re.IGNORECASE):
+            return False
+
+    return True
+
+
+def check_authorization_header(headers: Dict[str, str]):
+    authorization = headers.get("Authorization")
+    if not authorization:
+        return
+
+    if authorization.startswith("Bearer "):
+        if not is_valid_jwt(authorization):
+            raise ValueError(
+                "create_client called with global Authorization header that does not contain a JWT"
+            )
+
+    return True