CertManagement.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

# See COPYING file for copyrights details.


import os
import shutil
import time
import json
from zipfile import ZipFile
from cryptography import x509
try:
    from twisted.internet.ssl import PrivateCertificate
    TWISTED_AVAILABLE = True
except ImportError:
    TWISTED_AVAILABLE = False
from cryptography.x509.oid import ExtensionOID
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
try:
    import OpenSSL
    OPENSSL_AVAILABLE = True
except ImportError:
    OPENSSL_AVAILABLE = False

from util.paths import AppDataPath

# Certificate Management Data model :
# [[CN, Desc, LastConnect]]
COL_CN, COL_DESC, COL_LAST = list(range(3))
REPLACE, KEEP, CANCEL = list(range(3))

keystore_path = AppDataPath("keystore")

def _certpath():
    certpath = os.path.join(keystore_path, "cert")
    return certpath


def _mgtpath():
    return os.path.join(_certpath(), 'management.json')


def _ensureCertdir():
    certpath = _certpath()
    if not os.path.exists(certpath):
        os.makedirs(certpath)
    return certpath


def _default(CN):
    return [CN,
            '',  # default description
            None]  # last connection date


def _dataByCN(data):
    return {row[COL_CN]: row for row in data}


def _LoadData():
    """ load known metadata """
    if os.path.isdir(_certpath()):
        _path = _mgtpath()
        if os.path.exists(_path):
            return json.loads(open(_path).read())
    return []


def _get_host_name_from_certificate_file(file_path):
    with open(file_path, "rb") as cert_file:
        cert_data = cert_file.read()
    cert = x509.load_pem_x509_certificate(cert_data, default_backend())
    # Support for legacy common name
    common_names = cert.subject.get_attributes_for_oid(x509.NameOID.COMMON_NAME)
    if common_names:
        return common_names[0].value
    # Get the subjectAltName extension from the certificate
    ext = cert.extensions.get_extension_for_oid(ExtensionOID.SUBJECT_ALTERNATIVE_NAME)
    # Get the dNSName entries from the SAN extension
    SAN = ext.value.get_values_for_type(x509.DNSName)
    return SAN[0]
    # In case CN is not enough: fpr=cert.fingerprint(hash_algorithm).hex()


def GetData(log):
    loaded_data = _LoadData()
    if loaded_data:
        certpath = _certpath()
        cert_files = os.listdir(certpath)
        input_by_CN = _dataByCN(loaded_data)
        output = []
        # Go through all certificate files available an build data
        # out of data recoverd from json and list of certificates.
        # This implicitly filters out metadata from certificates 
        # whose file is missing
        for filename in cert_files:
            if filename.endswith('.crt'):
                try:
                    filepath = os.path.join(certpath, filename)
                    CN = _get_host_name_from_certificate_file(filepath)
                except:
                    log.write_error("Could not load certificate %s: %s"%(filepath,str(e)))
                else:    
                    if CN+".crt" == filename:
                        output.append(input_by_CN.get(CN, _default(CN)))
                    else:
                        log.write_error("Certificate %s is missnamed: should be %s.crt"%(filepath,CN))
        return output
    return []


def DeleteCert(CN):
    secret_path = os.path.join(_certpath(), CN+'.crt')
    os.remove(secret_path)


def SaveData(data):
    _ensureCertdir()
    with open(_mgtpath(), 'w') as f:
        f.write(json.dumps(data))


def _touchCN(CN):
    # here we directly use _LoadData, avoiding filtering that could be long
    data = _LoadData()
    idata = _dataByCN(data)
    dataForCN = idata.get(CN, None) if data else None

    _is_new_CN = dataForCN is None
    if _is_new_CN:
        dataForCN = _default(CN)

    # FIXME : could store time instead os a string and use DVC model's cmp
    # then date display could be smarter, etc - sortable sting hack for now
    dataForCN[COL_LAST] = time.strftime('%y/%m/%d-%H:%M:%S')

    if _is_new_CN:
        data.append(dataForCN)

    SaveData(data)

def GetCertPath(CN):
    # find Certificate from project
    crtpath = os.path.join(_certpath(), CN+'.crt')
    if not os.path.exists(crtpath):
        raise ValueError(
            "Error: Certificate for %s is missing!\n" % CN + \
            "Provide valid certicate in identity manager.")
    _touchCN(CN)
    return crtpath


def ImportCert(filepath, log, sircb):
    certpath = _ensureCertdir()
    try:
        CN = _get_host_name_from_certificate_file(filepath)
    except Exception as e:
        log.write_error("Could not load certificate %s: %s"%(filepath,str(e)))
        return False
    else:    
        data = _LoadData()
        idata = _dataByCN(data)
        dataForCN = idata.get(CN, None) if data else None
        new_filename = os.path.join(certpath, CN+".crt")
        if dataForCN:
           if sircb(dataForCN) != REPLACE:
                log.write_warning("New certificate %s for %s was discarded."%(filepath, CN))
                return False
        shutil.copyfile(filepath, new_filename)
        _touchCN(CN)
    return True


def GetClientCert():
    own_keystore = os.path.join(keystore_path, "own")
    if not os.path.exists(own_keystore):
        os.makedirs(own_keystore)
    return os.path.join(own_keystore, "client.crt")

def GetClientCertificateInfo():
    if not TWISTED_AVAILABLE:
        return "Twisted module not available - client certificate info disabled"
    if not OPENSSL_AVAILABLE:
        return "pyOpenSSL module not available - client certificate info disabled"
    file_path = GetClientCert()
    if os.path.exists(file_path):
        info = ""
        try:
            with open(file_path, "rb") as cert_file:
                cert_data = cert_file.read()
            client_cert = PrivateCertificate.loadPEM(cert_data)
            cert = client_cert.original.to_cryptography()

            # Support for legacy common name
            common_names = cert.subject.get_attributes_for_oid(x509.NameOID.COMMON_NAME)
            for cn in common_names:
                info += "Common Name: %s\n"%cn.value
            try:
                ext = cert.extensions.get_extension_for_oid(ExtensionOID.SUBJECT_ALTERNATIVE_NAME)
                SAN = ext.value.get_values_for_type(x509.DNSName)
                for SANEntry in SAN:
                    info += "SubjectAltName: %s\n"%SANEntry
            except x509.extensions.ExtensionNotFound:
                info += "No SubjectAltName\n"

            info += "Fingerprint: %s\n"%cert.fingerprint(hashes.SHA256()).hex()
            info += "Creation date: %s\n"%cert.not_valid_before.isoformat()
            info += "Expiration date: %s\n"%cert.not_valid_after.isoformat()
        except OpenSSL.crypto.Error:
            info += "Imported PEM is invalid, it must contain Certificate and Private Key.\n"
        except Exception as e:
            info += "Error while loading certificate: %s\n"%str(e)
            print(e.__class__)
        return info
    return "No client certificate available"

def ImportClientCert(filepath):
    certpath = GetClientCert()
    shutil.copyfile(filepath, certpath)

def RemoveClientCert():
    certpath = GetClientCert()
    if os.path.exists(certpath ):
        os.remove(certpath)