# Documentation for Zulip's authentication backends is split across a few places:
#
# * https://zulip.readthedocs.io/en/latest/production/authentication-methods.html and
#   zproject/prod_settings_template.py have user-level configuration documentation.
# * https://zulip.readthedocs.io/en/latest/development/authentication.html
#   has developer-level documentation, especially on testing authentication backends
#   in the Zulip development environment.
#
# Django upstream's documentation for authentication backends is also
# helpful background.  The most important detail to understand for
# reading this file is that the Django authenticate() function will
# call the authenticate methods of all backends registered in
# settings.AUTHENTICATION_BACKENDS that have a function signature
# matching the args/kwargs passed in the authenticate() call.
import binascii
import json
import logging
from abc import ABC, abstractmethod
from collections.abc import Callable, Generator
from contextlib import contextmanager
from dataclasses import dataclass
from email.headerregistry import Address
from typing import Any, TypedDict, TypeVar, cast

import magic
import orjson
from decorator import decorator
from django.conf import settings
from django.contrib.auth import authenticate, get_backends, logout
from django.contrib.auth.backends import RemoteUserBackend
from django.contrib.staticfiles.storage import staticfiles_storage
from django.core.exceptions import ValidationError
from django.core.validators import validate_email
from django.db import transaction
from django.dispatch import Signal, receiver
from django.http import HttpRequest, HttpResponse, HttpResponseRedirect
from django.shortcuts import render
from django.urls import reverse
from django.utils.translation import gettext as _
from django_auth_ldap.backend import LDAPBackend, _LDAPUser, _LDAPUserGroups, ldap_error
from django_auth_ldap.config import LDAPSettings
from lxml.etree import XMLSyntaxError
from onelogin.saml2 import compat as onelogin_saml2_compat
from onelogin.saml2.auth import OneLogin_Saml2_Auth
from onelogin.saml2.errors import OneLogin_Saml2_Error, OneLogin_Saml2_ValidationError
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.settings import OneLogin_Saml2_Settings
from onelogin.saml2.utils import OneLogin_Saml2_Utils
from onelogin.saml2.xml_utils import OneLogin_Saml2_XML
from requests import HTTPError
from social_core.backends.apple import AppleIdAuth
from social_core.backends.azuread import AzureADOAuth2
from social_core.backends.base import BaseAuth
from social_core.backends.github import GithubOAuth2, GithubOrganizationOAuth2, GithubTeamOAuth2
from social_core.backends.gitlab import GitLabOAuth2
from social_core.backends.google import GoogleOAuth2
from social_core.backends.open_id_connect import OpenIdConnectAuth
from social_core.backends.saml import SAMLAuth, SAMLIdentityProvider
from social_core.exceptions import (
    AuthCanceled,
    AuthFailed,
    AuthMissingParameter,
    AuthStateForbidden,
    SocialAuthBaseException,
)
from social_core.pipeline.partial import partial
from social_core.storage import UserProtocol
from social_core.strategy import HttpResponseProtocol
from social_django.utils import load_backend, load_strategy
from typing_extensions import override
from zxcvbn import zxcvbn

from zerver.actions.create_user import do_create_user, do_reactivate_user
from zerver.actions.custom_profile_fields import do_update_user_custom_profile_data_if_changed
from zerver.actions.user_groups import (
    bulk_add_members_to_user_groups,
    bulk_remove_members_from_user_groups,
    check_add_user_group_core,
)
from zerver.actions.user_settings import do_change_user_delivery_email, do_regenerate_api_key
from zerver.actions.users import do_change_user_role, do_deactivate_user
from zerver.lib.avatar import avatar_url, is_avatar_new
from zerver.lib.avatar_hash import user_avatar_content_hash
from zerver.lib.dev_ldap_directory import init_fakeldap
from zerver.lib.email_validation import email_allowed_for_realm, validate_email_not_already_in_realm
from zerver.lib.exceptions import JsonableError
from zerver.lib.mobile_auth_otp import is_valid_otp
from zerver.lib.rate_limiter import RateLimitedObject, client_is_exempt_from_rate_limiting
from zerver.lib.redis_utils import get_dict_from_redis, get_redis_client, put_dict_in_redis
from zerver.lib.request import RequestNotes
from zerver.lib.sessions import delete_user_sessions
from zerver.lib.subdomains import get_subdomain
from zerver.lib.types import ProfileDataElementUpdateDict
from zerver.lib.user_groups import check_user_group_name, get_role_based_system_groups_dict
from zerver.lib.users import check_full_name, validate_user_custom_profile_field
from zerver.models import (
    CustomProfileField,
    NamedUserGroup,
    PreregistrationRealm,
    PreregistrationUser,
    Realm,
    UserGroupMembership,
    UserProfile,
)
from zerver.models.custom_profile_fields import custom_profile_fields_for_realm
from zerver.models.groups import SystemGroups
from zerver.models.realms import (
    DisposableEmailError,
    DomainNotAllowedForRealmError,
    EmailContainsPlusError,
    get_realm,
    supported_auth_backends,
)
from zerver.models.users import (
    ExternalAuthID,
    PasswordTooWeakError,
    get_user_by_delivery_email,
    get_user_profile_by_id,
    remote_user_to_email,
)
from zproject.settings_types import OIDCIdPConfigDict

redis_client = get_redis_client()


def all_default_backend_names() -> list[str]:
    if not settings.BILLING_ENABLED or settings.DEVELOPMENT:
        # If billing isn't enabled, it's a self-hosted server
        # and has access to all authentication backends.
        #
        # In DEVELOPMENT, we have BILLING_ENABLED=True, but
        # nonetheless we want to enable all backends by default
        # for convenience - we shouldn't add additional steps to the
        # process of setting up a backend for testing.
        return list(AUTH_BACKEND_NAME_MAP.keys())

    # By default, only enable backends that are available without requiring a plan.
    return [
        name
        for name, backend in AUTH_BACKEND_NAME_MAP.items()
        if backend.available_for_cloud_plans is None
    ]


# This first batch of methods is used by other code in Zulip to check
# whether a given authentication backend is enabled for a given realm.
# In each case, we both needs to check at the server level (via
# `settings.AUTHENTICATION_BACKENDS`, queried via
# `django.contrib.auth.get_backends`) and at the realm level (via the
# `RealmAuthenticationMethod` table).
def pad_method_dict(method_dict: dict[str, bool]) -> dict[str, bool]:
    """Pads an authentication methods dict to contain all auth backends
    supported by the software, regardless of whether they are
    configured on this server"""
    for key in AUTH_BACKEND_NAME_MAP:
        if key not in method_dict:
            method_dict[key] = False
    return method_dict


def auth_enabled_helper(
    backends_to_check: list[str],
    realm: Realm | None,
    realm_authentication_methods: dict[str, bool] | None = None,
) -> bool:
    """
    realm_authentication_methods can be passed if already fetched to avoid
    a database query.
    """
    if realm is not None:
        if realm_authentication_methods is not None:
            # Copy the dict to avoid mutating the original if it was passed in as argument.
            enabled_method_dict = realm_authentication_methods.copy()
        else:
            enabled_method_dict = realm.authentication_methods_dict()
    else:
        enabled_method_dict = dict.fromkeys(AUTH_BACKEND_NAME_MAP, True)

    pad_method_dict(enabled_method_dict)
    for supported_backend in supported_auth_backends():
        for backend_name in backends_to_check:
            backend = AUTH_BACKEND_NAME_MAP[backend_name]
            if enabled_method_dict[backend_name] and isinstance(supported_backend, backend):
                return True
    return False


def ldap_auth_enabled(
    realm: Realm | None = None, realm_authentication_methods: dict[str, bool] | None = None
) -> bool:
    return auth_enabled_helper(["LDAP"], realm, realm_authentication_methods)


def email_auth_enabled(
    realm: Realm | None = None, realm_authentication_methods: dict[str, bool] | None = None
) -> bool:
    return auth_enabled_helper(["Email"], realm, realm_authentication_methods)


def password_auth_enabled(
    realm: Realm | None = None, realm_authentication_methods: dict[str, bool] | None = None
) -> bool:
    return ldap_auth_enabled(realm, realm_authentication_methods) or email_auth_enabled(
        realm, realm_authentication_methods
    )


def dev_auth_enabled(
    realm: Realm | None = None, realm_authentication_methods: dict[str, bool] | None = None
) -> bool:
    return auth_enabled_helper(["Dev"], realm, realm_authentication_methods)


