Produits Entr'ouvert » Calebasse

# -*- coding: utf-8 -*-

import os

import sys

import re

import base64

from datetime import datetime

import gzip

import StringIO

import ldap

from M2Crypto import X509, SSL, Rand, SMIME, BIO

MODE_TEST = False

MODE_COMPRESS = True

MODE_ENCRYPT = True

LDAP_HOST = 'ldap://annuaire.gip-cps.fr'

if MODE_TEST:

    LDAP_BASEDN = 'o=gip-cps-test,c=fr'

    CAPATH = '/var/lib/calebasse/test-gip-cps.capath/'

else:

    # production

    LDAP_BASEDN = 'o=gip-cps,c=fr'

    CAPATH = '/var/lib/calebasse/gip-cps.capath/'

LDAP_BASEDN_RSS = 'ou=339172288100045,l=Sarthe (72),' + LDAP_BASEDN

LDAP_X509_ATTR = 'userCertificate;binary'

LDAP_CA_ATTRS = {

        'cert': ('cACertificate;binary', 'CERTIFICATE'),

        'crl': ('certificateRevocationList;binary', 'X509 CRL'),

        'delta-crl': ('deltaRevocationList;binary', 'X509 CRL'),

    }

RANDFILE = '/var/tmp/randpool.dat'

if MODE_TEST:

    MAILPATH = '/var/lib/calebasse/test-mail.out/'

    MESSAGE_ID_RIGHT = 'teletransmission-test.aps42.org'

else:

    # production

    MAILPATH = '/var/lib/calebasse/mail.out/'

    MESSAGE_ID_RIGHT = 'teletransmission.aps42.org'

SENDER = 'teletransmission@aps42.org'

VVVVVV = '100500'  # ETS-DT-001-TransportsFlux_SpecsTechCommune_v1.1.pdf

NUMERO_EMETTEUR = '00000420788606'

EXERCICE = NUMERO_EMETTEUR

#

# get a certificate from gip-cps LDAP

#

def get_certificate(large_regime, dest_organism):

    """

    return a M2Crypto.X509 object, containing the certificate of the health center

    example :

      x509 = get_certificate('01', '422') # CPAM Saint Etienne

      print x509.as_text()

      print x509.as_pem()

    """

    l = ldap.initialize(LDAP_HOST)

    cn = large_regime + dest_organism + '@' + dest_organism + '.' + large_regime + '.rss.fr'

    results = l.search_s(LDAP_BASEDN_RSS, ldap.SCOPE_SUBTREE, '(cn=' + cn + ')')

    if len(results) > 1:

        raise LookupError("non unique result for cn=%s" % cn)

    if len(results) < 1:

        raise LookupError("no result for cn=%s" % cn)

    dn = results[0][0]

    attrs = results[0][1]

    if LDAP_X509_ATTR not in attrs:

        raise LookupError("no certificate in dn:%s" % dn)

    certificates = {}

    for der in results[0][1][LDAP_X509_ATTR]:

        x509 = X509.load_cert_der_string(der)

        serial = x509.get_serial_number()

        startdate =  x509.get_not_after().get_datetime().replace(tzinfo=None)

        enddate = x509.get_not_before().get_datetime().replace(tzinfo=None)

        now = datetime.utcnow().replace(tzinfo=None)

        if startdate >= now >= enddate:

            # TODO : add capath + crl validation

            # os.execute(openssl verify -CApath CAPATH -crl_check cert.pem)

            certificates[serial] = x509

    if certificates:

        return certificates[max(certificates)]

    return None

#

# IRIS/B2 mail construction

#

def mime(message):

    # compress

    if MODE_COMPRESS:

        sio = StringIO.StringIO()

        gzf = gzip.GzipFile(fileobj=sio, mode='wb', filename='B2.gz')

        gzf.write(message)

        gzf.close()

        zmessage = sio.getvalue()

        sio.close()

    else:

        zmessage = message

    if MODE_TEST:

        content_description = 'IRISTEST/B2'

    else:

        content_description = 'IRIS/B2'

    if MODE_COMPRESS:

        content_description += '/Z'

    return "Content-Type: application/EDI-consent\n" + \

            "Content-Transfer-Encoding: base64\n" + \

            "Content-Description: " + content_description + "\n" + \

            "\n" + base64.encodestring(zmessage)

