From 59de9f24439109a382ff040cfb5e7d420b0967ac Mon Sep 17 00:00:00 2001 From: Herculino Trotta Date: Sat, 10 Oct 2026 17:01:52 -0300 Subject: [PATCH] feat(auth): configure OIDC email trust Allow verified-only linking with OIDC_TRUST_EMAIL=false while retaining case-insensitive matching. Keep unconditional trust enabled by default for compatibility with #518; that mode retains the documented takeover risk. --- app/WYGIWYH/settings.py | 5 ++ app/apps/users/adapters.py | 73 ++++++--------------------- app/apps/users/tests.py | 100 ++++++++++++++++++++++++++++++++++++- 3 files changed, 119 insertions(+), 59 deletions(-) diff --git a/app/WYGIWYH/settings.py b/app/WYGIWYH/settings.py index c8292782..f295efc6 100644 --- a/app/WYGIWYH/settings.py +++ b/app/WYGIWYH/settings.py @@ -372,6 +372,11 @@ if ( "secret": os.getenv("OIDC_CLIENT_SECRET"), "settings": { "server_url": os.getenv("OIDC_SERVER_URL"), + # When true, every email from the provider is treated as + # verified and auto-linked to a matching local account. When + # false, only emails with email_verified=true are linked. + "verified_email": os.getenv("OIDC_TRUST_EMAIL", "true").lower() + == "true", }, } ) diff --git a/app/apps/users/adapters.py b/app/apps/users/adapters.py index 41b02c12..7295d8df 100644 --- a/app/apps/users/adapters.py +++ b/app/apps/users/adapters.py @@ -1,75 +1,32 @@ import logging + from allauth.socialaccount.adapter import DefaultSocialAccountAdapter from django.contrib.auth import get_user_model -User = get_user_model() logger = logging.getLogger(__name__) class AutoConnectSocialAccountAdapter(DefaultSocialAccountAdapter): - """ - Custom adapter to automatically connect social accounts to existing users - with the same email address. + """Link verified (or provider-trusted) emails case-insensitively. - SECURITY WARNING: - This adapter automatically connects OIDC accounts to existing local accounts - based on email matching. - - If your OIDC provider allows unverified emails, this could lead to - ACCOUNT TAKEOVER attacks where an attacker creates an OIDC account - with someone else's email and gains access to their account. + OIDC_TRUST_EMAIL=true makes allauth treat every provider email as verified. + Only use that mode with an issuer that guarantees email ownership. """ def pre_social_login(self, request, sociallogin): - """ - Invoked just after a user successfully authenticates via a - social provider, but before the login is actually processed. - - If a user with the same email already exists, connect the social - account to that existing user instead of creating a new account. - """ - # If the social account is already connected to a user, do nothing if sociallogin.is_existing: return - # Check if we have an email from the social provider - if not sociallogin.email_addresses: - logger.warning( - "OIDC login attempted without email address. " - f"Provider: {sociallogin.account.provider}" - ) - return + for address in sociallogin.email_addresses: + if not address.verified: + continue + try: + user = get_user_model().objects.get(email__iexact=address.email) + except get_user_model().DoesNotExist: + continue + except get_user_model().MultipleObjectsReturned: + logger.error("Multiple local users match OIDC email; blocking auto-connect.") + return - # Get the email from the social login - email = sociallogin.email_addresses[0].email.lower() - - # Try to find an existing user with this email - try: - user = User.objects.get(email__iexact=email) - - # Log this connection for security audit trail - logger.info( - f"Auto-connecting OIDC account to existing user. " - f"Email: {email}, Provider: {sociallogin.account.provider}, " - f"User ID: {user.id}" - ) - - # Connect the social account to the existing user sociallogin.connect(request, user) - - except User.DoesNotExist: - # No user with this email exists, proceed with normal signup flow - logger.debug( - f"No existing user found for email {email}. " - "Proceeding with new account creation." - ) - pass - except User.MultipleObjectsReturned: - # Multiple users with the same email (shouldn't happen with unique constraint) - logger.error( - f"Multiple users found with email {email}. " - "This should not happen with unique constraint. " - "Blocking auto-connect." - ) - # Let the default behavior handle this - pass + return diff --git a/app/apps/users/tests.py b/app/apps/users/tests.py index 5bdb6150..1ca0ea01 100644 --- a/app/apps/users/tests.py +++ b/app/apps/users/tests.py @@ -1,5 +1,5 @@ from django.contrib.auth import get_user_model -from django.test import TestCase +from django.test import TestCase, override_settings from django.urls import reverse from django.utils import timezone @@ -71,3 +71,101 @@ class UserAPITokenViewsTests(TestCase): self.assertEqual(response.status_code, 404) self.assertTrue(APIToken.objects.filter(id=token.id).exists()) + + +def _oidc_providers(trust_email=False): + return { + "openid_connect": { + "APPS": [ + { + "provider_id": "test-oidc", + "name": "Test OIDC", + "client_id": "client", + "secret": "secret", + "settings": { + "server_url": "https://idp.example.com", + "verified_email": trust_email, + }, + } + ] + } + } + + +class OIDCEmailAutoConnectTests(TestCase): + """GHSA-prgm-wrjv-fvrh: OIDC logins must only link to existing local + accounts by email when the email is verified (or explicitly trusted).""" + + def setUp(self): + self.victim = get_user_model().objects.create_superuser( + email="Victim@example.com", password="victim-password" + ) + + def _login(self, claims): + from allauth.core import context + from allauth.socialaccount.adapter import get_adapter + from allauth.socialaccount.helpers import complete_social_login + from allauth.socialaccount.models import SocialLogin + from django.contrib.messages.middleware import MessageMiddleware + from django.contrib.sessions.middleware import SessionMiddleware + from django.contrib.auth.models import AnonymousUser + from django.test import RequestFactory + + request = RequestFactory().get("/auth/oidc/test-oidc/login/callback/") + SessionMiddleware(lambda r: None).process_request(request) + MessageMiddleware(lambda r: None).process_request(request) + request.user = AnonymousUser() + + provider = get_adapter().get_provider(request, "test-oidc") + sociallogin = provider.sociallogin_from_response(request, claims) + sociallogin.state = SocialLogin.state_from_request(request) + with context.request_context(request): + complete_social_login(request, sociallogin) + return request + + def _assert_not_linked(self, request): + from allauth.socialaccount.models import SocialAccount + + self.assertFalse(SocialAccount.objects.filter(user=self.victim).exists()) + self.assertNotEqual( + request.session.get("_auth_user_id"), str(self.victim.pk) + ) + + def _assert_linked(self, request): + from allauth.socialaccount.models import SocialAccount + + self.assertTrue( + SocialAccount.objects.filter(user=self.victim, uid="oidc-uid").exists() + ) + self.assertEqual(request.session.get("_auth_user_id"), str(self.victim.pk)) + + @override_settings(SOCIALACCOUNT_PROVIDERS=_oidc_providers()) + def test_unverified_email_is_not_linked(self): + request = self._login( + {"sub": "oidc-uid", "email": "victim@example.com", "email_verified": False} + ) + self._assert_not_linked(request) + + @override_settings(SOCIALACCOUNT_PROVIDERS=_oidc_providers()) + def test_missing_email_verified_claim_is_not_linked(self): + request = self._login({"sub": "oidc-uid", "email": "VICTIM@example.com"}) + self._assert_not_linked(request) + + @override_settings(SOCIALACCOUNT_PROVIDERS=_oidc_providers()) + def test_verified_email_is_linked(self): + request = self._login( + {"sub": "oidc-uid", "email": "victim@example.com", "email_verified": True} + ) + self._assert_linked(request) + + @override_settings(SOCIALACCOUNT_PROVIDERS=_oidc_providers(trust_email=True)) + def test_trusted_provider_links_without_email_verified_claim(self): + request = self._login({"sub": "oidc-uid", "email": "victim@example.com"}) + self._assert_linked(request) + + @override_settings(SOCIALACCOUNT_PROVIDERS=_oidc_providers(trust_email=True)) + def test_trusted_provider_links_explicitly_unverified_email(self): + request = self._login( + {"sub": "oidc-uid", "email": "victim@example.com", "email_verified": False} + ) + self._assert_linked(request)