Repository: pennersr/django-allauth
Pull request #3629 (2024-02-06 08:07:08)

Don't check Redirect Url if there's no Redirect
## Don't check Redirect Url if there's no Redirect

 - [x] Make sure you use [semantic commit messages](https://seesparkbox.com/foundry/semantic_commit_messages).
       Examples: `"fix(google): Fixed foobar bug"`, `"feat(accounts): Added foobar feature"`.
 - [x] All Python code must formatted using Black, and clean from pep8 and isort issues.
 - [x] JavaScript code should adhere to [StandardJS](https://standardjs.com).
 - [x] If your changes are significant, please update `ChangeLog.rst`.
 - [x] If your change is substantial, feel free to add yourself to `AUTHORS`.



coveralls: 
[![Coverage Status](https://coveralls.io/builds/65522457/badge)](https://coveralls.io/builds/65522457)

coverage: 95.798% (-0.007%) from 95.805%
when pulling **661f8a8b45db104f71cbbe6bfd79f4cffd6a898b on Akay7:fixRedirectUrl**
into **ad893888ac5504bc2847837d592f31dcbfa48bb0 on pennersr:main**.


Patch:
diff --git a/ChangeLog.rst b/ChangeLog.rst
index 1355853e..95c16796 100644
--- a/ChangeLog.rst
+++ b/ChangeLog.rst
@@ -9,6 +9,20 @@ 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
+---------------
+
+- MFA: It was possible to reuse a valid TOTP code within its time window. This
+  has now been addressed. As a result, a user can now only login once per 30
+  seconds (``MFA_TOTP_PERIOD``).
+
 
 Backwards incompatible changes
 ------------------------------
diff --git a/allauth/account/utils.py b/allauth/account/utils.py
index a1ff612f..c44431ef 100644
--- a/allauth/account/utils.py
+++ b/allauth/account/utils.py
@@ -39,7 +39,7 @@ def get_next_redirect_url(request, redirect_field_name="next"):
     via the request.
     """
     redirect_to = get_request_param(request, redirect_field_name)
-    if not get_adapter().is_safe_url(redirect_to):
+    if redirect_to and not get_adapter().is_safe_url(redirect_to):
         redirect_to = None
     return redirect_to
 
diff --git a/allauth/mfa/totp.py b/allauth/mfa/totp.py
index 6e2620de..8a33480d 100644
--- a/allauth/mfa/totp.py
+++ b/allauth/mfa/totp.py
@@ -7,6 +7,7 @@ import time
 from io import BytesIO
 from urllib.parse import quote
 
+from django.core.cache import cache
 from django.utils.http import urlencode
 
 import qrcode
@@ -104,5 +105,20 @@ class TOTP:
         self.instance.delete()
 
     def validate_code(self, code):
+        if self._is_code_used(code):
+            return False
+
         secret = decrypt(self.instance.data["secret"])
-        return validate_totp_code(secret, code)
+        valid = validate_totp_code(secret, code)
+        if valid:
+            self._mark_code_used(code)
+        return valid
+
+    def _get_used_cache_key(self, code):
+        return f"allauth.mfa.totp.used?user={self.instance.user_id}&code={code}"
+
+    def _is_code_used(self, code):
+        return cache.get(self._get_used_cache_key(code)) == "y"
+
+    def _mark_code_used(self, code):
+        cache.set(self._get_used_cache_key(code), "y", timeout=app_settings.TOTP_PERIOD)
diff --git a/allauth/socialaccount/providers/google/provider.py b/allauth/socialaccount/providers/google/provider.py
index a727c8bd..eaf92bd0 100644
--- a/allauth/socialaccount/providers/google/provider.py
+++ b/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 @@ class GoogleProvider(OAuth2Provider):
         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 @@ class GoogleProvider(OAuth2Provider):
     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
 
 
diff --git a/allauth/socialaccount/providers/google/views.py b/allauth/socialaccount/providers/google/views.py
index de6becd9..c16dab38 100644
--- a/allauth/socialaccount/providers/google/views.py
+++ b/allauth/socialaccount/providers/google/views.py
@@ -24,9 +24,17 @@ from allauth.socialaccount.providers.oauth2.views import (
 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 @@ class GoogleOAuth2Adapter(OAuth2Adapter):
             )
         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)


Test patch:
diff --git a/allauth/mfa/tests/test_views.py b/allauth/mfa/tests/test_views.py
index 026ce720..580a8df8 100644
--- a/allauth/mfa/tests/test_views.py
+++ b/allauth/mfa/tests/test_views.py
@@ -2,6 +2,8 @@ from unittest.mock import ANY, patch
 
 import django
 from django.conf import settings
+from django.core.cache import cache
+from django.test import Client
 from django.urls import reverse
 
 import pytest
@@ -301,3 +303,44 @@ def test_cannot_deactivate_totp(auth_client, user_with_totp, user_password):
         assert resp.context["form"].errors == {
             "__all__": [get_adapter().error_messages["cannot_delete_authenticator"]],
         }
+
+
+def test_totp_code_reuse(
+    user_with_totp, user_password, totp_validation_bypass, enable_cache
+):
+    for code, time_lapse, expect_success in [
+        # First use of code, SUCCESS
+        ("123", False, True),
+        # Second use, no time elapsed: FAIL
+        ("123", False, False),
+        # Different code, no time elapsed: SUCCESS
+        ("456", False, True),
+        # Again, previous code, no time elapsed: FAIL
+        ("123", False, False),
+        # Previous code, but time elapsed: SUCCESS
+        ("123", True, True),
+    ]:
+        if time_lapse:
+            cache.clear()
+        client = Client()
+        resp = client.post(
+            reverse("account_login"),
+            {"login": user_with_totp.username, "password": user_password},
+        )
+        assert resp.status_code == 302
+        assert resp["location"] == reverse("mfa_authenticate")
+        # Note that this bypass only bypasses the actual code check, not the
+        # re-use check we're testing here.
+        with totp_validation_bypass():
+            resp = client.post(
+                reverse("mfa_authenticate"),
+                {"code": code},
+            )
+        if expect_success:
+            assert resp.status_code == 302
+            assert resp["location"] == settings.LOGIN_REDIRECT_URL
+        else:
+            assert resp.status_code == 200
+            assert resp.context["form"].errors == {
+                "code": [get_adapter().error_messages["incorrect_code"]]
+            }
diff --git a/allauth/socialaccount/providers/google/tests.py b/allauth/socialaccount/providers/google/tests.py
index f4ba3036..eb059eee 100644
--- a/allauth/socialaccount/providers/google/tests.py
+++ b/allauth/socialaccount/providers/google/tests.py
@@ -13,11 +13,13 @@ from django.urls import reverse
 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 @@ class GoogleTests(OAuth2TestsMixin, TestCase):
         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