m2crypto use inside Zope

## Decoding and verifing of signed files.
##
## This file is part of PAFlow 2.
##
## Authors: Claudio Battaglino <[email protected]>
##          Marco Bizzarri <[email protected]>
##          Paolo Bizzarri <[email protected]>
##          Giuseppe Ciuni <[email protected]>
##          Maurizio Giosia <[email protected]>
##          Ivan Ricotti <[email protected]>
##          Daniele Tarini <[email protected]>
##
## Copyright (C) 2003-2004 Icube Srl
##
## PAFlow is free software; you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation; either version 2 of the License, or
## (at your option) any later version.
##
## PAFlow is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
## GNU General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with PAFlow; if not, write to the Free Software
## Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

from cStringIO import StringIO
from email import message_from_file
from os import listdir, environ
try:
    from M2Crypto import BIO, SMIME, X509
    from M2Crypto import BIO, EVP, X509, Err, util, m2
except:
    raise M2CryptoNotFound

from Globals import INSTANCE_HOME
from App.FindHomes import CLIENT_HOME

class SignatureGeneralError:
    pass

class SignatureVerifyError:
    pass

def load_pkcs7_bio_der(p7_bio):
    p7_ptr = m2.pkcs7_read_bio_der(p7_bio._ptr())
    if p7_ptr is None:
        raise Err.get_error()
    return SMIME.PKCS7(p7_ptr, 1)


class CertificateHelper:
    __allow_access_to_unprotected_subobjects__ = 1

    def __init__(self):
        self._common_name = []
        self._issuer_name = []
        self._serial_number = 0
        self._not_before = ''
        self._not_after = ''

    def set_common_name(self, common_name):
        self._common_name = common_name

    def set_serial_number(self, serial_number):
        self._serial_number = serial_number

    def set_not_before(self, not_before):
        self._not_before = not_before

    def set_not_after(self, not_after):
        self._not_after = not_after

    def set_issuer_name(self, issuer_name):
        self._issuer_name = issuer_name

    def get_common_name(self):
        return self._common_name

    def get_serial_number(self):
        return self._serial_number

    def get_not_before(self):
        return self._not_before

    def get_not_after(self):
        return self._not_after

    def get_issuer_name(self):
        return self._issuer_name


class Verifier:
    _nids = ('C', 'ST', 'L', 'O', 'OU', 'CN', 'emailAddress')

    def __init__(self, cert_dir=None):
        if cert_dir is None:
            self._cert_dir = INSTANCE_HOME + '/var/certs'
        else:
            self._cert_dir = cert_dir
        self._store = X509.X509_Store()
        for filename in listdir(self._cert_dir):
            self._store.add_x509(X509.load_cert(self._cert_dir + '/' + filename))

    def _fillCertificateHelper(self, certificate_helper, x509):
        certificate_helper.set_not_before(str(x509.get_not_before()))
        certificate_helper.set_not_after(str(x509.get_not_after()))
        certificate_helper.set_serial_number(str(x509.get_serial_number()))
        x509_name = x509.get_subject()
        common_name = []
        for n in self._nids:
            if getattr(x509_name, n):
                common_name.append((n, getattr(x509_name, n)))
        certificate_helper.set_common_name(common_name)
        x509_issuer_name = x509.get_issuer()
        issuer_name = []
        for n in self._nids:
            if getattr(x509_issuer_name, n):
                issuer_name.append((n, getattr(x509_issuer_name, n)))
        certificate_helper.set_issuer_name(issuer_name)

    def _prepareCertificates(self, signers_stack):
        result = []
        for x509 in signers_stack:
            certificate_helper = CertificateHelper()
            self._fillCertificateHelper(certificate_helper, x509)
            result.append(certificate_helper)
        return result

    def run1(self, file):
        p7_bio = BIO.MemoryBuffer(file.read())
        try:
            p7 = load_pkcs7_bio_der(p7_bio)
        except:
            return []
        s = SMIME.SMIME()
        s.set_x509_store(self._store)
        s.set_x509_stack(p7.get0_signers(X509.X509_Stack()))
        try:
            file = s.verify(p7)
        except:
            return []
        signers_stack = p7.get0_signers(X509.X509_Stack())

        return self._prepareCertificates(signers_stack)

    def run2(self, file):
        bio = BIO.MemoryBuffer(file.read())
        try:
            p7, data = SMIME.smime_load_pkcs7_bio(bio)
        except:
            return []
        signers_stack = p7.get0_signers(X509.X509_Stack())
        s = SMIME.SMIME()
        s.set_x509_stack(signers_stack)
        s.set_x509_store(self._store)
        file.seek(0)
        msg = message_from_file(file)
        txt = msg.get_payload(0).get_payload()
        new_data = BIO.MemoryBuffer(txt)
        try:
            v = s.verify(p7, new_data)
        except:
            return []

        return self._prepareCertificates(signers_stack)

    def run(self, file):
        result1 = self.run1(file)
        if result1:
            return result1
        file.seek(0)
        result2 = self.run2(file)
        if result2:
            return result2
        return []