diff --git a/modoboa/lib/exceptions.py b/modoboa/lib/exceptions.py index 82c1135f0..aa2bb26c4 100644 --- a/modoboa/lib/exceptions.py +++ b/modoboa/lib/exceptions.py @@ -66,3 +66,12 @@ class PermDeniedException(ModoboaException): if self.msg: return _("Permission denied: %s" % self.msg) return _("Permission denied") + + +class ImapError(ModoboaException): + + def __init__(self, reason): + self.reason = reason + + def __str__(self): + return str(self.reason) diff --git a/modoboa/lib/imaputils.py b/modoboa/lib/imaputils.py new file mode 100644 index 000000000..14e19bd63 --- /dev/null +++ b/modoboa/lib/imaputils.py @@ -0,0 +1,302 @@ +# coding: utf-8 +""" +:mod:`imaputils` --- Extra IMAPv4 utilities +------------------------------------------- +""" +import imaplib +import re +import ssl +from functools import wraps + +from django.utils.translation import ugettext as _ + +from modoboa.lib import parameters, imap_utf7 # NOQA +from modoboa.lib.connections import ConnectionsManager +from modoboa.lib.exceptions import InternalError + +from .exceptions import ImapError + +# imaplib.Debug = 4 + + +class capability(object): + + """ + Simple decorator to check if the server presents the required + capability. If not, a fallback method is called instead. + + :param name: the capability name (upper case) + :param fallback_method: a method's name + """ + + def __init__(self, name, fallback_method): + self.name = name + self.fallback_method = fallback_method + + def __call__(self, method): + @wraps(method) + def wrapped_func(cls, *args, **kwargs): + if self.name in cls.capabilities: + return method(cls, *args, **kwargs) + return getattr(cls, self.fallback_method)(cls, **kwargs) + + return wrapped_func + + +class IMAPconnector(object): + + """The IMAPv4 connector.""" + + __metaclass__ = ConnectionsManager + + 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.address = parameters.get_admin("IMAP_SERVER", "core") + self.port = int(parameters.get_admin("IMAP_PORT", "core")) + 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 + """ + if name in ['FETCH', 'SORT', 'STORE', 'COPY', 'SEARCH']: + try: + typ, data = self.m.uid(name, *args) + except imaplib.IMAP4.error as e: + raise ImapError(e) + if typ == "NO": + raise ImapError(data) + return data + + try: + typ, data = self.m._simple_command(name, *args) + except imaplib.IMAP4.error as e: + raise ImapError(e) + if typ == "NO": + raise ImapError(data) + 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]) + if m is None: + raise InternalError(_("Failed to retrieve hierarchy delimiter")) + self.__hdelimiter = m.group('delimiter') + return self.__hdelimiter + + def refresh(self, user, password): + """Check if current connection needs a refresh + + Is it really secure? + """ + if self.m is not None: + try: + self._cmd("NOOP") + except ImapError: + if hasattr(self, "current_mailbox"): + del self.current_mailbox + else: + return + + self.login(user, password) + + def login(self, user, passwd): + """Custom login method + + We connect to the server, issue a LOGIN command. If + successfull, we try to record a eventuel CAPABILITY untagged + response. Otherwise, we issue the command. + + :param user: username + :param passwd: password + """ + import socket + + if isinstance(user, unicode): + user = user.encode("utf-8") + if isinstance(passwd, unicode): + passwd = passwd.encode("utf-8") + try: + secured = parameters.get_admin("IMAP_SECURED", "core") + if secured == "yes": + self.m = imaplib.IMAP4_SSL(self.address, self.port) + else: + self.m = imaplib.IMAP4(self.address, self.port) + except (socket.error, imaplib.IMAP4.error, ssl.SSLError) as error: + raise ImapError(_("Connection to IMAP server failed: %s" % error)) + + data = self._cmd("LOGIN", user, passwd) + self.m.state = "AUTH" + if "CAPABILITY" in self.m.untagged_responses: + self.capabilities = \ + self.m.untagged_responses.pop('CAPABILITY')[0].split() + else: + data = self._cmd("CAPABILITY") + self.capabilities = data[0].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_simple(self, topmailbox='INBOX', mailboxes=None, **kwargs): + # data = self._cmd("LIST", "", "*") + if not mailboxes: + mailboxes = [] + (status, data) = self.m.list() + newmboxes = [] + for mb in data: + flags, delimiter, name = self.list_response_pattern.match( + mb).groups() + name = name.strip('"').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 = dict(name=name) + newmboxes += [descr] + + if re.search("\%s" % delimiter, name): + parts = name.split(delimiter) + if "path" not in descr: + descr["path"] = parts[0] + descr["sub"] = [] + if self._parse_mailbox_name(descr["sub"], parts[0], delimiter, + parts[1:]): + descr["class"] = "subfolders" + continue + + from operator import itemgetter + mailboxes += sorted(newmboxes, key=itemgetter("name")) + + @capability('LIST-EXTENDED', '_listmboxes_simple') + def _listmboxes(self, topmailbox, mailboxes): + """Retrieve mailboxes list.""" + pattern = ( + "{0}{1}%".format(topmailbox.encode("imap4-utf-7"), 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 [str, unicode]: + flags, delimiter, name, childinfo = \ + self.listextended_response_pattern.match(mb).groups() + else: + flags, delimiter, namelen = ( + self.list_response_pattern_literal.match(mb[0]).groups() + ) + name = mb[1][0:int(namelen)] + flags = flags.split(' ') + 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 = dict(name=name) + newmboxes += [descr] + + if r'\Marked' in flags or r'\UnMarked' not in flags: + descr["send_status"] = True + if r'\HasChildren' in flags: + if r'\NonExistent' in flags: + descr["removed"] = True + descr["path"] = name + descr["sub"] = [] + + from operator import itemgetter + mailboxes += sorted(newmboxes, key=itemgetter("name")) + + def getmboxes(self, user, topmailbox=""): + """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", "class": "fa fa-inbox"}, + {"name": parameters.get_user(user, "DRAFTS_FOLDER"), + "class": "fa fa-file"}, + {"name": 'Junk', "class": "fa fa-fire"}, + {"name": parameters.get_user(user, "SENT_FOLDER"), + "class": "fa fa-envelope"}, + {"name": parameters.get_user(user, "TRASH_FOLDER"), + "class": "fa fa-trash"} + ] + self._listmboxes(topmailbox, md_mailboxes) + + return md_mailboxes + + +def get_imapconnector(request): + """Simple shortcut to create a connector + + :param request: a ``Request`` object + """ + imapc = IMAPconnector(user=request.user.username, + password=request.session["password"]) + return imapc + + +if __name__ == "__main__": + import doctest + doctest.testmod()