internet/community.crypto
3
0
Fork 0
mirror of https://github.com/ansible-collections/community.crypto.git synced 2026-04-16 22:01:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove PyOpenSSL backends (except for openssl_pkcs12) (#273)

Browse Source 
* Remove Ubuntu 16.04 (Xenial Xerus) from CI.

* Removing PyOpenSSL backend from everywhere but openssl_pkcs12.

* Remove PyOpenSSL support from module_utils that's not needed for openssl_pkcs12.

* Add changelog fragment.
This commit is contained in:
Felix Fontein
2021-09-28 17:46:35 +02:00
committed by GitHub
parent 24e7d07973
commit f644db3c79
72 changed files with 227 additions and 2638 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
plugins/module_utils/crypto/__init__.py
View File

@@ -25,7 +25,6 @@ __metaclass__ = type
# please stop doing so.
from .basic import (
HAS_PYOPENSSL,
CRYPTOGRAPHY_HAS_X25519,
CRYPTOGRAPHY_HAS_X25519_FULL,
CRYPTOGRAPHY_HAS_X448,
@@ -74,12 +73,6 @@ from ._objects import OID_LOOKUP as _OID_LOOKUP
from ._objects import NORMALIZE_NAMES as _NORMALIZE_NAMES
from ._objects import NORMALIZE_NAMES_SHORT as _NORMALIZE_NAMES_SHORT
from .pyopenssl_support import (
pyopenssl_normalize_name,
pyopenssl_get_extensions_from_cert,
pyopenssl_get_extensions_from_csr,
)
from .support import (
get_fingerprint_of_bytes,
get_fingerprint,

8
plugins/module_utils/crypto/basic.py
View File

@@ -22,14 +22,6 @@ __metaclass__ = type
from distutils.version import LooseVersion
try:
import OpenSSL # noqa
from OpenSSL import crypto # noqa
HAS_PYOPENSSL = True
except ImportError:
# Error handled in the calling module.
HAS_PYOPENSSL = False
try:
import cryptography
from cryptography import x509

88
plugins/module_utils/crypto/module_backends/certificate.py
View File

@@ -38,18 +38,6 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.module_bac
)
MINIMAL_CRYPTOGRAPHY_VERSION = '1.6'
MINIMAL_PYOPENSSL_VERSION = '0.15'
PYOPENSSL_IMP_ERR = None
try:
import OpenSSL
from OpenSSL import crypto
PYOPENSSL_VERSION = LooseVersion(OpenSSL.__version__)
except ImportError:
PYOPENSSL_IMP_ERR = traceback.format_exc()
PYOPENSSL_FOUND = False
else:
PYOPENSSL_FOUND = True
CRYPTOGRAPHY_IMP_ERR = None
CRYPTOGRAPHY_VERSION = None
@@ -173,43 +161,12 @@ class CertificateBackend(object):
def _check_privatekey(self):
"""Check whether provided parameters match, assuming self.existing_certificate and self.privatekey have been populated."""
if self.backend == 'pyopenssl':
ctx = OpenSSL.SSL.Context(OpenSSL.SSL.TLSv1_2_METHOD)
ctx.use_privatekey(self.privatekey)
ctx.use_certificate(self.existing_certificate)
try:
ctx.check_privatekey()
return True
except OpenSSL.SSL.Error:
return False
elif self.backend == 'cryptography':
if self.backend == 'cryptography':
return cryptography_compare_public_keys(self.existing_certificate.public_key(), self.privatekey.public_key())
def _check_csr(self):
"""Check whether provided parameters match, assuming self.existing_certificate and self.csr have been populated."""
if self.backend == 'pyopenssl':
# Verify that CSR is signed by certificate's private key
try:
self.csr.verify(self.existing_certificate.get_pubkey())
except OpenSSL.crypto.Error:
return False
# Check subject
if self.check_csr_subject and self.csr.get_subject() != self.existing_certificate.get_subject():
return False
# Check extensions
if not self.check_csr_extensions:
return True
csr_extensions = self.csr.get_extensions()
cert_extension_count = self.existing_certificate.get_extension_count()
if len(csr_extensions) != cert_extension_count:
return False
for extension_number in range(0, cert_extension_count):
cert_extension = self.existing_certificate.get_extension(extension_number)
csr_extension = filter(lambda extension: extension.get_short_name() == cert_extension.get_short_name(), csr_extensions)
if cert_extension.get_data() != list(csr_extension)[0].get_data():
return False
return True
elif self.backend == 'cryptography':
if self.backend == 'cryptography':
# Verify that CSR is signed by certificate's private key
if not self.csr.is_signature_valid:
return False
@@ -244,10 +201,6 @@ class CertificateBackend(object):
def _check_subject_key_identifier(self):
"""Check whether Subject Key Identifier matches, assuming self.existing_certificate has been populated."""
if self.backend != 'cryptography':
# We do not support SKI with pyOpenSSL backend
return True
# Get hold of certificate's SKI
try:
ext = self.existing_certificate.extensions.get_extension_for_class(x509.SubjectKeyIdentifier)
@@ -328,10 +281,6 @@ class CertificateProvider(object):
def needs_version_two_certs(self, module):
"""Whether the provider needs to create a version 2 certificate."""
def needs_pyopenssl_get_extensions(self, module):
"""Whether the provider needs to use get_extensions() with pyOpenSSL."""
return True
@abc.abstractmethod
def create_backend(self, module, backend):
"""Create an implementation for a backend.
@@ -352,45 +301,22 @@ def select_backend(module, backend, provider):
if backend == 'auto':
# Detect what backend we can use
can_use_cryptography = CRYPTOGRAPHY_FOUND and CRYPTOGRAPHY_VERSION >= LooseVersion(MINIMAL_CRYPTOGRAPHY_VERSION)
can_use_pyopenssl = PYOPENSSL_FOUND and PYOPENSSL_VERSION >= LooseVersion(MINIMAL_PYOPENSSL_VERSION)
# If cryptography is available we'll use it
if can_use_cryptography:
backend = 'cryptography'
elif can_use_pyopenssl:
backend = 'pyopenssl'
if provider.needs_version_two_certs(module):
module.warn('crypto backend forced to pyopenssl. The cryptography library does not support v2 certificates')
backend = 'pyopenssl'
# Fail if no backend has been found
if backend == 'auto':
module.fail_json(msg=("Can't detect any of the required Python libraries "
"cryptography (>= {0}) or PyOpenSSL (>= {1})").format(
MINIMAL_CRYPTOGRAPHY_VERSION,
MINIMAL_PYOPENSSL_VERSION))
module.fail_json(msg=("Can't detect the required Python library "
"cryptography (>= {0})").format(MINIMAL_CRYPTOGRAPHY_VERSION))
if backend == 'pyopenssl':
module.deprecate('The module is using the PyOpenSSL backend. This backend has been deprecated',
version='2.0.0', collection_name='community.crypto')
if not PYOPENSSL_FOUND:
module.fail_json(msg=missing_required_lib('pyOpenSSL >= {0}'.format(MINIMAL_PYOPENSSL_VERSION)),
exception=PYOPENSSL_IMP_ERR)
if provider.needs_pyopenssl_get_extensions(module):
try:
getattr(crypto.X509Req, 'get_extensions')
except AttributeError:
module.fail_json(msg='You need to have PyOpenSSL>=0.15')
elif backend == 'cryptography':
if backend == 'cryptography':
if not CRYPTOGRAPHY_FOUND:
module.fail_json(msg=missing_required_lib('cryptography >= {0}'.format(MINIMAL_CRYPTOGRAPHY_VERSION)),
exception=CRYPTOGRAPHY_IMP_ERR)
if provider.needs_version_two_certs(module):
module.fail_json(msg='The cryptography backend does not support v2 certificates, '
'use select_crypto_backend=pyopenssl for v2 certificates')
module.fail_json(msg='The cryptography backend does not support v2 certificates')
return provider.create_backend(module, backend)
@@ -402,7 +328,7 @@ def get_certificate_argument_spec():
force=dict(type='bool', default=False,),
csr_path=dict(type='path'),
csr_content=dict(type='str'),
select_crypto_backend=dict(type='str', default='auto', choices=['auto', 'cryptography', 'pyopenssl']),
select_crypto_backend=dict(type='str', default='auto', choices=['auto', 'cryptography']),
# General properties of a certificate
privatekey_path=dict(type='path'),

164
plugins/module_utils/crypto/module_backends/certificate_assertonly.py
View File

@@ -25,10 +25,6 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptograp
cryptography_parse_key_usage_params,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.pyopenssl_support import (
pyopenssl_normalize_name_attribute,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.certificate import (
CertificateBackend,
CertificateProvider,
@@ -460,164 +456,6 @@ class AssertOnlyCertificateBackendCryptography(AssertOnlyCertificateBackend):
return self.existing_certificate.not_valid_before, valid_in_date, self.existing_certificate.not_valid_after
class AssertOnlyCertificateBackendPyOpenSSL(AssertOnlyCertificateBackend):
"""validate the supplied certificate."""
def __init__(self, module):
super(AssertOnlyCertificateBackendPyOpenSSL, self).__init__(module, 'pyopenssl')
# Ensure inputs are properly sanitized before comparison.
for param in ['signature_algorithms', 'key_usage', 'extended_key_usage',
'subject_alt_name', 'subject', 'issuer', 'not_before',
'not_after', 'valid_at', 'invalid_at']:
attr = getattr(self, param)
if isinstance(attr, list) and attr:
if isinstance(attr[0], str):
setattr(self, param, [to_bytes(item) for item in attr])
elif isinstance(attr[0], tuple):
setattr(self, param, [(to_bytes(item[0]), to_bytes(item[1])) for item in attr])
elif isinstance(attr, tuple):
setattr(self, param, dict((to_bytes(k), to_bytes(v)) for (k, v) in attr.items()))
elif isinstance(attr, dict):
setattr(self, param, dict((to_bytes(k), to_bytes(v)) for (k, v) in attr.items()))
elif isinstance(attr, str):
setattr(self, param, to_bytes(attr))
def _validate_privatekey(self):
ctx = OpenSSL.SSL.Context(OpenSSL.SSL.TLSv1_2_METHOD)
ctx.use_privatekey(self.privatekey)
ctx.use_certificate(self.existing_certificate)
try:
ctx.check_privatekey()
return True
except OpenSSL.SSL.Error:
return False
def _validate_csr_signature(self):
try:
self.csr.verify(self.existing_certificate.get_pubkey())
except OpenSSL.crypto.Error:
return False
def _validate_csr_subject(self):
if self.csr.get_subject() != self.existing_certificate.get_subject():
return False
def _validate_csr_extensions(self):
csr_extensions = self.csr.get_extensions()
cert_extension_count = self.existing_certificate.get_extension_count()
if len(csr_extensions) != cert_extension_count:
return False
for extension_number in range(0, cert_extension_count):
cert_extension = self.existing_certificate.get_extension(extension_number)
csr_extension = filter(lambda extension: extension.get_short_name() == cert_extension.get_short_name(), csr_extensions)
if cert_extension.get_data() != list(csr_extension)[0].get_data():
return False
return True
def _validate_signature_algorithms(self):
if self.existing_certificate.get_signature_algorithm() not in self.signature_algorithms:
return self.existing_certificate.get_signature_algorithm()
def _validate_subject(self):
expected_subject = [(OpenSSL._util.lib.OBJ_txt2nid(sub[0]), sub[1]) for sub in self.subject]
cert_subject = self.existing_certificate.get_subject().get_components()
current_subject = [(OpenSSL._util.lib.OBJ_txt2nid(sub[0]), sub[1]) for sub in cert_subject]
if not compare_sets(expected_subject, current_subject, self.subject_strict):
return expected_subject, current_subject
def _validate_issuer(self):
expected_issuer = [(OpenSSL._util.lib.OBJ_txt2nid(iss[0]), iss[1]) for iss in self.issuer]
cert_issuer = self.existing_certificate.get_issuer().get_components()
current_issuer = [(OpenSSL._util.lib.OBJ_txt2nid(iss[0]), iss[1]) for iss in cert_issuer]
if not compare_sets(expected_issuer, current_issuer, self.issuer_strict):
return self.issuer, cert_issuer
def _validate_has_expired(self):
# The following 3 lines are the same as the current PyOpenSSL code for cert.has_expired().
# Older version of PyOpenSSL have a buggy implementation,
# to avoid issues with those we added the code from a more recent release here.
time_string = to_native(self.existing_certificate.get_notAfter())
not_after = datetime.datetime.strptime(time_string, "%Y%m%d%H%M%SZ")
cert_expired = not_after < datetime.datetime.utcnow()
return cert_expired
def _validate_version(self):
# Version numbers in certs are off by one:
# v1: 0, v2: 1, v3: 2 ...
return self.existing_certificate.get_version() + 1
def _validate_key_usage(self):
found = False
for extension_idx in range(0, self.existing_certificate.get_extension_count()):
extension = self.existing_certificate.get_extension(extension_idx)
if extension.get_short_name() == b'keyUsage':
found = True
expected_extension = crypto.X509Extension(b"keyUsage", False, b', '.join(self.key_usage))
key_usage = [usage.strip() for usage in to_text(expected_extension, errors='surrogate_or_strict').split(',')]
current_ku = [usage.strip() for usage in to_text(extension, errors='surrogate_or_strict').split(',')]
if not compare_sets(key_usage, current_ku, self.key_usage_strict):
return self.key_usage, str(extension).split(', ')
if not found:
# This is only bad if the user specified a non-empty list
if self.key_usage:
return NO_EXTENSION
def _validate_extended_key_usage(self):
found = False
for extension_idx in range(0, self.existing_certificate.get_extension_count()):
extension = self.existing_certificate.get_extension(extension_idx)
if extension.get_short_name() == b'extendedKeyUsage':
found = True
extKeyUsage = [OpenSSL._util.lib.OBJ_txt2nid(keyUsage) for keyUsage in self.extended_key_usage]
current_xku = [OpenSSL._util.lib.OBJ_txt2nid(usage.strip()) for usage in
to_bytes(extension, errors='surrogate_or_strict').split(b',')]
if not compare_sets(extKeyUsage, current_xku, self.extended_key_usage_strict):
return self.extended_key_usage, str(extension).split(', ')
if not found:
# This is only bad if the user specified a non-empty list
if self.extended_key_usage:
return NO_EXTENSION
def _validate_subject_alt_name(self):
found = False
for extension_idx in range(0, self.existing_certificate.get_extension_count()):
extension = self.existing_certificate.get_extension(extension_idx)
if extension.get_short_name() == b'subjectAltName':
found = True
l_altnames = [pyopenssl_normalize_name_attribute(altname.strip()) for altname in
to_text(extension, errors='surrogate_or_strict').split(', ')]
sans = [pyopenssl_normalize_name_attribute(to_text(san, errors='surrogate_or_strict')) for san in self.subject_alt_name]
if not compare_sets(sans, l_altnames, self.subject_alt_name_strict):
return self.subject_alt_name, l_altnames
if not found:
# This is only bad if the user specified a non-empty list
if self.subject_alt_name:
return NO_EXTENSION
def _validate_not_before(self):
return self.existing_certificate.get_notBefore()
def _validate_not_after(self):
return self.existing_certificate.get_notAfter()
def _validate_valid_at(self):
rt = get_relative_time_option(self.valid_at, "valid_at", backend=self.backend)
rt = to_bytes(rt, errors='surrogate_or_strict')
return self.existing_certificate.get_notBefore(), rt, self.existing_certificate.get_notAfter()
def _validate_invalid_at(self):
rt = get_relative_time_option(self.invalid_at, "invalid_at", backend=self.backend)
rt = to_bytes(rt, errors='surrogate_or_strict')
return self.existing_certificate.get_notBefore(), rt, self.existing_certificate.get_notAfter()
def _validate_valid_in(self):
valid_in_asn1 = get_relative_time_option(self.valid_in, "valid_in", backend=self.backend)
valid_in_date = to_bytes(valid_in_asn1, errors='surrogate_or_strict')
return self.existing_certificate.get_notBefore(), valid_in_date, self.existing_certificate.get_notAfter()
class AssertOnlyCertificateProvider(CertificateProvider):
def validate_module_args(self, module):
module.deprecate("The 'assertonly' provider is deprecated; please see the examples of "
@@ -630,8 +468,6 @@ class AssertOnlyCertificateProvider(CertificateProvider):
def create_backend(self, module, backend):
if backend == 'cryptography':
return AssertOnlyCertificateBackendCryptography(module)
if backend == 'pyopenssl':
return AssertOnlyCertificateBackendPyOpenSSL(module)
def add_assertonly_provider_to_argument_spec(argument_spec):

19
plugins/module_utils/crypto/module_backends/certificate_entrust.py
View File

@@ -58,18 +58,7 @@ class EntrustCertificateBackend(CertificateBackend):
# We want to always force behavior of trying to use the organization provided in the CSR.
# To that end we need to parse out the organization from the CSR.
self.csr_org = None
if self.backend == 'pyopenssl':
csr_subject = self.csr.get_subject()
csr_subject_components = csr_subject.get_components()
for k, v in csr_subject_components:
if k.upper() == 'O':
# Entrust does not support multiple validated organizations in a single certificate
if self.csr_org is not None:
self.module.fail_json(msg=("Entrust provider does not currently support multiple validated organizations. Multiple organizations "
"found in Subject DN: '{0}'. ".format(csr_subject)))
else:
self.csr_org = v
elif self.backend == 'cryptography':
if self.backend == 'cryptography':
csr_subject_orgs = self.csr.subject.get_attributes_for_oid(NameOID.ORGANIZATION_NAME)
if len(csr_subject_orgs) == 1:
self.csr_org = csr_subject_orgs[0].value
@@ -162,11 +151,7 @@ class EntrustCertificateBackend(CertificateBackend):
if self.existing_certificate:
serial_number = None
expiry = None
if self.backend == 'pyopenssl':
serial_number = "{0:X}".format(self.existing_certificate.get_serial_number())
time_string = to_native(self.existing_certificate.get_notAfter())
expiry = datetime.datetime.strptime(time_string, "%Y%m%d%H%M%SZ")
elif self.backend == 'cryptography':
if self.backend == 'cryptography':
serial_number = "{0:X}".format(cryptography_serial_number_of_cert(self.existing_certificate))
expiry = self.existing_certificate.not_valid_after

201
plugins/module_utils/crypto/module_backends/certificate_info.py
View File

@@ -33,37 +33,11 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptograp
cryptography_serial_number_of_cert,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.pyopenssl_support import (
pyopenssl_get_extensions_from_cert,
pyopenssl_normalize_name,
pyopenssl_normalize_name_attribute,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.publickey_info import (
get_publickey_info,
)
MINIMAL_CRYPTOGRAPHY_VERSION = '1.6'
MINIMAL_PYOPENSSL_VERSION = '0.15'
PYOPENSSL_IMP_ERR = None
try:
import OpenSSL
from OpenSSL import crypto
PYOPENSSL_VERSION = LooseVersion(OpenSSL.__version__)
if OpenSSL.SSL.OPENSSL_VERSION_NUMBER >= 0x10100000:
# OpenSSL 1.1.0 or newer
OPENSSL_MUST_STAPLE_NAME = b"tlsfeature"
OPENSSL_MUST_STAPLE_VALUE = b"status_request"
else:
# OpenSSL 1.0.x or older
OPENSSL_MUST_STAPLE_NAME = b"1.3.6.1.5.5.7.1.24"
OPENSSL_MUST_STAPLE_VALUE = b"DER:30:03:02:01:05"
except ImportError:
PYOPENSSL_IMP_ERR = traceback.format_exc()
PYOPENSSL_FOUND = False
else:
PYOPENSSL_FOUND = True
CRYPTOGRAPHY_IMP_ERR = None
try:
@@ -209,20 +183,19 @@ class CertificateInfoRetrieval(object):
result['fingerprints'] = get_fingerprint_of_bytes(
self._get_der_bytes(), prefer_one=prefer_one_fingerprint)
if self.backend != 'pyopenssl':
ski = self._get_subject_key_identifier()
if ski is not None:
ski = to_native(binascii.hexlify(ski))
ski = ':'.join([ski[i:i + 2] for i in range(0, len(ski), 2)])
result['subject_key_identifier'] = ski
ski = self._get_subject_key_identifier()
if ski is not None:
ski = to_native(binascii.hexlify(ski))
ski = ':'.join([ski[i:i + 2] for i in range(0, len(ski), 2)])
result['subject_key_identifier'] = ski
aki, aci, acsn = self._get_authority_key_identifier()
if aki is not None:
aki = to_native(binascii.hexlify(aki))
aki = ':'.join([aki[i:i + 2] for i in range(0, len(aki), 2)])
result['authority_key_identifier'] = aki
result['authority_cert_issuer'] = aci
result['authority_cert_serial_number'] = acsn
aki, aci, acsn = self._get_authority_key_identifier()
if aki is not None:
aki = to_native(binascii.hexlify(aki))
aki = ':'.join([aki[i:i + 2] for i in range(0, len(aki), 2)])
result['authority_key_identifier'] = aki
result['authority_cert_issuer'] = aci
result['authority_cert_serial_number'] = acsn
result['serial_number'] = self._get_serial_number()
result['extensions_by_oid'] = self._get_all_extensions()
@@ -392,136 +365,9 @@ class CertificateInfoRetrievalCryptography(CertificateInfoRetrieval):
return None
class CertificateInfoRetrievalPyOpenSSL(CertificateInfoRetrieval):
"""validate the supplied certificate."""
def __init__(self, module, content):
super(CertificateInfoRetrievalPyOpenSSL, self).__init__(module, 'pyopenssl', content)
def _get_der_bytes(self):
return crypto.dump_certificate(crypto.FILETYPE_ASN1, self.cert)
def _get_signature_algorithm(self):
return to_text(self.cert.get_signature_algorithm())
def __get_name(self, name):
result = []
for sub in name.get_components():
result.append([pyopenssl_normalize_name(sub[0]), to_text(sub[1])])
return result
def _get_subject_ordered(self):
return self.__get_name(self.cert.get_subject())
def _get_issuer_ordered(self):
return self.__get_name(self.cert.get_issuer())
def _get_version(self):
# Version numbers in certs are off by one:
# v1: 0, v2: 1, v3: 2 ...
return self.cert.get_version() + 1
def _get_extension(self, short_name):
for extension_idx in range(0, self.cert.get_extension_count()):
extension = self.cert.get_extension(extension_idx)
if extension.get_short_name() == short_name:
result = [
pyopenssl_normalize_name(usage.strip()) for usage in to_text(extension, errors='surrogate_or_strict').split(',')
]
return sorted(result), bool(extension.get_critical())
return None, False
def _get_key_usage(self):
return self._get_extension(b'keyUsage')
def _get_extended_key_usage(self):
return self._get_extension(b'extendedKeyUsage')
def _get_basic_constraints(self):
return self._get_extension(b'basicConstraints')
def _get_ocsp_must_staple(self):
extensions = [self.cert.get_extension(i) for i in range(0, self.cert.get_extension_count())]
oms_ext = [
ext for ext in extensions
if to_bytes(ext.get_short_name()) == OPENSSL_MUST_STAPLE_NAME and to_bytes(ext) == OPENSSL_MUST_STAPLE_VALUE
]
if OpenSSL.SSL.OPENSSL_VERSION_NUMBER < 0x10100000:
# Older versions of libssl don't know about OCSP Must Staple
oms_ext.extend([ext for ext in extensions if ext.get_short_name() == b'UNDEF' and ext.get_data() == b'\x30\x03\x02\x01\x05'])
if oms_ext:
return True, bool(oms_ext[0].get_critical())
else:
return None, False
def _get_subject_alt_name(self):
for extension_idx in range(0, self.cert.get_extension_count()):
extension = self.cert.get_extension(extension_idx)
if extension.get_short_name() == b'subjectAltName':
result = [pyopenssl_normalize_name_attribute(altname.strip()) for altname in
to_text(extension, errors='surrogate_or_strict').split(', ')]
return result, bool(extension.get_critical())
return None, False
def get_not_before(self):
time_string = to_native(self.cert.get_notBefore())
return datetime.datetime.strptime(time_string, "%Y%m%d%H%M%SZ")
def get_not_after(self):
time_string = to_native(self.cert.get_notAfter())
return datetime.datetime.strptime(time_string, "%Y%m%d%H%M%SZ")
def _get_public_key_pem(self):
try:
return crypto.dump_publickey(
crypto.FILETYPE_PEM,
self.cert.get_pubkey(),
)
except AttributeError:
try:
# pyOpenSSL < 16.0:
bio = crypto._new_mem_buf()
rc = crypto._lib.PEM_write_bio_PUBKEY(bio, self.cert.get_pubkey()._pkey)
if rc != 1:
crypto._raise_current_error()
return crypto._bio_to_string(bio)
except AttributeError:
self.module.warn('Your pyOpenSSL version does not support dumping public keys. '
'Please upgrade to version 16.0 or newer, or use the cryptography backend.')
def _get_public_key_object(self):
return self.cert.get_pubkey()
def _get_subject_key_identifier(self):
# Won't be implemented
return None
def _get_authority_key_identifier(self):
# Won't be implemented
return None, None, None
def _get_serial_number(self):
return self.cert.get_serial_number()
def _get_all_extensions(self):
return pyopenssl_get_extensions_from_cert(self.cert)
def _get_ocsp_uri(self):
for i in range(self.cert.get_extension_count()):
ext = self.cert.get_extension(i)
if ext.get_short_name() == b'authorityInfoAccess':
v = str(ext)
m = re.search('^OCSP - URI:(.*)$', v, flags=re.MULTILINE)
if m:
return m.group(1)
return None
def get_certificate_info(module, backend, content, prefer_one_fingerprint=False):
if backend == 'cryptography':
info = CertificateInfoRetrievalCryptography(module, content)
elif backend == 'pyopenssl':
info = CertificateInfoRetrievalPyOpenSSL(module, content)
return info.get_info(prefer_one_fingerprint=prefer_one_fingerprint)
@@ -529,34 +375,17 @@ def select_backend(module, backend, content):
if backend == 'auto':
# Detection what is possible
can_use_cryptography = CRYPTOGRAPHY_FOUND and CRYPTOGRAPHY_VERSION >= LooseVersion(MINIMAL_CRYPTOGRAPHY_VERSION)
can_use_pyopenssl = PYOPENSSL_FOUND and PYOPENSSL_VERSION >= LooseVersion(MINIMAL_PYOPENSSL_VERSION)
# First try cryptography, then pyOpenSSL
# Try cryptography
if can_use_cryptography:
backend = 'cryptography'
elif can_use_pyopenssl:
backend = 'pyopenssl'
# Success?
if backend == 'auto':
module.fail_json(msg=("Can't detect any of the required Python libraries "
"cryptography (>= {0}) or PyOpenSSL (>= {1})").format(
MINIMAL_CRYPTOGRAPHY_VERSION,
MINIMAL_PYOPENSSL_VERSION))
"cryptography (>= {0})").format(MINIMAL_CRYPTOGRAPHY_VERSION))
if backend == 'pyopenssl':
if not PYOPENSSL_FOUND:
module.fail_json(msg=missing_required_lib('pyOpenSSL >= {0}'.format(MINIMAL_PYOPENSSL_VERSION)),
exception=PYOPENSSL_IMP_ERR)
try:
getattr(crypto.X509Req, 'get_extensions')
except AttributeError:
module.fail_json(msg='You need to have PyOpenSSL>=0.15 to generate CSRs')
module.deprecate('The module is using the PyOpenSSL backend. This backend has been deprecated',
version='2.0.0', collection_name='community.crypto')
return backend, CertificateInfoRetrievalPyOpenSSL(module, content)
elif backend == 'cryptography':
if backend == 'cryptography':
if not CRYPTOGRAPHY_FOUND:
module.fail_json(msg=missing_required_lib('cryptography >= {0}'.format(MINIMAL_CRYPTOGRAPHY_VERSION)),
exception=CRYPTOGRAPHY_IMP_ERR)

96
plugins/module_utils/crypto/module_backends/certificate_ownca.py
View File

@@ -227,100 +227,6 @@ def generate_serial_number():
return result
class OwnCACertificateBackendPyOpenSSL(CertificateBackend):
def __init__(self, module):
super(OwnCACertificateBackendPyOpenSSL, self).__init__(module, 'pyopenssl')
self.notBefore = get_relative_time_option(self.module.params['ownca_not_before'], 'ownca_not_before', backend=self.backend)
self.notAfter = get_relative_time_option(self.module.params['ownca_not_after'], 'ownca_not_after', backend=self.backend)
self.digest = self.module.params['ownca_digest']
self.version = self.module.params['ownca_version']
self.serial_number = generate_serial_number()
if self.module.params['ownca_create_subject_key_identifier'] != 'create_if_not_provided':
self.module.fail_json(msg='ownca_create_subject_key_identifier cannot be used with the pyOpenSSL backend!')
if self.module.params['ownca_create_authority_key_identifier']:
self.module.warn('ownca_create_authority_key_identifier is ignored by the pyOpenSSL backend!')
self.ca_cert_path = self.module.params['ownca_path']
self.ca_cert_content = self.module.params['ownca_content']
if self.ca_cert_content is not None:
self.ca_cert_content = self.ca_cert_content.encode('utf-8')
self.ca_privatekey_path = self.module.params['ownca_privatekey_path']
self.ca_privatekey_content = self.module.params['ownca_privatekey_content']
if self.ca_privatekey_content is not None:
self.ca_privatekey_content = self.ca_privatekey_content.encode('utf-8')
self.ca_privatekey_passphrase = self.module.params['ownca_privatekey_passphrase']
if self.csr_content is None and not os.path.exists(self.csr_path):
raise CertificateError(
'The certificate signing request file {0} does not exist'.format(self.csr_path)
)
if self.ca_cert_content is None and not os.path.exists(self.ca_cert_path):
raise CertificateError(
'The CA certificate file {0} does not exist'.format(self.ca_cert_path)
)
if self.ca_privatekey_content is None and not os.path.exists(self.ca_privatekey_path):
raise CertificateError(
'The CA private key file {0} does not exist'.format(self.ca_privatekey_path)
)
self._ensure_csr_loaded()
self.ca_cert = load_certificate(
path=self.ca_cert_path,
content=self.ca_cert_content,
)
try:
self.ca_privatekey = load_privatekey(
path=self.ca_privatekey_path,
content=self.ca_privatekey_content,
passphrase=self.ca_privatekey_passphrase
)
except OpenSSLBadPassphraseError as exc:
self.module.fail_json(msg=str(exc))
def generate_certificate(self):
"""(Re-)Generate certificate."""
cert = crypto.X509()
cert.set_serial_number(self.serial_number)
cert.set_notBefore(to_bytes(self.notBefore))
cert.set_notAfter(to_bytes(self.notAfter))
cert.set_subject(self.csr.get_subject())
cert.set_issuer(self.ca_cert.get_subject())
cert.set_version(self.version - 1)
cert.set_pubkey(self.csr.get_pubkey())
cert.add_extensions(self.csr.get_extensions())
cert.sign(self.ca_privatekey, self.digest)
self.cert = cert
def get_certificate_data(self):
"""Return bytes for self.cert."""
return crypto.dump_certificate(crypto.FILETYPE_PEM, self.cert)
def dump(self, include_certificate):
result = super(OwnCACertificateBackendPyOpenSSL, self).dump(include_certificate)
result.update({
'ca_cert': self.ca_cert_path,
'ca_privatekey': self.ca_privatekey_path,
})
if self.module.check_mode:
result.update({
'notBefore': self.notBefore,
'notAfter': self.notAfter,
'serial_number': self.serial_number,
})
else:
if self.cert is None:
self.cert = self.existing_certificate
result.update({
'notBefore': self.cert.get_notBefore(),
'notAfter': self.cert.get_notAfter(),
'serial_number': self.cert.get_serial_number(),
})
return result
class OwnCACertificateProvider(CertificateProvider):
def validate_module_args(self, module):
if module.params['ownca_path'] is None and module.params['ownca_content'] is None:
@@ -334,8 +240,6 @@ class OwnCACertificateProvider(CertificateProvider):
def create_backend(self, module, backend):
if backend == 'cryptography':
return OwnCACertificateBackendCryptography(module)
if backend == 'pyopenssl':
return OwnCACertificateBackendPyOpenSSL(module)
def add_ownca_provider_to_argument_spec(argument_spec):

72
plugins/module_utils/crypto/module_backends/certificate_selfsigned.py
View File

@@ -163,76 +163,6 @@ def generate_serial_number():
return result
class SelfSignedCertificateBackendPyOpenSSL(CertificateBackend):
def __init__(self, module):
super(SelfSignedCertificateBackendPyOpenSSL, self).__init__(module, 'pyopenssl')
if module.params['selfsigned_create_subject_key_identifier'] != 'create_if_not_provided':
module.fail_json(msg='selfsigned_create_subject_key_identifier cannot be used with the pyOpenSSL backend!')
self.notBefore = get_relative_time_option(module.params['selfsigned_not_before'], 'selfsigned_not_before', backend=self.backend)
self.notAfter = get_relative_time_option(module.params['selfsigned_not_after'], 'selfsigned_not_after', backend=self.backend)
self.digest = module.params['selfsigned_digest']
self.version = module.params['selfsigned_version']
self.serial_number = generate_serial_number()
if self.csr_path is not None and not os.path.exists(self.csr_path):
raise CertificateError(
'The certificate signing request file {0} does not exist'.format(self.csr_path)
)
if self.privatekey_content is None and not os.path.exists(self.privatekey_path):
raise CertificateError(
'The private key file {0} does not exist'.format(self.privatekey_path)
)
self._ensure_private_key_loaded()
self._ensure_csr_loaded()
if self.csr is None:
# Create empty CSR on the fly
self.csr = crypto.X509Req()
self.csr.set_pubkey(self.privatekey)
self.csr.sign(self.privatekey, self.digest)
def generate_certificate(self):
"""(Re-)Generate certificate."""
cert = crypto.X509()
cert.set_serial_number(self.serial_number)
cert.set_notBefore(to_bytes(self.notBefore))
cert.set_notAfter(to_bytes(self.notAfter))
cert.set_subject(self.csr.get_subject())
cert.set_issuer(self.csr.get_subject())
cert.set_version(self.version - 1)
cert.set_pubkey(self.csr.get_pubkey())
cert.add_extensions(self.csr.get_extensions())
cert.sign(self.privatekey, self.digest)
self.cert = cert
def get_certificate_data(self):
"""Return bytes for self.cert."""
return crypto.dump_certificate(crypto.FILETYPE_PEM, self.cert)
def dump(self, include_certificate):
result = super(SelfSignedCertificateBackendPyOpenSSL, self).dump(include_certificate)
if self.module.check_mode:
result.update({
'notBefore': self.notBefore,
'notAfter': self.notAfter,
'serial_number': self.serial_number,
})
else:
if self.cert is None:
self.cert = self.existing_certificate
result.update({
'notBefore': self.cert.get_notBefore(),
'notAfter': self.cert.get_notAfter(),
'serial_number': self.cert.get_serial_number(),
})
return result
class SelfSignedCertificateProvider(CertificateProvider):
def validate_module_args(self, module):
if module.params['privatekey_path'] is None and module.params['privatekey_content'] is None:
@@ -244,8 +174,6 @@ class SelfSignedCertificateProvider(CertificateProvider):
def create_backend(self, module, backend):
if backend == 'cryptography':
return SelfSignedCertificateBackendCryptography(module)
if backend == 'pyopenssl':
return SelfSignedCertificateBackendPyOpenSSL(module)
def add_selfsigned_provider_to_argument_spec(argument_spec):

218
plugins/module_utils/crypto/module_backends/csr.py
View File

@@ -43,11 +43,6 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptograp
REVOCATION_REASON_MAP,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.pyopenssl_support import (
pyopenssl_normalize_name_attribute,
pyopenssl_parse_name_constraints,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.csr_info import (
get_csr_info,
)
@@ -55,28 +50,8 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.module_bac
from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.common import ArgumentSpec
MINIMAL_PYOPENSSL_VERSION = '0.15'
MINIMAL_CRYPTOGRAPHY_VERSION = '1.3'
PYOPENSSL_IMP_ERR = None
try:
import OpenSSL
from OpenSSL import crypto
PYOPENSSL_VERSION = LooseVersion(OpenSSL.__version__)
except ImportError:
PYOPENSSL_IMP_ERR = traceback.format_exc()
PYOPENSSL_FOUND = False
else:
PYOPENSSL_FOUND = True
if OpenSSL.SSL.OPENSSL_VERSION_NUMBER >= 0x10100000:
# OpenSSL 1.1.0 or newer
OPENSSL_MUST_STAPLE_NAME = b"tlsfeature"
OPENSSL_MUST_STAPLE_VALUE = b"status_request"
else:
# OpenSSL 1.0.x or older
OPENSSL_MUST_STAPLE_NAME = b"1.3.6.1.5.5.7.1.24"
OPENSSL_MUST_STAPLE_VALUE = b"DER:30:03:02:01:05"
CRYPTOGRAPHY_IMP_ERR = None
try:
import cryptography
@@ -273,174 +248,6 @@ class CertificateSigningRequestBackend(object):
return result
# Implementation with using pyOpenSSL
class CertificateSigningRequestPyOpenSSLBackend(CertificateSigningRequestBackend):
def __init__(self, module):
for o in ('create_subject_key_identifier', ):
if module.params[o]:
module.fail_json(msg='You cannot use {0} with the pyOpenSSL backend!'.format(o))
for o in ('subject_key_identifier', 'authority_key_identifier', 'authority_cert_issuer', 'authority_cert_serial_number', 'crl_distribution_points'):
if module.params[o] is not None:
module.fail_json(msg='You cannot use {0} with the pyOpenSSL backend!'.format(o))
super(CertificateSigningRequestPyOpenSSLBackend, self).__init__(module, 'pyopenssl')
def generate_csr(self):
"""(Re-)Generate CSR."""
self._ensure_private_key_loaded()
req = crypto.X509Req()
req.set_version(self.version - 1)
subject = req.get_subject()
for entry in self.subject:
if entry[1] is not None:
# Workaround for https://github.com/pyca/pyopenssl/issues/165
nid = OpenSSL._util.lib.OBJ_txt2nid(to_bytes(entry[0]))
if nid == 0:
raise CertificateSigningRequestError('Unknown subject field identifier "{0}"'.format(entry[0]))
res = OpenSSL._util.lib.X509_NAME_add_entry_by_NID(subject._name, nid, OpenSSL._util.lib.MBSTRING_UTF8, to_bytes(entry[1]), -1, -1, 0)
if res == 0:
raise CertificateSigningRequestError('Invalid value for subject field identifier "{0}": {1}'.format(entry[0], entry[1]))
extensions = []
if self.subjectAltName:
altnames = ', '.join(self.subjectAltName)
try:
extensions.append(crypto.X509Extension(b"subjectAltName", self.subjectAltName_critical, altnames.encode('ascii')))
except OpenSSL.crypto.Error as e:
raise CertificateSigningRequestError(
'Error while parsing Subject Alternative Names {0} (check for missing type prefix, such as "DNS:"!): {1}'.format(
', '.join(["{0}".format(san) for san in self.subjectAltName]), str(e)
)
)
if self.keyUsage:
usages = ', '.join(self.keyUsage)
extensions.append(crypto.X509Extension(b"keyUsage", self.keyUsage_critical, usages.encode('ascii')))
if self.extendedKeyUsage:
usages = ', '.join(self.extendedKeyUsage)
extensions.append(crypto.X509Extension(b"extendedKeyUsage", self.extendedKeyUsage_critical, usages.encode('ascii')))
if self.basicConstraints:
usages = ', '.join(self.basicConstraints)
extensions.append(crypto.X509Extension(b"basicConstraints", self.basicConstraints_critical, usages.encode('ascii')))
if self.name_constraints_permitted or self.name_constraints_excluded:
usages = ', '.join(
['permitted;{0}'.format(name) for name in self.name_constraints_permitted] +
['excluded;{0}'.format(name) for name in self.name_constraints_excluded]
)
extensions.append(crypto.X509Extension(b"nameConstraints", self.name_constraints_critical, usages.encode('ascii')))
if self.ocspMustStaple:
extensions.append(crypto.X509Extension(OPENSSL_MUST_STAPLE_NAME, self.ocspMustStaple_critical, OPENSSL_MUST_STAPLE_VALUE))
if extensions:
req.add_extensions(extensions)
req.set_pubkey(self.privatekey)
req.sign(self.privatekey, self.digest)
self.csr = req
def get_csr_data(self):
"""Return bytes for self.csr."""
return crypto.dump_certificate_request(crypto.FILETYPE_PEM, self.csr)
def _check_csr(self):
def _check_subject(csr):
subject = [(OpenSSL._util.lib.OBJ_txt2nid(to_bytes(sub[0])), to_bytes(sub[1])) for sub in self.subject]
current_subject = [(OpenSSL._util.lib.OBJ_txt2nid(to_bytes(sub[0])), to_bytes(sub[1])) for sub in csr.get_subject().get_components()]
if not set(subject) == set(current_subject):
return False
return True
def _check_subjectAltName(extensions):
altnames_ext = next((ext for ext in extensions if ext.get_short_name() == b'subjectAltName'), '')
altnames = [pyopenssl_normalize_name_attribute(altname.strip()) for altname in
to_text(altnames_ext, errors='surrogate_or_strict').split(',') if altname.strip()]
if self.subjectAltName:
if (set(altnames) != set([pyopenssl_normalize_name_attribute(to_text(name)) for name in self.subjectAltName]) or
altnames_ext.get_critical() != self.subjectAltName_critical):
return False
else:
if altnames:
return False
return True
def _check_keyUsage_(extensions, extName, expected, critical):
usages_ext = [ext for ext in extensions if ext.get_short_name() == extName]
if (not usages_ext and expected) or (usages_ext and not expected):
return False
elif not usages_ext and not expected:
return True
else:
current = [OpenSSL._util.lib.OBJ_txt2nid(to_bytes(usage.strip())) for usage in str(usages_ext[0]).split(',')]
expected = [OpenSSL._util.lib.OBJ_txt2nid(to_bytes(usage)) for usage in expected]
return set(current) == set(expected) and usages_ext[0].get_critical() == critical
def _check_keyUsage(extensions):
usages_ext = [ext for ext in extensions if ext.get_short_name() == b'keyUsage']
if (not usages_ext and self.keyUsage) or (usages_ext and not self.keyUsage):
return False
elif not usages_ext and not self.keyUsage:
return True
else:
# OpenSSL._util.lib.OBJ_txt2nid() always returns 0 for all keyUsage values
# (since keyUsage has a fixed bitfield for these values and is not extensible).
# Therefore, we create an extension for the wanted values, and compare the
# data of the extensions (which is the serialized bitfield).
expected_ext = crypto.X509Extension(b"keyUsage", False, ', '.join(self.keyUsage).encode('ascii'))
return usages_ext[0].get_data() == expected_ext.get_data() and usages_ext[0].get_critical() == self.keyUsage_critical
def _check_extenededKeyUsage(extensions):
return _check_keyUsage_(extensions, b'extendedKeyUsage', self.extendedKeyUsage, self.extendedKeyUsage_critical)
def _check_basicConstraints(extensions):
return _check_keyUsage_(extensions, b'basicConstraints', self.basicConstraints, self.basicConstraints_critical)
def _check_nameConstraints(extensions):
nc_ext = next((ext for ext in extensions if ext.get_short_name() == b'nameConstraints'), '')
permitted, excluded = pyopenssl_parse_name_constraints(nc_ext)
if self.name_constraints_permitted or self.name_constraints_excluded:
if set(permitted) != set([pyopenssl_normalize_name_attribute(to_text(name)) for name in self.name_constraints_permitted]):
return False
if set(excluded) != set([pyopenssl_normalize_name_attribute(to_text(name)) for name in self.name_constraints_excluded]):
return False
if nc_ext.get_critical() != self.name_constraints_critical:
return False
else:
if permitted or excluded:
return False
return True
def _check_ocspMustStaple(extensions):
oms_ext = [ext for ext in extensions if to_bytes(ext.get_short_name()) == OPENSSL_MUST_STAPLE_NAME and to_bytes(ext) == OPENSSL_MUST_STAPLE_VALUE]
if OpenSSL.SSL.OPENSSL_VERSION_NUMBER < 0x10100000:
# Older versions of libssl don't know about OCSP Must Staple
oms_ext.extend([ext for ext in extensions if ext.get_short_name() == b'UNDEF' and ext.get_data() == b'\x30\x03\x02\x01\x05'])
if self.ocspMustStaple:
return len(oms_ext) > 0 and oms_ext[0].get_critical() == self.ocspMustStaple_critical
else:
return len(oms_ext) == 0
def _check_extensions(csr):
extensions = csr.get_extensions()
return (_check_subjectAltName(extensions) and _check_keyUsage(extensions) and
_check_extenededKeyUsage(extensions) and _check_basicConstraints(extensions) and
_check_ocspMustStaple(extensions) and _check_nameConstraints(extensions))
def _check_signature(csr):
try:
return csr.verify(self.privatekey)
except crypto.Error:
return False
return _check_subject(self.existing_csr) and _check_extensions(self.existing_csr) and _check_signature(self.existing_csr)
def parse_crl_distribution_points(module, crl_distribution_points):
result = []
for index, parse_crl_distribution_point in enumerate(crl_distribution_points):
@@ -762,34 +569,17 @@ def select_backend(module, backend):
if backend == 'auto':
# Detection what is possible
can_use_cryptography = CRYPTOGRAPHY_FOUND and CRYPTOGRAPHY_VERSION >= LooseVersion(MINIMAL_CRYPTOGRAPHY_VERSION)
can_use_pyopenssl = PYOPENSSL_FOUND and PYOPENSSL_VERSION >= LooseVersion(MINIMAL_PYOPENSSL_VERSION)
# First try cryptography, then pyOpenSSL
# Try cryptography
if can_use_cryptography:
backend = 'cryptography'
elif can_use_pyopenssl:
backend = 'pyopenssl'
# Success?
if backend == 'auto':
module.fail_json(msg=("Can't detect any of the required Python libraries "
"cryptography (>= {0}) or PyOpenSSL (>= {1})").format(
MINIMAL_CRYPTOGRAPHY_VERSION,
MINIMAL_PYOPENSSL_VERSION))
"cryptography (>= {0})").format(MINIMAL_CRYPTOGRAPHY_VERSION))
if backend == 'pyopenssl':
if not PYOPENSSL_FOUND:
module.fail_json(msg=missing_required_lib('pyOpenSSL >= {0}'.format(MINIMAL_PYOPENSSL_VERSION)),
exception=PYOPENSSL_IMP_ERR)
try:
getattr(crypto.X509Req, 'get_extensions')
except AttributeError:
module.fail_json(msg='You need to have PyOpenSSL>=0.15 to generate CSRs')
module.deprecate('The module is using the PyOpenSSL backend. This backend has been deprecated',
version='2.0.0', collection_name='community.crypto')
return backend, CertificateSigningRequestPyOpenSSLBackend(module)
elif backend == 'cryptography':
if backend == 'cryptography':
if not CRYPTOGRAPHY_FOUND:
module.fail_json(msg=missing_required_lib('cryptography >= {0}'.format(MINIMAL_CRYPTOGRAPHY_VERSION)),
exception=CRYPTOGRAPHY_IMP_ERR)
@@ -853,7 +643,7 @@ def get_csr_argument_spec():
),
mutually_exclusive=[('full_name', 'relative_name')]
),
select_crypto_backend=dict(type='str', default='auto', choices=['auto', 'cryptography', 'pyopenssl']),
select_crypto_backend=dict(type='str', default='auto', choices=['auto', 'cryptography']),
),
required_together=[
['authority_cert_issuer', 'authority_cert_serial_number'],

180
plugins/module_utils/crypto/module_backends/csr_info.py
View File

@@ -30,38 +30,11 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptograp
cryptography_oid_to_name,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.pyopenssl_support import (
pyopenssl_get_extensions_from_csr,
pyopenssl_normalize_name,
pyopenssl_normalize_name_attribute,
pyopenssl_parse_name_constraints,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.publickey_info import (
get_publickey_info,
)
MINIMAL_CRYPTOGRAPHY_VERSION = '1.3'
MINIMAL_PYOPENSSL_VERSION = '0.15'
PYOPENSSL_IMP_ERR = None
try:
import OpenSSL
from OpenSSL import crypto
PYOPENSSL_VERSION = LooseVersion(OpenSSL.__version__)
if OpenSSL.SSL.OPENSSL_VERSION_NUMBER >= 0x10100000:
# OpenSSL 1.1.0 or newer
OPENSSL_MUST_STAPLE_NAME = b"tlsfeature"
OPENSSL_MUST_STAPLE_VALUE = b"status_request"
else:
# OpenSSL 1.0.x or older
OPENSSL_MUST_STAPLE_NAME = b"1.3.6.1.5.5.7.1.24"
OPENSSL_MUST_STAPLE_VALUE = b"DER:30:03:02:01:05"
except ImportError:
PYOPENSSL_IMP_ERR = traceback.format_exc()
PYOPENSSL_FOUND = False
else:
PYOPENSSL_FOUND = True
CRYPTOGRAPHY_IMP_ERR = None
try:
@@ -173,20 +146,19 @@ class CSRInfoRetrieval(object):
'public_key_fingerprints': public_key_info['fingerprints'],
})
if self.backend != 'pyopenssl':
ski = self._get_subject_key_identifier()
if ski is not None:
ski = to_native(binascii.hexlify(ski))
ski = ':'.join([ski[i:i + 2] for i in range(0, len(ski), 2)])
result['subject_key_identifier'] = ski
ski = self._get_subject_key_identifier()
if ski is not None:
ski = to_native(binascii.hexlify(ski))
ski = ':'.join([ski[i:i + 2] for i in range(0, len(ski), 2)])
result['subject_key_identifier'] = ski
aki, aci, acsn = self._get_authority_key_identifier()
if aki is not None:
aki = to_native(binascii.hexlify(aki))
aki = ':'.join([aki[i:i + 2] for i in range(0, len(aki), 2)])
result['authority_key_identifier'] = aki
result['authority_cert_issuer'] = aci
result['authority_cert_serial_number'] = acsn
aki, aci, acsn = self._get_authority_key_identifier()
if aki is not None:
aki = to_native(binascii.hexlify(aki))
aki = ':'.join([aki[i:i + 2] for i in range(0, len(aki), 2)])
result['authority_key_identifier'] = aki
result['authority_cert_issuer'] = aci
result['authority_cert_serial_number'] = acsn
result['extensions_by_oid'] = self._get_all_extensions()
@@ -332,112 +304,9 @@ class CSRInfoRetrievalCryptography(CSRInfoRetrieval):
return self.csr.is_signature_valid
class CSRInfoRetrievalPyOpenSSL(CSRInfoRetrieval):
"""validate the supplied CSR."""
def __init__(self, module, content, validate_signature):
super(CSRInfoRetrievalPyOpenSSL, self).__init__(module, 'pyopenssl', content, validate_signature)
def __get_name(self, name):
result = []
for sub in name.get_components():
result.append([pyopenssl_normalize_name(sub[0]), to_text(sub[1])])
return result
def _get_subject_ordered(self):
return self.__get_name(self.csr.get_subject())
def _get_extension(self, short_name):
for extension in self.csr.get_extensions():
if extension.get_short_name() == short_name:
result = [
pyopenssl_normalize_name(usage.strip()) for usage in to_text(extension, errors='surrogate_or_strict').split(',')
]
return sorted(result), bool(extension.get_critical())
return None, False
def _get_key_usage(self):
return self._get_extension(b'keyUsage')
def _get_extended_key_usage(self):
return self._get_extension(b'extendedKeyUsage')
def _get_basic_constraints(self):
return self._get_extension(b'basicConstraints')
def _get_ocsp_must_staple(self):
extensions = self.csr.get_extensions()
oms_ext = [
ext for ext in extensions
if to_bytes(ext.get_short_name()) == OPENSSL_MUST_STAPLE_NAME and to_bytes(ext) == OPENSSL_MUST_STAPLE_VALUE
]
if OpenSSL.SSL.OPENSSL_VERSION_NUMBER < 0x10100000:
# Older versions of libssl don't know about OCSP Must Staple
oms_ext.extend([ext for ext in extensions if ext.get_short_name() == b'UNDEF' and ext.get_data() == b'\x30\x03\x02\x01\x05'])
if oms_ext:
return True, bool(oms_ext[0].get_critical())
else:
return None, False
def _get_subject_alt_name(self):
for extension in self.csr.get_extensions():
if extension.get_short_name() == b'subjectAltName':
result = [pyopenssl_normalize_name_attribute(altname.strip()) for altname in
to_text(extension, errors='surrogate_or_strict').split(', ')]
return result, bool(extension.get_critical())
return None, False
def _get_name_constraints(self):
for extension in self.csr.get_extensions():
if extension.get_short_name() == b'nameConstraints':
permitted, excluded = pyopenssl_parse_name_constraints(extension)
return permitted, excluded, bool(extension.get_critical())
return None, None, False
def _get_public_key_pem(self):
try:
return crypto.dump_publickey(
crypto.FILETYPE_PEM,
self.csr.get_pubkey(),
)
except AttributeError:
try:
bio = crypto._new_mem_buf()
rc = crypto._lib.PEM_write_bio_PUBKEY(bio, self.csr.get_pubkey()._pkey)
if rc != 1:
crypto._raise_current_error()
return crypto._bio_to_string(bio)
except AttributeError:
self.module.warn('Your pyOpenSSL version does not support dumping public keys. '
'Please upgrade to version 16.0 or newer, or use the cryptography backend.')
def _get_public_key_object(self):
return self.csr.get_pubkey()
def _get_subject_key_identifier(self):
# Won't be implemented
return None
def _get_authority_key_identifier(self):
# Won't be implemented
return None, None, None
def _get_all_extensions(self):
return pyopenssl_get_extensions_from_csr(self.csr)
def _is_signature_valid(self):
try:
return bool(self.csr.verify(self.csr.get_pubkey()))
except crypto.Error:
# OpenSSL error means that key is not consistent
return False
def get_csr_info(module, backend, content, validate_signature=True, prefer_one_fingerprint=False):
if backend == 'cryptography':
info = CSRInfoRetrievalCryptography(module, content, validate_signature=validate_signature)
elif backend == 'pyopenssl':
info = CSRInfoRetrievalPyOpenSSL(module, content, validate_signature=validate_signature)
return info.get_info(prefer_one_fingerprint=prefer_one_fingerprint)
@@ -445,34 +314,17 @@ def select_backend(module, backend, content, validate_signature=True):
if backend == 'auto':
# Detection what is possible
can_use_cryptography = CRYPTOGRAPHY_FOUND and CRYPTOGRAPHY_VERSION >= LooseVersion(MINIMAL_CRYPTOGRAPHY_VERSION)
can_use_pyopenssl = PYOPENSSL_FOUND and PYOPENSSL_VERSION >= LooseVersion(MINIMAL_PYOPENSSL_VERSION)
# First try cryptography, then pyOpenSSL
# Try cryptography
if can_use_cryptography:
backend = 'cryptography'
elif can_use_pyopenssl:
backend = 'pyopenssl'
# Success?
if backend == 'auto':
module.fail_json(msg=("Can't detect any of the required Python libraries "
"cryptography (>= {0}) or PyOpenSSL (>= {1})").format(
MINIMAL_CRYPTOGRAPHY_VERSION,
MINIMAL_PYOPENSSL_VERSION))
module.fail_json(msg=("Can't detect the required Python library "
"cryptography (>= {0})").format(MINIMAL_CRYPTOGRAPHY_VERSION))
if backend == 'pyopenssl':
if not PYOPENSSL_FOUND:
module.fail_json(msg=missing_required_lib('pyOpenSSL >= {0}'.format(MINIMAL_PYOPENSSL_VERSION)),
exception=PYOPENSSL_IMP_ERR)
try:
getattr(crypto.X509Req, 'get_extensions')
except AttributeError:
module.fail_json(msg='You need to have PyOpenSSL>=0.15 to generate CSRs')
module.deprecate('The module is using the PyOpenSSL backend. This backend has been deprecated',
version='2.0.0', collection_name='community.crypto')
return backend, CSRInfoRetrievalPyOpenSSL(module, content, validate_signature=validate_signature)
elif backend == 'cryptography':
if backend == 'cryptography':
if not CRYPTOGRAPHY_FOUND:
module.fail_json(msg=missing_required_lib('cryptography >= {0}'.format(MINIMAL_CRYPTOGRAPHY_VERSION)),
exception=CRYPTOGRAPHY_IMP_ERR)

99
plugins/module_utils/crypto/module_backends/privatekey.py
View File

@@ -46,20 +46,8 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.module_bac
from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.common import ArgumentSpec
MINIMAL_PYOPENSSL_VERSION = '0.6'
MINIMAL_CRYPTOGRAPHY_VERSION = '1.2.3'
PYOPENSSL_IMP_ERR = None
try:
import OpenSSL
from OpenSSL import crypto
PYOPENSSL_VERSION = LooseVersion(OpenSSL.__version__)
except ImportError:
PYOPENSSL_IMP_ERR = traceback.format_exc()
PYOPENSSL_FOUND = False
else:
PYOPENSSL_FOUND = True
CRYPTOGRAPHY_IMP_ERR = None
try:
import cryptography
@@ -263,61 +251,6 @@ class PrivateKeyBackend:
return result
# Implementation with using pyOpenSSL
class PrivateKeyPyOpenSSLBackend(PrivateKeyBackend):
def __init__(self, module):
super(PrivateKeyPyOpenSSLBackend, self).__init__(module=module, backend='pyopenssl')
if self.type == 'RSA':
self.openssl_type = crypto.TYPE_RSA
elif self.type == 'DSA':
self.openssl_type = crypto.TYPE_DSA
else:
self.module.fail_json(msg="PyOpenSSL backend only supports RSA and DSA keys.")
if self.format != 'auto_ignore':
self.module.fail_json(msg="PyOpenSSL backend only supports auto_ignore format.")
def generate_private_key(self):
"""(Re-)Generate private key."""
self.private_key = crypto.PKey()
try:
self.private_key.generate_key(self.openssl_type, self.size)
except (TypeError, ValueError) as exc:
raise PrivateKeyError(exc)
def _ensure_existing_private_key_loaded(self):
if self.existing_private_key is None and self.has_existing():
try:
self.existing_private_key = load_privatekey(
None, self.passphrase, content=self.existing_private_key_bytes, backend=self.backend)
except OpenSSLBadPassphraseError as exc:
raise PrivateKeyError(exc)
def get_private_key_data(self):
"""Return bytes for self.private_key"""
if self.cipher and self.passphrase:
return crypto.dump_privatekey(crypto.FILETYPE_PEM, self.private_key,
self.cipher, to_bytes(self.passphrase))
else:
return crypto.dump_privatekey(crypto.FILETYPE_PEM, self.private_key)
def _check_passphrase(self):
try:
load_privatekey(None, self.passphrase, content=self.existing_private_key_bytes, backend=self.backend)
return True
except Exception as dummy:
return False
def _check_size_and_type(self):
return self.size == self.existing_private_key.bits() and self.openssl_type == self.existing_private_key.type()
def _check_format(self):
# Not supported by this backend
return True
# Implementation with using cryptography
class PrivateKeyCryptographyBackend(PrivateKeyBackend):
@@ -550,36 +483,16 @@ def select_backend(module, backend):
if backend == 'auto':
# Detection what is possible
can_use_cryptography = CRYPTOGRAPHY_FOUND and CRYPTOGRAPHY_VERSION >= LooseVersion(MINIMAL_CRYPTOGRAPHY_VERSION)
can_use_pyopenssl = PYOPENSSL_FOUND and PYOPENSSL_VERSION >= LooseVersion(MINIMAL_PYOPENSSL_VERSION)
# Decision
if module.params['cipher'] and module.params['passphrase'] and module.params['cipher'] != 'auto':
# First try pyOpenSSL, then cryptography
if can_use_pyopenssl:
backend = 'pyopenssl'
elif can_use_cryptography:
backend = 'cryptography'
else:
# First try cryptography, then pyOpenSSL
if can_use_cryptography:
backend = 'cryptography'
elif can_use_pyopenssl:
backend = 'pyopenssl'
if can_use_cryptography:
backend = 'cryptography'
# Success?
if backend == 'auto':
module.fail_json(msg=("Can't detect any of the required Python libraries "
"cryptography (>= {0}) or PyOpenSSL (>= {1})").format(
MINIMAL_CRYPTOGRAPHY_VERSION,
MINIMAL_PYOPENSSL_VERSION))
if backend == 'pyopenssl':
if not PYOPENSSL_FOUND:
module.fail_json(msg=missing_required_lib('pyOpenSSL >= {0}'.format(MINIMAL_PYOPENSSL_VERSION)),
exception=PYOPENSSL_IMP_ERR)
module.deprecate('The module is using the PyOpenSSL backend. This backend has been deprecated',
version='2.0.0', collection_name='community.crypto')
return backend, PrivateKeyPyOpenSSLBackend(module)
elif backend == 'cryptography':
module.fail_json(msg=("Can't detect the required Python library "
"cryptography (>= {0})").format(MINIMAL_CRYPTOGRAPHY_VERSION))
if backend == 'cryptography':
if not CRYPTOGRAPHY_FOUND:
module.fail_json(msg=missing_required_lib('cryptography >= {0}'.format(MINIMAL_CRYPTOGRAPHY_VERSION)),
exception=CRYPTOGRAPHY_IMP_ERR)
@@ -605,7 +518,7 @@ def get_privatekey_argument_spec():
cipher=dict(type='str'),
format=dict(type='str', default='auto_ignore', choices=['pkcs1', 'pkcs8', 'raw', 'auto', 'auto_ignore']),
format_mismatch=dict(type='str', default='regenerate', choices=['regenerate', 'convert']),
select_crypto_backend=dict(type='str', choices=['auto', 'pyopenssl', 'cryptography'], default='auto'),
select_crypto_backend=dict(type='str', choices=['auto', 'cryptography'], default='auto'),
regenerate=dict(
type='str',
default='full_idempotence',

160
plugins/module_utils/crypto/module_backends/privatekey_info.py
View File

@@ -36,24 +36,10 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.math impor
from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.publickey_info import (
_get_cryptography_public_key_info,
_bigint_to_int,
_get_pyopenssl_public_key_info,
)
MINIMAL_CRYPTOGRAPHY_VERSION = '1.2.3'
MINIMAL_PYOPENSSL_VERSION = '0.15'
PYOPENSSL_IMP_ERR = None
try:
import OpenSSL
from OpenSSL import crypto
PYOPENSSL_VERSION = LooseVersion(OpenSSL.__version__)
except ImportError:
PYOPENSSL_IMP_ERR = traceback.format_exc()
PYOPENSSL_FOUND = False
else:
PYOPENSSL_FOUND = True
CRYPTOGRAPHY_IMP_ERR = None
try:
@@ -265,135 +251,10 @@ class PrivateKeyInfoRetrievalCryptography(PrivateKeyInfoRetrieval):
return _is_cryptography_key_consistent(self.key, key_public_data, key_private_data)
class PrivateKeyInfoRetrievalPyOpenSSL(PrivateKeyInfoRetrieval):
"""validate the supplied private key."""
def __init__(self, module, content, **kwargs):
super(PrivateKeyInfoRetrievalPyOpenSSL, self).__init__(module, 'pyopenssl', content, **kwargs)
def _get_public_key(self, binary):
try:
return crypto.dump_publickey(
crypto.FILETYPE_ASN1 if binary else crypto.FILETYPE_PEM,
self.key
)
except AttributeError:
try:
# pyOpenSSL < 16.0:
bio = crypto._new_mem_buf()
if binary:
rc = crypto._lib.i2d_PUBKEY_bio(bio, self.key._pkey)
else:
rc = crypto._lib.PEM_write_bio_PUBKEY(bio, self.key._pkey)
if rc != 1:
crypto._raise_current_error()
return crypto._bio_to_string(bio)
except AttributeError:
self.module.warn('Your pyOpenSSL version does not support dumping public keys. '
'Please upgrade to version 16.0 or newer, or use the cryptography backend.')
def _get_key_info(self):
key_type, key_public_data, try_fallback = _get_pyopenssl_public_key_info(self.key)
key_private_data = dict()
openssl_key_type = self.key.type()
if crypto.TYPE_RSA == openssl_key_type:
try:
# Use OpenSSL directly to extract key data
key = OpenSSL._util.lib.EVP_PKEY_get1_RSA(self.key._pkey)
key = OpenSSL._util.ffi.gc(key, OpenSSL._util.lib.RSA_free)
# OpenSSL 1.1 and newer have functions to extract the parameters
# from the EVP PKEY data structures. Older versions didn't have
# these getters, and it was common use to simply access the values
# directly. Since there's no guarantee that these data structures
# will still be accessible in the future, we use the getters for
# 1.1 and later, and directly access the values for 1.0.x and
# earlier.
if OpenSSL.SSL.OPENSSL_VERSION_NUMBER >= 0x10100000:
# Get modulus and exponents
n = OpenSSL._util.ffi.new("BIGNUM **")
e = OpenSSL._util.ffi.new("BIGNUM **")
d = OpenSSL._util.ffi.new("BIGNUM **")
OpenSSL._util.lib.RSA_get0_key(key, n, e, d)
key_private_data['exponent'] = _bigint_to_int(d[0])
# Get factors
p = OpenSSL._util.ffi.new("BIGNUM **")
q = OpenSSL._util.ffi.new("BIGNUM **")
OpenSSL._util.lib.RSA_get0_factors(key, p, q)
key_private_data['p'] = _bigint_to_int(p[0])
key_private_data['q'] = _bigint_to_int(q[0])
else:
# Get private exponent
key_private_data['exponent'] = _bigint_to_int(key.d)
# Get factors
key_private_data['p'] = _bigint_to_int(key.p)
key_private_data['q'] = _bigint_to_int(key.q)
except AttributeError:
try_fallback = True
elif crypto.TYPE_DSA == openssl_key_type:
try:
# Use OpenSSL directly to extract key data
key = OpenSSL._util.lib.EVP_PKEY_get1_DSA(self.key._pkey)
key = OpenSSL._util.ffi.gc(key, OpenSSL._util.lib.DSA_free)
# OpenSSL 1.1 and newer have functions to extract the parameters
# from the EVP PKEY data structures. Older versions didn't have
# these getters, and it was common use to simply access the values
# directly. Since there's no guarantee that these data structures
# will still be accessible in the future, we use the getters for
# 1.1 and later, and directly access the values for 1.0.x and
# earlier.
if OpenSSL.SSL.OPENSSL_VERSION_NUMBER >= 0x10100000:
# Get private key exponents
y = OpenSSL._util.ffi.new("BIGNUM **")
x = OpenSSL._util.ffi.new("BIGNUM **")
OpenSSL._util.lib.DSA_get0_key(key, y, x)
key_private_data['x'] = _bigint_to_int(x[0])
else:
# Get private key exponents
key_private_data['x'] = _bigint_to_int(key.priv_key)
except AttributeError:
try_fallback = True
else:
# Return 'unknown'
key_type = 'unknown ({0})'.format(self.key.type())
# If needed and if possible, fall back to cryptography
if try_fallback and PYOPENSSL_VERSION >= LooseVersion('16.1.0') and CRYPTOGRAPHY_FOUND:
return _get_cryptography_private_key_info(self.key.to_cryptography_key())
return key_type, key_public_data, key_private_data
def _is_key_consistent(self, key_public_data, key_private_data):
openssl_key_type = self.key.type()
if crypto.TYPE_RSA == openssl_key_type:
try:
return self.key.check()
except crypto.Error:
# OpenSSL error means that key is not consistent
return False
if crypto.TYPE_DSA == openssl_key_type:
result = _check_dsa_consistency(key_public_data, key_private_data)
if result is not None:
return result
signature = crypto.sign(self.key, SIGNATURE_TEST_DATA, 'sha256')
# Verify wants a cert (where it can get the public key from)
cert = crypto.X509()
cert.set_pubkey(self.key)
try:
crypto.verify(cert, signature, SIGNATURE_TEST_DATA, 'sha256')
return True
except crypto.Error:
return False
# If needed and if possible, fall back to cryptography
if PYOPENSSL_VERSION >= LooseVersion('16.1.0') and CRYPTOGRAPHY_FOUND:
return _is_cryptography_key_consistent(self.key.to_cryptography_key(), key_public_data, key_private_data)
return None
def get_privatekey_info(module, backend, content, passphrase=None, return_private_key_data=False, prefer_one_fingerprint=False):
if backend == 'cryptography':
info = PrivateKeyInfoRetrievalCryptography(
module, content, passphrase=passphrase, return_private_key_data=return_private_key_data)
elif backend == 'pyopenssl':
info = PrivateKeyInfoRetrievalPyOpenSSL(
module, content, passphrase=passphrase, return_private_key_data=return_private_key_data)
return info.get_info(prefer_one_fingerprint=prefer_one_fingerprint)
@@ -401,30 +262,17 @@ def select_backend(module, backend, content, passphrase=None, return_private_key
if backend == 'auto':
# Detection what is possible
can_use_cryptography = CRYPTOGRAPHY_FOUND and CRYPTOGRAPHY_VERSION >= LooseVersion(MINIMAL_CRYPTOGRAPHY_VERSION)
can_use_pyopenssl = PYOPENSSL_FOUND and PYOPENSSL_VERSION >= LooseVersion(MINIMAL_PYOPENSSL_VERSION)
# First try cryptography, then pyOpenSSL
# Try cryptography
if can_use_cryptography:
backend = 'cryptography'
elif can_use_pyopenssl:
backend = 'pyopenssl'
# Success?
if backend == 'auto':
module.fail_json(msg=("Can't detect any of the required Python libraries "
"cryptography (>= {0}) or PyOpenSSL (>= {1})").format(
MINIMAL_CRYPTOGRAPHY_VERSION,
MINIMAL_PYOPENSSL_VERSION))
module.fail_json(msg=("Can't detect the required Python library "
"cryptography (>= {0})").format(MINIMAL_CRYPTOGRAPHY_VERSION))
if backend == 'pyopenssl':
if not PYOPENSSL_FOUND:
module.fail_json(msg=missing_required_lib('pyOpenSSL >= {0}'.format(MINIMAL_PYOPENSSL_VERSION)),
exception=PYOPENSSL_IMP_ERR)
module.deprecate('The module is using the PyOpenSSL backend. This backend has been deprecated',
version='2.0.0', collection_name='community.crypto')
return backend, PrivateKeyInfoRetrievalPyOpenSSL(
module, content, passphrase=passphrase, return_private_key_data=return_private_key_data)
elif backend == 'cryptography':
if backend == 'cryptography':
if not CRYPTOGRAPHY_FOUND:
module.fail_json(msg=missing_required_lib('cryptography >= {0}'.format(MINIMAL_CRYPTOGRAPHY_VERSION)),
exception=CRYPTOGRAPHY_IMP_ERR)

159
plugins/module_utils/crypto/module_backends/publickey_info.py
View File

@@ -31,18 +31,6 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.support im
MINIMAL_CRYPTOGRAPHY_VERSION = '1.2.3'
MINIMAL_PYOPENSSL_VERSION = '16.0.0' # when working with public key objects, the minimal required version is 0.15
PYOPENSSL_IMP_ERR = None
try:
import OpenSSL
from OpenSSL import crypto
PYOPENSSL_VERSION = LooseVersion(OpenSSL.__version__)
except ImportError:
PYOPENSSL_IMP_ERR = traceback.format_exc()
PYOPENSSL_FOUND = False
else:
PYOPENSSL_FOUND = True
CRYPTOGRAPHY_IMP_ERR = None
try:
@@ -93,98 +81,6 @@ def _get_cryptography_public_key_info(key):
return key_type, key_public_data
def _bigint_to_int(bn):
'''Convert OpenSSL BIGINT to Python integer'''
if bn == OpenSSL._util.ffi.NULL:
return None
hexstr = OpenSSL._util.lib.BN_bn2hex(bn)
try:
return int(OpenSSL._util.ffi.string(hexstr), 16)
finally:
OpenSSL._util.lib.OPENSSL_free(hexstr)
def _get_pyopenssl_public_key_info(key):
key_public_data = dict()
try_fallback = True
openssl_key_type = key.type()
if crypto.TYPE_RSA == openssl_key_type:
key_type = 'RSA'
key_public_data['size'] = key.bits()
try:
# Use OpenSSL directly to extract key data
key = OpenSSL._util.lib.EVP_PKEY_get1_RSA(key._pkey)
key = OpenSSL._util.ffi.gc(key, OpenSSL._util.lib.RSA_free)
# OpenSSL 1.1 and newer have functions to extract the parameters
# from the EVP PKEY data structures. Older versions didn't have
# these getters, and it was common use to simply access the values
# directly. Since there's no guarantee that these data structures
# will still be accessible in the future, we use the getters for
# 1.1 and later, and directly access the values for 1.0.x and
# earlier.
if OpenSSL.SSL.OPENSSL_VERSION_NUMBER >= 0x10100000:
# Get modulus and exponents
n = OpenSSL._util.ffi.new("BIGNUM **")
e = OpenSSL._util.ffi.new("BIGNUM **")
d = OpenSSL._util.ffi.new("BIGNUM **")
OpenSSL._util.lib.RSA_get0_key(key, n, e, d)
key_public_data['modulus'] = _bigint_to_int(n[0])
key_public_data['exponent'] = _bigint_to_int(e[0])
else:
# Get modulus and exponents
key_public_data['modulus'] = _bigint_to_int(key.n)
key_public_data['exponent'] = _bigint_to_int(key.e)
try_fallback = False
except AttributeError:
# Use fallback if available
pass
elif crypto.TYPE_DSA == openssl_key_type:
key_type = 'DSA'
key_public_data['size'] = key.bits()
try:
# Use OpenSSL directly to extract key data
key = OpenSSL._util.lib.EVP_PKEY_get1_DSA(key._pkey)
key = OpenSSL._util.ffi.gc(key, OpenSSL._util.lib.DSA_free)
# OpenSSL 1.1 and newer have functions to extract the parameters
# from the EVP PKEY data structures. Older versions didn't have
# these getters, and it was common use to simply access the values
# directly. Since there's no guarantee that these data structures
# will still be accessible in the future, we use the getters for
# 1.1 and later, and directly access the values for 1.0.x and
# earlier.
if OpenSSL.SSL.OPENSSL_VERSION_NUMBER >= 0x10100000:
# Get public parameters (primes and group element)
p = OpenSSL._util.ffi.new("BIGNUM **")
q = OpenSSL._util.ffi.new("BIGNUM **")
g = OpenSSL._util.ffi.new("BIGNUM **")
OpenSSL._util.lib.DSA_get0_pqg(key, p, q, g)
key_public_data['p'] = _bigint_to_int(p[0])
key_public_data['q'] = _bigint_to_int(q[0])
key_public_data['g'] = _bigint_to_int(g[0])
# Get public key exponents
y = OpenSSL._util.ffi.new("BIGNUM **")
x = OpenSSL._util.ffi.new("BIGNUM **")
OpenSSL._util.lib.DSA_get0_key(key, y, x)
key_public_data['y'] = _bigint_to_int(y[0])
else:
# Get public parameters (primes and group element)
key_public_data['p'] = _bigint_to_int(key.p)
key_public_data['q'] = _bigint_to_int(key.q)
key_public_data['g'] = _bigint_to_int(key.g)
# Get public key exponents
key_public_data['y'] = _bigint_to_int(key.pub_key)
try_fallback = False
except AttributeError:
# Use fallback if available
pass
else:
# Return 'unknown'
key_type = 'unknown ({0})'.format(key.type())
return key_type, key_public_data, try_fallback
class PublicKeyParseError(OpenSSLObjectError):
def __init__(self, msg, result):
super(PublicKeyParseError, self).__init__(msg)
@@ -242,46 +138,9 @@ class PublicKeyInfoRetrievalCryptography(PublicKeyInfoRetrieval):
return _get_cryptography_public_key_info(self.key)
class PublicKeyInfoRetrievalPyOpenSSL(PublicKeyInfoRetrieval):
"""validate the supplied public key."""
def __init__(self, module, content=None, key=None):
super(PublicKeyInfoRetrievalPyOpenSSL, self).__init__(module, 'pyopenssl', content=content, key=key)
def _get_public_key(self, binary):
try:
return crypto.dump_publickey(
crypto.FILETYPE_ASN1 if binary else crypto.FILETYPE_PEM,
self.key
)
except AttributeError:
try:
# pyOpenSSL < 16.0:
bio = crypto._new_mem_buf()
if binary:
rc = crypto._lib.i2d_PUBKEY_bio(bio, self.key._pkey)
else:
rc = crypto._lib.PEM_write_bio_PUBKEY(bio, self.key._pkey)
if rc != 1:
crypto._raise_current_error()
return crypto._bio_to_string(bio)
except AttributeError:
self.module.warn('Your pyOpenSSL version does not support dumping public keys. '
'Please upgrade to version 16.0 or newer, or use the cryptography backend.')
def _get_key_info(self):
key_type, key_public_data, try_fallback = _get_pyopenssl_public_key_info(self.key)
# If needed and if possible, fall back to cryptography
if try_fallback and PYOPENSSL_VERSION >= LooseVersion('16.1.0') and CRYPTOGRAPHY_FOUND:
return _get_cryptography_public_key_info(self.key.to_cryptography_key())
return key_type, key_public_data
def get_publickey_info(module, backend, content=None, key=None, prefer_one_fingerprint=False):
if backend == 'cryptography':
info = PublicKeyInfoRetrievalCryptography(module, content=content, key=key)
elif backend == 'pyopenssl':
info = PublicKeyInfoRetrievalPyOpenSSL(module, content=content, key=key)
return info.get_info(prefer_one_fingerprint=prefer_one_fingerprint)
@@ -289,29 +148,17 @@ def select_backend(module, backend, content=None, key=None):
if backend == 'auto':
# Detection what is possible
can_use_cryptography = CRYPTOGRAPHY_FOUND and CRYPTOGRAPHY_VERSION >= LooseVersion(MINIMAL_CRYPTOGRAPHY_VERSION)
can_use_pyopenssl = PYOPENSSL_FOUND and PYOPENSSL_VERSION >= LooseVersion(MINIMAL_PYOPENSSL_VERSION)
# First try cryptography, then pyOpenSSL
# Try cryptography
if can_use_cryptography:
backend = 'cryptography'
elif can_use_pyopenssl:
backend = 'pyopenssl'
# Success?
if backend == 'auto':
module.fail_json(msg=("Can't detect any of the required Python libraries "
"cryptography (>= {0}) or PyOpenSSL (>= {1})").format(
MINIMAL_CRYPTOGRAPHY_VERSION,
MINIMAL_PYOPENSSL_VERSION))
"cryptography (>= {0})").format(MINIMAL_CRYPTOGRAPHY_VERSION))
if backend == 'pyopenssl':
if not PYOPENSSL_FOUND:
module.fail_json(msg=missing_required_lib('pyOpenSSL >= {0}'.format(MINIMAL_PYOPENSSL_VERSION)),
exception=PYOPENSSL_IMP_ERR)
module.deprecate('The module is using the PyOpenSSL backend. This backend has been deprecated',
version='2.0.0', collection_name='community.crypto')
return backend, PublicKeyInfoRetrievalPyOpenSSL(module, content=content, key=key)
elif backend == 'cryptography':
if backend == 'cryptography':
if not CRYPTOGRAPHY_FOUND:
module.fail_json(msg=missing_required_lib('cryptography >= {0}'.format(MINIMAL_CRYPTOGRAPHY_VERSION)),
exception=CRYPTOGRAPHY_IMP_ERR)

154
plugins/module_utils/crypto/pyopenssl_support.py
View File

@@ -1,154 +0,0 @@
# -*- coding: utf-8 -*-
#
# (c) 2019, Felix Fontein <felix@fontein.de>
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import base64
from ansible.module_utils.common.text.converters import to_bytes, to_text
from ansible_collections.community.crypto.plugins.module_utils.compat import ipaddress as compat_ipaddress
try:
import OpenSSL
except ImportError:
# Error handled in the calling module.
pass
from ._objects import (
NORMALIZE_NAMES_SHORT,
NORMALIZE_NAMES,
)
from ._obj2txt import obj2txt
from .basic import (
OpenSSLObjectError,
)
def pyopenssl_normalize_name(name, short=False):
nid = OpenSSL._util.lib.OBJ_txt2nid(to_bytes(name))
if nid != 0:
b_name = OpenSSL._util.lib.OBJ_nid2ln(nid)
name = to_text(OpenSSL._util.ffi.string(b_name))
if short:
return NORMALIZE_NAMES_SHORT.get(name, name)
else:
return NORMALIZE_NAMES.get(name, name)
def pyopenssl_normalize_name_attribute(san):
# apparently openssl returns 'IP address' not 'IP' as specifier when converting the subjectAltName to string
# although it won't accept this specifier when generating the CSR. (https://github.com/openssl/openssl/issues/4004)
if san.startswith('IP Address:'):
san = 'IP:' + san[len('IP Address:'):]
if san.startswith('IP:'):
address = san[3:]
if '/' in address:
ip = compat_ipaddress.ip_network(address)
san = 'IP:{0}/{1}'.format(ip.network_address.compressed, ip.prefixlen)
else:
ip = compat_ipaddress.ip_address(address)
san = 'IP:{0}'.format(ip.compressed)
if san.startswith('Registered ID:'):
san = 'RID:' + san[len('Registered ID:'):]
# Some versions of OpenSSL apparently forgot the colon. Happens in CI with Ubuntu 16.04 and FreeBSD 11.1
if san.startswith('Registered ID'):
san = 'RID:' + san[len('Registered ID'):]
return san
def pyopenssl_get_extensions_from_cert(cert):
# While pyOpenSSL allows us to get an extension's DER value, it won't
# give us the dotted string for an OID. So we have to do some magic to
# get hold of it.
result = dict()
ext_count = cert.get_extension_count()
for i in range(0, ext_count):
ext = cert.get_extension(i)
entry = dict(
critical=bool(ext.get_critical()),
value=base64.b64encode(ext.get_data()),
)
oid = obj2txt(
OpenSSL._util.lib,
OpenSSL._util.ffi,
OpenSSL._util.lib.X509_EXTENSION_get_object(ext._extension)
)
# This could also be done a bit simpler:
#
# oid = obj2txt(OpenSSL._util.lib, OpenSSL._util.ffi, OpenSSL._util.lib.OBJ_nid2obj(ext._nid))
#
# Unfortunately this gives the wrong result in case the linked OpenSSL
# doesn't know the OID. That's why we have to get the OID dotted string
# similarly to how cryptography does it.
result[oid] = entry
return result
def pyopenssl_get_extensions_from_csr(csr):
# While pyOpenSSL allows us to get an extension's DER value, it won't
# give us the dotted string for an OID. So we have to do some magic to
# get hold of it.
result = dict()
for ext in csr.get_extensions():
entry = dict(
critical=bool(ext.get_critical()),
value=base64.b64encode(ext.get_data()),
)
oid = obj2txt(
OpenSSL._util.lib,
OpenSSL._util.ffi,
OpenSSL._util.lib.X509_EXTENSION_get_object(ext._extension)
)
# This could also be done a bit simpler:
#
# oid = obj2txt(OpenSSL._util.lib, OpenSSL._util.ffi, OpenSSL._util.lib.OBJ_nid2obj(ext._nid))
#
# Unfortunately this gives the wrong result in case the linked OpenSSL
# doesn't know the OID. That's why we have to get the OID dotted string
# similarly to how cryptography does it.
result[oid] = entry
return result
def pyopenssl_parse_name_constraints(name_constraints_extension):
lines = to_text(name_constraints_extension, errors='surrogate_or_strict').splitlines()
exclude = None
excluded = []
permitted = []
for line in lines:
if line.startswith(' ') or line.startswith('\t'):
name = pyopenssl_normalize_name_attribute(line.strip())
if exclude is True:
excluded.append(name)
elif exclude is False:
permitted.append(name)
else:
raise OpenSSLObjectError('Unexpected nameConstraint line: "{0}"'.format(line))
else:
line_lc = line.lower()
if line_lc.startswith('exclud'):
exclude = True
elif line_lc.startswith('includ') or line_lc.startswith('permitt'):
exclude = False
else:
raise OpenSSLObjectError('Cannot parse nameConstraint line: "{0}"'.format(line))
return permitted, excluded

40
plugins/module_utils/crypto/support.py
View File

@@ -98,25 +98,10 @@ def get_fingerprint_of_bytes(source, prefer_one=False):
return fingerprint
def get_fingerprint_of_privatekey(privatekey, backend='pyopenssl', prefer_one=False):
def get_fingerprint_of_privatekey(privatekey, backend='cryptography', prefer_one=False):
"""Generate the fingerprint of the public key. """
if backend == 'pyopenssl':
try:
publickey = crypto.dump_publickey(crypto.FILETYPE_ASN1, privatekey)
except AttributeError:
# If PyOpenSSL < 16.0 crypto.dump_publickey() will fail.
try:
bio = crypto._new_mem_buf()
rc = crypto._lib.i2d_PUBKEY_bio(bio, privatekey._pkey)
if rc != 1:
crypto._raise_current_error()
publickey = crypto._bio_to_string(bio)
except AttributeError:
# By doing this we prevent the code from raising an error
# yet we return no value in the fingerprint hash.
return None
elif backend == 'cryptography':
if backend == 'cryptography':
publickey = privatekey.public_key().public_bytes(
serialization.Encoding.DER,
serialization.PublicFormat.SubjectPublicKeyInfo
@@ -125,7 +110,7 @@ def get_fingerprint_of_privatekey(privatekey, backend='pyopenssl', prefer_one=Fa
return get_fingerprint_of_bytes(publickey, prefer_one=prefer_one)
def get_fingerprint(path, passphrase=None, content=None, backend='pyopenssl', prefer_one=False):
def get_fingerprint(path, passphrase=None, content=None, backend='cryptography', prefer_one=False):
"""Generate the fingerprint of the public key. """
privatekey = load_privatekey(path, passphrase=passphrase, content=content, check_passphrase=False, backend=backend)
@@ -133,7 +118,7 @@ def get_fingerprint(path, passphrase=None, content=None, backend='pyopenssl', pr
return get_fingerprint_of_privatekey(privatekey, backend=backend, prefer_one=prefer_one)
def load_privatekey(path, passphrase=None, check_passphrase=True, content=None, backend='pyopenssl'):
def load_privatekey(path, passphrase=None, check_passphrase=True, content=None, backend='cryptography'):
"""Load the specified OpenSSL private key.
The content can also be specified via content; in that case,
@@ -213,14 +198,9 @@ def load_publickey(path=None, content=None, backend=None):
return serialization.load_pem_public_key(content, backend=cryptography_backend())
except Exception as e:
raise OpenSSLObjectError('Error while deserializing key: {0}'.format(e))
else:
try:
return crypto.load_publickey(crypto.FILETYPE_PEM, content)
except crypto.Error as e:
raise OpenSSLObjectError('Error while deserializing key: {0}'.format(e))
def load_certificate(path, content=None, backend='pyopenssl'):
def load_certificate(path, content=None, backend='cryptography'):
"""Load the specified certificate."""
try:
@@ -240,7 +220,7 @@ def load_certificate(path, content=None, backend='pyopenssl'):
raise OpenSSLObjectError(exc)
def load_certificate_request(path, content=None, backend='pyopenssl'):
def load_certificate_request(path, content=None, backend='cryptography'):
"""Load the specified certificate signing request."""
try:
if content is None:
@@ -250,9 +230,7 @@ def load_certificate_request(path, content=None, backend='pyopenssl'):
csr_content = content
except (IOError, OSError) as exc:
raise OpenSSLObjectError(exc)
if backend == 'pyopenssl':
return crypto.load_certificate_request(crypto.FILETYPE_PEM, csr_content)
elif backend == 'cryptography':
if backend == 'cryptography':
try:
return x509.load_pem_x509_csr(csr_content, cryptography_backend())
except ValueError as exc:
@@ -322,9 +300,7 @@ def get_relative_time_option(input_string, input_name, backend='cryptography'):
elif backend == 'cryptography':
return result_datetime
# Absolute time
if backend == 'pyopenssl':
return input_string
elif backend == 'cryptography':
if backend == 'cryptography':
for date_fmt in ['%Y%m%d%H%M%SZ', '%Y%m%d%H%MZ', '%Y%m%d%H%M%S%z', '%Y%m%d%H%M%z']:
try:
return datetime.datetime.strptime(result, date_fmt)