Source code for galaxy.security.validate_user_input

"""
Utilities for validating inputs related to user objects.

The validate_* methods in this file return simple messages that do not contain
user inputs - so these methods do not need to be escaped.
"""

import logging
from typing import (
    Any,
    Protocol,
    TYPE_CHECKING,
)

import dns.resolver
from dns.exception import DNSException
from sqlalchemy import (
    func,
    select,
)
from typing_extensions import LiteralString

from galaxy.objectstore import ObjectStore
from galaxy.util.user_input import (
    EMAIL_MAX_LEN,
    FILL_CHAR,
    is_valid_email_str,
    PASSWORD_MIN_LEN,
    PUBLICNAME_MAX_LEN,
    transform_publicname,
    VALID_EMAIL_RE,
    VALID_PUBLICNAME_RE,
    VALID_PUBLICNAME_SUB,
    validate_email_str,
    validate_password_str,
    validate_publicname_str,
)

if TYPE_CHECKING:
    from galaxy.model import User



[docs]
class UserValidationContext(Protocol):
    """What the user input validators need from a transaction.

    Galaxy and the tool shed keep separate context hierarchies with their own
    app, config and User classes. Both offer a session and an app, so the
    validators ask for that much rather than naming either hierarchy and
    forcing one side to depend on the other.
    """

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

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



log = logging.getLogger(__name__)



[docs]
def validate_email(
    trans: UserValidationContext, email, user=None, check_dup=True, allow_empty=False, validate_domain=False
):
    """
    Validates the email format.
    Checks whether the domain is blocklisted in the disposable domains configuration.
    Checks whether the email address is banned.
    Optionally checks if email exists.
    """
    if (user and user.email == email) or (email == "" and allow_empty):
        return ""
    message = validate_email_str(email)
    if not message and validate_domain:
        domain = extract_domain(email)
        message = validate_email_domain_name(domain)

    if not message:
        if is_email_banned(email, trans.app.config.email_ban_file, trans.app.config.canonical_email_rules):
            message = "This email address has been banned."

    if not message and check_dup:
        user_class = trans.app.model.User
        stmt = select(user_class).filter(func.lower(user_class.email) == email.lower())
        if user is not None:
            # Changing only the case of one's own address is not a duplicate.
            stmt = stmt.filter(user_class.id != user.id)
        if trans.sa_session.scalars(stmt.limit(1)).first():
            message = f"User with email '{email}' already exists."

    if not message:
        # If the allowlist is not empty filter out any domain not in the list and ignore blocklist.
        if trans.app.config.email_domain_allowlist_content is not None:
            domain = extract_domain(email)
            if domain not in trans.app.config.email_domain_allowlist_content:
                message = "Please enter an allowed domain email address for this server."
        # If the blocklist is not empty filter out the disposable domains.
        elif trans.app.config.email_domain_blocklist_content is not None:
            domain = extract_domain(email, base_only=True)
            if domain in trans.app.config.email_domain_blocklist_content:
                message = "Please enter your permanent email address."

    return message




[docs]
def validate_email_domain_name(domain: str) -> LiteralString:
    message = ""
    try:
        dns.resolver.resolve(domain, "MX")
    except DNSException:
        try:
            # Per RFC 5321, try to fall back to the A record (implicit MX) for
            # the domain, see https://www.rfc-editor.org/rfc/rfc5321#section-5.1
            dns.resolver.resolve(domain, "A")
        except DNSException:
            message = "The email domain cannot be resolved."
    return message




[docs]
def extract_domain(email, base_only=False):
    domain = email.rsplit("@", 1)[-1]
    parts = domain.split(".")
    if len(parts) > 2 and base_only:
        return (".").join(parts[-2:])
    return domain




[docs]
def validate_publicname(trans: UserValidationContext, publicname, user=None):
    """
    Check that publicname respects the minimum and maximum string length, the
    allowed characters, and that the username is not taken already.
    """
    if user and user.username == publicname:
        return ""
    if message := validate_publicname_str(publicname):
        return message

    stmt = select(trans.app.model.User).filter_by(username=publicname).limit(1)
    if trans.sa_session.scalars(stmt).first():
        return "Public name is taken; please choose another."
    return ""




[docs]
def validate_password(trans: UserValidationContext, password, confirm):
    if password != confirm:
        return "Passwords do not match."
    return validate_password_str(password)




[docs]
def validate_preferred_object_store_id(
    user: "User | None", object_store: ObjectStore, preferred_object_store_id: str | None
) -> str:
    return object_store.validate_selected_object_store_id(user, preferred_object_store_id) or ""




[docs]
def is_email_banned(email: str, filepath: str | None, canonical_email_rules: dict | None) -> bool:
    if not filepath:
        return False
    normalizer = EmailAddressNormalizer(canonical_email_rules)
    email = normalizer.normalize(email)
    banned_emails = _read_email_ban_list(filepath)
    for address in banned_emails:
        if email == normalizer.normalize(address):
            return True
    return False



def _read_email_ban_list(filepath: str) -> list[str]:
    with open(filepath) as f:
        return [line.strip() for line in f if not line.startswith("#")]



[docs]
class EmailAddressNormalizer:
    IGNORE_CASE_RULE = "ignore_case"
    IGNORE_DOTS_RULE = "ignore_dots"
    SUB_ADDRESSING_RULE = "sub_addressing"
    SUB_ADDRESSING_DELIM = "sub_addressing_delim"
    SUB_ADDRESSING_DELIM_DEFAULT = "+"
    ALL = "all"


[docs]
    def __init__(self, canonical_email_rules: dict | None) -> None:
        self.config = canonical_email_rules



[docs]
    def normalize(self, email: str) -> str:
        """Transform email to its canonical form."""

        email_localpart, email_domain = email.split("@")
        # the domain part of an email address is case-insensitive (RFC1035)
        email_domain = email_domain.lower()

        # Step 1: If no rules are set, do not modify local-part
        if not self.config:
            return f"{email_localpart}@{email_domain}"

        # Step 2: Apply rules defined for all services before applying rules defined for specific services
        if self.ALL in self.config:
            email_localpart = self._apply_rules(email_localpart, self.ALL)

        # Step 3: Apply rules definied for each email service if email matches service
        for service in (s for s in self.config if s != self.ALL):
            service = service.lower()  # ensure domain is lowercase
            apply_rules = False

            if email_domain == service:
                apply_rules = True
            elif self.config[service].get("aliases"):
                service_aliases = [
                    a.lower() for a in self.config[service]["aliases"]
                ]  # ensure domain aliases are lowercase
                if email_domain in service_aliases:
                    # email domain is an alias of the service. Change it to the service's primary domain name.
                    email_domain = service
                    apply_rules = True

            if apply_rules:
                email_localpart = self._apply_rules(email_localpart, service)

        return f"{email_localpart}@{email_domain}"


    def _apply_rules(self, email_localpart: str, service: str) -> str:
        assert self.config
        config = self.config[service]

        if config.get(self.IGNORE_CASE_RULE, False):
            email_localpart = email_localpart.lower()
        if config.get(self.IGNORE_DOTS_RULE, False):
            email_localpart = email_localpart.replace(".", "")
        if config.get(self.SUB_ADDRESSING_RULE, False):
            delim = config.get(self.SUB_ADDRESSING_DELIM, self.SUB_ADDRESSING_DELIM_DEFAULT)
            if email_localpart.find(delim) > -1:
                email_localpart = email_localpart[: email_localpart.index(delim)]

        return email_localpart



__all__ = (
    "EMAIL_MAX_LEN",
    "EmailAddressNormalizer",
    "extract_domain",
    "FILL_CHAR",
    "is_email_banned",
    "is_valid_email_str",
    "PASSWORD_MIN_LEN",
    "PUBLICNAME_MAX_LEN",
    "transform_publicname",
    "UserValidationContext",
    "VALID_EMAIL_RE",
    "VALID_PUBLICNAME_RE",
    "VALID_PUBLICNAME_SUB",
    "validate_email",
    "validate_email_domain_name",
    "validate_email_str",
    "validate_password",
    "validate_password_str",
    "validate_preferred_object_store_id",
    "validate_publicname",
    "validate_publicname_str",
)