internet/community.crypto
3
0
Fork 0
mirror of https://github.com/ansible-collections/community.crypto.git synced 2026-05-06 13:22:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor x509_certificate module, add x509_certificate_pipe module (#135)

Browse Source 
* Move documentation to doc fragment.

* Prepare module backends.

* Linting.

* Fix comments.

* First shot at actually moving code.

* Forgot SKI check.

* Remove unused imports.

* Improve check mode.

* Fix 'returned'.

* Move csr_* checks.

* Explicitly specify parameter.

* Add x509_certificate_pipe module.

* Update other seealsos.

* Forgot to remove doc fragment.

* Adjust to work with macOS 10.15.

* Update plugins/module_utils/crypto/module_backends/certificate_entrust.py

Co-authored-by: Chris Trufan <31186388+ctrufan@users.noreply.github.com>

* Add changelog fragments for entrust bugfix and module refactorings.

* Restore old behavior of Entrust backend when existing certificate cannot be parsed.

* Update plugins/modules/x509_certificate_pipe.py

Co-authored-by: Chris Trufan <31186388+ctrufan@users.noreply.github.com>

* Remove Entrust provider from x509_certificate_pipe for now.

* Add own CA tests.

* One more fix for Entrust provider, when csr_content is used.

* Update plugins/modules/x509_certificate_pipe.py

Co-authored-by: Chris Trufan <31186388+ctrufan@users.noreply.github.com>

* Fix another broken example.

* Revert "Remove Entrust provider from x509_certificate_pipe for now."

This reverts commit 6ee5d7d4f99f0fe2218276a2d3f1f38b676c29b9.

* ci_complete

* Apply suggestions from code review

Co-authored-by: MarkusTeufelberger <mteufelberger@mgit.at>

* Improve example.

* Improve readability of example, add another one.

* Extend descriptions of csr_* for selfsigned.

* Improve documentation.

* Move deprecation message up.

* Explain empty choices.

Co-authored-by: Chris Trufan <31186388+ctrufan@users.noreply.github.com>
Co-authored-by: MarkusTeufelberger <mteufelberger@mgit.at>
This commit is contained in:
Felix Fontein
2020-11-24 17:21:52 +01:00
committed by GitHub
parent 86b39733e1
commit 69335a8bac
22 changed files with 3244 additions and 2369 deletions
Download Patch File Download Diff File Expand all files Collapse all files

392
plugins/module_utils/crypto/module_backends/certificate.py Normal file
View File

