mirror of
https://github.com/ansible-collections/community.general.git
synced 2025-07-23 21:30:22 -07:00
ACME: use Cryptography (if a new enough version is available) instead of OpenSSL (#42170)
* Collecting PEM -> DER conversions. * Using cryptography instead of OpenSSL binary in some situations. * Moving key-to-disk writing for key content to parse_account_key. * Rename parse_account_key -> parse_key. * Move OpenSSL specific code for key parsing and request signing into global functions. * Also using cryptography for key parsing and request signing. * Remove assert statements. * Fixing handling of key contents for cryptography code path. * Allow to disable the use of cryptography. * Updating documentation. * 1.5 seems to work as well (earlier versions don't have EC sign function). Making Python 2.x adjustments. * Changing option to select_crypto_backend. * Python 2.6 compatibility. * Trying to test both backends separately for acme_account. * Also testing both backends separately for acme_certificate and acme_certificate_revoke. * Adding changelog entry which informs about select_crypto_backend option in case autodetect fails. * Fixing YAML.
This commit is contained in:
Felix Fontein
René Moser
parent
7f41f0168a
commit
aef16ee195
13 changed files with 1031 additions and 671 deletions
|
|
@ -17,17 +17,38 @@ __metaclass__ = type
|
|||
import base64
|
||||
import binascii
|
||||
import copy
|
||||
import datetime
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import traceback
|
||||
|
||||
from ansible.module_utils._text import to_native, to_text, to_bytes
|
||||
from ansible.module_utils.urls import fetch_url as _fetch_url
|
||||
|
||||
try:
|
||||
import cryptography
|
||||
import cryptography.hazmat.backends
|
||||
import cryptography.hazmat.primitives.serialization
|
||||
import cryptography.hazmat.primitives.asymmetric.rsa
|
||||
import cryptography.hazmat.primitives.asymmetric.ec
|
||||
import cryptography.hazmat.primitives.asymmetric.padding
|
||||
import cryptography.hazmat.primitives.hashes
|
||||
import cryptography.hazmat.primitives.asymmetric.utils
|
||||
import cryptography.x509
|
||||
import cryptography.x509.oid
|
||||
from distutils.version import LooseVersion
|
||||
CRYPTOGRAPHY_VERSION = cryptography.__version__
|
||||
HAS_CURRENT_CRYPTOGRAPHY = (LooseVersion(CRYPTOGRAPHY_VERSION) >= LooseVersion('1.5'))
|
||||
if HAS_CURRENT_CRYPTOGRAPHY:
|
||||
_cryptography_backend = cryptography.hazmat.backends.default_backend()
|
||||
except Exception as _:
|
||||
HAS_CURRENT_CRYPTOGRAPHY = False
|
||||
|
||||
|
||||
class ModuleFailException(Exception):
|
||||
'''
|
||||
|
|
@ -83,6 +104,14 @@ def simple_get(module, url):
|
|||
return result
|
||||
|
||||
|
||||
def read_file(fn, mode='b'):
|
||||
try:
|
||||
with open(fn, 'r' + mode) as f:
|
||||
return f.read()
|
||||
except Exception as e:
|
||||
raise ModuleFailException('Error while reading file "{0}": {1}'.format(fn, e))
|
||||
|
||||
|
||||
# function source: network/basics/uri.py
|
||||
def write_file(module, dest, content):
|
||||
'''
|
||||
|
|
@ -141,6 +170,296 @@ def write_file(module, dest, content):
|
|||
return changed
|
||||
|
||||
|
||||
def pem_to_der(pem_filename):
|
||||
'''
|
||||
Load PEM file, 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())
|
||||
return base64.b64decode(''.join(certificate_lines))
|
||||
|
||||
|
||||
def _parse_key_openssl(openssl_binary, module, key_file=None, key_content=None):
|
||||
'''
|
||||
Parses an RSA or Elliptic Curve key file in PEM format and returns a pair
|
||||
(error, key_data).
|
||||
'''
|
||||
# If key_file isn't given, but key_content, write that to a temporary file
|
||||
if key_file is None:
|
||||
fd, tmpsrc = tempfile.mkstemp()
|
||||
module.add_cleanup_file(tmpsrc) # Ansible will delete the file on exit
|
||||
f = os.fdopen(fd, 'wb')
|
||||
try:
|
||||
f.write(key_content.encode('utf-8'))
|
||||
key_file = tmpsrc
|
||||
except Exception as err:
|
||||
try:
|
||||
f.close()
|
||||
except Exception as e:
|
||||
pass
|
||||
raise ModuleFailException("failed to create temporary content file: %s" % to_native(err), exception=traceback.format_exc())
|
||||
f.close()
|
||||
# Parse key
|
||||
account_key_type = None
|
||||
with open(key_file, "rt") as f:
|
||||
for line in f:
|
||||
m = re.match(r"^\s*-{5,}BEGIN\s+(EC|RSA)\s+PRIVATE\s+KEY-{5,}\s*$", line)
|
||||
if m is not None:
|
||||
account_key_type = m.group(1).lower()
|
||||
break
|
||||
if account_key_type is None:
|
||||
# This happens for example if openssl_privatekey created this key
|
||||
# (as opposed to the OpenSSL binary). For now, we assume this is
|
||||
# an RSA key.
|
||||
# FIXME: add some kind of auto-detection
|
||||
account_key_type = "rsa"
|
||||
if account_key_type not in ("rsa", "ec"):
|
||||
return 'unknown key type "%s"' % account_key_type, {}
|
||||
|
||||
openssl_keydump_cmd = [openssl_binary, account_key_type, "-in", key_file, "-noout", "-text"]
|
||||
dummy, out, dummy = module.run_command(openssl_keydump_cmd, check_rc=True)
|
||||
|
||||
if account_key_type == 'rsa':
|
||||
pub_hex, pub_exp = re.search(
|
||||
r"modulus:\n\s+00:([a-f0-9\:\s]+?)\npublicExponent: ([0-9]+)",
|
||||
to_text(out, errors='surrogate_or_strict'), re.MULTILINE | re.DOTALL).groups()
|
||||
pub_exp = "{0:x}".format(int(pub_exp))
|
||||
if len(pub_exp) % 2:
|
||||
pub_exp = "0{0}".format(pub_exp)
|
||||
|
||||
return None, {
|
||||
'key_file': key_file,
|
||||
'type': 'rsa',
|
||||
'alg': 'RS256',
|
||||
'jwk': {
|
||||
"kty": "RSA",
|
||||
"e": nopad_b64(binascii.unhexlify(pub_exp.encode("utf-8"))),
|
||||
"n": nopad_b64(binascii.unhexlify(re.sub(r"(\s|:)", "", pub_hex).encode("utf-8"))),
|
||||
},
|
||||
'hash': 'sha256',
|
||||
}
|
||||
elif account_key_type == 'ec':
|
||||
pub_data = re.search(
|
||||
r"pub:\s*\n\s+04:([a-f0-9\:\s]+?)\nASN1 OID: (\S+)(?:\nNIST CURVE: (\S+))?",
|
||||
to_text(out, errors='surrogate_or_strict'), re.MULTILINE | re.DOTALL)
|
||||
if pub_data is None:
|
||||
return 'cannot parse elliptic curve key', {}
|
||||
pub_hex = binascii.unhexlify(re.sub(r"(\s|:)", "", pub_data.group(1)).encode("utf-8"))
|
||||
asn1_oid_curve = pub_data.group(2).lower()
|
||||
nist_curve = pub_data.group(3).lower() if pub_data.group(3) else None
|
||||
if asn1_oid_curve == 'prime256v1' or nist_curve == 'p-256':
|
||||
bits = 256
|
||||
alg = 'ES256'
|
||||
hash = 'sha256'
|
||||
point_size = 32
|
||||
curve = 'P-256'
|
||||
elif asn1_oid_curve == 'secp384r1' or nist_curve == 'p-384':
|
||||
bits = 384
|
||||
alg = 'ES384'
|
||||
hash = 'sha384'
|
||||
point_size = 48
|
||||
curve = 'P-384'
|
||||
elif asn1_oid_curve == 'secp521r1' or nist_curve == 'p-521':
|
||||
# Not yet supported on Let's Encrypt side, see
|
||||
# https://github.com/letsencrypt/boulder/issues/2217
|
||||
bits = 521
|
||||
alg = 'ES512'
|
||||
hash = 'sha512'
|
||||
point_size = 66
|
||||
curve = 'P-521'
|
||||
else:
|
||||
return 'unknown elliptic curve: %s / %s' % (asn1_oid_curve, nist_curve), {}
|
||||
bytes = (bits + 7) // 8
|
||||
if len(pub_hex) != 2 * bytes:
|
||||
return 'bad elliptic curve point (%s / %s)' % (asn1_oid_curve, nist_curve), {}
|
||||
return None, {
|
||||
'key_file': key_file,
|
||||
'type': 'ec',
|
||||
'alg': alg,
|
||||
'jwk': {
|
||||
"kty": "EC",
|
||||
"crv": curve,
|
||||
"x": nopad_b64(pub_hex[:bytes]),
|
||||
"y": nopad_b64(pub_hex[bytes:]),
|
||||
},
|
||||
'hash': hash,
|
||||
'point_size': point_size,
|
||||
}
|
||||
|
||||
|
||||
def _sign_request_openssl(openssl_binary, module, payload64, protected64, key_data):
|
||||
openssl_sign_cmd = [openssl_binary, "dgst", "-{0}".format(key_data['hash']), "-sign", key_data['key_file']]
|
||||
sign_payload = "{0}.{1}".format(protected64, payload64).encode('utf8')
|
||||
dummy, out, dummy = module.run_command(openssl_sign_cmd, data=sign_payload, check_rc=True, binary_data=True)
|
||||
|
||||
if key_data['type'] == 'ec':
|
||||
dummy, der_out, dummy = module.run_command(
|
||||
[openssl_binary, "asn1parse", "-inform", "DER"],
|
||||
data=out, binary_data=True)
|
||||
expected_len = 2 * key_data['point_size']
|
||||
sig = re.findall(
|
||||
r"prim:\s+INTEGER\s+:([0-9A-F]{1,%s})\n" % expected_len,
|
||||
to_text(der_out, errors='surrogate_or_strict'))
|
||||
if len(sig) != 2:
|
||||
raise ModuleFailException(
|
||||
"failed to generate Elliptic Curve signature; cannot parse DER output: {0}".format(
|
||||
to_text(der_out, errors='surrogate_or_strict')))
|
||||
sig[0] = (expected_len - len(sig[0])) * '0' + sig[0]
|
||||
sig[1] = (expected_len - len(sig[1])) * '0' + sig[1]
|
||||
out = binascii.unhexlify(sig[0]) + binascii.unhexlify(sig[1])
|
||||
|
||||
return {
|
||||
"protected": protected64,
|
||||
"payload": payload64,
|
||||
"signature": nopad_b64(to_bytes(out)),
|
||||
}
|
||||
|
||||
|
||||
if sys.version_info[0] >= 3:
|
||||
# Python 3 (and newer)
|
||||
def _count_bytes(n):
|
||||
return (n.bit_length() + 7) // 8 if n > 0 else 0
|
||||
|
||||
def _convert_int_to_bytes(count, no):
|
||||
return no.to_bytes(count, byteorder='big')
|
||||
|
||||
def _pad_hex(n, digits):
|
||||
res = hex(n)[2:]
|
||||
if len(res) < digits:
|
||||
res = '0' * (digits - len(res)) + res
|
||||
return res
|
||||
else:
|
||||
# Python 2
|
||||
def _count_bytes(n):
|
||||
if n <= 0:
|
||||
return 0
|
||||
h = '%x' % n
|
||||
return (len(h) + 1) // 2
|
||||
|
||||
def _convert_int_to_bytes(count, n):
|
||||
h = '%x' % n
|
||||
if len(h) > 2 * count:
|
||||
raise Exception('Number {1} needs more than {0} bytes!'.format(count, n))
|
||||
return ('0' * (2 * count - len(h)) + h).decode('hex')
|
||||
|
||||
def _pad_hex(n, digits):
|
||||
h = '%x' % n
|
||||
if len(h) < digits:
|
||||
h = '0' * (digits - len(h)) + h
|
||||
return h
|
||||
|
||||
|
||||
def _parse_key_cryptography(module, key_file=None, key_content=None):
|
||||
'''
|
||||
Parses an RSA or Elliptic Curve key file in PEM format and returns a pair
|
||||
(error, key_data).
|
||||
'''
|
||||
# If key_content isn't given, read key_file
|
||||
if key_content is None:
|
||||
key_content = read_file(key_file)
|
||||
else:
|
||||
key_content = to_bytes(key_content)
|
||||
# Parse key
|
||||
try:
|
||||
key = cryptography.hazmat.primitives.serialization.load_pem_private_key(key_content, password=None, backend=_cryptography_backend)
|
||||
except Exception as e:
|
||||
return 'error while loading key: {0}'.format(e), None
|
||||
if isinstance(key, cryptography.hazmat.primitives.asymmetric.rsa.RSAPrivateKey):
|
||||
pk = key.public_key().public_numbers()
|
||||
return None, {
|
||||
'key_obj': key,
|
||||
'type': 'rsa',
|
||||
'alg': 'RS256',
|
||||
'jwk': {
|
||||
"kty": "RSA",
|
||||
"e": nopad_b64(_convert_int_to_bytes(_count_bytes(pk.e), pk.e)),
|
||||
"n": nopad_b64(_convert_int_to_bytes(_count_bytes(pk.n), pk.n)),
|
||||
},
|
||||
'hash': 'sha256',
|
||||
}
|
||||
elif isinstance(key, cryptography.hazmat.primitives.asymmetric.ec.EllipticCurvePrivateKey):
|
||||
pk = key.public_key().public_numbers()
|
||||
if pk.curve.name == 'secp256r1':
|
||||
bits = 256
|
||||
alg = 'ES256'
|
||||
hash = 'sha256'
|
||||
point_size = 32
|
||||
curve = 'P-256'
|
||||
elif pk.curve.name == 'secp384r1':
|
||||
bits = 384
|
||||
alg = 'ES384'
|
||||
hash = 'sha384'
|
||||
point_size = 48
|
||||
curve = 'P-384'
|
||||
elif pk.curve.name == 'secp521r1':
|
||||
# Not yet supported on Let's Encrypt side, see
|
||||
# https://github.com/letsencrypt/boulder/issues/2217
|
||||
bits = 521
|
||||
alg = 'ES512'
|
||||
hash = 'sha512'
|
||||
point_size = 66
|
||||
curve = 'P-521'
|
||||
else:
|
||||
return 'unknown elliptic curve: {0}'.format(pk.curve.name), {}
|
||||
bytes = (bits + 7) // 8
|
||||
return None, {
|
||||
'key_obj': key,
|
||||
'type': 'ec',
|
||||
'alg': alg,
|
||||
'jwk': {
|
||||
"kty": "EC",
|
||||
"crv": curve,
|
||||
"x": nopad_b64(_convert_int_to_bytes(bytes, pk.x)),
|
||||
"y": nopad_b64(_convert_int_to_bytes(bytes, pk.y)),
|
||||
},
|
||||
'hash': hash,
|
||||
'point_size': point_size,
|
||||
}
|
||||
else:
|
||||
return 'unknown key type "{0}"'.format(type(key)), {}
|
||||
|
||||
|
||||
def _sign_request_cryptography(module, payload64, protected64, key_data):
|
||||
sign_payload = "{0}.{1}".format(protected64, payload64).encode('utf8')
|
||||
if isinstance(key_data['key_obj'], cryptography.hazmat.primitives.asymmetric.rsa.RSAPrivateKey):
|
||||
padding = cryptography.hazmat.primitives.asymmetric.padding.PKCS1v15()
|
||||
hash = cryptography.hazmat.primitives.hashes.SHA256()
|
||||
signature = key_data['key_obj'].sign(sign_payload, padding, hash)
|
||||
elif isinstance(key_data['key_obj'], cryptography.hazmat.primitives.asymmetric.ec.EllipticCurvePrivateKey):
|
||||
if key_data['hash'] == 'sha256':
|
||||
hash = cryptography.hazmat.primitives.hashes.SHA256
|
||||
elif key_data['hash'] == 'sha384':
|
||||
hash = cryptography.hazmat.primitives.hashes.SHA384
|
||||
elif key_data['hash'] == 'sha512':
|
||||
hash = cryptography.hazmat.primitives.hashes.SHA512
|
||||
ecdsa = cryptography.hazmat.primitives.asymmetric.ec.ECDSA(hash())
|
||||
r, s = cryptography.hazmat.primitives.asymmetric.utils.decode_dss_signature(key_data['key_obj'].sign(sign_payload, ecdsa))
|
||||
rr = _pad_hex(r, 2 * key_data['point_size'])
|
||||
ss = _pad_hex(s, 2 * key_data['point_size'])
|
||||
signature = binascii.unhexlify(rr) + binascii.unhexlify(ss)
|
||||
|
||||
return {
|
||||
"protected": protected64,
|
||||
"payload": payload64,
|
||||
"signature": nopad_b64(signature),
|
||||
}
|
||||
|
||||
|
||||
class ACMEDirectory(object):
|
||||
'''
|
||||
The ACME server directory. Gives access to the available resources,
|
||||
|
|
@ -199,24 +518,8 @@ class ACMEAccount(object):
|
|||
|
||||
self._openssl_bin = module.get_bin_path('openssl', True)
|
||||
|
||||
# Create a key file from content, key (path) and key content are mutually exclusive
|
||||
if self.key_content is not None:
|
||||
fd, tmpsrc = tempfile.mkstemp()
|
||||
module.add_cleanup_file(tmpsrc) # Ansible will delete the file on exit
|
||||
f = os.fdopen(fd, 'wb')
|
||||
try:
|
||||
f.write(self.key_content.encode('utf-8'))
|
||||
self.key = tmpsrc
|
||||
except Exception as err:
|
||||
try:
|
||||
f.close()
|
||||
except Exception as e:
|
||||
pass
|
||||
raise ModuleFailException("failed to create temporary content file: %s" % to_native(err), exception=traceback.format_exc())
|
||||
f.close()
|
||||
|
||||
if self.key is not None:
|
||||
error, self.key_data = self.parse_account_key(self.key)
|
||||
if self.key is not None or self.key_content is not None:
|
||||
error, self.key_data = self.parse_key(self.key, self.key_content)
|
||||
if error:
|
||||
raise ModuleFailException("error while parsing account key: %s" % error)
|
||||
self.jwk = self.key_data['jwk']
|
||||
|
|
@ -234,136 +537,37 @@ class ACMEAccount(object):
|
|||
thumbprint = nopad_b64(hashlib.sha256(accountkey_json.encode('utf8')).digest())
|
||||
return "{0}.{1}".format(token, thumbprint)
|
||||
|
||||
def parse_account_key(self, key):
|
||||
def parse_key(self, key_file=None, key_content=None):
|
||||
'''
|
||||
Parses an RSA or Elliptic Curve key file in PEM format and returns a pair
|
||||
(error, key_data).
|
||||
'''
|
||||
account_key_type = None
|
||||
with open(key, "rt") as f:
|
||||
for line in f:
|
||||
m = re.match(r"^\s*-{5,}BEGIN\s+(EC|RSA)\s+PRIVATE\s+KEY-{5,}\s*$", line)
|
||||
if m is not None:
|
||||
account_key_type = m.group(1).lower()
|
||||
break
|
||||
if account_key_type is None:
|
||||
# This happens for example if openssl_privatekey created this key
|
||||
# (as opposed to the OpenSSL binary). For now, we assume this is
|
||||
# an RSA key.
|
||||
# FIXME: add some kind of auto-detection
|
||||
account_key_type = "rsa"
|
||||
if account_key_type not in ("rsa", "ec"):
|
||||
return 'unknown key type "%s"' % account_key_type, {}
|
||||
if key_file is None and key_content is None:
|
||||
raise AssertionError('One of key_file and key_content must be specified!')
|
||||
if HAS_CURRENT_CRYPTOGRAPHY:
|
||||
return _parse_key_cryptography(self.module, key_file, key_content)
|
||||
else:
|
||||
return _parse_key_openssl(self._openssl_bin, self.module, key_file, key_content)
|
||||
|
||||
openssl_keydump_cmd = [self._openssl_bin, account_key_type, "-in", key, "-noout", "-text"]
|
||||
dummy, out, dummy = self.module.run_command(openssl_keydump_cmd, check_rc=True)
|
||||
|
||||
if account_key_type == 'rsa':
|
||||
pub_hex, pub_exp = re.search(
|
||||
r"modulus:\n\s+00:([a-f0-9\:\s]+?)\npublicExponent: ([0-9]+)",
|
||||
to_text(out, errors='surrogate_or_strict'), re.MULTILINE | re.DOTALL).groups()
|
||||
pub_exp = "{0:x}".format(int(pub_exp))
|
||||
if len(pub_exp) % 2:
|
||||
pub_exp = "0{0}".format(pub_exp)
|
||||
|
||||
return None, {
|
||||
'type': 'rsa',
|
||||
'alg': 'RS256',
|
||||
'jwk': {
|
||||
"kty": "RSA",
|
||||
"e": nopad_b64(binascii.unhexlify(pub_exp.encode("utf-8"))),
|
||||
"n": nopad_b64(binascii.unhexlify(re.sub(r"(\s|:)", "", pub_hex).encode("utf-8"))),
|
||||
},
|
||||
'hash': 'sha256',
|
||||
}
|
||||
elif account_key_type == 'ec':
|
||||
pub_data = re.search(
|
||||
r"pub:\s*\n\s+04:([a-f0-9\:\s]+?)\nASN1 OID: (\S+)(?:\nNIST CURVE: (\S+))?",
|
||||
to_text(out, errors='surrogate_or_strict'), re.MULTILINE | re.DOTALL)
|
||||
if pub_data is None:
|
||||
return 'cannot parse elliptic curve key', {}
|
||||
pub_hex = binascii.unhexlify(re.sub(r"(\s|:)", "", pub_data.group(1)).encode("utf-8"))
|
||||
asn1_oid_curve = pub_data.group(2).lower()
|
||||
nist_curve = pub_data.group(3).lower() if pub_data.group(3) else None
|
||||
if asn1_oid_curve == 'prime256v1' or nist_curve == 'p-256':
|
||||
bits = 256
|
||||
alg = 'ES256'
|
||||
hash = 'sha256'
|
||||
point_size = 32
|
||||
curve = 'P-256'
|
||||
elif asn1_oid_curve == 'secp384r1' or nist_curve == 'p-384':
|
||||
bits = 384
|
||||
alg = 'ES384'
|
||||
hash = 'sha384'
|
||||
point_size = 48
|
||||
curve = 'P-384'
|
||||
elif asn1_oid_curve == 'secp521r1' or nist_curve == 'p-521':
|
||||
# Not yet supported on Let's Encrypt side, see
|
||||
# https://github.com/letsencrypt/boulder/issues/2217
|
||||
bits = 521
|
||||
alg = 'ES512'
|
||||
hash = 'sha512'
|
||||
point_size = 66
|
||||
curve = 'P-521'
|
||||
else:
|
||||
return 'unknown elliptic curve: %s / %s' % (asn1_oid_curve, nist_curve), {}
|
||||
bytes = (bits + 7) // 8
|
||||
if len(pub_hex) != 2 * bytes:
|
||||
return 'bad elliptic curve point (%s / %s)' % (asn1_oid_curve, nist_curve), {}
|
||||
return None, {
|
||||
'type': 'ec',
|
||||
'alg': alg,
|
||||
'jwk': {
|
||||
"kty": "EC",
|
||||
"crv": curve,
|
||||
"x": nopad_b64(pub_hex[:bytes]),
|
||||
"y": nopad_b64(pub_hex[bytes:]),
|
||||
},
|
||||
'hash': hash,
|
||||
'point_size': point_size,
|
||||
}
|
||||
|
||||
def sign_request(self, protected, payload, key_data, key):
|
||||
def sign_request(self, protected, payload, key_data):
|
||||
try:
|
||||
payload64 = nopad_b64(self.module.jsonify(payload).encode('utf8'))
|
||||
protected64 = nopad_b64(self.module.jsonify(protected).encode('utf8'))
|
||||
except Exception as e:
|
||||
raise ModuleFailException("Failed to encode payload / headers as JSON: {0}".format(e))
|
||||
|
||||
openssl_sign_cmd = [self._openssl_bin, "dgst", "-{0}".format(key_data['hash']), "-sign", key]
|
||||
sign_payload = "{0}.{1}".format(protected64, payload64).encode('utf8')
|
||||
dummy, out, dummy = self.module.run_command(openssl_sign_cmd, data=sign_payload, check_rc=True, binary_data=True)
|
||||
if HAS_CURRENT_CRYPTOGRAPHY:
|
||||
return _sign_request_cryptography(self.module, payload64, protected64, key_data)
|
||||
else:
|
||||
return _sign_request_openssl(self._openssl_bin, self.module, payload64, protected64, key_data)
|
||||
|
||||
if key_data['type'] == 'ec':
|
||||
dummy, der_out, dummy = self.module.run_command(
|
||||
[self._openssl_bin, "asn1parse", "-inform", "DER"],
|
||||
data=out, binary_data=True)
|
||||
expected_len = 2 * key_data['point_size']
|
||||
sig = re.findall(
|
||||
r"prim:\s+INTEGER\s+:([0-9A-F]{1,%s})\n" % expected_len,
|
||||
to_text(der_out, errors='surrogate_or_strict'))
|
||||
if len(sig) != 2:
|
||||
raise ModuleFailException(
|
||||
"failed to generate Elliptic Curve signature; cannot parse DER output: {0}".format(
|
||||
to_text(der_out, errors='surrogate_or_strict')))
|
||||
sig[0] = (expected_len - len(sig[0])) * '0' + sig[0]
|
||||
sig[1] = (expected_len - len(sig[1])) * '0' + sig[1]
|
||||
out = binascii.unhexlify(sig[0]) + binascii.unhexlify(sig[1])
|
||||
|
||||
return {
|
||||
"protected": protected64,
|
||||
"payload": payload64,
|
||||
"signature": nopad_b64(to_bytes(out)),
|
||||
}
|
||||
|
||||
def send_signed_request(self, url, payload, key_data=None, key=None, jws_header=None):
|
||||
def send_signed_request(self, url, payload, key_data=None, jws_header=None):
|
||||
'''
|
||||
Sends a JWS signed HTTP POST request to the ACME server and returns
|
||||
the response as dictionary
|
||||
https://tools.ietf.org/html/draft-ietf-acme-acme-12#section-6.2
|
||||
'''
|
||||
key_data = key_data or self.key_data
|
||||
key = key or self.key
|
||||
jws_header = jws_header or self.jws_header
|
||||
failed_tries = 0
|
||||
while True:
|
||||
|
|
@ -372,7 +576,7 @@ class ACMEAccount(object):
|
|||
if self.version != 1:
|
||||
protected["url"] = url
|
||||
|
||||
data = self.sign_request(protected, payload, key_data, key)
|
||||
data = self.sign_request(protected, payload, key_data)
|
||||
if self.version == 1:
|
||||
data["header"] = jws_header
|
||||
data = self.module.jsonify(data)
|
||||
|
|
@ -530,3 +734,67 @@ class ACMEAccount(object):
|
|||
result, dummy = self.send_signed_request(self.uri, upd_reg)
|
||||
changed = True
|
||||
return new_account or changed
|
||||
|
||||
|
||||
def cryptography_get_csr_domains(module, csr_filename):
|
||||
'''
|
||||
Return a set of requested domains (CN and SANs) for the CSR.
|
||||
'''
|
||||
domains = set([])
|
||||
csr = cryptography.x509.load_pem_x509_csr(read_file(csr_filename), _cryptography_backend)
|
||||
for sub in csr.subject:
|
||||
if sub.oid == cryptography.x509.oid.NameOID.COMMON_NAME:
|
||||
domains.add(sub.value)
|
||||
for extension in csr.extensions:
|
||||
if extension.oid == cryptography.x509.oid.ExtensionOID.SUBJECT_ALTERNATIVE_NAME:
|
||||
for name in extension.value:
|
||||
if isinstance(name, cryptography.x509.DNSName):
|
||||
domains.add(name.value)
|
||||
return domains
|
||||
|
||||
|
||||
def cryptography_get_cert_days(module, cert_file):
|
||||
'''
|
||||
Return the days the certificate in cert_file remains valid and -1
|
||||
if the file was not found. If cert_file contains more than one
|
||||
certificate, only the first one will be considered.
|
||||
'''
|
||||
if not os.path.exists(cert_file):
|
||||
return -1
|
||||
|
||||
try:
|
||||
cert = cryptography.x509.load_pem_x509_certificate(read_file(cert_file), _cryptography_backend)
|
||||
except Exception as e:
|
||||
raise ModuleFailException('Cannot parse certificate {0}: {1}'.format(cert_file, e))
|
||||
now = datetime.datetime.now()
|
||||
return (cert.not_valid_after - now).days
|
||||
|
||||
|
||||
def set_crypto_backend(module):
|
||||
'''
|
||||
Sets which crypto backend to use (default: auto detection).
|
||||
|
||||
Does not care whether a new enough cryptoraphy is available or not. Must
|
||||
be called before any real stuff is done which might evaluate
|
||||
``HAS_CURRENT_CRYPTOGRAPHY``.
|
||||
'''
|
||||
global HAS_CURRENT_CRYPTOGRAPHY
|
||||
# Choose backend
|
||||
backend = module.params['select_crypto_backend']
|
||||
if backend == 'auto':
|
||||
pass
|
||||
elif backend == 'openssl':
|
||||
HAS_CURRENT_CRYPTOGRAPHY = False
|
||||
elif backend == 'cryptography':
|
||||
try:
|
||||
cryptography.__version__
|
||||
except Exception as _:
|
||||
module.fail_json(msg='Cannot find cryptography module!')
|
||||
HAS_CURRENT_CRYPTOGRAPHY = True
|
||||
else:
|
||||
module.fail_json(msg='Unknown crypto backend "{0}"!'.format(backend))
|
||||
# Inform about choices
|
||||
if HAS_CURRENT_CRYPTOGRAPHY:
|
||||
module.debug('Using cryptography backend (library version {0})'.format(CRYPTOGRAPHY_VERSION))
|
||||
else:
|
||||
module.debug('Using OpenSSL binary backend')
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue