Warning

This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.managers.users

"""
Manager and Serializer for Users.
"""

import hashlib
import logging
import random
import string
import time
from collections.abc import Callable
from typing import (
    Any,
    Protocol,
    TYPE_CHECKING,
)

from markupsafe import escape
from sqlalchemy import (
    and_,
    exc,
    false,
    select,
    true,
    update,
)
from sqlalchemy.exc import NoResultFound

from galaxy import (
    exceptions,
    model,
    schema,
    util,
)
from galaxy.config import (
    GALAXY_APP_NAME,
    templates,
)
from galaxy.managers import (
    base,
    deletable,
)
from galaxy.managers.base import combine_lists
from galaxy.managers.context import (
    ProvidesAppContext,
    ProvidesHistoryContext,
    ProvidesUserContext,
)
from galaxy.managers.extra_preferences import (
    EXTRA_PREFERENCES_KEY,
    ExtraPreferencesManager,
)
from galaxy.model import (
    Job,
    User,
    UserAddress,
    UserQuotaUsage,
)
from galaxy.model.db.user import (
    _cleanup_nonprivate_user_roles,
    get_user_by_email,
    get_user_by_username,
    get_user_groups,
)
from galaxy.security.validate_user_input import (
    UserValidationContext,
    validate_email,
    validate_password,
    validate_preferred_object_store_id,
    validate_publicname,
)
from galaxy.structured_app import (
    BasicSharedApp,
    MinimalManagerApp,
)
from galaxy.util import now
from galaxy.util.hash_util import new_secure_hash_v2
from galaxy.util.user_input import (
    canonicalize_display_name,
    canonicalize_email,
    VALID_EMAIL_RE,
    validate_display_name_str,
)

if TYPE_CHECKING:
    from galaxy.webapps.base.webapp import GalaxyWebTransaction

log = logging.getLogger(__name__)

PASSWORD_RESET_TEMPLATE = """
To reset your %s password use the following link, which will expire %s.

%s

If you did not make this request, no action is necessary on your part, though
you may want to notify an administrator.

If you're having trouble using the link when clicking it from email client, you
can also copy and paste it into your browser.
"""
TXT_ACTIVATION_EMAIL_TEMPLATE_RELPATH = "mail/activation-email.txt"
HTML_ACTIVATION_EMAIL_TEMPLATE_RELPATH = "mail/activation-email.html"




class PasswordChangeContext(UserValidationContext, Protocol):
    """What setting a password needs from a transaction, in Galaxy or the tool shed."""

    @property
    def galaxy_session(self) -> Any: ...



    def log_event(self, message: str) -> None: ...






class UserManager(base.ModelManager, deletable.PurgableManagerMixin):
    foreign_key_name = "user"

    # TODO: there is quite a bit of functionality around the user (authentication, permissions, quotas, groups/roles)
    #   most of which it may be unneccessary to have here

    # TODO: incorp BaseAPIController.validate_in_users_and_groups
    # TODO: incorporate UsesFormDefinitionsMixin?


    def __init__(self, app: BasicSharedApp, app_type="galaxy"):
        self.model_class = app.model.User
        self.app_type = app_type
        super().__init__(app)




    def register(
        self, trans: "GalaxyWebTransaction", email=None, username=None, password=None, confirm=None, subscribe=False
    ):
        """
        Register a new user.
        """
        if not trans.app.config.allow_local_account_creation and not trans.user_is_admin:
            message = "User registration is disabled.  Please contact your local Galaxy administrator for an account."
            if trans.app.config.error_email_to is not None:
                message += f" Contact: {trans.app.config.error_email_to}"
            return None, message
        if not email or not username or not password or not confirm:
            return None, "Please provide email, username and password."

        email = util.restore_text(email).strip()
        username = util.restore_text(username).strip()
        # We could add a separate option here for enabling or disabling domain validation (DNS resolution test)
        validate_domain = trans.app.config.user_activation_on
        message = "\n".join(
            (
                validate_email(trans, email, validate_domain=validate_domain),
                validate_password(trans, password, confirm),
                validate_publicname(trans, username),
            )
        ).rstrip()

        if message:
            return None, message
        message, status = trans.app.auth_manager.check_registration_allowed(email, username, password, trans.request)
        if message and not trans.user_is_admin:
            return None, message
        if subscribe:
            message = self.send_subscription_email(email)
            if message:
                return None, message
        user = self.create(email=email, username=username, password=password, trans=trans, send_activation_email=True)
        return user, None




    def create(
        self,
        email=None,
        username=None,
        password=None,
        *,
        trans=None,
        trusted_email=False,
        send_activation_email=False,
        **kwargs,
    ):
        """
        Create a new user.

        The account is active unless email activation is enabled
        (``user_activation_on``). ``trusted_email`` activates the account
        regardless, for emails already verified by a trusted source such as an
        OIDC identity provider. When the account is created inactive and
        ``send_activation_email`` is set, an activation email is sent (requires
        ``trans``).
        """
        self._error_on_duplicate_email(email)
        user = self.model_class(email=email, username=username)
        if password:
            user.set_password_cleartext(password)
        else:
            user.set_random_password()
        user.active = trusted_email or not self.app.config.user_activation_on
        session = self.session()
        session.add(user)
        try:
            # Creating a private role will commit the session
            self.app.security_agent.create_user_role(user, self.app)
        except exc.IntegrityError as db_err:
            raise exceptions.Conflict(str(db_err))
        if send_activation_email and not user.active:
            self.send_activation_email(trans, email, username)
        return user




    def update_email(
        self,
        trans: ProvidesUserContext,
        user: User,
        new_email: str,
        *,
        commit: bool = True,
        send_activation_email: bool = True,
        asserted_by_identity_provider: bool = False,
    ) -> None:
        """
        Update a user's email address, keeping the private role in sync and honoring activation settings.
        Raises RequestParameterInvalidException on validation errors.

        Non-admins are held to the address policy below, unless the address comes from an external identity
        provider (``asserted_by_identity_provider``) rather than from the user.
        """
        new_email = canonicalize_email(new_email)
        if user.email == new_email:
            return
        user_chosen = not asserted_by_identity_provider and not trans.user_is_admin
        if user_chosen and self.logins_resolve_accounts_by_email():
            # Refused before validation, so that the error does not reveal whether the address is registered.
            raise exceptions.ConfigDoesNotAllowException("Email changes are not allowed in this Galaxy instance")
        if message := validate_email(trans, new_email, user):
            raise exceptions.RequestParameterInvalidException(message)
        if user_chosen and self._is_admin_email(new_email):
            # admin_users grants administrator rights by address alone. The message is the one for a
            # taken address, so that it does not reveal which addresses belong to administrators.
            raise exceptions.RequestParameterInvalidException(f"User with email '{new_email}' already exists.")
        private_role = trans.app.security_agent.get_private_user_role(user)
        assert private_role is not None
        private_role.name = new_email
        private_role.description = f"Private role for {new_email}"
        user.email = new_email
        session = self.session()
        session.add_all([user, private_role])
        if trans.app.config.user_activation_on:
            user.active = False
            # A token mailed to the previous address must not verify this one.
            user.activation_token = None
            if send_activation_email:
                if not self.send_activation_email(trans, user.email, user.username):
                    session.rollback()
                    error_message = "Unable to send activation email, please contact your local Galaxy administrator."
                    if trans.app.config.error_email_to is not None:
                        error_message += f" Contact: {trans.app.config.error_email_to}"
                    raise exceptions.InternalServerError(error_message)
        if commit:
            session.commit()




    def update_username(self, trans: ProvidesAppContext, user: User, new_username: str, *, commit: bool = True) -> None:
        """
        Update a user's public name after validating it. Raises RequestParameterInvalidException on validation errors.
        """
        if message := validate_publicname(trans, new_username, user):
            raise exceptions.RequestParameterInvalidException(message)
        if user.username == new_username:
            return
        user.username = new_username
        session = self.session()
        session.add(user)
        if commit:
            session.commit()




    def update_display_name(self, user: User, new_display_name: str | None, *, commit: bool = True) -> None:
        """
        Update a user's display name after validating it. Raises RequestParameterInvalidException on validation errors.

        Unlike the username this needs no transaction: display names are not unique, so there is nothing to look up.
        The name is stored in the form ``canonicalize_display_name`` returns, so a blank name clears the field.
        """
        normalized = canonicalize_display_name(new_display_name)
        if message := validate_display_name_str(normalized):
            raise exceptions.RequestParameterInvalidException(message)
        if user.display_name == normalized:
            return
        user.display_name = normalized
        session = self.session()
        session.add(user)
        if commit:
            session.commit()




    def delete(self, user, flush=True):
        """Mark the given user deleted."""
        if not self.app.config.allow_user_deletion:
            raise exceptions.ConfigDoesNotAllowException(
                "The configuration of this Galaxy instance does not allow admins to delete users."
            )
        super().delete(user, flush=flush)
        self._stop_all_jobs_from_user(user)


    def _stop_all_jobs_from_user(self, user):
        active_jobs = self._get_all_active_jobs_from_user(user)
        session = self.session()
        for job in active_jobs:
            job.mark_deleted(self.app.config.track_jobs_in_database)
        session.commit()

    def _get_all_active_jobs_from_user(self, user: User) -> list[Job]:
        """Get all jobs that are not ready yet and belong to the given user."""
        stmt = select(Job).where(and_(Job.user_id == user.id, Job.state.in_(Job.non_ready_states)))
        jobs = self.session().scalars(stmt)
        return jobs  # type: ignore[return-value]



    def undelete(self, user, flush=True):
        """Remove the deleted flag for the given user."""
        if not self.app.config.allow_user_deletion:
            raise exceptions.ConfigDoesNotAllowException(
                "The configuration of this Galaxy instance does not allow admins to undelete users."
            )
        if user.purged:
            raise exceptions.ItemDeletionException("Purged user cannot be undeleted.")
        super().undelete(user, flush=flush)




    def purge(self, user, flush=True):
        """Purge the given user. They must have the deleted flag already."""
        if not self.app.config.allow_user_deletion:
            raise exceptions.ConfigDoesNotAllowException(
                "The configuration of this Galaxy instance does not allow admins to delete or purge users."
            )
        if not user.deleted:
            raise exceptions.MessageException(f"User '{user.email}' has not been deleted, so they cannot be purged.")
        private_role = self.app.security_agent.get_private_user_role(user)
        if private_role is None:
            raise exceptions.InconsistentDatabase(
                f"User {user.email} private role is missing while attempting to purge deleted user."
            )
        # Delete History
        for active_history in user.active_histories:
            self.session().refresh(active_history)
            for hda in active_history.active_datasets:
                # Delete HistoryDatasetAssociation
                hda.deleted = True
                self.session().add(hda)
            active_history.deleted = True
            self.session().add(active_history)
        # Delete UserGroupAssociations
        for uga in user.groups:
            self.session().delete(uga)
        _cleanup_nonprivate_user_roles(self.session(), user, private_role.id)
        # Delete UserAddresses
        for address in user.addresses:
            self.session().delete(address)
        # Delete UserAuthnzTokens, unlinking any external identities
        for authnz in user.social_auth:
            self.session().delete(authnz)
        compliance_log = logging.getLogger("COMPLIANCE")
        compliance_log.info(f"delete-user-event: {user.username}")
        # Maybe there is some case in the future where an admin needs
        # to prove that a user was using a server for some reason (e.g.
        # a court case.) So we make this painfully hard to recover (and
        # not immediately reversable) in line with GDPR, but still
        # leave open the possibility to prove someone was part of the
        # server just in case. By knowing the exact email + approximate
        # time of deletion, one could run through hashes for every
        # second of the surrounding days/weeks.
        pseudorandom_value = str(int(time.time()))
        # Replace email + username with a (theoretically) unreversable
        # hash. If provided with the username we can probably re-hash
        # to identify if it is needed for some reason.
        #
        # Deleting multiple times will re-hash the username/email
        email_hash = new_secure_hash_v2(user.email + pseudorandom_value)
        uname_hash = new_secure_hash_v2(user.username + pseudorandom_value)
        # Redact all roles user has
        for role in user.all_roles():
            if self.app.config.redact_username_during_deletion:
                role.name = role.name.replace(user.username, uname_hash)
                if role.description:
                    role.description = role.description.replace(user.username, uname_hash)

            if self.app.config.redact_email_during_deletion:
                role.name = role.name.replace(user.email, email_hash)
                if role.description:
                    role.description = role.description.replace(user.email, email_hash)
            self.session().add(role)
        private_role.name = email_hash
        private_role.description = f"Private Role for {email_hash}"
        self.session().add(private_role)
        # Redact user's email and username
        user.email = email_hash
        user.username = uname_hash
        # The display name is free text and typically holds a real name, so it is
        # dropped outright - there is nothing here worth keeping a hash of.
        user.display_name = None
        # Redact user addresses as well
        if self.app.config.redact_user_address_during_deletion:
            stmt = select(UserAddress).where(UserAddress.user_id == user.id)
            for addr in self.session().scalars(stmt):
                addr.desc = new_secure_hash_v2(addr.desc + pseudorandom_value)
                addr.name = new_secure_hash_v2(addr.name + pseudorandom_value)
                addr.institution = new_secure_hash_v2(addr.institution + pseudorandom_value)
                addr.address = new_secure_hash_v2(addr.address + pseudorandom_value)
                addr.city = new_secure_hash_v2(addr.city + pseudorandom_value)
                addr.state = new_secure_hash_v2(addr.state + pseudorandom_value)
                addr.postal_code = new_secure_hash_v2(addr.postal_code + pseudorandom_value)
                addr.country = new_secure_hash_v2(addr.country + pseudorandom_value)
                addr.phone = new_secure_hash_v2(addr.phone + pseudorandom_value)
                self.session().add(addr)
        # Purge the user
        super().purge(user, flush=flush)


    def _error_on_duplicate_email(self, email: str) -> None:
        """
        Check for a duplicate email and raise if found.

        :raises exceptions.Conflict: if any are found
        """
        if self.by_email(email, case_sensitive=False) is not None:
            raise exceptions.Conflict("Email must be unique", email=email)



    def by_id(self, user_id: int) -> model.User | None:
        return self.app.model.session.get(self.model_class, user_id)


    # ---- filters


    def by_email(self, email: str, case_sensitive: bool = True, deleted: bool | None = None) -> model.User | None:
        """
        Find a user by their email.
        """
        return get_user_by_email(
            self.session(), email, self.model_class, case_sensitive=case_sensitive, deleted=deleted
        )




    def by_api_key(self, api_key: str, sa_session=None):
        """
        Find a user by API key.
        """
        if self.check_bootstrap_admin_api_key(api_key=api_key):
            return schema.BootstrapAdminUser()
        sa_session = sa_session or self.app.model.session
        try:
            stmt = select(self.app.model.APIKeys).filter_by(key=api_key, deleted=False)
            provided_key = sa_session.execute(stmt).scalar_one()
        except NoResultFound:
            raise exceptions.AuthenticationFailed("Provided API key is not valid.")
        if provided_key.user.deleted:
            raise exceptions.AuthenticationFailed("User account is deactivated, please contact an administrator.")
        sa_session.refresh(provided_key.user)
        newest_key = provided_key.user.api_keys[0]
        if newest_key.key != provided_key.key:
            raise exceptions.AuthenticationFailed("Provided API key has expired.")
        return provided_key.user




    def by_credentials(self, api_key: str | None, access_token: str | None) -> model.User | None:
        """Find the user for a Galaxy API key or, failing that, an OIDC access token.

        The API key is tried first, and a bad one raises. An access token returns None when no OIDC
        provider is configured or none of them accepts the token.
        """
        if api_key:
            return self.by_api_key(api_key=api_key)
        if access_token:
            return self.by_oidc_access_token(access_token=access_token)
        return None




    def by_oidc_access_token(self, access_token: str):
        if hasattr(self.app, "authnz_manager") and self.app.authnz_manager:
            user = self.app.authnz_manager.match_access_token_to_user(self.app.model.session, access_token)
            return user
        else:
            return None




    def check_bootstrap_admin_api_key(self, api_key):
        bootstrap_admin_api_key = getattr(self.app.config, "bootstrap_admin_api_key", None)
        if not bootstrap_admin_api_key:
            return False
        # Hash keys to make them the same size, so we can do safe comparison.
        bootstrap_hash = hashlib.sha256(util.smart_str(bootstrap_admin_api_key)).hexdigest()
        provided_hash = hashlib.sha256(util.smart_str(api_key)).hexdigest()
        return util.safe_str_cmp(bootstrap_hash, provided_hash)


    # ---- admin


    def is_admin(self, user: model.User | None, trans: ProvidesUserContext | None = None) -> bool:
        """Return True if this user is an admin (or session is authenticated as admin).

        Do not pass trans to simply check if an existing user object is an admin user,
        pass trans when checking permissions.
        """
        if user is None:
            # Anonymous session or master_api_key used, if master_api_key is detected
            # return True.
            return bool(trans and trans.user_is_admin)
        return self.app.config.is_admin_user(user)




    def logins_resolve_accounts_by_email(self) -> bool:
        """
        Whether a login can land in an existing account by its email address alone, so that a self-chosen
        address would claim the account that someone else's first login lands in. Remote user logins look
        accounts up by email, accounts come only from external logins when local accounts are disabled, and
        with a single OIDC provider and no other authenticators (``fixed_delegated_auth``) a first OIDC login
        is associated with the account that has its email.
        """
        config = self.app.config
        return config.use_remote_user or config.disable_local_accounts or config.fixed_delegated_auth


    def _is_admin_email(self, email: str) -> bool:
        return email.lower() in (admin_email.lower() for admin_email in self.app.config.admin_users_list)



    def admins(self, filters=None, **kwargs):
        """
        Return a list of admin Users.

        Deleted (and therefore purged) accounts are excluded even when their
        email is still listed in ``admin_users``: they can no longer act as
        admins and must not receive admin-targeted notifications.
        """
        admin_emails = self.app.config.admin_users_list
        filters = combine_lists(self.model_class.email.in_(admin_emails), filters)
        filters = combine_lists(self.model_class.deleted == false(), filters)
        return super().list(filters=filters, **kwargs)




    def error_unless_admin(self, user, msg="Administrators only", **kwargs):
        """
        Raise an error if `user` is not an admin.

        :raises exceptions.AdminRequiredException: if `user` is not an admin.
        """
        # useful in admin only methods
        if not self.is_admin(user, trans=kwargs.get("trans", None)):
            raise exceptions.AdminRequiredException(msg, **kwargs)
        return user


    # ---- anonymous


    def is_anonymous(self, user: model.User | None) -> bool:
        """
        Return True if `user` is anonymous.
        """
        # define here for single point of change and make more readable
        return user is None




    def error_if_anonymous(self, user, msg="Log-in required", **kwargs):
        """
        Raise an error if `user` is anonymous.
        """
        if user is None:
            # TODO: code is correct (401) but should be named AuthenticationRequired (401 and 403 are flipped)
            raise exceptions.AuthenticationFailed(msg, **kwargs)
        return user




    def get_user_by_identity(self, identity):
        """Get user by username or email."""
        user = None
        if VALID_EMAIL_RE.match(identity):
            # VALID_PUBLICNAME and VALID_EMAIL do not overlap, so 'identity' here is an email address
            user = self.by_email(identity)
            if not user:
                # Try a case-insensitive match on the email
                user = self.by_email(identity, case_sensitive=False)
        else:
            user = get_user_by_username(self.session(), identity, self.model_class)
        return user


    # ---- current


    def current_user(self, trans: ProvidesUserContext):
        # define here for single point of change and make more readable
        # TODO: trans
        return trans.user




    def user_can_do_run_as(self, user: model.User | schema.BootstrapAdminUser | None) -> bool:
        run_as_users = [u for u in self.app.config.get("api_allow_run_as", "").split(",") if u]
        if not run_as_users:
            return False
        if user is None:
            return False
        # Can do if explicitly in list or master_api_key supplied.
        return user.email in run_as_users or user.bootstrap_admin_user


    # ---- preferences


    def preferences(self, user):
        return dict(user.preferences.items())


    # ---- roles and permissions


    def private_role(self, user):
        return self.app.security_agent.get_private_user_role(user)




    def sharing_roles(self, user):
        return self.app.security_agent.get_sharing_roles(user)




    def default_permissions(self, user):
        return self.app.security_agent.user_get_default_permissions(user)




    def quota(self, user, total=False, quota_source_label=None):
        if total:
            return self.app.quota_agent.get_quota_nice_size(user, quota_source_label=quota_source_label)
        return self.app.quota_agent.get_percent(user=user, quota_source_label=quota_source_label)




    def quota_bytes(self, user, quota_source_label: str | None = None):
        return self.app.quota_agent.get_quota(user=user, quota_source_label=quota_source_label)




    def change_password(self, trans, password=None, confirm=None, token=None, id=None, current=None):
        """
        Allows to change a user password with a token.
        """
        if not token and not id:
            return None, "Please provide a token or a user and password."
        if token:
            token_result = trans.sa_session.get(self.app.model.PasswordResetToken, token)
            if not token_result or not token_result.expiration_time > now():
                return None, "Invalid or expired password reset token, please request a new one."
            user = token_result.user
            message = self.__set_password(trans, user, password, confirm)
            if message:
                return None, message
            log.info("Password reset token redeemed for user %s.", user.id)
            return user, "Password has been changed. Token has been invalidated."
        else:
            if not isinstance(id, int):
                id = self.app.security.decode_id(id)
            user = self.by_id(id)
            if user:
                message = self.app.auth_manager.check_change_password(user, current, trans.request)
                if message:
                    return None, message
                message = self.__set_password(trans, user, password, confirm)
                if message:
                    return None, message
                return user, "Password has been changed."
            else:
                return user, "User not found."


    def __set_password(self, trans: ProvidesUserContext, user, password, confirm):
        if not password:
            return "Please provide a new password."
        if user:
            try:
                self.set_password(trans, user, password, confirm=confirm)
            except exceptions.RequestParameterInvalidException as e:
                return e.err_msg
        else:
            return "Failed to determine user, access denied."



    def set_password(self, trans: PasswordChangeContext, user: User, password: str, confirm: str | None = None) -> None:
        """Validate and set a new password, and log the user out of every other session."""
        if message := validate_password(trans, password, password if confirm is None else confirm):
            raise exceptions.RequestParameterInvalidException(message)
        user.set_password_cleartext(password)
        # The tool shed maps its own GalaxySession, so take the class from the app's model.
        GalaxySession = self.app.model.GalaxySession
        stmt = update(GalaxySession).where(GalaxySession.user_id == user.id, GalaxySession.is_valid == true())
        if trans.galaxy_session:
            stmt = stmt.where(GalaxySession.id != trans.galaxy_session.id)
        trans.sa_session.execute(stmt.values(is_valid=False))
        self.expire_reset_tokens(trans, user)
        trans.sa_session.add(user)
        trans.sa_session.commit()
        trans.log_event("User change password")
        log.info("Password changed for user %s.", user.id)




    def set_roles(self, user: User, role_ids: list[int]) -> None:
        """Replace the user's role associations; the private role is kept."""
        self.app.security_agent.set_user_group_and_role_associations(user, role_ids=role_ids)
        # The associations are written with bulk statements, so the loaded collection is stale.
        self.session().refresh(user)




    def set_groups(self, user: User, group_ids: list[int]) -> None:
        """Replace the groups the user is a member of."""
        self.app.security_agent.set_user_group_and_role_associations(user, group_ids=group_ids)
        self.session().refresh(user)




    def get_groups(self, user: User) -> list[tuple[int, str]]:
        """Return (id, name) of the non-deleted groups the user is a member of."""
        return get_user_groups(self.session(), user.id)




    def impersonate(self, trans: "GalaxyWebTransaction", user):
        if not trans.app.config.allow_user_impersonation:
            raise exceptions.MessageException("User impersonation is not enabled in this instance of Galaxy.")
        if user:
            trans.handle_user_logout()
            trans.handle_user_login(user)
        else:
            raise exceptions.MessageException("Please provide a valid user.")




    def send_activation_email(self, trans, email, username):
        """
        Send the verification email containing the activation link to the user's email.
        """
        activation_token = self.__get_activation_token(trans, email)
        activation_link = trans.url_builder(
            "/user/activate",
            activation_token=activation_token,
            email=escape(email),
            qualified=True,
        )
        template_context = {
            "name": escape(username),
            "user_email": escape(email),
            "date": now().strftime("%D"),
            "hostname": trans.request.host,
            "activation_url": activation_link,
            "terms_url": self.app.config.terms_url,
            "contact_email": self.app.config.error_email_to,
            "instance_resource_url": self.app.config.instance_resource_url,
            "custom_message": self.app.config.custom_activation_email_message,
            "expiry_days": self.app.config.activation_grace_period,
        }
        # The HTML body is autoescaped to prevent XSS via attacker-influenced
        # values such as the Host-derived ``hostname``; pre-escaped ``Markup``
        # values (``name``/``user_email``) pass through unchanged under
        # autoescape, and the admin's ``custom_message`` is marked ``| safe`` in
        # the template. The plain-text body is never autoescaped (see
        # templates.render's ``.txt`` guard).
        body = templates.render(
            TXT_ACTIVATION_EMAIL_TEMPLATE_RELPATH, template_context, self.app.config.templates_dir, autoescape=False
        )
        html = templates.render(
            HTML_ACTIVATION_EMAIL_TEMPLATE_RELPATH, template_context, self.app.config.templates_dir, autoescape=True
        )
        to = email
        subject = "Galaxy Account Activation"
        try:
            util.send_mail(self.app.config.email_from, to, subject, body, self.app.config, html=html)
            self.session().commit()
            return True
        except Exception:
            log.debug(body)
            log.exception("Unable to send the activation email.")
            return False


    def __get_activation_token(self, trans: ProvidesAppContext, email):
        """
        Check for the activation token. Create new activation token and store it in the database if no token found.
        Flushes but does not commit—the caller is responsible for committing the transaction.
        """
        session = trans.sa_session
        # Flush pending changes so the user is visible to the DB query below.
        session.flush()
        user = get_user_by_email(session, email, self.app.model.User)
        assert user is not None, f"User with email '{email}' not found while generating activation token."
        activation_token = user.activation_token
        if activation_token is None:
            activation_token = util.hash_util.new_secure_hash_v2(str(random.getrandbits(256)))
            user.activation_token = activation_token
            session.add(user)
            session.flush()
        return activation_token



    def request_password_reset(self, trans, email: str, *, reset_url_for: Callable[[str], str]):
        """Email a reset link built by ``reset_url_for``, which must use a trusted base URL.

        Raise ConfigDoesNotAllowException for missing mail configuration or
        RequestParameterInvalidException for invalid addresses. Unknown accounts
        and delivery failures return silently to prevent account enumeration.
        """
        app_name = "Galaxy" if self.app.name == GALAXY_APP_NAME else "Tool Shed"
        if self.app.config.smtp_server is None:
            raise exceptions.ConfigDoesNotAllowException(
                f"Mail is not configured for this {app_name} instance and password reset information cannot be sent. "
                "Please contact your local administrator."
            )
        if message := validate_email(trans, email, check_dup=False):
            raise exceptions.RequestParameterInvalidException(message)
        reset_user, prt = self.get_reset_token(trans, email)
        if reset_user is None or prt is None:
            log.warning("Password reset requested for an address without an active account.")
            return
        expiration_time = prt.expiration_time
        assert expiration_time is not None
        body = PASSWORD_RESET_TEMPLATE % (
            app_name,
            expiration_time.strftime(self.app.config.pretty_datetime_format),
            reset_url_for(prt.token),
        )
        try:
            util.send_mail(self.app.config.email_from, email, f"{app_name} Password Reset", body, self.app.config)
        except Exception:
            # Keep delivery errors private and invalidate the undelivered token.
            log.exception("Failed to send password reset email.")
            prt.expiration_time = now()
            trans.sa_session.add(prt)
            trans.sa_session.commit()
            return
        trans.log_event(f"User reset password: {email}")
        log.info("Sent a password reset email for user %s.", reset_user.id)




    def send_reset_email(self, trans: "GalaxyWebTransaction", payload, **kwd):
        """Reset the user's password. Send an email with token that allows a password change."""
        email = payload.get("email")
        if not email:
            return "Please provide your email."
        try:
            self.request_password_reset(
                trans,
                email,
                reset_url_for=lambda token: trans.url_builder("/login/start", token=token, qualified=True),
            )
        except exceptions.MessageException as e:
            return str(e)
        return None




    def expire_reset_tokens(self, trans: UserValidationContext, user) -> None:
        stmt = select(self.app.model.PasswordResetToken).where(
            and_(
                self.app.model.PasswordResetToken.user_id == user.id,
                self.app.model.PasswordResetToken.expiration_time > now(),
            )
        )
        for token in trans.sa_session.scalars(stmt):
            token.expiration_time = now()
            trans.sa_session.add(token)




    def get_reset_token(self, trans: ProvidesAppContext, email):
        reset_user = self.by_email(email)
        if not reset_user:
            reset_user = self.by_email(email, case_sensitive=False)
        if reset_user and not reset_user.deleted:
            # Only the most recent link works, so repeated requests cannot pile up live tokens.
            self.expire_reset_tokens(trans, reset_user)
            prt = self.app.model.PasswordResetToken(reset_user)
            trans.sa_session.add(prt)
            trans.sa_session.commit()
            return reset_user, prt
        return None, None




    def send_subscription_email(self, email):
        if self.app.config.smtp_server is None:
            return "Subscribing to the mailing list has failed because mail is not configured for this Galaxy instance. Please contact your local Galaxy administrator."
        else:
            body = (self.app.config.mailing_join_body or "") + "\n"
            to = self.app.config.mailing_join_addr
            frm = email
            subject = self.app.config.mailing_join_subject or ""
            try:
                util.send_mail(frm, to, subject, body, self.app.config)
            except Exception:
                log.exception("Subscribing to the mailing list has failed.")
                return "Subscribing to the mailing list has failed."




    def activate(self, user):
        user.active = True
        self.session().add(user)
        session = self.session()
        session.commit()




    def get_or_create_remote_user(self, remote_user_email):
        """
        Create a remote user with the email remote_user_email and return it
        """
        if not self.app.config.use_remote_user:
            return None
        if getattr(self.app.config, "normalize_remote_user_email", False):
            remote_user_email = remote_user_email.lower()
        user = self.by_email(remote_user_email)
        if not user:
            # Try a case-insensitive match on the email
            user = self.by_email(remote_user_email, case_sensitive=False)
        if user:
            # Ensure a private role and default permissions are set for remote users (remote user creation bug existed prior to 2009)
            self.app.security_agent.get_private_user_role(user, auto_create=True)
            if self.app_type == "galaxy":
                self.app.security_agent.create_user_role(user, self.app)
        elif user is None:
            session = self.session()
            random.seed()
            username = username_from_email(session, remote_user_email, self.model_class)
            user = self.model_class(email=remote_user_email, username=username)
            user.set_random_password(length=12)
            user.external = True
            session.add(user)
            session.commit()
            self.app.security_agent.create_private_user_role(user)
            # We set default user permissions, before we log in and set the default history permissions
            if self.app_type == "galaxy":
                self.app.security_agent.create_user_role(user, self.app)
            # self.log_event( "Automatically created account '%s'", user.email )
        return user






class UserSerializer(base.ModelSerializer, deletable.PurgableSerializerMixin):
    model_manager_class = UserManager



    def __init__(self, app: MinimalManagerApp, extra_preferences_manager: ExtraPreferencesManager):
        """
        Convert a User and associated data to a dictionary representation.
        """
        super().__init__(app)
        self.user_manager = self.manager
        self.extra_preferences_manager = extra_preferences_manager

        self.default_view = "summary"
        self.add_view("summary", ["id", "email", "username"])
        self.add_view(
            "detailed",
            [
                "active",
                "deleted",
                "display_name",
                "is_admin",
                "nice_total_disk_usage",
                "preferences",
                "preferred_object_store_id",
                "purged",
                "quota",
                "quota_bytes",
                "quota_percent",
                "total_disk_usage",
            ],
            include_keys_from="summary",
        )




    def add_serializers(self):
        super().add_serializers()
        deletable.PurgableSerializerMixin.add_serializers(self)

        self.serializers.update(
            {
                "id": self.serialize_id,
                "create_time": self.serialize_date,
                "update_time": self.serialize_date,
                "active": lambda i, k, **c: bool(i.active),
                "is_admin": lambda i, k, **c: self.user_manager.is_admin(i),
                "preferences": lambda i, k, **c: self.serialize_preferences(i),
                "total_disk_usage": lambda i, k, **c: float(i.total_disk_usage),
                "quota_percent": lambda i, k, **c: self.user_manager.quota(i),
                "quota": lambda i, k, **c: self.user_manager.quota(i, total=True),
                "quota_bytes": lambda i, k, **c: self.user_manager.quota_bytes(i),
            }
        )




    def serialize_disk_usage(self, user: model.User) -> list[UserQuotaUsage]:
        usages = user.dictify_usage(self.app.object_store)
        rval: list[UserQuotaUsage] = []
        for usage in usages:
            quota_source_label = usage.quota_source_label
            quota_percent = self.user_manager.quota(user, quota_source_label=quota_source_label)
            quota = self.user_manager.quota(user, total=True, quota_source_label=quota_source_label)
            quota_bytes = self.user_manager.quota_bytes(user, quota_source_label=quota_source_label)
            rval.append(
                UserQuotaUsage(
                    quota_source_label=quota_source_label,
                    total_disk_usage=usage.total_disk_usage,
                    quota_percent=quota_percent,
                    quota=quota,
                    quota_bytes=quota_bytes,
                )
            )
        return rval




    def serialize_disk_usage_for(self, user: model.User, label: str | None) -> UserQuotaUsage:
        usage = user.dictify_usage_for(label)
        quota_source_label = usage.quota_source_label
        quota_percent = self.user_manager.quota(user, quota_source_label=quota_source_label)
        quota = self.user_manager.quota(user, total=True, quota_source_label=quota_source_label)
        quota_bytes = self.user_manager.quota_bytes(user, quota_source_label=quota_source_label)
        return UserQuotaUsage(
            quota_source_label=quota_source_label,
            total_disk_usage=usage.total_disk_usage,
            quota_percent=quota_percent,
            quota=quota,
            quota_bytes=quota_bytes,
        )




    def serialize_preferences(self, user: User) -> dict[str, str]:
        preferences = self.user_manager.preferences(user)
        if EXTRA_PREFERENCES_KEY in preferences:
            # Values of password and secret inputs are written by the user but never sent back.
            preferences[EXTRA_PREFERENCES_KEY] = self.extra_preferences_manager.redact(
                preferences[EXTRA_PREFERENCES_KEY]
            )
        return preferences






class UserDeserializer(base.ModelDeserializer):
    """
    Service object for validating and deserializing dictionaries that
    update/alter users.
    """

    model_manager_class = UserManager



    def add_deserializers(self):
        super().add_deserializers()
        user_deserializers: dict[str, base.Deserializer] = {
            "active": self.deserialize_active,
            "username": self.deserialize_username,
            "display_name": self.deserialize_display_name,
            "preferred_object_store_id": self.deserialize_preferred_object_store_id,
            "email": self.deserialize_email,
        }
        self.deserializers.update(user_deserializers)




    def deserialize_active(self, item, key, active, trans: ProvidesUserContext | None = None, **context):
        # Activation is the email-verification gate under user_activation_on: a user
        # who could flip it themselves would skip verifying a new address, or the
        # address they registered with once the grace period ends.
        if trans is None or not trans.user_is_admin:
            raise exceptions.AdminRequiredException("Only an administrator can change whether a user is active.")
        return self.deserialize_bool(item, key, active, **context)




    def deserialize_email(self, item, key, email, trans: ProvidesUserContext | None = None, **context):
        if trans is None:
            raise base.ModelDeserializingError("Email addresses cannot be changed in this context.")
        # update_email keeps the private role in sync and honours user_activation_on.
        # Without activation the change is committed by ModelDeserializer.deserialize.
        # With activation, send_activation_email commits once the mail is sent, and a
        # failed send rolls the session back, which also drops the fields this payload
        # set before the email.
        self.manager.update_email(trans, item, email, commit=False, send_activation_email=True)
        return email




    def deserialize_preferred_object_store_id(
        self, item: Any, key: Any, val: Any, trans: ProvidesUserContext | None = None, **context
    ):
        preferred_object_store_id = val
        validation_error = validate_preferred_object_store_id(
            trans.user if trans else None, self.app.object_store, preferred_object_store_id
        )
        if validation_error:
            raise base.ModelDeserializingError(validation_error)
        return self.default_deserializer(item, key, preferred_object_store_id, **context)




    def deserialize_display_name(self, item, key, display_name, **context):
        # update_display_name strips, validates and assigns. commit=False because
        # ModelDeserializer.deserialize commits once at the end.
        self.manager.update_display_name(item, display_name, commit=False)
        return item.display_name




    def deserialize_username(self, item, key, username, trans: ProvidesAppContext | None = None, **context):
        # TODO: validate_publicname requires trans and should(?) raise exceptions
        # move validation to UserValidator and use self.app, exceptions instead
        validation_error = validate_publicname(trans, username, user=item)  # type: ignore[arg-type]
        if validation_error:
            raise base.ModelDeserializingError(validation_error)
        return self.default_deserializer(item, key, username, trans=trans, **context)






class CurrentUserSerializer(UserSerializer):
    model_manager_class = UserManager



    def serialize(self, user, keys, **kwargs):
        """
        Override to return at least some usage info if user is anonymous.
        """
        kwargs["current_user"] = user
        if self.user_manager.is_anonymous(user):
            return self.serialize_current_anonymous_user(user, keys, **kwargs)
        return super(UserSerializer, self).serialize(user, keys, **kwargs)




    def serialize_current_anonymous_user(self, user, keys, trans: ProvidesHistoryContext | None = None, **kwargs):
        # use the current history if any to get usage stats for trans' anonymous user
        # TODO: might be better as sep. Serializer class
        usage: float = 0
        percent = None

        if trans is not None and trans.history:
            usage = self.app.quota_agent.get_usage(trans, history=trans.history) or 0
            percent = self.app.quota_agent.get_percent(trans=trans, usage=usage)

        # a very small subset of keys available
        values = {
            "id": None,
            "total_disk_usage": float(usage),
            "nice_total_disk_usage": util.nice_size(usage),
            "quota_percent": percent,
        }
        serialized = {}
        for key in keys:
            if key in values:
                serialized[key] = values[key]
        return serialized






class AdminUserFilterParser(base.ModelFilterParser, deletable.PurgableFiltersMixin):
    model_manager_class = UserManager
    model_class = model.User

    def _add_parsers(self):
        super()._add_parsers()
        deletable.PurgableFiltersMixin._add_parsers(self)

        # PRECONDITION: user making the query has been verified as an admin
        self.orm_filter_parsers.update(
            {
                "email": {"op": ("eq", "contains", "like")},
                "username": {"op": ("eq", "contains", "like")},
                "active": {"op": ("eq")},
                "disk_usage": {"op": ("le", "ge")},
            }
        )

        self.fn_filter_parsers.update({})





def username_from_email(session, email, model_class=User):
    """Get next available username generated based on email"""
    username = email.split("@", 1)[0].lower()
    username = filter_out_invalid_username_characters(username)
    if username_exists(session, username, model_class):
        username = generate_next_available_username(session, username, model_class)
    return username





def filter_out_invalid_username_characters(username):
    """Replace invalid characters in username"""
    for char in [x for x in username if x not in f"{string.ascii_lowercase + string.digits}-."]:
        username = username.replace(char, "-")
    return username





def username_exists(session, username: str, model_class=User):
    return bool(get_user_by_username(session, username, model_class))





def generate_next_available_username(session, username, model_class=User):
    """Generate unique username; user can change it later"""
    i = 1
    while session.execute(select(model_class).where(model_class.username == f"{username}-{i}")).first():
        i += 1
    return f"{username}-{i}"