diff --git a/app/api/v1/mail/ApiMailMail.py b/app/api/v1/mail/ApiMailMail.py index 73492451..b5eab9ed 100644 --- a/app/api/v1/mail/ApiMailMail.py +++ b/app/api/v1/mail/ApiMailMail.py @@ -10,7 +10,7 @@ from app.interface.mail.InterfaceApiMailMail import InterfaceApiMailMail from app.utils.logger.logger import logger_api -from app.utils.api.paginate_sort_filter import collection_paginate, CustomPaginateResponse +from app.utils.api.paginate_sort_filter import collection_paginate, CustomPaginateResponse, DeletedFilterQueryArgsSchema from .schemas.mail import ( MailDetailResponseSchema, MailListResponseSchema, @@ -60,8 +60,9 @@ class ApiMailFolderIdMail(MethodView): """ @blp.response(200, MailListResponseSchema, example=MailListResponseSchema.example()) + @blp.arguments(DeletedFilterQueryArgsSchema, location="query", arg_name="deleted_query") @collection_paginate(blp, sort_value_set=MailListResponseSchema.sort_by_values(), filter_value_set=MailListResponseSchema.filter_by_values()) - def get(self, collection_param: CollectionPaginateArgs, account_id: str, folder_name: str) -> CustomPaginateResponse: + def get(self, collection_param: CollectionPaginateArgs, deleted_query: dict, account_id: str, folder_name: str) -> CustomPaginateResponse: """Fetch the list of mails in a specific folder. The filtering for this endpoint is special:\r\n @@ -100,10 +101,17 @@ def get(self, collection_param: CollectionPaginateArgs, account_id: str, folder_ If you want just to list the mails while not needing the actual content, set `fields="contents"` and `fields_action="exclude"`. + The `deleted` query parameter (boolean, default `false`) controls whether mails + flagged `\\Deleted` are included: `false` (default) excludes them, `true` includes + them alongside non-deleted mails. It is applied as an IMAP search criterion, not + a post-fetch filter, so pagination is unaffected by it. + --- :param collection_param: pagination, sorting and filtering args :type collection_param: CollectionPaginateArgs + :param deleted_query: parsed "deleted" query param + :type deleted_query: dict :param account_id: The account identifier :type account_id: str :param folder_name: The folder identifier @@ -114,7 +122,7 @@ def get(self, collection_param: CollectionPaginateArgs, account_id: str, folder_ logger_api.debug("Calling ApiMailFolderIdMail: Fetching mail list for account_id: %s, folder_name: %s, params: %s", account_id, folder_name, collection_param) interface: InterfaceApiMailMail = g.inter - item_count, response, status_code = interface.get_mail_list(account_id, folder_name, collection_param) + item_count, response, status_code = interface.get_mail_list(account_id, folder_name, collection_param, deleted_query["deleted"]) return item_count, response, status_code @@ -140,6 +148,8 @@ def post(self, data: dict, account_id: str, folder_name: str) -> ResponseReturnV * **ham**: Mark the selected mails as not spam. * **copy**: Copy the selected mails to another folder. The destination folder name must be provided in the ``data`` field as a string. * **delete**: Delete the selected mails, following the user's mail delete behavior preference. + * **illegal**: Report the selected mails as illegal content and move them to the Junk folder. + * **phishing**: Report the selected mails as phishing and move them to the Junk folder. :param data: The batch action data containing 'uids', 'action' and optional 'data' field :type data: dict @@ -229,6 +239,8 @@ def post(self, data: dict, account_id: str, folder_name: str, mail_uid: str) -> * **ham**: Mark the mail as not spam. * **copy**: Copy the mail to another folder. The destination folder name must be provided in the ``data`` field as a string. * **delete**: Delete the mail, following the user's mail delete behavior preference. + * **illegal**: Report the mail as illegal content and move it to the Junk folder. + * **phishing**: Report the mail as phishing and move it to the Junk folder. :param data: The action data containing 'action' and optional 'data' field :type data: dict diff --git a/app/api/v1/mail/ApiMailMailbox.py b/app/api/v1/mail/ApiMailMailbox.py index 2bfbc290..d40e031a 100644 --- a/app/api/v1/mail/ApiMailMailbox.py +++ b/app/api/v1/mail/ApiMailMailbox.py @@ -8,6 +8,7 @@ from app.interface.mail.InterfaceApiMailMailbox import InterfaceApiMailMailbox from app.utils.logger.logger import logger_api from app.utils.api.ApiBaseResponse import ApiBaseResponse +from app.utils.api.paginate_sort_filter import collection_paginate, CustomPaginateResponse, DeletedFilterQueryArgsSchema from app.api.v1.mail.schemas.mailbox import ( MailboxCreateSchema, MailboxUpdateSchema, @@ -18,11 +19,16 @@ DelegationResponseSchema, MailboxPurgeSchema, MailboxPurgeResponseSchema, + MailboxBatchActionSchema, + MailboxBatchActionResponseSchema, + MailboxSearchSchema, + MailboxSearchResponseSchema, ) if TYPE_CHECKING: from app.config.settings.ProcessSetting import ProcessSetting from app.auth.User import User + from app.utils.api.paginate_sort_filter import CollectionPaginateArgs blp = Blueprint("Mail Account", __name__, url_prefix="/mailboxes") @@ -140,6 +146,51 @@ def post(self, data: dict, account_id: str) -> ResponseReturnValue: return interface.create_mailbox_delegate(account_id, data) +@blp.route("//batch-action") +class ApiMailBoxesAccountBatchAction(MethodView): + """ + Resource: Batch actions across the whole mailbox + """ + @blp.arguments(MailboxBatchActionSchema, example=MailboxBatchActionSchema.example(), error_status_code=400) + @blp.response(200, MailboxBatchActionResponseSchema, example=MailboxBatchActionResponseSchema.example()) + def post(self, data: dict, account_id: str) -> ResponseReturnValue: + """Perform an action (tag, untag, move, spam, ham, copy) on mails from several folders of the account at once. + + Behaves like the per-folder batch action endpoint, except that ``uids`` maps folder names + to their list of mail UIDs, so mails from multiple folders can be processed in a single call. + Each folder is processed independently: a failure on one folder does not prevent the others + from being processed, and the per-folder outcome is reported in the response's ``results`` + and ``errors`` fields. + + **Supported actions:** + + * **tag**: Add one or more tags to the selected mails. Tags are provided in the ``data`` field as a list of strings. + * **untag**: Remove one or more tags from the selected mails. Tags to remove are provided in the ``data`` field as a list of strings. + * **move**: Move the selected mails to another folder. The destination folder name must be provided in the ``data`` field as a string. + * **spam**: Mark the selected mails as spam. + * **ham**: Mark the selected mails as not spam. + * **copy**: Copy the selected mails to another folder. The destination folder name must be provided in the ``data`` field as a string. + * **delete**: Delete the selected mails, following the user's mail delete behavior preference. + * **illegal**: Report the selected mails as illegal content and move them to the Junk folder. + * **phishing**: Report the selected mails as phishing and move them to the Junk folder. + + :param data: The batch action data containing 'uids' (folder name -> list of uids), 'action' and optional 'data' field + :type data: dict + :param account_id: The account identifier + :type account_id: str + :return: A response indicating the per-folder result of the action + :rtype: ResponseReturnValue + """ + logger_api.debug( + "Calling ApiMailBoxesAccountBatchAction.post for account_id: %s, uids: %s with action: %s", + account_id, + data["uids"], + data["action"] + ) + interface: InterfaceApiMailMailbox = g.inter + return interface.mailbox_batch_action(account_id, data) + + @blp.route("//purge") class ApiMailBoxesAccountPurge(MethodView): """ @@ -155,3 +206,47 @@ def post(self, purge_data: dict, account_id: str) -> ResponseReturnValue: interface: InterfaceApiMailMailbox = g.inter return interface.purge_mailbox(account_id, purge_data) + +@blp.route("//search") +class ApiMailBoxesAccountSearch(MethodView): + """ + Resource: Advanced Mail Search + """ + @blp.arguments(MailboxSearchSchema, example=MailboxSearchSchema.example(), error_status_code=400) + @blp.response(200, MailboxSearchResponseSchema) + @blp.arguments(DeletedFilterQueryArgsSchema, location="query", arg_name="deleted_query") + @collection_paginate(blp, can_sort=True, sort_value_set={"date", "relevance", "sender", "subject", "size"}, + can_filter=True, filter_value_set={"contents"}) + def post(self, search_params: dict, collection_param: "CollectionPaginateArgs", deleted_query: dict, account_id: str) -> CustomPaginateResponse: + """ + Advanced mail search across one or multiple folders. + + * **operator**: str, 'AND' (default) or 'OR' - how the criteria below are combined. + With 'AND' every provided criterion must match, with 'OR' at least one must match. + * **text**: str, full text search in subject/sender/recipients/body + * **folders**: list[str], list of folder paths to search in (e.g. ["INBOX", "Sent"] or ["all"] for all folders) + * **include_subfolders**: bool, default True - when True, also search the subfolders of each folder listed in "folders"; when False, search only the exact folders listed. Ignored when "folders" is empty or ["all"]. + * **date_range**: dict, date range for the search (e.g. {"from": "2023-01-01", "to": "2023-01-31"}) + * **has_attachments**: bool, whether to search for emails with attachments + * **to**: str, email address to search for in either the recipient (To) or copy (Cc) headers + * **bcc**: str, blind copy (Bcc) email address to search for + * **from**: list[str], list of sender email addresses to search for + * **subject** : str, keywords to search for in the email subject + * **attachment_type**: list[str], list of attachment types to search for (e.g. ["pdf", "jpg"]) + * **is_read**: bool, whether to search for read or unread emails + * **labels**: list[str], list of labels/tags to search for + * **size**: dict, filter by mail size (e.g. {"value": 15, "operator": ">", "unit": "kb"}). + ``operator`` is ">" (larger than) or "<" (smaller than), ``unit`` is "kb", "mb" or "gb" + (default "kb"). Uses the native IMAP LARGER/SMALLER search keys. + + All search criteria are optional and combined using the "operator" field (AND by default, OR to match any criterion). + Pagination, sorting and field filtering are controlled via query parameters (page, page_size, sort_by, sort_order, fields, fields_action). + + The `deleted` query parameter (boolean, default `false`) controls whether mails + flagged `\\Deleted` are included: `false` (default) excludes them, `true` includes + them alongside non-deleted mails. It is applied as part of the IMAP search + criteria, not a post-fetch filter, so pagination is unaffected by it. + """ + logger_api.debug("Calling ApiMailBoxesAccountSearch.post for account_id: %s with params: %s", account_id, search_params) + interface: InterfaceApiMailMailbox = g.inter + return interface.search_mailbox(account_id, search_params, collection_param, deleted_query["deleted"]) diff --git a/app/api/v1/mail/schemas/mail.py b/app/api/v1/mail/schemas/mail.py index 96e39bb8..321d04c2 100644 --- a/app/api/v1/mail/schemas/mail.py +++ b/app/api/v1/mail/schemas/mail.py @@ -52,7 +52,7 @@ class MailActionSchema(Schema): """ action = fields.String( required=True, - validate=validate.OneOf(['tag', 'untag', 'move', 'spam', 'ham', 'copy', 'delete']) + validate=validate.OneOf(['tag', 'untag', 'move', 'spam', 'ham', 'copy', 'delete', 'illegal', 'phishing']) ) data = fields.Raw(required=False, allow_none=True) @@ -76,7 +76,7 @@ class MailBatchActionSchema(Schema): uids = fields.List(fields.Integer(), required=True, validate=validate.Length(min=1)) action = fields.String( required=True, - validate=validate.OneOf(['tag', 'untag', 'move', 'spam', 'ham', 'copy', 'delete']) + validate=validate.OneOf(['tag', 'untag', 'move', 'spam', 'ham', 'copy', 'delete', 'illegal', 'phishing']) ) data = fields.Raw(required=False, allow_none=True) diff --git a/app/api/v1/mail/schemas/mailbox.py b/app/api/v1/mail/schemas/mailbox.py index a53274f9..1ada6f5c 100644 --- a/app/api/v1/mail/schemas/mailbox.py +++ b/app/api/v1/mail/schemas/mailbox.py @@ -583,6 +583,67 @@ def example(cls) -> dict: } +class MailboxBatchActionSchema(Schema): + """ + Schema for POST /mailboxes//batch-action - Perform an action on multiple mails + spanning multiple folders of the same account in a single call. + """ + uids = fields.Dict( + keys=fields.String(), + values=fields.List(fields.Integer(), validate=validate.Length(min=1)), + required=True, + validate=validate.Length(min=1) + ) + action = fields.String( + required=True, + validate=validate.OneOf(['tag', 'untag', 'move', 'spam', 'ham', 'copy', 'delete', 'illegal', 'phishing']) + ) + data = fields.Raw(required=False, allow_none=True) + + @classmethod + def example(cls) -> dict: + """Example data for mailbox batch action. + + :return: Example mailbox batch action payload + :rtype: dict + """ + return { + "uids": { + "INBOX": [42, 43, 27, 21], + "Trash": [42, 43] + }, + "action": "tag", + "data": ["important"] + } + + +class MailboxBatchActionResponseSchema(ApiBaseResponse): + """ + Schema for POST /mailboxes//batch-action response + """ + data = fields.Dict(required=False, allow_none=True) + + @classmethod + def example(cls) -> dict: + """Example response for mailbox batch action. + + :return: Example mailbox batch action response + :rtype: dict + """ + return { + "error_code": 0, + "error_msg": "", + "data": { + "action": "tag", + "results": { + "INBOX": {"action": "tag", "mail_uid": ["42", "43", "27", "21"], "tags_added": ["important"]}, + "Trash": {"action": "tag", "mail_uid": ["42", "43"], "tags_added": ["important"]} + }, + "errors": {} + } + } + + class MailboxPurgeResponseSchema(ApiBaseResponse): """ Schema for POST /mailboxes//purge response @@ -604,3 +665,125 @@ def example(cls) -> dict: } } +class DateRangeSchema(Schema): + """ + Schema for date range filter in advanced search + """ + start = fields.String(required=False, allow_none=True, metadata={"description": "Start date in ISO 8601 format (e.g. 2026-05-01T00:00:00Z), or a bare date (e.g. 2026-05-01). A bare date includes the entire day regardless of time."}) + end = fields.String(required=False, allow_none=True, metadata={"description": "End date in ISO 8601 format (e.g. 2026-05-19T23:59:59Z), or a bare date (e.g. 2026-05-19). A bare date includes the entire day regardless of time."}) + + @classmethod + def example(cls) -> dict: + return { + "start": "2026-05-01T00:00:00Z", + "end": "2026-05-19T23:59:59Z" + } + + +class SizeFilterSchema(Schema): + """ + Schema for the size filter in advanced search + """ + value = fields.Integer(required=True, validate=validate.Range(min=0), metadata={"description": "Size threshold, expressed in the given unit"}) + operator = fields.String(required=True, validate=validate.OneOf([">", "<"]), metadata={"description": "'>' for mails larger than value, '<' for mails smaller than value"}) + unit = fields.String(required=False, load_default="kb", validate=validate.OneOf(["kb", "mb", "gb"]), metadata={"description": "Unit of 'value': kb, mb or gb"}) + + @classmethod + def example(cls) -> dict: + return { + "value": 15, + "operator": ">", + "unit": "kb" + } + + +class MailboxSearchSchema(Schema): + """ + Schema for POST /mailboxes//search - Advanced mail search. + + All fields are optional. When multiple criteria are provided, they are combined + using the "operator" field: "AND" (default, every criterion must match) or + "OR" (at least one criterion must match). + """ + operator = fields.String( + required=False, + allow_none=True, + load_default="AND", + validate=validate.OneOf(["AND", "OR"]), + metadata={"description": "Logical operator combining the search criteria below: 'AND' (default) requires every provided criterion to match, 'OR' requires at least one to match"} + ) + text = fields.String(required=False, allow_none=True, load_default=None, metadata={"description": "Full-text search in body and headers"}) + from_ = fields.String(required=False, allow_none=True, load_default=None, data_key="from", metadata={"description": "Filter by sender email address"}) + to = fields.String(required=False, allow_none=True, load_default=None, metadata={"description": "Filter by recipient email address (matches either the To or the Cc header)"}) + bcc = fields.String(required=False, allow_none=True, load_default=None, metadata={"description": "Filter by Bcc recipient email address"}) + subject = fields.String(required=False, allow_none=True, load_default=None, metadata={"description": "Filter by subject (substring match)"}) + has_attachment = fields.Boolean(required=False, allow_none=True, load_default=None, metadata={"description": "Filter mails that have (or don't have) attachments"}) + attachment_type = fields.List(fields.String(), required=False, allow_none=True, load_default=None, metadata={"description": "Filter by attachment file extensions (e.g. ['pdf', 'jpg'])"}) + date_range = fields.Nested(DateRangeSchema, required=False, allow_none=True, load_default=None, metadata={"description": "Filter by date range"}) + is_read = fields.Boolean(required=False, allow_none=True, load_default=None, metadata={"description": "Filter by read/unread status"}) + is_flagged = fields.Boolean(required=False, allow_none=True, load_default=None, metadata={"description": "Filter by starred (flagged) status"}) + folders = fields.List(fields.String(), required=False, allow_none=True, load_default=None, metadata={"description": "Folders to search in (use ['all'] for entire mailbox)"}) + include_subfolders = fields.Boolean(required=False, allow_none=True, load_default=True, metadata={"description": "If True (default), also search in the subfolders of each folder listed in 'folders'. If False, search only in the exact folders listed"}) + labels = fields.List(fields.String(), required=False, allow_none=True, load_default=None, metadata={"description": "Filter by IMAP keyword labels"}) + size = fields.Nested(SizeFilterSchema, required=False, allow_none=True, load_default=None, metadata={"description": "Filter by mail size"}) + + @classmethod + def example(cls) -> dict: + """Example data for advanced mail search. + + :return: Example search payload + :rtype: dict + """ + return { + "operator": "AND", + "text": "contrat urgent", + "from": "customer@entreprise.com", + "to": "jdoe@domaine.com", + "bcc": "hidden@domaine.com", + "subject": "Projet X", + "has_attachment": True, + "attachment_type": ["pdf", "jpg"], + "date_range": { + "start": "2025-05-01T00:00:00Z", + "end": "2026-05-19T23:59:59Z" + }, + "is_read": False, + "is_flagged": True, + "folders": ["INBOX", "Archive"], + "include_subfolders": True, + "labels": ["important", "work"], + "size": { + "value": 15, + "operator": ">", + "unit": "kb" + }, + } + + +class MailboxSearchResponseSchema(ApiBaseResponse): + """ + Schema for the response of the advanced mail search endpoint. + """ + data = fields.Dict(required=False, allow_none=True, metadata={"description": "Search results with mails list and total count"}) + + @classmethod + def example(cls) -> dict: + return { + "error_code": 0, + "error_msg": "", + "data": { + "total": 2, + "mails": [ + { + "uid": "42", + "subject": "Projet X - Contrat urgent", + "from": {"name": "Client", "email": "client@entreprise.com"}, + "date": "Tue, 19 May 2026 10:00:00 +0000", + "seen": False, + "flagged": True, + "has_attachment": True, + "folder": "INBOX" + } + ] + } + } diff --git a/app/interface/mail/InterfaceApiMailMail.py b/app/interface/mail/InterfaceApiMailMail.py index cd762d7a..c4903468 100644 --- a/app/interface/mail/InterfaceApiMailMail.py +++ b/app/interface/mail/InterfaceApiMailMail.py @@ -36,9 +36,9 @@ def __init__(self, process_setting: ProcessSetting, user_domain_settings: dict, self.mail_module = ModuleMail(self.user, self.mail_settings, self.process_setting) - def get_mail_list(self, account_id: str, folder_name: str, collection_param: CollectionPaginateArgs) -> tuple[int, dict[str, Any], int]: + def get_mail_list(self, account_id: str, folder_name: str, collection_param: CollectionPaginateArgs, deleted: bool = False) -> tuple[int, dict[str, Any], int]: """Retrieve a list of mails in a specific folder. - + :param account_id: The ID of the account :type account_id: str :param folder_name: The ID of the folder @@ -47,11 +47,15 @@ def get_mail_list(self, account_id: str, folder_name: str, collection_param: Col :type first: int :param last: The last item index (0-based, exclusive) :type last: int + :param deleted: If False (default), mails flagged as deleted are excluded. If + True, they are included alongside non-deleted mails (no filtering on the + deleted flag). + :type deleted: bool :return: A tuple of (total_count, API response dict, status code) :rtype: tuple[int, dict[str, Any], int] """ try: - result, total_count = self.mail_module.get_folder_mails(account_id, folder_name, collection_param) + result, total_count = self.mail_module.get_folder_mails(account_id, folder_name, collection_param, deleted) return total_count, *create_api_base_response(result) except RequestException as ex: logger_api.error("Request exception in get_mail_list: %s", str(ex)) diff --git a/app/interface/mail/InterfaceApiMailMailbox.py b/app/interface/mail/InterfaceApiMailMailbox.py index 24934126..709832ce 100644 --- a/app/interface/mail/InterfaceApiMailMailbox.py +++ b/app/interface/mail/InterfaceApiMailMailbox.py @@ -2,6 +2,8 @@ from typing import TYPE_CHECKING, Any from http import HTTPStatus +from marshmallow import ValidationError + from app.config.settings.DomainSettings import UserModuleSettings, UserModuleSettingsObj, MailSettings, MailSettingsObj from app.module.mail.ModuleMail import ModuleMail from app.module.mail.ModuleMailOutgoing import ModuleMailOutgoing @@ -15,6 +17,7 @@ if TYPE_CHECKING: from app.config.settings.ProcessSetting import ProcessSetting from app.auth.User import User + from app.utils.api.paginate_sort_filter import CollectionPaginateArgs class InterfaceApiMailMailbox: @@ -243,6 +246,27 @@ def purge_mailbox(self, account_id: str, purge_data: dict[str, Any]) -> tuple[di return create_api_base_response(None, ex.error) + def mailbox_batch_action(self, account_id: str, batch_action_data: dict[str, Any]) -> tuple[dict[str, Any], int]: + """Perform an action on multiple mails spanning multiple folders of the same account. + + :param account_id: The account identifier + :type account_id: str + :param batch_action_data: Dictionary containing 'uids' (folder name -> list of uids), + 'action' and optional 'data' fields + :type batch_action_data: dict[str, Any] + :return: A tuple of (API response dict, status code) + :rtype: tuple[dict[str, Any], int] + """ + try: + result = self.mail_module.perform_mailbox_batch_action(account_id, batch_action_data) + return create_api_base_response(result) + except ValidationError as ex: + logger_api.error("Validation error in mailbox_batch_action: %s", ex.messages) + return create_api_base_response(None, err.ERROR_VALIDATION_ERROR) + except RequestException as ex: + logger_api.error("Request exception in mailbox_batch_action for user %s, account %s: %s", self.user.uid, account_id, str(ex)) + return create_api_base_response(None, ex.error) + def save_draft(self, account_id: str, mail_data: dict, key: str | None = None) -> tuple[dict, int]: """Save a mail as a draft in the account's Drafts folder. @@ -296,3 +320,29 @@ def send_mail(self, account_id: str, mail_data: dict, draft_uid: str | None = No logger_api.warning("Failed to delete draft mail uid %s for user %s, account %s: %s", draft_uid, self.user.uid, account_id, str(ex)) return create_api_base_response(None) + + def search_mailbox(self, account_id: str, search_params: dict, collection_param: "CollectionPaginateArgs", deleted: bool = False) -> tuple[int, dict, int]: + """Advanced mail search across one or multiple folders for the given account. + + :param account_id: The account identifier ("0" for main account) + :type account_id: str + :param search_params: Validated search parameters (from MailboxSearchSchema) + :type search_params: dict + :param collection_param: Pagination, sorting and filtering parameters. + :type collection_param: CollectionPaginateArgs + :param deleted: If False (default), mails flagged as deleted are excluded. If + True, they are included alongside non-deleted mails (no filtering on the + deleted flag). + :type deleted: bool + :return: A tuple of (total_count, API response dict, status code) + :rtype: tuple[int, dict, int] + """ + if account_id != cs.DEFAULT_IDENTITY_KEY_VALUE and not self.user_module_settings.SOGO_D_ALLOW_EXT_MAIL_ACCOUNT: + return 0, *create_api_base_response(error=err.ERROR_EXTERNAL_ACCOUNT_FORBIDDEN) + + try: + result, total = self.mail_module.search_mails(account_id, search_params, collection_param, deleted) + except RequestException as ex: + logger_api.error("Request exception in search_mailbox for user %s, account %s: %s", self.user.uid, account_id, str(ex)) + return 0, *create_api_base_response(None, ex.error) + return total, *create_api_base_response(result) diff --git a/app/manager/mail/ClientImap.py b/app/manager/mail/ClientImap.py index a3a4a3fb..185c0d65 100644 --- a/app/manager/mail/ClientImap.py +++ b/app/manager/mail/ClientImap.py @@ -1,7 +1,7 @@ from __future__ import annotations from typing import Any, Callable, TypeVar, ParamSpec, Iterator, cast -from datetime import datetime +from datetime import datetime, timedelta from email import message_from_bytes from email.header import decode_header, make_header from email.message import EmailMessage @@ -16,7 +16,7 @@ from app.manager.mail.ClientMailServer import ClientMailServer from app.utils import errors as err from app.utils import constants as cs -from app.utils.strings import quote, imap_join_folders +from app.utils.strings import quote, imap_join_folders, escape_imap_string # Maximum debug output from imaplib #TODO all imap are logged, including login/auth password used SecretString (on ldap branch not in develoope now) @@ -134,6 +134,50 @@ def _convert_imap_to_rights(imap_rights: str) -> dict[str, int]: return sogo_rights + +def _group_imap_search_parts(parts: list[str]) -> str: + """Group several IMAP search-key strings belonging to the same search field into one search-key. + + A parenthesized list of search-keys is itself a single search-key that matches + when all the keys it contains match (RFC 3501), so this lets a multi-valued field + (e.g. several ``to`` addresses or ``labels``) keep its own AND semantics while being + usable as one atomic group when combined with sibling fields, whatever the top-level + operator (AND/OR) is. + + :param parts: IMAP search-key strings for a single field (e.g. one per address). + :type parts: list[str] + :return: A single search-key string. + :rtype: str + """ + if len(parts) == 1: + return parts[0] + return "(" + " ".join(parts) + ")" + + +def _combine_imap_search_or(criteria: list[str]) -> str: + """Combine independent IMAP search-key strings with OR. + + RFC 3501's ``OR`` search-key takes exactly two search-keys, so combining more than + two terms requires right-nesting: ``OR a (OR b c)``. Parentheses are only added + around a nested ``OR`` (never around a single trailing search-key), keeping the + output minimal while staying unambiguous. This also works transparently for 0 or 1 + criteria. + + :param criteria: Independent IMAP search-key strings to OR together. + :type criteria: list[str] + :return: A single combined search-key string ("" if criteria is empty). + :rtype: str + """ + if not criteria: + return "" + if len(criteria) == 1: + return criteria[0] + rest = _combine_imap_search_or(criteria[1:]) + if len(criteria) > 2: + rest = f"({rest})" + return f"OR {criteria[0]} {rest}" + + class ImapFolder: """ Simple class to parse folder response and store useful values @@ -247,7 +291,7 @@ def parse_uids_from_bytes(byte_data: bytes) -> Iterator[str]: """ current_uid: list[bytes] = [] for byte in byte_data: - if byte == b' ': + if byte == ord(' '): if current_uid: # Avoid yielding empty strings yield b''.join(current_uid).decode('utf-8') current_uid = [] @@ -492,10 +536,14 @@ def _imap_list_folders(self, folder_path: str = '"*"') -> Iterator[ImapFolder]: raise RequestException("Failed to list mailboxes", err.ERROR_IMAP_FAILED) if success: success_list, datas_list = self._exec_imap4_method(self.connection.response, 'LIST') - if not success_list or not datas_list: + if not success_list: raise RequestException(f"Failed to list mailboxes: {datas_list}", err.ERROR_IMAP_FAILED) + if not datas_list: + # No mailbox matched the pattern (e.g. no children, or the + # base folder doesn't exist): nothing to yield, not an error. + return success_status, datas_status = self._exec_imap4_method(self.connection.response, 'STATUS') - if not success_status or not datas_status: + if not success_status: raise RequestException(f"Failed to status mailboxes: {datas_status}", err.ERROR_IMAP_FAILED) idx_status = 0 for data in datas_list: @@ -828,7 +876,7 @@ def expunge_folder(self, folder_path: str, do_children: bool = True) -> int: folder_path = quote(folder_path) self.select_mailbox(folder_path) - success, datas = self.connection.expunge() + success, datas = self._exec_imap4_method(self.connection.expunge) if not success: raise RequestException(f"Failed to expunge mailbox {folder_path}", err.ERROR_IMAP_FAILED) expunged_count += len(datas) @@ -879,6 +927,33 @@ def purge_folder(self, folder_path: str, before_date: str = "", do_children: boo raise BugException("Not authenticated meaning self.connect() and self.login() was not called beforehands") + def get_folder_with_subfolders(self, folder_path: str, include_subfolders: bool = True) -> list[str]: + """Return the given folder path, optionally followed by the paths of all its subfolders. + + :param folder_path: The folder to start from. + :type folder_path: str + :param include_subfolders: If True, also list every subfolder (at any depth) below folder_path. + :type include_subfolders: bool + :return: List of folder paths, folder_path first. + :rtype: list[str] + :raises RequestException: If not connected to the server. + """ + if self.connection is not None and self.authenticated: + folder_path = self._fix_folder_path(folder_path) + folder_paths = [folder_path] + + if include_subfolders: + delimiter = self._get_delimiter_for(folder_path) + pattern_folder_path = quote(f"{folder_path}{delimiter}*") + for folder in self._imap_list_folders(pattern_folder_path): + if folder.can_be_select: + folder_paths.append(folder.path) + + return folder_paths + else: + raise BugException("Not authenticated meaning self.connect() and self.login() was not called beforehands") + + def _is_folder_subscribed(self, folder_path: str) -> bool: """Check if a folder is subscribed. @@ -1140,6 +1215,8 @@ def uid_store_flags(self, mail_uid: str|list|Iterator, flags: list[str], operati if self.connection is not None and self.authenticated: if isinstance(mail_uid, (Iterator, list)): mail_uid = ','.join(mail_uid) + if not mail_uid: + return 0 flags_str = '(' + ' '.join(flags) + ')' success, datas = self._exec_imap4_method(self.connection.uid, 'STORE', mail_uid, operation, flags_str) if not success: @@ -1205,7 +1282,40 @@ def _parse_mail_with_content_fetching(self, message_parts: tuple[bytes, bytes]) "size": size } - def fetch_all_mails_with_content(self, folder_path: str, number_of_mails: int, offset: int) -> Iterator[dict]: + def _search_deleted_filtered_uids(self, folder_path: str, deleted: bool) -> list[str]: + """Select ``folder_path`` and run an IMAP UID SEARCH filtering on the deleted flag. + + Filtering on the deleted flag is done as an IMAP search criterion (not a + post-fetch filter) so that pagination on the result is always accurate: + slicing a fixed-size page out of this list can never yield fewer than the + requested number of mails just because some in-range mails were filtered out. + + :param folder_path: IMAP folder path (already quoted as needed by the caller). + :type folder_path: str + :param deleted: If False, mails flagged \\Deleted are excluded. If True, they + are included alongside non-deleted mails (no filtering on the deleted flag). + :type deleted: bool + :raises RequestException: If selecting the folder or the SEARCH command fails. + :return: Matching UIDs, most recent (highest UID) first. + :rtype: list[str] + """ + if self.connection is None or not self.authenticated: + raise BugException("Not authenticated meaning self.connect() and self.login() was not called beforehands") + self.select_mailbox(folder_path) + + criteria = "ALL" if deleted else "NOT DELETED" + success, datas = self._exec_imap4_method(self.connection.uid, 'SEARCH', criteria) + if not success: + raise RequestException( + f"IMAP SEARCH failed in folder '{folder_path}' with criteria: {criteria}", + err.ERROR_MAIL_SEARCH_FAILED + ) + + uid_list = datas[0].decode().split() if datas and datas[0] else [] + uid_list.reverse() # Recent mails have the highest UID, so most recent first + return uid_list + + def fetch_all_mails_with_content(self, folder_path: str, number_of_mails: int, offset: int, deleted: bool = False) -> Iterator[dict]: """ https://datatracker.ietf.org/doc/html/rfc9051#name-fetch-response Fetch a specific number of mails from a mailbox with full details. @@ -1217,7 +1327,7 @@ def fetch_all_mails_with_content(self, folder_path: str, number_of_mails: int, o "size": size, int } - Always yield the totla number of mails of the folder + Always yield the total number of mails matching the ``deleted`` filter Then yield mail by mail, from the most recent to the oldest :param mailbox: The mailbox to fetch mails from. @@ -1226,6 +1336,10 @@ def fetch_all_mails_with_content(self, folder_path: str, number_of_mails: int, o :type number_of_mails: int :param offset: The offset of the mail to fetch. :type number_of_mails: int + :param deleted: If False (default), mails flagged \\Deleted are excluded. If + True, they are included alongside non-deleted mails (no filtering on the + deleted flag). + :type deleted: bool :raises RequestException: If fetching mails fails :return: A tuple of (list of mail dicts with full details, total count) :rtype: tuple[list[dict[str, Any]], int] @@ -1235,30 +1349,35 @@ def fetch_all_mails_with_content(self, folder_path: str, number_of_mails: int, o if not folder_path.isascii(): raise RequestException(f"Mailbox name is not ascii: {folder_path}", err.ERROR_IMAP_NOT_ASCII) folder_path = quote(folder_path) - mailbox_len = self.select_mailbox(folder_path) + uid_list = self._search_deleted_filtered_uids(folder_path, deleted) #yield length - yield {"nb_mails": mailbox_len} - if mailbox_len == 0: + yield {"nb_mails": len(uid_list)} + if not uid_list: + return + + start = max(0, offset - 1) + page_uids = uid_list[start: start + number_of_mails] + if not page_uids: return - #Recent mail have the higest ID number, so we fetch from descending - range_arg = f'{max(1, mailbox_len - offset + 1 - number_of_mails + 1)}:{mailbox_len - offset + 1}' # Fetch full message, flags, and UID (size is included in BODY.PEEK[] response) - # If success, datas will be of length 2*nb_mails. Each mail is - # a tuple (b'metadata', b'body') and a singular b')' - success, datas = self._exec_imap4_method(self.connection.fetch, range_arg, '(BODY.PEEK[] FLAGS UID)') + success, datas = self._exec_imap4_method(self.connection.uid, 'FETCH', ",".join(page_uids), '(BODY.PEEK[] FLAGS UID)') if not success: - if isinstance(datas[0], str) and "Invalid messageset" in datas[0]: - #Goes here means our range args is wrong, it should'nt happen - raise BugException(f"Try to fetch mail with an unvalid messageset: {range_arg}") raise RequestException(f"Fail to fetch mails: {datas}", err.ERROR_IMAP_FAILED) - for part in reversed(datas): + mails_by_uid: dict[str, dict] = {} + for part in datas: if not isinstance(part, tuple): continue - yield self._parse_mail_with_content_fetching(part) + mail_dict = self._parse_mail_with_content_fetching(part) + mails_by_uid[str(mail_dict["uid"])] = mail_dict + + for uid in page_uids: + mail_dict = mails_by_uid.get(uid) + if mail_dict is not None: + yield mail_dict else: raise BugException("Not authenticated meaning self.connect() and self.login() was not called beforehands") @@ -1359,7 +1478,7 @@ def _parse_mail_without_content_fetching(self, message_parts: tuple[bytes, bytes return ret - def fetch_all_mails_without_content(self, folder_path: str, number_of_mails: int, offset: int) -> Iterator[dict]: + def fetch_all_mails_without_content(self, folder_path: str, number_of_mails: int, offset: int, deleted: bool = False) -> Iterator[dict]: """ https://datatracker.ietf.org/doc/html/rfc9051#name-fetch-response Fetch a specific number of mails from a mailbox with full details. @@ -1372,7 +1491,7 @@ def fetch_all_mails_without_content(self, folder_path: str, number_of_mails: int "has_attachment": bool } - Always yield the total number of mails of the folder + Always yield the total number of mails matching the ``deleted`` filter Then yield mail by mail, from the most recent to the oldest :param mailbox: The mailbox to fetch mails from. @@ -1381,6 +1500,10 @@ def fetch_all_mails_without_content(self, folder_path: str, number_of_mails: int :type number_of_mails: int :param offset: The offset of the mail to fetch. :type number_of_mails: int + :param deleted: If False (default), mails flagged \\Deleted are excluded. If + True, they are included alongside non-deleted mails (no filtering on the + deleted flag). + :type deleted: bool :raises RequestException: If fetching mails fails :return: A tuple of (list of mail dicts with full details, total count) :rtype: tuple[list[dict[str, Any]], int] @@ -1390,26 +1513,27 @@ def fetch_all_mails_without_content(self, folder_path: str, number_of_mails: int if not folder_path.isascii(): raise RequestException(f"Mailbox name is not ascii: {folder_path}", err.ERROR_IMAP_NOT_ASCII) folder_path = quote(folder_path) - mailbox_len = self.select_mailbox(folder_path) + uid_list = self._search_deleted_filtered_uids(folder_path, deleted) #yield length - yield {"nb_mails": mailbox_len} - if mailbox_len == 0: + yield {"nb_mails": len(uid_list)} + if not uid_list: return - #Recent mail have the higest ID number, so we fetch from descending - range_arg = f'{max(1, mailbox_len - offset + 1 - number_of_mails + 1)}:{mailbox_len - offset + 1}' - # Fetch headers, bodystruscture, flags, UID and size - # If success, datas will be of length 2*nb_mails. Each mail is + start = max(0, offset - 1) + page_uids = uid_list[start: start + number_of_mails] + if not page_uids: + return + + # Fetch headers, bodystructure, flags, UID and size + # If success, datas will be of length 2*len(page_uids). Each mail is # a tuple (b'metadata', b'body') and a singular b'BODYSTRUCTURE' - success, datas = self._exec_imap4_method(self.connection.fetch, range_arg, '(BODY.PEEK[HEADER] BODYSTRUCTURE FLAGS UID RFC822.SIZE)') + success, datas = self._exec_imap4_method(self.connection.uid, 'FETCH', ",".join(page_uids), '(BODY.PEEK[HEADER] BODYSTRUCTURE FLAGS UID RFC822.SIZE)') if not success: - if isinstance(datas[0], str) and "Invalid messageset" in datas[0]: - #Goes here means our range args is wrong, it should'nt happen - raise BugException(f"Try to fetch mail with an unvalid messageset: {range_arg}") raise RequestException(f"Fail to fetch mails: {datas}", err.ERROR_IMAP_FAILED) + mails_by_uid: dict[str, dict] = {} # Iterate from the end, two items at a time for i in range(len(datas) - 1, -1, -2): pair = datas[i-1:i+1] # Get the current and previous item @@ -1417,7 +1541,13 @@ def fetch_all_mails_without_content(self, folder_path: str, number_of_mails: int message_parts = cast(tuple[bytes, bytes], pair[0]) has_attachment = self._parse_body_structure_for_attachment(bodystruct) #b'1 (FLAGS (\\Draft) UID 47 RFC822.SIZE 74732 BODY[HEADER] {1080} - yield self._parse_mail_without_content_fetching(message_parts, has_attachment) + mail_dict = self._parse_mail_without_content_fetching(message_parts, has_attachment) + mails_by_uid[str(mail_dict["uid"])] = mail_dict + + for uid in page_uids: + mail_dict_tmp = mails_by_uid.get(uid) + if mail_dict_tmp is not None: + yield mail_dict_tmp else: raise BugException("Not authenticated meaning self.connect() and self.login() was not called beforehands") @@ -1648,7 +1778,9 @@ def copy_mail_to_mailbox(self, folder_path: str, mail_uid: str|list[str], dest_f :param type: bool, default to False :raises RequestException: If the operation fails. """ - logger_imap.debug("Copying mail UID '%s' from '%s' to '%s'", mail_uid, folder_path, dest_folder_path) + print("HAAAAAA") + print("Copying mail UID '%s' from '%s' to '%s'", mail_uid, folder_path, dest_folder_path) + logger_imap.info("Copying mail UID '%s' from '%s' to '%s'", mail_uid, folder_path, dest_folder_path) if self.connection is not None and self.authenticated: if not folder_path.isascii() or not dest_folder_path.isascii(): raise RequestException(f"Mailbox name is not ascii: {folder_path} and/or {dest_folder_path}", err.ERROR_IMAP_NOT_ASCII) @@ -1976,6 +2108,260 @@ def delete_mail_permanently_from_folder_type(self, folder_type: str, mail_uid: s folder_path = self.folders_map_type_to_name[folder_type] self.delete_mails_by_uid(folder_path, mail_uid, move_to_trash=False, permanently=True) + def build_search_criteria(self, search_params: dict, deleted: bool) -> str: + """Build an IMAP SEARCH criteria string from the generic search_params dict. + + Each populated field produces one independent search-key "group". Groups are + then combined using ``search_params["operator"]``: + + ``to`` matches a mail whose ``To`` *or* ``Cc`` header contains the address + (OR-ed across the two headers). ``bcc`` only matches against the ``Bcc`` + header. + + * "AND" (default): groups are simply space-joined (IMAP's implicit AND). + * "OR": groups are combined with a right-nested IMAP ``OR`` operator, so that + a mail matches if it satisfies *any* of the provided criteria. + + When ``deleted`` is False, ``NOT DELETED`` is a system-level filter (not a + user search criterion) and is therefore always AND-ed in regardless of the + operator. When ``deleted`` is True, no filter on the deleted flag is applied + at all, so mails are matched whether or not they are flagged \\Deleted. + + ``size`` filters by mail size using the native IMAP ``LARGER``/``SMALLER`` search + keys: operator ">" maps to ``LARGER`` and "<" maps to ``SMALLER``. ``value`` is + converted to bytes according to ``unit`` (kb/mb/gb, binary multiples of 1024). + + ``date_range.start``/``date_range.end`` accept either a full ISO 8601 timestamp + or a bare date (``YYYY-MM-DD``). IMAP's ``SINCE``/``BEFORE`` only compare dates + (time is ignored), so a bare ``start`` date is already inclusive of that whole + day. A bare ``end`` date is made inclusive of that whole day by searching + ``BEFORE`` the following day. + + :param search_params: Validated search parameters (from MailboxSearchSchema). + :type search_params: dict + :param deleted: If False, mails flagged \\Deleted are excluded from the + results. If True, they are included alongside non-deleted mails (no + filtering on the deleted flag). + :type deleted: bool + :raises RequestException: If a date value cannot be parsed. + :return: IMAP SEARCH criteria string (e.g. "(NOT DELETED SUBJECT \"foo\")" or "ALL"). + :rtype: str + """ + operator = search_params.get("operator") or "AND" + field_groups: list[str] = [] + + if search_params.get("text"): + field_groups.append(f'TEXT "{search_params["text"]}"') + + if search_params.get("from_"): + escaped = escape_imap_string(search_params["from_"]) + field_groups.append(f'FROM "{escaped}"') + + if search_params.get("to"): + escaped = escape_imap_string(search_params["to"]) + field_groups.append(f'(OR TO "{escaped}" CC "{escaped}")') + + if search_params.get("bcc"): + escaped = escape_imap_string(search_params["bcc"]) + field_groups.append(f'BCC "{escaped}"') + + if search_params.get("subject"): + field_groups.append(f'SUBJECT "{search_params["subject"]}"') + + if search_params.get("is_read") is True: + field_groups.append("SEEN") + elif search_params.get("is_read") is False: + field_groups.append("UNSEEN") + + if search_params.get("is_flagged") is True: + field_groups.append("FLAGGED") + elif search_params.get("is_flagged") is False: + field_groups.append("UNFLAGGED") + + if search_params.get("has_attachment") is True: + field_groups.append('HEADER Content-Type "multipart/mixed"') + + if search_params.get("labels"): + label_parts = [f'KEYWORD "{label}"' for label in search_params["labels"]] + field_groups.append(_group_imap_search_parts(label_parts)) + + if search_params.get("size"): + size = search_params["size"] + unit_multiplier = {"kb": 1024, "mb": 1024 ** 2, "gb": 1024 ** 3}[size.get("unit") or "kb"] + size_in_bytes = size["value"] * unit_multiplier + size_keyword = "LARGER" if size["operator"] == ">" else "SMALLER" + field_groups.append(f"{size_keyword} {size_in_bytes}") + + if search_params.get("date_range"): + date_range = search_params["date_range"] + date_parts: list[str] = [] + if date_range.get("start"): + try: + dt = datetime.fromisoformat(date_range["start"].replace("Z", "+00:00")) + date_parts.append(f'SINCE {dt.strftime("%d-%b-%Y")}') + except (ValueError, AttributeError) as exc: + raise RequestException( + f"Invalid start date: {date_range['start']}", + err.ERROR_MAIL_SEARCH_INVALID_DATE + ) from exc + if date_range.get("end"): + end_str = date_range["end"] + try: + dt = datetime.fromisoformat(end_str.replace("Z", "+00:00")) + except (ValueError, AttributeError) as exc: + raise RequestException( + f"Invalid end date: {end_str}", + err.ERROR_MAIL_SEARCH_INVALID_DATE + ) from exc + if "T" not in end_str: + # Date-only end (no time given): IMAP BEFORE excludes the given day, + # so push to the next day to make the end date itself inclusive. + dt = dt + timedelta(days=1) + date_parts.append(f'BEFORE {dt.strftime("%d-%b-%Y")}') + if date_parts: + field_groups.append(_group_imap_search_parts(date_parts)) + + combined = _combine_imap_search_or(field_groups) if operator == "OR" else " ".join(field_groups) + + criteria_parts: list[str] = [] if deleted else ["NOT DELETED"] + if combined: + criteria_parts.append(combined) + + return "(" + " ".join(criteria_parts) + ")" if criteria_parts else "ALL" + + def _search_uids_in_folder(self, folder_path: str, criteria: str) -> str | None: + """Execute an IMAP SEARCH in a single folder and return the UID set string, or None if no results. + + :param folder_path: IMAP folder path to search in. + :type folder_path: str + :param criteria: IMAP SEARCH criteria string. + :type criteria: str + :raises RequestException: If the SEARCH command fails. + :raises BugException: If not authenticated. + :return: Space-separated UID string, or None if no matches. + :rtype: str | None + """ + if self.connection is None or not self.authenticated: + raise BugException("Not authenticated meaning self.connect() and self.login() was not called beforehands") + + if not folder_path.isascii(): + raise RequestException(f"Mailbox name is not ascii: {folder_path}", err.ERROR_IMAP_NOT_ASCII) + + try: + self.select_mailbox(folder_path, readonly=True) + except RequestException: + logger_imap.warning("Folder '%s' not found or not selectable, skipping", folder_path) + return None + + success, datas = self._exec_imap4_method(self.connection.uid, 'SEARCH', criteria) + if not success: + raise RequestException( + f"IMAP SEARCH failed in folder '{folder_path}' with criteria: {criteria}", + err.ERROR_MAIL_SEARCH_FAILED + ) + + if not datas or not datas[0]: + return None + + uid_set = datas[0].decode().strip() + return uid_set if uid_set else None + + def search_mails_without_content(self, folders: list[str], criteria: str) -> Iterator[tuple[str, dict]]: + """Execute an IMAP SEARCH with the given criteria string on each folder and + fetch the matching mails (headers only, no body content). + + Yields tuples of (folder_path, mail_dict) for every matching mail across + all requested folders. ``mail_dict`` has the same shape as + ``_parse_mail_without_content_fetching`` output, enriched with + ``"folder"`` (the IMAP folder path). + + :param folders: List of IMAP folder paths to search in. + :type folders: list[str] + :param criteria: IMAP SEARCH criteria string + :type criteria: str + :raises RequestException: If a SEARCH or FETCH command fails. + :raises BugException: If not authenticated. + :return: Yields (folder_path, mail_dict) tuples. + :rtype: Iterator[tuple[str, dict]] + """ + logger_imap.debug("Searching mails (without content) in folders %s with criteria: %s", folders, criteria) + if self.connection is None or not self.authenticated: + raise BugException("Not authenticated meaning self.connect() and self.login() was not called beforehands") + + for folder_path in folders: + uid_set = self._search_uids_in_folder(folder_path, criteria) + if uid_set is None: + continue + + # Fetch headers + bodystructure for all matching UIDs in one round-trip + success, fetch_datas = self._exec_imap4_method( + self.connection.uid, 'FETCH', uid_set.replace(' ', ','), + '(BODY.PEEK[HEADER] BODYSTRUCTURE FLAGS UID RFC822.SIZE)' + ) + if not success: + raise RequestException( + f"IMAP FETCH failed in folder '{folder_path}' for UIDs {uid_set}", + err.ERROR_MAIL_SEARCH_FAILED + ) + + for i in range(len(fetch_datas) - 1, -1, -2): + pair = fetch_datas[i - 1:i + 1] + if len(pair) < 2: + continue + bodystruct = pair[1] + message_parts = cast(tuple[bytes, bytes], pair[0]) + if not isinstance(message_parts, tuple): + continue + has_attachment = self._parse_body_structure_for_attachment(bodystruct) + mail_dict = self._parse_mail_without_content_fetching(message_parts, has_attachment) + mail_dict["folder"] = folder_path + yield folder_path, mail_dict + + def search_mails_with_content(self, folders: list[str], criteria: str) -> Iterator[tuple[str, dict]]: + """Execute an IMAP SEARCH with the given criteria string on each folder and + fetch the matching mails with full body content. + + Yields tuples of (folder_path, mail_dict) for every matching mail across + all requested folders. ``mail_dict`` has the same shape as + ``_parse_mail_with_content_fetching`` output, enriched with + ``"folder"`` (the IMAP folder path). + + :param folders: List of IMAP folder paths to search in. + :type folders: list[str] + :param criteria: IMAP SEARCH criteria string + :type criteria: str + :raises RequestException: If a SEARCH or FETCH command fails. + :raises BugException: If not authenticated. + :return: Yields (folder_path, mail_dict) tuples. + :rtype: Iterator[tuple[str, dict]] + """ + logger_imap.debug("Searching mails (with content) in folders %s with criteria: %s", folders, criteria) + if self.connection is None or not self.authenticated: + raise BugException("Not authenticated meaning self.connect() and self.login() was not called beforehands") + + for folder_path in folders: + uid_set = self._search_uids_in_folder(folder_path, criteria) + if uid_set is None: + continue + + # Fetch full body for all matching UIDs in one round-trip + success, fetch_datas = self._exec_imap4_method( + self.connection.uid, 'FETCH', uid_set.replace(' ', ','), + '(BODY.PEEK[] FLAGS UID)' + ) + if not success: + raise RequestException( + f"IMAP FETCH failed in folder '{folder_path}' for UIDs {uid_set}", + err.ERROR_MAIL_SEARCH_FAILED + ) + + for part in fetch_datas: + if not isinstance(part, tuple): + continue + mail_dict = self._parse_mail_with_content_fetching(part) + mail_dict["folder"] = folder_path + yield folder_path, mail_dict + def logout(self) -> None: """ Log out from the IMAP server. diff --git a/app/manager/mail/ClientMailServer.py b/app/manager/mail/ClientMailServer.py index 79fd8ef2..840bf2b2 100644 --- a/app/manager/mail/ClientMailServer.py +++ b/app/manager/mail/ClientMailServer.py @@ -2,6 +2,8 @@ from typing import Any, Iterator from email.message import EmailMessage +from app.utils import constants as cs + class ClientMailServer(metaclass=ABCMeta): """ Abstract class for mail clients. @@ -14,6 +16,69 @@ def __init__(self) -> None: self.connected = False self.authenticated = False + @staticmethod + def parse_fields_param(fields: str | None, fields_action: str | None) -> dict[str, bool]: + """Parse the generic "fields"/"fields_action" query params (see CollectionPaginateArgs) + into flags telling the caller how mails should be fetched. + + This centralizes the handling of every field name that impacts *how* mails are + fetched from the mail server (as opposed to fields that are simply stripped from + the already-built response). For now one field name is recognized: + + * ``"contents"``: whether the mail content (body, attachments, ...) should be fetched. + Fetching content is a heavy operation, both for the mail server and for this API. + + ``"contents"`` is present by default (opt-out): with ``fields_action == "include"`` it + must be explicitly listed to stay on; with ``fields_action == "exclude"`` it is turned + off only if listed. + + When ``fields`` is empty/None, the default value is used: fetch content. + + :param fields: Comma separated list of field names, or None. + :type fields: str | None + :param fields_action: "include" or "exclude". + :type fields_action: str | None + :return: dict with key "with_content" + :rtype: dict[str, bool] + """ + requested: set[str] = set(fields.split(",")) if fields else set() + + if not requested: + return {"with_content": True} + + if fields_action == "include": + with_content = cs.MAIL_FIELD_CONTENTS in requested + else: + with_content = cs.MAIL_FIELD_CONTENTS not in requested + + return {"with_content": with_content} + + @abstractmethod + def build_search_criteria(self, search_params: dict, deleted: bool) -> Any: + """Build a protocol-specific search criteria object/string from the generic, + protocol-agnostic ``search_params`` dict (as validated by MailboxSearchSchema). + + This keeps every bit of protocol-specific search syntax (IMAP SEARCH syntax, + JMAP filter objects, ...) confined to the concrete client implementation, so + that callers (e.g. ModuleMail) stay protocol agnostic. + + :param search_params: Validated search parameters (from MailboxSearchSchema), + with keys like "text", "from_", "to" (matches To or Cc), "bcc", "subject", + "is_read", "is_flagged", "has_attachment", "labels", "date_range", "size" + (dict with "value", "operator" ">"/"<" and "unit" kb/mb/gb), and + "operator" ("AND"/"OR", controlling how the other criteria are combined). + :type search_params: dict + :param deleted: If False, mails flagged as deleted are excluded from the + results. If True, they are included alongside non-deleted mails (no + filtering on the deleted flag). + :type deleted: bool + :raises RequestException: If a value in search_params cannot be translated + (e.g. an invalid date). + :return: A protocol-specific criteria value to pass to search_mails_with_content + / search_mails_without_content. + :rtype: Any + """ + @abstractmethod def connect(self) -> None: """Connect to the mail server.""" @@ -143,12 +208,12 @@ def delete_acl(self, folder_path: str, identifier: str) -> None: """ @abstractmethod - def fetch_all_mails_with_content(self, folder_path: str, number_of_mails: int, offset: int) -> Iterator[dict]: + def fetch_all_mails_with_content(self, folder_path: str, number_of_mails: int, offset: int, deleted: bool = False) -> Iterator[dict]: """ https://datatracker.ietf.org/doc/html/rfc9051#name-fetch-response Fetch a specific number of mails from a mailbox with full details. - First yield the total number of mails into the folder: + First yield the total number of mails matching the ``deleted`` filter: {"nb_mails": 500} If not 0, yield a dict for each mail, from most recent to oldest { @@ -164,13 +229,18 @@ def fetch_all_mails_with_content(self, folder_path: str, number_of_mails: int, o :type number_of_mails: int :param offset: The offset of the mail to fetch. :type number_of_mails: int + :param deleted: If False (default), mails flagged \\Deleted are excluded. If + True, they are included alongside non-deleted mails (no filtering on the + deleted flag). This is applied as a search criterion before pagination, + so the page always contains up to ``number_of_mails`` matching mails. + :type deleted: bool :raises RequestException: If fetching mails fails :return: A tuple of (list of mail dicts with full details, total count) :rtype: tuple[list[dict[str, Any]], int] """ @abstractmethod - def fetch_all_mails_without_content(self, folder_path: str, number_of_mails: int, offset: int) -> Iterator[dict]: + def fetch_all_mails_without_content(self, folder_path: str, number_of_mails: int, offset: int, deleted: bool = False) -> Iterator[dict]: """ https://datatracker.ietf.org/doc/html/rfc9051#name-fetch-response Fetch a specific number of mails from a mailbox with full details. @@ -183,7 +253,7 @@ def fetch_all_mails_without_content(self, folder_path: str, number_of_mails: int "has_attachment": bool } - Always yield the total number of mails of the folder + Always yield the total number of mails matching the ``deleted`` filter Then yield mail by mail, from the most recent to the oldest :param mailbox: The mailbox to fetch mails from. @@ -192,6 +262,11 @@ def fetch_all_mails_without_content(self, folder_path: str, number_of_mails: int :type number_of_mails: int :param offset: The offset of the mail to fetch. :type number_of_mails: int + :param deleted: If False (default), mails flagged \\Deleted are excluded. If + True, they are included alongside non-deleted mails (no filtering on the + deleted flag). This is applied as a search criterion before pagination, + so the page always contains up to ``number_of_mails`` matching mails. + :type deleted: bool :raises RequestException: If fetching mails fails :return: A tuple of (list of mail dicts with full details, total count) :rtype: tuple[list[dict[str, Any]], int] @@ -349,4 +424,45 @@ def get_quota(self) -> dict[str, Any] | None: "storage_limit": int, # storage limit in KB (0 if unlimited) } :rtype: dict[str, Any] | None - """ \ No newline at end of file + """ + + @abstractmethod + def search_mails_without_content(self, folders: list[str], criteria: str) -> Iterator[tuple[str, dict]]: + """Search mails (headers only, no body) across the given folders. + + Yields tuples of (folder_path, mail_dict) for every matching mail. + + :param folders: List of folder paths to search in. + :type folders: list[str] + :param criteria: IMAP SEARCH criteria string. + :type criteria: str + :return: Yields (folder_path, mail_dict) tuples. + :rtype: Iterator[tuple[str, dict]] + """ + + @abstractmethod + def search_mails_with_content(self, folders: list[str], criteria: str) -> Iterator[tuple[str, dict]]: + """Search mails with full body content across the given folders. + + Yields tuples of (folder_path, mail_dict) for every matching mail. + + :param folders: List of folder paths to search in. + :type folders: list[str] + :param criteria: IMAP SEARCH criteria string. + :type criteria: str + :return: Yields (folder_path, mail_dict) tuples. + :rtype: Iterator[tuple[str, dict]] + """ + + @abstractmethod + def get_folder_with_subfolders(self, folder_path: str, include_subfolders: bool = True) -> list[str]: + """Return the given folder path, optionally followed by the paths of all its subfolders. + + :param folder_path: The folder to start from. + :type folder_path: str + :param include_subfolders: If True, also list every subfolder (at any depth) below folder_path. + :type include_subfolders: bool + :return: List of folder paths, folder_path first. + :rtype: list[str] + :raises RequestException: If not connected to the server. + """ diff --git a/app/module/mail/ModuleMail.py b/app/module/mail/ModuleMail.py index a5d9ee16..49a19558 100644 --- a/app/module/mail/ModuleMail.py +++ b/app/module/mail/ModuleMail.py @@ -22,7 +22,7 @@ from app.utils.maths.crypto_utils import decrypt_password from app.utils.module.importManager import import_and_instantiate_manager from app.utils.logger.logger import logger_mail_server -from app.utils.strings import get_imap_config_from_url, get_domain_from_mail, get_domain_from_contact +from app.utils.strings import get_imap_config_from_url, get_domain_from_mail, get_domain_from_contact, encode_imap_tag, decode_imap_tag from app.utils.constants import DELETE_MAIL_BEHAVIOR_MAP if TYPE_CHECKING: @@ -654,7 +654,7 @@ def _parse_mail(self, mail_dict:dict) -> dict: "answered": flags_dict.get('answered', False), "forwarded": flags_dict.get('forwarded', False), "deleted": flags_dict.get('deleted', False), - "flags": flags_dict.get('all', []), + "flags": [decode_imap_tag(flag) for flag in flags_dict.get('all', [])], "to": to, "from": from_, "cc": cc, @@ -673,7 +673,7 @@ def _parse_mail(self, mail_dict:dict) -> dict: "mail_type_data": mail_type_data } - def get_folder_mails(self, account_id: str, folder_name: str, collection_param: CollectionPaginateArgs) -> tuple[list[dict[str, Any]], int]: + def get_folder_mails(self, account_id: str, folder_name: str, collection_param: CollectionPaginateArgs, deleted: bool = False) -> tuple[list[dict[str, Any]], int]: """Retrieve a list of mails in a specific folder with full details. :param folder_name: The name of the folder to fetch mails from. @@ -682,6 +682,11 @@ def get_folder_mails(self, account_id: str, folder_name: str, collection_param: :type first: int :param last: The ending index for pagination (exclusive). :type last: int + :param deleted: If False (default), mails flagged as deleted are excluded. If + True, they are included alongside non-deleted mails (no filtering on the + deleted flag). Applied as an IMAP search criterion before pagination, so + the returned page is never short because of it. + :type deleted: bool :raises RequestException: If fetching mails fails :return: A tuple of (list of mail dicts with full details, total mail count) :rtype: tuple[list[dict[str, Any]], int] @@ -690,18 +695,14 @@ def get_folder_mails(self, account_id: str, folder_name: str, collection_param: offset = collection_param.first_item + 1 #First mail is index 1 not zero nb_mails = collection_param.last_item - collection_param.first_item + 1 logger_mail_server.info("Try to fetch %s mails with offset %s", nb_mails, offset) - mail_iter: Iterator|None = None - without_content = False - if collection_param.fields: - requested = collection_param.fields.split(",") - if collection_param.fields_action == "include" and "contents" not in requested: - without_content = True - mail_iter = client.fetch_all_mails_without_content(folder_name, number_of_mails=nb_mails, offset=offset) - if collection_param.fields_action == "exclude" and "contents" in requested: - without_content = True - mail_iter = client.fetch_all_mails_without_content(folder_name, number_of_mails=nb_mails, offset=offset) - if mail_iter is None: - mail_iter = client.fetch_all_mails_with_content(folder_name, number_of_mails=nb_mails, offset=offset) + + fields_params = client.parse_fields_param(collection_param.fields, collection_param.fields_action) + without_content = not fields_params["with_content"] + + if without_content: + mail_iter = client.fetch_all_mails_without_content(folder_name, number_of_mails=nb_mails, offset=offset, deleted=deleted) + else: + mail_iter = client.fetch_all_mails_with_content(folder_name, number_of_mails=nb_mails, offset=offset, deleted=deleted) total_count = next(mail_iter)["nb_mails"] mails = [] @@ -738,6 +739,126 @@ def _get_delete_behavior(self) -> tuple[bool, bool]: delete_behavior: str = mail_general_prefs["SOGO_U_MAIL_DELETE_BEHAVIOR"] return DELETE_MAIL_BEHAVIOR_MAP.get(delete_behavior, (True, True)) + @staticmethod + def _flatten_selectable_folder_paths(folders: list[dict[str, Any]]) -> list[str]: + """Recursively flatten a folder tree (as returned by ClientImap.list_folders) into a flat + list of selectable folder paths, including every nested subfolder at any depth. + + :param folders: List of folder dicts, each possibly containing nested "children" folders. + :type folders: list[dict[str, Any]] + :return: Flat list of selectable folder paths. + :rtype: list[str] + """ + paths: list[str] = [] + for folder in folders: + if folder.get(cs.FOLDER_SELECTABLE, True): + paths.append(folder[cs.FOLDER_PATH]) + children = folder.get(cs.FOLDER_CHILDREN) or [] + if children: + paths.extend(ModuleMail._flatten_selectable_folder_paths(children)) + return paths + + def search_mails(self, account_id: str, search_params: dict, collection_param: CollectionPaginateArgs, deleted: bool = False) -> tuple[list[dict[str, Any]], int]: + """Execute an advanced search across one or multiple folders. + + Delegates the building of the protocol-specific search criteria (IMAP SEARCH + syntax, JMAP filter, ...) to the mail client, queries each requested folder + and returns a paginated list of matching mails together with the total count. + + :param account_id: The account identifier. + :type account_id: str + :param search_params: Validated search parameters (from MailboxSearchSchema). + :type search_params: dict + :param collection_param: Pagination, sorting and filtering parameters. + :type collection_param: CollectionPaginateArgs + :param deleted: If False (default), mails flagged as deleted are excluded. If + True, they are included alongside non-deleted mails (no filtering on the + deleted flag). Applied as part of the IMAP search criteria, not as a + post-fetch filter. + :type deleted: bool + :return: A tuple of (list of mail dicts, total count). + :rtype: tuple[list[dict[str, Any]], int] + :raises RequestException: If mail server operations fail or search_params are invalid. + """ + client = self._open_client_for(account_id) + + # --- Determine content handling from generic fields param --- + fields_params = client.parse_fields_param(collection_param.fields, collection_param.fields_action) + without_content = not fields_params["with_content"] + + # --- Build the protocol-specific search criteria (delegated to the client) --- + criteria = client.build_search_criteria(search_params, deleted) + + # --- Determine folders to search --- + folder_list = search_params.get("folders") or [] + if not folder_list or folder_list == ["all"]: + raw_folders = client.list_folders() + folders_to_search = self._flatten_selectable_folder_paths(raw_folders) + else: + include_subfolders = search_params.get("include_subfolders", True) + folders_to_search = [] + seen_folders: set[str] = set() + for folder in folder_list: + for folder_path in client.get_folder_with_subfolders(folder, include_subfolders): + if folder_path not in seen_folders: + seen_folders.add(folder_path) + folders_to_search.append(folder_path) + + # --- Determine if content should be fetched --- + mail_iter = ( + client.search_mails_without_content(folders_to_search, criteria) + if without_content + else client.search_mails_with_content(folders_to_search, criteria) + ) + + # --- Collect all results --- + all_mails: list[dict] = [] + for _folder_path, mail_dict in mail_iter: + parsed = self._parse_mail(mail_dict) + parsed["folder"] = mail_dict.get("folder", _folder_path) + if without_content: + parsed.pop("contents", None) + parsed.pop("attachments", None) + parsed.pop("certificates", None) + parsed.pop("mail_type_data", None) + all_mails.append(parsed) + + # --- Post-filter by attachment_type (IMAP has no direct extension filter) --- + # Applied as an additional AND filter regardless of search_params["operator"], + # since it runs after the mail-server search rather than as part of the IMAP criteria. + if search_params.get("attachment_type"): + ext_filter = {e.lower() for e in search_params["attachment_type"]} + all_mails = [ + m for m in all_mails + if any( + att.get("extension", "").lower() in ext_filter + for att in m.get("attachments", []) + ) + ] + + # --- Sort --- + sort_by: str = collection_param.sort_by or "date" + sort_order: str = collection_param.sort_order or "desc" + reverse = sort_order == "desc" + + sort_key_map: dict[str, Any] = { + "date": lambda m: m.get("date", ""), + "sender": lambda m: m.get("from", {}).get("email", ""), + "subject": lambda m: m.get("subject", ""), + "size": lambda m: m.get("size", 0), + "relevance": lambda m: m.get("date", ""), # fallback: by date TODO: what?? + } + key_fn = sort_key_map.get(sort_by, sort_key_map["date"]) + all_mails.sort(key=key_fn, reverse=reverse) + + # --- Paginate --- + total = len(all_mails) + offset = collection_param.first_item + nb_mails = collection_param.page_size + page = all_mails[offset: offset + nb_mails] + + return page, total + def delete_mails(self, account_id:str, folder_path: str, mail_uids: str|list[str]) -> None: """Delete multiple mails by UIDs in a single client session. @@ -1488,6 +1609,10 @@ def perform_mail_action(self, account_id:str, folder_name: str, mail_uid: str, a return self._action_copy(client, folder_name, mail_uid, data) elif action == "delete": return self._action_delete(client, folder_name, mail_uid, account_id=account_id) + elif action == "illegal": + return self._action_illegal(client, folder_name, mail_uid) + elif action == "phishing": + return self._action_phishing(client, folder_name, mail_uid) else: raise RequestException(f"Invalid action: {action}", err.ERROR_INVALID_ACTION) @@ -1525,9 +1650,44 @@ def perform_mail_batch_action(self, account_id: str, folder_name: str, batch_act return self._action_copy(client, folder_name, mail_uids, data) elif action == "delete": return self._action_delete(client, folder_name, mail_uids, account_id=account_id) + elif action == "illegal": + return self._action_illegal(client, folder_name, mail_uids) + elif action == "phishing": + return self._action_phishing(client, folder_name, mail_uids) else: raise RequestException(f"Invalid action: {action}", err.ERROR_INVALID_ACTION) + def perform_mailbox_batch_action(self, account_id: str, batch_action_data: dict) -> dict[str, Any]: + """Perform an action on multiple mails spanning multiple folders of the same account. + + Loops over ``perform_mail_batch_action`` for each folder listed in ``uids``. A failure on + one folder is recorded in ``errors`` but does not prevent the remaining folders from being + processed. + + :param account_id: The account identifier + :type account_id: str + :param batch_action_data: dictionary containing 'uids' (folder name -> list of uids), + 'action' and optional 'data' fields + :type batch_action_data: dict[str, Any] + :return: Dict with the action, the per-folder results, and the per-folder errors + :rtype: dict[str, Any] + """ + action: str = batch_action_data["action"] + data = batch_action_data.get("data") + uids_by_folder: dict = batch_action_data["uids"] + + results: dict[str, Any] = {} + errors: dict[str, str] = {} + + for folder_name, uids in uids_by_folder.items(): + try: + results[folder_name] = self.perform_mail_batch_action(account_id, folder_name, {"uids": uids, "action": action, "data": data}) + except RequestException as ex: + logger_mail_server.warning("perform_mailbox_batch_action: action '%s' failed for folder '%s': %s", action, folder_name, str(ex)) + errors[folder_name] = ex.error.c + + return {"action": action, "results": results, "errors": errors} + def download_attachment(self, account_id: str, folder_name: str, mail_uid: str, filename: str) -> tuple[bytes, str]: """Download a specific attachment from a mail. @@ -1592,7 +1752,10 @@ def _action_tag(self, client: ClientMailServer, folder_name: str, mail_uid: str| else: raise RequestException("Tags must be a string or list of strings", err.ERROR_MISSING_ACTION_DATA) - client.add_flags_to_mail(folder_name, mail_uid, tag_list) + # IMAP flags are atoms and cannot contain spaces/special chars; encode (reversibly) before sending + encoded_tags = [encode_imap_tag(tag) for tag in tag_list] + + client.add_flags_to_mail(folder_name, mail_uid, encoded_tags) return {"action": "tag", "mail_uid": mail_uid, "tags_added": tag_list} @@ -1620,7 +1783,10 @@ def _action_untag(self, client: ClientMailServer, folder_name: str, mail_uid: st else: raise RequestException("Tags must be a string or list of strings", err.ERROR_MISSING_ACTION_DATA) - client.remove_flags_to_mail(folder_name, mail_uid, tag_list) + # IMAP flags are atoms and cannot contain spaces/special chars; encode (reversibly) before sending + encoded_tags = [encode_imap_tag(tag) for tag in tag_list] + + client.remove_flags_to_mail(folder_name, mail_uid, encoded_tags) return {"action": "untag", "mail_uid": mail_uid, "tags_removed": tag_list} @@ -1679,6 +1845,40 @@ def _action_ham(self, client: ClientMailServer, folder_name: str, mail_uid: str| return {"action": "ham", "mail_uid": mail_uid, "moved_to": inbox_folder} + def _action_illegal(self, client: ClientMailServer, folder_name: str, mail_uid: str|list[str]) -> dict[str, Any]: + """Report a mail or a list of mails as illegal content, copy them to the Junk folder + and permanently remove them (no Trash copy) from their source folder. + + :param folder_name: The name of the folder + :type folder_name: str + :param mail_uid: The unique identifier of the mail, or a list of them + :type mail_uid: str|list[str] + :return: Result with illegal action info + :rtype: dict[str, Any] + :raises RequestException: If operation fails + """ + junk_folder = self.domain_mail_folder_name.get(cs.MAIL_FOLDER_JUNK, "Junk") + client.copy_mail_to_mailbox(folder_name, mail_uid, junk_folder, create_dest=True) + client.delete_mails_by_uid(folder_name, mail_uid, move_to_trash=False, permanently=True) + return {"action": "illegal", "mail_uid": mail_uid, "moved_to": junk_folder} + + def _action_phishing(self, client: ClientMailServer, folder_name: str, mail_uid: str|list[str]) -> dict[str, Any]: + """Report a mail or a list of mails as phishing, copy them to the Junk folder + and permanently remove them (no Trash copy) from their source folder. + + :param folder_name: The name of the folder + :type folder_name: str + :param mail_uid: The unique identifier of the mail, or a list of them + :type mail_uid: str|list[str] + :return: Result with phishing action info + :rtype: dict[str, Any] + :raises RequestException: If operation fails + """ + junk_folder = self.domain_mail_folder_name.get(cs.MAIL_FOLDER_JUNK, "Junk") + client.copy_mail_to_mailbox(folder_name, mail_uid, junk_folder, create_dest=True) + client.delete_mails_by_uid(folder_name, mail_uid, move_to_trash=False, permanently=True) + return {"action": "phishing", "mail_uid": mail_uid, "moved_to": junk_folder} + def _action_copy(self, client: ClientMailServer, folder_name: str, mail_uid: str|list[str], destination: Any) -> dict[str, Any]: """Copy a mail or a list of mails to another folder. diff --git a/app/utils/api/paginate_sort_filter.py b/app/utils/api/paginate_sort_filter.py index 348bb1f6..879d0ecd 100644 --- a/app/utils/api/paginate_sort_filter.py +++ b/app/utils/api/paginate_sort_filter.py @@ -118,6 +118,25 @@ def make_paginator(self, data: dict, **kwargs: dict) -> CollectionPaginateArgs: +class DeletedFilterQueryArgsSchema(Schema): + """Deserializes the "deleted" query param used by mail-listing endpoints. + + Kept separate from the "fields" include/exclude query param (see + ``collection_pagination_parameters``) because, unlike a display field, it must be + applied as a search criterion before pagination: a post-fetch filter could yield + fewer results than the requested page size. + """ + + class Meta: + """Set behavior for unknown param""" + unknown = EXCLUDE + + deleted = ma_fields.Boolean( + load_default=False, + metadata={"description": "If false (default), exclude mails flagged \\Deleted. If true, include them alongside non-deleted mails."} + ) + + class PaginationMetadataSchema(Schema): """Pagination metadata schema diff --git a/app/utils/constants.py b/app/utils/constants.py index 31f24c37..439f9a2e 100644 --- a/app/utils/constants.py +++ b/app/utils/constants.py @@ -113,6 +113,9 @@ # tmp_draft TMP_DRAFT_KEY_SIZE = 32 # Length of the unique hash key for a tmp_draft entry +# Mail "fields" query param values (used by collection_paginate on mail list/search endpoints) +MAIL_FIELD_CONTENTS = "contents" # Include/exclude the mail content (body, attachments, ...) + # Mail deletion behavior mapping to (move_to_trash, permanently) flags DELETE_MAIL_BEHAVIOR_MAP = { # behavior move_to_trash permanently diff --git a/app/utils/errors.py b/app/utils/errors.py index 5ccdaabf..07563efa 100644 --- a/app/utils/errors.py +++ b/app/utils/errors.py @@ -173,6 +173,10 @@ def __init__(self, c:str, m:str, h:int = HTTPStatus.INTERNAL_SERVER_ERROR): ERROR_SIEVE_PUSH_FAILED = E("S001508", "Failed To Push Filters To Sieve", HTTPStatus.INTERNAL_SERVER_ERROR) ERROR_SIEVE_CAPABILITY_NOT_FOUND = E("S001509", "Sieve capability not found in server response", HTTPStatus.INTERNAL_SERVER_ERROR) +#Search +ERROR_MAIL_SEARCH_FAILED = E("S000338", "IMAP search command failed", HTTPStatus.INTERNAL_SERVER_ERROR) +ERROR_MAIL_SEARCH_INVALID_DATE = E("S000339", "Invalid date format in search parameters, expected ISO 8601", HTTPStatus.BAD_REQUEST) + #Quota ERROR_IMAP_QUOTA_NOT_SUPPORTED = E("S000336", "IMAP server does not support QUOTA extension", HTTPStatus.NOT_IMPLEMENTED) ERROR_IMAP_QUOTA_FAILED = E("S000337", "IMAP GETQUOTAROOT command failed", HTTPStatus.INTERNAL_SERVER_ERROR) diff --git a/app/utils/strings.py b/app/utils/strings.py index d9423695..9120b3b2 100644 --- a/app/utils/strings.py +++ b/app/utils/strings.py @@ -1,3 +1,4 @@ +import base64 import re import unicodedata @@ -188,9 +189,70 @@ def imap_join_folders(delimiter: str, first_path: str, second_path: str) -> str: second_path = second_path[1:-1] return quote(f"{first_path}{delimiter}{second_path}") +# Prefix marking a tag as base32-encoded. Kept short and IMAP-atom-safe (letters/digits only) +# so it never collides with a plain user tag that happens to look like base32. +_IMAP_TAG_ENCODED_PREFIX = "B32-" + + +def encode_imap_tag(tag: str) -> str: + """Encode a user-provided tag into a value that is safe to use as an IMAP flag/keyword. + + Per RFC 3501, a flag is an "atom" and cannot contain spaces, control characters or any of + the special chars ( ) { % * " \\ ] plus SP and CTL. IMAP servers (Dovecot included) will + otherwise silently split on whitespace, turning a single tag like "test avec espace" into + three distinct flags ("test", "avec", "espace"). + + To keep the round-trip lossless (spaces, accents, underscores, punctuation...), the tag is + base32-encoded (padding stripped) and prefixed with a marker. Base32 only produces + ``[A-Z2-7]`` characters, which are always valid IMAP atom characters. + + Tags that are already plain IMAP-safe atoms (letters/digits/._- only, no spaces) are + returned unchanged to keep flags human-readable on the wire when possible. + System flags (starting with '\\', e.g. \\Seen, \\Deleted) are always returned unchanged. + + :param tag: The raw tag value to encode. + :type tag: str + :return: A value safe to use as a single IMAP flag. + :rtype: str + """ + if tag.startswith('\\'): + return tag + if re.fullmatch(r'[A-Za-z0-9._-]+', tag): + return tag + encoded = base64.b32encode(tag.encode('utf-8')).decode('ascii').rstrip('=') + return _IMAP_TAG_ENCODED_PREFIX + encoded + + +def decode_imap_tag(flag: str) -> str: + """Decode an IMAP flag/keyword previously encoded with :func:`encode_imap_tag`. + + Flags that don't carry the encoding prefix (system flags, or plain tags that were kept + as-is because they were already IMAP-safe) are returned unchanged. + + :param flag: The IMAP flag value as received from the server. + :type flag: str + :return: The original, human-readable tag value. + :rtype: str + """ + if not flag.startswith(_IMAP_TAG_ENCODED_PREFIX): + return flag + encoded = flag[len(_IMAP_TAG_ENCODED_PREFIX):] + padding = '=' * (-len(encoded) % 8) + try: + return base64.b32decode(encoded + padding).decode('utf-8') + except (ValueError, UnicodeDecodeError): + # Not actually one of our encoded tags (unlikely collision); return as-is. + return flag + + def string_to_sort_score(s: str) -> int: """Convert a string to an integer score for sorting purposes.""" score = 0 for char in s: score = (score << 8) | ord(char) # Décalage de 8 bits pour chaque caractère return score + +def escape_imap_string(value: str) -> str: + """Escape special characters for IMAP protocol.""" + # RFC 9051: Escape backslashes and double quotes + return value.replace("\\", "\\\\").replace('"', '\\"') diff --git a/tests/test_interface/test_mail/test_InterfaceApiMailMail.py b/tests/test_interface/test_mail/test_InterfaceApiMailMail.py index de5fe5a1..e493cad1 100644 --- a/tests/test_interface/test_mail/test_InterfaceApiMailMail.py +++ b/tests/test_interface/test_mail/test_InterfaceApiMailMail.py @@ -46,9 +46,9 @@ def __init__(self): self.perform_mail_batch_action_args = None self.perform_mail_batch_action_result = {"action": "tag", "mail_uid": [42, 43], "tags_added": ["Important"]} - def get_folder_mails(self, account_id, folder_name, collection_param): + def get_folder_mails(self, account_id, folder_name, collection_param, deleted=False): """Fetch a list of mails from a folder.""" - self.get_folder_mails_args = (account_id, folder_name, collection_param.first_item, collection_param.last_item) + self.get_folder_mails_args = (account_id, folder_name, collection_param.first_item, collection_param.last_item, deleted) return self.get_folder_mails_result def get_mail_detail(self, account_id, folder_name, mail_uid): @@ -104,7 +104,7 @@ def test_get_mail_list_success(): assert status_code == 200 assert total == 100 assert result["data"] == [{"uid": 1, "subject": "Test"}] - assert fake_module.get_folder_mails_args == (0, "INBOX", 0, 10) + assert fake_module.get_folder_mails_args == (0, "INBOX", 0, 10, False) def test_get_mail_list_module_exception(): """Test error handling when module raises RequestException.""" diff --git a/tests/test_interface/test_mail/test_InterfaceApiMailMailbox.py b/tests/test_interface/test_mail/test_InterfaceApiMailMailbox.py index 4c61a409..9dd03fe0 100644 --- a/tests/test_interface/test_mail/test_InterfaceApiMailMailbox.py +++ b/tests/test_interface/test_mail/test_InterfaceApiMailMailbox.py @@ -762,3 +762,179 @@ def save_mail_to_folder(self, account_id, message, folder): result, status_code = interface.send_mail(account_id="abc123", mail_data=mail_data) assert status_code == 200 +# ========== Tests for search_mailbox ========== + +def create_interface_with_search(monkeypatch, fake_module, allow_external=True, + search_result=None, search_total=0, + search_raises=None): + """Helper to create interface with a FakeModuleMail that supports search_mails.""" + patch_module_on_interface(monkeypatch, fake_module) + + class FakeUserModuleSettings: + def __init__(self, data): + self.SOGO_D_ALLOW_EXT_MAIL_ACCOUNT = allow_external + + monkeypatch.setattr( + "app.interface.mail.InterfaceApiMailMailbox.UserModuleSettingsObj", + FakeUserModuleSettings + ) + + class FakeMailSettings: + def __init__(self, data): + pass + + monkeypatch.setattr( + "app.interface.mail.InterfaceApiMailMailbox.MailSettingsObj", + FakeMailSettings + ) + + _search_result = search_result if search_result is not None else [] + _search_total = search_total + _search_raises = search_raises + + class FakeModuleMail: + def __init__(self, user, mail_settings, process_setting=None): + self.search_mails_args = None + + def get_mailbox_quota(self, account_id): + return None + + def search_mails(self, account_id, search_params, collection_param, deleted=False): + self.search_mails_args = (account_id, search_params, collection_param, deleted) + if _search_raises is not None: + raise _search_raises + return _search_result, _search_total + + fake_mail_module_instance = FakeModuleMail.__new__(FakeModuleMail) + fake_mail_module_instance.search_mails_args = None + + class FakeModuleMailTracked(FakeModuleMail): + """Tracked version that exposes the instance for assertions.""" + _instance = None + def __init__(self, user, mail_settings, process_setting=None): + super().__init__(user, mail_settings, process_setting) + FakeModuleMailTracked._instance = self + + monkeypatch.setattr( + "app.interface.mail.InterfaceApiMailMailbox.ModuleMail", + FakeModuleMailTracked + ) + + # Patch ModuleMailOutgoing to avoid real instantiation + class FakeModuleMailOutgoing: + def __init__(self, user, mail_settings): + pass + + monkeypatch.setattr( + "app.interface.mail.InterfaceApiMailMailbox.ModuleMailOutgoing", + FakeModuleMailOutgoing + ) + + process_setting = FakeProcessSetting() + user = FakeUser() + user_domain = {"USER_MODULE_SETTINGS": {}, "MAIL_SETTINGS": {}} + + interface = InterfaceApiMailMailbox( + process_setting=process_setting, + user=user, + user_domain=user_domain + ) + return interface, FakeModuleMailTracked + + +def _make_collection_param(): + """Build a minimal CollectionPaginateArgs for search tests.""" + from app.utils.api.paginate_sort_filter import CollectionPaginateArgs + return CollectionPaginateArgs(page=1, page_size=10) + + +def test_search_mailbox_main_account_success(monkeypatch): + """Test advanced search on main account returns results.""" + fake_module = FakeModuleUserProfile() + mails = [{"uid": "1", "subject": "Hello"}, {"uid": "2", "subject": "World"}] + interface, tracked = create_interface_with_search( + monkeypatch, fake_module, search_result=mails, search_total=2 + ) + + search_params = {"text": "Hello"} + total, result, status_code = interface.search_mailbox("0", search_params, _make_collection_param()) + + assert status_code == 200 + assert total == 2 + assert result["data"] == mails + + +def test_search_mailbox_empty_results(monkeypatch): + """Test advanced search with no matching mails returns empty list.""" + fake_module = FakeModuleUserProfile() + interface, _ = create_interface_with_search( + monkeypatch, fake_module, search_result=[], search_total=0 + ) + + total, result, status_code = interface.search_mailbox("0", {}, _make_collection_param()) + + assert status_code == 200 + assert total == 0 + assert result["data"] == [] + + +def test_search_mailbox_external_account_forbidden(monkeypatch): + """Test that searching external account when not allowed returns 403.""" + fake_module = FakeModuleUserProfile() + interface, _ = create_interface_with_search( + monkeypatch, fake_module, allow_external=False + ) + + total, result, status_code = interface.search_mailbox("abc123", {}, _make_collection_param()) + + assert status_code == 403 + assert total == 0 + assert result["error_code"] == err.ERROR_EXTERNAL_ACCOUNT_FORBIDDEN.c + + +def test_search_mailbox_external_account_allowed(monkeypatch): + """Test that searching external account when allowed succeeds.""" + fake_module = FakeModuleUserProfile() + mails = [{"uid": "5", "subject": "External mail"}] + interface, _ = create_interface_with_search( + monkeypatch, fake_module, allow_external=True, + search_result=mails, search_total=1 + ) + + total, result, status_code = interface.search_mailbox("abc123", {}, _make_collection_param()) + + assert status_code == 200 + assert total == 1 + assert result["data"] == mails + + +def test_search_mailbox_module_error_returns_error_response(monkeypatch): + """Test that a RequestException from search_mails is caught and returned as error.""" + fake_module = FakeModuleUserProfile() + interface, _ = create_interface_with_search( + monkeypatch, fake_module, + search_raises=RequestException("IMAP error", err.ERROR_IMAP_CONNECTION_FAILED) + ) + + total, result, status_code = interface.search_mailbox("0", {}, _make_collection_param()) + + assert total == 0 + assert status_code >= 400 + assert "error_code" in result + + +def test_search_mailbox_passes_params_to_module(monkeypatch): + """Test that search_params and collection_param are forwarded to module.search_mails.""" + fake_module = FakeModuleUserProfile() + interface, tracked = create_interface_with_search( + monkeypatch, fake_module, search_result=[], search_total=0 + ) + + search_params = {"text": "invoice", "folders": ["INBOX"]} + collection = _make_collection_param() + interface.search_mailbox("0", search_params, collection) + + instance = tracked._instance + assert instance is not None + assert instance.search_mails_args[1] == search_params + assert instance.search_mails_args[2] is collection diff --git a/tests/test_manager/test_mail/test_clientImap.py b/tests/test_manager/test_mail/test_clientImap.py index d0680066..9be804ab 100644 --- a/tests/test_manager/test_mail/test_clientImap.py +++ b/tests/test_manager/test_mail/test_clientImap.py @@ -81,6 +81,7 @@ def __init__(self): b'(\\HasNoChildren) "." "Sent"']) self.expunge_response = ("OK", [b"1", b"2"]) self.uid_response = ("OK", [b""]) + self.uid_search_response = ("OK", [b""]) self.append_response = ("OK", [b""]) self.getacl_response = ("OK", [b"INBOX user1 lrswipkxtea user2 lr"]) self.setacl_response = ("OK", [b""]) @@ -167,6 +168,8 @@ def expunge(self): return self.expunge_response def uid(self, command, *args): + if command == 'SEARCH': + return self.uid_search_response return self.uid_response def fetch(self, message_set, parts): @@ -277,13 +280,10 @@ def test_single_uid(self): assert list(parse_uids_from_bytes(b"42")) == ["42"] def test_multiple_uids(self): - # bytes iteration yields integers; comparison `byte == b' '` never matches, - # so the entire input is returned as one string - assert list(parse_uids_from_bytes(b"1 2 3 4 5")) == ["1 2 3 4 5"] + assert list(parse_uids_from_bytes(b"1 2 3 4 5")) == ["1", "2", "3", "4", "5"] def test_trailing_space(self): - # same reason: no splitting occurs, trailing space is included in the single string - assert list(parse_uids_from_bytes(b"10 20 ")) == ["10 20 "] + assert list(parse_uids_from_bytes(b"10 20 ")) == ["10", "20"] def test_empty_bytes(self): assert list(parse_uids_from_bytes(b"")) == [] @@ -604,7 +604,8 @@ def test_fetch_mails_success(): """Test fetching mails from a mailbox.""" fake_conn = FakeIMAPConnection() fake_conn.select_response = ('OK', [b'10']) - fake_conn.fetch_response = ('OK', [ + fake_conn.uid_search_response = ('OK', [b'100 101']) + fake_conn.uid_response = ('OK', [ (b'1 (UID 100 FLAGS (\\Seen))', b'Subject: Test\r\n\r\nBody'), (b'2 (UID 101 FLAGS ())', b'Subject: Test2\r\n\r\nBody2') ]) @@ -612,7 +613,7 @@ def test_fetch_mails_success(): results = list(client.fetch_all_mails_with_content('INBOX', number_of_mails=2, offset=0)) # First yielded item is the total count dict - assert results[0] == 10 or isinstance(results[0], (int, dict)) + assert results[0] == {"nb_mails": 2} def test_fetch_mail_success(): @@ -946,6 +947,20 @@ def test_not_authenticated_raises_bug_exception(self): with pytest.raises(BugException): client.uid_store_flags("100", ["\\Seen"]) + def test_empty_uid_list_is_noop(self): + """No mails to store flags on should skip the IMAP command entirely, + instead of sending an empty uidset which the server rejects.""" + fake_conn = FakeIMAPConnection() + client = authenticated_client(fake_conn) + count = client.uid_store_flags([], ["\\Deleted"]) + assert count == 0 + + def test_empty_uid_iterator_is_noop(self): + fake_conn = FakeIMAPConnection() + client = authenticated_client(fake_conn) + count = client.uid_store_flags(iter([]), ["\\Deleted"]) + assert count == 0 + def test_uid_store_flags_with_list_sends_single_joined_command(self): """Test that storing flags on a list of UIDs sends a single IMAP UID STORE command with a comma-joined UID set, instead of one command per mail.""" @@ -974,7 +989,8 @@ class TestFetchAllMails: def test_yields_count_dict_first(self): fake_conn = FakeIMAPConnection() fake_conn.select_response = ("OK", [b"2"]) - fake_conn.fetch_response = ( + fake_conn.uid_search_response = ("OK", [b"100 101"]) + fake_conn.uid_response = ( "OK", [ (b"1 (UID 100 FLAGS (\\Seen) BODY[] {10}", b"Subject: A\r\n\r\nA"), @@ -998,6 +1014,72 @@ def test_empty_mailbox_yields_only_count(self): results = list(client.fetch_all_mails_with_content("INBOX", number_of_mails=10, offset=0)) assert results == [{"nb_mails": 0}] + def test_deleted_true_uses_all_search_criteria(self): + """When deleted=True, deleted and non-deleted mails are both fetched (no filter).""" + fake_conn = FakeIMAPConnection() + fake_conn.select_response = ("OK", [b"5"]) + captured_criteria = [] + original_uid = fake_conn.uid + + def tracking_uid(command, *args): + if command == "SEARCH": + captured_criteria.append(args[0]) + return original_uid(command, *args) + + fake_conn.uid = tracking_uid + fake_conn.uid_search_response = ("OK", [b"100"]) + fake_conn.uid_response = ( + "OK", + [(b"1 (UID 100 FLAGS (\\Deleted) BODY[] {10}", b"Subject: A\r\n\r\nA"), b")"], + ) + client = authenticated_client(fake_conn) + + results = list(client.fetch_all_mails_with_content("INBOX", number_of_mails=10, offset=0, deleted=True)) + assert captured_criteria == ["ALL"] + assert results[0] == {"nb_mails": 1} + + def test_deleted_false_uses_not_deleted_search_criteria(self): + fake_conn = FakeIMAPConnection() + fake_conn.select_response = ("OK", [b"5"]) + captured_criteria = [] + original_uid = fake_conn.uid + + def tracking_uid(command, *args): + if command == "SEARCH": + captured_criteria.append(args[0]) + return original_uid(command, *args) + + fake_conn.uid = tracking_uid + client = authenticated_client(fake_conn) + + list(client.fetch_all_mails_with_content("INBOX", number_of_mails=10, offset=0)) + assert captured_criteria == ["NOT DELETED"] + + def test_pagination_does_not_short_change_page_when_filtering(self): + """A page must contain up to number_of_mails matching mails: the filter is + applied by the IMAP SEARCH, not after fetching a fixed sequence-number window.""" + fake_conn = FakeIMAPConnection() + fake_conn.select_response = ("OK", [b"3"]) + # Only 3 of, say, 10 mails in the mailbox are not deleted (already filtered by SEARCH) + fake_conn.uid_search_response = ("OK", [b"10 20 30"]) + fake_conn.uid_response = ( + "OK", + [ + (b"1 (UID 10 FLAGS () BODY[] {5}", b"Subject: A\r\n\r\nA"), + b")", + (b"2 (UID 20 FLAGS () BODY[] {5}", b"Subject: B\r\n\r\nB"), + b")", + (b"3 (UID 30 FLAGS () BODY[] {5}", b"Subject: C\r\n\r\nC"), + b")", + ], + ) + client = authenticated_client(fake_conn) + + results = list(client.fetch_all_mails_with_content("INBOX", number_of_mails=3, offset=1)) + assert results[0] == {"nb_mails": 3} + mail_dicts = [r for r in results[1:] if isinstance(r, dict) and "uid" in r] + assert len(mail_dicts) == 3 + def test_not_authenticated_raises_bug_exception(self): client = make_client() client.connection = None @@ -1166,18 +1248,16 @@ class TestGetMailUidsBeforeDate: def test_returns_uids(self): fake_conn = FakeIMAPConnection() fake_conn.select_response = ("OK", [b"5"]) - fake_conn.uid_response = ("OK", [b"1 2 3"]) + fake_conn.uid_search_response = ("OK", [b"1 2 3"]) client = authenticated_client(fake_conn) uids = list(client.get_mail_uids_before_date("INBOX")) - # parse_uids_from_bytes iterates bytes as integers so space splitting - # never triggers; the whole payload is returned as one string - assert uids == ["1 2 3"] + assert uids == ["1", "2", "3"] def test_with_before_date(self): fake_conn = FakeIMAPConnection() fake_conn.select_response = ("OK", [b"5"]) - fake_conn.uid_response = ("OK", [b"1"]) + fake_conn.uid_search_response = ("OK", [b"1"]) client = authenticated_client(fake_conn) uids = list(client.get_mail_uids_before_date("INBOX", before_date="2024-01-01")) @@ -1421,6 +1501,35 @@ def test_imap_list_folders_not_authenticated_raises(self): with pytest.raises(BugException): list(client._imap_list_folders()) + def test_extended_empty_match_returns_no_folders(self): + """LIST-EXTENDED/LIST-STATUS matching zero mailboxes (e.g. a folder with no + children, or a nonexistent base folder) must yield an empty result, not raise. + + Regression test: this used to raise RequestException(ERROR_IMAP_FAILED), + which surfaced as a 500 when searching folders that don't exist. + """ + fake_conn = FakeIMAPConnection() + client = authenticated_client(fake_conn) + client.capabilities = {"LIST-EXTENDED", "LIST-STATUS"} + fake_conn.response = lambda name: ("OK", [None]) + + folders = list(client._imap_list_folders('"NonExistent.*"')) + assert folders == [] + + def test_extended_non_empty_match_is_parsed(self): + fake_conn = FakeIMAPConnection() + client = authenticated_client(fake_conn) + client.capabilities = {"LIST-EXTENDED", "LIST-STATUS"} + responses = { + "LIST": ("OK", [b'(\\HasNoChildren) "." INBOX.Sent']), + "STATUS": ("OK", [b"INBOX.Sent (MESSAGES 3 UNSEEN 1)"]), + } + fake_conn.response = lambda name: responses[name] + + folders = list(client._imap_list_folders('"INBOX.*"')) + assert len(folders) == 1 + assert folders[0].path == "INBOX.Sent" + # =========================================================================== # Tests: list_folders @@ -1532,13 +1641,18 @@ class TestFetchAllMailsWithoutContent: def test_fetch_all_mails_without_content_success(self): fake_conn = FakeIMAPConnection() fake_conn.select_response = ("OK", [b"2"]) + fake_conn.uid_search_response = ("OK", [b"100 101"]) fake_conn.uid_response = ("OK", [ - b"1 (UID 100 FLAGS (\\Seen) RFC822.SIZE 1000)", - b"2 (UID 101 FLAGS () RFC822.SIZE 2000)", + (b"1 (UID 100 FLAGS (\\Seen) RFC822.SIZE 1000 BODY[HEADER] {20}", b"Subject: A\r\n\r\n"), + b'BODYSTRUCTURE ("TEXT" "PLAIN" NIL NIL NIL "7BIT" 5 1 NIL NIL NIL)', + (b"2 (UID 101 FLAGS () RFC822.SIZE 2000 BODY[HEADER] {20}", b"Subject: B\r\n\r\n"), + b'BODYSTRUCTURE ("TEXT" "PLAIN" NIL NIL NIL "7BIT" 5 1 NIL NIL NIL)', ]) client = authenticated_client(fake_conn) results = list(client.fetch_all_mails_without_content("INBOX", number_of_mails=2, offset=0)) - assert len(results) > 0 + assert results[0] == {"nb_mails": 2} + mail_dicts = [r for r in results[1:] if isinstance(r, dict) and "uid" in r] + assert len(mail_dicts) == 2 def test_fetch_all_mails_without_content_not_authenticated_raises(self): client = make_client() @@ -1546,6 +1660,37 @@ def test_fetch_all_mails_without_content_not_authenticated_raises(self): with pytest.raises(BugException): list(client.fetch_all_mails_without_content("INBOX", number_of_mails=5, offset=0)) + def test_deleted_true_uses_all_search_criteria(self): + fake_conn = FakeIMAPConnection() + fake_conn.select_response = ("OK", [b"5"]) + captured_criteria = [] + original_uid = fake_conn.uid + + def tracking_uid(command, *args): + if command == "SEARCH": + captured_criteria.append(args[0]) + return original_uid(command, *args) + + fake_conn.uid = tracking_uid + fake_conn.uid_search_response = ("OK", [b"100"]) + fake_conn.uid_response = ("OK", [ + (b"1 (UID 100 FLAGS (\\Deleted) RFC822.SIZE 1000 BODY[HEADER] {20}", b"Subject: A\r\n\r\n"), + b'BODYSTRUCTURE ("TEXT" "PLAIN" NIL NIL NIL "7BIT" 5 1 NIL NIL NIL)', + ]) + client = authenticated_client(fake_conn) + + results = list(client.fetch_all_mails_without_content("INBOX", number_of_mails=10, offset=0, deleted=True)) + assert captured_criteria == ["ALL"] + assert results[0] == {"nb_mails": 1} + + def test_empty_search_result_yields_only_count(self): + fake_conn = FakeIMAPConnection() + fake_conn.select_response = ("OK", [b"0"]) + client = authenticated_client(fake_conn) + + results = list(client.fetch_all_mails_without_content("INBOX", number_of_mails=10, offset=0)) + assert results == [{"nb_mails": 0}] + # =========================================================================== # Tests: fetch_mails_by_uids @@ -1663,3 +1808,82 @@ def test_is_folder_subscribed_not_authenticated_raises(self): client.connection = None with pytest.raises(BugException): client._is_folder_subscribed("INBOX") + + +class TestBuildSearchCriteria: + """Tests for ClientImap.build_search_criteria, in particular the AND/OR operator.""" + + def test_default_operator_is_and(self): + client = make_client() + criteria = client.build_search_criteria( + {"subject": "Projet X", "from_": "a@b.com"}, deleted=False + ) + assert criteria == '(NOT DELETED FROM "a@b.com" SUBJECT "Projet X")' + + def test_explicit_and_operator_same_as_default(self): + client = make_client() + criteria = client.build_search_criteria( + {"operator": "AND", "subject": "Projet X", "from_": "a@b.com"}, deleted=False + ) + assert criteria == '(NOT DELETED FROM "a@b.com" SUBJECT "Projet X")' + + def test_or_operator_combines_two_fields(self): + client = make_client() + criteria = client.build_search_criteria( + {"operator": "OR", "subject": "Projet X", "from_": "a@b.com"}, deleted=False + ) + assert criteria == '(NOT DELETED OR FROM "a@b.com" SUBJECT "Projet X")' + + def test_or_operator_combines_more_than_two_fields(self): + client = make_client() + criteria = client.build_search_criteria( + {"operator": "OR", "subject": "Projet X", "from_": "a@b.com", "is_read": False}, + deleted=False, + ) + assert criteria == '(NOT DELETED OR FROM "a@b.com" (OR SUBJECT "Projet X" UNSEEN))' + + def test_or_operator_with_single_field_has_no_or_keyword(self): + client = make_client() + criteria = client.build_search_criteria( + {"operator": "OR", "subject": "Projet X"}, deleted=False + ) + assert criteria == '(NOT DELETED SUBJECT "Projet X")' + + def test_deleted_true_applies_no_deleted_filter_regardless_of_operator(self): + client = make_client() + criteria = client.build_search_criteria( + {"operator": "OR", "subject": "Projet X", "from_": "a@b.com"}, deleted=True + ) + assert criteria == '(OR FROM "a@b.com" SUBJECT "Projet X")' + + def test_or_operator_combines_to_with_another_field(self): + client = make_client() + criteria = client.build_search_criteria( + {"operator": "OR", "to": "x@y.com", "subject": "Projet X"}, + deleted=False, + ) + assert criteria == ( + '(NOT DELETED OR (OR TO "x@y.com" CC "x@y.com") SUBJECT "Projet X")' + ) + + def test_to_field_matches_to_or_cc_header(self): + client = make_client() + criteria = client.build_search_criteria( + {"to": "x@y.com"}, deleted=False + ) + assert criteria == '(NOT DELETED (OR TO "x@y.com" CC "x@y.com"))' + + def test_bcc_field(self): + client = make_client() + criteria = client.build_search_criteria( + {"bcc": "x@y.com"}, deleted=False + ) + assert criteria == '(NOT DELETED BCC "x@y.com")' + + def test_no_user_criteria_and_deleted_true_returns_all(self): + client = make_client() + assert client.build_search_criteria({}, deleted=True) == "ALL" + + def test_no_user_criteria_still_applies_not_deleted(self): + client = make_client() + assert client.build_search_criteria({}, deleted=False) == "(NOT DELETED)" diff --git a/tests/test_module/test_mail/test_moduleMail.py b/tests/test_module/test_mail/test_moduleMail.py index d7bbbcd5..8eab4508 100644 --- a/tests/test_module/test_mail/test_moduleMail.py +++ b/tests/test_module/test_mail/test_moduleMail.py @@ -55,6 +55,24 @@ def list_folders(self): def get_one_folder(self, folder_path): return self.get_one_folder_result + def get_folder_with_subfolders(self, folder_path, include_subfolders=True): + """Return the given folder path, optionally followed by the paths of all its subfolders. + + For the fake client, this just returns [folder_path] unless specific subfolders + are configured in the test via get_folder_with_subfolders_result. + + :param folder_path: The folder to start from + :type folder_path: str + :param include_subfolders: If True, also list every subfolder below folder_path + :type include_subfolders: bool + :return: List of folder paths + :rtype: list[str] + """ + if hasattr(self, 'get_folder_with_subfolders_result'): + return self.get_folder_with_subfolders_result + # Default: just return the folder itself + return [folder_path] + def create_folder(self, folder_name, parent_path=None): self.create_folder_calls.append((folder_name, parent_path)) if self.create_folder_result is not None: @@ -74,7 +92,7 @@ def purge_folder(self, folder_path, before_date=None, do_children=False, permane # ---- mail methods ---- - def fetch_all_mails_with_content(self, folder_name, number_of_mails, offset=0): + def fetch_all_mails_with_content(self, folder_name, number_of_mails, offset=0, deleted=False): """Returns an iterator: first item has {'nb_mails': int}, then mail dicts.""" yield {'nb_mails': 0} @@ -146,7 +164,7 @@ def list_mailboxes_detailed(self): {'name': 'Sent', 'path': 'Sent'} ] - def fetch_all_mails_without_content(self, mailbox, number_of_mails, offset=0): + def fetch_all_mails_without_content(self, mailbox, number_of_mails, offset=0, deleted=False): """Fetch all mails from a mailbox without content (used by get_folder_mails).""" yield {'nb_mails': 0} @@ -158,6 +176,82 @@ def delete_mail_permanently_from_folder_type(self, folder_type, uid): """Delete a mail permanently from a folder type.""" pass + def search_mails_with_content(self, folders, criteria): + """Search mails with full content across folders. Yields (folder_path, mail_dict).""" + return iter(self.search_mails_result if hasattr(self, 'search_mails_result') else []) + + def search_mails_without_content(self, folders, criteria): + """Search mails without body content across folders. Yields (folder_path, mail_dict).""" + return iter(self.search_mails_result if hasattr(self, 'search_mails_result') else []) + + def build_search_criteria(self, search_params, deleted): + """Build search criteria from params (simplified for fake client). + + This is called by ModuleMail to convert generic search params into + protocol-specific criteria. The fake implementation just returns + a simple string representation. + """ + from datetime import datetime + + # For the fake client, we just build a simple string + # In reality, this would be IMAP-specific syntax + criteria_parts = [] + + if search_params.get("text"): + criteria_parts.append(f"TEXT:{search_params['text']}") + + if search_params.get("from"): + criteria_parts.append(f"FROM:{search_params['from']}") + + if search_params.get("to"): + criteria_parts.append(f"TO:{search_params['to']}") + + if search_params.get("subject"): + criteria_parts.append(f"SUBJECT:{search_params['subject']}") + + if search_params.get("date_range"): + dr = search_params["date_range"] + if dr.get("start"): + # Validate start date format (YYYY-MM-DD) + try: + datetime.strptime(dr['start'], '%Y-%m-%d') + except ValueError: + raise RequestException(f"Invalid start date format: {dr['start']}, expected YYYY-MM-DD") + criteria_parts.append(f"SINCE:{dr['start']}") + if dr.get("end"): + # Validate end date format (YYYY-MM-DD) + try: + datetime.strptime(dr['end'], '%Y-%m-%d') + except ValueError: + raise RequestException(f"Invalid end date format: {dr['end']}, expected YYYY-MM-DD") + criteria_parts.append(f"BEFORE:{dr['end']}") + + if not deleted: + criteria_parts.append("NOT_DELETED") + + return " ".join(criteria_parts) if criteria_parts else "ALL" + + @staticmethod + def parse_fields_param(fields, fields_action): + """Parse the generic "fields"/"fields_action" query params into flags. + + This is a static method from ClientMailServer base class that handles: + - "contents": whether to fetch mail content (heavy operation) + """ + from app.utils import constants as cs + + requested = set(fields.split(",")) if fields else set() + + if not requested: + return {"with_content": True} + + if fields_action == "include": + with_content = cs.MAIL_FIELD_CONTENTS in requested + else: + with_content = cs.MAIL_FIELD_CONTENTS not in requested + + return {"with_content": with_content} + def _make_email_message(subject='Test', from_='sender@example.com', to='recipient@example.com', @@ -290,7 +384,7 @@ def test_get_folder_mails_success(monkeypatch): mail1 = _make_email_message(subject='Test1') mail2 = _make_email_message(subject='Test2') - def fetch_all(folder_name, number_of_mails, offset=0): + def fetch_all(folder_name, number_of_mails, offset=0, deleted=False): yield {'nb_mails': 100} yield {'uid': '1', 'mail': mail1, 'flags': {'seen': True, 'flagged': False, 'answered': False, 'forwarded': False, 'deleted': False, 'all': ['\\Seen']}, 'size': 120} yield {'uid': '2', 'mail': mail2, 'flags': {'seen': False, 'flagged': False, 'answered': False, 'forwarded': False, 'deleted': False, 'all': []}, 'size': 120} @@ -309,7 +403,7 @@ def test_get_folder_mails_empty_folder(monkeypatch): """Test getting mails from empty folder.""" module, fake_client = _make_module(monkeypatch) - def fetch_all(folder_name, number_of_mails, offset=0): + def fetch_all(folder_name, number_of_mails, offset=0, deleted=False): yield {'nb_mails': 0} fake_client.fetch_all_mails_with_content = fetch_all @@ -1072,7 +1166,7 @@ def test_get_folder_mails_without_content_include_filter(monkeypatch): mail1 = _make_email_message(subject='Test1') - def fetch_all_without_content(mailbox, number_of_mails, offset=0): + def fetch_all_without_content(mailbox, number_of_mails, offset=0, deleted=False): yield {'nb_mails': 50} yield {'uid': '1', 'mail': mail1, 'flags': {'seen': True, 'flagged': False, 'answered': False, 'forwarded': False, 'deleted': False, 'all': ['\\Seen']}, 'size': 120} @@ -1099,7 +1193,7 @@ def test_get_folder_mails_without_content_exclude_filter(monkeypatch): mail1 = _make_email_message(subject='Test1') - def fetch_all_without_content(mailbox, number_of_mails, offset=0): + def fetch_all_without_content(mailbox, number_of_mails, offset=0, deleted=False): yield {'nb_mails': 25} yield {'uid': '1', 'mail': mail1, 'flags': {'seen': False, 'flagged': False, 'answered': False, 'forwarded': False, 'deleted': False, 'all': []}, 'size': 120} @@ -1190,3 +1284,316 @@ def get_acl_after_share(folder_path): identifiers = [call[1] for call in fake_client.set_acl_calls] assert 'user1@example.com' in identifiers assert 'user2@example.com' in identifiers +# =========================================================================== +# Tests: search_mails +# =========================================================================== + +def _make_search_mail_dict(uid='1', subject='Test', from_='sender@example.com', + to='recipient@example.com', seen=False, folder='INBOX'): + """Build a mail dict as returned by search_mails_with_content.""" + msg = _make_email_message(subject=subject, from_=from_, to=to) + return (folder, { + 'uid': uid, + 'mail': msg, + 'flags': { + 'seen': seen, + 'flagged': False, + 'answered': False, + 'forwarded': False, + 'deleted': False, + 'all': ['\\Seen'] if seen else [] + }, + 'size': 200, + 'folder': folder, + }) + + +def test_search_mails_returns_matching_results(monkeypatch): + """Test basic search returns parsed mail list and total count.""" + module, fake_client = _make_module(monkeypatch) + fake_client.search_mails_result = [ + _make_search_mail_dict(uid='1', subject='Hello'), + _make_search_mail_dict(uid='2', subject='World'), + ] + + params = {"text": "Hello"} + collection = CollectionPaginateArgs(page=1, page_size=10) + result, total = module.search_mails(ACCOUNT_ID, params, collection) + + assert total == 2 + assert len(result) == 2 + assert result[0]['subject'] == 'Hello' + assert result[1]['subject'] == 'World' + + +def test_search_mails_empty_results(monkeypatch): + """Test search with no matching mails returns empty list.""" + module, fake_client = _make_module(monkeypatch) + fake_client.search_mails_result = [] + + params = {"text": "nonexistent"} + collection = CollectionPaginateArgs(page=1, page_size=10) + result, total = module.search_mails(ACCOUNT_ID, params, collection) + + assert total == 0 + assert result == [] + + +def test_search_mails_empty_params_searches_all(monkeypatch): + """Test search with no criteria still runs and returns results.""" + module, fake_client = _make_module(monkeypatch) + fake_client.search_mails_result = [ + _make_search_mail_dict(uid='10', subject='Any mail'), + ] + + params = {} + collection = CollectionPaginateArgs(page=1, page_size=10) + result, total = module.search_mails(ACCOUNT_ID, params, collection) + + assert total == 1 + assert result[0]['subject'] == 'Any mail' + + +def test_search_mails_pagination(monkeypatch): + """Test search respects pagination (page_size).""" + module, fake_client = _make_module(monkeypatch) + fake_client.search_mails_result = [ + _make_search_mail_dict(uid=str(i), subject=f'Mail {i}') + for i in range(5) + ] + + collection = CollectionPaginateArgs(page=1, page_size=3) + result, total = module.search_mails(ACCOUNT_ID, {}, collection) + + assert total == 5 + assert len(result) == 3 + + +def test_search_mails_pagination_second_page(monkeypatch): + """Test search returns correct slice for page 2.""" + module, fake_client = _make_module(monkeypatch) + fake_client.search_mails_result = [ + _make_search_mail_dict(uid=str(i), subject=f'Mail {i}') + for i in range(5) + ] + + collection = CollectionPaginateArgs(page=2, page_size=3) + result, total = module.search_mails(ACCOUNT_ID, {}, collection) + + assert total == 5 + assert len(result) == 2 # only 2 items remain on page 2 + + +def test_search_mails_sort_by_subject_asc(monkeypatch): + """Test search sorts results by subject ascending.""" + module, fake_client = _make_module(monkeypatch) + fake_client.search_mails_result = [ + _make_search_mail_dict(uid='1', subject='Zebra'), + _make_search_mail_dict(uid='2', subject='Apple'), + _make_search_mail_dict(uid='3', subject='Mango'), + ] + + collection = CollectionPaginateArgs(page=1, page_size=10, sort_by='subject', sort_order='asc') + result, total = module.search_mails(ACCOUNT_ID, {}, collection) + + assert total == 3 + assert result[0]['subject'] == 'Apple' + assert result[1]['subject'] == 'Mango' + assert result[2]['subject'] == 'Zebra' + + +def test_search_mails_sort_by_subject_desc(monkeypatch): + """Test search sorts results by subject descending.""" + module, fake_client = _make_module(monkeypatch) + fake_client.search_mails_result = [ + _make_search_mail_dict(uid='1', subject='Apple'), + _make_search_mail_dict(uid='2', subject='Zebra'), + ] + + collection = CollectionPaginateArgs(page=1, page_size=10, sort_by='subject', sort_order='desc') + result, _ = module.search_mails(ACCOUNT_ID, {}, collection) + + assert result[0]['subject'] == 'Zebra' + assert result[1]['subject'] == 'Apple' + + +def test_search_mails_specific_folders(monkeypatch): + """Test search against specific folder list does not call list_folders.""" + module, fake_client = _make_module(monkeypatch) + fake_client.search_mails_result = [ + _make_search_mail_dict(uid='1', subject='In Sent', folder='Sent'), + ] + list_folders_called = [] + original_list = fake_client.list_folders + fake_client.list_folders = lambda: list_folders_called.append(1) or original_list() + + params = {"folders": ["Sent"]} + collection = CollectionPaginateArgs(page=1, page_size=10) + result, total = module.search_mails(ACCOUNT_ID, params, collection) + + assert len(list_folders_called) == 0 # list_folders must NOT be called + assert total == 1 + assert result[0]['folder'] == 'Sent' + + +def test_search_mails_folders_all_calls_list_folders(monkeypatch): + """Test search with folders=['all'] calls list_folders to enumerate all folders.""" + module, fake_client = _make_module(monkeypatch) + fake_client.search_mails_result = [] + fake_client.list_folders_result = [ + {'name': 'INBOX', 'path': 'INBOX', 'selectable': True}, + {'name': 'Sent', 'path': 'Sent', 'selectable': True}, + ] + list_folders_called = [] + original_list = fake_client.list_folders + def tracking_list(): + list_folders_called.append(1) + return original_list() + fake_client.list_folders = tracking_list + + params = {"folders": ["all"]} + collection = CollectionPaginateArgs(page=1, page_size=10) + module.search_mails(ACCOUNT_ID, params, collection) + + assert len(list_folders_called) == 1 + + +def test_search_mails_without_content_when_contents_excluded(monkeypatch): + """Test that search_mails_without_content is used when 'contents' is excluded via fields.""" + module, fake_client = _make_module(monkeypatch) + fake_client.search_mails_result = [ + _make_search_mail_dict(uid='1', subject='Test'), + ] + without_content_called = [] + original = fake_client.search_mails_without_content + def tracking_without(folders, criteria): + without_content_called.append(1) + return original(folders, criteria) + fake_client.search_mails_without_content = tracking_without + + # fields_action='include' + 'contents' not in fields => without_content=True + collection = CollectionPaginateArgs(page=1, page_size=10, fields='subject,from', fields_action='include') + module.search_mails(ACCOUNT_ID, {}, collection) + + assert len(without_content_called) == 1 + + +def test_search_mails_with_content_when_contents_included(monkeypatch): + """Test that search_mails_with_content is used when 'contents' is explicitly included.""" + module, fake_client = _make_module(monkeypatch) + fake_client.search_mails_result = [ + _make_search_mail_dict(uid='1', subject='Test'), + ] + with_content_called = [] + original = fake_client.search_mails_with_content + def tracking_with(folders, criteria): + with_content_called.append(1) + return original(folders, criteria) + fake_client.search_mails_with_content = tracking_with + + # fields_action='include' + 'contents' in fields => with_content + collection = CollectionPaginateArgs(page=1, page_size=10, fields='subject,contents', fields_action='include') + module.search_mails(ACCOUNT_ID, {}, collection) + + assert len(with_content_called) == 1 + + +def test_search_mails_invalid_start_date_raises(monkeypatch): + """Test that an invalid start date in date_range raises RequestException.""" + module, fake_client = _make_module(monkeypatch) + fake_client.search_mails_result = [] + + params = {"date_range": {"start": "not-a-date"}} + collection = CollectionPaginateArgs(page=1, page_size=10) + + with pytest.raises(RequestException): + module.search_mails(ACCOUNT_ID, params, collection) + + +def test_search_mails_invalid_end_date_raises(monkeypatch): + """Test that an invalid end date in date_range raises RequestException.""" + module, fake_client = _make_module(monkeypatch) + fake_client.search_mails_result = [] + + params = {"date_range": {"end": "not-a-date"}} + collection = CollectionPaginateArgs(page=1, page_size=10) + + with pytest.raises(RequestException): + module.search_mails(ACCOUNT_ID, params, collection) + + +def test_search_mails_valid_date_range(monkeypatch): + """Test that a valid date_range does not raise and returns results.""" + module, fake_client = _make_module(monkeypatch) + fake_client.search_mails_result = [ + _make_search_mail_dict(uid='1', subject='Recent mail'), + ] + + params = {"date_range": {"start": "2024-01-01", "end": "2024-12-31"}} + collection = CollectionPaginateArgs(page=1, page_size=10) + result, total = module.search_mails(ACCOUNT_ID, params, collection) + + assert total == 1 + assert result[0]['subject'] == 'Recent mail' + + +def _make_email_message_with_pdf_attachment(subject='Has PDF'): + """Build a multipart email message with a real PDF attachment.""" + from email.mime.multipart import MIMEMultipart + from email.mime.text import MIMEText + from email.mime.base import MIMEBase + from email import encoders + + msg = MIMEMultipart() + msg['Subject'] = subject + msg['From'] = 'sender@example.com' + msg['To'] = 'recipient@example.com' + msg['Date'] = 'Mon, 1 Jan 2024 10:00:00 +0000' + + msg.attach(MIMEText('Body content', 'plain')) + + pdf_part = MIMEBase('application', 'pdf') + pdf_part.set_payload(b'%PDF-1.4 fake pdf content') + encoders.encode_base64(pdf_part) + pdf_part.add_header('Content-Disposition', 'attachment', filename='doc.pdf') + msg.attach(pdf_part) + + return msg + + +def test_search_mails_attachment_type_filter(monkeypatch): + """Test post-filtering by attachment_type extension.""" + module, fake_client = _make_module(monkeypatch) + + msg_with_pdf = _make_email_message_with_pdf_attachment(subject='Has PDF') + msg_no_pdf = _make_email_message(subject='No PDF') + + flags = {'seen': False, 'flagged': False, 'answered': False, + 'forwarded': False, 'deleted': False, 'all': []} + + fake_client.search_mails_result = [ + ('INBOX', {'uid': '1', 'mail': msg_with_pdf, 'flags': flags, 'size': 300, 'folder': 'INBOX'}), + ('INBOX', {'uid': '2', 'mail': msg_no_pdf, 'flags': flags, 'size': 100, 'folder': 'INBOX'}), + ] + + params = {"attachment_type": ["pdf"]} + collection = CollectionPaginateArgs(page=1, page_size=10) + result, total = module.search_mails(ACCOUNT_ID, params, collection) + + # Only the mail that has a pdf attachment should survive the post-filter + assert total == 1 + assert result[0]['subject'] == 'Has PDF' + + +def test_search_mails_client_error_propagates(monkeypatch): + """Test that a RequestException from the client propagates.""" + module, fake_client = _make_module(monkeypatch) + + def failing_search(folders, criteria): + raise RequestException("IMAP search failed") + + fake_client.search_mails_with_content = failing_search + + collection = CollectionPaginateArgs(page=1, page_size=10) + with pytest.raises(RequestException, match="IMAP search failed"): + module.search_mails(ACCOUNT_ID, {}, collection)