def smime(message, x509):

    """

    output s/mime message (headers+payload), encrypted with x509 certificate

    """

    # encrypt

    if RANDFILE:

        Rand.load_file(RANDFILE, -1)

    s = SMIME.SMIME()

    sk = X509.X509_Stack()

    sk.push(x509)

    s.set_x509_stack(sk)

    s.set_cipher(SMIME.Cipher('des_ede3_cbc'))

    bio = BIO.MemoryBuffer(message)

    pkcs7 = s.encrypt(bio)

    if RANDFILE:

        Rand.save_file(RANDFILE)

    out = BIO.MemoryBuffer()

    s.write(out, pkcs7)

    return out.read()

def build_mail(large_regime, dest_organism, b2_filename):

    """

    build a mail to healt center, with b2-compressed+encrypted information

    """

    b2_fd = open(b2_filename, 'r')

    b2 = b2_fd.read()

    b2_fd.close()

    utcnow = datetime.utcnow()

    stamp = utcnow.strftime('%Y%m%d%H%M%S') + utcnow.strftime('%f')[:5]

    # count invoice in the b2 file = lines start with "5"

    nb_invoices = [b2[x*128] for x in range(len(b2)/128)].count('5')

    subject = 'IR%s/%s/%s/%0.5d' % (VVVVVV, EXERCICE, stamp, nb_invoices)

    delimiter = '_ENTROUVERT-CALEBASSE-B2-%s' % stamp

    mail = {

            'From': SENDER,

            'To': large_regime + dest_organism + '@' + dest_organism + '.' + large_regime + '.rss.fr',

            'Message-ID': '<%s@%s>' % (stamp, MESSAGE_ID_RIGHT),

            'Subject': subject,

            'Mime-Version': '1.0',

            'Content-Type': 'multipart/mixed; boundary="%s"' % delimiter,

            }

    mime_part = mime(b2)

    if MODE_ENCRYPT:

        mime_part = smime(mime(b2), get_certificate(large_regime, dest_organism))

    fd = StringIO.StringIO()

    for k,v in mail.items():

        fd.write('%s: %s\n' % (k,v))

    fd.write('\n')

    fd.write('--%s\n' % delimiter)

    fd.write(mime_part)

    fd.write('--%s--\n' % delimiter)

    ret = fd.getvalue()

    fd.close()

    return ret

#

# CApath construction

#

def der2pem(der, type_='CERTIFICATE'):

    return "-----BEGIN %s-----\n%s\n-----END %s-----\n" % \

        (type_, base64.encodestring(der).rstrip(), type_)

def build_capath(path=CAPATH):

    """

    get all pkiCA from the gip-cps.fr ldap, store them in path

    note: the gip-cps.fr ldap is limited to 10 objects in a response... by chance, there is less than 10 pkiCA ;)

    """

    l = ldap.initialize(LDAP_HOST)

    results = l.search_s(LDAP_BASEDN,ldap.SCOPE_SUBTREE,'(objectclass=pkiCA)')

    for ca in results:

        dn = ca[0]

        for attr in LDAP_CA_ATTRS:

            if LDAP_CA_ATTRS[attr][0] in ca[1]:

                n = 0

                for der in ca[1][LDAP_CA_ATTRS[attr][0]]:

                    filename = os.path.join(path, '%s.%d.%s.pem' % (dn, n, attr))

                    print "create ", filename

                    fd = open(filename, 'w')

                    fd.write(der2pem(der, LDAP_CA_ATTRS[attr][1]))

                    fd.close()

                    n += 1

    os.chdir(path)

    os.system('c_rehash .')

#

#

#

if __name__ == '__main__' and (len(sys.argv) > 1):

    action = sys.argv[1]

    if action == 'build_capath':

        build_capath()

    if action == 'x509':

        x509 = get_certificate(sys.argv[2], sys.argv[3])

        print x509.as_text()

        print x509.as_pem()

    if MODE_TEST and (action == 'test'):

        print build_mail('01', '996', 'test.b2')