fix(oidc): validate ID tokens on refresh

CHANGELOG.md

@@ -5,6 +5,13 @@ 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]
+
+### Security
+
+- OpenID Connect backends now validate ID tokens returned during token refresh
+  and reject changes to the authenticated identity.
+
 ## [5.0.2](https://github.com/python-social-auth/social-core/releases/tag/5.0.2) - 2026-06-26
 
 ### Security

social_core/backends/open_id_connect.py

@@ -78,6 +78,7 @@ 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 = ""
@@ -256,17 +257,20 @@ def get_nonce(self, nonce):
     def remove_nonce(self, nonce_id) -> None:
         self.strategy.storage.association.remove([nonce_id])
 
-    def validate_claims(self, id_token) -> None:
+    def validate_temporal_claims(self, id_token) -> None:
         utc_timestamp = timegm(datetime.datetime.now(datetime.timezone.utc).timetuple())
 
         if "nbf" in id_token and utc_timestamp < id_token["nbf"]:
             raise AuthTokenError(self, "Incorrect id_token: nbf")
 
         # Verify the token was issued in the last 10 minutes
         iat_leeway = self.setting("ID_TOKEN_MAX_AGE", self.ID_TOKEN_MAX_AGE)
-        if utc_timestamp > id_token["iat"] + iat_leeway:
+        if "iat" not in id_token or utc_timestamp > id_token["iat"] + iat_leeway:
             raise AuthTokenError(self, "Incorrect id_token: iat")
 
+    def validate_claims(self, id_token) -> None:
+        self.validate_temporal_claims(id_token)
+
         # Validate the nonce to ensure the request was not modified
         nonce = id_token.get("nonce")
         if not nonce:
@@ -278,6 +282,10 @@ 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")
 
@@ -308,11 +316,8 @@ def find_valid_key(self, id_token):
                     return key
         return None
 
-    def validate_and_return_id_token(self, id_token, access_token):
-        """
-        Validates the id_token according to the steps at
-        http://openid.net/specs/openid-connect-core-1_0.html#IDTokenValidation.
-        """
+    def decode_and_validate_id_token(self, id_token, access_token):
+        """Validate an ID token's signature and self-contained claims."""
         client_id, _client_secret = self.get_key_and_secret()
 
         try:
@@ -349,13 +354,29 @@ def validate_and_return_id_token(self, id_token, access_token):
 
         # pyjwt does not validate OIDC claims
         # see https://github.com/jpadilla/pyjwt/pull/296
+        self.validate_authorized_party(claims, client_id)
         if not self.validate_at_hash(claims, access_token, key):
             raise AuthTokenError(self, "Invalid access token")
 
+        return claims
+
+    def validate_and_return_id_token(self, id_token, access_token):
+        """
+        Validates the id_token according to the steps at
+        http://openid.net/specs/openid-connect-core-1_0.html#IDTokenValidation.
+        """
+        claims = self.decode_and_validate_id_token(id_token, access_token)
         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)
+
+        return claims
+
     def request_access_token(
         self,
         url: str,
@@ -390,6 +411,148 @@ def request_access_token(
         )
         return response
 
+    def process_refresh_token_response(self, response, *args, **kwargs) -> dict:
+        data = super().process_refresh_token_response(response, *args, **kwargs)
+        id_token = data.get("id_token")
+        if id_token is None:
+            return data
+
+        access_token = data.get("access_token")
+        if access_token is None:
+            raise AuthTokenError(
+                self,
+                "Missing access_token in OpenID Connect refresh response",
+            )
+
+        self.id_token = self.validate_and_return_refresh_id_token(
+            id_token, access_token
+        )
+        return data
+
+    @staticmethod
+    def id_token_audiences(audience) -> set[str]:
+        if isinstance(audience, str):
+            return {audience}
+        if isinstance(audience, list) and all(
+            isinstance(item, str) for item in audience
+        ):
+            return set(cast("list[str]", audience))
+        raise ValueError
+
+    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):
+            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"):
+            if claim not in claims:
+                raise AuthTokenError(self, f"Incorrect id_token: {claim}")
+            context[claim] = claims[claim]
+
+        for claim in ("auth_time", "nonce", "azp"):
+            if claim in claims:
+                context[claim] = claims[claim]
+        return context
+
+    def validate_id_token_context(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")
+
+        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
+        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")
+
+    def extra_data(
+        self,
+        user,
+        uid: str,
+        response: dict[str, Any],
+        details: dict[str, Any],
+        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
+
+        return data
+
     def user_data(self, access_token: str, *args, **kwargs) -> dict[str, Any] | None:
         return self.validate_userinfo_sub(
             self.get_json(

social_core/tests/backends/open_id_connect.py

@@ -132,6 +132,11 @@ def prepare_access_token_body(  # NOQA: PLR0913, PLR0917
         at_hash=None,
         subject=None,
         access_token: str | None = "foobar",  # noqa: S107
+        refresh_token: str | None = None,
+        include_nonce: bool = True,
+        auth_time: int | None = None,
+        include_azp: bool = True,
+        authorized_party: str | None = None,
     ):
         """
         Prepares a provider access token response. Arguments:
@@ -145,6 +150,8 @@ def prepare_access_token_body(  # NOQA: PLR0913, PLR0917
         body = {"token_type": "bearer"}
         if access_token is not None:
             body["access_token"] = access_token
+        if refresh_token is not None:
+            body["refresh_token"] = refresh_token
         client_key = client_key or self.client_key
         now = datetime.datetime.now(datetime.timezone.utc)
         expiration_datetime = expiration_datetime or (
@@ -161,6 +168,16 @@ def prepare_access_token_body(  # NOQA: PLR0913, PLR0917
             issuer,
             subject,
         )
+        if isinstance(client_key, list):
+            id_token["azp"] = client_key[0]
+        if not include_nonce:
+            id_token.pop("nonce")
+        if not include_azp:
+            id_token.pop("azp")
+        elif authorized_party is not None:
+            id_token["azp"] = authorized_party
+        if auth_time is not None:
+            id_token["auth_time"] = auth_time
         if at_hash is not None:
             id_token["at_hash"] = at_hash
         elif access_token is not None:

social_core/tests/backends/test_cas.py

@@ -2,6 +2,8 @@
 
 import responses
 
+from social_core.exceptions import AuthTokenError
+
 from .oauth import BaseAuthUrlTestMixin
 from .open_id_connect import OpenIdConnectTest
 
@@ -71,3 +73,10 @@ 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},
+            )

social_core/tests/backends/test_google.py

@@ -173,6 +173,13 @@ 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_pkce_can_be_enabled_by_setting(self) -> None:
         self.strategy.set_settings(
             {