tobias/docker-offlineimap
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Adapt the code to work with the new imaplib2

Browse Source 
imaplib renamed self.sslobj to self.sock and our overriden open()
functions were failing for that reason when updating imaplib2 to
v2.28. It turns out that all of our custom initializations are being
done by stock imaplib2 now anyway, so there is no need to override them
anymore. This lets us simplify the code we have to worry about.

Move the verifycert() function to the imapserver.py file, it is now a
callback function that is being handed to imaplib from there, so it
makes sense to also define it in our imapserver function...
(this also lets us easily make use of the verifycert function in the
starttls case in the future)

TODO: we need to examine if and why we still need to override the
select() function, it is the only reason why we still wrap the IMAP4
classes.

Signed-off-by: Sebastian Spaeth <Sebastian@SSpaeth.de>
Signed-off-by: Nicolas Sebrecht <nicolas.s-dev@laposte.net>
This commit is contained in:
Sebastian Spaeth 2011-08-15 11:55:42 +02:00 committed by Nicolas Sebrecht
parent 1c0c19ad86
commit 3a91e296f0
3 changed files with 55 additions and 154 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Changelog.draft.rst
View File

@ -23,6 +23,8 @@ Changes
* Refactor our IMAPServer class. Background work without user-visible
changes.
* Updated bundled imaplib2 to version 2.28
Bug Fixes
---------

156
offlineimap/imaplibutil.py
View File

