fix(google): Gracefully handle cases where id_token is absent

ChangeLog.rst

@@ -9,6 +9,12 @@ Note worthy changes
   password was changed", including information on user agent / IP address from where the change
   originated, will be emailed.
 
+- Google: Starting from 0.52.0, the ``id_token`` is being used for extracting
+  user information.  To accommodate for scenario's where django-allauth is used
+  in contexts where the ``id_token`` is not posted, the provider now looks up
+  the required information from the ``/userinfo`` endpoint based on the access
+  token if the ``id_token`` is absent.
+
 
 Security notice
 ---------------

allauth/socialaccount/providers/google/provider.py

@@ -10,6 +10,35 @@ class Scope(object):
 
 
 class GoogleAccount(ProviderAccount):
+    """
+    The account data can be in two formats. One, originating from
+    the /v2/userinfo endpoint:
+
+        {'email': 'john.doe@gmail.com',
+         'given_name': 'John',
+         'id': '12345678901234567890',
+         'locale': 'en',
+         'name': 'John',
+         'picture': 'https://lh3.googleusercontent.com/a/code',
+         'verified_email': True}
+
+    The second, which is the payload of the id_token:
+
+        {'at_hash': '-someHASH',
+         'aud': '123-pqr.apps.googleusercontent.com',
+         'azp': '123-pqr.apps.googleusercontent.com',
+         'email': 'john.doe@gmail.com',
+         'email_verified': True,
+         'exp': 1707297277,
+         'given_name': 'John',
+         'iat': 1707293677,
+         'iss': 'https://accounts.google.com',
+         'locale': 'en',
+         'name': 'John',
+         'picture': 'https://lh3.googleusercontent.com/a/code',
+         'sub': '12345678901234567890'}
+    """
+
     def get_profile_url(self):
         return self.account.extra_data.get("link")
 
@@ -39,7 +68,9 @@ def get_auth_params(self, request, action):
         return ret
 
     def extract_uid(self, data):