def google_auth_enabled(
    realm: Realm | None = None, realm_authentication_methods: dict[str, bool] | None = None
) -> bool:
    return auth_enabled_helper(["Google"], realm, realm_authentication_methods)


def github_auth_enabled(
    realm: Realm | None = None, realm_authentication_methods: dict[str, bool] | None = None
) -> bool:
    return auth_enabled_helper(["GitHub"], realm, realm_authentication_methods)


def gitlab_auth_enabled(
    realm: Realm | None = None, realm_authentication_methods: dict[str, bool] | None = None
) -> bool:
    return auth_enabled_helper(["GitLab"], realm, realm_authentication_methods)


def apple_auth_enabled(
    realm: Realm | None = None, realm_authentication_methods: dict[str, bool] | None = None
) -> bool:
    return auth_enabled_helper(["Apple"], realm, realm_authentication_methods)


def saml_auth_enabled(
    realm: Realm | None = None, realm_authentication_methods: dict[str, bool] | None = None
) -> bool:
    return auth_enabled_helper(["SAML"], realm, realm_authentication_methods)


def require_email_format_usernames(realm: Realm | None = None) -> bool:
    if ldap_auth_enabled(realm) and (settings.LDAP_EMAIL_ATTR or settings.LDAP_APPEND_DOMAIN):
        return False
    return True


def is_user_active(user_profile: UserProfile, return_data: dict[str, Any] | None = None) -> bool:
    if user_profile.realm.deactivated:
        if return_data is not None:
            return_data["inactive_realm"] = True
        return False
    if not user_profile.is_active:
        if return_data is not None:
            if user_profile.is_mirror_dummy:
                # Record whether it's a mirror dummy account
                return_data["is_mirror_dummy"] = True
            return_data["inactive_user"] = True
            return_data["inactive_user_id"] = user_profile.id
        return False

    return True


def common_get_active_user(
    email: str, realm: Realm, return_data: dict[str, Any] | None = None
) -> UserProfile | None:
    """This is the core common function used by essentially all
    authentication backends to check if there's an active user account
    with a given email address in the organization, handling both
    user-level and realm-level deactivation correctly.
    """
    try:
        user_profile = get_user_by_delivery_email(email, realm)
    except UserProfile.DoesNotExist:
        # If the user doesn't have an account in the target realm, we
        # check whether they might have an account in another realm,
        # and if so, provide a helpful error message via
        # `invalid_subdomain`.
        if not UserProfile.objects.filter(delivery_email__iexact=email).exists():
            return None
        if return_data is not None:
            return_data["invalid_subdomain"] = True
            return_data["matching_user_ids_in_different_realms"] = list(
                UserProfile.objects.filter(delivery_email__iexact=email).values("realm_id", "id")
            )
        return None
    if not is_user_active(user_profile, return_data):
        return None

    return user_profile


def is_subdomain_in_allowed_subdomains_list(subdomain: str, allowed_subdomains: list[str]) -> bool:
    if subdomain in allowed_subdomains:
        return True

    # The root subdomain is a special case, as sending an
    # empty string in the list of values of the attribute may
    # not be viable. So, any of the ROOT_SUBDOMAIN_ALIASES can
    # be used to signify the user is authorized for the root
    # subdomain.
    if (
        subdomain == Realm.SUBDOMAIN_FOR_ROOT_DOMAIN
        and not settings.ROOT_DOMAIN_LANDING_PAGE
        and any(alias in allowed_subdomains for alias in settings.ROOT_SUBDOMAIN_ALIASES)
    ):
        return True

    return False


AuthFuncT = TypeVar("AuthFuncT", bound=Callable[..., UserProfile | None])


class RateLimitedAuthenticationByUsername(RateLimitedObject):
    def __init__(self, username: str) -> None:
        self.username = username
        super().__init__()

    @override
    def key(self) -> str:
        return f"{type(self).__name__}:{self.username}"

    @override
    def rules(self) -> list[tuple[int, int]]:
        return settings.RATE_LIMITING_RULES["authenticate_by_username"]


def rate_limit_authentication_by_username(request: HttpRequest, username: str) -> None:
    RateLimitedAuthenticationByUsername(username).rate_limit_request(request)


def auth_rate_limiting_already_applied(request: HttpRequest) -> bool:
    request_notes = RequestNotes.get_notes(request)

    return any(
        isinstance(r.entity, RateLimitedAuthenticationByUsername)
        for r in request_notes.ratelimits_applied
    )


# Django's authentication mechanism uses introspection on the various authenticate() functions
# defined by backends, so we need a decorator that doesn't break function signatures.
# @decorator does this for us.
# The usual @wraps from functools breaks signatures, so it can't be used here.
@decorator
def custom_auth_decorator(auth_func: AuthFuncT, *args: Any, **kwargs: Any) -> UserProfile | None:
    custom_auth_wrapper_func = settings.CUSTOM_AUTHENTICATION_WRAPPER_FUNCTION
    if custom_auth_wrapper_func is None:
        return auth_func(*args, **kwargs)
    else:
        return custom_auth_wrapper_func(auth_func, *args, **kwargs)


@decorator
def rate_limit_auth(auth_func: AuthFuncT, *args: Any, **kwargs: Any) -> UserProfile | None:
    if not settings.RATE_LIMITING_AUTHENTICATE:
        return auth_func(*args, **kwargs)

    request = args[1]
    username = kwargs["username"]
    if RequestNotes.get_notes(request).client is None or not client_is_exempt_from_rate_limiting(
        request
    ):
        # Django cycles through enabled authentication backends until one succeeds,
        # or all of them fail. If multiple backends are tried like this, we only want
        # to execute rate_limit_authentication_* once, on the first attempt:
        if auth_rate_limiting_already_applied(request):
            pass
        else:
            # Apply rate limiting. If this request is above the limit,
            # RateLimitedError will be raised, interrupting the authentication process.
            # From there, the code calling authenticate() can either catch the exception
            # and handle it on its own, or it will be processed by RateLimitMiddleware.
            rate_limit_authentication_by_username(request, username)

    result = auth_func(*args, **kwargs)
    if result is not None:
        # Authentication succeeded, clear the rate-limiting record.
        RateLimitedAuthenticationByUsername(username).clear_history()

    return result


@decorator
def log_auth_attempts(auth_func: AuthFuncT, *args: Any, **kwargs: Any) -> UserProfile | None:
    result = auth_func(*args, **kwargs)

    backend_instance = args[0]
    request = args[1]
    username = kwargs["username"]
    realm = kwargs["realm"]
    return_data = kwargs["return_data"]

    log_auth_attempt(
        backend_instance.logger,
        request,
        realm,
        username,
        succeeded=result is not None,
        return_data=return_data,
    )

    return result


def log_auth_attempt(
    logger: logging.Logger,
    request: HttpRequest,
    realm: Realm,
    username: str,
    succeeded: bool,
    return_data: dict[str, Any],
) -> None:
    ip_addr = request.META.get("REMOTE_ADDR")
    outcome = "success" if succeeded else "failed"
    logger.info(
        "Authentication attempt from %s: subdomain=%s;username=%s;outcome=%s;return_data=%s",
        ip_addr,
        realm.subdomain,
        username,
        outcome,
        return_data,
    )


class ZulipAuthMixin:
    """This common mixin is used to override Django's default behavior for
    looking up a logged-in user by ID to use a version that fetches
    from memcached before checking the database (avoiding a database
    query in most cases).
    """

    name = "undefined"
    _logger: logging.Logger | None = None

    # Describes which plans gives access to this authentication method on zulipchat.com.
    # None means the backend is available regardless of the plan.
    # Otherwise, it should be a list of Realm.plan_type values that give access to the backend.
    available_for_cloud_plans: list[int] | None = None

    @property
    def logger(self) -> logging.Logger:
        if self._logger is None:
            self._logger = logging.getLogger(f"zulip.auth.{self.name}")
        return self._logger

    def get_user(self, user_profile_id: int) -> UserProfile | None:
        """Override the Django method for getting a UserProfile object from
        the user_profile_id,."""
        try:
            return get_user_profile_by_id(user_profile_id)
        except UserProfile.DoesNotExist:
            return None


class ZulipDummyBackend(ZulipAuthMixin):
    """Used when we want to log you in without checking any
    authentication (i.e. new user registration or when otherwise
    authentication has already been checked earlier in the process).

    We ensure that this backend only ever successfully authenticates
    when explicitly requested by including the use_dummy_backend kwarg.
    """

    name = "dummy"

    @custom_auth_decorator
    def authenticate(
        self,
        request: HttpRequest | None = None,
        *,
        username: str,
        realm: Realm,
        use_dummy_backend: bool = False,
        return_data: dict[str, Any] | None = None,
    ) -> UserProfile | None:
        if use_dummy_backend:
            return common_get_active_user(username, realm, return_data)
        return None


def check_password_strength(password: str) -> bool:
    """
    Returns True if the password is strong enough,
    False otherwise.
    """
    if len(password) < settings.PASSWORD_MIN_LENGTH:
        return False

    if password == "":
        # zxcvbn throws an exception when passed the empty string, so
        # we need a special case for the empty string password here.
        return False

    if (
        int(zxcvbn(password, max_length=settings.PASSWORD_MAX_LENGTH)["guesses"])
        < settings.PASSWORD_MIN_GUESSES
    ):
        return False

    return True


class EmailAuthBackend(ZulipAuthMixin):
    """
    Email+Password authentication backend (the default).

    Allows a user to sign in using an email/password pair.
    """

    name = "email"

    @rate_limit_auth
    @log_auth_attempts
    @custom_auth_decorator
    def authenticate(
        self,
        request: HttpRequest,
        *,
        username: str,
        password: str,
        realm: Realm,
        return_data: dict[str, Any] | None = None,
    ) -> UserProfile | None:
        """Authenticate a user based on email address as the user name."""
        if not password_auth_enabled(realm):
            if return_data is not None:
                return_data["password_auth_disabled"] = True
            return None
        if not email_auth_enabled(realm):
            if return_data is not None:
                return_data["email_auth_disabled"] = True
            return None
        if password == "":
            # Never allow an empty password.  This is defensive code;
            # a user having password "" should only be possible
            # through a bug somewhere else.
            return None

        user_profile = common_get_active_user(username, realm, return_data=return_data)
        if user_profile is None:
            return None

        try:
            is_password_correct = user_profile.check_password(password)
        except PasswordTooWeakError:
            # In some rare cases when password hasher is changed and the user has
            # a weak password, PasswordTooWeakError will be raised.
            self.logger.info(
                "User %s password can't be rehashed due to being too weak.", user_profile.id
            )
            if return_data is not None:
                return_data["password_reset_needed"] = True
                return None
            else:
                # Since we can't communicate the situation via return_data,
                # we have to raise an error - a silent failure would not be right
                # because the password actually is correct, just can't be re-hashed.
                raise JsonableError(_("You need to reset your password."))

        if is_password_correct:
            return user_profile
        return None


def is_valid_email(email: str) -> bool:
    try:
        validate_email(email)
    except ValidationError:
        return False
    return True


def check_ldap_config() -> None:
    if not settings.LDAP_APPEND_DOMAIN:
        # Email search needs to be configured in this case.
        assert settings.AUTH_LDAP_USERNAME_ATTR and settings.AUTH_LDAP_REVERSE_EMAIL_SEARCH

    # These two are alternatives approaches to deactivating users based on an ldap attribute
    # and thus don't make sense to have enabled together.
    assert not (
        settings.AUTH_LDAP_USER_ATTR_MAP.get("userAccountControl")
        and settings.AUTH_LDAP_USER_ATTR_MAP.get("deactivated")
    )


def ldap_should_sync_active_status() -> bool:
    if "userAccountControl" in settings.AUTH_LDAP_USER_ATTR_MAP:
        return True

    if "deactivated" in settings.AUTH_LDAP_USER_ATTR_MAP:
        return True

    return False


def find_ldap_users_by_email(email: str) -> list[_LDAPUser]:
    """
    Returns list of _LDAPUsers matching the email search
    """
    return LDAPReverseEmailSearch().search_for_users(email)


def email_belongs_to_ldap(realm: Realm, email: str) -> bool:
    """Used to make determinations on whether a user's email address is
    managed by LDAP.  For environments using both LDAP and
    Email+Password authentication, we do not allow EmailAuthBackend
    authentication for email addresses managed by LDAP (to avoid a
    security issue where one create separate credentials for an LDAP
    user), and this function is used to enforce that rule.
    """
    if not ldap_auth_enabled(realm):
        return False

    check_ldap_config()
    if settings.LDAP_APPEND_DOMAIN:
        # Check if the email ends with LDAP_APPEND_DOMAIN
        return Address(addr_spec=email).domain.lower() == settings.LDAP_APPEND_DOMAIN

    # If we don't have an LDAP domain, we have to do a lookup for the email.
    if find_ldap_users_by_email(email):
        return True
    else:
        return False


