mirror of
https://github.com/jwinterm/modoboa.git
synced 2026-10-18 05:51:26 +02:00
Finally, not imap tools in the core.
This commit is contained in:
Antoine Nguyen
@@ -1,214 +0,0 @@
|
||||
# -*- coding: iso-8859-1 -*-
|
||||
|
||||
"""
|
||||
Imap folder names are encoded using a special version of utf-7 as
|
||||
defined in RFC 2060 section 5.1.3.
|
||||
|
||||
5.1.3. Mailbox International Naming Convention
|
||||
|
||||
By convention, international mailbox names are specified using a
|
||||
modified version of the UTF-7 encoding described in [UTF-7]. The
|
||||
purpose of these modifications is to correct the following problems
|
||||
with UTF-7:
|
||||
|
||||
1) UTF-7 uses the "+" character for shifting; this conflicts with
|
||||
the common use of "+" in mailbox names, in particular USENET
|
||||
newsgroup names.
|
||||
|
||||
2) UTF-7's encoding is BASE64 which uses the "/" character; this
|
||||
conflicts with the use of "/" as a popular hierarchy delimiter.
|
||||
|
||||
3) UTF-7 prohibits the unencoded usage of "\"; this conflicts with
|
||||
the use of "\" as a popular hierarchy delimiter.
|
||||
|
||||
4) UTF-7 prohibits the unencoded usage of "~"; this conflicts with
|
||||
the use of "~" in some servers as a home directory indicator.
|
||||
|
||||
5) UTF-7 permits multiple alternate forms to represent the same
|
||||
string; in particular, printable US-ASCII chararacters can be
|
||||
represented in encoded form.
|
||||
|
||||
In modified UTF-7, printable US-ASCII characters except for "&"
|
||||
represent themselves; that is, characters with octet values 0x20-0x25
|
||||
and 0x27-0x7e. The character "&" (0x26) is represented by the two-
|
||||
octet sequence "&-".
|
||||
|
||||
All other characters (octet values 0x00-0x1f, 0x7f-0xff, and all
|
||||
Unicode 16-bit octets) are represented in modified BASE64, with a
|
||||
further modification from [UTF-7] that "," is used instead of "/".
|
||||
Modified BASE64 MUST NOT be used to represent any printing US-ASCII
|
||||
character which can represent itself.
|
||||
|
||||
"&" is used to shift to modified BASE64 and "-" to shift back to US-
|
||||
ASCII. All names start in US-ASCII, and MUST end in US-ASCII (that
|
||||
is, a name that ends with a Unicode 16-bit octet MUST end with a "-
|
||||
").
|
||||
|
||||
For example, here is a mailbox name which mixes English, Japanese,
|
||||
and Chinese text: ~peter/mail/&ZeVnLIqe-/&U,BTFw-
|
||||
|
||||
Found here:
|
||||
http://svn.plone.org/svn/collective/mxmImapClient/trunk/imapUTF7.py
|
||||
|
||||
"""
|
||||
import binascii
|
||||
import codecs
|
||||
|
||||
# encoding
|
||||
|
||||
|
||||
def modified_base64(s):
|
||||
s = s.encode('utf-16be')
|
||||
return binascii.b2a_base64(s).rstrip('\n=').replace('/', ',')
|
||||
|
||||
|
||||
def doB64(_in, r):
|
||||
if _in:
|
||||
r.append('&%s-' % modified_base64(''.join(_in)))
|
||||
del _in[:]
|
||||
|
||||
|
||||
def encoder(s):
|
||||
r = []
|
||||
_in = []
|
||||
for c in s:
|
||||
ordC = ord(c)
|
||||
if 0x20 <= ordC <= 0x25 or 0x27 <= ordC <= 0x7e:
|
||||
doB64(_in, r)
|
||||
r.append(c)
|
||||
elif c == '&':
|
||||
doB64(_in, r)
|
||||
r.append('&-')
|
||||
else:
|
||||
_in.append(c)
|
||||
doB64(_in, r)
|
||||
return (str(''.join(r)), len(s))
|
||||
|
||||
|
||||
# decoding
|
||||
|
||||
def modified_unbase64(s):
|
||||
b = binascii.a2b_base64(s.replace(',', '/') + '===')
|
||||
return unicode(b, 'utf-16be')
|
||||
|
||||
|
||||
def decoder(s):
|
||||
r = []
|
||||
decode = []
|
||||
for c in s:
|
||||
if c == '&' and not decode:
|
||||
decode.append('&')
|
||||
elif c == '-' and decode:
|
||||
if len(decode) == 1:
|
||||
r.append('&')
|
||||
else:
|
||||
r.append(modified_unbase64(''.join(decode[1:])))
|
||||
decode = []
|
||||
elif decode:
|
||||
decode.append(c)
|
||||
else:
|
||||
r.append(c)
|
||||
if decode:
|
||||
r.append(modified_unbase64(''.join(decode[1:])))
|
||||
bin_str = ''.join(r)
|
||||
return (bin_str, len(s))
|
||||
|
||||
|
||||
class StreamReader(codecs.StreamReader):
|
||||
def decode(self, s, errors='strict'):
|
||||
return decoder(s)
|
||||
|
||||
|
||||
class StreamWriter(codecs.StreamWriter):
|
||||
def decode(self, s, errors='strict'):
|
||||
return encoder(s)
|
||||
|
||||
|
||||
def imap4_utf_7(name):
|
||||
if name == 'imap4-utf-7':
|
||||
return (encoder, decoder, StreamReader, StreamWriter)
|
||||
codecs.register(imap4_utf_7)
|
||||
|
||||
|
||||
# testing methods
|
||||
|
||||
def imapUTF7Encode(ust):
|
||||
"Returns imap utf-7 encoded version of string"
|
||||
return ust.encode('imap4-utf-7')
|
||||
|
||||
|
||||
def imapUTF7EncodeSequence(seq):
|
||||
"Returns imap utf-7 encoded version of strings in sequence"
|
||||
return [imapUTF7Encode(itm) for itm in seq]
|
||||
|
||||
|
||||
def imapUTF7Decode(st):
|
||||
"Returns utf7 encoded version of imap utf-7 string"
|
||||
return st.decode('imap4-utf-7')
|
||||
|
||||
|
||||
def imapUTF7DecodeSequence(seq):
|
||||
"Returns utf7 encoded version of imap utf-7 strings in sequence"
|
||||
return [imapUTF7Decode(itm) for itm in seq]
|
||||
|
||||
|
||||
def utf8Decode(st):
|
||||
"Returns utf7 encoded version of imap utf-7 string"
|
||||
return st.decode('utf-8')
|
||||
|
||||
|
||||
def utf7SequenceToUTF8(seq):
|
||||
"Returns utf7 encoded version of imap utf-7 strings in sequence"
|
||||
return [itm.decode('imap4-utf-7').encode('utf-8') for itm in seq]
|
||||
|
||||
|
||||
__all__ = ['imapUTF7Encode', 'imapUTF7Decode', ]
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
# print u'bøx'.encode('imap4-utf-7')
|
||||
# print 'expected b&APg-x'
|
||||
|
||||
# print u'båx'.encode('imap4-utf-7')
|
||||
# print 'expected b&AOU-x'
|
||||
|
||||
print '#######'
|
||||
print 'bøx'
|
||||
e = imapUTF7Encode(u'bøx')
|
||||
print e
|
||||
print imapUTF7Decode(e).encode('latin-1')
|
||||
|
||||
print '#######'
|
||||
print 'båx'
|
||||
e = imapUTF7Encode(u'båx')
|
||||
print e
|
||||
print imapUTF7Decode(e).encode('latin-1')
|
||||
|
||||
print '#######'
|
||||
print '~/bågø'
|
||||
e = imapUTF7Encode(u'~/bågø')
|
||||
print e
|
||||
print imapUTF7Decode(e).encode('latin-1')
|
||||
|
||||
print '#######'
|
||||
print 'Ting & Såger'
|
||||
e = imapUTF7Encode(u'Ting & Såger')
|
||||
print e
|
||||
print imapUTF7Decode(e).encode('latin-1')
|
||||
|
||||
# e = imapUTF7Decode('b&AOU-x')
|
||||
# print e.encode('latin-1')
|
||||
|
||||
# e = imapUTF7Decode('b&APg-x')
|
||||
# print e.encode('latin-1')
|
||||
|
||||
print '#######'
|
||||
print '~/Følder/mailbåx & stuff + more'
|
||||
n = u'~/Følder/mailbåx & stuff + more'
|
||||
e = imapUTF7Encode(n)
|
||||
print e
|
||||
print imapUTF7Decode(e).encode('latin-1')
|
||||
|
||||
print '#######'
|
||||
print '~peter/mail/&ZeVnLIqe-/&U,BTFw-'
|
||||
print imapUTF7Decode('~peter/mail/&ZeVnLIqe-/&U,BTFw-').encode('utf-8')
|
||||
@@ -1,302 +0,0 @@
|
||||
# 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<flags>.*?)\) "(?P<delimiter>.*)" "?(?P<name>[^"]*)"?'
|
||||
)
|
||||
list_response_pattern_literal = re.compile(
|
||||
r'\((?P<flags>.*?)\) "(?P<delimiter>.*)" \{(?P<namelen>\d+)\}')
|
||||
list_response_pattern = re.compile(list_base_pattern)
|
||||
listextended_response_pattern = \
|
||||
re.compile(list_base_pattern + r'\s*(?P<childinfo>.*)')
|
||||
|
||||
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()
|
||||
Reference in New Issue
Block a user