@@ -0,0 +1,392 @@
# -*- coding: utf-8 -*-
# Copyright: (c) 2016-2017, Yanis Guenane <yanis+ansible@guenane.org>
# Copyright: (c) 2017, Markus Teufelberger <mteufelberger+ansible@mgit.at>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import abc
import traceback
from distutils.version import LooseVersion
from ansible.module_utils import six
from ansible.module_utils.basic import missing_required_lib
from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.common import ArgumentSpec
from ansible_collections.community.crypto.plugins.module_utils.crypto.basic import (
OpenSSLObjectError,
OpenSSLBadPassphraseError,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.support import (
load_privatekey,
load_certificate,
load_certificate_request,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import (
cryptography_compare_public_keys,
)
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
try:
import cryptography
from cryptography import x509
CRYPTOGRAPHY_VERSION = LooseVersion(cryptography.__version__)
except ImportError:
CRYPTOGRAPHY_IMP_ERR = traceback.format_exc()
CRYPTOGRAPHY_FOUND = False
else:
CRYPTOGRAPHY_FOUND = True
class CertificateError(OpenSSLObjectError):
pass
@six.add_metaclass(abc.ABCMeta)
class CertificateBackend(object):
def __init__(self, module, backend):
self.module = module
self.backend = backend
self.force = module.params['force']
self.privatekey_path = module.params['privatekey_path']
self.privatekey_content = module.params['privatekey_content']
if self.privatekey_content is not None:
self.privatekey_content = self.privatekey_content.encode('utf-8')
self.privatekey_passphrase = module.params['privatekey_passphrase']
self.csr_path = module.params['csr_path']
self.csr_content = module.params['csr_content']
if self.csr_content is not None:
self.csr_content = self.csr_content.encode('utf-8')
# The following are default values which make sure check() works as
# before if providers do not explicitly change these properties.
self.create_subject_key_identifier = 'never_create'
self.create_authority_key_identifier = False
self.privatekey = None
self.csr = None
self.cert = None
self.existing_certificate = None
self.existing_certificate_bytes = None
self.check_csr_subject = True
self.check_csr_extensions = True
@abc.abstractmethod
def generate_certificate(self):
"""(Re-)Generate certificate."""
pass
@abc.abstractmethod
def get_certificate_data(self):
"""Return bytes for self.cert."""
pass
def set_existing(self, certificate_bytes):
"""Set existing certificate bytes. None indicates that the key does not exist."""
self.existing_certificate_bytes = certificate_bytes
def has_existing(self):
"""Query whether an existing certificate is/has been there."""
return self.existing_certificate_bytes is not None
def _ensure_private_key_loaded(self):
"""Load the provided private key into self.privatekey."""
if self.privatekey is not None:
return
if self.privatekey_path is None and self.privatekey_content is None:
return
try:
self.privatekey = load_privatekey(
path=self.privatekey_path,
content=self.privatekey_content,
passphrase=self.privatekey_passphrase,
backend=self.backend,
)
except OpenSSLBadPassphraseError as exc:
raise CertificateError(exc)
def _ensure_csr_loaded(self):
"""Load the CSR into self.csr."""
if self.csr is not None:
return
if self.csr_path is None and self.csr_content is None:
return
self.csr = load_certificate_request(
path=self.csr_path,
content=self.csr_content,
backend=self.backend,
)
def _ensure_existing_certificate_loaded(self):
"""Load the existing certificate into self.existing_certificate."""
if self.existing_certificate is not None:
return
if self.existing_certificate_bytes is None:
return
self.existing_certificate = load_certificate(
path=None,
content=self.existing_certificate_bytes,
backend=self.backend,
)
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':
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':
# Verify that CSR is signed by certificate's private key
if not self.csr.is_signature_valid:
return False
if not cryptography_compare_public_keys(self.csr.public_key(), self.existing_certificate.public_key()):
return False
# Check subject
if self.check_csr_subject and self.csr.subject != self.existing_certificate.subject:
return False
# Check extensions
if not self.check_csr_extensions:
return True
cert_exts = list(self.existing_certificate.extensions)
csr_exts = list(self.csr.extensions)
if self.create_subject_key_identifier != 'never_create':
# Filter out SubjectKeyIdentifier extension before comparison
cert_exts = list(filter(lambda x: not isinstance(x.value, x509.SubjectKeyIdentifier), cert_exts))
csr_exts = list(filter(lambda x: not isinstance(x.value, x509.SubjectKeyIdentifier), csr_exts))
if self.create_authority_key_identifier:
# Filter out AuthorityKeyIdentifier extension before comparison
cert_exts = list(filter(lambda x: not isinstance(x.value, x509.AuthorityKeyIdentifier), cert_exts))
csr_exts = list(filter(lambda x: not isinstance(x.value, x509.AuthorityKeyIdentifier), csr_exts))
if len(cert_exts) != len(csr_exts):
return False
for cert_ext in cert_exts:
try:
csr_ext = self.csr.extensions.get_extension_for_oid(cert_ext.oid)
if cert_ext != csr_ext:
return False
except cryptography.x509.ExtensionNotFound as dummy:
return False
return True
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)
except cryptography.x509.ExtensionNotFound as dummy:
return False
# Get hold of CSR's SKI for 'create_if_not_provided'
csr_ext = None
if self.create_subject_key_identifier == 'create_if_not_provided':
try:
csr_ext = self.csr.extensions.get_extension_for_class(x509.SubjectKeyIdentifier)
except cryptography.x509.ExtensionNotFound as dummy:
pass
if csr_ext is None:
# If CSR had no SKI, or we chose to ignore it ('always_create'), compare with created SKI
if ext.value.digest != x509.SubjectKeyIdentifier.from_public_key(self.existing_certificate.public_key()).digest:
return False
else:
# If CSR had SKI and we didn't ignore it ('create_if_not_provided'), compare SKIs
if ext.value.digest != csr_ext.value.digest:
return False
return True
def needs_regeneration(self):
"""Check whether a regeneration is necessary."""
if self.force or self.existing_certificate_bytes is None:
return True
try:
self._ensure_existing_certificate_loaded()
except Exception as dummy:
return True
# Check whether private key matches
self._ensure_private_key_loaded()
if self.privatekey is not None and not self._check_privatekey():
return True
# Check whether CSR matches
self._ensure_csr_loaded()
if self.csr is not None and not self._check_csr():
return True
# Check SubjectKeyIdentifier
if self.create_subject_key_identifier != 'never_create' and not self._check_subject_key_identifier():
return True
return False
def dump(self, include_certificate):
"""Serialize the object into a dictionary."""
result = {
'privatekey': self.privatekey_path,
'csr': self.csr_path
}
if include_certificate:
# Get hold of certificate bytes
certificate_bytes = self.existing_certificate_bytes
if self.cert is not None:
certificate_bytes = self.get_certificate_data()
# Store result
result['certificate'] = certificate_bytes.decode('utf-8') if certificate_bytes else None
return result
@six.add_metaclass(abc.ABCMeta)
class CertificateProvider(object):
@abc.abstractmethod
def validate_module_args(self, module):
"""Check module arguments"""
@abc.abstractmethod
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.
Return value must be instance of CertificateBackend.
"""
def select_backend(module, backend, provider):
"""
:type module: AnsibleModule
:type backend: str
:type provider: CertificateProvider
"""
provider.validate_module_args(module)
backend = module.params['select_crypto_backend']
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))
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 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')
return provider.create_backend(module, backend)
def get_certificate_argument_spec():
return ArgumentSpec(
argument_spec=dict(
provider=dict(type='str', choices=[]), # choices will be filled by add_XXX_provider_to_argument_spec() in certificate_xxx.py
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']),
# General properties of a certificate
privatekey_path=dict(type='path'),
privatekey_content=dict(type='str', no_log=True),
privatekey_passphrase=dict(type='str', no_log=True),
),
mutually_exclusive=[
['csr_path', 'csr_content'],
['privatekey_path', 'privatekey_content'],
],
)

119
plugins/module_utils/crypto/module_backends/certificate_acme.py Normal file
View File

@@ -0,0 +1,119 @@
# -*- coding: utf-8 -*-
# Copyright: (c) 2016-2017, Yanis Guenane <yanis+ansible@guenane.org>
# Copyright: (c) 2017, Markus Teufelberger <mteufelberger+ansible@mgit.at>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import os
import tempfile
import traceback
from ansible.module_utils._text import to_native, to_bytes
from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.certificate import (
CertificateError,
CertificateBackend,
CertificateProvider,
)
class AcmeCertificateBackend(CertificateBackend):
def __init__(self, module, backend):
super(AcmeCertificateBackend, self).__init__(module, backend)
self.accountkey_path = module.params['acme_accountkey_path']
self.challenge_path = module.params['acme_challenge_path']
self.use_chain = module.params['acme_chain']
self.acme_directory = module.params['acme_directory']
if self.csr_content is None and self.csr_path is None:
raise CertificateError(
'csr_path or csr_content is required for ownca provider'
)
if self.csr_content is None and not os.path.exists(self.csr_path):
raise CertificateError(
'The certificate signing request file %s does not exist' % self.csr_path
)
if not os.path.exists(self.accountkey_path):
raise CertificateError(
'The account key %s does not exist' % self.accountkey_path
)
if not os.path.exists(self.challenge_path):
raise CertificateError(
'The challenge path %s does not exist' % self.challenge_path
)
self.acme_tiny_path = self.module.get_bin_path('acme-tiny', required=True)
def generate_certificate(self):
"""(Re-)Generate certificate."""
command = [self.acme_tiny_path]
if self.use_chain:
command.append('--chain')
command.extend(['--account-key', self.accountkey_path])
if self.csr_content is not None:
# We need to temporarily write the CSR to disk
fd, tmpsrc = tempfile.mkstemp()
self.module.add_cleanup_file(tmpsrc) # Ansible will delete the file on exit
f = os.fdopen(fd, 'wb')
try:
f.write(self.csr_content)
except Exception as err:
try:
f.close()
except Exception as dummy:
pass
self.module.fail_json(
msg="failed to create temporary CSR file: %s" % to_native(err),
exception=traceback.format_exc()
)
f.close()
command.extend(['--csr', tmpsrc])
else:
command.extend(['--csr', self.csr_path])
command.extend(['--acme-dir', self.challenge_path])
command.extend(['--directory-url', self.acme_directory])
try:
self.cert = to_bytes(self.module.run_command(command, check_rc=True)[1])
except OSError as exc:
raise CertificateError(exc)
def get_certificate_data(self):
"""Return bytes for self.cert."""
return self.cert
def dump(self, include_certificate):
result = super(AcmeCertificateBackend, self).dump(include_certificate)
result['accountkey'] = self.accountkey_path
return result
class AcmeCertificateProvider(CertificateProvider):
def validate_module_args(self, module):
if module.params['acme_accountkey_path'] is None:
module.fail_json(msg='The acme_accountkey_path option must be specified for the acme provider.')
if module.params['acme_challenge_path'] is None:
module.fail_json(msg='The acme_challenge_path option must be specified for the acme provider.')
def needs_version_two_certs(self, module):
return False
def create_backend(self, module, backend):
return AcmeCertificateBackend(module, backend)
def add_acme_provider_to_argument_spec(argument_spec):
argument_spec.argument_spec['provider']['choices'].append('acme')
argument_spec.argument_spec.update(dict(
acme_accountkey_path=dict(type='path'),
acme_challenge_path=dict(type='path'),
acme_chain=dict(type='bool', default=False),
acme_directory=dict(type='str', default="https://acme-v02.api.letsencrypt.org/directory"),
))

664
plugins/module_utils/crypto/module_backends/certificate_assertonly.py Normal file
View File

@@ -0,0 +1,664 @@
# -*- coding: utf-8 -*-
# Copyright: (c) 2016-2017, Yanis Guenane <yanis+ansible@guenane.org>
# Copyright: (c) 2017, Markus Teufelberger <mteufelberger+ansible@mgit.at>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import abc
import datetime
from ansible.module_utils._text import to_native, to_bytes, to_text
from ansible_collections.community.crypto.plugins.module_utils.crypto.support import (
parse_name_field,
get_relative_time_option,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import (
cryptography_compare_public_keys,
cryptography_get_name,
cryptography_name_to_oid,
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,
)
try:
import OpenSSL
from OpenSSL import crypto
except ImportError:
pass
try:
import cryptography
from cryptography import x509
from cryptography.x509 import NameAttribute, Name
except ImportError:
pass
def compare_sets(subset, superset, equality=False):
if equality:
return set(subset) == set(superset)
else:
return all(x in superset for x in subset)
def compare_dicts(subset, superset, equality=False):
if equality:
return subset == superset
else:
return all(superset.get(x) == v for x, v in subset.items())
NO_EXTENSION = 'no extension'
class AssertOnlyCertificateBackend(CertificateBackend):
def __init__(self, module, backend):
super(AssertOnlyCertificateBackend, self).__init__(module, backend)
self.signature_algorithms = module.params['signature_algorithms']
if module.params['subject']:
self.subject = parse_name_field(module.params['subject'])
else:
self.subject = []
self.subject_strict = module.params['subject_strict']
if module.params['issuer']:
self.issuer = parse_name_field(module.params['issuer'])
else:
self.issuer = []
self.issuer_strict = module.params['issuer_strict']
self.has_expired = module.params['has_expired']
self.version = module.params['version']
self.key_usage = module.params['key_usage']
self.key_usage_strict = module.params['key_usage_strict']
self.extended_key_usage = module.params['extended_key_usage']
self.extended_key_usage_strict = module.params['extended_key_usage_strict']
self.subject_alt_name = module.params['subject_alt_name']
self.subject_alt_name_strict = module.params['subject_alt_name_strict']
self.not_before = module.params['not_before']
self.not_after = module.params['not_after']
self.valid_at = module.params['valid_at']
self.invalid_at = module.params['invalid_at']
self.valid_in = module.params['valid_in']
if self.valid_in and not self.valid_in.startswith("+") and not self.valid_in.startswith("-"):
try:
int(self.valid_in)
except ValueError:
module.fail_json(msg='The supplied value for "valid_in" (%s) is not an integer or a valid timespec' % self.valid_in)
self.valid_in = "+" + self.valid_in + "s"
# Load objects
self._ensure_private_key_loaded()
self._ensure_csr_loaded()
@abc.abstractmethod
def _validate_privatekey(self):
pass
@abc.abstractmethod
def _validate_csr_signature(self):
pass
@abc.abstractmethod
def _validate_csr_subject(self):
pass
@abc.abstractmethod
def _validate_csr_extensions(self):
pass
@abc.abstractmethod
def _validate_signature_algorithms(self):
pass
@abc.abstractmethod
def _validate_subject(self):
pass
@abc.abstractmethod
def _validate_issuer(self):
pass
@abc.abstractmethod
def _validate_has_expired(self):
pass
@abc.abstractmethod
def _validate_version(self):
pass
@abc.abstractmethod
def _validate_key_usage(self):
pass
@abc.abstractmethod
def _validate_extended_key_usage(self):
pass
@abc.abstractmethod
def _validate_subject_alt_name(self):
pass
@abc.abstractmethod
def _validate_not_before(self):
pass
@abc.abstractmethod
def _validate_not_after(self):
pass
@abc.abstractmethod
def _validate_valid_at(self):
pass
@abc.abstractmethod
def _validate_invalid_at(self):
pass
@abc.abstractmethod
def _validate_valid_in(self):
pass
def assertonly(self):
messages = []
if self.privatekey_path is not None or self.privatekey_content is not None:
if not self._validate_privatekey():
messages.append(
'Certificate %s and private key %s do not match' %
(self.path, self.privatekey_path or '(provided in module options)')
)
if self.csr_path is not None or self.csr_content is not None:
if not self._validate_csr_signature():
messages.append(
'Certificate %s and CSR %s do not match: private key mismatch' %
(self.path, self.csr_path or '(provided in module options)')
)
if not self._validate_csr_subject():
messages.append(
'Certificate %s and CSR %s do not match: subject mismatch' %
(self.path, self.csr_path or '(provided in module options)')
)
if not self._validate_csr_extensions():
messages.append(
'Certificate %s and CSR %s do not match: extensions mismatch' %
(self.path, self.csr_path or '(provided in module options)')
)
if self.signature_algorithms is not None:
wrong_alg = self._validate_signature_algorithms()
if wrong_alg:
messages.append(
'Invalid signature algorithm (got %s, expected one of %s)' %
(wrong_alg, self.signature_algorithms)
)
if self.subject is not None:
failure = self._validate_subject()
if failure:
dummy, cert_subject = failure
messages.append(
'Invalid subject component (got %s, expected all of %s to be present)' %
(cert_subject, self.subject)
)
if self.issuer is not None:
failure = self._validate_issuer()
if failure:
dummy, cert_issuer = failure
messages.append(
'Invalid issuer component (got %s, expected all of %s to be present)' % (cert_issuer, self.issuer)
)
if self.has_expired is not None:
cert_expired = self._validate_has_expired()
if cert_expired != self.has_expired:
messages.append(
'Certificate expiration check failed (certificate expiration is %s, expected %s)' %
(cert_expired, self.has_expired)
)
if self.version is not None:
cert_version = self._validate_version()
if cert_version != self.version:
messages.append(
'Invalid certificate version number (got %s, expected %s)' %
(cert_version, self.version)
)
if self.key_usage is not None:
failure = self._validate_key_usage()
if failure == NO_EXTENSION:
messages.append('Found no keyUsage extension')
elif failure:
dummy, cert_key_usage = failure
messages.append(
'Invalid keyUsage components (got %s, expected all of %s to be present)' %
(cert_key_usage, self.key_usage)
)
if self.extended_key_usage is not None:
failure = self._validate_extended_key_usage()
if failure == NO_EXTENSION:
messages.append('Found no extendedKeyUsage extension')
elif failure:
dummy, ext_cert_key_usage = failure
messages.append(
'Invalid extendedKeyUsage component (got %s, expected all of %s to be present)' % (ext_cert_key_usage, self.extended_key_usage)
)
if self.subject_alt_name is not None:
failure = self._validate_subject_alt_name()
if failure == NO_EXTENSION:
messages.append('Found no subjectAltName extension')
elif failure:
dummy, cert_san = failure
messages.append(
'Invalid subjectAltName component (got %s, expected all of %s to be present)' %
(cert_san, self.subject_alt_name)
)
if self.not_before is not None:
cert_not_valid_before = self._validate_not_before()
if cert_not_valid_before != get_relative_time_option(self.not_before, 'not_before', backend=self.backend):
messages.append(
'Invalid not_before component (got %s, expected %s to be present)' %
(cert_not_valid_before, self.not_before)
)
if self.not_after is not None:
cert_not_valid_after = self._validate_not_after()
if cert_not_valid_after != get_relative_time_option(self.not_after, 'not_after', backend=self.backend):
messages.append(
'Invalid not_after component (got %s, expected %s to be present)' %
(cert_not_valid_after, self.not_after)
)
if self.valid_at is not None:
not_before, valid_at, not_after = self._validate_valid_at()
if not (not_before <= valid_at <= not_after):
messages.append(
'Certificate is not valid for the specified date (%s) - not_before: %s - not_after: %s' %
(self.valid_at, not_before, not_after)
)
if self.invalid_at is not None:
not_before, invalid_at, not_after = self._validate_invalid_at()
if not_before <= invalid_at <= not_after:
messages.append(
'Certificate is not invalid for the specified date (%s) - not_before: %s - not_after: %s' %
(self.invalid_at, not_before, not_after)
)
if self.valid_in is not None:
not_before, valid_in, not_after = self._validate_valid_in()
if not not_before <= valid_in <= not_after:
messages.append(
'Certificate is not valid in %s from now (that would be %s) - not_before: %s - not_after: %s' %
(self.valid_in, valid_in, not_before, not_after)
)
return messages
def needs_regeneration(self):
self._ensure_existing_certificate_loaded()
if self.existing_certificate is None:
self.messages = ['Certificate not provided']
else:
self.messages = self.assertonly()
return len(self.messages) != 0
def generate_certificate(self):
self.module.fail_json(msg=' | '.join(self.messages))
def get_certificate_data(self):
return self.existing_certificate_bytes
class AssertOnlyCertificateBackendCryptography(AssertOnlyCertificateBackend):
"""Validate the supplied cert, using the cryptography backend"""
def __init__(self, module):
super(AssertOnlyCertificateBackendCryptography, self).__init__(module, 'cryptography')
def _validate_privatekey(self):
return cryptography_compare_public_keys(self.existing_certificate.public_key(), self.privatekey.public_key())
def _validate_csr_signature(self):
if not self.csr.is_signature_valid:
return False
return cryptography_compare_public_keys(self.csr.public_key(), self.existing_certificate.public_key())
def _validate_csr_subject(self):
return self.csr.subject == self.existing_certificate.subject
def _validate_csr_extensions(self):
cert_exts = self.existing_certificate.extensions
csr_exts = self.csr.extensions
if len(cert_exts) != len(csr_exts):
return False
for cert_ext in cert_exts:
try:
csr_ext = csr_exts.get_extension_for_oid(cert_ext.oid)
if cert_ext != csr_ext:
return False
except cryptography.x509.ExtensionNotFound as dummy:
return False
return True
def _validate_signature_algorithms(self):
if self.existing_certificate.signature_algorithm_oid._name not in self.signature_algorithms:
return self.existing_certificate.signature_algorithm_oid._name
def _validate_subject(self):
expected_subject = Name([NameAttribute(oid=cryptography_name_to_oid(sub[0]), value=to_text(sub[1]))
for sub in self.subject])
cert_subject = self.existing_certificate.subject
if not compare_sets(expected_subject, cert_subject, self.subject_strict):
return expected_subject, cert_subject
def _validate_issuer(self):
expected_issuer = Name([NameAttribute(oid=cryptography_name_to_oid(iss[0]), value=to_text(iss[1]))
for iss in self.issuer])
cert_issuer = self.existing_certificate.issuer
if not compare_sets(expected_issuer, cert_issuer, self.issuer_strict):
return self.issuer, cert_issuer
def _validate_has_expired(self):
cert_not_after = self.existing_certificate.not_valid_after
cert_expired = cert_not_after < datetime.datetime.utcnow()
return cert_expired
def _validate_version(self):
if self.existing_certificate.version == x509.Version.v1:
return 1
if self.existing_certificate.version == x509.Version.v3:
return 3
return "unknown"
def _validate_key_usage(self):
try:
current_key_usage = self.existing_certificate.extensions.get_extension_for_class(x509.KeyUsage).value
test_key_usage = dict(
digital_signature=current_key_usage.digital_signature,
content_commitment=current_key_usage.content_commitment,
key_encipherment=current_key_usage.key_encipherment,
data_encipherment=current_key_usage.data_encipherment,
key_agreement=current_key_usage.key_agreement,
key_cert_sign=current_key_usage.key_cert_sign,
crl_sign=current_key_usage.crl_sign,
encipher_only=False,
decipher_only=False
)
if test_key_usage['key_agreement']:
test_key_usage.update(dict(
encipher_only=current_key_usage.encipher_only,
decipher_only=current_key_usage.decipher_only
))
key_usages = cryptography_parse_key_usage_params(self.key_usage)
if not compare_dicts(key_usages, test_key_usage, self.key_usage_strict):
return self.key_usage, [k for k, v in test_key_usage.items() if v is True]
except cryptography.x509.ExtensionNotFound:
# 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):
try:
current_ext_keyusage = self.existing_certificate.extensions.get_extension_for_class(x509.ExtendedKeyUsage).value
usages = [cryptography_name_to_oid(usage) for usage in self.extended_key_usage]
expected_ext_keyusage = x509.ExtendedKeyUsage(usages)
if not compare_sets(expected_ext_keyusage, current_ext_keyusage, self.extended_key_usage_strict):
return [eku.value for eku in expected_ext_keyusage], [eku.value for eku in current_ext_keyusage]
except cryptography.x509.ExtensionNotFound:
# 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):
try:
current_san = self.existing_certificate.extensions.get_extension_for_class(x509.SubjectAlternativeName).value
expected_san = [cryptography_get_name(san) for san in self.subject_alt_name]
if not compare_sets(expected_san, current_san, self.subject_alt_name_strict):
return self.subject_alt_name, current_san
except cryptography.x509.ExtensionNotFound:
# 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.not_valid_before
def _validate_not_after(self):
return self.existing_certificate.not_valid_after
def _validate_valid_at(self):
rt = get_relative_time_option(self.valid_at, 'valid_at', backend=self.backend)
return self.existing_certificate.not_valid_before, rt, self.existing_certificate.not_valid_after
def _validate_invalid_at(self):
rt = get_relative_time_option(self.invalid_at, 'invalid_at', backend=self.backend)
return self.existing_certificate.not_valid_before, rt, self.existing_certificate.not_valid_after
def _validate_valid_in(self):
valid_in_date = get_relative_time_option(self.valid_in, "valid_in", backend=self.backend)
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 "
"the 'x509_certificate' module on how to replace it with other modules",
version='2.0.0', collection_name='community.crypto')
def needs_version_two_certs(self, module):
return False
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):
argument_spec.argument_spec['provider']['choices'].append('assertonly')
argument_spec.argument_spec.update(dict(
signature_algorithms=dict(type='list', elements='str', removed_in_version='2.0.0', removed_from_collection='community.crypto'),
subject=dict(type='dict', removed_in_version='2.0.0', removed_from_collection='community.crypto'),
subject_strict=dict(type='bool', default=False, removed_in_version='2.0.0', removed_from_collection='community.crypto'),
issuer=dict(type='dict', removed_in_version='2.0.0', removed_from_collection='community.crypto'),
issuer_strict=dict(type='bool', default=False, removed_in_version='2.0.0', removed_from_collection='community.crypto'),
has_expired=dict(type='bool', default=False, removed_in_version='2.0.0', removed_from_collection='community.crypto'),
version=dict(type='int', removed_in_version='2.0.0', removed_from_collection='community.crypto'),
key_usage=dict(type='list', elements='str', aliases=['keyUsage'],
removed_in_version='2.0.0', removed_from_collection='community.crypto'),
key_usage_strict=dict(type='bool', default=False, aliases=['keyUsage_strict'],
removed_in_version='2.0.0', removed_from_collection='community.crypto'),
extended_key_usage=dict(type='list', elements='str', aliases=['extendedKeyUsage'],
removed_in_version='2.0.0', removed_from_collection='community.crypto'),
extended_key_usage_strict=dict(type='bool', default=False, aliases=['extendedKeyUsage_strict'],
removed_in_version='2.0.0', removed_from_collection='community.crypto'),
subject_alt_name=dict(type='list', elements='str', aliases=['subjectAltName'],
removed_in_version='2.0.0', removed_from_collection='community.crypto'),
subject_alt_name_strict=dict(type='bool', default=False, aliases=['subjectAltName_strict'],
removed_in_version='2.0.0', removed_from_collection='community.crypto'),
not_before=dict(type='str', aliases=['notBefore'], removed_in_version='2.0.0', removed_from_collection='community.crypto'),
not_after=dict(type='str', aliases=['notAfter'], removed_in_version='2.0.0', removed_from_collection='community.crypto'),
valid_at=dict(type='str', removed_in_version='2.0.0', removed_from_collection='community.crypto'),
invalid_at=dict(type='str', removed_in_version='2.0.0', removed_from_collection='community.crypto'),
valid_in=dict(type='str', removed_in_version='2.0.0', removed_from_collection='community.crypto'),
))

225
plugins/module_utils/crypto/module_backends/certificate_entrust.py Normal file
View File

@@ -0,0 +1,225 @@
# -*- coding: utf-8 -*-
# Copyright: (c) 2016-2017, Yanis Guenane <yanis+ansible@guenane.org>
# Copyright: (c) 2017, Markus Teufelberger <mteufelberger+ansible@mgit.at>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import datetime
import time
import os
from ansible.module_utils._text import to_native, to_bytes
from ansible_collections.community.crypto.plugins.module_utils.ecs.api import ECSClient, RestOperationException, SessionConfigurationException
from ansible_collections.community.crypto.plugins.module_utils.crypto.support import (
load_certificate,
get_relative_time_option,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import (
cryptography_serial_number_of_cert,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.certificate import (
CertificateError,
CertificateBackend,
CertificateProvider,
)
try:
from cryptography.x509.oid import NameOID
except ImportError:
pass
class EntrustCertificateBackend(CertificateBackend):
def __init__(self, module, backend):
super(EntrustCertificateBackend, self).__init__(module, backend)
self.trackingId = None
self.notAfter = get_relative_time_option(module.params['entrust_not_after'], 'entrust_not_after', backend=self.backend)
if self.csr_content is None and self.csr_path is None:
raise CertificateError(
'csr_path or csr_content is required for entrust provider'
)
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)
)
self._ensure_csr_loaded()
# ECS API defaults to using the validated organization tied to the account.
# 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':
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
elif len(csr_subject_orgs) > 1:
self.module.fail_json(msg=("Entrust provider does not currently support multiple validated organizations. Multiple organizations found in "
"Subject DN: '{0}'. ".format(self.csr.subject)))
# If no organization in the CSR, explicitly tell ECS that it should be blank in issued cert, not defaulted to
# organization tied to the account.
if self.csr_org is None:
self.csr_org = ''
try:
self.ecs_client = ECSClient(
entrust_api_user=self.module.params['entrust_api_user'],
entrust_api_key=self.module.params['entrust_api_key'],
entrust_api_cert=self.module.params['entrust_api_client_cert_path'],
entrust_api_cert_key=self.module.params['entrust_api_client_cert_key_path'],
entrust_api_specification_path=self.module.params['entrust_api_specification_path']
)
except SessionConfigurationException as e:
module.fail_json(msg='Failed to initialize Entrust Provider: {0}'.format(to_native(e.message)))
def generate_certificate(self):
"""(Re-)Generate certificate."""
body = {}
# Read the CSR that was generated for us
if self.csr_content is not None:
# csr_content contains bytes
body['csr'] = to_native(self.csr_content)
else:
with open(self.csr_path, 'r') as csr_file:
body['csr'] = csr_file.read()
body['certType'] = self.module.params['entrust_cert_type']
# Handle expiration (30 days if not specified)
expiry = self.notAfter
if not expiry:
gmt_now = datetime.datetime.fromtimestamp(time.mktime(time.gmtime()))
expiry = gmt_now + datetime.timedelta(days=365)
expiry_iso3339 = expiry.strftime("%Y-%m-%dT%H:%M:%S.00Z")
body['certExpiryDate'] = expiry_iso3339
body['org'] = self.csr_org
body['tracking'] = {
'requesterName': self.module.params['entrust_requester_name'],
'requesterEmail': self.module.params['entrust_requester_email'],
'requesterPhone': self.module.params['entrust_requester_phone'],
}
try:
result = self.ecs_client.NewCertRequest(Body=body)
self.trackingId = result.get('trackingId')
except RestOperationException as e:
self.module.fail_json(msg='Failed to request new certificate from Entrust Certificate Services (ECS): {0}'.format(to_native(e.message)))
self.cert_bytes = to_bytes(result.get('endEntityCert'))
self.cert = load_certificate(path=None, content=self.cert_bytes, backend=self.backend)
def get_certificate_data(self):
"""Return bytes for self.cert."""
return self.cert_bytes
def needs_regeneration(self):
parent_check = super(EntrustCertificateBackend, self).needs_regeneration()
try:
cert_details = self._get_cert_details()
except RestOperationException as e:
self.module.fail_json(msg='Failed to get status of existing certificate from Entrust Certificate Services (ECS): {0}.'.format(to_native(e.message)))
# Always issue a new certificate if the certificate is expired, suspended or revoked
status = cert_details.get('status', False)
if status == 'EXPIRED' or status == 'SUSPENDED' or status == 'REVOKED':
return True
# If the requested cert type was specified and it is for a different certificate type than the initial certificate, a new one is needed
if self.module.params['entrust_cert_type'] and cert_details.get('certType') and self.module.params['entrust_cert_type'] != cert_details.get('certType'):
return True
return parent_check
def _get_cert_details(self):
cert_details = {}
try:
self._ensure_existing_certificate_loaded()
except Exception as dummy:
return
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':
serial_number = "{0:X}".format(cryptography_serial_number_of_cert(self.existing_certificate))
expiry = self.existing_certificate.not_valid_after
# get some information about the expiry of this certificate
expiry_iso3339 = expiry.strftime("%Y-%m-%dT%H:%M:%S.00Z")
cert_details['expiresAfter'] = expiry_iso3339
# If a trackingId is not already defined (from the result of a generate)
# use the serial number to identify the tracking Id
if self.trackingId is None and serial_number is not None:
cert_results = self.ecs_client.GetCertificates(serialNumber=serial_number).get('certificates', {})
# Finding 0 or more than 1 result is a very unlikely use case, it simply means we cannot perform additional checks
# on the 'state' as returned by Entrust Certificate Services (ECS). The general certificate validity is
# still checked as it is in the rest of the module.
if len(cert_results) == 1:
self.trackingId = cert_results[0].get('trackingId')
if self.trackingId is not None:
cert_details.update(self.ecs_client.GetCertificate(trackingId=self.trackingId))
return cert_details
class EntrustCertificateProvider(CertificateProvider):
def validate_module_args(self, module):
pass
def needs_version_two_certs(self, module):
return False
def create_backend(self, module, backend):
return EntrustCertificateBackend(module, backend)
def add_entrust_provider_to_argument_spec(argument_spec):
argument_spec.argument_spec['provider']['choices'].append('entrust')
argument_spec.argument_spec.update(dict(
entrust_cert_type=dict(type='str', default='STANDARD_SSL',
choices=['STANDARD_SSL', 'ADVANTAGE_SSL', 'UC_SSL', 'EV_SSL', 'WILDCARD_SSL',
'PRIVATE_SSL', 'PD_SSL', 'CDS_ENT_LITE', 'CDS_ENT_PRO', 'SMIME_ENT']),
entrust_requester_email=dict(type='str'),
entrust_requester_name=dict(type='str'),
entrust_requester_phone=dict(type='str'),
entrust_api_user=dict(type='str'),
entrust_api_key=dict(type='str', no_log=True),
entrust_api_client_cert_path=dict(type='path'),
entrust_api_client_cert_key_path=dict(type='path', no_log=True),
entrust_api_specification_path=dict(type='path', default='https://cloud.entrust.net/EntrustCloud/documentation/cms-api-2.1.0.yaml'),
entrust_not_after=dict(type='str', default='+365d'),
))
argument_spec.required_if.append(
['provider', 'entrust', ['entrust_requester_email', 'entrust_requester_name', 'entrust_requester_phone',
'entrust_api_user', 'entrust_api_key', 'entrust_api_client_cert_path',
'entrust_api_client_cert_key_path']]
)

363
plugins/module_utils/crypto/module_backends/certificate_ownca.py Normal file
View File

@@ -0,0 +1,363 @@
# -*- coding: utf-8 -*-
# Copyright: (c) 2016-2017, Yanis Guenane <yanis+ansible@guenane.org>
# Copyright: (c) 2017, Markus Teufelberger <mteufelberger+ansible@mgit.at>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import os
from distutils.version import LooseVersion
from random import randrange
from ansible.module_utils._text import to_bytes
from ansible_collections.community.crypto.plugins.module_utils.crypto.basic import (
OpenSSLBadPassphraseError,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.support import (
load_privatekey,
load_certificate,
get_relative_time_option,
select_message_digest,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import (
cryptography_key_needs_digest_for_signing,
cryptography_serial_number_of_cert,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.certificate import (
CRYPTOGRAPHY_VERSION,
CertificateError,
CertificateBackend,
CertificateProvider,
)
try:
from OpenSSL import crypto
except ImportError:
pass
try:
import cryptography
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.serialization import Encoding
except ImportError:
pass
class OwnCACertificateBackendCryptography(CertificateBackend):
def __init__(self, module):
super(OwnCACertificateBackendCryptography, self).__init__(module, 'cryptography')
self.create_subject_key_identifier = module.params['ownca_create_subject_key_identifier']
self.create_authority_key_identifier = module.params['ownca_create_authority_key_identifier']
self.notBefore = get_relative_time_option(module.params['ownca_not_before'], 'ownca_not_before', backend=self.backend)
self.notAfter = get_relative_time_option(module.params['ownca_not_after'], 'ownca_not_after', backend=self.backend)
self.digest = select_message_digest(module.params['ownca_digest'])
self.version = module.params['ownca_version']
self.serial_number = x509.random_serial_number()
self.ca_cert_path = module.params['ownca_path']
self.ca_cert_content = 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 = module.params['ownca_privatekey_path']
self.ca_privatekey_content = 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 = module.params['ownca_privatekey_passphrase']
if self.csr_content is None and self.csr_path is None:
raise CertificateError(
'csr_path or csr_content is required for ownca provider'
)
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,
backend=self.backend
)
try:
self.ca_private_key = load_privatekey(
path=self.ca_privatekey_path,
content=self.ca_privatekey_content,
passphrase=self.ca_privatekey_passphrase,
backend=self.backend
)
except OpenSSLBadPassphraseError as exc:
module.fail_json(msg=str(exc))
if cryptography_key_needs_digest_for_signing(self.ca_private_key):
if self.digest is None:
raise CertificateError(
'The digest %s is not supported with the cryptography backend' % module.params['ownca_digest']
)
else:
self.digest = None
def generate_certificate(self):
"""(Re-)Generate certificate."""
cert_builder = x509.CertificateBuilder()
cert_builder = cert_builder.subject_name(self.csr.subject)
cert_builder = cert_builder.issuer_name(self.ca_cert.subject)
cert_builder = cert_builder.serial_number(self.serial_number)
cert_builder = cert_builder.not_valid_before(self.notBefore)
cert_builder = cert_builder.not_valid_after(self.notAfter)
cert_builder = cert_builder.public_key(self.csr.public_key())
has_ski = False
for extension in self.csr.extensions:
if isinstance(extension.value, x509.SubjectKeyIdentifier):
if self.create_subject_key_identifier == 'always_create':
continue
has_ski = True
if self.create_authority_key_identifier and isinstance(extension.value, x509.AuthorityKeyIdentifier):
continue
cert_builder = cert_builder.add_extension(extension.value, critical=extension.critical)
if not has_ski and self.create_subject_key_identifier != 'never_create':
cert_builder = cert_builder.add_extension(
x509.SubjectKeyIdentifier.from_public_key(self.csr.public_key()),
critical=False
)
if self.create_authority_key_identifier:
try:
ext = self.ca_cert.extensions.get_extension_for_class(x509.SubjectKeyIdentifier)
cert_builder = cert_builder.add_extension(
x509.AuthorityKeyIdentifier.from_issuer_subject_key_identifier(ext.value)
if CRYPTOGRAPHY_VERSION >= LooseVersion('2.7') else
x509.AuthorityKeyIdentifier.from_issuer_subject_key_identifier(ext),
critical=False
)
except cryptography.x509.ExtensionNotFound:
cert_builder = cert_builder.add_extension(
x509.AuthorityKeyIdentifier.from_issuer_public_key(self.ca_cert.public_key()),
critical=False
)
try:
certificate = cert_builder.sign(
private_key=self.ca_private_key, algorithm=self.digest,
backend=default_backend()
)
except TypeError as e:
if str(e) == 'Algorithm must be a registered hash algorithm.' and self.digest is None:
self.module.fail_json(msg='Signing with Ed25519 and Ed448 keys requires cryptography 2.8 or newer.')
raise
self.cert = certificate
def get_certificate_data(self):
"""Return bytes for self.cert."""
return self.cert.public_bytes(Encoding.PEM)
def needs_regeneration(self):
if super(OwnCACertificateBackendCryptography, self).needs_regeneration():
return True
# Check AuthorityKeyIdentifier
if self.create_authority_key_identifier:
try:
ext = self.ca_cert.extensions.get_extension_for_class(x509.SubjectKeyIdentifier)
expected_ext = (
x509.AuthorityKeyIdentifier.from_issuer_subject_key_identifier(ext.value)
if CRYPTOGRAPHY_VERSION >= LooseVersion('2.7') else
x509.AuthorityKeyIdentifier.from_issuer_subject_key_identifier(ext)
)
except cryptography.x509.ExtensionNotFound:
expected_ext = x509.AuthorityKeyIdentifier.from_issuer_public_key(self.ca_cert.public_key())
self._ensure_existing_certificate_loaded()
try:
ext = self.existing_certificate.extensions.get_extension_for_class(x509.AuthorityKeyIdentifier)
if ext.value != expected_ext:
return True
except cryptography.x509.ExtensionNotFound as dummy:
return True
return False
def dump(self, include_certificate):
result = super(OwnCACertificateBackendCryptography, 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.strftime("%Y%m%d%H%M%SZ"),
'notAfter': self.notAfter.strftime("%Y%m%d%H%M%SZ"),
'serial_number': self.serial_number,
})
else:
if self.cert is None:
self.cert = self.existing_certificate
result.update({
'notBefore': self.cert.not_valid_before.strftime("%Y%m%d%H%M%SZ"),
'notAfter': self.cert.not_valid_after.strftime("%Y%m%d%H%M%SZ"),
'serial_number': cryptography_serial_number_of_cert(self.cert),
})
return result
def generate_serial_number():
"""Generate a serial number for a certificate"""
while True:
result = randrange(0, 1 << 160)
if result >= 1000:
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:
module.fail_json(msg='One of ownca_path and ownca_content must be specified for the ownca provider.')
if module.params['ownca_privatekey_path'] is None and module.params['ownca_privatekey_content'] is None:
module.fail_json(msg='One of ownca_privatekey_path and ownca_privatekey_content must be specified for the ownca provider.')
def needs_version_two_certs(self, module):
return module.params['ownca_version'] == 2
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):
argument_spec.argument_spec['provider']['choices'].append('ownca')
argument_spec.argument_spec.update(dict(
ownca_path=dict(type='path'),
ownca_content=dict(type='str'),
ownca_privatekey_path=dict(type='path'),
ownca_privatekey_content=dict(type='str', no_log=True),
ownca_privatekey_passphrase=dict(type='str', no_log=True),
ownca_digest=dict(type='str', default='sha256'),
ownca_version=dict(type='int', default=3),
ownca_not_before=dict(type='str', default='+0s'),
ownca_not_after=dict(type='str', default='+3650d'),
ownca_create_subject_key_identifier=dict(
type='str',
default='create_if_not_provided',
choices=['create_if_not_provided', 'always_create', 'never_create']
),
ownca_create_authority_key_identifier=dict(type='bool', default=True),
))
argument_spec.mutually_exclusive.extend([
['ownca_path', 'ownca_content'],
['ownca_privatekey_path', 'ownca_privatekey_content'],
])

263
plugins/module_utils/crypto/module_backends/certificate_selfsigned.py Normal file
View File

@@ -0,0 +1,263 @@
# -*- coding: utf-8 -*-
# Copyright: (c) 2016-2017, Yanis Guenane <yanis+ansible@guenane.org>
# Copyright: (c) 2017, Markus Teufelberger <mteufelberger+ansible@mgit.at>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import os
from random import randrange
from ansible.module_utils._text import to_bytes
from ansible_collections.community.crypto.plugins.module_utils.crypto.support import (
get_relative_time_option,
select_message_digest,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import (
cryptography_key_needs_digest_for_signing,
cryptography_serial_number_of_cert,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.certificate import (
CertificateError,
CertificateBackend,
CertificateProvider,
)
try:
from OpenSSL import crypto
except ImportError:
pass
try:
import cryptography
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.serialization import Encoding
except ImportError:
pass
class SelfSignedCertificateBackendCryptography(CertificateBackend):
def __init__(self, module):
super(SelfSignedCertificateBackendCryptography, self).__init__(module, 'cryptography')
self.create_subject_key_identifier = module.params['selfsigned_create_subject_key_identifier']
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 = select_message_digest(module.params['selfsigned_digest'])
self.version = module.params['selfsigned_version']
self.serial_number = x509.random_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._module = module
self._ensure_private_key_loaded()
self._ensure_csr_loaded()
if self.csr is None:
# Create empty CSR on the fly
csr = cryptography.x509.CertificateSigningRequestBuilder()
csr = csr.subject_name(cryptography.x509.Name([]))
digest = None
if cryptography_key_needs_digest_for_signing(self.privatekey):
digest = self.digest
if digest is None:
self.module.fail_json(msg='Unsupported digest "{0}"'.format(module.params['selfsigned_digest']))
try:
self.csr = csr.sign(self.privatekey, digest, default_backend())
except TypeError as e:
if str(e) == 'Algorithm must be a registered hash algorithm.' and digest is None:
self.module.fail_json(msg='Signing with Ed25519 and Ed448 keys requires cryptography 2.8 or newer.')
raise
if cryptography_key_needs_digest_for_signing(self.privatekey):
if self.digest is None:
raise CertificateError(
'The digest %s is not supported with the cryptography backend' % module.params['selfsigned_digest']
)
else:
self.digest = None
def generate_certificate(self):
"""(Re-)Generate certificate."""
try:
cert_builder = x509.CertificateBuilder()
cert_builder = cert_builder.subject_name(self.csr.subject)
cert_builder = cert_builder.issuer_name(self.csr.subject)
cert_builder = cert_builder.serial_number(self.serial_number)
cert_builder = cert_builder.not_valid_before(self.notBefore)
cert_builder = cert_builder.not_valid_after(self.notAfter)
cert_builder = cert_builder.public_key(self.privatekey.public_key())
has_ski = False
for extension in self.csr.extensions:
if isinstance(extension.value, x509.SubjectKeyIdentifier):
if self.create_subject_key_identifier == 'always_create':
continue
has_ski = True
cert_builder = cert_builder.add_extension(extension.value, critical=extension.critical)
if not has_ski and self.create_subject_key_identifier != 'never_create':
cert_builder = cert_builder.add_extension(
x509.SubjectKeyIdentifier.from_public_key(self.privatekey.public_key()),
critical=False
)
except ValueError as e:
raise CertificateError(str(e))
try:
certificate = cert_builder.sign(
private_key=self.privatekey, algorithm=self.digest,
backend=default_backend()
)
except TypeError as e:
if str(e) == 'Algorithm must be a registered hash algorithm.' and self.digest is None:
self.module.fail_json(msg='Signing with Ed25519 and Ed448 keys requires cryptography 2.8 or newer.')
raise
self.cert = certificate
def get_certificate_data(self):
"""Return bytes for self.cert."""
return self.cert.public_bytes(Encoding.PEM)
def dump(self, include_certificate):
result = super(SelfSignedCertificateBackendCryptography, self).dump(include_certificate)
if self.module.check_mode:
result.update({
'notBefore': self.notBefore.strftime("%Y%m%d%H%M%SZ"),
'notAfter': self.notAfter.strftime("%Y%m%d%H%M%SZ"),
'serial_number': self.serial_number,
})
else:
if self.cert is None:
self.cert = self.existing_certificate
result.update({
'notBefore': self.cert.not_valid_before.strftime("%Y%m%d%H%M%SZ"),
'notAfter': self.cert.not_valid_after.strftime("%Y%m%d%H%M%SZ"),
'serial_number': cryptography_serial_number_of_cert(self.cert),
})
return result
def generate_serial_number():
"""Generate a serial number for a certificate"""
while True:
result = randrange(0, 1 << 160)
if result >= 1000:
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:
module.fail_json(msg='One of privatekey_path and privatekey_content must be specified for the selfsigned provider.')
def needs_version_two_certs(self, module):
return module.params['selfsigned_version'] == 2
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):
argument_spec.argument_spec['provider']['choices'].append('selfsigned')
argument_spec.argument_spec.update(dict(
selfsigned_version=dict(type='int', default=3),
selfsigned_digest=dict(type='str', default='sha256'),
selfsigned_not_before=dict(type='str', default='+0s', aliases=['selfsigned_notBefore']),
selfsigned_not_after=dict(type='str', default='+3650d', aliases=['selfsigned_notAfter']),
selfsigned_create_subject_key_identifier=dict(
type='str',
default='create_if_not_provided',
choices=['create_if_not_provided', 'always_create', 'never_create']
),
))

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

@@ -173,7 +173,7 @@ class CertificateSigningRequestBackend(object):
@abc.abstractmethod
def generate_csr(self):
"""(Re-)Generate private key."""
"""(Re-)Generate CSR."""
pass
@abc.abstractmethod
@@ -182,11 +182,11 @@ class CertificateSigningRequestBackend(object):
pass
def set_existing(self, csr_bytes):
"""Set existing private key bytes. None indicates that the key does not exist."""
"""Set existing CSR bytes. None indicates that the CSR does not exist."""
self.existing_csr_bytes = csr_bytes
def has_existing(self):
"""Query whether an existing private key is/has been there."""
"""Query whether an existing CSR is/has been there."""
return self.existing_csr_bytes is not None
def _ensure_private_key_loaded(self):
@@ -253,7 +253,7 @@ class CertificateSigningRequestPyOpenSSLBackend(CertificateSigningRequestBackend
super(CertificateSigningRequestPyOpenSSLBackend, self).__init__(module, 'pyopenssl')
def generate_csr(self):
"""(Re-)Generate private key."""
"""(Re-)Generate CSR."""
self._ensure_private_key_loaded()
req = crypto.X509Req()
@@ -418,7 +418,7 @@ class CertificateSigningRequestCryptographyBackend(CertificateSigningRequestBack
module.warn('The cryptography backend only supports version 1. (The only valid value according to RFC 2986.)')
def generate_csr(self):
"""(Re-)Generate private key."""
"""(Re-)Generate CSR."""
self._ensure_private_key_loaded()
csr = cryptography.x509.CertificateSigningRequestBuilder()