From f77f0acb9284ff59455205e109b17cd86661a13d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Michal=20=C4=8Ciha=C5=99?= Date: Fri, 7 Aug 2026 08:35:00 +0200 Subject: [PATCH] fix(oidc): simplify refresh identity validation Require core claims before login and validate refreshed ID tokens without relying on the authentication-request nonce. Persist an immutable identity context while retaining the latest ID token. This removes UID and backend-specific migration logic, accepts valid authorized-party claim changes and audience reordering, and avoids false mismatches for custom user IDs and repeated refreshes. Legacy associations establish their baseline on the first fully validated refresh; malformed stored contexts require reauthentication. --- CHANGELOG.md | 9 + social_core/backends/open_id_connect.py | 136 ++++----- social_core/exceptions.py | 7 + social_core/tests/backends/open_id_connect.py | 31 +- social_core/tests/backends/test_cas.py | 17 +- social_core/tests/backends/test_google.py | 36 ++- .../tests/backends/test_open_id_connect.py | 264 +++++++++--------- social_core/tests/test_exceptions.py | 9 + 8 files changed, 279 insertions(+), 230 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bf1d27637..f23a3b272 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/social_core/backends/open_id_connect.py b/social_core/backends/open_id_connect.py index 261be727c..645a9ebbf 100644 --- a/social_core/backends/open_id_connect.py +++ b/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,6 +364,7 @@ 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 @@ -373,7 +372,8 @@ def validate_and_return_id_token(self, id_token, access_token): 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,69 +441,39 @@ 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: @@ -511,20 +481,12 @@ def validate_id_token_audience_context(self, previous, current) -> None: 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 diff --git a/social_core/exceptions.py b/social_core/exceptions.py index 91e2666ff..2d8b99369 100644 --- a/social_core/exceptions.py +++ b/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.""" diff --git a/social_core/tests/backends/open_id_connect.py b/social_core/tests/backends/open_id_connect.py index f502895b3..b3effac5f 100644 --- a/social_core/tests/backends/open_id_connect.py +++ b/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): diff --git a/social_core/tests/backends/test_cas.py b/social_core/tests/backends/test_cas.py index b8c6d3402..bd17384a7 100644 --- a/social_core/tests/backends/test_cas.py +++ b/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") diff --git a/social_core/tests/backends/test_google.py b/social_core/tests/backends/test_google.py index 5800cfb9c..327903b9a 100644 --- a/social_core/tests/backends/test_google.py +++ b/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( diff --git a/social_core/tests/backends/test_open_id_connect.py b/social_core/tests/backends/test_open_id_connect.py index 977fdcda9..21f66d8ab 100644 --- a/social_core/tests/backends/test_open_id_connect.py +++ b/social_core/tests/backends/test_open_id_connect.py @@ -4,17 +4,29 @@ import datetime import json from typing import Protocol, cast -from unittest.mock import patch import jwt import responses from social_core.backends.open_id_connect import OpenIdConnectAuth -from social_core.exceptions import AuthInvalidParameter, AuthTokenError +from social_core.exceptions import ( + AuthInvalidParameter, + AuthReauthenticationRequired, + AuthTokenError, +) from social_core.utils import get_querystring, parse_qs from .oauth import BaseAuthUrlTestMixin -from .open_id_connect import OpenIdConnectTest +from .open_id_connect import STORED_ID_TOKEN_CONTEXT_KEY, OpenIdConnectTest + + +def decode_id_token_context(id_token: str) -> dict: + claims = jwt.decode(id_token, options={"verify_signature": False}) + return { + claim: claims[claim] + for claim in ("iss", "sub", "aud", "auth_time", "nonce") + if claim in claims + } class OpenIdConnectPkceAssertionsCapable(Protocol): @@ -134,31 +146,6 @@ 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) @@ -168,11 +155,21 @@ def assert_refresh_rejected(self, body: str, message: str) -> None: self.assertEqual(social.extra_data, original_extra_data) - def test_refresh_without_id_token_preserves_context(self) -> None: + def assert_refresh_requires_reauthentication(self, social) -> None: + original_extra_data = copy.deepcopy(social.extra_data) + + with self.assertRaises(AuthReauthenticationRequired): + self.refresh_social(social, self.refresh_response()) + + self.assertEqual(social.extra_data, original_extra_data) + + def test_refresh_without_id_token_preserves_id_token(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] + original_context = decode_id_token_context(original_id_token) + self.assertEqual( + social.extra_data[STORED_ID_TOKEN_CONTEXT_KEY], + original_context, ) self.refresh_social( @@ -188,23 +185,45 @@ def test_refresh_without_id_token_preserves_context(self) -> None: 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], + social.extra_data[STORED_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() + def test_refresh_retains_original_claims_across_refreshes(self) -> None: + auth_time = 1_700_000_000 + social = self.login_for_refresh(auth_time=auth_time) + original_context = copy.deepcopy(social.extra_data[STORED_ID_TOKEN_CONTEXT_KEY]) - self.refresh_social(social, body) + first_body = self.refresh_response() + self.refresh_social(social, first_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], + social.extra_data[STORED_ID_TOKEN_CONTEXT_KEY], + original_context, + ) + self.assertEqual( + social.extra_data["id_token"], + json.loads(first_body)["id_token"], + ) + + second_body = self.prepare_access_token_body( + access_token="second-refreshed-access-token", # noqa: S106 + nonce=original_context["nonce"], + auth_time=auth_time, + ) + self.refresh_social(social, second_body) + + self.assertEqual( + social.extra_data["access_token"], + "second-refreshed-access-token", + ) + self.assertEqual( + social.extra_data["id_token"], + json.loads(second_body)["id_token"], + ) + self.assertEqual( + social.extra_data[STORED_ID_TOKEN_CONTEXT_KEY], original_context, ) @@ -224,25 +243,21 @@ def test_refresh_rejects_invalid_signature(self) -> None: "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_stale_issue_time(self) -> None: + issue_datetime = datetime.datetime.now( + datetime.timezone.utc + ) - datetime.timedelta( + seconds=self.backend.ID_TOKEN_MAX_AGE * 2, ) - - def test_refresh_rejects_invalid_issuer(self) -> None: self.assert_refresh_rejected( - self.refresh_response(issuer="https://invalid.example.com"), - "Invalid issuer", + self.refresh_response(issue_datetime=issue_datetime), + "Incorrect id_token: iat", ) - def test_refresh_rejects_invalid_audience(self) -> None: + def test_refresh_rejects_missing_expiration(self) -> None: self.assert_refresh_rejected( - self.refresh_response(client_key="invalid-client"), - "Invalid audience", + self.refresh_response(exclude_claims=("exp",)), + "Incorrect id_token: exp", ) def test_refresh_rejects_changed_audience_set(self) -> None: @@ -251,54 +266,56 @@ def test_refresh_rejects_changed_audience_set(self) -> None: "Incorrect refreshed id_token: aud", ) - def test_refresh_rejects_missing_azp_for_multiple_audiences(self) -> None: - self.assert_refresh_rejected( + def test_refresh_accepts_reordered_audience_set(self) -> None: + social = self.login_for_refresh( + client_key=[self.client_key, "another-audience"], + include_azp=False, + ) + + self.refresh_social( + social, self.refresh_response( - client_key=[self.client_key, "another-audience"], + client_key=["another-audience", self.client_key], include_azp=False, ), - "Incorrect id_token: azp", ) + self.assertEqual(social.extra_data["access_token"], "refreshed-access-token") + + def test_refresh_accepts_missing_azp_for_multiple_audiences(self) -> None: + audiences = [self.client_key, "another-audience"] + social = self.login_for_refresh(client_key=audiences, include_azp=False) + + self.refresh_social( + social, + self.refresh_response(client_key=audiences, include_azp=False), + ) + + self.assertEqual(social.extra_data["access_token"], "refreshed-access-token") + 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: + def test_refresh_accepts_added_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.refresh_social(social, self.refresh_response()) - self.assertEqual(social.extra_data, original_extra_data) + self.assertEqual(social.extra_data["access_token"], "refreshed-access-token") - def test_refresh_rejects_omitted_azp(self) -> None: + def test_refresh_accepts_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", + self.refresh_social( + social, + self.refresh_response(include_azp=False), ) + self.assertEqual(social.extra_data["access_token"], "refreshed-access-token") + def test_refresh_rejects_changed_subject(self) -> None: self.assert_refresh_rejected( self.refresh_response(subject="different-subject"), @@ -320,20 +337,6 @@ def test_refresh_rejects_changed_auth_time(self) -> None: 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( @@ -343,31 +346,20 @@ def test_refresh_rejects_changed_nonce(self) -> None: "Incorrect refreshed id_token: nonce", ) - def test_refresh_seeds_missing_legacy_context(self) -> None: + def test_refresh_without_context_establishes_new_baseline(self) -> None: social = self.login_for_refresh() - social.extra_data.pop(self.backend.ID_TOKEN_CONTEXT_KEY) - - self.refresh_social(social, self.refresh_response()) + social.extra_data.pop("id_token") + social.extra_data.pop(STORED_ID_TOKEN_CONTEXT_KEY) + self.refresh_social( + social, + self.refresh_response(subject="migrated-subject"), + ) self.assertEqual( - social.extra_data[self.backend.ID_TOKEN_CONTEXT_KEY]["sub"], - "1234", + social.extra_data[STORED_ID_TOKEN_CONTEXT_KEY]["sub"], + "migrated-subject", ) - 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) + migrated_extra_data = copy.deepcopy(social.extra_data) with self.assertRaisesRegex( AuthTokenError, @@ -375,26 +367,16 @@ def test_legacy_refresh_rejects_changed_subject(self) -> None: ): self.refresh_social( social, - self.refresh_response(subject="different-subject"), + self.refresh_response(subject="another-subject"), ) - self.assertEqual(social.extra_data, original_extra_data) + self.assertEqual(social.extra_data, migrated_extra_data) - def test_legacy_refresh_requires_subject_identity_key(self) -> None: + def test_refresh_requires_reauthentication_for_incomplete_context(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()) + social.extra_data[STORED_ID_TOKEN_CONTEXT_KEY].pop("sub") - self.assertEqual(social.extra_data, original_extra_data) + self.assert_refresh_requires_reauthentication(social) class ExampleOpenIdConnectAuth(OpenIdConnectAuth): @@ -453,6 +435,18 @@ def test_user_id_comes_from_id_token_when_userinfo_omits_sub(self) -> None: self.assertEqual(user.social[0].uid, "1234") + def test_missing_id_token_subject_raises_error(self) -> None: + self.authtoken_raised( + "Incorrect id_token: sub", + exclude_claims=("sub",), + ) + + def test_missing_id_token_expiration_raises_error(self) -> None: + self.authtoken_raised( + "Incorrect id_token: exp", + exclude_claims=("exp",), + ) + def test_matching_userinfo_sub_succeeds(self) -> None: self.userinfo_response["sub"] = "1234" @@ -607,6 +601,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, ...] = (), ): if at_hash is None and access_token is not None: at_hash = OpenIdConnectAuth.calc_at_hash(access_token, "RS256", "sha512") @@ -626,6 +621,7 @@ def prepare_access_token_body( # NOQA: PLR0913, PLR0917 auth_time=auth_time, include_azp=include_azp, authorized_party=authorized_party, + exclude_claims=exclude_claims, ) def test_everything_works(self) -> None: diff --git a/social_core/tests/test_exceptions.py b/social_core/tests/test_exceptions.py index 22897b7a9..1f0ba2e52 100644 --- a/social_core/tests/test_exceptions.py +++ b/social_core/tests/test_exceptions.py @@ -8,6 +8,7 @@ AuthFailed, AuthForbidden, AuthMissingParameter, + AuthReauthenticationRequired, AuthStateForbidden, AuthStateMissing, AuthTokenError, @@ -57,6 +58,14 @@ class AuthTokenErrorTest(BaseExceptionTestCase): expected_message = "Token error: Incorrect tokens" +class AuthReauthenticationRequiredTest(BaseExceptionTestCase): + exception = AuthReauthenticationRequired(BaseAuth(TestStrategy(TestStorage))) + expected_message = "Token error: reauthentication required" + + def test_is_token_error(self) -> None: + self.assertIsInstance(self.exception, AuthTokenError) + + class AuthMissingParameterTest(BaseExceptionTestCase): exception = AuthMissingParameter(BaseAuth(TestStrategy(TestStorage)), "username") expected_message = "Missing needed parameter username"