diff --git a/backend/open_webui/routers/auths.py b/backend/open_webui/routers/auths.py
index 75a3e0961d..16ae0422e8 100644
--- a/backend/open_webui/routers/auths.py
+++ b/backend/open_webui/routers/auths.py
@@ -1752,6 +1752,7 @@ async def token_exchange(
         user=user,
         user_data=user_data,
         provider=provider,
+        access_token=form_data.token,
         db=db,
     )
     if await Config.get('oauth.enable_group_mapping'):
diff --git a/backend/open_webui/utils/oauth.py b/backend/open_webui/utils/oauth.py
index 3bcda023e3..10dedb1e6c 100644
--- a/backend/open_webui/utils/oauth.py
+++ b/backend/open_webui/utils/oauth.py
@@ -188,6 +188,16 @@ def _default_value(value):
     return getattr(value, 'value', value)
 
 
+def _get_roles_claim(claims: dict, claim: str) -> list | str | int | None:
+    """Read nested or flat claims, preserving explicit empty values and zero."""
+    value = claims
+    for key in claim.split('.'):
+        value = value.get(key) if isinstance(value, dict) else None
+    if not isinstance(value, (list, str, int)):
+        value = claims.get(claim)
+    return value if isinstance(value, (list, str, int)) else None
+
+
 async def get_oauth_runtime_config() -> SimpleNamespace:
     keys = [key for key, _default in OAUTH_RUNTIME_CONFIG.values()]
     stored = await Config.get_many(*keys)
@@ -1505,7 +1515,7 @@ class OAuthManager:
             log.error(f'Exception during token refresh for provider {provider}: {e}')
             return None
 
-    async def get_user_role(self, user, user_data):
+    async def get_user_role(self, user, user_data, *, access_token: str | None = None):
         auth_config = await get_oauth_runtime_config()
         user_count = await Users.get_num_users()
         if user and user_count == 1:
@@ -1529,18 +1539,15 @@ class OAuthManager:
             # Keep existing users at their current role unless the provider sent roles.
             role = user.role if user else auth_config.DEFAULT_USER_ROLE
 
-            # Next block extracts the roles from the user data, accepting nested claims of any depth
-            if oauth_claim and oauth_allowed_roles and oauth_admin_roles:
-                claim_data = user_data
-                nested_claims = oauth_claim.split('.')
-                for nested_claim in nested_claims:
-                    claim_data = claim_data.get(nested_claim, {})
-
-                # Try flat claim structure as alternative
-                if not claim_data:
-                    claim_data = user_data.get(oauth_claim, {})
-
-                oauth_roles = []
+            if oauth_claim:
+                claim_data = _get_roles_claim(user_data, oauth_claim)
+                if claim_data is None and access_token is not None:
+                    # The exchange endpoint has already validated this token with the provider's userinfo endpoint.
+                    try:
+                        token_claims = jwt.decode(access_token, options={'verify_signature': False})
+                        claim_data = _get_roles_claim(token_claims, oauth_claim)
+                    except jwt.PyJWTError as e:
+                        log.debug('Token exchange: cannot decode token claims: %s', e)
 
                 if isinstance(claim_data, list):
                     oauth_roles = claim_data
@@ -1553,6 +1560,10 @@ class OAuthManager:
                 elif isinstance(claim_data, int):
                     oauth_roles = [str(claim_data)]
 
+            if access_token is not None and not oauth_roles and oauth_allowed_roles and '*' not in oauth_allowed_roles:
+                log.warning('Token exchange denied: no readable roles claim in userinfo or the token')
+                raise HTTPException(status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.ACCESS_PROHIBITED)
+
             log.debug('Oauth Roles claim: %s', oauth_claim)
             log.debug('User roles from oauth: %s', oauth_roles)
             log.debug('Accepted user roles: %s', oauth_allowed_roles)
@@ -1599,9 +1610,10 @@ class OAuthManager:
         user_data,
         provider,
         *,
+        access_token: str | None = None,
         db=None,
     ):
-        determined_role = await self.get_user_role(user, user_data)
+        determined_role = await self.get_user_role(user, user_data, access_token=access_token)
         if user.role == determined_role:
             return user