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

Allow to pass CSR to acme_certificate as csr_content (#115)

Browse Source 
* Allow to pass CSR to acme_certificate as csr_content.

* Make sure contents are bytes.

* No need to write CSR to disk.

* Forgot version_added.

* Fix documentation.
This commit is contained in:
Felix Fontein
2020-10-09 14:01:34 +02:00
committed by GitHub
parent 8e10e1e590
commit 42dd19c387
5 changed files with 80 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
plugins/module_utils/acme.py
View File

@@ -140,27 +140,31 @@ def write_file(module, dest, content):
return changed
def pem_to_der(pem_filename):
def pem_to_der(pem_filename, pem_content=None):
'''
Load PEM file, and convert to DER.
Load PEM file, or use PEM file's content, and convert to DER.
If PEM contains multiple entities, the first entity will be used.
'''
certificate_lines = []
try:
with open(pem_filename, "rt") as f:
header_line_count = 0
for line in f:
if line.startswith('-----'):
header_line_count += 1
if header_line_count == 2:
# If certificate file contains other certs appended
# (like intermediate certificates), ignore these.
break
continue
certificate_lines.append(line.strip())
except Exception as err:
raise ModuleFailException("cannot load PEM file {0}: {1}".format(pem_filename, to_native(err)), exception=traceback.format_exc())
if pem_content is not None:
lines = pem_content.splitlines()
else:
try:
with open(pem_filename, "rt") as f:
lines = list(f)
except Exception as err:
raise ModuleFailException("cannot load PEM file {0}: {1}".format(pem_filename, to_native(err)), exception=traceback.format_exc())
header_line_count = 0
for line in lines:
if line.startswith('-----'):
header_line_count += 1
if header_line_count == 2:
# If certificate file contains other certs appended
# (like intermediate certificates), ignore these.
break
continue
certificate_lines.append(line.strip())
return base64.b64decode(''.join(certificate_lines))
@@ -989,14 +993,20 @@ def _normalize_ip(ip):
return ip
def openssl_get_csr_identifiers(openssl_binary, module, csr_filename):
def openssl_get_csr_identifiers(openssl_binary, module, csr_filename, csr_content=None):
'''
Return a set of requested identifiers (CN and SANs) for the CSR.
Each identifier is a pair (type, identifier), where type is either
'dns' or 'ip'.
'''
openssl_csr_cmd = [openssl_binary, "req", "-in", csr_filename, "-noout", "-text"]
dummy, out, dummy = module.run_command(openssl_csr_cmd, check_rc=True)
filename = csr_filename
data = None
if csr_content is not None:
filename = '-'
data = csr_content.encode('utf-8')
openssl_csr_cmd = [openssl_binary, "req", "-in", filename, "-noout", "-text"]
dummy, out, dummy = module.run_command(openssl_csr_cmd, data=data, check_rc=True)
identifiers = set([])
common_name = re.search(r"Subject:.* CN\s?=\s?([^\s,;/]+)", to_text(out, errors='surrogate_or_strict'))
@@ -1018,14 +1028,18 @@ def openssl_get_csr_identifiers(openssl_binary, module, csr_filename):
return identifiers
def cryptography_get_csr_identifiers(module, csr_filename):
def cryptography_get_csr_identifiers(module, csr_filename, csr_content=None):
'''
Return a set of requested identifiers (CN and SANs) for the CSR.
Each identifier is a pair (type, identifier), where type is either
'dns' or 'ip'.
'''
identifiers = set([])
csr = cryptography.x509.load_pem_x509_csr(read_file(csr_filename), _cryptography_backend)
if csr_content is None:
csr_content = read_file(csr_filename)
else:
csr_content = to_bytes(csr_content)
csr = cryptography.x509.load_pem_x509_csr(csr_content, _cryptography_backend)
for sub in csr.subject:
if sub.oid == cryptography.x509.oid.NameOID.COMMON_NAME:
identifiers.add(('dns', sub.value))

34
plugins/modules/acme_certificate.py
View File

@@ -125,9 +125,23 @@ options:
account key. This is a bad idea from a security point of view, and
the CA should not accept the CSR. The ACME server should return an
error in this case."
- Precisely one of I(csr) or I(csr_content) must be specified.
type: path
required: true
aliases: ['src']
csr_content:
description:
- "Content of the CSR for the new certificate."
- "Can be created with C(openssl req ...)."
- "The CSR may contain multiple Subject Alternate Names, but each one
will lead to an individual challenge that must be fulfilled for the
CSR to be signed."
- "I(Note): the private key used to create the CSR I(must not) be the
account key. This is a bad idea from a security point of view, and
the CA should not accept the CSR. The ACME server should return an
error in this case."
- Precisely one of I(csr) or I(csr_content) must be specified.
type: str
version_added: 1.2.0
data:
description:
- "The data to validate ongoing challenges. This must be specified for
@@ -279,7 +293,7 @@ EXAMPLES = r'''
- name: Create a challenge for sample.com using a account key file.
community.crypto.acme_certificate:
account_key_src: /etc/pki/cert/private/account.key
csr: /etc/pki/cert/csr/sample.com.csr
csr_content: "{{ lookup('file', '/etc/pki/cert/csr/sample.com.csr') }}"
dest: /etc/httpd/ssl/sample.com.crt
fullchain_dest: /etc/httpd/ssl/sample.com-fullchain.crt
register: sample_com_challenge
@@ -576,6 +590,7 @@ class ACMEClient(object):
self.version = module.params['acme_version']
self.challenge = module.params['challenge']
self.csr = module.params['csr']
self.csr_content = module.params['csr_content']
self.dest = module.params.get('dest')
self.fullchain_dest = module.params.get('fullchain_dest')
self.chain_dest = module.params.get('chain_dest')
@@ -613,7 +628,7 @@ class ACMEClient(object):
# signed ACME request.
pass
if not os.path.exists(self.csr):
if self.csr is not None and not os.path.exists(self.csr):
raise ModuleFailException("CSR %s not found" % (self.csr))
self._openssl_bin = module.get_bin_path('openssl', True)
@@ -626,9 +641,9 @@ class ACMEClient(object):
Parse the CSR and return the list of requested identifiers
'''
if HAS_CURRENT_CRYPTOGRAPHY:
return cryptography_get_csr_identifiers(self.module, self.csr)
return cryptography_get_csr_identifiers(self.module, self.csr, self.csr_content)
else:
return openssl_get_csr_identifiers(self._openssl_bin, self.module, self.csr)
return openssl_get_csr_identifiers(self._openssl_bin, self.module, self.csr, self.csr_content)
def _add_or_update_auth(self, identifier_type, identifier, auth):
'''
@@ -767,7 +782,7 @@ class ACMEClient(object):
Return the certificate object as dict
https://tools.ietf.org/html/rfc8555#section-7.4
'''
csr = pem_to_der(self.csr)
csr = pem_to_der(self.csr, self.csr_content)
new_cert = {
"csr": nopad_b64(csr),
}
@@ -844,7 +859,7 @@ class ACMEClient(object):
Return the certificate object as dict
https://tools.ietf.org/html/draft-ietf-acme-acme-02#section-6.5
'''
csr = pem_to_der(self.csr)
csr = pem_to_der(self.csr, self.csr_content)
new_cert = {
"resource": "new-cert",
"csr": nopad_b64(csr),
@@ -1177,7 +1192,8 @@ def main():
agreement=dict(type='str'),
terms_agreed=dict(type='bool', default=False),
challenge=dict(type='str', default='http-01', choices=['http-01', 'dns-01', 'tls-alpn-01']),
csr=dict(type='path', required=True, aliases=['src']),
csr=dict(type='path', aliases=['src']),
csr_content=dict(type='str'),
data=dict(type='dict'),
dest=dict(type='path', aliases=['cert']),
fullchain_dest=dict(type='path', aliases=['fullchain']),
@@ -1199,9 +1215,11 @@ def main():
required_one_of=(
['account_key_src', 'account_key_content'],
['dest', 'fullchain_dest'],
['csr', 'csr_content'],
),
mutually_exclusive=(
['account_key_src', 'account_key_content'],
['csr', 'csr_content'],
),
supports_check_mode=True,
)