import base64
import copy
import json
import os
import re
import secrets
import time
from abc import ABC, abstractmethod
from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence
from contextlib import contextmanager
from datetime import timedelta
from email.headerregistry import Address
from typing import TYPE_CHECKING, Any
from unittest import mock
from urllib.parse import parse_qs, urlencode, urlsplit

import jwt
import ldap
import orjson
import requests
import responses
import time_machine
from bs4 import BeautifulSoup
from bs4.element import Tag
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from django.conf import settings
from django.contrib.auth import authenticate
from django.core import mail
from django.http import HttpRequest
from django.test import override_settings
from django.urls import reverse
from django.utils.timezone import now as timezone_now
from django_auth_ldap.backend import LDAPSearch, _LDAPUser
from jwt.exceptions import PyJWTError
from onelogin.saml2.auth import OneLogin_Saml2_Auth
from onelogin.saml2.logout_request import OneLogin_Saml2_Logout_Request
from onelogin.saml2.logout_response import OneLogin_Saml2_Logout_Response
from onelogin.saml2.response import OneLogin_Saml2_Response
from onelogin.saml2.utils import OneLogin_Saml2_Utils
from social_core.backends.github import GithubOrganizationOAuth2, GithubTeamOAuth2
from social_core.backends.oauth import BaseOAuth2
from social_core.backends.saml import SAMLAuth
from social_core.exceptions import AuthFailed, AuthStateForbidden
from social_django.storage import BaseDjangoStorage
from social_django.strategy import DjangoStrategy
from typing_extensions import override

from confirmation.models import Confirmation, create_confirmation_link
from zerver.actions.create_realm import do_create_realm
from zerver.actions.create_user import do_create_user, do_reactivate_user
from zerver.actions.invites import do_invite_users
from zerver.actions.realm_settings import (
    do_deactivate_realm,
    do_reactivate_realm,
    do_set_realm_authentication_methods,
    do_set_realm_property,
)
from zerver.actions.user_groups import (
    bulk_add_members_to_user_groups,
    create_user_group_in_database,
)
from zerver.actions.user_settings import do_change_password, do_change_user_setting
from zerver.actions.users import change_user_is_active, do_deactivate_user
from zerver.lib.avatar import avatar_url
from zerver.lib.avatar_hash import user_avatar_path
from zerver.lib.dev_ldap_directory import generate_dev_ldap_dir
from zerver.lib.email_validation import (
    get_existing_user_errors,
    get_realm_email_validator,
    validate_email_is_valid,
)
from zerver.lib.exceptions import JsonableError, RateLimitedError
from zerver.lib.initial_password import initial_password
from zerver.lib.mobile_auth_otp import otp_decrypt_api_key
from zerver.lib.storage import static_path
from zerver.lib.stream_subscription import get_subscribed_stream_ids_for_user
from zerver.lib.streams import ensure_stream
from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.test_helpers import (
    HostRequestMock,
    create_s3_buckets,
    load_subdomain_token,
    ratelimit_rule,
    read_test_image_file,
    use_s3_backend,
)
from zerver.lib.thumbnail import DEFAULT_AVATAR_SIZE, MEDIUM_AVATAR_SIZE, resize_avatar
from zerver.lib.types import Validator
from zerver.lib.url_encoding import append_url_query_string
from zerver.lib.user_groups import (
    get_system_user_group_by_name,
    get_system_user_group_for_user,
    is_user_in_group,
)
from zerver.lib.users import get_users_for_api
from zerver.lib.utils import assert_is_not_none
from zerver.lib.validator import (
    check_bool,
    check_dict_only,
    check_int,
    check_list,
    check_none_or,
    check_string,
    validate_login_email,
)
from zerver.models import (
    CustomProfileField,
    CustomProfileFieldValue,
    MultiuseInvite,
    NamedUserGroup,
    PreregistrationUser,
    Realm,
    RealmDomain,
    Stream,
    UserProfile,
)
from zerver.models.groups import SystemGroups, UserGroupMembership
from zerver.models.realms import clear_supported_auth_backends_cache, get_realm
from zerver.models.users import ExternalAuthID, PasswordTooWeakError, get_user_by_delivery_email
from zerver.signals import JUST_CREATED_THRESHOLD
from zerver.views.auth import log_into_subdomain, maybe_send_to_registration
from zproject.backends import (
    AUTH_BACKEND_NAME_MAP,
    AppleAuthBackend,
    AuthFuncT,
    AzureADAuthBackend,
    DevAuthBackend,
    EmailAuthBackend,
    ExternalAuthDataDict,
    ExternalAuthMethod,
    ExternalAuthResult,
    GenericOpenIdConnectBackend,
    GitHubAuthBackend,
    GitLabAuthBackend,
    GoogleAuthBackend,
    NoMatchingLDAPUserError,
    OutsideLDAPDomainError,
    PopulateUserLDAPError,
    RateLimitedAuthenticationByUsername,
    SAMLAuthBackend,
    SAMLDocument,
    SocialAuthMixin,
    ZulipAuthMixin,
    ZulipDummyBackend,
    ZulipLDAPAuthBackend,
    ZulipLDAPConfigurationError,
    ZulipLDAPError,
    ZulipLDAPUser,
    ZulipLDAPUserPopulator,
    ZulipRemoteUserBackend,
    apple_auth_enabled,
    auth_enabled_helper,
    check_password_strength,
    dev_auth_enabled,
    email_belongs_to_ldap,
    get_external_method_dicts,
    github_auth_enabled,
    gitlab_auth_enabled,
    google_auth_enabled,
    password_auth_enabled,
    query_ldap,
    require_email_format_usernames,
    saml_auth_enabled,
    sync_user_from_ldap,
)
from zproject.config import get_from_file_if_exists

