source: TI12-security/trunk/python/ndg.security.common/ndg/security/common/m2CryptoSSLUtility.py @ 2932

"""Extend M2Crypto SSL functionality for cert verification and custom
timeout settings.

NERC Data Grid Project"""
__author__ = "P J Kershaw"
__date__ = "02/07/07"
__copyright__ = "(C) 2007 STFC & NERC"
__license__ = \
"""This software may be distributed under the terms of the Q Public 
License, version 1.0 or later."""
__contact__ = "P.J.Kershaw@rl.ac.uk"
__revision__ = '$Id$'

import httplib
import socket

from M2Crypto import SSL, X509
from M2Crypto.httpslib import HTTPSConnection as _HTTPSConnection

from ndg.security.common.X509 import X509Cert, X509Stack

class InvalidCertSignature(SSL.Checker.SSLVerificationError):
    """Raise if verification against CA cert public key fails"""

class InvalidCertDN(SSL.Checker.SSLVerificationError):
    """Raise if verification against a list acceptable DNs fails"""
   

class HostCheck(SSL.Checker.Checker, object):
    """Override SSL.Checker.Checker to enable alternate Common Name
    setting match for peer cert"""

    def __init__(self, 
                 peerCertDN=None, 
                 peerCertCN=None,
                 acceptedDNs=[], 
                 caCertList=[],
                 caCertFilePathList=[], 
                 **kw):
        """Override parent class __init__ to enable setting of myProxyServerDN
        setting
        
        @type peerCertDN: string/list
        @param peerCertDN: Set the expected Distinguished Name of the
        server to avoid errors matching hostnames.  This is useful
        where the hostname is not fully qualified.  

        *param acceptedDNs: a list of acceptable DNs.  This enables validation where the expected DN is
        where against a limited list of certs.
        
        @type peerCertCN: string
        @param peerCertCN: enable alternate Common Name to peer
        hostname
        
        @type caCertList: list type of M2Crypto.X509.X509 types
        @param caCertList: CA X.509 certificates - if set the peer cert's 
        CA signature is verified against one of these.  At least one must
        verify
        
        @type caCertFilePathList: list string types
        @param caCertFilePathList: same as caCertList except input as list
        of CA cert file paths"""
        
        SSL.Checker.Checker.__init__(self, **kw)
        
        self.peerCertDN = peerCertDN
        self.peerCertCN = peerCertCN
        self.acceptedDNs = acceptedDNs
        
        if caCertList:
            self.caCertList = caCertList
        elif caCertFilePathList:
            self.caCertFilePathList = caCertFilePathList
            
        
    def __call__(self, peerCert, host=None):
        """Carry out checks on server ID
        @param peerCert: MyProxy server host certificate as M2Crypto.X509.X509
        instance
        @param host: name of host to check
        """
        peerCertDN = '/'+peerCert.get_subject().as_text().replace(', ', '/')
        try:
            SSL.Checker.Checker.__call__(self, peerCert, host=self.peerCertCN)
            
        except SSL.Checker.WrongHost, e:
            # Try match against peerCertDN set   
            if peerCertDN != self.peerCertDN:
                raise e

        # At least one match should be found in the list
        if self.acceptedDNs and \
           not len([dn for dn in self.acceptedDNs if peerCertDN==dn]):
            raise InvalidCertDN, \
                "Peer cert DN %s doesn't match verification list" % peerCertDN

        if len(self.__caCertStack) > 0:
            try:
                self.__caCertStack.verifyCertChain(\
                           x509Cert2Verify=X509Cert(m2CryptoX509=peerCert))
            except Exception, e:
                raise InvalidCertSignature, \
            "Peer certificate verification against CA cert failed: "+str(e) 
              
        # They match - drop the exception and return all OK instead          
        return True
    
    
    def __setCACertList(self, caCertList):
        """Set list of CA certs - peer cert must validate against at least one
        of these"""
        self.__caCertStack = X509Stack()
        for caCert in caCertList:
            self.__caCertStack.push(caCert)

    caCertList = property(fset=__setCACertList,
              doc="list of CA certs - peer cert must validate against one")


    #_________________________________________________________________________
    def __setCACertsFromFileList(self, caCertFilePathList):
        '''Read CA certificates from file and add them to the X.509
        stack
        
        @type caCertFilePathList: list or tuple
        @param caCertFilePathList: list of file paths for CA certificates to
        be used to verify certificate used to sign message'''
        
        if not isinstance(caCertFilePathList, list) and \
           not isinstance(caCertFilePathList, tuple):
            raise AttributeError, \
                        'Expecting a list or tuple for "caCertFilePathList"'

        self.__caCertStack = X509Stack()

        for caCertFilePath in caCertFilePathList:
            self.__caCertStack.push(X509.load_cert(caCertFilePath))
        
    caCertFilePathList = property(fset=__setCACertsFromFileList,
    doc="list of CA cert file paths - peer cert must validate against one")


class HTTPSConnection(_HTTPSConnection):
    """Modified version of M2Crypto equivalent to enable custom checks with
    the peer and timeout settings
    
    @type defReadTimeout: M2Crypto.SSL.timeout
    @cvar defReadTimeout: default timeout for read operations
    @type defWriteTimeout: M2Crypto.SSL.timeout
    @cvar defWriteTimeout: default timeout for write operations"""    
    defReadTimeout = SSL.timeout(sec=20.)
    defWriteTimeout = SSL.timeout(sec=20.)
    
    def __init__(self, *args, **kw):
        '''Overload to enable setting of post connection check
        callback to SSL.Connection
        
        type *args: tuple
        param *args: args which apply to M2Crypto.httpslib.HTTPSConnection
        type **kw: dict
        param **kw: additional keywords
        @type postConnectionCheck: SSL.Checker.Checker derivative
        @keyword postConnectionCheck: set class for checking peer
        @type readTimeout: M2Crypto.SSL.timeout
        @keyword readTimeout: readTimeout - set timeout for read
        @type writeTimeout: M2Crypto.SSL.timeout
        @keyword writeTimeout: similar to read timeout'''
        
        if 'postConnectionCheck' in kw:
            self._postConnectionCheck = kw['postConnectionCheck']
            del kw['postConnectionCheck']
        else:
            self._postConnectionCheck = SSL.Checker.Checker
        
        if 'readTimeout' in kw:
            if not isinstance(readTimeout, SSL.timeout):
                raise AttributeError, "readTimeout must be of type " + \
                                      "M2Crypto.SSL.timeout" 
            self.readTimeout = readTimeout
            del kw['readTimeout']
        else:
            self.readTimeout = HTTPSConnection.defReadTimeout
              
        if 'writeTimeout' in kw:
            if not isinstance(writeTimeout, SSL.timeout):
                raise AttributeError, "writeTimeout must be of type " + \
                                      "M2Crypto.SSL.timeout" 
            self.writeTimeout = writeTimeout
            del kw['writeTimeout']
        else:
            self.writeTimeout = HTTPSConnection.defWriteTimeout
            
        _HTTPSConnection.__init__(self, *args, **kw)
        
        
    def connect(self):
        '''Overload M2Crypto.httpslib.HTTPSConnection to enable
        custom post connection check of peer certificate and socket timeout'''
        self.sock = SSL.Connection(self.ssl_ctx)
        self.sock.set_post_connection_check_callback(
                                                 self._postConnectionCheck)

        self.sock.set_socket_read_timeout(self.readTimeout)
        self.sock.set_socket_write_timeout(self.writeTimeout)

        self.sock.connect((self.host, self.port))