[PATCH] Implement SSL certificate checking
Sebastian
Sebastian at SSpaeth.de
Tue Dec 14 13:15:01 GMT 2010
If we specify a cacertfile= setting in the Repository section of an IMAP server,
validate it on python>=2.6 or fail with python<=2.5.
The hostname check should propbably be made optional.
This should make debian happy.
Signed-off-By: Sebastian Spaeth <Sebastian at SSpaeth.de>
---
This is a big patch for inspection and review (sorry didn't try to break it up yet). It implements SSL Certificate checking so taht Debian does not pull us out. Cert checking gets enabled when the user specifies sslcacert=path/to/file in the Repository section of offlineimaprc. It also checks that the hostname of the certificate matches the connection host name.
If the user specifies that option under python 2.4/2.5, we fail with an Exception as those versions do not support certificate checking.
offlineimap/imaplibutil.py | 186 ++++++++++++++++++++++++++--------------
offlineimap/imapserver.py | 15 ++--
offlineimap/repository/IMAP.py | 3 +
3 files changed, 132 insertions(+), 72 deletions(-)
diff --git a/offlineimap/imaplibutil.py b/offlineimap/imaplibutil.py
index a60242b..839712c 100644
--- a/offlineimap/imaplibutil.py
+++ b/offlineimap/imaplibutil.py
@@ -25,9 +25,9 @@ from imaplib import IMAP4_PORT, IMAP4_SSL_PORT, InternalDate, Mon2num
try:
import ssl
- ssl_wrap = ssl.wrap_socket
except ImportError:
- ssl_wrap = socket.ssl
+ #fails on python <2.6
+ pass
class IMAP4_Tunnel(IMAP4):
"""IMAP4 client class over a tunnel
@@ -62,45 +62,7 @@ class IMAP4_Tunnel(IMAP4):
self.infd.close()
self.outfd.close()
self.process.wait()
-
-class sslwrapper:
- def __init__(self, sslsock):
- self.sslsock = sslsock
- self.readbuf = ''
-
- def write(self, s):
- return self.sslsock.write(s)
-
- def _read(self, n):
- return self.sslsock.read(n)
- def read(self, n):
- if len(self.readbuf):
- # Return the stuff in readbuf, even if less than n.
- # It might contain the rest of the line, and if we try to
- # read more, might block waiting for data that is not
- # coming to arrive.
- bytesfrombuf = min(n, len(self.readbuf))
- retval = self.readbuf[:bytesfrombuf]
- self.readbuf = self.readbuf[bytesfrombuf:]
- return retval
- retval = self._read(n)
- if len(retval) > n:
- self.readbuf = retval[n:]
- return retval[:n]
- return retval
-
- def readline(self):
- retval = ''
- while 1:
- linebuf = self.read(1024)
- nlindex = linebuf.find("\n")
- if nlindex != -1:
- retval += linebuf[:nlindex + 1]
- self.readbuf = linebuf[nlindex + 1:] + self.readbuf
- return retval
- else:
- retval += linebuf
def new_mesg(self, s, secs=None):
if secs is None:
@@ -108,27 +70,38 @@ def new_mesg(self, s, secs=None):
tm = time.strftime('%M:%S', time.localtime(secs))
UIBase.getglobalui().debug('imap', ' %s.%02d %s' % (tm, (secs*100)%100, s))
-class WrappedIMAP4_SSL(IMAP4_SSL):
- def open(self, host = '', port = IMAP4_SSL_PORT):
- IMAP4_SSL.open(self, host, port)
- self.sslobj = sslwrapper(self.sslobj)
- def readline(self):
- return self.sslobj.readline()
+class WrappedIMAP4_SSL(IMAP4_SSL):
+ """Provides an improved version of the standard IMAP4_SSL
+
+ It provides a better readline() implementation as impaplibs()
+ readline is extremly inefficient. It can connect to IPv6
+ addresses. It performs SSL certificate checking on python >=2.6
+ and optional hostname check. It also replaces the
+ IMAP4_SSL().sslobj with an ssl implementation that works on both
+ python 2.4/5 and 2."""
+ 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 new_open(self, host = '', port = IMAP4_PORT):
- """Setup connection to remote server on "host:port"
- (default: localhost:standard IMAP4 port).
- This connection will be used by the routines:
- read, readline, send, shutdown.
- """
+ def open(self, host = '', port = IMAP4_SSL_PORT):
+ """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 at 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 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:
@@ -142,22 +115,106 @@ def new_open(self, host = '', port = IMAP4_PORT):
if last_error != 0:
# FIXME
raise socket.error(last_error)
- self.file = self.sock.makefile('rb')
-def new_open_ssl(self, host = '', port = IMAP4_SSL_PORT):
- """Setup connection to remote server on "host:port".
- (default: localhost:standard IMAP4 SSL port).
+ #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, now check
+ #that hostnames also match.
+ error = self._verifycert(self.sslobj.getpeercert(), host)
+ if error:
+ raise ssl.SSLError("SSL Certificate host name mismatch: %s" % error)
+
+ def _verifycert(self, cert, hostname):
+ '''Verify that cert (in socket.getpeercert() format) matches hostname.
+ CRLs and subjectAltName 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()
+ for s in cert.get('subject', []):
+ key, value = s[0]
+ if key == 'commonName':
+ certname = value.lower()
+ if (certname == dnsname or
+ '.' in dnsname and certname == '*.' + dnsname.split('.', 1)[1]):
+ return None
+ return ('certificate is for %s') % certname
+ return ('no commonName found in certificate')
+
+ def _read_upto (self, n):
+ """Read up to n bytes, emptying existing _readbuffer first"""
+ bytesfrombuf = min(n, len(self._readbuf))
+ if bytesfrombuf:
+ # Return the stuff in readbuf, even if less than n.
+ # It might contain the rest of the line, and if we try to
+ # read more, might block waiting for data that is not
+ # coming to arrive.
+ retval = self._readbuf[:bytesfrombuf]
+ self._readbuf = self._readbuf[bytesfrombuf:]
+ return retval
+ return self.sslobj.read(min(n, 16384))
+
+ def read(self, n):
+ """Read exactly n bytes (as done in IMAP4_SSL.read() API)."""
+ chunks = []
+ read = 0
+ while read < n:
+ data = self._read_upto (n-read)
+ read += len(data)
+ chunks.append(data)
+
+ return ''.join(chunks)
+
+ def readline(self):
+ retval = ''
+ while 1:
+ linebuf = self._read_upto(1024)
+ nlindex = linebuf.find("\n")
+ if nlindex != -1:
+ retval += linebuf[:nlindex + 1]
+ self._readbuf = linebuf[nlindex + 1:] + self._readbuf
+ return retval
+ else:
+ retval += linebuf
+
+
+def new_open(self, host = '', port = IMAP4_PORT):
+ """Setup connection to remote server on "host:port"
+ (default: localhost:standard IMAP4 port).
This connection will be used by the routines:
read, readline, send, shutdown.
"""
self.host = host
self.port = port
- #This connects to the first ip found ipv4/ipv6
- #Added by Adriaan Peeters <apeeters at 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 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:
@@ -171,8 +228,7 @@ def new_open_ssl(self, host = '', port = IMAP4_SSL_PORT):
if last_error != 0:
# FIXME
raise socket.error(last_error)
- self.sslobj = ssl_wrap(self.sock, self.keyfile, self.certfile)
- self.sslobj = sslwrapper(self.sslobj)
+ self.file = self.sock.makefile('rb')
mustquote = re.compile(r"[^\w!#$%&'+,.:;<=>?^`|~-]")
diff --git a/offlineimap/imapserver.py b/offlineimap/imapserver.py
index d9dd6ba..22869ff 100644
--- a/offlineimap/imapserver.py
+++ b/offlineimap/imapserver.py
@@ -81,12 +81,8 @@ class UsefulIMAP4(UsefulIMAPMixIn, imaplib.IMAP4):
return imaplib.IMAP4.read (self, size)
class UsefulIMAP4_SSL(UsefulIMAPMixIn, imaplibutil.WrappedIMAP4_SSL):
- def open(self, host = '', port = imaplib.IMAP4_SSL_PORT):
- imaplibutil.new_open_ssl(self, host, port)
-
# This is the same hack as above, to be used in the case of an SSL
# connexion.
-
def read(self, size):
if (system() == 'Darwin') and (size>0) :
read = 0
@@ -107,7 +103,8 @@ class IMAPServer:
def __init__(self, config, reposname,
username = None, password = None, hostname = None,
port = None, ssl = 1, maxconnections = 1, tunnel = None,
- reference = '""', sslclientcert = None, sslclientkey = None):
+ reference = '""', sslclientcert = None, sslclientkey = None,
+ sslcacertfile= None):
self.reposname = reposname
self.config = config
self.username = username
@@ -120,6 +117,7 @@ class IMAPServer:
self.usessl = ssl
self.sslclientcert = sslclientcert
self.sslclientkey = sslclientkey
+ self.sslcacertfile = sslcacertfile
self.delim = None
self.root = None
if port == None:
@@ -260,7 +258,8 @@ class IMAPServer:
elif self.usessl:
UIBase.getglobalui().connecting(self.hostname, self.port)
imapobj = UsefulIMAP4_SSL(self.hostname, self.port,
- self.sslclientkey, self.sslclientcert)
+ self.sslclientkey, self.sslclientcert,
+ cacertfile = self.sslcacertfile)
else:
UIBase.getglobalui().connecting(self.hostname, self.port)
imapobj = UsefulIMAP4(self.hostname, self.port)
@@ -421,6 +420,7 @@ class ConfigedIMAPServer(IMAPServer):
ssl = self.repos.getssl()
sslclientcert = self.repos.getsslclientcert()
sslclientkey = self.repos.getsslclientkey()
+ sslcacertfile = self.repos.getsslcacertfile()
reference = self.repos.getreference()
server = None
password = None
@@ -442,4 +442,5 @@ class ConfigedIMAPServer(IMAPServer):
self.repos.getmaxconnections(),
reference = reference,
sslclientcert = sslclientcert,
- sslclientkey = sslclientkey)
+ sslclientkey = sslclientkey,
+ sslcacertfile = sslcacertfile)
diff --git a/offlineimap/repository/IMAP.py b/offlineimap/repository/IMAP.py
index 52a8e61..3bfa5db 100644
--- a/offlineimap/repository/IMAP.py
+++ b/offlineimap/repository/IMAP.py
@@ -139,6 +139,9 @@ class IMAPRepository(BaseRepository):
def getsslclientkey(self):
return self.getconf('sslclientkey', None)
+ def getsslcacertfile(self):
+ return self.getconf('sslcacertfile', None)
+
def getpreauthtunnel(self):
return self.getconf('preauthtunnel', None)
--
1.7.1
More information about the OfflineIMAP-project
mailing list