if TYPE_CHECKING:
    from django.test.client import _MonkeyPatchedWSGIResponse as TestHttpResponse

APPLE_ID_TOKEN_GENERATION_KEY = get_from_file_if_exists(
    "zerver/tests/fixtures/apple/token_gen_private_key"
)
EXAMPLE_JWK = get_from_file_if_exists("zerver/tests/fixtures/example_jwk")


class AuthBackendTest(ZulipTestCase):
    def get_email(self) -> str:
        return self.example_email("hamlet")

    def verify_backend(
        self,
        backend: Any,
        *,
        good_kwargs: dict[str, Any],
        bad_kwargs: dict[str, Any] | None = None,
    ) -> None:
        clear_supported_auth_backends_cache()
        user_profile = self.example_user("hamlet")

        # If bad_kwargs was specified, verify auth fails in that case
        if bad_kwargs is not None:
            self.assertIsNone(backend.authenticate(**bad_kwargs))

        # Verify auth works
        result = backend.authenticate(**good_kwargs)
        self.assertEqual(user_profile, result)

        # Verify auth fails with a deactivated user
        do_deactivate_user(user_profile, acting_user=None)
        result = backend.authenticate(**good_kwargs)

        # Just takes you back to the login page treating as
        # invalid auth; this is correct because the form will
        # provide the appropriate validation error for deactivated
        # account.
        self.assertIsNone(result)

        # Reactivate the user and verify auth works again
        do_reactivate_user(user_profile, acting_user=None)
        result = backend.authenticate(**good_kwargs)
        self.assertEqual(user_profile, result)

        # Verify auth fails with a deactivated realm
        do_deactivate_realm(
            user_profile.realm,
            acting_user=None,
            deactivation_reason="owner_request",
            email_owners=False,
        )
        result = backend.authenticate(**good_kwargs)

        self.assertIsNone(result)

        # Verify auth works again after reactivating the realm
        do_reactivate_realm(user_profile.realm)
        result = backend.authenticate(**good_kwargs)
        self.assertEqual(user_profile, result)

        # ZulipDummyBackend isn't a real backend so the remainder
        # doesn't make sense for it
        if isinstance(backend, ZulipDummyBackend):
            return

        # Verify auth fails if the auth backend is disabled on server
        with self.settings(AUTHENTICATION_BACKENDS=("zproject.backends.ZulipDummyBackend",)):
            clear_supported_auth_backends_cache()
            result = backend.authenticate(**good_kwargs)

            self.assertIsNone(result)
        clear_supported_auth_backends_cache()

        # Verify auth fails if the auth backend is disabled for the realm
        for backend_name in AUTH_BACKEND_NAME_MAP:
            if isinstance(backend, AUTH_BACKEND_NAME_MAP[backend_name]):
                break

        authentication_methods = user_profile.realm.authentication_methods_dict()
        authentication_methods[backend_name] = False

        do_set_realm_authentication_methods(
            user_profile.realm, authentication_methods, acting_user=None
        )

        if "realm" in good_kwargs:
            # Because this test is a little unfaithful to the ordering
            # (i.e. we fetched the realm object before this function
            # was called, when in fact it should be fetched after we
            # changed the allowed authentication methods), we need to
            # propagate the changes we just made to the actual realm
            # object in good_kwargs.
            good_kwargs["realm"] = user_profile.realm

        self.assertIsNone(result)

        authentication_methods[backend_name] = True
        do_set_realm_authentication_methods(
            user_profile.realm, authentication_methods, acting_user=None
        )

    def test_dummy_backend(self) -> None:
        realm = get_realm("zulip")
        username = self.get_email()
        self.verify_backend(
            ZulipDummyBackend(),
            good_kwargs=dict(username=username, realm=realm, use_dummy_backend=True),
            bad_kwargs=dict(username=username, realm=realm, use_dummy_backend=False),
        )

    def setup_subdomain(self, user_profile: UserProfile) -> None:
        realm = user_profile.realm
        realm.string_id = "zulip"
        realm.save()

    def test_email_auth_backend(self) -> None:
        username = self.get_email()
        user_profile = self.example_user("hamlet")
        password = "testpassword"
        user_profile.set_password(password)
        user_profile.save()

        with (
            mock.patch("zproject.backends.email_auth_enabled", return_value=False),
            mock.patch("zproject.backends.password_auth_enabled", return_value=True),
        ):
            return_data: dict[str, bool] = {}
            user = EmailAuthBackend().authenticate(
                request=mock.MagicMock(),
                username=user_profile.delivery_email,
                realm=get_realm("zulip"),
                password=password,
                return_data=return_data,
            )
            self.assertEqual(user, None)
            self.assertTrue(return_data["email_auth_disabled"])

        self.verify_backend(
            EmailAuthBackend(),
            good_kwargs=dict(
                request=mock.MagicMock(),
                password=password,
                username=username,
                realm=get_realm("zulip"),
                return_data={},
            ),
            bad_kwargs=dict(
                request=mock.MagicMock(),
                password=password,
                username=username,
                realm=get_realm("zephyr"),
                return_data={},
            ),
        )

    def test_email_auth_backend_empty_password(self) -> None:
        user_profile = self.example_user("hamlet")
        password = "testpassword"
        user_profile.set_password(password)
        user_profile.save()

        # First, verify authentication works with a nonempty
        # password so we know we've set up the test correctly.
        self.assertIsNotNone(
            EmailAuthBackend().authenticate(
                request=mock.MagicMock(),
                username=self.example_email("hamlet"),
                password=password,
                realm=get_realm("zulip"),
            )
        )

        # Now do the same test with the empty string as the password.
        password = ""
        with self.assertRaises(PasswordTooWeakError):
            # UserProfile.set_password protects against setting an empty password.
            user_profile.set_password(password)
        # We do want to force an empty password for this test, so we bypass the protection
        # by using Django's version of this method.
        super(UserProfile, user_profile).set_password(password)
        user_profile.save()
        self.assertIsNone(
            EmailAuthBackend().authenticate(
                request=mock.MagicMock(),
                username=self.example_email("hamlet"),
                password=password,
                realm=get_realm("zulip"),
            )
        )

    def test_email_auth_backend_disabled_password_auth(self) -> None:
        user_profile = self.example_user("hamlet")
        password = "testpassword"
        user_profile.set_password(password)
        user_profile.save()
        # Verify if a realm has password auth disabled, correct password is rejected
        with mock.patch("zproject.backends.password_auth_enabled", return_value=False):
            self.assertIsNone(
                EmailAuthBackend().authenticate(
                    request=mock.MagicMock(),
                    username=self.example_email("hamlet"),
                    password=password,
                    realm=get_realm("zulip"),
                )
            )

    def test_email_auth_backend_password_hasher_change(self) -> None:
        user_profile = self.example_user("hamlet")
        password = "a_password_of_22_chars"

        with self.settings(PASSWORD_HASHERS=("django.contrib.auth.hashers.MD5PasswordHasher",)):
            user_profile.set_password(password)
            user_profile.save()

        with (
            self.settings(
                PASSWORD_HASHERS=(
                    "django.contrib.auth.hashers.PBKDF2PasswordHasher",
                    "django.contrib.auth.hashers.MD5PasswordHasher",
                ),
                PASSWORD_MIN_LENGTH=30,
            ),
            self.assertLogs("zulip.auth.email", level="INFO"),
            self.assertRaises(JsonableError) as m,
        ):
            EmailAuthBackend().authenticate(
                request=mock.MagicMock(),
                username=self.example_email("hamlet"),
                password=password,
                realm=get_realm("zulip"),
            )
        self.assertEqual(str(m.exception), "You need to reset your password.")

    def test_login_preview(self) -> None:
        # Test preview=true displays organization login page
        # instead of redirecting to app
        self.login("iago")
        realm = get_realm("zulip")
        result = self.client_get("/login/", {"preview": "true"})
        self.assertEqual(result.status_code, 200)
        self.assert_in_response(realm.description, result)
        assert realm.name is not None
        self.assert_in_response(realm.name, result)
        self.assert_in_response("Log in to Zulip", result)

        data = dict(
            description=orjson.dumps("New realm description").decode(),
            name=orjson.dumps("New Zulip").decode(),
        )
        result = self.client_patch("/json/realm", data)
        self.assert_json_success(result)

        result = self.client_get("/login/", {"preview": "true"})
        self.assertEqual(result.status_code, 200)
        self.assert_in_response("New realm description", result)
        self.assert_in_response("New Zulip", result)

        result = self.client_get("/login/")
        self.assertEqual(result.status_code, 302)
        self.assertEqual(result["Location"], "http://zulip.testserver/")

    def test_invalid_login_on_self_hosting_management_subdomain(self) -> None:
        result = self.client_get("/login/", subdomain="selfhosting")
        self.assertEqual(result.status_code, 404)
        self.assert_in_response("No organization found", result)

    @override_settings(AUTHENTICATION_BACKENDS=("zproject.backends.ZulipDummyBackend",))
    def test_no_backend_enabled(self) -> None:
        result = self.client_get("/login/")
        self.assert_in_success_response(["No authentication backends are enabled"], result)

        result = self.client_get("/register/")
        self.assert_in_success_response(["No authentication backends are enabled"], result)

    @override_settings(AUTHENTICATION_BACKENDS=("zproject.backends.GoogleAuthBackend",))
    def test_any_backend_enabled(self) -> None:
        # testing to avoid false error messages.
        result = self.client_get("/login/")
        self.assert_not_in_success_response(["No authentication backends are enabled"], result)

        result = self.client_get("/register/")
        self.assert_not_in_success_response(["No authentication backends are enabled"], result)

    @override_settings(
        AUTHENTICATION_BACKENDS=("zproject.backends.ZulipLDAPAuthBackend",), LDAP_EMAIL_ATTR="mail"
    )
    def test_ldap_backend(self) -> None:
        self.init_default_ldap_database()
        user_profile = self.example_user("hamlet")
        email = user_profile.delivery_email
        password = self.ldap_password("hamlet")
        self.setup_subdomain(user_profile)

        username = self.get_email()
        backend = ZulipLDAPAuthBackend()
        orig_authenticate = backend.authenticate

        def wrapped_authenticate(*args: Any, **kwargs: Any) -> UserProfile | None:
            with self.artificial_transaction_savepoint():
                return orig_authenticate(*args, **kwargs)

        backend.authenticate = wrapped_authenticate  # type: ignore[method-assign]

        # Test LDAP auth fails when LDAP server rejects password
        self.assertIsNone(
            backend.authenticate(
                request=mock.MagicMock(),
                username=email,
                password="wrongpass",
                realm=get_realm("zulip"),
            )
        )

        self.verify_backend(
            backend,
            bad_kwargs=dict(
                request=mock.MagicMock(),
                username=username,
                password=password,
                realm=get_realm("zephyr"),
            ),
            good_kwargs=dict(
                request=mock.MagicMock(),
                username=username,
                password=password,
                realm=get_realm("zulip"),
            ),
        )

    def test_devauth_backend(self) -> None:
        self.verify_backend(
            DevAuthBackend(),
            good_kwargs=dict(dev_auth_username=self.get_email(), realm=get_realm("zulip")),
            bad_kwargs=dict(dev_auth_username=self.get_email(), realm=get_realm("zephyr")),
        )

    @override_settings(AUTHENTICATION_BACKENDS=("zproject.backends.ZulipRemoteUserBackend",))
    def test_remote_user_backend(self) -> None:
        username = self.get_email()
        self.verify_backend(
            ZulipRemoteUserBackend(),
            good_kwargs=dict(remote_user=username, realm=get_realm("zulip")),
            bad_kwargs=dict(remote_user=username, realm=get_realm("zephyr")),
        )

    @override_settings(AUTHENTICATION_BACKENDS=("zproject.backends.ZulipRemoteUserBackend",))
    def test_remote_user_backend_invalid_realm(self) -> None:
        username = self.get_email()
        self.verify_backend(
            ZulipRemoteUserBackend(),
            good_kwargs=dict(remote_user=username, realm=get_realm("zulip")),
            bad_kwargs=dict(remote_user=username, realm=get_realm("zephyr")),
        )

    @override_settings(AUTHENTICATION_BACKENDS=("zproject.backends.ZulipRemoteUserBackend",))
    @override_settings(SSO_APPEND_DOMAIN="zulip.com")
    def test_remote_user_backend_sso_append_domain(self) -> None:
        username = Address(addr_spec=self.get_email()).username
        self.verify_backend(
            ZulipRemoteUserBackend(),
            good_kwargs=dict(remote_user=username, realm=get_realm("zulip")),
            bad_kwargs=dict(remote_user=username, realm=get_realm("zephyr")),
        )