-        return data["sub"]
+        if "sub" in data:
+            return data["sub"]
+        return data["id"]
 
     def extract_common_fields(self, data):
         return dict(
@@ -51,8 +82,9 @@ def extract_common_fields(self, data):
     def extract_email_addresses(self, data):
         ret = []
         email = data.get("email")
-        if email and data.get("email_verified"):
-            ret.append(EmailAddress(email=email, verified=True, primary=True))
+        if email:
+            verified = bool(data.get("email_verified") or data.get("verified_email"))
+            ret.append(EmailAddress(email=email, verified=verified, primary=True))
         return ret
 
 

allauth/socialaccount/providers/google/tests.py

@@ -13,11 +13,13 @@
 import pytest
 
 from allauth.account import app_settings as account_settings
-from allauth.account.adapter import get_adapter
+from allauth.account.adapter import get_adapter as get_account_adapter
 from allauth.account.models import EmailAddress, EmailConfirmation
 from allauth.account.signals import user_signed_up
-from allauth.socialaccount.models import SocialAccount
+from allauth.socialaccount.adapter import get_adapter
+from allauth.socialaccount.models import SocialAccount, SocialToken
 from allauth.socialaccount.providers.apple.client import jwt_encode
+from allauth.socialaccount.providers.google.views import GoogleOAuth2Adapter
 from allauth.socialaccount.tests import OAuth2TestsMixin
 from allauth.tests import TestCase, mocked_response
 
@@ -155,7 +157,7 @@ def test_email_verified_stashed(self):
         self.client.cookies[settings.SESSION_COOKIE_NAME] = store.session_key
         request = RequestFactory().get("/")
         request.session = self.client.session
-        adapter = get_adapter()
+        adapter = get_account_adapter()
         adapter.stash_verified_email(request, self.email)
         request.session.save()
 
@@ -271,3 +273,70 @@ def test_login_by_token(db, client, settings_with_google_provider):
                     assert resp.status_code == 302
                     socialaccount = SocialAccount.objects.get(uid="123sub")
                     assert socialaccount.user.email == "raymond@example.com"
+
+
+@pytest.mark.parametrize(
+    "id_key,verified_key",
+    [
+        ("id", "email_verified"),
+        ("sub", "verified_email"),
+    ],
+)
+@pytest.mark.parametrize("verified", [False, True])
+def test_extract_data(
+    id_key, verified_key, verified, settings_with_google_provider, db
+):
+    data = {
+        "email": "a@b.com",
+    }
+    data[id_key] = "123"
+    data[verified_key] = verified
+    provider = get_adapter().get_provider(None, GoogleProvider.id)
+    assert provider.extract_uid(data) == "123"
+    emails = provider.extract_email_addresses(data)
+    assert len(emails) == 1
+    assert emails[0].verified == verified
+    assert emails[0].email == "a@b.com"
+
+
+@pytest.mark.parametrize(
+    "fetch_userinfo,id_token_has_picture,response,expected_uid, expected_picture",
+    [
+        (True, True, {"id_token": "123"}, "uid-from-id-token", "pic-from-id-token"),
+        (True, False, {"id_token": "123"}, "uid-from-id-token", "pic-from-userinfo"),
+        (True, True, {"access_token": "123"}, "uid-from-userinfo", "pic-from-userinfo"),
+    ],
+)
+def test_complete_login_variants(
+    response,
+    settings_with_google_provider,
+    db,
+    fetch_userinfo,
+    expected_uid,
+    expected_picture,
+    id_token_has_picture,
+):
+    with patch.object(
+        GoogleOAuth2Adapter,
+        "_fetch_user_info",
+        return_value={
+            "id": "uid-from-userinfo",
+            "picture": "pic-from-userinfo",
+        },
+    ):
+        id_token = {"sub": "uid-from-id-token"}
+        if id_token_has_picture:
+            id_token["picture"] = "pic-from-id-token"
+        with patch.object(
+            GoogleOAuth2Adapter,
+            "_decode_id_token",
+            return_value=id_token,
+        ):
+            request = None
+            app = None
+            adapter = GoogleOAuth2Adapter(request)
+            adapter.fetch_userinfo = fetch_userinfo
+            token = SocialToken()
+            login = adapter.complete_login(request, app, token, response)
+            assert login.account.uid == expected_uid
+            assert login.account.extra_data["picture"] == expected_picture

allauth/socialaccount/providers/google/views.py

@@ -24,9 +24,17 @@
 from .provider import GoogleProvider
 
 
-CERTS_URL = "https://www.googleapis.com/oauth2/v1/certs"
+CERTS_URL = (
+    getattr(settings, "SOCIALACCOUNT_PROVIDERS", {})
+    .get("google", {})
+    .get("CERTS_URL", "https://www.googleapis.com/oauth2/v1/certs")
+)
 
-IDENTITY_URL = "https://www.googleapis.com/oauth2/v2/userinfo"
+IDENTITY_URL = (
+    getattr(settings, "SOCIALACCOUNT_PROVIDERS", {})
+    .get("google", {})
+    .get("IDENTITY_URL", "https://www.googleapis.com/oauth2/v2/userinfo")
+)
 
 ACCESS_TOKEN_URL = (
     getattr(settings, "SOCIALACCOUNT_PROVIDERS", {})
@@ -62,9 +70,24 @@ class GoogleOAuth2Adapter(OAuth2Adapter):
     fetch_userinfo = FETCH_USERINFO
 
     def complete_login(self, request, app, token, response, **kwargs):
+        data = None
+        id_token = response.get("id_token")
+        if id_token:
+            data = self._decode_id_token(app, id_token)
+            if self.fetch_userinfo and "picture" not in data:
+                info = self._fetch_user_info(token.token)
+                picture = info.get("picture")
+                if picture:
+                    data["picture"] = picture
+        else:
+            data = self._fetch_user_info(token.token)
+        login = self.get_provider().sociallogin_from_response(request, data)
+        return login
+
+    def _decode_id_token(self, app, id_token):
         try:
-            identity_data = jwt.decode(
-                response["id_token"],
+            data = jwt.decode(
+                id_token,
                 # Since the token was received by direct communication
                 # protected by TLS between this library and Google, we
                 # are allowed to skip checking the token signature
@@ -82,22 +105,20 @@ def complete_login(self, request, app, token, response, **kwargs):
             )
         except jwt.PyJWTError as e:
             raise OAuth2Error("Invalid id_token") from e
-
-        if self.fetch_userinfo and "picture" not in identity_data:
-            resp = (
-                get_adapter()
-                .get_requests_session()
-                .get(
-                    self.identity_url,
-                    headers={"Authorization": "Bearer {}".format(token)},
-                )
+        return data
+
+    def _fetch_user_info(self, access_token):
+        resp = (
+            get_adapter()
+            .get_requests_session()
+            .get(
+                self.identity_url,
+                headers={"Authorization": "Bearer {}".format(access_token)},
             )
-            if not resp.ok:
-                raise OAuth2Error("Request to user info failed")
-            identity_data["picture"] = resp.json()["picture"]
-
-        login = self.get_provider().sociallogin_from_response(request, identity_data)
-        return login
+        )
+        if not resp.ok:
+            raise OAuth2Error("Request to user info failed")
+        return resp.json()
 
 
 oauth2_login = OAuth2LoginView.adapter_view(GoogleOAuth2Adapter)