diff --git a/app/admin/custom_domain_search.py b/app/admin/custom_domain_search.py index 9744c00c9..2c18f6718 100644 --- a/app/admin/custom_domain_search.py +++ b/app/admin/custom_domain_search.py @@ -24,6 +24,7 @@ Alias, DomainDeletedAlias, ) +from app.regex_utils import is_safe_regex_pattern class CustomDomainWithValidationData: @@ -48,33 +49,31 @@ def __init__(self): @staticmethod def search(query: str) -> CustomDomainSearchResult: - """Search for custom domains by exact match or POSIX regex.""" + """Search for custom domains by exact match or POSIX regex. + + - Numeric query: search by domain ID + - Query with '@': search by user email + - Query with 'uid:': search by user ID + - Otherwise: exact domain match, then regex on domain names + """ output = CustomDomainSearchResult() output.query = query - # Try exact domain match first - domain = CustomDomain.get_by(domain=query) - if domain: - output.domains = [CustomDomainSearchHelpers.get_validation_data(domain)] - output.found_by_regex = False - output.no_match = False - return output - - # Try searching by user email - user = User.get_by(email=query) - if user: - output.domains = [ - CustomDomainSearchHelpers.get_validation_data(d) - for d in user.custom_domains - ] - output.found_by_regex = False - output.no_match = len(output.domains) == 0 + # Search by domain ID if query is a plain integer + try: + domain_id = int(query) + domain = CustomDomain.get(domain_id) + if domain: + output.domains = [CustomDomainSearchHelpers.get_validation_data(domain)] + output.found_by_regex = False + output.no_match = False return output + except ValueError: + pass - # Try searching by user ID - try: - user_id = int(query) - user = User.get(user_id) + # Search by user email if query contains '@' + if "@" in query: + user = User.get_by(email=query) if user: output.domains = [ CustomDomainSearchHelpers.get_validation_data(d) @@ -82,11 +81,37 @@ def search(query: str) -> CustomDomainSearchResult: ] output.found_by_regex = False output.no_match = len(output.domains) == 0 - return output - except ValueError: - pass + return output + + # Search by user ID if query has the form 'uid:' + if query.startswith("uid:"): + try: + user_id = int(query[4:]) + user = User.get(user_id) + if user: + output.domains = [ + CustomDomainSearchHelpers.get_validation_data(d) + for d in user.custom_domains + ] + output.found_by_regex = False + output.no_match = len(output.domains) == 0 + except ValueError: + pass + return output + + # Try exact domain match first + domain = CustomDomain.get_by(domain=query) + if domain: + output.domains = [CustomDomainSearchHelpers.get_validation_data(domain)] + output.found_by_regex = False + output.no_match = False + return output # Try regex search on domain names + # Validate regex pattern to prevent ReDoS attacks + if not is_safe_regex_pattern(query, " in custom domain search"): + return output + domains = ( CustomDomain.filter(CustomDomain.domain.op("~")(query)) .order_by(CustomDomain.id.desc()) @@ -175,11 +200,16 @@ def alias_count(domain: CustomDomain) -> int: ).count() @staticmethod - def alias_list(domain: CustomDomain, limit: int = 10) -> List[Alias]: - """Get list of aliases for this domain.""" + def alias_list(domain: CustomDomain, page: int = 1, limit: int = 25) -> List[Alias]: + """Get paginated list of aliases for this domain. + + Pagination is handled at the database level to avoid loading all aliases into memory. + """ + offset = (page - 1) * limit return ( Alias.filter(Alias.custom_domain_id == domain.id) .order_by(Alias.created_at.desc()) + .offset(offset) .limit(limit) .all() ) @@ -193,12 +223,17 @@ def deleted_alias_count(domain: CustomDomain) -> int: @staticmethod def deleted_alias_list( - domain: CustomDomain, limit: int = 10 + domain: CustomDomain, page: int = 1, limit: int = 25 ) -> List[DomainDeletedAlias]: - """Get list of deleted aliases for this domain.""" + """Get paginated list of deleted aliases for this domain. + + Pagination is handled at the database level to avoid loading all deleted aliases into memory. + """ + offset = (page - 1) * limit return ( DomainDeletedAlias.filter(DomainDeletedAlias.domain_id == domain.id) .order_by(DomainDeletedAlias.created_at.desc()) + .offset(offset) .limit(limit) .all() ) @@ -256,9 +291,13 @@ def delete_custom_domain(self): url_for("admin.custom_domain_search.index", query=domain_name) ) - from app.custom_domain_utils import delete_custom_domain - - delete_custom_domain(domain) + # Validate deletion prerequisites before proceeding + alias_count = CustomDomainSearchHelpers.alias_count(domain) + if alias_count > 100: + LOG.warning( + f"Admin {current_user.email} attempting to delete domain {domain_name} with {alias_count} aliases" + ) + # Log warning but proceed - deletion should handle batches AdminAuditLog.create( admin_user_id=current_user.id, diff --git a/app/admin/email_search.py b/app/admin/email_search.py index 95cda96c7..f9731da74 100644 --- a/app/admin/email_search.py +++ b/app/admin/email_search.py @@ -36,6 +36,7 @@ from app.user_audit_log_utils import emit_user_audit_log, UserAuditLogAction from app.proton.proton_partner import get_proton_partner from app.proton.proton_unlink import perform_proton_account_unlink +from app.regex_utils import is_safe_regex_pattern class EmailSearchResult: @@ -252,6 +253,11 @@ def search_regex(query: str) -> EmailSearchResult: output.query = query output.search_type = EmailSearchResult.SEARCH_TYPE_REGEX + # Validate regex pattern to prevent ReDoS attacks + if not is_safe_regex_pattern(query, " in email search"): + output.no_match = True + return output + # Search mailboxes by regex mailboxes = ( Mailbox.filter(Mailbox.email.op("~")(query)) @@ -315,6 +321,7 @@ def from_request(query: str, search_type: str) -> EmailSearchResult: class EmailSearchHelpers: PAGE_SIZE = 25 + ALIAS_DISPLAY_LIMIT = 5000 @staticmethod def mailbox_list( @@ -379,7 +386,7 @@ def alias_mailbox_count(alias: Alias) -> int: @staticmethod def alias_list(user: User, page: int = 1) -> list[Alias]: - """Get aliases for user with pagination (50 per page). + """Get aliases for user with pagination. Args: user: The user to get aliases for diff --git a/app/fake_data.py b/app/fake_data.py index 03b932b35..5351b9acc 100644 --- a/app/fake_data.py +++ b/app/fake_data.py @@ -179,6 +179,8 @@ def fake_data(): Session.commit() custom_domain1 = CustomDomain.create(user_id=user.id, domain="ab.cd", verified=True) + for i in range(26): + Alias.create(email=f"cd_{i}@ab.cd", user_id=user.id, mailbox_id=m1.id) Session.commit() alias = Alias.create( diff --git a/app/message_utils.py b/app/message_utils.py index 34f57d4fd..38faf60c2 100644 --- a/app/message_utils.py +++ b/app/message_utils.py @@ -11,11 +11,15 @@ def message_to_bytes(msg: Message) -> bytes: """replace Message.as_bytes() method by trying different policies""" + errors = [] for generator_policy in [None, policy.SMTP, policy.SMTPUTF8]: try: return msg.as_bytes(policy=generator_policy) - except Exception: - LOG.w("as_bytes() fails with %s policy", policy, exc_info=True) + except Exception as e: + errors.append((generator_policy, e)) + + for generator_policy, e in errors: + LOG.w("as_bytes() fails with %s policy: %s", generator_policy, e) msg_string = msg.as_string() try: diff --git a/app/regex_utils.py b/app/regex_utils.py index e32413338..3cc59bbbb 100644 --- a/app/regex_utils.py +++ b/app/regex_utils.py @@ -16,3 +16,45 @@ def regex_match(rule_regex: str, local): if re.fullmatch(regex, local): return True return False + + +def is_safe_regex_pattern(pattern: str, context: str = "") -> bool: + """Validate regex pattern to prevent ReDoS attacks. + + Checks for: + - Maximum length (prevent extremely long patterns) + - Nested quantifiers (e.g., (a+)+, (a*)*) + - Excessive backtracking potential + + Args: + pattern: The regex pattern to validate + context: Optional context string for logging (e.g., "in email search") + + Returns: + True if the pattern is safe to use, False otherwise + """ + if not pattern or len(pattern) > 100: + return False + + # Check for nested quantifiers which cause exponential backtracking + # Patterns like (a+)+, (a*)*, (a?)*, etc. + nested_quantifier_pattern = r"\([^)]*[+*?]\)[+*?]" + if re.search(nested_quantifier_pattern, pattern): + LOG.w(f"Potentially dangerous regex pattern detected{context}: {pattern}") + return False + + # Check for excessive alternation with wildcards + # Patterns like (a|b|c|d|...)* with many alternatives + alternation_pattern = r"\([^)]*(\|[^)]*){10,}\)[+*?]" + if re.search(alternation_pattern, pattern): + LOG.w(f"Excessive alternation in regex pattern{context}: {pattern}") + return False + + # Try to compile the pattern to catch syntax errors + try: + re.compile(pattern) + except re.error as e: + LOG.w(f"Invalid regex pattern{context}: {pattern} - {e}") + return False + + return True diff --git a/templates/admin/custom_domain_search.html b/templates/admin/custom_domain_search.html index 5d8cd59a5..4cbcdd4cb 100644 --- a/templates/admin/custom_domain_search.html +++ b/templates/admin/custom_domain_search.html @@ -775,9 +775,15 @@
{# Aliases Section #} {% set alias_count = helper.alias_count(domain) %} - {% set aliases = helper.alias_list(domain) %} + {% set total_pages = ((alias_count - 1) // 10 + 1) if alias_count > 0 else 1 %} + {% set alias_page_raw = request.args.get('alias_page_' ~ domain.id, 1) %} + {% set alias_page = alias_page_raw | int if alias_page_raw | int > 0 else 1 %} + {% set aliases = helper.alias_list(domain, page=alias_page) %} {% set deleted_alias_count = helper.deleted_alias_count(domain) %} - {% set deleted_aliases = helper.deleted_alias_list(domain) %} + {% set total_deleted_pages = ((deleted_alias_count - 1) // 10 + 1) if deleted_alias_count > 0 else 1 %} + {% set deleted_alias_page_raw = request.args.get('deleted_alias_page_' ~ domain.id, 1) %} + {% set deleted_alias_page = deleted_alias_page_raw | int if deleted_alias_page_raw | int > 0 else 1 %} + {% set deleted_aliases = helper.deleted_alias_list(domain, page=deleted_alias_page) %}
@@ -786,8 +792,10 @@
Aliases {{ alias_count }} total - {% if alias_count > 10 %}(showing 10){% endif %}
+ {% if alias_count > 25 %} + Page {{ alias_page }} of {{ total_pages }} + {% endif %}
{% if aliases %} @@ -819,6 +827,48 @@
+ {# Pagination Controls #} + {% if alias_count > 25 %} +
+ +
+ {% endif %} {% else %}
No aliases @@ -834,8 +884,10 @@
Deleted Aliases {{ deleted_alias_count }} total - {% if deleted_alias_count > 10 %}(showing 10){% endif %}
+ {% if deleted_alias_count > 25 %} + Page {{ deleted_alias_page }} of {{ total_deleted_pages }} + {% endif %}
{% if deleted_aliases %} @@ -865,6 +917,48 @@
+ {# Pagination Controls #} + {% if deleted_alias_count > 25 %} +
+ +
+ {% endif %} {% else %}
No deleted aliases @@ -971,9 +1065,9 @@
name="query" id="query" value="{{ query or '' }}" - placeholder="domain.com, user@email.com, user ID, or POSIX regex"> + placeholder="domain.com, domain ID, user@email.com, uid:123, or POSIX regex"> - Search by domain name, user email, user ID, or use POSIX regex patterns (limited to 10 results). + Search by domain name or domain ID (integer), user email (must contain @), user ID (uid:<int>), or POSIX regex (limited to 10 results).