class RateLimitAuthenticationTests(ZulipTestCase):
    @override_settings(RATE_LIMITING_AUTHENTICATE=True)
    def do_test_auth_rate_limiting(
        self,
        attempt_authentication_func: Callable[[HttpRequest, str, str], UserProfile | None],
        username: str,
        correct_password: str,
        wrong_password: str,
        expected_user_profile: UserProfile,
    ) -> None:
        # We have to mock RateLimitedAuthenticationByUsername.key to avoid key collisions
        # if tests run in parallel.
        original_key_method = RateLimitedAuthenticationByUsername.key
        salt = secrets.token_hex(16)

        def _mock_key(self: RateLimitedAuthenticationByUsername) -> str:
            return f"{salt}:{original_key_method(self)}"

        def attempt_authentication(username: str, password: str) -> UserProfile | None:
            request = HttpRequest()
            request.session = mock.MagicMock()
            return attempt_authentication_func(request, username, password)

        with (
            mock.patch.object(RateLimitedAuthenticationByUsername, "key", new=_mock_key),
            ratelimit_rule(10, 2, domain="authenticate_by_username"),
        ):
            try:
                start_time = time.time()
                with mock.patch("time.time", return_value=start_time):
                    self.assertIsNone(attempt_authentication(username, wrong_password))
                    self.assertIsNone(attempt_authentication(username, wrong_password))
                    # 2 failed attempts is the limit, so the next ones should get blocked,
                    # even with the correct password.
                    with self.assertRaises(RateLimitedError):
                        attempt_authentication(username, correct_password)
                    with self.assertRaises(RateLimitedError):
                        attempt_authentication(username, wrong_password)

                # After enough time passes, more authentication attempts can be made:
                with mock.patch("time.time", return_value=start_time + 11.0):
                    self.assertIsNone(attempt_authentication(username, wrong_password))

                    # Correct password
                    self.assertEqual(
                        attempt_authentication(username, correct_password), expected_user_profile
                    )
                    # A correct login attempt should reset the rate limits for this user profile,
                    # so the next two attempts shouldn't get limited:
                    self.assertIsNone(attempt_authentication(username, wrong_password))
                    self.assertIsNone(attempt_authentication(username, wrong_password))
                    # But the third attempt goes over the limit:
                    with self.assertRaises(RateLimitedError):
                        attempt_authentication(username, wrong_password)

                    # Resetting the password also clears the rate-limit
                    do_change_password(expected_user_profile, correct_password)
                    self.assertIsNone(attempt_authentication(username, wrong_password))

            finally:
                # Clean up to avoid affecting other tests.
                RateLimitedAuthenticationByUsername(username).clear_history()

    def test_email_auth_backend_user_based_rate_limiting(self) -> None:
        user_profile = self.example_user("hamlet")
        password = "testpassword"
        user_profile.set_password(password)
        user_profile.save()

        def attempt_authentication(
            request: HttpRequest, username: str, password: str
        ) -> UserProfile | None:
            return EmailAuthBackend().authenticate(
                request=request,
                username=username,
                realm=get_realm("zulip"),
                password=password,
                return_data={},
            )

        self.do_test_auth_rate_limiting(
            attempt_authentication,
            user_profile.delivery_email,
            password,
            "wrong_password",
            user_profile,
        )

    @override_settings(
        AUTHENTICATION_BACKENDS=("zproject.backends.ZulipLDAPAuthBackend",), LDAP_EMAIL_ATTR="mail"
    )
    def test_ldap_backend_user_based_rate_limiting(self) -> None:
        self.init_default_ldap_database()
        user_profile = self.example_user("hamlet")
        password = self.ldap_password("hamlet")

        def attempt_authentication(
            request: HttpRequest, username: str, password: str
        ) -> UserProfile | None:
            return ZulipLDAPAuthBackend().authenticate(
                request=request,
                username=username,
                realm=get_realm("zulip"),
                password=password,
                return_data={},
            )

        self.do_test_auth_rate_limiting(
            attempt_authentication,
            user_profile.delivery_email,
            password,
            "wrong_password",
            user_profile,
        )

    @override_settings(
        AUTHENTICATION_BACKENDS=(
            "zproject.backends.EmailAuthBackend",
            "zproject.backends.ZulipLDAPAuthBackend",
        ),
        LDAP_EMAIL_ATTR="mail",
    )
    def test_email_and_ldap_backends_user_based_rate_limiting(self) -> None:
        self.init_default_ldap_database()
        user_profile = self.example_user("hamlet")
        ldap_password = self.ldap_password("hamlet")

        email_password = "email_password"
        user_profile.set_password(email_password)
        user_profile.save()

        def attempt_authentication(
            request: HttpRequest, username: str, password: str
        ) -> UserProfile | None:
            user = authenticate(
                request=request,
                username=username,
                realm=get_realm("zulip"),
                password=password,
                return_data={},
            )
            if user is not None:
                assert isinstance(user, UserProfile)
            return user

        self.do_test_auth_rate_limiting(
            attempt_authentication,
            user_profile.delivery_email,
            email_password,
            "wrong_password",
            user_profile,
        )
        self.do_test_auth_rate_limiting(
            attempt_authentication,
            user_profile.delivery_email,
            ldap_password,
            "wrong_password",
            user_profile,
        )


