Custom implementation of salt.states.x509

x509.py

import salt.states.x509
from salt.states.x509 import *
from salt.states.x509 import _get_file_args

del certificate_managed


def __init__(opts):
    salt.states.x509.__salt__ = __salt__
    salt.states.x509.__states__ = __states__


def certificate_managed(name,
                        days_remaining=90,
                        managed_private_key=None,
                        append_certs=None,
                        **kwargs):
    """
    Fix for https://github.com/saltstack/salt/issues/52180

    For simplicity, managed_private_key and issuer checks are removed.
    See :py:func:`salt.states.x509.certificate_managed`.
    """
    if 'path' in kwargs:
        name = kwargs.pop('path')

    file_args, kwargs = _get_file_args(name, **kwargs)

    current_days_remaining = 0
    current_comp = {}

    if os.path.isfile(name):
        try:
            current = __salt__['x509.read_certificate'](certificate=name)
            current_comp = copy.deepcopy(current)
            if 'serial_number' not in kwargs:
                current_comp.pop('Serial Number')
                if 'signing_cert' not in kwargs:
                    try:
                        current_comp['X509v3 Extensions']['authorityKeyIdentifier'] = (
                            re.sub(r'serial:([0-9A-F]{2}:)*[0-9A-F]{2}', 'serial:--',
                                   current_comp['X509v3 Extensions']['authorityKeyIdentifier']))
                    except KeyError:
                        pass
            current_comp.pop('Not Before')
            current_comp.pop('MD5 Finger Print')
            current_comp.pop('SHA1 Finger Print')
            current_comp.pop('SHA-256 Finger Print')
            current_notafter = current_comp.pop('Not After')
            current_days_remaining = (
                    datetime.datetime.strptime(current_notafter, '%Y-%m-%d %H:%M:%S') -
                    datetime.datetime.now()).days
            if days_remaining == 0:
                days_remaining = current_days_remaining - 1
        except salt.exceptions.SaltInvocationError:
            current = '{0} is not a valid Certificate.'.format(name)
    else:
        current = '{0} does not exist.'.format(name)

    if 'ca_server' in kwargs and 'signing_policy' not in kwargs:
        raise salt.exceptions.SaltInvocationError(
            'signing_policy must be specified if ca_server is.')

    new_cert_pem = __salt__['x509.create_certificate'](text=True, **kwargs)
    new = __salt__['x509.read_certificate'](certificate=new_cert_pem)

    if isinstance(new, dict):
        new_comp = copy.deepcopy(new)
        if 'serial_number' not in kwargs:
            new_comp.pop('Serial Number')
            if 'signing_cert' not in kwargs:
                try:
                    new_comp['X509v3 Extensions']['authorityKeyIdentifier'] = (
                        re.sub(r'serial:([0-9A-F]{2}:)*[0-9A-F]{2}', 'serial:--',
                               new_comp['X509v3 Extensions']['authorityKeyIdentifier']))
                except KeyError:
                    pass
        new_comp.pop('Not Before')
        new_comp.pop('Not After')
        new_comp.pop('MD5 Finger Print')
        new_comp.pop('SHA1 Finger Print')
        new_comp.pop('SHA-256 Finger Print')
    else:
        new_comp = new

    new_certificate = False
    if current_comp == new_comp and current_days_remaining > days_remaining:
        certificate = __salt__['x509.get_pem_entry'](name, pem_type='CERTIFICATE')
    else:
        new_certificate = True
        certificate = new_cert_pem

    file_args['contents'] = ''
    file_args['contents'] += salt.utils.stringutils.to_str(certificate)

    if not append_certs:
        append_certs = []
    for append_cert in append_certs:
        file_args[
            'contents'] += __salt__['x509.get_pem_entry'](append_cert, pem_type='CERTIFICATE')

    file_args['show_changes'] = False
    ret = __states__['file.managed'](**file_args)

    if ret['changes']:
        ret['changes'] = {'Certificate': ret['changes']}
    else:
        ret['changes'] = {}
    if new_certificate:
        ret['changes']['Certificate'] = {
            'Old': current,
            'New': new}

    return ret