fix(client): add claims_cls parameter for parse_id_token, #725

Author: lepture

.gitignore

@@ -19,3 +19,4 @@ venv/
 .pytest_cache/
 *.egg
 .idea/
+uv.lock

authlib/integrations/base_client/async_openid.py

@@ -34,17 +34,18 @@ async def userinfo(self, **kwargs):
         data = resp.json()
         return UserInfo(data)
 
-    async def parse_id_token(self, token, nonce, claims_options=None):
+    async def parse_id_token(self, token, nonce, claims_options=None, claims_cls=None, leeway=120):
         """Return an instance of UserInfo from token's ``id_token``."""
         claims_params = dict(
             nonce=nonce,
             client_id=self.client_id,
         )
-        if "access_token" in token:
-            claims_params["access_token"] = token["access_token"]
-            claims_cls = CodeIDToken
-        else:
-            claims_cls = ImplicitIDToken
+        if claims_cls is None:
+            if "access_token" in token:
+                claims_params["access_token"] = token["access_token"]
+                claims_cls = CodeIDToken
+            else:
+                claims_cls = ImplicitIDToken
 
         metadata = await self.load_server_metadata()
         if claims_options is None and "issuer" in metadata:
@@ -78,5 +79,5 @@ async def parse_id_token(self, token, nonce, claims_options=None):
         # https://github.com/lepture/authlib/issues/259
         if claims.get("nonce_supported") is False:
             claims.params["nonce"] = None
-        claims.validate(leeway=120)
+        claims.validate(leeway=leeway)
         return UserInfo(claims)

authlib/integrations/base_client/sync_openid.py

@@ -33,7 +33,7 @@ def userinfo(self, **kwargs):
         data = resp.json()
         return UserInfo(data)
 
-    def parse_id_token(self, token, nonce, claims_options=None, leeway=120):
+    def parse_id_token(self, token, nonce, claims_options=None, claims_cls=None, leeway=120):
         """Return an instance of UserInfo from token's ``id_token``."""
         if "id_token" not in token:
             return None
@@ -44,11 +44,13 @@ def parse_id_token(self, token, nonce, claims_options=None, leeway=120):
             nonce=nonce,
             client_id=self.client_id,
         )
-        if "access_token" in token:
-            claims_params["access_token"] = token["access_token"]
-            claims_cls = CodeIDToken
-        else:
-            claims_cls = ImplicitIDToken
+
+        if claims_cls is None:
+            if "access_token" in token:
+                claims_params["access_token"] = token["access_token"]
+                claims_cls = CodeIDToken
+            else:
+                claims_cls = ImplicitIDToken
 
         metadata = self.load_server_metadata()
         if claims_options is None and "issuer" in metadata:

authlib/integrations/django_client/apps.py

@@ -78,15 +78,22 @@ def authorize_access_token(self, request, **kwargs):
                 "state": request.POST.get("state"),
             }
 
-        claims_options = kwargs.pop("claims_options", None)
         state_data = self.framework.get_state_data(request.session, params.get("state"))
         self.framework.clear_state_data(request.session, params.get("state"))
         params = self._format_state_params(state_data, params)
+
+        claims_options = kwargs.pop("claims_options", None)
+        claims_cls = kwargs.pop("claims_cls", None)
+        leeway = kwargs.pop("leeway", 120)
         token = self.fetch_access_token(**params, **kwargs)
 
         if "id_token" in token and "nonce" in state_data:
             userinfo = self.parse_id_token(
-                token, nonce=state_data["nonce"], claims_options=claims_options
+                token,
+                nonce=state_data["nonce"],
+                claims_options=claims_options,
+                claims_cls=claims_cls,
+                leeway=leeway,
             )
             token["userinfo"] = userinfo
         return token

authlib/integrations/flask_client/apps.py

@@ -100,16 +100,23 @@ def authorize_access_token(self, **kwargs):
                 "state": request.form.get("state"),
             }
 
-        claims_options = kwargs.pop("claims_options", None)
         state_data = self.framework.get_state_data(session, params.get("state"))
         self.framework.clear_state_data(session, params.get("state"))
         params = self._format_state_params(state_data, params)
+
+        claims_options = kwargs.pop("claims_options", None)
+        claims_cls = kwargs.pop("claims_cls", None)
+        leeway = kwargs.pop("leeway", 120)
         token = self.fetch_access_token(**params, **kwargs)
         self.token = token
 
         if "id_token" in token and "nonce" in state_data:
             userinfo = self.parse_id_token(
-                token, nonce=state_data["nonce"], claims_options=claims_options
+                token,
+                nonce=state_data["nonce"],
+                claims_options=claims_options,
+                claims_cls=claims_cls,
+                leeway=leeway,
             )
             token["userinfo"] = userinfo
         return token

authlib/integrations/starlette_client/apps.py

@@ -78,15 +78,22 @@ async def authorize_access_token(self, request, **kwargs):
         else:
             session = request.session
 
-        claims_options = kwargs.pop("claims_options", None)
         state_data = await self.framework.get_state_data(session, params.get("state"))
         await self.framework.clear_state_data(session, params.get("state"))
         params = self._format_state_params(state_data, params)
+
+        claims_options = kwargs.pop("claims_options", None)
+        claims_cls = kwargs.pop("claims_cls", None)
+        leeway = kwargs.pop("leeway", 120)
         token = await self.fetch_access_token(**params, **kwargs)
 
         if "id_token" in token and "nonce" in state_data:
             userinfo = await self.parse_id_token(
-                token, nonce=state_data["nonce"], claims_options=claims_options
+                token,
+                nonce=state_data["nonce"],
+                claims_options=claims_options,
+                claims_cls=claims_cls,
+                leeway=leeway,
             )
             token["userinfo"] = userinfo
         return token