From 4a4de817073712d5bafa30747c55ddb1052c7d4e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Michal=20=C4=8Ciha=C5=99?= Date: Thu, 6 Aug 2026 10:33:00 +0200 Subject: [PATCH] fix(oidc): validate ID tokens on refresh Refresh responses could replace a trusted ID token without validating its signature or identity continuity. Validate refreshed tokens and bind their subject, audiences, authorized party, nonce, and authentication time before persistence. Closes #589 --- CHANGELOG.md | 7 + social_core/backends/open_id_connect.py | 177 ++++++++++- social_core/tests/backends/open_id_connect.py | 17 ++ social_core/tests/backends/test_cas.py | 9 + social_core/tests/backends/test_google.py | 7 + .../tests/backends/test_open_id_connect.py | 275 ++++++++++++++++++ 6 files changed, 485 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2e2fdd264..6bdf2c604 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/social_core/backends/open_id_connect.py b/social_core/backends/open_id_connect.py index 952bf2ef9..261be727c 100644 --- a/social_core/backends/open_id_connect.py +++ b/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,7 +257,7 @@ 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"]: @@ -264,9 +265,12 @@ def validate_claims(self, id_token) -> None: # 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( diff --git a/social_core/tests/backends/open_id_connect.py b/social_core/tests/backends/open_id_connect.py index c80e9d66c..f502895b3 100644 --- a/social_core/tests/backends/open_id_connect.py +++ b/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: diff --git a/social_core/tests/backends/test_cas.py b/social_core/tests/backends/test_cas.py index 30f99b6b3..b8c6d3402 100644 --- a/social_core/tests/backends/test_cas.py +++ b/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}, + ) diff --git a/social_core/tests/backends/test_google.py b/social_core/tests/backends/test_google.py index fbf3abbdd..5800cfb9c 100644 --- a/social_core/tests/backends/test_google.py +++ b/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( { diff --git a/social_core/tests/backends/test_open_id_connect.py b/social_core/tests/backends/test_open_id_connect.py index 21ed6023d..977fdcda9 100644 --- a/social_core/tests/backends/test_open_id_connect.py +++ b/social_core/tests/backends/test_open_id_connect.py @@ -1,7 +1,10 @@ from __future__ import annotations +import copy +import datetime import json from typing import Protocol, cast +from unittest.mock import patch import jwt import responses @@ -131,6 +134,268 @@ def test_pkce_can_be_enabled_by_setting(self) -> None: self.assert_pkce_enabled() + 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 assert_refresh_rejected(self, body: str, message: str) -> None: + social = self.login_for_refresh() + original_extra_data = copy.deepcopy(social.extra_data) + + with self.assertRaisesRegex(AuthTokenError, message): + self.refresh_social(social, body) + + self.assertEqual(social.extra_data, original_extra_data) + + def test_refresh_without_id_token_preserves_context(self) -> None: + social = self.login_for_refresh() + original_id_token = social.extra_data["id_token"] + original_context = copy.deepcopy( + social.extra_data[self.backend.ID_TOKEN_CONTEXT_KEY] + ) + + self.refresh_social( + social, + json.dumps( + { + "access_token": "refreshed-access-token", + "token_type": "bearer", + } + ), + ) + + self.assertEqual(social.extra_data["access_token"], "refreshed-access-token") + self.assertEqual(social.extra_data["id_token"], original_id_token) + self.assertEqual( + social.extra_data[self.backend.ID_TOKEN_CONTEXT_KEY], + original_context, + ) + + def test_refresh_validates_id_token_without_nonce(self) -> None: + social = self.login_for_refresh() + original_context = copy.deepcopy( + social.extra_data[self.backend.ID_TOKEN_CONTEXT_KEY] + ) + body = self.refresh_response() + + self.refresh_social(social, body) + + self.assertEqual(social.extra_data["access_token"], "refreshed-access-token") + self.assertEqual(social.extra_data["id_token"], json.loads(body)["id_token"]) + self.assertEqual( + social.extra_data[self.backend.ID_TOKEN_CONTEXT_KEY], + original_context, + ) + + def test_refresh_rejects_missing_access_token(self) -> None: + body = self.prepare_access_token_body( + access_token=None, + include_nonce=False, + ) + self.assert_refresh_rejected( + body, + "Missing access_token in OpenID Connect refresh response", + ) + + def test_refresh_rejects_invalid_signature(self) -> None: + self.assert_refresh_rejected( + self.refresh_response(tamper_message=True), + "Signature verification failed", + ) + + def test_refresh_rejects_expired_id_token(self) -> None: + expiration = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta( + seconds=30 + ) + self.assert_refresh_rejected( + self.refresh_response(expiration_datetime=expiration), + "Signature has expired", + ) + + def test_refresh_rejects_invalid_issuer(self) -> None: + self.assert_refresh_rejected( + self.refresh_response(issuer="https://invalid.example.com"), + "Invalid issuer", + ) + + def test_refresh_rejects_invalid_audience(self) -> None: + self.assert_refresh_rejected( + self.refresh_response(client_key="invalid-client"), + "Invalid audience", + ) + + def test_refresh_rejects_changed_audience_set(self) -> None: + self.assert_refresh_rejected( + self.refresh_response(client_key=[self.client_key, "another-audience"]), + "Incorrect refreshed id_token: aud", + ) + + def test_refresh_rejects_missing_azp_for_multiple_audiences(self) -> None: + self.assert_refresh_rejected( + self.refresh_response( + client_key=[self.client_key, "another-audience"], + include_azp=False, + ), + "Incorrect id_token: azp", + ) + + def test_refresh_rejects_invalid_azp(self) -> None: + self.assert_refresh_rejected( + self.refresh_response(authorized_party="another-audience"), + "Incorrect id_token: azp", + ) + + def test_refresh_rejects_changed_azp(self) -> None: + social = self.login_for_refresh(include_azp=False) + original_extra_data = copy.deepcopy(social.extra_data) + + with self.assertRaisesRegex( + AuthTokenError, + "Incorrect refreshed id_token: azp", + ): + self.refresh_social(social, self.refresh_response()) + + self.assertEqual(social.extra_data, original_extra_data) + + def test_refresh_rejects_omitted_azp(self) -> None: + social = self.login_for_refresh() + original_extra_data = copy.deepcopy(social.extra_data) + + with self.assertRaisesRegex( + AuthTokenError, + "Incorrect refreshed id_token: azp", + ): + self.refresh_social( + social, + self.refresh_response(include_azp=False), + ) + + self.assertEqual(social.extra_data, original_extra_data) + + def test_refresh_rejects_invalid_at_hash(self) -> None: + self.assert_refresh_rejected( + self.refresh_response(at_hash="invalid-hash"), + "Invalid access token", + ) + + def test_refresh_rejects_changed_subject(self) -> None: + self.assert_refresh_rejected( + self.refresh_response(subject="different-subject"), + "Incorrect refreshed id_token: sub", + ) + + def test_refresh_rejects_changed_auth_time(self) -> None: + social = self.login_for_refresh(auth_time=1_700_000_000) + original_extra_data = copy.deepcopy(social.extra_data) + + with self.assertRaisesRegex( + AuthTokenError, + "Incorrect refreshed id_token: auth_time", + ): + self.refresh_social( + social, + self.refresh_response(auth_time=1_700_000_001), + ) + + self.assertEqual(social.extra_data, original_extra_data) + + def test_refresh_accepts_matching_nonce(self) -> None: + social = self.login_for_refresh() + context = social.extra_data[self.backend.ID_TOKEN_CONTEXT_KEY] + + self.refresh_social( + social, + self.prepare_access_token_body( + access_token="refreshed-access-token", # noqa: S106 + nonce=context["nonce"], + ), + ) + + self.assertEqual(social.extra_data["access_token"], "refreshed-access-token") + + def test_refresh_rejects_changed_nonce(self) -> None: + self.assert_refresh_rejected( + self.prepare_access_token_body( + access_token="refreshed-access-token", # noqa: S106 + nonce="different-nonce", + ), + "Incorrect refreshed id_token: nonce", + ) + + def test_refresh_seeds_missing_legacy_context(self) -> None: + social = self.login_for_refresh() + social.extra_data.pop(self.backend.ID_TOKEN_CONTEXT_KEY) + + self.refresh_social(social, self.refresh_response()) + + self.assertEqual( + social.extra_data[self.backend.ID_TOKEN_CONTEXT_KEY]["sub"], + "1234", + ) + original_extra_data = copy.deepcopy(social.extra_data) + with self.assertRaisesRegex( + AuthTokenError, + "Incorrect refreshed id_token: sub", + ): + self.refresh_social( + social, + self.refresh_response(subject="different-subject"), + ) + self.assertEqual(social.extra_data, original_extra_data) + + def test_legacy_refresh_rejects_changed_subject(self) -> None: + social = self.login_for_refresh() + social.extra_data.pop(self.backend.ID_TOKEN_CONTEXT_KEY) + original_extra_data = copy.deepcopy(social.extra_data) + + with self.assertRaisesRegex( + AuthTokenError, + "Incorrect refreshed id_token: sub", + ): + self.refresh_social( + social, + self.refresh_response(subject="different-subject"), + ) + + self.assertEqual(social.extra_data, original_extra_data) + + def test_legacy_refresh_requires_subject_identity_key(self) -> None: + social = self.login_for_refresh() + social.extra_data.pop(self.backend.ID_TOKEN_CONTEXT_KEY) + original_extra_data = copy.deepcopy(social.extra_data) + + with ( + patch.object(OpenIdConnectAuth, "id_key", return_value="username"), + self.assertRaisesRegex( + AuthTokenError, + "reauthentication required", + ), + ): + self.refresh_social(social, self.refresh_response()) + + self.assertEqual(social.extra_data, original_extra_data) + class ExampleOpenIdConnectAuth(OpenIdConnectAuth): name = "example123" @@ -337,6 +602,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, ): if at_hash is None and access_token is not None: at_hash = OpenIdConnectAuth.calc_at_hash(access_token, "RS256", "sha512") @@ -351,6 +621,11 @@ def prepare_access_token_body( # NOQA: PLR0913, PLR0917 at_hash=at_hash, subject=subject, access_token=access_token, + refresh_token=refresh_token, + include_nonce=include_nonce, + auth_time=auth_time, + include_azp=include_azp, + authorized_party=authorized_party, ) def test_everything_works(self) -> None: