Admin fixes

app/admin/custom_domain_search.py

@@ -24,6 +24,7 @@
     Alias,
     DomainDeletedAlias,
 )
+from app.regex_utils import is_safe_regex_pattern
 
 
 class CustomDomainWithValidationData:
@@ -48,45 +49,69 @@ 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:<int>': 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)
                     for d in user.custom_domains
                 ]
                 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:<int>'
+        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,

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

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(

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:

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