ldap_logger = logging.getLogger("zulip.ldap")


class LDAPReverseEmailSearch(_LDAPUser):
    """
    This class is a workaround - we want to use
    django-auth-ldap to query the ldap directory for
    users with the specified email address, but it doesn't
    provide an API for that or an isolated class for handling
    the connection. Because connection-handling is tightly integrated
    into the _LDAPUser class, we have to make this strange inheritance here,
    in order to be able to comfortably have an ldap connection and make search
    queries.

    We may be able to get rid of this in the future if we can get
    https://github.com/django-auth-ldap/django-auth-ldap/pull/150 merged upstream.
    """

    def __init__(self) -> None:
        # Superclass __init__ requires a username argument - it doesn't actually
        # impact anything for us in this class, given its very limited use
        # for only making a search query, so we pass an empty string.
        super().__init__(LDAPBackend(), username="")

    def search_for_users(self, email: str) -> list[_LDAPUser]:
        search = settings.AUTH_LDAP_REVERSE_EMAIL_SEARCH
        USERNAME_ATTR = settings.AUTH_LDAP_USERNAME_ATTR

        assert search is not None
        results = search.execute(self.connection, {"email": email})

        ldap_users = []
        for result in results:
            user_dn, user_attrs = result
            username = user_attrs[USERNAME_ATTR][0]
            ldap_user = _LDAPUser(self.backend, username=username)
            ldap_user._user_dn = user_dn
            ldap_user._user_attrs = user_attrs

            ldap_users.append(ldap_user)

        return ldap_users


class ZulipLDAPError(_LDAPUser.AuthenticationFailed):
    """Since this inherits from _LDAPUser.AuthenticationFailed, these will
    be caught and logged at debug level inside django-auth-ldap's authenticate()"""


class NoMatchingLDAPUserError(ZulipLDAPError):
    pass


class OutsideLDAPDomainError(NoMatchingLDAPUserError):
    pass


class ZulipLDAPConfigurationError(Exception):
    pass


LDAP_USER_ACCOUNT_CONTROL_DISABLED_MASK = 2


def ldap_external_auth_id_sync_enabled() -> bool:
    external_auth_id_attr = settings.AUTH_LDAP_USER_ATTR_MAP.get("unique_account_id")
    if external_auth_id_attr is None:
        return False

    if not isinstance(external_auth_id_attr, str):
        raise AssertionError(
            "unique_account_id in AUTH_LDAP_USER_ATTR_MAP set to invalid value! must be a string."
        )

    return True


class ZulipLDAPSettings(LDAPSettings):
    NON_SYNCABLE_ATTRS = [
        "unique_account_id",
        "avatar",
        "userAccountControl",
        "deactivated",
        "org_membership",
    ]

    def __init__(
        self, prefix: str = "AUTH_LDAP_", defaults: dict[object, object] | None = None
    ) -> None:
        """
        django-auth-ldap populate_user() codepath iterates through USER_ATTR_MAP dict, and calls
        setattr(user, key, value) for each key, value pair. The challenge here is that Zulip adds some custom
        mappings in AUTH_LDAP_USER_ATTR_MAP - for which the keys are special keywords (e.g. "unique_account_id")
        with a meaning understood by Zulip's code for handling them - and aren't actual UserProfile attributes
        that can be set in a valid way.

        To avoid ending up with strange UserProfile objects, which have attributes set on them which shouldn't
        exist at all, we have to patch this settings class - to drop such special keywords from USER_ATTR_MAP
        for the purposes of django-auth-ldap's internal code.
        """
        if defaults is None:  # nocoverage
            defaults = dict()

        super().__init__(prefix, defaults)
        self.USER_ATTR_MAP: dict[object, object] = {
            key: value
            for key, value in self.USER_ATTR_MAP.items()
            if key not in self.NON_SYNCABLE_ATTRS
        }


