fix(api): strengthen social account linking (#12013)

api/changelog.d/social-account-linking.security.md

@@ -0,0 +1 @@
+Social account linking now requires a verified matching email from both the identity provider and the existing user account

api/src/backend/api/adapters.py

@@ -1,3 +1,7 @@
+import logging
+
+from allauth.account.models import EmailAddress
+from allauth.core.exceptions import ImmediateHttpResponse
 from allauth.socialaccount.adapter import DefaultSocialAccountAdapter
 from api.db_router import MainRouter
 from api.db_utils import rls_transaction
@@ -11,6 +15,9 @@ from api.models import (
 )
 from api.utils import accept_invitation_for_user
 from django.db import transaction
+from django.http import HttpResponseForbidden
+
+logger = logging.getLogger(__name__)
 
 
 class ProwlerSocialAccountAdapter(DefaultSocialAccountAdapter):
@@ -38,8 +45,13 @@ class ProwlerSocialAccountAdapter(DefaultSocialAccountAdapter):
         return None
 
     def pre_social_login(self, request, sociallogin):
-        # Link existing accounts with the same email address
-        email = sociallogin.account.extra_data.get("email")
+        # The provider account is already bound, so no email-based linking is needed.
+        if sociallogin.account.pk:
+            return
+
+        # Prefer the normalized email populated by allauth. GitHub can return the
+        # primary email separately from the profile stored in extra_data.
+        email = sociallogin.user.email or sociallogin.account.extra_data.get("email")
         if sociallogin.provider.id == "saml":
             # For SAML, the asserted NameID email cannot be trusted on its own:
             # any tenant can claim any email domain in its SAML configuration. To
@@ -80,8 +92,25 @@ class ProwlerSocialAccountAdapter(DefaultSocialAccountAdapter):
         if email:
             existing_user = self.get_user_by_email(email)
             if existing_user:
+                email_is_verified = EmailAddress.objects.filter(
+                    user=existing_user,
+                    email__iexact=email,
+                    verified=True,
+                ).exists()
+                provider_verified_email = any(
+                    address.verified and address.email.casefold() == email.casefold()
+                    for address in sociallogin.email_addresses
+                )
+                if not email_is_verified or not provider_verified_email:
+                    raise ImmediateHttpResponse(HttpResponseForbidden())
                 sociallogin.connect(request, existing_user)
 
+    def send_notification_mail(self, *args, **kwargs):
+        try:
+            return super().send_notification_mail(*args, **kwargs)
+        except OSError:
+            logger.exception("Failed to send social account connection notification")
+
     def save_user(self, request, sociallogin, form=None):
         """
         Called after the user data is fully populated from the provider

api/src/backend/api/tests/test_adapters.py

@@ -2,11 +2,18 @@ from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
 import pytest
-from allauth.socialaccount.models import SocialLogin
+from allauth.account import app_settings as account_app_settings
+from allauth.account.models import EmailAddress
+from allauth.core import context
+from allauth.core.exceptions import ImmediateHttpResponse
+from allauth.socialaccount import app_settings as socialaccount_app_settings
+from allauth.socialaccount.internal.flows.login import complete_login
+from allauth.socialaccount.models import SocialAccount, SocialLogin
 from api.adapters import ProwlerSocialAccountAdapter
 from api.db_router import MainRouter
 from api.models import Invitation, Membership, SAMLConfiguration, Tenant
 from django.contrib.auth import get_user_model
+from django.core import mail
 
 User = get_user_model()
 
@@ -40,6 +47,7 @@ def _saml_request(rf, organization_slug):
 def _saml_sociallogin(user):
     sociallogin = MagicMock(spec=SocialLogin)
     sociallogin.account = MagicMock()
+    sociallogin.account.pk = None
     sociallogin.provider = MagicMock()
     sociallogin.provider.id = "saml"
     sociallogin.account.extra_data = {}
@@ -48,6 +56,59 @@ def _saml_sociallogin(user):
     return sociallogin
 
 
+def _oauth_sociallogin(
+    user,
+    *,
+    provider="google",
+    provider_email_verified=True,
+    include_extra_email=True,
+):
+    sociallogin = MagicMock(spec=SocialLogin)
+    sociallogin.account = MagicMock()
+    sociallogin.account.pk = None
+    sociallogin.provider = MagicMock()
+    sociallogin.provider.id = provider
+    sociallogin.account.extra_data = (
+        {"email": user.email} if include_extra_email else {}
+    )
+    sociallogin.email_addresses = [
+        EmailAddress(
+            email=user.email,
+            verified=provider_email_verified,
+            primary=True,
+        )
+    ]
+    sociallogin.user = user
+    sociallogin.connect = MagicMock()
+    return sociallogin
+
+
+def _real_oauth_sociallogin(user, uid):
+    provider = MagicMock()
+    provider.id = "google"
+    provider.app = None
+    provider.get_settings.return_value = {}
+    return SocialLogin(
+        user=user,
+        account=SocialAccount(
+            provider="google",
+            uid=uid,
+            extra_data={"email": user.email},
+        ),
+        email_addresses=[EmailAddress(email=user.email, verified=True, primary=True)],
+        provider=provider,
+    )
+
+
+def _verify_local_email(user):
+    return EmailAddress.objects.create(
+        user=user,
+        email=user.email,
+        verified=True,
+        primary=True,
+    )
+
+
 @pytest.mark.django_db
 class TestProwlerSocialAccountAdapter:
     def test_get_user_by_email_returns_user(self, create_test_user):
@@ -157,6 +218,7 @@ class TestProwlerSocialAccountAdapter:
 
         sociallogin = MagicMock(spec=SocialLogin)
         sociallogin.account = MagicMock()
+        sociallogin.account.pk = None
         sociallogin.provider = MagicMock()
         sociallogin.user = MagicMock()
         sociallogin.user.email = ""
@@ -168,25 +230,144 @@ class TestProwlerSocialAccountAdapter:
 
         sociallogin.connect.assert_not_called()
 
-    def test_pre_social_login_non_saml_links_by_email(self, create_test_user, rf):
-        """Non-SAML providers (e.g. Google/GitHub) still link to an existing
-        local account by email; the tenant binding only applies to SAML."""
+    def test_pre_social_login_blocks_unverified_local_email(self, create_test_user, rf):
+        """A verified OAuth email must not claim an unverified local account."""
         adapter = ProwlerSocialAccountAdapter()
+        sociallogin = _oauth_sociallogin(create_test_user)
 
-        sociallogin = MagicMock(spec=SocialLogin)
-        sociallogin.account = MagicMock()
-        sociallogin.provider = MagicMock()
-        sociallogin.provider.id = "google"
-        sociallogin.account.extra_data = {"email": create_test_user.email}
-        sociallogin.user = create_test_user
-        sociallogin.connect = MagicMock()
+        with pytest.raises(ImmediateHttpResponse) as exc_info:
+            adapter.pre_social_login(rf.get("/"), sociallogin)
+
+        assert exc_info.value.response.status_code == 403
+        sociallogin.connect.assert_not_called()
+
+    def test_complete_oauth_login_does_not_link_unverified_local_email(
+        self, create_test_user, rf
+    ):
+        """Regression test for the complete pre-hijack account-linking flow."""
+        incoming_user = User(email=create_test_user.email)
+        incoming_user.set_unusable_password()
+        sociallogin = _real_oauth_sociallogin(
+            incoming_user,
+            uid="victim-google-account",
+        )
+        request = rf.get("/")
+        request.session = {}
+
+        with pytest.raises(ImmediateHttpResponse) as exc_info:
+            complete_login(request, sociallogin, raises=True)
+
+        assert exc_info.value.response.status_code == 403
+        assert not SocialAccount.objects.filter(
+            provider="google", uid="victim-google-account"
+        ).exists()
+
+    def test_pre_social_login_allows_already_connected_account(
+        self, create_test_user, rf
+    ):
+        """Existing provider bindings do not need to relink on every login."""
+        adapter = ProwlerSocialAccountAdapter()
+        sociallogin = _oauth_sociallogin(create_test_user)
+        sociallogin.account.pk = "existing-social-account"
 
         adapter.pre_social_login(rf.get("/"), sociallogin)
 
-        call_args = sociallogin.connect.call_args
-        assert call_args is not None
-        _, called_user = call_args[0]
-        assert called_user.email == create_test_user.email
+        sociallogin.connect.assert_not_called()
+
+    def test_pre_social_login_blocks_unverified_provider_email(
+        self, create_test_user, rf
+    ):
+        """An OAuth provider must prove ownership of the matching email."""
+        _verify_local_email(create_test_user)
+        adapter = ProwlerSocialAccountAdapter()
+        sociallogin = _oauth_sociallogin(
+            create_test_user,
+            provider="github",
+            provider_email_verified=False,
+        )
+
+        with pytest.raises(ImmediateHttpResponse) as exc_info:
+            adapter.pre_social_login(rf.get("/"), sociallogin)
+
+        assert exc_info.value.response.status_code == 403
+        sociallogin.connect.assert_not_called()
+
+    def test_pre_social_login_links_verified_emails(self, create_test_user, rf):
+        _verify_local_email(create_test_user)
+        adapter = ProwlerSocialAccountAdapter()
+        sociallogin = _oauth_sociallogin(create_test_user)
+        request = rf.get("/")
+
+        adapter.pre_social_login(request, sociallogin)
+
+        sociallogin.connect.assert_called_once_with(request, create_test_user)
+
+    def test_verified_social_account_link_notifies_owner(self, create_test_user, rf):
+        _verify_local_email(create_test_user)
+        sociallogin = _real_oauth_sociallogin(
+            create_test_user,
+            uid="verified-google-account",
+        )
+
+        request = rf.get("/")
+        with context.request_context(request):
+            ProwlerSocialAccountAdapter().pre_social_login(request, sociallogin)
+
+        assert SocialAccount.objects.filter(
+            provider="google",
+            uid="verified-google-account",
+            user=create_test_user,
+        ).exists()
+        assert len(mail.outbox) == 1
+        assert mail.outbox[0].to == [create_test_user.email]
+
+    def test_notification_delivery_failure_does_not_break_verified_link(
+        self, create_test_user, rf
+    ):
+        _verify_local_email(create_test_user)
+        sociallogin = _real_oauth_sociallogin(
+            create_test_user,
+            uid="verified-google-account-without-smtp",
+        )
+        request = rf.get("/")
+
+        with (
+            context.request_context(request),
+            patch(
+                "allauth.socialaccount.adapter."
+                "DefaultSocialAccountAdapter.send_notification_mail",
+                side_effect=ConnectionRefusedError,
+            ),
+        ):
+            ProwlerSocialAccountAdapter().pre_social_login(request, sociallogin)
+
+        assert SocialAccount.objects.filter(
+            provider="google",
+            uid="verified-google-account-without-smtp",
+            user=create_test_user,
+        ).exists()
+
+    def test_pre_social_login_uses_verified_email_missing_from_extra_data(
+        self, create_test_user, rf
+    ):
+        """GitHub can return its verified primary email outside extra_data."""
+        _verify_local_email(create_test_user)
+        adapter = ProwlerSocialAccountAdapter()
+        sociallogin = _oauth_sociallogin(
+            create_test_user,
+            provider="github",
+            include_extra_email=False,
+        )
+        request = rf.get("/")
+
+        adapter.pre_social_login(request, sociallogin)
+
+        sociallogin.connect.assert_called_once_with(request, create_test_user)
+
+    def test_social_account_linking_settings_are_fail_closed(self):
+        assert not socialaccount_app_settings.EMAIL_AUTHENTICATION
+        assert not socialaccount_app_settings.EMAIL_AUTHENTICATION_AUTO_CONNECT
+        assert account_app_settings.EMAIL_NOTIFICATIONS
 
     def test_save_user_social_with_invitation_joins_invited_tenant(
         self, rf, create_test_user, tenants_fixture

api/src/backend/config/settings/social_login.py

@@ -13,16 +13,17 @@ GITHUB_OAUTH_CALLBACK_URL = env("SOCIAL_GITHUB_OAUTH_CALLBACK_URL", default="")
 ACCOUNT_LOGIN_METHODS = {"email"}  # Use Email / Password authentication
 ACCOUNT_SIGNUP_FIELDS = ["email*", "password1*", "password2*"]
 ACCOUNT_EMAIL_VERIFICATION = "none"  # Do not require email confirmation
+ACCOUNT_EMAIL_NOTIFICATIONS = True
 ACCOUNT_USER_MODEL_USERNAME_FIELD = None
 REST_AUTH = {
     "TOKEN_MODEL": None,
     "REST_USE_JWT": True,
 }
 # django-allauth (social)
-# Authenticate if local account with this email address already exists
-SOCIALACCOUNT_EMAIL_AUTHENTICATION = True
-# Connect local account and social account if local account with that email address already exists
-SOCIALACCOUNT_EMAIL_AUTHENTICATION_AUTO_CONNECT = True
+# Email-based account matching is handled by ProwlerSocialAccountAdapter, which
+# verifies both the provider email and the existing account email before linking.
+SOCIALACCOUNT_EMAIL_AUTHENTICATION = False
+SOCIALACCOUNT_EMAIL_AUTHENTICATION_AUTO_CONNECT = False
 SOCIALACCOUNT_ADAPTER = "api.adapters.ProwlerSocialAccountAdapter"