Sign Up
Log In
Log In
or
Sign Up
Places
All Projects
Status Monitor
Collapse sidebar
systemsmanagement:Uyuni:Snapshots:2021.08
salt
x509-fixes-111.patch
Overview
Repositories
Revisions
Requests
Users
Attributes
Meta
File x509-fixes-111.patch of Package salt
From d947df3004323a3d400aaf7b5d94580965bf4a11 Mon Sep 17 00:00:00 2001 From: Florian Bergmann <bergmannf@users.noreply.github.com> Date: Fri, 14 Sep 2018 10:30:39 +0200 Subject: [PATCH] X509 fixes (#111) * Return proper content type for the x509 certificate * Remove parenthesis * Remove extra-variables during the import * Comment fix * Remove double returns * Change log level from trace to debug * Remove 'pass' and add logging instead * Remove unnecessary wrapping Remove wrapping * PEP 8: line too long PEP8: line too long * PEP8: Redefine RSAError variable in except clause * Do not return None if name was not found * Do not return None if no matched minions found * Fix unit tests --- salt/modules/publish.py | 17 +++---- salt/modules/x509.py | 100 ++++++++++++++++++++-------------------- salt/states/x509.py | 75 ++++++++++++++++++++++++++++-- 3 files changed, 129 insertions(+), 63 deletions(-) diff --git a/salt/modules/publish.py b/salt/modules/publish.py index 0926f3fa13..6d56c4d08e 100644 --- a/salt/modules/publish.py +++ b/salt/modules/publish.py @@ -1,15 +1,10 @@ -# -*- coding: utf-8 -*- """ Publish a command from a minion to a target """ -from __future__ import absolute_import, print_function, unicode_literals import logging - -# Import python libs import time -# Import salt libs import salt.crypt import salt.payload import salt.transport.client @@ -93,13 +88,13 @@ def _publish( matching_master_uris = [ master for master in __opts__["master_uri_list"] - if "//{0}:".format(via_master) in master + if "//{}:".format(via_master) in master ] if not matching_master_uris: raise SaltInvocationError( - "Could not find match for {0} in \ - list of configured masters {1} when using `via_master` option".format( + "Could not find match for {} in \ + list of configured masters {} when using `via_master` option".format( via_master, __opts__["master_uri_list"] ) ) @@ -143,7 +138,7 @@ def _publish( try: peer_data = channel.send(load) except SaltReqTimeoutError: - return "'{0}' publish timed out".format(fun) + return "'{}' publish timed out".format(fun) if not peer_data: return {} # CLI args are passed as strings, re-cast to keep time.sleep happy @@ -198,6 +193,8 @@ def _publish( else: return ret + return {} + def publish( tgt, fun, arg=None, tgt_type="glob", returner="", timeout=5, via_master=None @@ -347,4 +344,4 @@ def runner(fun, arg=None, timeout=5): try: return channel.send(load) except SaltReqTimeoutError: - return "'{0}' runner publish timed out".format(fun) + return "'{}' runner publish timed out".format(fun) diff --git a/salt/modules/x509.py b/salt/modules/x509.py index ec9c92c008..9b475f3096 100644 --- a/salt/modules/x509.py +++ b/salt/modules/x509.py @@ -30,16 +30,13 @@ from salt.utils.odict import OrderedDict try: import M2Crypto - - HAS_M2 = True except ImportError: - HAS_M2 = False + M2Crypto = None + try: import OpenSSL - - HAS_OPENSSL = True except ImportError: - HAS_OPENSSL = False + OpenSSL = None __virtualname__ = "x509" @@ -79,10 +76,10 @@ def __virtual__(): """ only load this module if m2crypto is available """ - if HAS_M2: - return __virtualname__ - else: - return (False, "Could not load x509 module, m2crypto unavailable") + return ( + __virtualname__ if M2Crypto is not None else False, + "Could not load x509 module, m2crypto unavailable", + ) class _Ctx(ctypes.Structure): @@ -140,8 +137,8 @@ def _new_extension(name, value, critical=0, issuer=None, _pyfree=1): x509_ext_ptr = M2Crypto.m2.x509v3_ext_conf(None, ctx, name, value) lhash = None except AttributeError: - lhash = M2Crypto.m2.x509v3_lhash() - ctx = M2Crypto.m2.x509v3_set_conf_lhash(lhash) + lhash = M2Crypto.m2.x509v3_lhash() # pylint: disable=no-member + ctx = M2Crypto.m2.x509v3_set_conf_lhash(lhash) # pylint: disable=no-member # ctx not zeroed _fix_ctx(ctx, issuer) x509_ext_ptr = M2Crypto.m2.x509v3_ext_conf(lhash, ctx, name, value) @@ -280,7 +277,7 @@ def _get_signing_policy(name): signing_policy = policies.get(name) if signing_policy: return signing_policy - return __salt__["config.get"]("x509_signing_policies", {}).get(name) + return __salt__["config.get"]("x509_signing_policies", {}).get(name) or {} def _pretty_hex(hex_str): @@ -318,9 +315,11 @@ def _text_or_file(input_): """ if _isfile(input_): with salt.utils.files.fopen(input_) as fp_: - return salt.utils.stringutils.to_str(fp_.read()) + out = salt.utils.stringutils.to_str(fp_.read()) else: - return salt.utils.stringutils.to_str(input_) + out = salt.utils.stringutils.to_str(input_) + + return out def _parse_subject(subject): @@ -339,7 +338,7 @@ def _parse_subject(subject): ret_list.append((nid_num, nid_name, val)) nids.append(nid_num) except TypeError as err: - log.trace("Missing attribute '%s'. Error: %s", nid_name, err) + log.debug("Missing attribute '%s'. Error: %s", nid_name, err) for nid_num, nid_name, val in sorted(ret_list): ret[nid_name] = val return ret @@ -536,8 +535,8 @@ def get_pem_entries(glob_path): if os.path.isfile(path): try: ret[path] = get_pem_entry(text=path) - except ValueError: - pass + except ValueError as err: + log.debug("Unable to get PEM entries from %s: %s", path, err) return ret @@ -615,8 +614,8 @@ def read_certificates(glob_path): if os.path.isfile(path): try: ret[path] = read_certificate(certificate=path) - except ValueError: - pass + except ValueError as err: + log.debug("Unable to read certificate %s: %s", path, err) return ret @@ -646,10 +645,9 @@ def read_csr(csr): "Subject": _parse_subject(csr.get_subject()), "Subject Hash": _dec2hex(csr.get_subject().as_hash()), "Public Key Hash": hashlib.sha1(csr.get_pubkey().get_modulus()).hexdigest(), + "X509v3 Extensions": _get_csr_extensions(csr), } - ret["X509v3 Extensions"] = _get_csr_extensions(csr) - return ret @@ -959,7 +957,7 @@ def create_crl( # pyOpenSSL Note due to current limitations in pyOpenSSL it is impossible # to specify a digest For signing the CRL. This will hopefully be fixed # soon: https://github.com/pyca/pyopenssl/pull/161 - if not HAS_OPENSSL: + if OpenSSL is None: raise salt.exceptions.SaltInvocationError( "Could not load OpenSSL module, OpenSSL unavailable" ) @@ -1033,7 +1031,9 @@ def create_crl( crltext = crl.export(**export_kwargs) except (TypeError, ValueError): log.warning( - "Error signing crl with specified digest. Are you using pyopenssl 0.15 or newer? The default md5 digest will be used." + "Error signing crl with specified digest. " + "Are you using pyopenssl 0.15 or newer? " + "The default md5 digest will be used." ) export_kwargs.pop("digest", None) crltext = crl.export(**export_kwargs) @@ -1109,6 +1109,7 @@ def get_signing_policy(signing_policy_name): signing_policy = _get_signing_policy(signing_policy_name) if not signing_policy: return "Signing policy {} does not exist.".format(signing_policy_name) + if isinstance(signing_policy, list): dict_ = {} for item in signing_policy: @@ -1125,7 +1126,7 @@ def get_signing_policy(signing_policy_name): signing_policy["signing_cert"], "CERTIFICATE" ) except KeyError: - pass + log.debug('Unable to get "certificate" PEM entry') return signing_policy @@ -1734,7 +1735,8 @@ def create_csr(path=None, text=False, **kwargs): if "private_key" not in kwargs and "public_key" in kwargs: kwargs["private_key"] = kwargs["public_key"] log.warning( - "OpenSSL no longer allows working with non-signed CSRs. A private_key must be specified. Attempting to use public_key as private_key" + "OpenSSL no longer allows working with non-signed CSRs. " + "A private_key must be specified. Attempting to use public_key as private_key" ) if "private_key" not in kwargs: @@ -1758,7 +1760,8 @@ def create_csr(path=None, text=False, **kwargs): ) ) - for entry in sorted(subject.nid): + # pylint: disable=unused-variable + for entry, num in subject.nid.items(): if entry in kwargs: setattr(subject, entry, kwargs[entry]) @@ -1794,7 +1797,6 @@ def create_csr(path=None, text=False, **kwargs): extstack.push(ext) csr.add_extensions(extstack) - csr.sign( _get_private_key_obj( kwargs["private_key"], passphrase=kwargs["private_key_passphrase"] @@ -1802,10 +1804,11 @@ def create_csr(path=None, text=False, **kwargs): kwargs["algorithm"], ) - if path: - return write_pem(text=csr.as_pem(), path=path, pem_type="CERTIFICATE REQUEST") - else: - return csr.as_pem() + return ( + write_pem(text=csr.as_pem(), path=path, pem_type="CERTIFICATE REQUEST") + if path + else csr.as_pem() + ) def verify_private_key(private_key, public_key, passphrase=None): @@ -1830,7 +1833,7 @@ def verify_private_key(private_key, public_key, passphrase=None): salt '*' x509.verify_private_key private_key=/etc/pki/myca.key \\ public_key=/etc/pki/myca.crt """ - return bool(get_public_key(private_key, passphrase) == get_public_key(public_key)) + return get_public_key(private_key, passphrase) == get_public_key(public_key) def verify_signature( @@ -1886,7 +1889,10 @@ def verify_crl(crl, cert): salt '*' x509.verify_crl crl=/etc/pki/myca.crl cert=/etc/pki/myca.crt """ if not salt.utils.path.which("openssl"): - raise salt.exceptions.SaltInvocationError("openssl binary not found in path") + raise salt.exceptions.SaltInvocationError( + 'External command "openssl" not found' + ) + crltext = _text_or_file(crl) crltext = get_pem_entry(crltext, pem_type="X509 CRL") crltempfile = tempfile.NamedTemporaryFile(delete=True) @@ -1908,10 +1914,7 @@ def verify_crl(crl, cert): crltempfile.close() certtempfile.close() - if "verify OK" in output: - return True - else: - return False + return "verify OK" in output def expired(certificate): @@ -1949,8 +1952,9 @@ def expired(certificate): ret["expired"] = True else: ret["expired"] = False - except ValueError: - pass + except ValueError as err: + log.debug("Failed to get data of expired certificate: %s", err) + log.trace(err, exc_info=True) return ret @@ -1973,6 +1977,7 @@ def will_expire(certificate, days): salt '*' x509.will_expire "/etc/pki/mycert.crt" days=30 """ + ts_pt = "%Y-%m-%d %H:%M:%S" ret = {} if os.path.isfile(certificate): @@ -1986,14 +1991,11 @@ def will_expire(certificate, days): _expiration_date = cert.get_not_after().get_datetime() ret["cn"] = _parse_subject(cert.get_subject())["CN"] - - if _expiration_date.strftime("%Y-%m-%d %H:%M:%S") <= _check_time.strftime( - "%Y-%m-%d %H:%M:%S" - ): - ret["will_expire"] = True - else: - ret["will_expire"] = False - except ValueError: - pass + ret["will_expire"] = _expiration_date.strftime( + ts_pt + ) <= _check_time.strftime(ts_pt) + except ValueError as err: + log.debug("Unable to return details of a sertificate expiration: %s", err) + log.trace(err, exc_info=True) return ret diff --git a/salt/states/x509.py b/salt/states/x509.py index 8269df2f37..5ad7195b7c 100644 --- a/salt/states/x509.py +++ b/salt/states/x509.py @@ -177,12 +177,14 @@ import os import re import salt.exceptions +import salt.utils.stringutils import salt.utils.versions +from salt.ext import six try: from M2Crypto.RSA import RSAError except ImportError: - pass + RSAError = Exception("RSA Error") log = logging.getLogger(__name__) @@ -194,7 +196,7 @@ def __virtual__(): if "x509.get_pem_entry" in __salt__: return "x509" else: - return (False, "Could not load x509 state: m2crypto unavailable") + return False, "Could not load x509 state: the x509 is not available" def _revoked_to_list(revs): @@ -696,7 +698,70 @@ def certificate_managed( "Old": invalid_reason, "New": "Certificate will be valid and up to date", } - return ret + private_key_args.update(managed_private_key) + kwargs["public_key_passphrase"] = private_key_args["passphrase"] + + if private_key_args["new"]: + rotate_private_key = True + private_key_args["new"] = False + + if _check_private_key( + private_key_args["name"], + bits=private_key_args["bits"], + passphrase=private_key_args["passphrase"], + new=private_key_args["new"], + overwrite=private_key_args["overwrite"], + ): + private_key = __salt__["x509.get_pem_entry"]( + private_key_args["name"], pem_type="RSA PRIVATE KEY" + ) + else: + new_private_key = True + private_key = __salt__["x509.create_private_key"]( + text=True, + bits=private_key_args["bits"], + passphrase=private_key_args["passphrase"], + cipher=private_key_args["cipher"], + verbose=private_key_args["verbose"], + ) + + kwargs["public_key"] = private_key + + 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 = "{} is not a valid Certificate.".format(name) + else: + current = "{} does not exist.".format(name) contents = __salt__["x509.create_certificate"](text=True, **kwargs) # Check the module actually returned a cert and not an error message as a string @@ -892,6 +957,8 @@ def pem_managed(name, text, backup=False, **kwargs): Any arguments supported by :py:func:`file.managed <salt.states.file.managed>` are supported. """ file_args, kwargs = _get_file_args(name, **kwargs) - file_args["contents"] = __salt__["x509.get_pem_entry"](text=text) + file_args["contents"] = salt.utils.stringutils.to_str( + __salt__["x509.get_pem_entry"](text=text) + ) return __states__["file.managed"](**file_args) -- 2.29.2
Locations
Projects
Search
Status Monitor
Help
OpenBuildService.org
Documentation
API Documentation
Code of Conduct
Contact
Support
@OBShq
Terms
openSUSE Build Service is sponsored by
The Open Build Service is an
openSUSE project
.
Sign Up
Log In
Places
Places
All Projects
Status Monitor
