diff --git a/modoboa/sievefilters/api/v2/vloaders.py b/modoboa/sievefilters/api/v2/vloaders.py index 72914a5f3..bffdb707c 100644 --- a/modoboa/sievefilters/api/v2/vloaders.py +++ b/modoboa/sievefilters/api/v2/vloaders.py @@ -2,7 +2,7 @@ from typing import List, Optional, TypedDict -from modoboa.sievefilters import imaputils +from modoboa.webmail.lib import imaputils class UserMailbox(TypedDict): @@ -13,7 +13,7 @@ class UserMailbox(TypedDict): def __build_folders_list( - folders: dict, user, imapc: imaputils.IMAPconnector, parentmb: Optional[str] = None + folders: list, user, imapc: imaputils.IMAPconnector, parentmb: Optional[str] = None ) -> List[UserMailbox]: ret: List[UserMailbox] = [] for fd in folders: diff --git a/modoboa/sievefilters/imaputils.py b/modoboa/sievefilters/imaputils.py deleted file mode 100644 index 215daf79d..000000000 --- a/modoboa/sievefilters/imaputils.py +++ /dev/null @@ -1,212 +0,0 @@ -"""IMAPv4 utilities.""" - -import base64 -import imaplib -import re -from operator import itemgetter -import ssl - -from django.utils.translation import gettext as _ - -from modoboa.lib import imap_utf7 # NOQA -from modoboa.lib.exceptions import ModoboaException, InternalError -from modoboa.parameters import tools as param_tools - - -class ImapError(ModoboaException): - - def __init__(self, reason): - self.reason = reason - - def __str__(self): - return str(self.reason) - - -class IMAPconnector: - """The IMAPv4 connector.""" - - list_base_pattern = r'\((?P.*?)\) "(?P.*)" "?(?P[^"]*)"?' - list_response_pattern_literal = re.compile( - r'\((?P.*?)\) "(?P.*)" \{(?P\d+)\}' - ) - list_response_pattern = re.compile(list_base_pattern) - listextended_response_pattern = re.compile( - list_base_pattern + r"\s*(?P.*)" - ) - - def __init__(self, user=None, password=None): - self.__hdelimiter = None - self.conf = dict(param_tools.get_global_parameters("sievefilters")) - self.address = self.conf["imap_server"] - self.port = self.conf["imap_port"] - self.login(user, password) - - def _cmd(self, name, *args, **kwargs): - """ - IMAP command wrapper. - - To simplify errors handling, this wrapper calls the - appropriate method (``uid`` or FIXME) and then check the - return code. If an error has occured, an ``ImapError`` - exception is raised. - - For specific commands commands (FETCH, ...), the result is - parsed using the IMAPclient module before being returned. - - :param name: the command's name - :return: the command's result - """ - try: - typ, data = self.m._simple_command(name, *args) - except imaplib.IMAP4.error as e: - raise ImapError(e) from None - if typ == "NO": - raise ImapError(data) from None - if "responses" not in kwargs: - if name not in self.m.untagged_responses: - return None - return self.m.untagged_responses.pop(name) - res = [] - for r in kwargs["responses"]: - if r not in self.m.untagged_responses: - return None - res.append(self.m.untagged_responses.pop(r)) - return res - - @property - def hdelimiter(self): - """Return the default hierachy delimiter. - - This is a simple way to retrieve the default delimiter (see - http://www.imapwiki.org/ClientImplementation/MailboxList). - - :return: a string - """ - if self.__hdelimiter is None: - data = self._cmd("LIST", '""', '""') - m = self.list_response_pattern.match(data[0].decode()) - if m is None: - raise InternalError(_("Failed to retrieve hierarchy delimiter")) - self.__hdelimiter = m.group("delimiter") - return self.__hdelimiter - - def login(self, user: str, passwd: str): - """ - Custom login method. - - Only OAUTHBEARER mechanism is supported here. passwd parameter is actually a - Bearer token. - """ - try: - if self.conf["imap_secured"]: - self.m = imaplib.IMAP4_SSL(self.address, self.port) - else: - self.m = imaplib.IMAP4(self.address, self.port) - except (OSError, imaplib.IMAP4.error, ssl.SSLError) as error: - raise ImapError(_(f"Connection to IMAP server failed: {error}")) from None - - token = ( - b"n,a=" - + user.encode("utf-8") - + b",\001auth=Bearer " - + passwd.encode("utf-8") - + b"\001\001" - ) - token = base64.b64encode(token) - data = self._cmd("AUTHENTICATE", b"OAUTHBEARER", token) - self.m.state = "AUTH" - if "CAPABILITY" in self.m.untagged_responses: - self.capabilities = ( - self.m.untagged_responses.pop("CAPABILITY")[0].decode().split() - ) - else: - data = self._cmd("CAPABILITY") - self.capabilities = data[0].decode().split() - - def logout(self): - """Logout from server.""" - try: - self._cmd("CHECK") - except ImapError: - pass - self._cmd("LOGOUT") - del self.m - self.m = None - if hasattr(self, "current_mailbox"): - del self.current_mailbox - - def _listmboxes(self, topmailbox, mailboxes): - """Retrieve mailboxes list.""" - pattern = ( - f'"{topmailbox.encode("imap4-utf-7").decode()}{self.hdelimiter}%"' - if topmailbox - else "%" - ) - resp = self._cmd("LIST", '""', pattern, "RETURN", "(CHILDREN)") - newmboxes = [] - for mb in resp: - if not mb: - continue - if type(mb) in [list, tuple]: - flags, delimiter, namelen = self.list_response_pattern_literal.match( - mb[0].decode() - ).groups() - name = mb[1][0 : int(namelen)] - else: - flags, delimiter, name, childinfo = ( - self.listextended_response_pattern.match(mb.decode()).groups() - ) - flags = flags.split(" ") - name = bytearray(name, "utf-8") - name = name.decode("imap4-utf-7") - mdm_found = False - for idx, mdm in enumerate(mailboxes): - if mdm["name"] == name: - mdm_found = True - descr = mailboxes[idx] - break - if not mdm_found: - descr = {"name": name} - newmboxes += [descr] - - if "\\HasChildren" in flags: - if "\\NonExistent" in flags: - descr["removed"] = True - descr["path"] = name - descr["sub"] = [] - - mailboxes += sorted(newmboxes, key=itemgetter("name")) - - def getmboxes(self, user, topmailbox: str = ""): - """Returns a list of mailboxes for a particular user - - By default, only the first level of mailboxes under - ``topmailbox`` is returned. - - :param user: a ``User`` instance - :param topmailbox: the mailbox where to start in the tree - :return: a list - """ - if topmailbox: - md_mailboxes = [] - else: - md_mailboxes = [ - {"name": "INBOX"}, - {"name": user.parameters.get_value("drafts_folder")}, - {"name": "Junk"}, - {"name": user.parameters.get_value("sent_folder")}, - {"name": user.parameters.get_value("trash_folder")}, - ] - self._listmboxes(topmailbox, md_mailboxes) - - return md_mailboxes - - -def get_imapconnector(request) -> IMAPconnector: - """ - Simple shortcut to create a connector - - :param request: a ``Request`` object - """ - imapc = IMAPconnector(user=request.user.username, password=str(request.auth)) - return imapc diff --git a/modoboa/webmail/tests/test_viewsets.py b/modoboa/webmail/tests/test_viewsets.py index 744215dcf..2659fc347 100644 --- a/modoboa/webmail/tests/test_viewsets.py +++ b/modoboa/webmail/tests/test_viewsets.py @@ -333,11 +333,11 @@ class ComposeSessionViewSetTestCase(WebmailTestCase): self.assertEqual(response.status_code, 200) manager = ComposeSessionManager(self.user.username) - # self.set_global_parameters({"max_attachment_size": "10"}) - # with self.settings(MEDIA_ROOT=self.workdir): - # response = self.client.post(url, {"attachment": get_gif()}) - # self.assertEqual(response.status_code, 400) - # self.assertContains(response, "Attachment is too big") + self.set_global_parameters({"max_attachment_size": "10"}) + with self.settings(MEDIA_ROOT=self.workdir): + response = self.client.post(url, {"attachment": get_gif()}) + self.assertEqual(response.status_code, 400) + self.assertIn("Attachment is too big", response.json()["attachment"][0]) self.set_global_parameters({"max_attachment_size": "10K"}) with self.settings(MEDIA_ROOT=self.workdir): @@ -370,17 +370,23 @@ class ComposeSessionViewSetTestCase(WebmailTestCase): self.authenticate() uid = self._create_compose_session() - url = reverse("v2:webmail-compose-session-send", args=[uid]) - response = self.client.post( - url, - { - "sender": self.user.email, - "to": ["test@example.test"], - "subject": "test", - "body": "Test", - }, - format="json", - ) + # Upload an attachment + url = reverse("v2:webmail-compose-session-attachments", args=[uid]) + with self.settings(MEDIA_ROOT=self.workdir): + response = self.client.post(url, {"attachment": get_gif()}) + + # Send the message + url = reverse("v2:webmail-compose-session-send", args=[uid]) + response = self.client.post( + url, + { + "sender": self.user.email, + "to": ["test@example.test"], + "subject": "test", + "body": "Test", + }, + format="json", + ) self.assertEqual(response.status_code, 204) self.assertEqual(len(mail.outbox), 1) self.assertEqual(mail.outbox[0].from_email, "user@test.com") diff --git a/modoboa/webmail/viewsets.py b/modoboa/webmail/viewsets.py index e89a332fb..bf090b8c0 100644 --- a/modoboa/webmail/viewsets.py +++ b/modoboa/webmail/viewsets.py @@ -3,6 +3,7 @@ from django import forms from django.core.validators import validate_email from django.http import Http404, HttpResponse +from django.utils.translation import gettext as _ from rest_framework import parsers, response, viewsets from rest_framework.decorators import action @@ -270,6 +271,13 @@ class ComposeSessionViewSet(viewsets.GenericViewSet): permission_classes = (IsAuthenticated, HasMailbox) + def initialize_request(self, request, *args, **kwargs): + response = super().initialize_request(request, *args, **kwargs) + if self.action == "attachments": + uploader = attachments.AttachmentUploadHandler() + request.upload_handlers.insert(0, uploader) + return response + def get_serializer_class(self): if self.action == "send": return serializers.SendEmailSerializer @@ -306,10 +314,15 @@ class ComposeSessionViewSet(viewsets.GenericViewSet): def attachments(self, request, pk): if request.method == "POST": manager = attachments.ComposeSessionManager(request.user.username) - uploader = attachments.AttachmentUploadHandler() - request.upload_handlers.insert(0, uploader) serializer = serializers.AttachmentUploadSerializer(data=request.FILES) - serializer.is_valid(raise_exception=True) + if not serializer.is_valid(): + errors = serializer.errors + if request.upload_handlers[0].toobig: + errors["attachment"] = [ + _("Attachment is too big (limit: %s)") + % request.upload_handlers[0].maxsize + ] + return response.Response(errors, status=400) result = attachments.save_attachment( request, pk, serializer.validated_data["attachment"] )