diff --git a/api/changelog.d/social-account-linking.security.md b/api/changelog.d/social-account-linking.security.md new file mode 100644 index 0000000000..ab468dbefb --- /dev/null +++ b/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 diff --git a/api/src/backend/api/adapters.py b/api/src/backend/api/adapters.py index fa0abacb20..d8c5b1b386 100644 --- a/api/src/backend/api/adapters.py +++ b/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 diff --git a/api/src/backend/api/tests/test_adapters.py b/api/src/backend/api/tests/test_adapters.py index 1df9908ee5..70cd055205 100644 --- a/api/src/backend/api/tests/test_adapters.py +++ b/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 diff --git a/api/src/backend/config/settings/social_login.py b/api/src/backend/config/settings/social_login.py index d6b5b53df2..9ab3e89d73 100644 --- a/api/src/backend/config/settings/social_login.py +++ b/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"