fix(oidc): simplify refresh identity validation

CHANGELOG.md

@@ -5,6 +5,15 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/)
 and this project adheres to [Semantic Versioning](http://semver.org/).
 
+## [Unreleased]
+
+### Fixed
+
+- Simplified OpenID Connect ID-token validation to require core claims before
+  login and preserve an immutable identity context across refreshes. This avoids
+  false mismatches for custom user IDs and valid authorized-party claim changes;
+  legacy associations establish the context on their first validated refresh.
+
 ## [5.1.0](https://github.com/python-social-auth/social-core/releases/tag/5.1.0) - 2026-08-06
 
 ### Added

social_core/backends/open_id_connect.py

@@ -21,10 +21,13 @@
 from social_core.exceptions import (
     AuthInvalidParameter,
     AuthMissingParameter,
+    AuthReauthenticationRequired,
     AuthTokenError,
 )
 from social_core.utils import cache
 
+_ID_TOKEN_CONTEXT_KEY = "_oidc_id_token_context"
+
 if TYPE_CHECKING:
     from collections.abc import Mapping
 
@@ -78,7 +81,6 @@ class OpenIdConnectAuth(BaseOAuth2PKCE):
     JWT_LEEWAY: float = 1.0  # seconds
     VALIDATE_AT_HASH: bool = True
     CUSTOM_AT_HASH_ALGO: str | None = None
-    ID_TOKEN_CONTEXT_KEY = "_oidc_id_token_context"
     # When these options are unspecified, server will choose via openid autoconfiguration
     ID_TOKEN_ISSUER = ""
     ACCESS_TOKEN_URL = ""
@@ -282,10 +284,6 @@ def validate_claims(self, id_token) -> None:
         else:
             raise AuthTokenError(self, "Incorrect id_token: nonce")
 
-    def validate_refresh_claims(self, id_token) -> None:
-        """Validate claims that do not depend on the authentication request."""
-        self.validate_temporal_claims(id_token)
-
     def find_valid_key(self, id_token):
         kid = jwt.get_unverified_header(id_token).get("kid")
 
@@ -366,14 +364,16 @@ def validate_and_return_id_token(self, id_token, access_token):
         http://openid.net/specs/openid-connect-core-1_0.html#IDTokenValidation.
         """
         claims = self.decode_and_validate_id_token(id_token, access_token)
+        self.validate_required_id_token_claims(claims)
         self.validate_claims(claims)
 
         return claims
 
     def validate_and_return_refresh_id_token(self, id_token, access_token):
         """Validate an ID token returned by a refresh request."""
         claims = self.decode_and_validate_id_token(id_token, access_token)
-        self.validate_refresh_claims(claims)
+        self.validate_required_id_token_claims(claims)
+        self.validate_temporal_claims(claims)
 
         return claims
 
@@ -441,90 +441,52 @@ def id_token_audiences(audience) -> set[str]:
 
     def validate_authorized_party(self, claims, client_id: str) -> None:
         """Validate the client authorized to use the ID token."""
-        audience = claims.get("aud")
-        try:
-            self.id_token_audiences(audience)
-        except ValueError as error:
-            raise AuthTokenError(self, "Incorrect id_token: aud") from error
-
-        has_authorized_party = "azp" in claims
-        authorized_party = claims.get("azp")
-        if (
-            isinstance(audience, list)
-            and len(audience) > 1
-            and not has_authorized_party
-        ) or (has_authorized_party and authorized_party != client_id):
+        if "azp" in claims and claims["azp"] != client_id:
             raise AuthTokenError(self, "Incorrect id_token: azp")
 
-    def id_token_context(self, claims) -> dict[str, Any]:
-        """Return identity claims that must remain stable across refreshes."""
-        context = {}
-        for claim in ("iss", "sub", "aud"):
+    def validate_required_id_token_claims(self, claims) -> None:
+        """Validate claims required in every ID token."""
+        for claim in ("iss", "sub", "aud", "exp"):
             if claim not in claims:
                 raise AuthTokenError(self, f"Incorrect id_token: {claim}")
-            context[claim] = claims[claim]
 
-        for claim in ("auth_time", "nonce", "azp"):
+    @staticmethod
+    def _id_token_context(claims) -> dict[str, Any]:
+        """Return original claims used to validate refresh continuity."""
+        context = {claim: claims[claim] for claim in ("iss", "sub", "aud")}
+        for claim in ("auth_time", "nonce"):
             if claim in claims:
                 context[claim] = claims[claim]
         return context
 
-    def validate_id_token_context(self, previous, current) -> None:
+    def validate_refresh_id_token_claims(self, previous, current) -> None:
         """Validate identity continuity for an ID token refresh."""
-        if not isinstance(previous, dict):
-            raise AuthTokenError(self, "Invalid stored OpenID Connect context")
-
-        for claim in ("iss", "sub", "aud"):
-            if claim not in previous:
-                raise AuthTokenError(self, "Invalid stored OpenID Connect context")
+        if not isinstance(previous, dict) or any(
+            claim not in previous for claim in ("iss", "sub", "aud")
+        ):
+            raise AuthReauthenticationRequired(self)
 
         for claim in ("iss", "sub"):
             if previous[claim] != current[claim]:
                 raise AuthTokenError(self, f"Incorrect refreshed id_token: {claim}")
 
-        self.validate_id_token_audience_context(previous, current)
-
-        # OIDC Core 1.0 section 12.2 requires exact azp continuity,
-        # including whether the claim is present.
-        if previous.get("azp") != current.get("azp"):
-            raise AuthTokenError(self, "Incorrect refreshed id_token: azp")
-
-        for claim in ("auth_time", "nonce"):
-            if claim in current and previous.get(claim) != current[claim]:
-                raise AuthTokenError(
-                    self,
-                    f"Incorrect refreshed id_token: {claim}",
-                )
-
-    def validate_id_token_audience_context(self, previous, current) -> None:
-        """Validate that refreshed ID token audiences are unchanged."""
         try:
             previous_audiences = self.id_token_audiences(previous["aud"])
         except ValueError as error:
-            raise AuthTokenError(
-                self, "Invalid stored OpenID Connect context"
-            ) from error
+            raise AuthReauthenticationRequired(self) from error
         try:
             current_audiences = self.id_token_audiences(current["aud"])
         except ValueError as error:
             raise AuthTokenError(self, "Incorrect id_token: aud") from error
         if previous_audiences != current_audiences:
             raise AuthTokenError(self, "Incorrect refreshed id_token: aud")
 
-    def validate_legacy_id_token_context(self, uid: str, current) -> None:
-        """Bind a legacy association to a refreshed ID token when possible."""
-        # ID_KEY alone cannot prove how a subclass derived its persisted UID.
-        if (
-            self.id_key() != "sub"
-            or type(self).get_user_id is not OpenIdConnectAuth.get_user_id
-        ):
-            raise AuthTokenError(
-                self,
-                "OpenID Connect identity context is unavailable; "
-                "reauthentication required",
-            )
-        if uid != current["sub"]:
-            raise AuthTokenError(self, "Incorrect refreshed id_token: sub")
+        for claim in ("auth_time", "nonce"):
+            if claim in current and previous.get(claim) != current[claim]:
+                raise AuthTokenError(
+                    self,
+                    f"Incorrect refreshed id_token: {claim}",
+                )
 
     def extra_data(
         self,
@@ -535,21 +497,35 @@ def extra_data(
         pipeline_kwargs: dict[str, Any],
     ) -> dict[str, Any]:
         data = super().extra_data(user, uid, response, details, pipeline_kwargs)
-        previous_context = details.get(self.ID_TOKEN_CONTEXT_KEY)
-
-        if response.get("id_token") is not None:
-            if self.id_token is None:
-                raise AuthTokenError(self, "ID token was not validated")
-            current_context = self.id_token_context(self.id_token)
-            if previous_context is not None:
-                self.validate_id_token_context(previous_context, current_context)
-                data[self.ID_TOKEN_CONTEXT_KEY] = previous_context
-            else:
-                if not pipeline_kwargs:
-                    self.validate_legacy_id_token_context(uid, current_context)
-                data[self.ID_TOKEN_CONTEXT_KEY] = current_context
-        elif previous_context is not None:
-            data[self.ID_TOKEN_CONTEXT_KEY] = previous_context
+        response_id_token = response.get("id_token")
+        if response_id_token is not None:
+            data["id_token"] = response_id_token
+        elif "id_token" in details:
+            data["id_token"] = details["id_token"]
+
+        previous_context = details.get(_ID_TOKEN_CONTEXT_KEY)
+        if pipeline_kwargs:
+            if response_id_token is not None:
+                if self.id_token is None:
+                    raise AuthTokenError(self, "ID token was not validated")
+                data[_ID_TOKEN_CONTEXT_KEY] = self._id_token_context(self.id_token)
+            return data
+
+        if previous_context is not None:
+            data[_ID_TOKEN_CONTEXT_KEY] = previous_context
+        if response_id_token is None:
+            return data
+        if self.id_token is None:
+            raise AuthTokenError(self, "ID token was not validated")
+
+        if previous_context is None:
+            # Legacy associations have no original claim context. Their refreshes
+            # were historically not continuity-checked, so establish the baseline
+            # from this fully validated refresh token and enforce it thereafter.
+            previous_context = self._id_token_context(self.id_token)
+        else:
+            self.validate_refresh_id_token_claims(previous_context, self.id_token)
+        data[_ID_TOKEN_CONTEXT_KEY] = previous_context
 
         return data
 

social_core/exceptions.py

@@ -109,6 +109,13 @@ def __str__(self) -> str:
         return f"Token error: {msg}"
 
 
+class AuthReauthenticationRequired(AuthTokenError):
+    """The stored authentication context cannot establish token continuity."""
+
+    def __init__(self, backend: BaseAuth) -> None:
+        super().__init__(backend, "reauthentication required")
+
+
 class AuthMissingParameter(AuthException):
     """Missing parameter needed to start or complete the process."""
 

social_core/tests/backends/open_id_connect.py

@@ -37,6 +37,7 @@
 }
 
 JWK_PUBLIC_KEY = {key: value for key, value in JWK_KEY.items() if key != "d"}
+STORED_ID_TOKEN_CONTEXT_KEY = "_oidc_id_token_context"
 OpenIdConnectAuthT = TypeVar("OpenIdConnectAuthT", bound=OpenIdConnectAuth)
 
 
@@ -120,7 +121,7 @@ def get_id_token(
             "sub": subject or "1234",
         }
 
-    def prepare_access_token_body(  # NOQA: PLR0913, PLR0917
+    def prepare_access_token_body(  # NOQA: C901, PLR0913, PLR0917
         self,
         client_key=None,
         tamper_message=False,
@@ -137,6 +138,7 @@ def prepare_access_token_body(  # NOQA: PLR0913, PLR0917
         auth_time: int | None = None,
         include_azp: bool = True,
         authorized_party: str | None = None,
+        exclude_claims: tuple[str, ...] = (),
     ):
         """
         Prepares a provider access token response. Arguments:
@@ -182,6 +184,8 @@ def prepare_access_token_body(  # NOQA: PLR0913, PLR0917
             id_token["at_hash"] = at_hash
         elif access_token is not None:
             id_token["at_hash"] = OpenIdConnectAuth.calc_at_hash(access_token, "RS256")
+        for claim in exclude_claims:
+            id_token.pop(claim)
 
         body["id_token"] = jwt.encode(
             id_token,
@@ -200,6 +204,31 @@ def prepare_access_token_body(  # NOQA: PLR0913, PLR0917
 
         return json.dumps(body)
 
+    def login_for_refresh(self, **id_token_kwargs):
+        self.access_token_kwargs = {
+            "refresh_token": "refresh-token",
+            **id_token_kwargs,
+        }
+        user = self.do_login()
+        return user.social[0]
+
+    def refresh_response(self, **id_token_kwargs) -> str:
+        return self.prepare_access_token_body(
+            access_token="refreshed-access-token",  # noqa: S106
+            include_nonce=False,
+            **id_token_kwargs,
+        )
+
+    def refresh_social(self, social, body: str) -> None:
+        responses.add(
+            self._method(self.backend.REFRESH_TOKEN_METHOD),
+            self.backend.refresh_token_url(),
+            status=200,
+            body=body,
+            content_type="application/json",
+        )
+        social.refresh_token(strategy=self.strategy)
+
     def authtoken_raised(self, expected_message, **access_token_kwargs) -> None:
         self.access_token_kwargs = access_token_kwargs
         with self.assertRaisesRegex(AuthTokenError, expected_message):

social_core/tests/backends/test_cas.py

@@ -2,8 +2,6 @@
 
 import responses
 
-from social_core.exceptions import AuthTokenError
-
 from .oauth import BaseAuthUrlTestMixin
 from .open_id_connect import OpenIdConnectTest
 
@@ -74,9 +72,12 @@ def pre_complete_callback(self, start_url) -> None:
     def test_everything_works(self) -> None:
         self.do_login()
 
-    def test_legacy_refresh_requires_reauthentication(self) -> None:
-        with self.assertRaisesRegex(AuthTokenError, "reauthentication required"):
-            self.backend.validate_legacy_id_token_context(
-                "cartman",
-                {"sub": self.user_id},
-            )
+    def test_refresh_with_uid_different_from_id_token_subject(self) -> None:
+        social = self.login_for_refresh()
+
+        self.refresh_social(
+            social,
+            self.refresh_response(subject=self.user_id),
+        )
+
+        self.assertEqual(social.extra_data["access_token"], "refreshed-access-token")

social_core/tests/backends/test_google.py

@@ -16,7 +16,7 @@
 
 from .base import BaseBackendTest
 from .oauth import BaseAuthUrlTestMixin, OAuth1AuthUrlTestMixin, OAuth1Test, OAuth2Test
-from .open_id_connect import OpenIdConnectTest
+from .open_id_connect import STORED_ID_TOKEN_CONTEXT_KEY, OpenIdConnectTest
 
 
 class GoogleOAuth2Test(OAuth2Test, BaseAuthUrlTestMixin):
@@ -173,12 +173,34 @@ class GoogleOpenIdConnectTest(OpenIdConnectTest):
         }
     )
 
-    def test_legacy_refresh_requires_reauthentication(self) -> None:
-        with self.assertRaisesRegex(AuthTokenError, "reauthentication required"):
-            self.backend.validate_legacy_id_token_context(
-                "foo@bar.com",
-                {"sub": "101010101010101010101"},
-            )
+    def test_refresh_preserves_oidc_data_with_google_extra_data(self) -> None:
+        responses.add(
+            responses.GET,
+            url="https://openidconnect.googleapis.com/v1/userinfo",
+            status=200,
+            body=json.dumps(
+                {
+                    "sub": "1234",
+                    "email": "foo@bar.com",
+                    "preferred_username": "foo",
+                }
+            ),
+            content_type="application/json",
+        )
+        self.access_token_kwargs = {"refresh_token": "refresh-token"}
+        self.expected_username = "foo"
+        social = self.do_login().social[0]
+        self.assertIn("id_token", social.extra_data)
+        original_context = social.extra_data[STORED_ID_TOKEN_CONTEXT_KEY]
+        self.assertEqual(original_context["sub"], "1234")
+
+        self.refresh_social(social, self.refresh_response())
+
+        self.assertEqual(social.extra_data["access_token"], "refreshed-access-token")
+        self.assertEqual(
+            social.extra_data[STORED_ID_TOKEN_CONTEXT_KEY],
+            original_context,
+        )
 
     def test_pkce_can_be_enabled_by_setting(self) -> None:
         self.strategy.set_settings(