class ZulipLDAPAuthBackendBase(ZulipAuthMixin, LDAPBackend):
    """Common code between LDAP authentication (ZulipLDAPAuthBackend) and
    using LDAP just to sync user data (ZulipLDAPUserPopulator).

    To fully understand our LDAP backend, you may want to skim
    django_auth_ldap/backend.py from the upstream django-auth-ldap
    library.  It's not a lot of code, and searching around in that
    file makes the flow for LDAP authentication clear.
    """

    name = "ldap"

    def __init__(self) -> None:
        # Used to initialize a fake LDAP directly for both manual
        # and automated testing in a development environment where
        # there is no actual LDAP server.
        if settings.DEVELOPMENT and settings.FAKE_LDAP_MODE:  # nocoverage
            init_fakeldap()

        check_ldap_config()

    @property
    def settings(self) -> ZulipLDAPSettings:
        # See ZulipLDAPSettings for explanation of why we need to patch this
        # to substitute our overridden LDAPSettings class.
        self._settings: ZulipLDAPSettings | None
        if self._settings is None:
            self._settings = ZulipLDAPSettings(self.settings_prefix, self.default_settings)

        return self._settings

    # Disable django-auth-ldap's permissions functions -- we don't use
    # the standard Django user/group permissions system because they
    # are prone to performance issues.
    def has_perm(self, user: UserProfile | None, perm: Any, obj: Any = None) -> bool:
        return False

    def has_module_perms(self, user: UserProfile | None, app_label: str | None) -> bool:
        return False

    def get_all_permissions(self, user: UserProfile | None, obj: Any = None) -> set[Any]:
        return set()

    def get_group_permissions(self, user: UserProfile | None, obj: Any = None) -> set[Any]:
        return set()

    def django_to_ldap_username(self, username: str) -> str:
        """
        Translates django username (user_profile.delivery_email or whatever the user typed in the login
        field when authenticating via the LDAP backend) into LDAP username.
        Guarantees that the username it returns actually has an entry in the LDAP directory.
        Raises NoMatchingLDAPUserError if that's not possible.
        """
        result = username
        if settings.LDAP_APPEND_DOMAIN:
            if is_valid_email(username):
                address = Address(addr_spec=username)
                if address.domain != settings.LDAP_APPEND_DOMAIN:
                    raise OutsideLDAPDomainError(
                        f"Email {username} does not match LDAP domain {settings.LDAP_APPEND_DOMAIN}."
                    )
                result = address.username
        else:
            # We can use find_ldap_users_by_email
            if is_valid_email(username):
                email_search_result = find_ldap_users_by_email(username)
                if not email_search_result:
                    result = username
                elif len(email_search_result) == 1:
                    return email_search_result[0]._username
                elif len(email_search_result) > 1:
                    # This is possible, but strange, so worth logging a warning about.
                    # We can't translate the email to a unique username,
                    # so we don't do anything else here.
                    logging.warning("Multiple users with email %s found in LDAP.", username)
                    result = username

        if _LDAPUser(self, result).attrs is None:
            # Check that there actually is an LDAP entry matching the result username
            # we want to return. Otherwise, raise an exception.
            error_message = (
                "No LDAP user matching django_to_ldap_username result: {}. Input username: {}"
            )
            raise NoMatchingLDAPUserError(
                error_message.format(result, username),
            )

        return result

    def user_email_from_ldapuser(self, username: str, ldap_user: _LDAPUser) -> str:
        if hasattr(ldap_user, "_username"):
            # In tests, we sometimes pass a simplified _LDAPUser without _username attr,
            # and with the intended username in the username argument.
            username = ldap_user._username

        if settings.LDAP_APPEND_DOMAIN:
            return Address(username=username, domain=settings.LDAP_APPEND_DOMAIN).addr_spec

        if settings.LDAP_EMAIL_ATTR is not None:
            # Get email from LDAP attributes.
            if settings.LDAP_EMAIL_ATTR not in ldap_user.attrs:
                raise ZulipLDAPError(
                    f"LDAP user doesn't have the needed {settings.LDAP_EMAIL_ATTR} attribute"
                )
            else:
                return ldap_user.attrs[settings.LDAP_EMAIL_ATTR][0]

        return username

    def ldap_to_django_username(self, username: str) -> str:
        """
        This is called inside django_auth_ldap with only one role:
        to convert _LDAPUser._username to django username (so in Zulip, the email)
        and pass that as "username" argument to get_or_build_user(username, ldapuser).
        In many cases, the email is stored in the _LDAPUser's attributes, so it can't be
        constructed just from the username. We choose to do nothing in this function,
