Sign Up
Log In
Log In
or
Sign Up
Places
All Projects
Status Monitor
Collapse sidebar
openSUSE:Leap:15.5:Update
salt.9310
x509-fixes-111.patch
Overview
Repositories
Revisions
Requests
Users
Attributes
Meta
File x509-fixes-111.patch of Package salt.9310
From 85f2b6749424035c88373559b530ce2a7caed709 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 | 8 +- salt/modules/x509.py | 132 ++++++++++++-------------------- salt/states/x509.py | 22 ++++-- tests/unit/modules/test_x509.py | 9 ++- 4 files changed, 74 insertions(+), 97 deletions(-) diff --git a/salt/modules/publish.py b/salt/modules/publish.py index 2de99583f4..ac31b4b65f 100644 --- a/salt/modules/publish.py +++ b/salt/modules/publish.py @@ -83,10 +83,8 @@ def _publish( in minion configuration but `via_master` was specified.') else: # Find the master in the list of master_uris generated by the minion base class - matching_master_uris = [master for master - in __opts__['master_uri_list'] - if '//{0}:'.format(via_master) - in master] + matching_master_uris = [master for master in __opts__['master_uri_list'] + if '//{0}:'.format(via_master) in master] if not matching_master_uris: raise SaltInvocationError('Could not find match for {0} in \ @@ -176,6 +174,8 @@ def _publish( else: return ret + return {} + def publish(tgt, fun, diff --git a/salt/modules/x509.py b/salt/modules/x509.py index 87a40104ca..15ccf0889d 100644 --- a/salt/modules/x509.py +++ b/salt/modules/x509.py @@ -36,14 +36,13 @@ from salt.state import STATE_INTERNAL_KEYWORDS as _STATE_INTERNAL_KEYWORDS # Import 3rd Party Libs 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' @@ -81,10 +80,7 @@ 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): @@ -127,10 +123,8 @@ def _new_extension(name, value, critical=0, issuer=None, _pyfree=1): doesn't support getting the publickeyidentifier from the issuer to create the authoritykeyidentifier extension. ''' - if name == 'subjectKeyIdentifier' and \ - value.strip('0123456789abcdefABCDEF:') is not '': - raise salt.exceptions.SaltInvocationError( - 'value must be precomputed hash') + if name == 'subjectKeyIdentifier' and value.strip('0123456789abcdefABCDEF:') is not '': + raise salt.exceptions.SaltInvocationError('value must be precomputed hash') # ensure name and value are bytes name = salt.utils.stringutils.to_str(name) @@ -145,7 +139,7 @@ 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() # pylint: disable=no-member + lhash = M2Crypto.m2.x509v3_lhash() # pylint: disable=no-member ctx = M2Crypto.m2.x509v3_set_conf_lhash( lhash) # pylint: disable=no-member # ctx not zeroed @@ -196,10 +190,8 @@ def _get_csr_extensions(csr): csrtempfile.flush() csryaml = _parse_openssl_req(csrtempfile.name) csrtempfile.close() - if csryaml and 'Requested Extensions' in \ - csryaml['Certificate Request']['Data']: - csrexts = \ - csryaml['Certificate Request']['Data']['Requested Extensions'] + if csryaml and 'Requested Extensions' in csryaml['Certificate Request']['Data']: + csrexts = csryaml['Certificate Request']['Data']['Requested Extensions'] if not csrexts: return ret @@ -294,7 +286,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): @@ -321,9 +313,11 @@ def _text_or_file(input_): ''' if os.path.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): @@ -341,7 +335,7 @@ def _parse_subject(subject): ret[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) return ret @@ -520,8 +514,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 @@ -599,8 +593,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 @@ -629,12 +623,10 @@ def read_csr(csr): # Get size returns in bytes. The world thinks of key sizes in bits. '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() + '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 @@ -936,7 +928,7 @@ def create_crl( # pylint: disable=too-many-arguments,too-many-locals # 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' ) @@ -961,8 +953,7 @@ def create_crl( # pylint: disable=too-many-arguments,too-many-locals continue if 'revocation_date' not in rev_item: - rev_item['revocation_date'] = datetime.datetime\ - .now().strftime('%Y-%m-%d %H:%M:%S') + rev_item['revocation_date'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') rev_date = datetime.datetime.strptime( rev_item['revocation_date'], '%Y-%m-%d %H:%M:%S') @@ -1001,8 +992,9 @@ def create_crl( # pylint: disable=too-many-arguments,too-many-locals try: 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.') + log.warning('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) @@ -1041,8 +1033,7 @@ def sign_remote_certificate(argdic, **kwargs): if 'signing_policy' in argdic: signing_policy = _get_signing_policy(argdic['signing_policy']) if not signing_policy: - return 'Signing policy {0} does not exist.'.format( - argdic['signing_policy']) + return 'Signing policy {0} does not exist.'.format(argdic['signing_policy']) if isinstance(signing_policy, list): dict_ = {} @@ -1079,6 +1070,7 @@ def get_signing_policy(signing_policy_name): signing_policy = _get_signing_policy(signing_policy_name) if not signing_policy: return 'Signing policy {0} does not exist.'.format(signing_policy_name) + if isinstance(signing_policy, list): dict_ = {} for item in signing_policy: @@ -1091,10 +1083,9 @@ def get_signing_policy(signing_policy_name): pass try: - signing_policy['signing_cert'] = get_pem_entry( - signing_policy['signing_cert'], 'CERTIFICATE') + signing_policy['signing_cert'] = get_pem_entry(signing_policy['signing_cert'], 'CERTIFICATE') except KeyError: - pass + log.debug('Unable to get "certificate" PEM entry') return signing_policy @@ -1345,8 +1336,7 @@ def create_certificate( signing_private_key='/etc/pki/myca.key' csr='/etc/pki/myca.csr'} ''' - if not path and not text and \ - ('testrun' not in kwargs or kwargs['testrun'] is False): + if not path and not text and ('testrun' not in kwargs or kwargs['testrun'] is False): raise salt.exceptions.SaltInvocationError( 'Either path or text must be specified.') if path and text: @@ -1375,8 +1365,7 @@ def create_certificate( # Including listen_in and preqreuired because they are not included # in STATE_INTERNAL_KEYWORDS # for salt 2014.7.2 - for ignore in list(_STATE_INTERNAL_KEYWORDS) + \ - ['listen_in', 'preqrequired', '__prerequired__']: + for ignore in list(_STATE_INTERNAL_KEYWORDS) + ['listen_in', 'preqrequired', '__prerequired__']: kwargs.pop(ignore, None) certs = __salt__['publish.publish']( @@ -1483,8 +1472,7 @@ def create_certificate( continue # Use explicitly set values first, fall back to CSR values. - extval = kwargs.get(extname) or kwargs.get(extlongname) or \ - csrexts.get(extname) or csrexts.get(extlongname) + extval = kwargs.get(extname) or kwargs.get(extlongname) or csrexts.get(extname) or csrexts.get(extlongname) critical = False if extval.startswith('critical '): @@ -1607,8 +1595,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") + 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") if 'private_key' not in kwargs: raise salt.exceptions.SaltInvocationError('private_key is required') @@ -1620,11 +1608,9 @@ def create_csr(path=None, text=False, **kwargs): kwargs['private_key_passphrase'] = None if 'public_key_passphrase' not in kwargs: kwargs['public_key_passphrase'] = None - if kwargs['public_key_passphrase'] and not kwargs[ - 'private_key_passphrase']: + if kwargs['public_key_passphrase'] and not kwargs['private_key_passphrase']: kwargs['private_key_passphrase'] = kwargs['public_key_passphrase'] - if kwargs['private_key_passphrase'] and not kwargs[ - 'public_key_passphrase']: + if kwargs['private_key_passphrase'] and not kwargs['public_key_passphrase']: kwargs['public_key_passphrase'] = kwargs['private_key_passphrase'] csr.set_pubkey(get_public_key(kwargs['public_key'], @@ -1668,18 +1654,10 @@ 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']), 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): @@ -1704,8 +1682,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(certificate, signing_pub_key=None, @@ -1759,9 +1736,8 @@ 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() @@ -1782,10 +1758,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): @@ -1822,8 +1795,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 @@ -1846,6 +1820,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): @@ -1855,18 +1830,13 @@ def will_expire(certificate, days): cert = _get_certificate_obj(certificate) - _check_time = datetime.datetime.utcnow() + \ - datetime.timedelta(days=days) + _check_time = datetime.datetime.utcnow() + datetime.timedelta(days=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 7bb941f393..3ba4f79c79 100644 --- a/salt/states/x509.py +++ b/salt/states/x509.py @@ -163,6 +163,7 @@ import copy # Import Salt Libs import salt.exceptions +import salt.utils.stringutils # Import 3rd-party libs from salt.ext import six @@ -170,7 +171,7 @@ from salt.ext import six try: from M2Crypto.RSA import RSAError except ImportError: - pass + RSAError = Exception('RSA Error') def __virtual__(): @@ -180,7 +181,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): @@ -267,7 +268,8 @@ def private_key_managed(name, new: Always create a new key. Defaults to False. - Combining new with :mod:`prereq <salt.states.requsities.preqreq>`, or when used as part of a `managed_private_key` can allow key rotation whenever a new certificiate is generated. + Combining new with :mod:`prereq <salt.states.requsities.preqreq>`, or when used as part of a + `managed_private_key` can allow key rotation whenever a new certificiate is generated. overwrite: Overwrite an existing private key if the provided passphrase cannot decrypt it. @@ -453,8 +455,10 @@ def certificate_managed(name, 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']) + 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 @@ -664,8 +668,10 @@ def crl_managed(name, else: current = '{0} does not exist.'.format(name) - new_crl = __salt__['x509.create_crl'](text=True, signing_private_key=signing_private_key, signing_private_key_passphrase=signing_private_key_passphrase, - signing_cert=signing_cert, revoked=revoked, days_valid=days_valid, digest=digest, include_expired=include_expired) + new_crl = __salt__['x509.create_crl'](text=True, signing_private_key=signing_private_key, + signing_private_key_passphrase=signing_private_key_passphrase, + signing_cert=signing_cert, revoked=revoked, days_valid=days_valid, + digest=digest, include_expired=include_expired) new = __salt__['x509.read_crl'](crl=new_crl) new_comp = new.copy() @@ -707,6 +713,6 @@ def pem_managed(name, Any arguments supported by :state:`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) diff --git a/tests/unit/modules/test_x509.py b/tests/unit/modules/test_x509.py index 98fc84146e..91e13476b5 100644 --- a/tests/unit/modules/test_x509.py +++ b/tests/unit/modules/test_x509.py @@ -67,10 +67,11 @@ class X509TestCase(TestCase, LoaderModuleMockMixin): subj = FakeSubject() x509._parse_subject(subj) - x509.log.trace.assert_called_once() - assert x509.log.trace.call_args[0][0] == "Missing attribute '%s'. Error: %s" - assert x509.log.trace.call_args[0][1] == list(subj.nid.keys())[0] - assert isinstance(x509.log.trace.call_args[0][2], TypeError) + x509.log.debug.assert_called_once() + + assert x509.log.debug.call_args[0][0] == "Missing attribute '%s'. Error: %s" + assert x509.log.debug.call_args[0][1] == list(subj.nid.keys())[0] + assert isinstance(x509.log.debug.call_args[0][2], TypeError) @skipIf(not HAS_M2CRYPTO, 'Skipping, M2Crypt is unavailble') def test_get_pem_entry(self): -- 2.18.0
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
