7c8b046b5f
* Copy in incidental posix tests. * Update incidental test aliases. * Update target names. * Add support plugins. * Fix paths. * Update ignores. * Update integration-aliases sanity test. * Add incidental tests to CI.
1159 lines
53 KiB
Python
1159 lines
53 KiB
Python
#!/usr/bin/python
|
|
# -*- coding: utf-8 -*-
|
|
|
|
# Copyrigt: (c) 2017, Yanis Guenane <yanis+ansible@guenane.org>
|
|
# 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
|
|
|
|
ANSIBLE_METADATA = {'metadata_version': '1.1',
|
|
'status': ['preview'],
|
|
'supported_by': 'community'}
|
|
|
|
DOCUMENTATION = r'''
|
|
---
|
|
module: openssl_csr
|
|
version_added: '2.4'
|
|
short_description: Generate OpenSSL Certificate Signing Request (CSR)
|
|
description:
|
|
- This module allows one to (re)generate OpenSSL certificate signing requests.
|
|
- It uses the pyOpenSSL python library to interact with openssl. This module supports
|
|
the subjectAltName, keyUsage, extendedKeyUsage, basicConstraints and OCSP Must Staple
|
|
extensions.
|
|
- "Please note that the module regenerates existing CSR if it doesn't match the module's
|
|
options, or if it seems to be corrupt. If you are concerned that this could overwrite
|
|
your existing CSR, consider using the I(backup) option."
|
|
- The module can use the cryptography Python library, or the pyOpenSSL Python
|
|
library. By default, it tries to detect which one is available. This can be
|
|
overridden with the I(select_crypto_backend) option. Please note that the
|
|
PyOpenSSL backend was deprecated in Ansible 2.9 and will be removed in Ansible 2.13."
|
|
requirements:
|
|
- Either cryptography >= 1.3
|
|
- Or pyOpenSSL >= 0.15
|
|
author:
|
|
- Yanis Guenane (@Spredzy)
|
|
options:
|
|
state:
|
|
description:
|
|
- Whether the certificate signing request should exist or not, taking action if the state is different from what is stated.
|
|
type: str
|
|
default: present
|
|
choices: [ absent, present ]
|
|
digest:
|
|
description:
|
|
- The digest used when signing the certificate signing request with the private key.
|
|
type: str
|
|
default: sha256
|
|
privatekey_path:
|
|
description:
|
|
- The path to the private key to use when signing the certificate signing request.
|
|
- Either I(privatekey_path) or I(privatekey_content) must be specified if I(state) is C(present), but not both.
|
|
type: path
|
|
privatekey_content:
|
|
description:
|
|
- The content of the private key to use when signing the certificate signing request.
|
|
- Either I(privatekey_path) or I(privatekey_content) must be specified if I(state) is C(present), but not both.
|
|
type: str
|
|
version_added: "2.10"
|
|
privatekey_passphrase:
|
|
description:
|
|
- The passphrase for the private key.
|
|
- This is required if the private key is password protected.
|
|
type: str
|
|
version:
|
|
description:
|
|
- The version of the certificate signing request.
|
|
- "The only allowed value according to L(RFC 2986,https://tools.ietf.org/html/rfc2986#section-4.1)
|
|
is 1."
|
|
- This option will no longer accept unsupported values from Ansible 2.14 on.
|
|
type: int
|
|
default: 1
|
|
force:
|
|
description:
|
|
- Should the certificate signing request be forced regenerated by this ansible module.
|
|
type: bool
|
|
default: no
|
|
path:
|
|
description:
|
|
- The name of the file into which the generated OpenSSL certificate signing request will be written.
|
|
type: path
|
|
required: true
|
|
subject:
|
|
description:
|
|
- Key/value pairs that will be present in the subject name field of the certificate signing request.
|
|
- If you need to specify more than one value with the same key, use a list as value.
|
|
type: dict
|
|
version_added: '2.5'
|
|
country_name:
|
|
description:
|
|
- The countryName field of the certificate signing request subject.
|
|
type: str
|
|
aliases: [ C, countryName ]
|
|
state_or_province_name:
|
|
description:
|
|
- The stateOrProvinceName field of the certificate signing request subject.
|
|
type: str
|
|
aliases: [ ST, stateOrProvinceName ]
|
|
locality_name:
|
|
description:
|
|
- The localityName field of the certificate signing request subject.
|
|
type: str
|
|
aliases: [ L, localityName ]
|
|
organization_name:
|
|
description:
|
|
- The organizationName field of the certificate signing request subject.
|
|
type: str
|
|
aliases: [ O, organizationName ]
|
|
organizational_unit_name:
|
|
description:
|
|
- The organizationalUnitName field of the certificate signing request subject.
|
|
type: str
|
|
aliases: [ OU, organizationalUnitName ]
|
|
common_name:
|
|
description:
|
|
- The commonName field of the certificate signing request subject.
|
|
type: str
|
|
aliases: [ CN, commonName ]
|
|
email_address:
|
|
description:
|
|
- The emailAddress field of the certificate signing request subject.
|
|
type: str
|
|
aliases: [ E, emailAddress ]
|
|
subject_alt_name:
|
|
description:
|
|
- SAN extension to attach to the certificate signing request.
|
|
- This can either be a 'comma separated string' or a YAML list.
|
|
- Values must be prefixed by their options. (i.e., C(email), C(URI), C(DNS), C(RID), C(IP), C(dirName),
|
|
C(otherName) and the ones specific to your CA)
|
|
- Note that if no SAN is specified, but a common name, the common
|
|
name will be added as a SAN except if C(useCommonNameForSAN) is
|
|
set to I(false).
|
|
- More at U(https://tools.ietf.org/html/rfc5280#section-4.2.1.6).
|
|
type: list
|
|
elements: str
|
|
aliases: [ subjectAltName ]
|
|
subject_alt_name_critical:
|
|
description:
|
|
- Should the subjectAltName extension be considered as critical.
|
|
type: bool
|
|
aliases: [ subjectAltName_critical ]
|
|
use_common_name_for_san:
|
|
description:
|
|
- If set to C(yes), the module will fill the common name in for
|
|
C(subject_alt_name) with C(DNS:) prefix if no SAN is specified.
|
|
type: bool
|
|
default: yes
|
|
version_added: '2.8'
|
|
aliases: [ useCommonNameForSAN ]
|
|
key_usage:
|
|
description:
|
|
- This defines the purpose (e.g. encipherment, signature, certificate signing)
|
|
of the key contained in the certificate.
|
|
type: list
|
|
elements: str
|
|
aliases: [ keyUsage ]
|
|
key_usage_critical:
|
|
description:
|
|
- Should the keyUsage extension be considered as critical.
|
|
type: bool
|
|
aliases: [ keyUsage_critical ]
|
|
extended_key_usage:
|
|
description:
|
|
- Additional restrictions (e.g. client authentication, server authentication)
|
|
on the allowed purposes for which the public key may be used.
|
|
type: list
|
|
elements: str
|
|
aliases: [ extKeyUsage, extendedKeyUsage ]
|
|
extended_key_usage_critical:
|
|
description:
|
|
- Should the extkeyUsage extension be considered as critical.
|
|
type: bool
|
|
aliases: [ extKeyUsage_critical, extendedKeyUsage_critical ]
|
|
basic_constraints:
|
|
description:
|
|
- Indicates basic constraints, such as if the certificate is a CA.
|
|
type: list
|
|
elements: str
|
|
version_added: '2.5'
|
|
aliases: [ basicConstraints ]
|
|
basic_constraints_critical:
|
|
description:
|
|
- Should the basicConstraints extension be considered as critical.
|
|
type: bool
|
|
version_added: '2.5'
|
|
aliases: [ basicConstraints_critical ]
|
|
ocsp_must_staple:
|
|
description:
|
|
- Indicates that the certificate should contain the OCSP Must Staple
|
|
extension (U(https://tools.ietf.org/html/rfc7633)).
|
|
type: bool
|
|
version_added: '2.5'
|
|
aliases: [ ocspMustStaple ]
|
|
ocsp_must_staple_critical:
|
|
description:
|
|
- Should the OCSP Must Staple extension be considered as critical
|
|
- Note that according to the RFC, this extension should not be marked
|
|
as critical, as old clients not knowing about OCSP Must Staple
|
|
are required to reject such certificates
|
|
(see U(https://tools.ietf.org/html/rfc7633#section-4)).
|
|
type: bool
|
|
version_added: '2.5'
|
|
aliases: [ ocspMustStaple_critical ]
|
|
select_crypto_backend:
|
|
description:
|
|
- Determines which crypto backend to use.
|
|
- The default choice is C(auto), which tries to use C(cryptography) if available, and falls back to C(pyopenssl).
|
|
- If set to C(pyopenssl), will try to use the L(pyOpenSSL,https://pypi.org/project/pyOpenSSL/) library.
|
|
- If set to C(cryptography), will try to use the L(cryptography,https://cryptography.io/) library.
|
|
- Please note that the C(pyopenssl) backend has been deprecated in Ansible 2.9, and will be removed in Ansible 2.13.
|
|
From that point on, only the C(cryptography) backend will be available.
|
|
type: str
|
|
default: auto
|
|
choices: [ auto, cryptography, pyopenssl ]
|
|
version_added: '2.8'
|
|
backup:
|
|
description:
|
|
- Create a backup file including a timestamp so you can get the original
|
|
CSR back if you overwrote it with a new one by accident.
|
|
type: bool
|
|
default: no
|
|
version_added: "2.8"
|
|
create_subject_key_identifier:
|
|
description:
|
|
- Create the Subject Key Identifier from the public key.
|
|
- "Please note that commercial CAs can ignore the value, respectively use a value of
|
|
their own choice instead. Specifying this option is mostly useful for self-signed
|
|
certificates or for own CAs."
|
|
- Note that this is only supported if the C(cryptography) backend is used!
|
|
type: bool
|
|
default: no
|
|
version_added: "2.9"
|
|
subject_key_identifier:
|
|
description:
|
|
- The subject key identifier as a hex string, where two bytes are separated by colons.
|
|
- "Example: C(00:11:22:33:44:55:66:77:88:99:aa:bb:cc:dd:ee:ff:00:11:22:33)"
|
|
- "Please note that commercial CAs ignore this value, respectively use a value of their
|
|
own choice. Specifying this option is mostly useful for self-signed certificates
|
|
or for own CAs."
|
|
- Note that this option can only be used if I(create_subject_key_identifier) is C(no).
|
|
- Note that this is only supported if the C(cryptography) backend is used!
|
|
type: str
|
|
version_added: "2.9"
|
|
authority_key_identifier:
|
|
description:
|
|
- The authority key identifier as a hex string, where two bytes are separated by colons.
|
|
- "Example: C(00:11:22:33:44:55:66:77:88:99:aa:bb:cc:dd:ee:ff:00:11:22:33)"
|
|
- If specified, I(authority_cert_issuer) must also be specified.
|
|
- "Please note that commercial CAs ignore this value, respectively use a value of their
|
|
own choice. Specifying this option is mostly useful for self-signed certificates
|
|
or for own CAs."
|
|
- Note that this is only supported if the C(cryptography) backend is used!
|
|
- The C(AuthorityKeyIdentifier) will only be added if at least one of I(authority_key_identifier),
|
|
I(authority_cert_issuer) and I(authority_cert_serial_number) is specified.
|
|
type: str
|
|
version_added: "2.9"
|
|
authority_cert_issuer:
|
|
description:
|
|
- Names that will be present in the authority cert issuer field of the certificate signing request.
|
|
- Values must be prefixed by their options. (i.e., C(email), C(URI), C(DNS), C(RID), C(IP), C(dirName),
|
|
C(otherName) and the ones specific to your CA)
|
|
- "Example: C(DNS:ca.example.org)"
|
|
- If specified, I(authority_key_identifier) must also be specified.
|
|
- "Please note that commercial CAs ignore this value, respectively use a value of their
|
|
own choice. Specifying this option is mostly useful for self-signed certificates
|
|
or for own CAs."
|
|
- Note that this is only supported if the C(cryptography) backend is used!
|
|
- The C(AuthorityKeyIdentifier) will only be added if at least one of I(authority_key_identifier),
|
|
I(authority_cert_issuer) and I(authority_cert_serial_number) is specified.
|
|
type: list
|
|
elements: str
|
|
version_added: "2.9"
|
|
authority_cert_serial_number:
|
|
description:
|
|
- The authority cert serial number.
|
|
- Note that this is only supported if the C(cryptography) backend is used!
|
|
- "Please note that commercial CAs ignore this value, respectively use a value of their
|
|
own choice. Specifying this option is mostly useful for self-signed certificates
|
|
or for own CAs."
|
|
- The C(AuthorityKeyIdentifier) will only be added if at least one of I(authority_key_identifier),
|
|
I(authority_cert_issuer) and I(authority_cert_serial_number) is specified.
|
|
type: int
|
|
version_added: "2.9"
|
|
return_content:
|
|
description:
|
|
- If set to C(yes), will return the (current or generated) CSR's content as I(csr).
|
|
type: bool
|
|
default: no
|
|
version_added: "2.10"
|
|
extends_documentation_fragment:
|
|
- files
|
|
notes:
|
|
- If the certificate signing request already exists it will be checked whether subjectAltName,
|
|
keyUsage, extendedKeyUsage and basicConstraints only contain the requested values, whether
|
|
OCSP Must Staple is as requested, and if the request was signed by the given private key.
|
|
seealso:
|
|
- module: openssl_certificate
|
|
- module: openssl_dhparam
|
|
- module: openssl_pkcs12
|
|
- module: openssl_privatekey
|
|
- module: openssl_publickey
|
|
'''
|
|
|
|
EXAMPLES = r'''
|
|
- name: Generate an OpenSSL Certificate Signing Request
|
|
openssl_csr:
|
|
path: /etc/ssl/csr/www.ansible.com.csr
|
|
privatekey_path: /etc/ssl/private/ansible.com.pem
|
|
common_name: www.ansible.com
|
|
|
|
- name: Generate an OpenSSL Certificate Signing Request with an inline key
|
|
openssl_csr:
|
|
path: /etc/ssl/csr/www.ansible.com.csr
|
|
privatekey_content: "{{ private_key_content }}"
|
|
common_name: www.ansible.com
|
|
|
|
- name: Generate an OpenSSL Certificate Signing Request with a passphrase protected private key
|
|
openssl_csr:
|
|
path: /etc/ssl/csr/www.ansible.com.csr
|
|
privatekey_path: /etc/ssl/private/ansible.com.pem
|
|
privatekey_passphrase: ansible
|
|
common_name: www.ansible.com
|
|
|
|
- name: Generate an OpenSSL Certificate Signing Request with Subject information
|
|
openssl_csr:
|
|
path: /etc/ssl/csr/www.ansible.com.csr
|
|
privatekey_path: /etc/ssl/private/ansible.com.pem
|
|
country_name: FR
|
|
organization_name: Ansible
|
|
email_address: jdoe@ansible.com
|
|
common_name: www.ansible.com
|
|
|
|
- name: Generate an OpenSSL Certificate Signing Request with subjectAltName extension
|
|
openssl_csr:
|
|
path: /etc/ssl/csr/www.ansible.com.csr
|
|
privatekey_path: /etc/ssl/private/ansible.com.pem
|
|
subject_alt_name: 'DNS:www.ansible.com,DNS:m.ansible.com'
|
|
|
|
- name: Generate an OpenSSL CSR with subjectAltName extension with dynamic list
|
|
openssl_csr:
|
|
path: /etc/ssl/csr/www.ansible.com.csr
|
|
privatekey_path: /etc/ssl/private/ansible.com.pem
|
|
subject_alt_name: "{{ item.value | map('regex_replace', '^', 'DNS:') | list }}"
|
|
with_dict:
|
|
dns_server:
|
|
- www.ansible.com
|
|
- m.ansible.com
|
|
|
|
- name: Force regenerate an OpenSSL Certificate Signing Request
|
|
openssl_csr:
|
|
path: /etc/ssl/csr/www.ansible.com.csr
|
|
privatekey_path: /etc/ssl/private/ansible.com.pem
|
|
force: yes
|
|
common_name: www.ansible.com
|
|
|
|
- name: Generate an OpenSSL Certificate Signing Request with special key usages
|
|
openssl_csr:
|
|
path: /etc/ssl/csr/www.ansible.com.csr
|
|
privatekey_path: /etc/ssl/private/ansible.com.pem
|
|
common_name: www.ansible.com
|
|
key_usage:
|
|
- digitalSignature
|
|
- keyAgreement
|
|
extended_key_usage:
|
|
- clientAuth
|
|
|
|
- name: Generate an OpenSSL Certificate Signing Request with OCSP Must Staple
|
|
openssl_csr:
|
|
path: /etc/ssl/csr/www.ansible.com.csr
|
|
privatekey_path: /etc/ssl/private/ansible.com.pem
|
|
common_name: www.ansible.com
|
|
ocsp_must_staple: yes
|
|
'''
|
|
|
|
RETURN = r'''
|
|
privatekey:
|
|
description:
|
|
- Path to the TLS/SSL private key the CSR was generated for
|
|
- Will be C(none) if the private key has been provided in I(privatekey_content).
|
|
returned: changed or success
|
|
type: str
|
|
sample: /etc/ssl/private/ansible.com.pem
|
|
filename:
|
|
description: Path to the generated Certificate Signing Request
|
|
returned: changed or success
|
|
type: str
|
|
sample: /etc/ssl/csr/www.ansible.com.csr
|
|
subject:
|
|
description: A list of the subject tuples attached to the CSR
|
|
returned: changed or success
|
|
type: list
|
|
elements: list
|
|
sample: "[('CN', 'www.ansible.com'), ('O', 'Ansible')]"
|
|
subjectAltName:
|
|
description: The alternative names this CSR is valid for
|
|
returned: changed or success
|
|
type: list
|
|
elements: str
|
|
sample: [ 'DNS:www.ansible.com', 'DNS:m.ansible.com' ]
|
|
keyUsage:
|
|
description: Purpose for which the public key may be used
|
|
returned: changed or success
|
|
type: list
|
|
elements: str
|
|
sample: [ 'digitalSignature', 'keyAgreement' ]
|
|
extendedKeyUsage:
|
|
description: Additional restriction on the public key purposes
|
|
returned: changed or success
|
|
type: list
|
|
elements: str
|
|
sample: [ 'clientAuth' ]
|
|
basicConstraints:
|
|
description: Indicates if the certificate belongs to a CA
|
|
returned: changed or success
|
|
type: list
|
|
elements: str
|
|
sample: ['CA:TRUE', 'pathLenConstraint:0']
|
|
ocsp_must_staple:
|
|
description: Indicates whether the certificate has the OCSP
|
|
Must Staple feature enabled
|
|
returned: changed or success
|
|
type: bool
|
|
sample: false
|
|
backup_file:
|
|
description: Name of backup file created.
|
|
returned: changed and if I(backup) is C(yes)
|
|
type: str
|
|
sample: /path/to/www.ansible.com.csr.2019-03-09@11:22~
|
|
csr:
|
|
description: The (current or generated) CSR's content.
|
|
returned: if I(state) is C(present) and I(return_content) is C(yes)
|
|
type: str
|
|
version_added: "2.10"
|
|
'''
|
|
|
|
import abc
|
|
import binascii
|
|
import os
|
|
import traceback
|
|
from distutils.version import LooseVersion
|
|
|
|
from ansible.module_utils import crypto as crypto_utils
|
|
from ansible.module_utils.basic import AnsibleModule, missing_required_lib
|
|
from ansible.module_utils._text import to_native, to_bytes, to_text
|
|
from ansible.module_utils.compat import ipaddress as compat_ipaddress
|
|
|
|
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
|
|
import cryptography.x509
|
|
import cryptography.x509.oid
|
|
import cryptography.exceptions
|
|
import cryptography.hazmat.backends
|
|
import cryptography.hazmat.primitives.serialization
|
|
import cryptography.hazmat.primitives.hashes
|
|
CRYPTOGRAPHY_VERSION = LooseVersion(cryptography.__version__)
|
|
except ImportError:
|
|
CRYPTOGRAPHY_IMP_ERR = traceback.format_exc()
|
|
CRYPTOGRAPHY_FOUND = False
|
|
else:
|
|
CRYPTOGRAPHY_FOUND = True
|
|
CRYPTOGRAPHY_MUST_STAPLE_NAME = cryptography.x509.oid.ObjectIdentifier("1.3.6.1.5.5.7.1.24")
|
|
CRYPTOGRAPHY_MUST_STAPLE_VALUE = b"\x30\x03\x02\x01\x05"
|
|
|
|
|
|
class CertificateSigningRequestError(crypto_utils.OpenSSLObjectError):
|
|
pass
|
|
|
|
|
|
class CertificateSigningRequestBase(crypto_utils.OpenSSLObject):
|
|
|
|
def __init__(self, module):
|
|
super(CertificateSigningRequestBase, self).__init__(
|
|
module.params['path'],
|
|
module.params['state'],
|
|
module.params['force'],
|
|
module.check_mode
|
|
)
|
|
self.digest = module.params['digest']
|
|
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.version = module.params['version']
|
|
self.subjectAltName = module.params['subject_alt_name']
|
|
self.subjectAltName_critical = module.params['subject_alt_name_critical']
|
|
self.keyUsage = module.params['key_usage']
|
|
self.keyUsage_critical = module.params['key_usage_critical']
|
|
self.extendedKeyUsage = module.params['extended_key_usage']
|
|
self.extendedKeyUsage_critical = module.params['extended_key_usage_critical']
|
|
self.basicConstraints = module.params['basic_constraints']
|
|
self.basicConstraints_critical = module.params['basic_constraints_critical']
|
|
self.ocspMustStaple = module.params['ocsp_must_staple']
|
|
self.ocspMustStaple_critical = module.params['ocsp_must_staple_critical']
|
|
self.create_subject_key_identifier = module.params['create_subject_key_identifier']
|
|
self.subject_key_identifier = module.params['subject_key_identifier']
|
|
self.authority_key_identifier = module.params['authority_key_identifier']
|
|
self.authority_cert_issuer = module.params['authority_cert_issuer']
|
|
self.authority_cert_serial_number = module.params['authority_cert_serial_number']
|
|
self.request = None
|
|
self.privatekey = None
|
|
self.csr_bytes = None
|
|
self.return_content = module.params['return_content']
|
|
|
|
if self.create_subject_key_identifier and self.subject_key_identifier is not None:
|
|
module.fail_json(msg='subject_key_identifier cannot be specified if create_subject_key_identifier is true')
|
|
|
|
self.backup = module.params['backup']
|
|
self.backup_file = None
|
|
|
|
self.subject = [
|
|
('C', module.params['country_name']),
|
|
('ST', module.params['state_or_province_name']),
|
|
('L', module.params['locality_name']),
|
|
('O', module.params['organization_name']),
|
|
('OU', module.params['organizational_unit_name']),
|
|
('CN', module.params['common_name']),
|
|
('emailAddress', module.params['email_address']),
|
|
]
|
|
|
|
if module.params['subject']:
|
|
self.subject = self.subject + crypto_utils.parse_name_field(module.params['subject'])
|
|
self.subject = [(entry[0], entry[1]) for entry in self.subject if entry[1]]
|
|
|
|
if not self.subjectAltName and module.params['use_common_name_for_san']:
|
|
for sub in self.subject:
|
|
if sub[0] in ('commonName', 'CN'):
|
|
self.subjectAltName = ['DNS:%s' % sub[1]]
|
|
break
|
|
|
|
if self.subject_key_identifier is not None:
|
|
try:
|
|
self.subject_key_identifier = binascii.unhexlify(self.subject_key_identifier.replace(':', ''))
|
|
except Exception as e:
|
|
raise CertificateSigningRequestError('Cannot parse subject_key_identifier: {0}'.format(e))
|
|
|
|
if self.authority_key_identifier is not None:
|
|
try:
|
|
self.authority_key_identifier = binascii.unhexlify(self.authority_key_identifier.replace(':', ''))
|
|
except Exception as e:
|
|
raise CertificateSigningRequestError('Cannot parse authority_key_identifier: {0}'.format(e))
|
|
|
|
@abc.abstractmethod
|
|
def _generate_csr(self):
|
|
pass
|
|
|
|
def generate(self, module):
|
|
'''Generate the certificate signing request.'''
|
|
if not self.check(module, perms_required=False) or self.force:
|
|
result = self._generate_csr()
|
|
if self.backup:
|
|
self.backup_file = module.backup_local(self.path)
|
|
if self.return_content:
|
|
self.csr_bytes = result
|
|
crypto_utils.write_file(module, result)
|
|
self.changed = True
|
|
|
|
file_args = module.load_file_common_arguments(module.params)
|
|
if module.set_fs_attributes_if_different(file_args, False):
|
|
self.changed = True
|
|
|
|
@abc.abstractmethod
|
|
def _load_private_key(self):
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def _check_csr(self):
|
|
pass
|
|
|
|
def check(self, module, perms_required=True):
|
|
"""Ensure the resource is in its desired state."""
|
|
state_and_perms = super(CertificateSigningRequestBase, self).check(module, perms_required)
|
|
|
|
self._load_private_key()
|
|
|
|
if not state_and_perms:
|
|
return False
|
|
|
|
return self._check_csr()
|
|
|
|
def remove(self, module):
|
|
if self.backup:
|
|
self.backup_file = module.backup_local(self.path)
|
|
super(CertificateSigningRequestBase, self).remove(module)
|
|
|
|
def dump(self):
|
|
'''Serialize the object into a dictionary.'''
|
|
|
|
result = {
|
|
'privatekey': self.privatekey_path,
|
|
'filename': self.path,
|
|
'subject': self.subject,
|
|
'subjectAltName': self.subjectAltName,
|
|
'keyUsage': self.keyUsage,
|
|
'extendedKeyUsage': self.extendedKeyUsage,
|
|
'basicConstraints': self.basicConstraints,
|
|
'ocspMustStaple': self.ocspMustStaple,
|
|
'changed': self.changed
|
|
}
|
|
if self.backup_file:
|
|
result['backup_file'] = self.backup_file
|
|
if self.return_content:
|
|
if self.csr_bytes is None:
|
|
self.csr_bytes = crypto_utils.load_file_if_exists(self.path, ignore_errors=True)
|
|
result['csr'] = self.csr_bytes.decode('utf-8') if self.csr_bytes else None
|
|
|
|
return result
|
|
|
|
|
|
class CertificateSigningRequestPyOpenSSL(CertificateSigningRequestBase):
|
|
|
|
def __init__(self, module):
|
|
if module.params['create_subject_key_identifier']:
|
|
module.fail_json(msg='You cannot use create_subject_key_identifier with the pyOpenSSL backend!')
|
|
for o in ('subject_key_identifier', 'authority_key_identifier', 'authority_cert_issuer', 'authority_cert_serial_number'):
|
|
if module.params[o] is not None:
|
|
module.fail_json(msg='You cannot use {0} with the pyOpenSSL backend!'.format(o))
|
|
super(CertificateSigningRequestPyOpenSSL, self).__init__(module)
|
|
|
|
def _generate_csr(self):
|
|
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.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.request = req
|
|
|
|
return crypto.dump_certificate_request(crypto.FILETYPE_PEM, self.request)
|
|
|
|
def _load_private_key(self):
|
|
try:
|
|
self.privatekey = crypto_utils.load_privatekey(
|
|
path=self.privatekey_path,
|
|
content=self.privatekey_content,
|
|
passphrase=self.privatekey_passphrase
|
|
)
|
|
except crypto_utils.OpenSSLBadPassphraseError as exc:
|
|
raise CertificateSigningRequestError(exc)
|
|
|
|
def _normalize_san(self, 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:'):
|
|
ip = compat_ipaddress.ip_address(san[3:])
|
|
san = 'IP:{0}'.format(ip.compressed)
|
|
return san
|
|
|
|
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 = [self._normalize_san(altname.strip()) for altname in
|
|
to_text(altnames_ext, errors='surrogate_or_strict').split(',') if altname.strip()]
|
|
if self.subjectAltName:
|
|
if (set(altnames) != set([self._normalize_san(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_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))
|
|
|
|
def _check_signature(csr):
|
|
try:
|
|
return csr.verify(self.privatekey)
|
|
except crypto.Error:
|
|
return False
|
|
|
|
try:
|
|
csr = crypto_utils.load_certificate_request(self.path, backend='pyopenssl')
|
|
except Exception as dummy:
|
|
return False
|
|
|
|
return _check_subject(csr) and _check_extensions(csr) and _check_signature(csr)
|
|
|
|
|
|
class CertificateSigningRequestCryptography(CertificateSigningRequestBase):
|
|
|
|
def __init__(self, module):
|
|
super(CertificateSigningRequestCryptography, self).__init__(module)
|
|
self.cryptography_backend = cryptography.hazmat.backends.default_backend()
|
|
self.module = module
|
|
if self.version != 1:
|
|
module.warn('The cryptography backend only supports version 1. (The only valid value according to RFC 2986.)')
|
|
|
|
def _generate_csr(self):
|
|
csr = cryptography.x509.CertificateSigningRequestBuilder()
|
|
try:
|
|
csr = csr.subject_name(cryptography.x509.Name([
|
|
cryptography.x509.NameAttribute(crypto_utils.cryptography_name_to_oid(entry[0]), to_text(entry[1])) for entry in self.subject
|
|
]))
|
|
except ValueError as e:
|
|
raise CertificateSigningRequestError(e)
|
|
|
|
if self.subjectAltName:
|
|
csr = csr.add_extension(cryptography.x509.SubjectAlternativeName([
|
|
crypto_utils.cryptography_get_name(name) for name in self.subjectAltName
|
|
]), critical=self.subjectAltName_critical)
|
|
|
|
if self.keyUsage:
|
|
params = crypto_utils.cryptography_parse_key_usage_params(self.keyUsage)
|
|
csr = csr.add_extension(cryptography.x509.KeyUsage(**params), critical=self.keyUsage_critical)
|
|
|
|
if self.extendedKeyUsage:
|
|
usages = [crypto_utils.cryptography_name_to_oid(usage) for usage in self.extendedKeyUsage]
|
|
csr = csr.add_extension(cryptography.x509.ExtendedKeyUsage(usages), critical=self.extendedKeyUsage_critical)
|
|
|
|
if self.basicConstraints:
|
|
params = {}
|
|
ca, path_length = crypto_utils.cryptography_get_basic_constraints(self.basicConstraints)
|
|
csr = csr.add_extension(cryptography.x509.BasicConstraints(ca, path_length), critical=self.basicConstraints_critical)
|
|
|
|
if self.ocspMustStaple:
|
|
try:
|
|
# This only works with cryptography >= 2.1
|
|
csr = csr.add_extension(cryptography.x509.TLSFeature([cryptography.x509.TLSFeatureType.status_request]), critical=self.ocspMustStaple_critical)
|
|
except AttributeError as dummy:
|
|
csr = csr.add_extension(
|
|
cryptography.x509.UnrecognizedExtension(CRYPTOGRAPHY_MUST_STAPLE_NAME, CRYPTOGRAPHY_MUST_STAPLE_VALUE),
|
|
critical=self.ocspMustStaple_critical
|
|
)
|
|
|
|
if self.create_subject_key_identifier:
|
|
csr = csr.add_extension(
|
|
cryptography.x509.SubjectKeyIdentifier.from_public_key(self.privatekey.public_key()),
|
|
critical=False
|
|
)
|
|
elif self.subject_key_identifier is not None:
|
|
csr = csr.add_extension(cryptography.x509.SubjectKeyIdentifier(self.subject_key_identifier), critical=False)
|
|
|
|
if self.authority_key_identifier is not None or self.authority_cert_issuer is not None or self.authority_cert_serial_number is not None:
|
|
issuers = None
|
|
if self.authority_cert_issuer is not None:
|
|
issuers = [crypto_utils.cryptography_get_name(n) for n in self.authority_cert_issuer]
|
|
csr = csr.add_extension(
|
|
cryptography.x509.AuthorityKeyIdentifier(self.authority_key_identifier, issuers, self.authority_cert_serial_number),
|
|
critical=False
|
|
)
|
|
|
|
digest = None
|
|
if crypto_utils.cryptography_key_needs_digest_for_signing(self.privatekey):
|
|
if self.digest == 'sha256':
|
|
digest = cryptography.hazmat.primitives.hashes.SHA256()
|
|
elif self.digest == 'sha384':
|
|
digest = cryptography.hazmat.primitives.hashes.SHA384()
|
|
elif self.digest == 'sha512':
|
|
digest = cryptography.hazmat.primitives.hashes.SHA512()
|
|
elif self.digest == 'sha1':
|
|
digest = cryptography.hazmat.primitives.hashes.SHA1()
|
|
elif self.digest == 'md5':
|
|
digest = cryptography.hazmat.primitives.hashes.MD5()
|
|
# FIXME
|
|
else:
|
|
raise CertificateSigningRequestError('Unsupported digest "{0}"'.format(self.digest))
|
|
try:
|
|
self.request = csr.sign(self.privatekey, digest, self.cryptography_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
|
|
|
|
return self.request.public_bytes(cryptography.hazmat.primitives.serialization.Encoding.PEM)
|
|
|
|
def _load_private_key(self):
|
|
try:
|
|
if self.privatekey_content is not None:
|
|
content = self.privatekey_content
|
|
else:
|
|
with open(self.privatekey_path, 'rb') as f:
|
|
content = f.read()
|
|
self.privatekey = cryptography.hazmat.primitives.serialization.load_pem_private_key(
|
|
content,
|
|
None if self.privatekey_passphrase is None else to_bytes(self.privatekey_passphrase),
|
|
backend=self.cryptography_backend
|
|
)
|
|
except Exception as e:
|
|
raise CertificateSigningRequestError(e)
|
|
|
|
def _check_csr(self):
|
|
def _check_subject(csr):
|
|
subject = [(crypto_utils.cryptography_name_to_oid(entry[0]), entry[1]) for entry in self.subject]
|
|
current_subject = [(sub.oid, sub.value) for sub in csr.subject]
|
|
return set(subject) == set(current_subject)
|
|
|
|
def _find_extension(extensions, exttype):
|
|
return next(
|
|
(ext for ext in extensions if isinstance(ext.value, exttype)),
|
|
None
|
|
)
|
|
|
|
def _check_subjectAltName(extensions):
|
|
current_altnames_ext = _find_extension(extensions, cryptography.x509.SubjectAlternativeName)
|
|
current_altnames = [str(altname) for altname in current_altnames_ext.value] if current_altnames_ext else []
|
|
altnames = [str(crypto_utils.cryptography_get_name(altname)) for altname in self.subjectAltName] if self.subjectAltName else []
|
|
if set(altnames) != set(current_altnames):
|
|
return False
|
|
if altnames:
|
|
if current_altnames_ext.critical != self.subjectAltName_critical:
|
|
return False
|
|
return True
|
|
|
|
def _check_keyUsage(extensions):
|
|
current_keyusage_ext = _find_extension(extensions, cryptography.x509.KeyUsage)
|
|
if not self.keyUsage:
|
|
return current_keyusage_ext is None
|
|
elif current_keyusage_ext is None:
|
|
return False
|
|
params = crypto_utils.cryptography_parse_key_usage_params(self.keyUsage)
|
|
for param in params:
|
|
if getattr(current_keyusage_ext.value, '_' + param) != params[param]:
|
|
return False
|
|
if current_keyusage_ext.critical != self.keyUsage_critical:
|
|
return False
|
|
return True
|
|
|
|
def _check_extenededKeyUsage(extensions):
|
|
current_usages_ext = _find_extension(extensions, cryptography.x509.ExtendedKeyUsage)
|
|
current_usages = [str(usage) for usage in current_usages_ext.value] if current_usages_ext else []
|
|
usages = [str(crypto_utils.cryptography_name_to_oid(usage)) for usage in self.extendedKeyUsage] if self.extendedKeyUsage else []
|
|
if set(current_usages) != set(usages):
|
|
return False
|
|
if usages:
|
|
if current_usages_ext.critical != self.extendedKeyUsage_critical:
|
|
return False
|
|
return True
|
|
|
|
def _check_basicConstraints(extensions):
|
|
bc_ext = _find_extension(extensions, cryptography.x509.BasicConstraints)
|
|
current_ca = bc_ext.value.ca if bc_ext else False
|
|
current_path_length = bc_ext.value.path_length if bc_ext else None
|
|
ca, path_length = crypto_utils.cryptography_get_basic_constraints(self.basicConstraints)
|
|
# Check CA flag
|
|
if ca != current_ca:
|
|
return False
|
|
# Check path length
|
|
if path_length != current_path_length:
|
|
return False
|
|
# Check criticality
|
|
if self.basicConstraints:
|
|
if bc_ext.critical != self.basicConstraints_critical:
|
|
return False
|
|
return True
|
|
|
|
def _check_ocspMustStaple(extensions):
|
|
try:
|
|
# This only works with cryptography >= 2.1
|
|
tlsfeature_ext = _find_extension(extensions, cryptography.x509.TLSFeature)
|
|
has_tlsfeature = True
|
|
except AttributeError as dummy:
|
|
tlsfeature_ext = next(
|
|
(ext for ext in extensions if ext.value.oid == CRYPTOGRAPHY_MUST_STAPLE_NAME),
|
|
None
|
|
)
|
|
has_tlsfeature = False
|
|
if self.ocspMustStaple:
|
|
if not tlsfeature_ext or tlsfeature_ext.critical != self.ocspMustStaple_critical:
|
|
return False
|
|
if has_tlsfeature:
|
|
return cryptography.x509.TLSFeatureType.status_request in tlsfeature_ext.value
|
|
else:
|
|
return tlsfeature_ext.value.value == CRYPTOGRAPHY_MUST_STAPLE_VALUE
|
|
else:
|
|
return tlsfeature_ext is None
|
|
|
|
def _check_subject_key_identifier(extensions):
|
|
ext = _find_extension(extensions, cryptography.x509.SubjectKeyIdentifier)
|
|
if self.create_subject_key_identifier or self.subject_key_identifier is not None:
|
|
if not ext or ext.critical:
|
|
return False
|
|
if self.create_subject_key_identifier:
|
|
digest = cryptography.x509.SubjectKeyIdentifier.from_public_key(self.privatekey.public_key()).digest
|
|
return ext.value.digest == digest
|
|
else:
|
|
return ext.value.digest == self.subject_key_identifier
|
|
else:
|
|
return ext is None
|
|
|
|
def _check_authority_key_identifier(extensions):
|
|
ext = _find_extension(extensions, cryptography.x509.AuthorityKeyIdentifier)
|
|
if self.authority_key_identifier is not None or self.authority_cert_issuer is not None or self.authority_cert_serial_number is not None:
|
|
if not ext or ext.critical:
|
|
return False
|
|
aci = None
|
|
csr_aci = None
|
|
if self.authority_cert_issuer is not None:
|
|
aci = [str(crypto_utils.cryptography_get_name(n)) for n in self.authority_cert_issuer]
|
|
if ext.value.authority_cert_issuer is not None:
|
|
csr_aci = [str(n) for n in ext.value.authority_cert_issuer]
|
|
return (ext.value.key_identifier == self.authority_key_identifier
|
|
and csr_aci == aci
|
|
and ext.value.authority_cert_serial_number == self.authority_cert_serial_number)
|
|
else:
|
|
return ext is None
|
|
|
|
def _check_extensions(csr):
|
|
extensions = csr.extensions
|
|
return (_check_subjectAltName(extensions) and _check_keyUsage(extensions) and
|
|
_check_extenededKeyUsage(extensions) and _check_basicConstraints(extensions) and
|
|
_check_ocspMustStaple(extensions) and _check_subject_key_identifier(extensions) and
|
|
_check_authority_key_identifier(extensions))
|
|
|
|
def _check_signature(csr):
|
|
if not csr.is_signature_valid:
|
|
return False
|
|
# To check whether public key of CSR belongs to private key,
|
|
# encode both public keys and compare PEMs.
|
|
key_a = csr.public_key().public_bytes(
|
|
cryptography.hazmat.primitives.serialization.Encoding.PEM,
|
|
cryptography.hazmat.primitives.serialization.PublicFormat.SubjectPublicKeyInfo
|
|
)
|
|
key_b = self.privatekey.public_key().public_bytes(
|
|
cryptography.hazmat.primitives.serialization.Encoding.PEM,
|
|
cryptography.hazmat.primitives.serialization.PublicFormat.SubjectPublicKeyInfo
|
|
)
|
|
return key_a == key_b
|
|
|
|
try:
|
|
csr = crypto_utils.load_certificate_request(self.path, backend='cryptography')
|
|
except Exception as dummy:
|
|
return False
|
|
|
|
return _check_subject(csr) and _check_extensions(csr) and _check_signature(csr)
|
|
|
|
|
|
def main():
|
|
module = AnsibleModule(
|
|
argument_spec=dict(
|
|
state=dict(type='str', default='present', choices=['absent', 'present']),
|
|
digest=dict(type='str', default='sha256'),
|
|
privatekey_path=dict(type='path'),
|
|
privatekey_content=dict(type='str'),
|
|
privatekey_passphrase=dict(type='str', no_log=True),
|
|
version=dict(type='int', default=1),
|
|
force=dict(type='bool', default=False),
|
|
path=dict(type='path', required=True),
|
|
subject=dict(type='dict'),
|
|
country_name=dict(type='str', aliases=['C', 'countryName']),
|
|
state_or_province_name=dict(type='str', aliases=['ST', 'stateOrProvinceName']),
|
|
locality_name=dict(type='str', aliases=['L', 'localityName']),
|
|
organization_name=dict(type='str', aliases=['O', 'organizationName']),
|
|
organizational_unit_name=dict(type='str', aliases=['OU', 'organizationalUnitName']),
|
|
common_name=dict(type='str', aliases=['CN', 'commonName']),
|
|
email_address=dict(type='str', aliases=['E', 'emailAddress']),
|
|
subject_alt_name=dict(type='list', elements='str', aliases=['subjectAltName']),
|
|
subject_alt_name_critical=dict(type='bool', default=False, aliases=['subjectAltName_critical']),
|
|
use_common_name_for_san=dict(type='bool', default=True, aliases=['useCommonNameForSAN']),
|
|
key_usage=dict(type='list', elements='str', aliases=['keyUsage']),
|
|
key_usage_critical=dict(type='bool', default=False, aliases=['keyUsage_critical']),
|
|
extended_key_usage=dict(type='list', elements='str', aliases=['extKeyUsage', 'extendedKeyUsage']),
|
|
extended_key_usage_critical=dict(type='bool', default=False, aliases=['extKeyUsage_critical', 'extendedKeyUsage_critical']),
|
|
basic_constraints=dict(type='list', elements='str', aliases=['basicConstraints']),
|
|
basic_constraints_critical=dict(type='bool', default=False, aliases=['basicConstraints_critical']),
|
|
ocsp_must_staple=dict(type='bool', default=False, aliases=['ocspMustStaple']),
|
|
ocsp_must_staple_critical=dict(type='bool', default=False, aliases=['ocspMustStaple_critical']),
|
|
backup=dict(type='bool', default=False),
|
|
create_subject_key_identifier=dict(type='bool', default=False),
|
|
subject_key_identifier=dict(type='str'),
|
|
authority_key_identifier=dict(type='str'),
|
|
authority_cert_issuer=dict(type='list', elements='str'),
|
|
authority_cert_serial_number=dict(type='int'),
|
|
select_crypto_backend=dict(type='str', default='auto', choices=['auto', 'cryptography', 'pyopenssl']),
|
|
return_content=dict(type='bool', default=False),
|
|
),
|
|
required_together=[('authority_cert_issuer', 'authority_cert_serial_number')],
|
|
required_if=[('state', 'present', ['privatekey_path', 'privatekey_content'], True)],
|
|
mutually_exclusive=(
|
|
['privatekey_path', 'privatekey_content'],
|
|
),
|
|
add_file_common_args=True,
|
|
supports_check_mode=True,
|
|
)
|
|
|
|
if module.params['version'] != 1:
|
|
module.deprecate('The version option will only support allowed values from Ansible 2.14 on. '
|
|
'Currently, only the value 1 is allowed by RFC 2986', version='2.14')
|
|
|
|
base_dir = os.path.dirname(module.params['path']) or '.'
|
|
if not os.path.isdir(base_dir):
|
|
module.fail_json(name=base_dir, msg='The directory %s does not exist or the file is not a directory' % base_dir)
|
|
|
|
backend = module.params['select_crypto_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
|
|
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))
|
|
try:
|
|
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.13')
|
|
csr = CertificateSigningRequestPyOpenSSL(module)
|
|
elif backend == 'cryptography':
|
|
if not CRYPTOGRAPHY_FOUND:
|
|
module.fail_json(msg=missing_required_lib('cryptography >= {0}'.format(MINIMAL_CRYPTOGRAPHY_VERSION)),
|
|
exception=CRYPTOGRAPHY_IMP_ERR)
|
|
csr = CertificateSigningRequestCryptography(module)
|
|
|
|
if module.params['state'] == 'present':
|
|
if module.check_mode:
|
|
result = csr.dump()
|
|
result['changed'] = module.params['force'] or not csr.check(module)
|
|
module.exit_json(**result)
|
|
|
|
csr.generate(module)
|
|
|
|
else:
|
|
if module.check_mode:
|
|
result = csr.dump()
|
|
result['changed'] = os.path.exists(module.params['path'])
|
|
module.exit_json(**result)
|
|
|
|
csr.remove(module)
|
|
|
|
result = csr.dump()
|
|
module.exit_json(**result)
|
|
except crypto_utils.OpenSSLObjectError as exc:
|
|
module.fail_json(msg=to_native(exc))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|