@ -127,162 +127,16 @@ def new_mesg(self, s, tn=None, secs=None):
tm = time.strftime('%M:%S', time.localtime(secs))
getglobalui().debug('imap', ' %s.%02d %s %s' % (tm, (secs*100)%100, tn, s))
class WrappedIMAP4_SSL(UsefulIMAPMixIn, IMAP4_SSL):
"""Provides an improved version of the standard IMAP4_SSL
"""Improved version of imaplib.IMAP4_SSL overriding select()"""
pass
It provides a better readline() implementation as impaplib's
readline() is extremly inefficient. It can also connect to IPv6
addresses."""
def __init__(self, *args, **kwargs):
self._readbuf = ''
self._cacertfile = kwargs.get('cacertfile', None)
if kwargs.has_key('cacertfile'):
del kwargs['cacertfile']
IMAP4_SSL.__init__(self, *args, **kwargs)
def open(self, host=None, port=None):
"""Do whatever IMAP4_SSL would do in open, but call sslwrap
with cert verification"""
#IMAP4_SSL.open(self, host, port) uses the below 2 lines:
self.host = host
self.port = port
#rather than just self.sock = socket.create_connection((host, port))
#we use the below part to be able to connect to ipv6 addresses too
#This connects to the first ip found ipv4/ipv6
#Added by Adriaan Peeters <apeeters@lashout.net> based on a socket
#example from the python documentation:
#http://www.python.org/doc/lib/socket-example.html
res = socket.getaddrinfo(host, port, socket.AF_UNSPEC,
socket.SOCK_STREAM)
# Try all the addresses in turn until we connect()
last_error = 0
for remote in res:
af, socktype, proto, canonname, sa = remote
self.sock = socket.socket(af, socktype, proto)
last_error = self.sock.connect_ex(sa)
if last_error == 0:
break
else:
self.sock.close()
if last_error != 0:
raise Exception("can't open socket; error: %s"\
% socket.error(last_error))
# Allow sending of keep-alive message seems to prevent some servers
# from closing SSL on us leading to deadlocks
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
#connected to socket, now wrap it in SSL
try:
if self._cacertfile:
requirecert = ssl.CERT_REQUIRED
else:
requirecert = ssl.CERT_NONE
self.sslobj = ssl.wrap_socket(self.sock, self.keyfile,
self.certfile,
ca_certs = self._cacertfile,
cert_reqs = requirecert)
except NameError:
#Python 2.4/2.5 don't have the ssl module, we need to
#socket.ssl() here but that doesn't allow cert
#verification!!!
if self._cacertfile:
#user configured a CA certificate, but python 2.4/5 doesn't
#allow us to easily check it. So bail out here.
raise Exception("SSL CA Certificates cannot be checked with python <=2.6. Abort")
self.sslobj = socket.ssl(self.sock, self.keyfile,
self.certfile)
else:
#ssl.wrap_socket worked and cert is verified (if configured),
#now check that hostnames also match if we have a CA cert.
if self._cacertfile:
error = self._verifycert(self.sslobj.getpeercert(), host)
if error:
raise ssl.SSLError("SSL Certificate host name mismatch: %s" % error)
# imaplib2 uses this to poll()
self.read_fd = self.sock.fileno()
#TODO: Done for now. We should implement a mutt-like behavior
#that offers the users to accept a certificate (presenting a
#fingerprint of it) (get via self.sslobj.getpeercert()), and
#save that, and compare on future connects, rather than having
#to trust what the CA certs say.
def _verifycert(self, cert, hostname):
'''Verify that cert (in socket.getpeercert() format) matches hostname.
CRLs are not handled.
Returns error message if any problems are found and None on success.
'''
if not cert:
return ('no certificate received')
dnsname = hostname.lower()
certnames = []
# cert expired?
notafter = cert.get('notAfter')
if notafter:
if time.time() >= ssl.cert_time_to_seconds(notafter):
return ('server certificate error: certificate expired %s'
) % notafter
# First read commonName
for s in cert.get('subject', []):
key, value = s[0]
if key == 'commonName':
certnames.append(value.lower())
if len(certnames) == 0:
return ('no commonName found in certificate')
# Then read subjectAltName
for key, value in cert.get('subjectAltName', []):
if key == 'DNS':
certnames.append(value.lower())
# And finally try to match hostname with one of these names
for certname in certnames:
if (certname == dnsname or
'.' in dnsname and certname == '*.' + dnsname.split('.', 1)[1]):
return None
return ('no matching domain name found in certificate')
class WrappedIMAP4(UsefulIMAPMixIn, IMAP4):
"""Improved version of imaplib.IMAP4 that can also connect to IPv6"""
"""Improved version of imaplib.IMAP4 overriding select()"""
pass
def open(self, host = '', port = IMAP4_PORT):
"""Setup connection to remote server on "host:port"
(default: localhost:standard IMAP4 port).
"""
#self.host and self.port are needed by the parent IMAP4 class
self.host = host
self.port = port
res = socket.getaddrinfo(host, port, socket.AF_UNSPEC,
socket.SOCK_STREAM)
# Try each address returned by getaddrinfo in turn until we
# manage to connect to one.
# Try all the addresses in turn until we connect()
last_error = 0
for remote in res:
af, socktype, proto, canonname, sa = remote
self.sock = socket.socket(af, socktype, proto)
last_error = self.sock.connect_ex(sa)
if last_error == 0:
break
else:
self.sock.close()
if last_error != 0:
raise Exception("can't open socket; error: %s"\
% socket.error(last_error))
self.file = self.sock.makefile('rb')
# imaplib2 uses this to poll()
self.read_fd = self.sock.fileno()
def Internaldate2epoch(resp):
"""Convert IMAP4 INTERNALDATE to UT.

51
offlineimap/imapserver.py
View File

@ -24,10 +24,11 @@ import offlineimap.accounts
import hmac
import socket
import base64
import time
from socket import gaierror
try:
from ssl import SSLError
from ssl import SSLError, cert_time_to_seconds
except ImportError:
# Protect against python<2.6, use dummy and won't get SSL errors.
SSLError = None
@ -204,9 +205,12 @@ class IMAPServer:
self.ui.connecting(self.hostname, self.port)
imapobj = imaplibutil.WrappedIMAP4_SSL(self.hostname,
self.port,
self.sslclientkey, self.sslclientcert,
self.sslclientkey,
self.sslclientcert,
self.sslcacertfile,
self.verifycert,
timeout=socket.getdefaulttimeout(),
cacertfile = self.sslcacertfile)
)
else:
self.ui.connecting(self.hostname, self.port)
imapobj = imaplibutil.WrappedIMAP4(self.hostname, self.port,
@ -403,6 +407,47 @@ class IMAPServer:
self.ui.debug('imap', 'keepalive: bottom of loop')
def verifycert(self, cert, hostname):
'''Verify that cert (in socket.getpeercert() format) matches hostname.
CRLs are not handled.
Returns error message if any problems are found and None on success.
'''
errstr = "CA Cert verifying failed: "
if not cert:
return ('%s no certificate received' % errstr)
dnsname = hostname.lower()
certnames = []
# cert expired?
notafter = cert.get('notAfter')
if notafter:
if time.time() >= cert_time_to_seconds(notafter):
return '%s certificate expired %s' % (errstr, notafter)
# First read commonName
for s in cert.get('subject', []):
key, value = s[0]
if key == 'commonName':
certnames.append(value.lower())
if len(certnames) == 0:
return ('%s no commonName found in certificate' % errstr)
# Then read subjectAltName
for key, value in cert.get('subjectAltName', []):
if key == 'DNS':
certnames.append(value.lower())
# And finally try to match hostname with one of these names
for certname in certnames:
if (certname == dnsname or
'.' in dnsname and certname == '*.' + dnsname.split('.', 1)[1]):
return None
return ('%s no matching domain name found in certificate' % errstr)
class IdleThread(object):
def __init__(self, parent, folder=None):
self.parent = parent