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

Revert "Revert all non-bugfixes merged since the last release."

Browse Source 
This reverts commit 82251c2d80.
This commit is contained in:
Felix Fontein
2024-05-11 17:05:03 +02:00
parent 3d8c68e189
commit 00d23753ca
66 changed files with 2908 additions and 299 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
plugins/module_utils/acme/acme.py
View File

@@ -21,6 +21,8 @@ from ansible.module_utils.common.text.converters import to_bytes
from ansible.module_utils.urls import fetch_url
from ansible.module_utils.six import PY3
from ansible_collections.community.crypto.plugins.module_utils.argspec import ArgumentSpec
from ansible_collections.community.crypto.plugins.module_utils.acme.backend_openssl_cli import (
OpenSSLCLIBackend,
)
@@ -42,7 +44,9 @@ from ansible_collections.community.crypto.plugins.module_utils.acme.errors impor
)
from ansible_collections.community.crypto.plugins.module_utils.acme.utils import (
compute_cert_id,
nopad_b64,
parse_retry_after,
)
try:
@@ -153,6 +157,9 @@ class ACMEDirectory(object):
self.module, msg='Was not able to obtain nonce, giving up after 5 retries', info=info, response=response)
retry_count += 1
def has_renewal_info_endpoint(self):
return 'renewalInfo' in self.directory
class ACMEClient(object):
'''
@@ -168,9 +175,9 @@ class ACMEClient(object):
self.backend = backend
self.version = module.params['acme_version']
# account_key path and content are mutually exclusive
self.account_key_file = module.params['account_key_src']
self.account_key_content = module.params['account_key_content']
self.account_key_passphrase = module.params['account_key_passphrase']
self.account_key_file = module.params.get('account_key_src')
self.account_key_content = module.params.get('account_key_content')
self.account_key_passphrase = module.params.get('account_key_passphrase')
# Grab account URI from module parameters.
# Make sure empty string is treated as None.
@@ -383,24 +390,94 @@ class ACMEClient(object):
self.module, msg=error_msg, info=info, content=content, content_json=result if parsed_json_result else None)
return result, info
def get_renewal_info(
self,
cert_id=None,
cert_info=None,
cert_filename=None,
cert_content=None,
include_retry_after=False,
retry_after_relative_with_timezone=True,
):
if not self.directory.has_renewal_info_endpoint():
raise ModuleFailException('The ACME endpoint does not support ACME Renewal Information retrieval')
if cert_id is None:
cert_id = compute_cert_id(self.backend, cert_info=cert_info, cert_filename=cert_filename, cert_content=cert_content)
url = '{base}{cert_id}'.format(base=self.directory.directory['renewalInfo'], cert_id=cert_id)
data, info = self.get_request(url, parse_json_result=True, fail_on_error=True, get_only=True)
# Include Retry-After header if asked for
if include_retry_after and 'retry-after' in info:
try:
data['retryAfter'] = parse_retry_after(
info['retry-after'],
relative_with_timezone=retry_after_relative_with_timezone,
)
except ValueError:
pass
return data
def get_default_argspec():
'''
Provides default argument spec for the options documented in the acme doc fragment.
DEPRECATED: will be removed in community.crypto 3.0.0
'''
return dict(
account_key_src=dict(type='path', aliases=['account_key']),
account_key_content=dict(type='str', no_log=True),
account_key_passphrase=dict(type='str', no_log=True),
account_uri=dict(type='str'),
acme_directory=dict(type='str', required=True),
acme_version=dict(type='int', required=True, choices=[1, 2]),
validate_certs=dict(type='bool', default=True),
select_crypto_backend=dict(type='str', default='auto', choices=['auto', 'openssl', 'cryptography']),
request_timeout=dict(type='int', default=10),
account_key_src=dict(type='path', aliases=['account_key']),
account_key_content=dict(type='str', no_log=True),
account_key_passphrase=dict(type='str', no_log=True),
account_uri=dict(type='str'),
)
def create_default_argspec(
with_account=True,
require_account_key=True,
with_certificate=False,
):
'''
Provides default argument spec for the options documented in the acme doc fragment.
'''
result = ArgumentSpec(
argument_spec=dict(
acme_directory=dict(type='str', required=True),
acme_version=dict(type='int', required=True, choices=[1, 2]),
validate_certs=dict(type='bool', default=True),
select_crypto_backend=dict(type='str', default='auto', choices=['auto', 'openssl', 'cryptography']),
request_timeout=dict(type='int', default=10),
),
)
if with_account:
result.update_argspec(
account_key_src=dict(type='path', aliases=['account_key']),
account_key_content=dict(type='str', no_log=True),
account_key_passphrase=dict(type='str', no_log=True),
account_uri=dict(type='str'),
)
if require_account_key:
result.update(required_one_of=[['account_key_src', 'account_key_content']])
result.update(mutually_exclusive=[['account_key_src', 'account_key_content']])
if with_certificate:
result.update_argspec(
csr=dict(type='path'),
csr_content=dict(type='str'),
)
result.update(
required_one_of=[['csr', 'csr_content']],
mutually_exclusive=[['csr', 'csr_content']],
)
return result
def create_backend(module, needs_acme_v2):
if not HAS_IPADDRESS:
module.fail_json(msg=missing_required_lib('ipaddress'), exception=IPADDRESS_IMPORT_ERROR)

97
plugins/module_utils/acme/backend_cryptography.py
View File

@@ -11,6 +11,7 @@ __metaclass__ = type
import base64
import binascii
import datetime
import os
import traceback
@@ -19,7 +20,9 @@ from ansible.module_utils.common.text.converters import to_bytes, to_native, to_
from ansible_collections.community.crypto.plugins.module_utils.version import LooseVersion
from ansible_collections.community.crypto.plugins.module_utils.acme.backends import (
CertificateInformation,
CryptoBackend,
_parse_acme_timestamp,
)
from ansible_collections.community.crypto.plugins.module_utils.acme.certificates import (
@@ -35,27 +38,40 @@ from ansible_collections.community.crypto.plugins.module_utils.acme.io import re
from ansible_collections.community.crypto.plugins.module_utils.acme.utils import nopad_b64
from ansible_collections.community.crypto.plugins.module_utils.crypto.basic import (
OpenSSLObjectError,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.math import (
convert_int_to_bytes,
convert_int_to_hex,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.support import (
get_now_datetime,
ensure_utc_timezone,
parse_name_field,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import (
CRYPTOGRAPHY_TIMEZONE,
cryptography_name_to_oid,
cryptography_serial_number_of_cert,
get_not_valid_after,
get_not_valid_before,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.pem import (
extract_first_pem,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.support import (
parse_name_field,
)
from ansible_collections.community.crypto.plugins.module_utils.time import (
ensure_utc_timezone,
from_epoch_seconds,
get_epoch_seconds,
get_now_datetime,
get_relative_time_option,
UTC,
)
CRYPTOGRAPHY_MINIMAL_VERSION = '1.5'
CRYPTOGRAPHY_ERROR = None
@@ -170,6 +186,32 @@ class CryptographyBackend(CryptoBackend):
def __init__(self, module):
super(CryptographyBackend, self).__init__(module)
def get_now(self):
return get_now_datetime(with_timezone=CRYPTOGRAPHY_TIMEZONE)
def parse_acme_timestamp(self, timestamp_str):
return _parse_acme_timestamp(timestamp_str, with_timezone=CRYPTOGRAPHY_TIMEZONE)
def parse_module_parameter(self, value, name):
try:
return get_relative_time_option(value, name, backend='cryptography', with_timezone=CRYPTOGRAPHY_TIMEZONE)
except OpenSSLObjectError as exc:
raise BackendException(to_native(exc))
def interpolate_timestamp(self, timestamp_start, timestamp_end, percentage):
start = get_epoch_seconds(timestamp_start)
end = get_epoch_seconds(timestamp_end)
return from_epoch_seconds(start + percentage * (end - start), with_timezone=CRYPTOGRAPHY_TIMEZONE)
def get_utc_datetime(self, *args, **kwargs):
kwargs_ext = dict(kwargs)
if CRYPTOGRAPHY_TIMEZONE and ('tzinfo' not in kwargs_ext and len(args) < 8):
kwargs_ext['tzinfo'] = UTC
result = datetime.datetime(*args, **kwargs_ext)
if CRYPTOGRAPHY_TIMEZONE and ('tzinfo' in kwargs or len(args) >= 8):
result = ensure_utc_timezone(result)
return result
def parse_key(self, key_file=None, key_content=None, passphrase=None):
'''
Parses an RSA or Elliptic Curve key file in PEM format and returns key_data.
@@ -376,7 +418,7 @@ class CryptographyBackend(CryptoBackend):
raise BackendException('Cannot parse certificate {0}: {1}'.format(cert_filename, e))
if now is None:
now = get_now_datetime(with_timezone=CRYPTOGRAPHY_TIMEZONE)
now = self.get_now()
elif CRYPTOGRAPHY_TIMEZONE:
now = ensure_utc_timezone(now)
return (get_not_valid_after(cert) - now).days
@@ -386,3 +428,44 @@ class CryptographyBackend(CryptoBackend):
Given a Criterium object, creates a ChainMatcher object.
'''
return CryptographyChainMatcher(criterium, self.module)
def get_cert_information(self, cert_filename=None, cert_content=None):
'''
Return some information on a X.509 certificate as a CertificateInformation object.
'''
if cert_filename is not None:
cert_content = read_file(cert_filename)
else:
cert_content = to_bytes(cert_content)
# Make sure we have at most one PEM. Otherwise cryptography 36.0.0 will barf.
cert_content = to_bytes(extract_first_pem(to_text(cert_content)) or '')
try:
cert = cryptography.x509.load_pem_x509_certificate(cert_content, _cryptography_backend)
except Exception as e:
if cert_filename is None:
raise BackendException('Cannot parse certificate: {0}'.format(e))
raise BackendException('Cannot parse certificate {0}: {1}'.format(cert_filename, e))
ski = None
try:
ext = cert.extensions.get_extension_for_class(cryptography.x509.SubjectKeyIdentifier)
ski = ext.value.digest
except cryptography.x509.ExtensionNotFound:
pass
aki = None
try:
ext = cert.extensions.get_extension_for_class(cryptography.x509.AuthorityKeyIdentifier)
aki = ext.value.key_identifier
except cryptography.x509.ExtensionNotFound:
pass
return CertificateInformation(
not_valid_after=get_not_valid_after(cert),
not_valid_before=get_not_valid_before(cert),
serial_number=cryptography_serial_number_of_cert(cert),
subject_key_identifier=ski,
authority_key_identifier=aki,
)

95
plugins/module_utils/acme/backend_openssl_cli.py
View File

@@ -20,6 +20,7 @@ import traceback
from ansible.module_utils.common.text.converters import to_native, to_text, to_bytes
from ansible_collections.community.crypto.plugins.module_utils.acme.backends import (
CertificateInformation,
CryptoBackend,
)
@@ -30,6 +31,8 @@ from ansible_collections.community.crypto.plugins.module_utils.acme.errors impor
from ansible_collections.community.crypto.plugins.module_utils.acme.utils import nopad_b64
from ansible_collections.community.crypto.plugins.module_utils.crypto.math import convert_bytes_to_int
try:
import ipaddress
except ImportError:
@@ -39,6 +42,33 @@ except ImportError:
_OPENSSL_ENVIRONMENT_UPDATE = dict(LANG='C', LC_ALL='C', LC_MESSAGES='C', LC_CTYPE='C')
def _extract_date(out_text, name, cert_filename_suffix=""):
try:
date_str = re.search(r"\s+%s\s*:\s+(.*)" % name, out_text).group(1)
return datetime.datetime.strptime(date_str, '%b %d %H:%M:%S %Y %Z')
except AttributeError:
raise BackendException("No '{0}' date found{1}".format(name, cert_filename_suffix))
except ValueError as exc:
raise BackendException("Failed to parse '{0}' date{1}: {2}".format(name, cert_filename_suffix, exc))
def _decode_octets(octets_text):
return binascii.unhexlify(re.sub(r"(\s|:)", "", octets_text).encode("utf-8"))
def _extract_octets(out_text, name, required=True, potential_prefixes=None):
regexp = r"\s+%s:\s*\n\s+%s([A-Fa-f0-9]{2}(?::[A-Fa-f0-9]{2})*)\s*\n" % (
name,
('(?:%s)' % '|'.join(re.escape(pp) for pp in potential_prefixes)) if potential_prefixes else '',
)
match = re.search(regexp, out_text, re.MULTILINE | re.DOTALL)
if match is not None:
return _decode_octets(match.group(1))
if not required:
return None
raise BackendException("No '{0}' octet string found".format(name))
class OpenSSLCLIBackend(CryptoBackend):
def __init__(self, module, openssl_binary=None):
super(OpenSSLCLIBackend, self).__init__(module)
@@ -89,10 +119,12 @@ class OpenSSLCLIBackend(CryptoBackend):
dummy, out, dummy = self.module.run_command(
openssl_keydump_cmd, check_rc=True, environ_update=_OPENSSL_ENVIRONMENT_UPDATE)
out_text = to_text(out, errors='surrogate_or_strict')
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_hex = re.search(r"modulus:\n\s+00:([a-f0-9\:\s]+?)\npublicExponent", out_text, re.MULTILINE | re.DOTALL).group(1)
pub_exp = re.search(r"\npublicExponent: ([0-9]+)", out_text, re.MULTILINE | re.DOTALL).group(1)
pub_exp = "{0:x}".format(int(pub_exp))
if len(pub_exp) % 2:
pub_exp = "0{0}".format(pub_exp)
@@ -104,17 +136,19 @@ class OpenSSLCLIBackend(CryptoBackend):
'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"))),
"n": nopad_b64(_decode_octets(pub_hex)),
},
'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)
out_text,
re.MULTILINE | re.DOTALL,
)
if pub_data is None:
raise KeyParsingError('cannot parse elliptic curve key')
pub_hex = binascii.unhexlify(re.sub(r"(\s|:)", "", pub_data.group(1)).encode("utf-8"))
pub_hex = _decode_octets(pub_data.group(1))
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':
@@ -303,13 +337,8 @@ class OpenSSLCLIBackend(CryptoBackend):
openssl_cert_cmd = [self.openssl_binary, "x509", "-in", filename, "-noout", "-text"]
dummy, out, dummy = self.module.run_command(
openssl_cert_cmd, data=data, check_rc=True, binary_data=True, environ_update=_OPENSSL_ENVIRONMENT_UPDATE)
try:
not_after_str = re.search(r"\s+Not After\s*:\s+(.*)", to_text(out, errors='surrogate_or_strict')).group(1)
not_after = datetime.datetime.strptime(not_after_str, '%b %d %H:%M:%S %Y %Z')
except AttributeError:
raise BackendException("No 'Not after' date found{0}".format(cert_filename_suffix))
except ValueError:
raise BackendException("Failed to parse 'Not after' date{0}".format(cert_filename_suffix))
out_text = to_text(out, errors='surrogate_or_strict')
not_after = _extract_date(out_text, 'Not After', cert_filename_suffix=cert_filename_suffix)
if now is None:
now = datetime.datetime.now()
return (not_after - now).days
@@ -319,3 +348,43 @@ class OpenSSLCLIBackend(CryptoBackend):
Given a Criterium object, creates a ChainMatcher object.
'''
raise BackendException('Alternate chain matching can only be used with the "cryptography" backend.')
def get_cert_information(self, cert_filename=None, cert_content=None):
'''
Return some information on a X.509 certificate as a CertificateInformation object.
'''
filename = cert_filename
data = None
if cert_filename is not None:
cert_filename_suffix = ' in {0}'.format(cert_filename)
else:
filename = '/dev/stdin'
data = to_bytes(cert_content)
cert_filename_suffix = ''
openssl_cert_cmd = [self.openssl_binary, "x509", "-in", filename, "-noout", "-text"]
dummy, out, dummy = self.module.run_command(
openssl_cert_cmd, data=data, check_rc=True, binary_data=True, environ_update=_OPENSSL_ENVIRONMENT_UPDATE)
out_text = to_text(out, errors='surrogate_or_strict')
not_after = _extract_date(out_text, 'Not After', cert_filename_suffix=cert_filename_suffix)
not_before = _extract_date(out_text, 'Not Before', cert_filename_suffix=cert_filename_suffix)
sn = re.search(
r" Serial Number: ([0-9]+)",
to_text(out, errors='surrogate_or_strict'), re.MULTILINE | re.DOTALL)
if sn:
serial = int(sn.group(1))
else:
serial = convert_bytes_to_int(_extract_octets(out_text, 'Serial Number', required=True))
ski = _extract_octets(out_text, 'X509v3 Subject Key Identifier', required=False)
aki = _extract_octets(out_text, 'X509v3 Authority Key Identifier', required=False, potential_prefixes=['keyid:', ''])
return CertificateInformation(
not_valid_after=not_after,
not_valid_before=not_before,
serial_number=serial,
subject_key_identifier=ski,
authority_key_identifier=aki,
)