class CheckPasswordStrengthTest(ZulipTestCase):
    def test_check_password_strength(self) -> None:
        with self.settings(PASSWORD_MIN_LENGTH=0, PASSWORD_MIN_GUESSES=0):
            # Never allow empty password.
            self.assertFalse(check_password_strength(""))

        with self.settings(PASSWORD_MIN_LENGTH=6, PASSWORD_MIN_GUESSES=1000):
            self.assertFalse(check_password_strength(""))
            self.assertFalse(check_password_strength("short"))
            # Long enough, but too easy:
            self.assertFalse(check_password_strength("longer"))
            # Good password:
            self.assertTrue(check_password_strength("f657gdGGk9"))


class DesktopFlowTestingLib(ZulipTestCase):
    def verify_desktop_flow_app_page(self, response: "TestHttpResponse") -> None:
        self.assertEqual(response.status_code, 200)
        self.assertIn(b"<h1>Finish desktop login</h1>", response.content)

    def verify_desktop_flow_end_page(
        self, response: "TestHttpResponse", email: str, desktop_flow_otp: str
    ) -> None:
        self.assertEqual(response.status_code, 200)

        soup = BeautifulSoup(response.content, "html.parser")
        input = soup.find("input", value=True)
        assert isinstance(input, Tag)
        desktop_data = input["value"]
        assert isinstance(desktop_data, str)
        a = soup.find("a", href=True)
        assert isinstance(a, Tag)
        browser_url = a["href"]

        self.assertEqual(browser_url, "/login/")
        decrypted_key = self.verify_desktop_data_and_return_key(desktop_data, desktop_flow_otp)

        # Flush the session to have a clean slate - since up till now
        # we were simulating the part of the flow that happens in the browser.
        # Now we will simulating the last part of the flow, which gets executed
        # in the desktop application - thus with a separate session.
        self.client.session.flush()

        result = self.client_get(
            f"http://zulip.testserver/accounts/login/subdomain/{decrypted_key}"
        )
        self.assertEqual(result.status_code, 302)
        realm = get_realm("zulip")
        user_profile = get_user_by_delivery_email(email, realm)
        self.assert_logged_in_user_id(user_profile.id)

    def verify_desktop_data_and_return_key(self, desktop_data: str, desktop_flow_otp: str) -> str:
        key = bytes.fromhex(desktop_flow_otp)
        data = bytes.fromhex(desktop_data)
        iv = data[:12]
        ciphertext = data[12:]
        return AESGCM(key).decrypt(iv, ciphertext, b"").decode()


class SocialAuthBase(DesktopFlowTestingLib, ZulipTestCase, ABC):
    """This is a base class for testing social-auth backends. These
    methods are often overridden by subclasses:

        register_extra_endpoints() - If the backend being tested calls some extra
                                     endpoints then they can be added here.

        get_account_data_dict() - Return the data returned by the user info endpoint
                                  according to the respective backend.
    """

    BACKEND_CLASS: "type[SocialAuthMixin]"
    LOGIN_URL: str
    SIGNUP_URL: str
    AUTHORIZATION_URL: str
    AUTH_FINISH_URL: str
    ACCESS_TOKEN_URL: str
    USER_INFO_URL: str
    CLIENT_KEY_SETTING: str
    CLIENT_SECRET_SETTING: str

    @abstractmethod
    def get_account_data_dict(self, email: str, name: str) -> dict[str, Any]:
        raise NotImplementedError

    @override
    def setUp(self) -> None:
        super().setUp()
        self.user_profile = self.example_user("hamlet")
        self.email = self.user_profile.delivery_email
        self.name = self.user_profile.full_name
        self.backend = self.BACKEND_CLASS
        self.backend.strategy = DjangoStrategy(storage=BaseDjangoStorage())
        self.logger_string = f"zulip.auth.{self.backend.name}"

        # This is a workaround for the fact that Python social auth
        # caches the set of authentication backends that are enabled
        # the first time that `social_django.utils` is imported.  See
        # https://github.com/python-social-auth/social-app-django/pull/162
        # for details.
        from social_core.backends.utils import load_backends

        load_backends(settings.AUTHENTICATION_BACKENDS, force_load=True)

    def logger_output(self, output_string: str, type: str) -> str:
        return f"{type.upper()}:zulip.auth.{self.backend.name}:{output_string}"

    def register_extra_endpoints(
        self,
        requests_mock: responses.RequestsMock,
        account_data_dict: dict[str, str],
        **extra_data: Any,
    ) -> None:
        pass

    def prepare_login_url_and_headers(
        self,
        subdomain: str,
        mobile_flow_otp: str | None = None,
        desktop_flow_otp: str | None = None,
        is_signup: bool = False,
        next: str = "",
        multiuse_object_key: str = "",
        alternative_start_url: str | None = None,
        *,
        user_agent: str | None = None,
        extra_headers: dict[str, Any] | None = None,
    ) -> tuple[str, dict[str, Any]]:
        url = self.LOGIN_URL
        if alternative_start_url is not None:
            url = alternative_start_url

        params = {}
        headers = {}
        if subdomain == "":
            # "testserver" may trip up some libraries' URL validation,
            # so let's use the equivalent www. version.
            headers["HTTP_HOST"] = "www.testserver"
        else:
            headers["HTTP_HOST"] = subdomain + ".testserver"
        if mobile_flow_otp is not None:
            params["mobile_flow_otp"] = mobile_flow_otp
            headers["HTTP_USER_AGENT"] = "ZulipAndroid"
        if desktop_flow_otp is not None:
            params["desktop_flow_otp"] = desktop_flow_otp
        if is_signup:
            url = self.SIGNUP_URL
        params["next"] = next
        params["multiuse_object_key"] = multiuse_object_key
        if len(params) > 0:
            query = urlencode(params)
            url = append_url_query_string(url, query)

        if user_agent is not None:
            headers["HTTP_USER_AGENT"] = user_agent

        if extra_headers is not None:
            headers.update(extra_headers)