102
plugins/module_utils/acme/backends.py
View File

@@ -9,9 +9,78 @@ from __future__ import absolute_import, division, print_function
__metaclass__ = type
from collections import namedtuple
import abc
import datetime
import re
from ansible.module_utils import six
from ansible.module_utils.common.text.converters import to_native
from ansible_collections.community.crypto.plugins.module_utils.acme.errors import (
BackendException,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.basic import (
OpenSSLObjectError,
)
from ansible_collections.community.crypto.plugins.module_utils.time import (
ensure_utc_timezone,
from_epoch_seconds,
get_epoch_seconds,
get_now_datetime,
get_relative_time_option,
remove_timezone,
)
CertificateInformation = namedtuple(
'CertificateInformation',
(
'not_valid_after',
'not_valid_before',
'serial_number',
'subject_key_identifier',
'authority_key_identifier',
),
)
_FRACTIONAL_MATCHER = re.compile(r'^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(|\.\d+)(Z|[+-]\d{2}:?\d{2}.*)$')
def _reduce_fractional_digits(timestamp_str):
"""
Given a RFC 3339 timestamp that includes too many digits for the fractional seconds part, reduces these to at most 6.
"""
# RFC 3339 (https://www.rfc-editor.org/info/rfc3339)
m = _FRACTIONAL_MATCHER.match(timestamp_str)
if not m:
raise BackendException('Cannot parse ISO 8601 timestamp {0!r}'.format(timestamp_str))
timestamp, fractional, timezone = m.groups()
if len(fractional) > 7:
# Python does not support anything smaller than microseconds
# (Golang supports nanoseconds, Boulder often emits more fractional digits, which Python chokes on)
fractional = fractional[:7]
return '%s%s%s' % (timestamp, fractional, timezone)
def _parse_acme_timestamp(timestamp_str, with_timezone):
"""
Parses a RFC 3339 timestamp.
"""
# RFC 3339 (https://www.rfc-editor.org/info/rfc3339)
timestamp_str = _reduce_fractional_digits(timestamp_str)
for format in ('%Y-%m-%dT%H:%M:%SZ', '%Y-%m-%dT%H:%M:%S.%fZ', '%Y-%m-%dT%H:%M:%S%z', '%Y-%m-%dT%H:%M:%S.%f%z'):
# Note that %z won't work with Python 2... https://stackoverflow.com/a/27829491
try:
result = datetime.datetime.strptime(timestamp_str, format)
except ValueError:
pass
else:
return ensure_utc_timezone(result) if with_timezone else remove_timezone(result)
raise BackendException('Cannot parse ISO 8601 timestamp {0!r}'.format(timestamp_str))
@six.add_metaclass(abc.ABCMeta)
@@ -19,6 +88,30 @@ class CryptoBackend(object):
def __init__(self, module):
self.module = module
def get_now(self):
return get_now_datetime(with_timezone=False)
def parse_acme_timestamp(self, timestamp_str):
# RFC 3339 (https://www.rfc-editor.org/info/rfc3339)
return _parse_acme_timestamp(timestamp_str, with_timezone=False)
def parse_module_parameter(self, value, name):
try:
return get_relative_time_option(value, name, backend='cryptography', with_timezone=False)
except OpenSSLObjectError as exc:
raise BackendException(to_native(exc))
def interpolate_timestamp(self, timestamp_start, timestamp_end, percentage):
start = get_epoch_seconds(timestamp_start)
end = get_epoch_seconds(timestamp_end)
return from_epoch_seconds(start + percentage * (end - start), with_timezone=False)
def get_utc_datetime(self, *args, **kwargs):
result = datetime.datetime(*args, **kwargs)
if 'tzinfo' in kwargs or len(args) >= 8:
result = remove_timezone(result)
return result
@abc.abstractmethod
def parse_key(self, key_file=None, key_content=None, passphrase=None):
'''
@@ -74,3 +167,12 @@ class CryptoBackend(object):
'''
Given a Criterium object, creates a ChainMatcher object.
'''
def get_cert_information(self, cert_filename=None, cert_content=None):
'''
Return some information on a X.509 certificate as a CertificateInformation object.
'''
# Not implementing this method in a backend is DEPRECATED and will be
# disallowed in community.crypto 3.0.0. This method will be marked as
# @abstractmethod by then.
raise BackendException('This backend does not support get_cert_information()')

12
plugins/module_utils/acme/challenges.py
View File

@@ -103,7 +103,7 @@ class Challenge(object):
# https://tools.ietf.org/html/rfc8555#section-8.4
resource = '_acme-challenge'
value = nopad_b64(hashlib.sha256(to_bytes(key_authorization)).digest())
record = (resource + identifier[1:]) if identifier.startswith('*.') else '{0}.{1}'.format(resource, identifier)
record = '{0}.{1}'.format(resource, identifier[2:] if identifier.startswith('*.') else identifier)
return {
'resource': resource,
'resource_value': value,
@@ -283,13 +283,21 @@ class Authorization(object):
return self.status == 'valid'
return self.wait_for_validation(client, challenge_type)
def can_deactivate(self):
'''
Deactivates this authorization.
https://community.letsencrypt.org/t/authorization-deactivation/19860/2
https://tools.ietf.org/html/rfc8555#section-7.5.2
'''
return self.status in ('valid', 'pending')
def deactivate(self, client):
'''
Deactivates this authorization.
https://community.letsencrypt.org/t/authorization-deactivation/19860/2
https://tools.ietf.org/html/rfc8555#section-7.5.2
'''
if self.status != 'valid':
if not self.can_deactivate():
return
authz_deactivate = {
'status': 'deactivated'

6
plugins/module_utils/acme/orders.py
View File

@@ -32,6 +32,7 @@ class Order(object):
self.identifiers = []
for identifier in data['identifiers']:
self.identifiers.append((identifier['type'], identifier['value']))
self.replaces_cert_id = data.get('replaces')
self.finalize_uri = data.get('finalize')
self.certificate_uri = data.get('certificate')
self.authorization_uris = data['authorizations']
@@ -44,6 +45,7 @@ class Order(object):
self.status = None
self.identifiers = []
self.replaces_cert_id = None
self.finalize_uri = None
self.certificate_uri = None
self.authorization_uris = []
@@ -62,7 +64,7 @@ class Order(object):
return result
@classmethod
def create(cls, client, identifiers):
def create(cls, client, identifiers, replaces_cert_id=None):
'''
Start a new certificate order (ACME v2 protocol).
https://tools.ietf.org/html/rfc8555#section-7.4
@@ -76,6 +78,8 @@ class Order(object):
new_order = {
"identifiers": acme_identifiers
}
if replaces_cert_id is not None:
new_order["replaces"] = replaces_cert_id
result, info = client.send_signed_request(
client.directory['newOrder'], new_order, error_msg='Failed to start new order', expected_status_codes=[201])
return cls.from_json(client, result, info['location'])

58
plugins/module_utils/acme/utils.py
View File

@@ -10,6 +10,7 @@ __metaclass__ = type
import base64
import datetime
import re
import textwrap
import traceback
@@ -19,6 +20,10 @@ from ansible.module_utils.six.moves.urllib.parse import unquote
from ansible_collections.community.crypto.plugins.module_utils.acme.errors import ModuleFailException
from ansible_collections.community.crypto.plugins.module_utils.crypto.math import convert_int_to_bytes
from ansible_collections.community.crypto.plugins.module_utils.time import get_now_datetime
def nopad_b64(data):
return base64.urlsafe_b64encode(data).decode('utf8').replace("=", "")
@@ -65,8 +70,61 @@ def pem_to_der(pem_filename=None, pem_content=None):
def process_links(info, callback):
'''
Process link header, calls callback for every link header with the URL and relation as options.
https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Link
'''
if 'link' in info:
link = info['link']
for url, relation in re.findall(r'<([^>]+)>;\s*rel="(\w+)"', link):
callback(unquote(url), relation)
def parse_retry_after(value, relative_with_timezone=True, now=None):
'''
Parse the value of a Retry-After header and return a timestamp.
https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Retry-After
'''
# First try a number of seconds
try:
delta = datetime.timedelta(seconds=int(value))
if now is None:
now = get_now_datetime(relative_with_timezone)
return now + delta
except ValueError:
pass
try:
return datetime.datetime.strptime(value, '%a, %d %b %Y %H:%M:%S GMT')
except ValueError:
pass
raise ValueError('Cannot parse Retry-After header value %s' % repr(value))
def compute_cert_id(
backend,
cert_info=None,
cert_filename=None,
cert_content=None,
none_if_required_information_is_missing=False,
):
# Obtain certificate info if not provided
if cert_info is None:
cert_info = backend.get_cert_information(cert_filename=cert_filename, cert_content=cert_content)
# Convert Authority Key Identifier to string
if cert_info.authority_key_identifier is None:
if none_if_required_information_is_missing:
return None
raise ModuleFailException('Certificate has no Authority Key Identifier extension')
aki = to_native(base64.urlsafe_b64encode(cert_info.authority_key_identifier)).replace('=', '')
# Convert serial number to string
serial_bytes = convert_int_to_bytes(cert_info.serial_number)
if ord(serial_bytes[:1]) >= 128:
serial_bytes = b'\x00' + serial_bytes
serial = to_native(base64.urlsafe_b64encode(serial_bytes)).replace('=', '')
# Compose cert ID
return '{aki}.{serial}'.format(aki=aki, serial=serial)

75
plugins/module_utils/argspec.py Normal file
View File

@@ -0,0 +1,75 @@
# -*- coding: utf-8 -*-
#
# Copyright (c) 2020, Felix Fontein <felix@fontein.de>
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import absolute_import, division, print_function
__metaclass__ = type
from ansible.module_utils.basic import AnsibleModule
def _ensure_list(value):
if value is None:
return []
return list(value)
class ArgumentSpec:
def __init__(self, argument_spec=None, mutually_exclusive=None, required_together=None, required_one_of=None, required_if=None, required_by=None):
self.argument_spec = argument_spec or {}
self.mutually_exclusive = _ensure_list(mutually_exclusive)
self.required_together = _ensure_list(required_together)
self.required_one_of = _ensure_list(required_one_of)
self.required_if = _ensure_list(required_if)
self.required_by = required_by or {}
def update_argspec(self, **kwargs):
self.argument_spec.update(kwargs)
return self
def update(self, mutually_exclusive=None, required_together=None, required_one_of=None, required_if=None, required_by=None):
if mutually_exclusive:
self.mutually_exclusive.extend(mutually_exclusive)
if required_together:
self.required_together.extend(required_together)
if required_one_of:
self.required_one_of.extend(required_one_of)
if required_if:
self.required_if.extend(required_if)
if required_by:
for k, v in required_by.items():
if k in self.required_by:
v = list(self.required_by[k]) + list(v)
self.required_by[k] = v
return self
def merge(self, other):
self.update_argspec(**other.argument_spec)
self.update(
mutually_exclusive=other.mutually_exclusive,
required_together=other.required_together,
required_one_of=other.required_one_of,
required_if=other.required_if,
required_by=other.required_by,
)
return self
def create_ansible_module_helper(self, clazz, args, **kwargs):
return clazz(
*args,
argument_spec=self.argument_spec,
mutually_exclusive=self.mutually_exclusive,
required_together=self.required_together,
required_one_of=self.required_one_of,
required_if=self.required_if,
required_by=self.required_by,
**kwargs)
def create_ansible_module(self, **kwargs):
return self.create_ansible_module_helper(AnsibleModule, (), **kwargs)
__all__ = ('ArgumentSpec', )

16
plugins/module_utils/crypto/math.py
View File

@@ -110,6 +110,9 @@ if sys.version_info[0] >= 3:
def _convert_int_to_bytes(count, no):
return no.to_bytes(count, byteorder='big')
def _convert_bytes_to_int(data):
return int.from_bytes(data, byteorder='big', signed=False)
def _to_hex(no):
return hex(no)[2:]
else:
@@ -122,6 +125,12 @@ else:
raise Exception('Number {1} needs more than {0} bytes!'.format(count, n))
return ('0' * (2 * count - len(h)) + h).decode('hex')
def _convert_bytes_to_int(data):
v = 0
for x in data:
v = (v << 8) | ord(x)
return v
def _to_hex(no):
return '%x' % no
@@ -155,3 +164,10 @@ def convert_int_to_hex(no, digits=None):
if digits is not None and len(value) < digits:
value = '0' * (digits - len(value)) + value
return value
def convert_bytes_to_int(data):
"""
Convert a byte string to an unsigned integer in network byte order.
"""
return _convert_bytes_to_int(data)

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

@@ -15,9 +15,9 @@ import traceback
from ansible.module_utils import six
from ansible.module_utils.basic import missing_required_lib
from ansible_collections.community.crypto.plugins.module_utils.version import LooseVersion
from ansible_collections.community.crypto.plugins.module_utils.argspec import ArgumentSpec
from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.common import ArgumentSpec
from ansible_collections.community.crypto.plugins.module_utils.version import LooseVersion
from ansible_collections.community.crypto.plugins.module_utils.crypto.basic import (
OpenSSLObjectError,

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

@@ -18,8 +18,6 @@ from ansible_collections.community.crypto.plugins.module_utils.ecs.api import EC
from ansible_collections.community.crypto.plugins.module_utils.crypto.support import (
load_certificate,
get_now_datetime,
get_relative_time_option,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import (
@@ -34,6 +32,11 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.module_bac
CertificateProvider,
)
from ansible_collections.community.crypto.plugins.module_utils.time import (
get_now_datetime,
get_relative_time_option,
)
try:
from cryptography.x509.oid import NameOID
except ImportError:

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

@@ -23,7 +23,6 @@ from ansible_collections.community.crypto.plugins.module_utils.version import Lo
from ansible_collections.community.crypto.plugins.module_utils.crypto.support import (
load_certificate,
get_fingerprint_of_bytes,
get_now_datetime,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import (
@@ -40,6 +39,10 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.module_bac
get_publickey_info,
)
from ansible_collections.community.crypto.plugins.module_utils.time import (
get_now_datetime,
)
MINIMAL_CRYPTOGRAPHY_VERSION = '1.6'
CRYPTOGRAPHY_IMP_ERR = None

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

@@ -22,7 +22,6 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.basic impo
from ansible_collections.community.crypto.plugins.module_utils.crypto.support import (
load_privatekey,
load_certificate,
get_relative_time_option,
select_message_digest,
)
@@ -45,6 +44,10 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.module_bac
CertificateProvider,
)
from ansible_collections.community.crypto.plugins.module_utils.time import (
get_relative_time_option,
)
try:
import cryptography
from cryptography import x509

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

@@ -14,7 +14,6 @@ import os
from random import randrange
from ansible_collections.community.crypto.plugins.module_utils.crypto.support import (
get_relative_time_option,
select_message_digest,
)
@@ -35,6 +34,10 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.module_bac
CertificateProvider,
)
from ansible_collections.community.crypto.plugins.module_utils.time import (
get_relative_time_option,
)
try:
import cryptography
from cryptography import x509

31
plugins/module_utils/crypto/module_backends/common.py
View File

@@ -10,26 +10,19 @@ __metaclass__ = type
from ansible.module_utils.basic import AnsibleModule
from ansible_collections.community.crypto.plugins.module_utils.argspec import ArgumentSpec as _ArgumentSpec
class ArgumentSpec:
def __init__(self, argument_spec, mutually_exclusive=None, required_together=None, required_one_of=None, required_if=None, required_by=None):
self.argument_spec = argument_spec
self.mutually_exclusive = mutually_exclusive or []
self.required_together = required_together or []
self.required_one_of = required_one_of or []
self.required_if = required_if or []
self.required_by = required_by or {}
class ArgumentSpec(_ArgumentSpec):
def create_ansible_module_helper(self, clazz, args, **kwargs):
return clazz(
*args,
argument_spec=self.argument_spec,
mutually_exclusive=self.mutually_exclusive,
required_together=self.required_together,
required_one_of=self.required_one_of,
required_if=self.required_if,
required_by=self.required_by,
**kwargs)
result = super(ArgumentSpec, self).create_ansible_module_helper(clazz, args, **kwargs)
result.deprecate(
"The crypto.module_backends.common module utils is deprecated and will be removed from community.crypto 3.0.0."
" Use the argspec module utils from community.crypto instead.",
version='3.0.0',
collection_name='community.crypto',
)
return result
def create_ansible_module(self, **kwargs):
return self.create_ansible_module_helper(AnsibleModule, (), **kwargs)
__all__ = ('AnsibleModule', 'ArgumentSpec')

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

@@ -17,6 +17,8 @@ from ansible.module_utils import six
from ansible.module_utils.basic import missing_required_lib
from ansible.module_utils.common.text.converters import to_native, to_text
from ansible_collections.community.crypto.plugins.module_utils.argspec import ArgumentSpec
from ansible_collections.community.crypto.plugins.module_utils.version import LooseVersion
from ansible_collections.community.crypto.plugins.module_utils.crypto.basic import (
@@ -49,8 +51,6 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.module_bac
get_csr_info,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.common import ArgumentSpec
MINIMAL_CRYPTOGRAPHY_VERSION = '1.3'

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

@@ -17,6 +17,8 @@ from ansible.module_utils import six
from ansible.module_utils.basic import missing_required_lib
from ansible.module_utils.common.text.converters import to_bytes
from ansible_collections.community.crypto.plugins.module_utils.argspec import ArgumentSpec
from ansible_collections.community.crypto.plugins.module_utils.version import LooseVersion
from ansible_collections.community.crypto.plugins.module_utils.crypto.basic import (
@@ -42,8 +44,6 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.module_bac
get_privatekey_info,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.common import ArgumentSpec
MINIMAL_CRYPTOGRAPHY_VERSION = '1.2.3'

6
plugins/module_utils/crypto/module_backends/privatekey_convert.py
View File

@@ -15,12 +15,14 @@ from ansible.module_utils import six
from ansible.module_utils.basic import missing_required_lib
from ansible.module_utils.common.text.converters import to_bytes
from ansible_collections.community.crypto.plugins.module_utils.version import LooseVersion
from ansible_collections.community.crypto.plugins.module_utils.argspec import ArgumentSpec
from ansible_collections.community.crypto.plugins.module_utils.io import (
load_file,
)
from ansible_collections.community.crypto.plugins.module_utils.version import LooseVersion
from ansible_collections.community.crypto.plugins.module_utils.crypto.basic import (
CRYPTOGRAPHY_HAS_X25519,
CRYPTOGRAPHY_HAS_X448,
@@ -37,8 +39,6 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.pem import
identify_private_key_format,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.common import ArgumentSpec
MINIMAL_CRYPTOGRAPHY_VERSION = '1.2.3'

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

@@ -9,19 +9,25 @@ __metaclass__ = type
import abc
import datetime
import errno
import hashlib
import os
import re
from ansible.module_utils import six
from ansible.module_utils.common.text.converters import to_native, to_bytes
from ansible.module_utils.common.text.converters import to_bytes
from ansible_collections.community.crypto.plugins.module_utils.crypto.pem import (
identify_pem_format,
)
from ansible_collections.community.crypto.plugins.module_utils.time import ( # noqa: F401, pylint: disable=unused-import
# These imports are for backwards compatibility
get_now_datetime,
ensure_utc_timezone,
convert_relative_to_datetime,
get_relative_time_option,
)
try:
from OpenSSL import crypto
HAS_PYOPENSSL = True
@@ -279,86 +285,6 @@ def parse_ordered_name_field(input_list, name_field_name):
return result
def get_now_datetime(with_timezone):
if with_timezone:
return datetime.datetime.now(tz=datetime.timezone.utc)
return datetime.datetime.utcnow()
def ensure_utc_timezone(timestamp):
if timestamp.tzinfo is not None:
return timestamp
return timestamp.astimezone(datetime.timezone.utc)
def convert_relative_to_datetime(relative_time_string, with_timezone=False):
"""Get a datetime.datetime or None from a string in the time format described in sshd_config(5)"""
parsed_result = re.match(
r"^(?P<prefix>[+-])((?P<weeks>\d+)[wW])?((?P<days>\d+)[dD])?((?P<hours>\d+)[hH])?((?P<minutes>\d+)[mM])?((?P<seconds>\d+)[sS]?)?$",
relative_time_string)
if parsed_result is None or len(relative_time_string) == 1:
# not matched or only a single "+" or "-"
return None
offset = datetime.timedelta(0)
if parsed_result.group("weeks") is not None:
offset += datetime.timedelta(weeks=int(parsed_result.group("weeks")))
if parsed_result.group("days") is not None:
offset += datetime.timedelta(days=int(parsed_result.group("days")))
if parsed_result.group("hours") is not None:
offset += datetime.timedelta(hours=int(parsed_result.group("hours")))
if parsed_result.group("minutes") is not None:
offset += datetime.timedelta(
minutes=int(parsed_result.group("minutes")))
if parsed_result.group("seconds") is not None:
offset += datetime.timedelta(
seconds=int(parsed_result.group("seconds")))
now = get_now_datetime(with_timezone=with_timezone)
if parsed_result.group("prefix") == "+":
return now + offset
else:
return now - offset
def get_relative_time_option(input_string, input_name, backend='cryptography', with_timezone=False):
"""Return an absolute timespec if a relative timespec or an ASN1 formatted
string is provided.
The return value will be a datetime object for the cryptography backend,
and a ASN1 formatted string for the pyopenssl backend."""
result = to_native(input_string)
if result is None:
raise OpenSSLObjectError(
'The timespec "%s" for %s is not valid' %
input_string, input_name)
# Relative time
if result.startswith("+") or result.startswith("-"):
result_datetime = convert_relative_to_datetime(result, with_timezone=with_timezone)
if backend == 'pyopenssl':
return result_datetime.strftime("%Y%m%d%H%M%SZ")
elif backend == 'cryptography':
return result_datetime
# Absolute time
if backend == 'cryptography':
for date_fmt in ['%Y%m%d%H%M%SZ', '%Y%m%d%H%MZ', '%Y%m%d%H%M%S%z', '%Y%m%d%H%M%z']:
try:
res = datetime.datetime.strptime(result, date_fmt)
except ValueError:
pass
else:
if with_timezone:
res = res.astimezone(datetime.timezone.utc)
return res
raise OpenSSLObjectError(
'The time spec "%s" for %s is invalid' %
(input_string, input_name)
)
def select_message_digest(digest_string):
digest = None
if digest_string == 'sha256':

18
plugins/module_utils/openssh/certificate.py
View File

@@ -31,11 +31,15 @@ from hashlib import sha256
from ansible.module_utils import six
from ansible.module_utils.common.text.converters import to_text
from ansible_collections.community.crypto.plugins.module_utils.crypto.support import convert_relative_to_datetime
from ansible_collections.community.crypto.plugins.module_utils.openssh.utils import (
OpensshParser,
_OpensshWriter,
)
from ansible_collections.community.crypto.plugins.module_utils.time import (
add_or_remove_timezone as _add_or_remove_timezone,
convert_relative_to_datetime,
UTC as _UTC,
)
# See https://cvsweb.openbsd.org/src/usr.bin/ssh/PROTOCOL.certkeys?annotate=HEAD
_USER_TYPE = 1
@@ -66,14 +70,8 @@ _ECDSA_CURVE_IDENTIFIERS_LOOKUP = {
_USE_TIMEZONE = sys.version_info >= (3, 6)
def _ensure_utc_timezone_if_use_timezone(value):
if not _USE_TIMEZONE or value.tzinfo is not None:
return value
return value.astimezone(_datetime.timezone.utc)
_ALWAYS = _ensure_utc_timezone_if_use_timezone(datetime(1970, 1, 1))
_FOREVER = datetime(9999, 12, 31, 23, 59, 59, 999999, _datetime.timezone.utc) if _USE_TIMEZONE else datetime.max
_ALWAYS = _add_or_remove_timezone(datetime(1970, 1, 1), with_timezone=_USE_TIMEZONE)
_FOREVER = datetime(9999, 12, 31, 23, 59, 59, 999999, _UTC) if _USE_TIMEZONE else datetime.max
_CRITICAL_OPTIONS = (
'force-command',
@@ -198,7 +196,7 @@ class OpensshCertificateTimeParameters(object):
else:
for time_format in ("%Y-%m-%d", "%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S"):
try:
result = _ensure_utc_timezone_if_use_timezone(datetime.strptime(time_string, time_format))
result = _add_or_remove_timezone(datetime.strptime(time_string, time_format), with_timezone=_USE_TIMEZONE)
except ValueError:
pass
if result is None:

171
plugins/module_utils/time.py Normal file
View File

@@ -0,0 +1,171 @@
# -*- coding: utf-8 -*-
#
# Copyright (c) 2024, Felix Fontein <felix@fontein.de>
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import datetime
import re
import sys
from ansible.module_utils.common.text.converters import to_native
from ansible_collections.community.crypto.plugins.module_utils.crypto.basic import (
OpenSSLObjectError,
)
try:
UTC = datetime.timezone.utc
except AttributeError:
_DURATION_ZERO = datetime.timedelta(0)
class _UTCClass(datetime.tzinfo):
def utcoffset(self, dt):
return _DURATION_ZERO
def dst(self, dt):
return _DURATION_ZERO
def tzname(self, dt):
return 'UTC'
def fromutc(self, dt):
return dt
def __repr__(self):
return 'UTC'
UTC = _UTCClass()
def get_now_datetime(with_timezone):
if with_timezone:
return datetime.datetime.now(tz=UTC)
return datetime.datetime.utcnow()
def ensure_utc_timezone(timestamp):
if timestamp.tzinfo is UTC:
return timestamp
if timestamp.tzinfo is None:
# We assume that naive datetime objects use timezone UTC!
return timestamp.replace(tzinfo=UTC)
return timestamp.astimezone(UTC)
def remove_timezone(timestamp):
# Convert to native datetime object
if timestamp.tzinfo is None:
return timestamp
if timestamp.tzinfo is not UTC:
timestamp = timestamp.astimezone(UTC)
return timestamp.replace(tzinfo=None)
def add_or_remove_timezone(timestamp, with_timezone):
return ensure_utc_timezone(timestamp) if with_timezone else remove_timezone(timestamp)
if sys.version_info < (3, 3):
def get_epoch_seconds(timestamp):
epoch = datetime.datetime(1970, 1, 1, tzinfo=UTC if timestamp.tzinfo is not None else None)
delta = timestamp - epoch
try:
return delta.total_seconds()
except AttributeError:
# Python 2.6 and earlier: total_seconds() does not yet exist, so we use the formula from
# https://docs.python.org/2/library/datetime.html#datetime.timedelta.total_seconds
return (delta.microseconds + (delta.seconds + delta.days * 24 * 3600) * 10**6) / 10**6
else:
def get_epoch_seconds(timestamp):
return timestamp.timestamp()
def from_epoch_seconds(timestamp, with_timezone):
if with_timezone:
return datetime.datetime.fromtimestamp(timestamp, UTC)
return datetime.datetime.utcfromtimestamp(timestamp)
def convert_relative_to_datetime(relative_time_string, with_timezone=False, now=None):
"""Get a datetime.datetime or None from a string in the time format described in sshd_config(5)"""
parsed_result = re.match(
r"^(?P<prefix>[+-])((?P<weeks>\d+)[wW])?((?P<days>\d+)[dD])?((?P<hours>\d+)[hH])?((?P<minutes>\d+)[mM])?((?P<seconds>\d+)[sS]?)?$",
relative_time_string)
if parsed_result is None or len(relative_time_string) == 1:
# not matched or only a single "+" or "-"
return None
offset = datetime.timedelta(0)
if parsed_result.group("weeks") is not None:
offset += datetime.timedelta(weeks=int(parsed_result.group("weeks")))
if parsed_result.group("days") is not None:
offset += datetime.timedelta(days=int(parsed_result.group("days")))
if parsed_result.group("hours") is not None:
offset += datetime.timedelta(hours=int(parsed_result.group("hours")))
if parsed_result.group("minutes") is not None:
offset += datetime.timedelta(
minutes=int(parsed_result.group("minutes")))
if parsed_result.group("seconds") is not None:
offset += datetime.timedelta(
seconds=int(parsed_result.group("seconds")))
if now is None:
now = get_now_datetime(with_timezone=with_timezone)
else:
now = add_or_remove_timezone(now, with_timezone=with_timezone)
if parsed_result.group("prefix") == "+":
return now + offset
else:
return now - offset
def get_relative_time_option(input_string, input_name, backend='cryptography', with_timezone=False, now=None):
"""Return an absolute timespec if a relative timespec or an ASN1 formatted
string is provided.
The return value will be a datetime object for the cryptography backend,
and a ASN1 formatted string for the pyopenssl backend."""
result = to_native(input_string)
if result is None:
raise OpenSSLObjectError(
'The timespec "%s" for %s is not valid' %
input_string, input_name)
# Relative time
if result.startswith("+") or result.startswith("-"):
result_datetime = convert_relative_to_datetime(result, with_timezone=with_timezone, now=now)
if backend == 'pyopenssl':
return result_datetime.strftime("%Y%m%d%H%M%SZ")
elif backend == 'cryptography':
return result_datetime
# Absolute time
if backend == 'pyopenssl':
return input_string
elif backend == 'cryptography':
for date_fmt, length in [
('%Y%m%d%H%M%SZ', 15), # this also parses '202401020304Z', but as datetime(2024, 1, 2, 3, 0, 4)
('%Y%m%d%H%MZ', 13),
('%Y%m%d%H%M%S%z', 14 + 5), # this also parses '202401020304+0000', but as datetime(2024, 1, 2, 3, 0, 4, tzinfo=...)
('%Y%m%d%H%M%z', 12 + 5),
]:
if len(result) != length:
continue
try:
res = datetime.datetime.strptime(result, date_fmt)
except ValueError:
pass
else:
return add_or_remove_timezone(res, with_timezone=with_timezone)
raise OpenSSLObjectError(
'The time spec "%s" for %s is invalid' %
(input_string, input_name)
)