Use timezone aware functionality when using cryptography >= 42.0.0 (#727)

* Use timezone aware functionality when using cryptography >= 42.0.0.

* Adjust OpenSSH certificate code to avoid functions deprecated in Python 3.12.

* Strip timezone info from isoformat() output.

* InvalidityDate.invalidity_date currently has no _utc variant.
pull/729/head
Felix Fontein 2024-04-18 07:49:53 +02:00 committed by GitHub
parent 1b75f1aa9c
commit ae548de502
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 215 additions and 64 deletions

View File

@ -0,0 +1,4 @@
minor_changes:
- "When using cryptography >= 42.0.0, use offset-aware ``datetime.datetime`` objects (with timezone UTC) instead of offset-naive UTC timestamps
(https://github.com/ansible-collections/community.crypto/issues/726, https://github.com/ansible-collections/community.crypto/pull/727)."
- "openssh_cert - avoid UTC functions deprecated in Python 3.12 when using Python 3 (https://github.com/ansible-collections/community.crypto/pull/727)."

View File

@ -11,7 +11,6 @@ __metaclass__ = type
import base64 import base64
import binascii import binascii
import datetime
import os import os
import traceback import traceback
@ -42,11 +41,15 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.math impor
) )
from ansible_collections.community.crypto.plugins.module_utils.crypto.support import ( from ansible_collections.community.crypto.plugins.module_utils.crypto.support import (
get_now_datetime,
ensure_utc_timezone,
parse_name_field, parse_name_field,
) )
from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import ( from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import (
CRYPTOGRAPHY_TIMEZONE,
cryptography_name_to_oid, cryptography_name_to_oid,
get_not_valid_after,
) )
from ansible_collections.community.crypto.plugins.module_utils.crypto.pem import ( from ansible_collections.community.crypto.plugins.module_utils.crypto.pem import (
@ -373,8 +376,10 @@ class CryptographyBackend(CryptoBackend):
raise BackendException('Cannot parse certificate {0}: {1}'.format(cert_filename, e)) raise BackendException('Cannot parse certificate {0}: {1}'.format(cert_filename, e))
if now is None: if now is None:
now = datetime.datetime.now() now = get_now_datetime(with_timezone=CRYPTOGRAPHY_TIMEZONE)
return (cert.not_valid_after - now).days elif CRYPTOGRAPHY_TIMEZONE:
now = ensure_utc_timezone(now)
return (get_not_valid_after(cert) - now).days
def create_chain_matcher(self, criterium): def create_chain_matcher(self, criterium):
''' '''

View File

@ -19,6 +19,7 @@ from .basic import (
) )
from .cryptography_support import ( from .cryptography_support import (
CRYPTOGRAPHY_TIMEZONE,
cryptography_decode_name, cryptography_decode_name,
) )
@ -27,6 +28,11 @@ from ._obj2txt import (
) )
# TODO: once cryptography has a _utc variant of InvalidityDate.invalidity_date, set this
# to True and adjust get_invalidity_date() accordingly.
# (https://github.com/pyca/cryptography/issues/10818)
CRYPTOGRAPHY_TIMEZONE_INVALIDITY_DATE = False
TIMESTAMP_FORMAT = "%Y%m%d%H%M%SZ" TIMESTAMP_FORMAT = "%Y%m%d%H%M%SZ"
@ -55,7 +61,7 @@ else:
def cryptography_decode_revoked_certificate(cert): def cryptography_decode_revoked_certificate(cert):
result = { result = {
'serial_number': cert.serial_number, 'serial_number': cert.serial_number,
'revocation_date': cert.revocation_date, 'revocation_date': get_revocation_date(cert),
'issuer': None, 'issuer': None,
'issuer_critical': False, 'issuer_critical': False,
'reason': None, 'reason': None,
@ -77,7 +83,7 @@ def cryptography_decode_revoked_certificate(cert):
pass pass
try: try:
ext = cert.extensions.get_extension_for_class(x509.InvalidityDate) ext = cert.extensions.get_extension_for_class(x509.InvalidityDate)
result['invalidity_date'] = ext.value.invalidity_date result['invalidity_date'] = get_invalidity_date(ext.value)
result['invalidity_date_critical'] = ext.critical result['invalidity_date_critical'] = ext.critical
except x509.ExtensionNotFound: except x509.ExtensionNotFound:
pass pass
@ -112,3 +118,38 @@ def cryptography_get_signature_algorithm_oid_from_crl(crl):
crl._x509_crl.sig_alg.algorithm crl._x509_crl.sig_alg.algorithm
) )
return x509.oid.ObjectIdentifier(dotted) return x509.oid.ObjectIdentifier(dotted)
def get_next_update(obj):
if CRYPTOGRAPHY_TIMEZONE:
return obj.next_update_utc
return obj.next_update
def get_last_update(obj):
if CRYPTOGRAPHY_TIMEZONE:
return obj.last_update_utc
return obj.last_update
def get_revocation_date(obj):
if CRYPTOGRAPHY_TIMEZONE:
return obj.revocation_date_utc
return obj.revocation_date
def get_invalidity_date(obj):
# TODO: special handling if CRYPTOGRAPHY_TIMEZONE_INVALIDITY_DATE is True
return obj.invalidity_date
def set_next_update(builder, value):
return builder.next_update(value)
def set_last_update(builder, value):
return builder.last_update(value)
def set_revocation_date(builder, value):
return builder.revocation_date(value)

View File

@ -29,7 +29,9 @@ try:
from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import padding from cryptography.hazmat.primitives.asymmetric import padding
import ipaddress import ipaddress
_HAS_CRYPTOGRAPHY = True
except ImportError: except ImportError:
_HAS_CRYPTOGRAPHY = False
# Error handled in the calling module. # Error handled in the calling module.
pass pass
@ -106,6 +108,11 @@ from ._objects import (
from ._obj2txt import obj2txt from ._obj2txt import obj2txt
CRYPTOGRAPHY_TIMEZONE = False
if _HAS_CRYPTOGRAPHY:
CRYPTOGRAPHY_TIMEZONE = LooseVersion(cryptography.__version__) >= LooseVersion('42.0.0')
DOTTED_OID = re.compile(r'^\d+(?:\.\d+)+$') DOTTED_OID = re.compile(r'^\d+(?:\.\d+)+$')
@ -807,3 +814,23 @@ def cryptography_verify_certificate_signature(certificate, signer_public_key):
certificate.signature_hash_algorithm, certificate.signature_hash_algorithm,
signer_public_key signer_public_key
) )
def get_not_valid_after(obj):
if CRYPTOGRAPHY_TIMEZONE:
return obj.not_valid_after_utc
return obj.not_valid_after
def get_not_valid_before(obj):
if CRYPTOGRAPHY_TIMEZONE:
return obj.not_valid_before_utc
return obj.not_valid_before
def set_not_valid_after(builder, value):
return builder.not_valid_after(value)
def set_not_valid_before(builder, value):
return builder.not_valid_before(value)

View File

@ -32,6 +32,8 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.support im
from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import ( from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import (
cryptography_compare_public_keys, cryptography_compare_public_keys,
get_not_valid_after,
get_not_valid_before,
) )
from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.certificate_info import ( from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.certificate_info import (
@ -251,12 +253,12 @@ class CertificateBackend(object):
# Check not before # Check not before
if not_before is not None and not self.ignore_timestamps: if not_before is not None and not self.ignore_timestamps:
if self.existing_certificate.not_valid_before != not_before: if get_not_valid_before(self.existing_certificate) != not_before:
return True return True
# Check not after # Check not after
if not_after is not None and not self.ignore_timestamps: if not_after is not None and not self.ignore_timestamps:
if self.existing_certificate.not_valid_after != not_after: if get_not_valid_after(self.existing_certificate) != not_after:
return True return True
return False return False

View File

@ -10,7 +10,6 @@ __metaclass__ = type
import datetime import datetime
import time
import os import os
from ansible.module_utils.common.text.converters import to_native, to_bytes from ansible.module_utils.common.text.converters import to_native, to_bytes
@ -19,11 +18,14 @@ from ansible_collections.community.crypto.plugins.module_utils.ecs.api import EC
from ansible_collections.community.crypto.plugins.module_utils.crypto.support import ( from ansible_collections.community.crypto.plugins.module_utils.crypto.support import (
load_certificate, load_certificate,
get_now_datetime,
get_relative_time_option, get_relative_time_option,
) )
from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import ( from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import (
CRYPTOGRAPHY_TIMEZONE,
cryptography_serial_number_of_cert, cryptography_serial_number_of_cert,
get_not_valid_after,
) )
from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.certificate import ( from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.certificate import (
@ -99,7 +101,7 @@ class EntrustCertificateBackend(CertificateBackend):
# Handle expiration (30 days if not specified) # Handle expiration (30 days if not specified)
expiry = self.notAfter expiry = self.notAfter
if not expiry: if not expiry:
gmt_now = datetime.datetime.fromtimestamp(time.mktime(time.gmtime())) gmt_now = get_now_datetime(with_timezone=CRYPTOGRAPHY_TIMEZONE)
expiry = gmt_now + datetime.timedelta(days=365) expiry = gmt_now + datetime.timedelta(days=365)
expiry_iso3339 = expiry.strftime("%Y-%m-%dT%H:%M:%S.00Z") expiry_iso3339 = expiry.strftime("%Y-%m-%dT%H:%M:%S.00Z")
@ -154,7 +156,7 @@ class EntrustCertificateBackend(CertificateBackend):
expiry = None expiry = None
if self.backend == 'cryptography': if self.backend == 'cryptography':
serial_number = "{0:X}".format(cryptography_serial_number_of_cert(self.existing_certificate)) serial_number = "{0:X}".format(cryptography_serial_number_of_cert(self.existing_certificate))
expiry = self.existing_certificate.not_valid_after expiry = get_not_valid_after(self.existing_certificate)
# get some information about the expiry of this certificate # get some information about the expiry of this certificate
expiry_iso3339 = expiry.strftime("%Y-%m-%dT%H:%M:%S.00Z") expiry_iso3339 = expiry.strftime("%Y-%m-%dT%H:%M:%S.00Z")

View File

@ -12,7 +12,6 @@ __metaclass__ = type
import abc import abc
import binascii import binascii
import datetime
import traceback import traceback
from ansible.module_utils import six from ansible.module_utils import six
@ -24,13 +23,17 @@ from ansible_collections.community.crypto.plugins.module_utils.version import Lo
from ansible_collections.community.crypto.plugins.module_utils.crypto.support import ( from ansible_collections.community.crypto.plugins.module_utils.crypto.support import (
load_certificate, load_certificate,
get_fingerprint_of_bytes, get_fingerprint_of_bytes,
get_now_datetime,
) )
from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import ( from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import (
CRYPTOGRAPHY_TIMEZONE,
cryptography_decode_name, cryptography_decode_name,
cryptography_get_extensions_from_cert, cryptography_get_extensions_from_cert,
cryptography_oid_to_name, cryptography_oid_to_name,
cryptography_serial_number_of_cert, cryptography_serial_number_of_cert,
get_not_valid_after,
get_not_valid_before,
) )
from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.publickey_info import ( from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.publickey_info import (
@ -169,7 +172,7 @@ class CertificateInfoRetrieval(object):
not_after = self.get_not_after() not_after = self.get_not_after()
result['not_before'] = not_before.strftime(TIMESTAMP_FORMAT) result['not_before'] = not_before.strftime(TIMESTAMP_FORMAT)
result['not_after'] = not_after.strftime(TIMESTAMP_FORMAT) result['not_after'] = not_after.strftime(TIMESTAMP_FORMAT)
result['expired'] = not_after < datetime.datetime.utcnow() result['expired'] = not_after < get_now_datetime(with_timezone=CRYPTOGRAPHY_TIMEZONE)
result['public_key'] = to_native(self._get_public_key_pem()) result['public_key'] = to_native(self._get_public_key_pem())
@ -322,10 +325,10 @@ class CertificateInfoRetrievalCryptography(CertificateInfoRetrieval):
return None, False return None, False
def get_not_before(self): def get_not_before(self):
return self.cert.not_valid_before return get_not_valid_before(self.cert)
def get_not_after(self): def get_not_after(self):
return self.cert.not_valid_after return get_not_valid_after(self.cert)
def _get_public_key_pem(self): def _get_public_key_pem(self):
return self.cert.public_key().public_bytes( return self.cert.public_key().public_bytes(

View File

@ -31,6 +31,10 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptograp
cryptography_key_needs_digest_for_signing, cryptography_key_needs_digest_for_signing,
cryptography_serial_number_of_cert, cryptography_serial_number_of_cert,
cryptography_verify_certificate_signature, cryptography_verify_certificate_signature,
get_not_valid_after,
get_not_valid_before,
set_not_valid_after,
set_not_valid_before,
) )
from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.certificate import ( from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.certificate import (
@ -120,8 +124,8 @@ class OwnCACertificateBackendCryptography(CertificateBackend):
cert_builder = cert_builder.subject_name(self.csr.subject) cert_builder = cert_builder.subject_name(self.csr.subject)
cert_builder = cert_builder.issuer_name(self.ca_cert.subject) cert_builder = cert_builder.issuer_name(self.ca_cert.subject)
cert_builder = cert_builder.serial_number(self.serial_number) cert_builder = cert_builder.serial_number(self.serial_number)
cert_builder = cert_builder.not_valid_before(self.notBefore) cert_builder = set_not_valid_before(cert_builder, self.notBefore)
cert_builder = cert_builder.not_valid_after(self.notAfter) cert_builder = set_not_valid_after(cert_builder, self.notAfter)
cert_builder = cert_builder.public_key(self.csr.public_key()) cert_builder = cert_builder.public_key(self.csr.public_key())
has_ski = False has_ski = False
for extension in self.csr.extensions: for extension in self.csr.extensions:
@ -220,8 +224,8 @@ class OwnCACertificateBackendCryptography(CertificateBackend):
if self.cert is None: if self.cert is None:
self.cert = self.existing_certificate self.cert = self.existing_certificate
result.update({ result.update({
'notBefore': self.cert.not_valid_before.strftime("%Y%m%d%H%M%SZ"), 'notBefore': get_not_valid_before(self.cert).strftime("%Y%m%d%H%M%SZ"),
'notAfter': self.cert.not_valid_after.strftime("%Y%m%d%H%M%SZ"), 'notAfter': get_not_valid_after(self.cert).strftime("%Y%m%d%H%M%SZ"),
'serial_number': cryptography_serial_number_of_cert(self.cert), 'serial_number': cryptography_serial_number_of_cert(self.cert),
}) })

View File

@ -22,6 +22,10 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptograp
cryptography_key_needs_digest_for_signing, cryptography_key_needs_digest_for_signing,
cryptography_serial_number_of_cert, cryptography_serial_number_of_cert,
cryptography_verify_certificate_signature, cryptography_verify_certificate_signature,
get_not_valid_after,
get_not_valid_before,
set_not_valid_after,
set_not_valid_before,
) )
from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.certificate import ( from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.certificate import (
@ -95,8 +99,8 @@ class SelfSignedCertificateBackendCryptography(CertificateBackend):
cert_builder = cert_builder.subject_name(self.csr.subject) cert_builder = cert_builder.subject_name(self.csr.subject)
cert_builder = cert_builder.issuer_name(self.csr.subject) cert_builder = cert_builder.issuer_name(self.csr.subject)
cert_builder = cert_builder.serial_number(self.serial_number) cert_builder = cert_builder.serial_number(self.serial_number)
cert_builder = cert_builder.not_valid_before(self.notBefore) cert_builder = set_not_valid_before(cert_builder, self.notBefore)
cert_builder = cert_builder.not_valid_after(self.notAfter) cert_builder = set_not_valid_after(cert_builder, self.notAfter)
cert_builder = cert_builder.public_key(self.privatekey.public_key()) cert_builder = cert_builder.public_key(self.privatekey.public_key())
has_ski = False has_ski = False
for extension in self.csr.extensions: for extension in self.csr.extensions:
@ -154,8 +158,8 @@ class SelfSignedCertificateBackendCryptography(CertificateBackend):
if self.cert is None: if self.cert is None:
self.cert = self.existing_certificate self.cert = self.existing_certificate
result.update({ result.update({
'notBefore': self.cert.not_valid_before.strftime("%Y%m%d%H%M%SZ"), 'notBefore': get_not_valid_before(self.cert).strftime("%Y%m%d%H%M%SZ"),
'notAfter': self.cert.not_valid_after.strftime("%Y%m%d%H%M%SZ"), 'notAfter': get_not_valid_after(self.cert).strftime("%Y%m%d%H%M%SZ"),
'serial_number': cryptography_serial_number_of_cert(self.cert), 'serial_number': cryptography_serial_number_of_cert(self.cert),
}) })

View File

@ -279,7 +279,19 @@ def parse_ordered_name_field(input_list, name_field_name):
return result return result
def convert_relative_to_datetime(relative_time_string): 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)""" """Get a datetime.datetime or None from a string in the time format described in sshd_config(5)"""
parsed_result = re.match( parsed_result = re.match(
@ -304,13 +316,14 @@ def convert_relative_to_datetime(relative_time_string):
offset += datetime.timedelta( offset += datetime.timedelta(
seconds=int(parsed_result.group("seconds"))) seconds=int(parsed_result.group("seconds")))
now = get_now_datetime(with_timezone=with_timezone)
if parsed_result.group("prefix") == "+": if parsed_result.group("prefix") == "+":
return datetime.datetime.utcnow() + offset return now + offset
else: else:
return datetime.datetime.utcnow() - offset return now - offset
def get_relative_time_option(input_string, input_name, backend='cryptography'): 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 """Return an absolute timespec if a relative timespec or an ASN1 formatted
string is provided. string is provided.
@ -323,7 +336,7 @@ def get_relative_time_option(input_string, input_name, backend='cryptography'):
input_string, input_name) input_string, input_name)
# Relative time # Relative time
if result.startswith("+") or result.startswith("-"): if result.startswith("+") or result.startswith("-"):
result_datetime = convert_relative_to_datetime(result) result_datetime = convert_relative_to_datetime(result, with_timezone=with_timezone)
if backend == 'pyopenssl': if backend == 'pyopenssl':
return result_datetime.strftime("%Y%m%d%H%M%SZ") return result_datetime.strftime("%Y%m%d%H%M%SZ")
elif backend == 'cryptography': elif backend == 'cryptography':
@ -332,9 +345,13 @@ def get_relative_time_option(input_string, input_name, backend='cryptography'):
if backend == 'cryptography': 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']: 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: try:
return datetime.datetime.strptime(result, date_fmt) res = datetime.datetime.strptime(result, date_fmt)
except ValueError: except ValueError:
pass pass
else:
if with_timezone:
res = res.astimezone(datetime.timezone.utc)
return res
raise OpenSSLObjectError( raise OpenSSLObjectError(
'The time spec "%s" for %s is invalid' % 'The time spec "%s" for %s is invalid' %

View File

@ -22,7 +22,9 @@ __metaclass__ = type
import abc import abc
import binascii import binascii
import datetime as _datetime
import os import os
import sys
from base64 import b64encode from base64 import b64encode
from datetime import datetime from datetime import datetime
from hashlib import sha256 from hashlib import sha256
@ -61,8 +63,17 @@ _ECDSA_CURVE_IDENTIFIERS_LOOKUP = {
b'nistp521': 'ecdsa-nistp521', b'nistp521': 'ecdsa-nistp521',
} }
_ALWAYS = datetime(1970, 1, 1) _USE_TIMEZONE = sys.version_info >= (3, 6)
_FOREVER = datetime.max
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
_CRITICAL_OPTIONS = ( _CRITICAL_OPTIONS = (
'force-command', 'force-command',
@ -136,7 +147,7 @@ class OpensshCertificateTimeParameters(object):
elif dt == _FOREVER: elif dt == _FOREVER:
result = 'forever' result = 'forever'
else: else:
result = dt.isoformat() if date_format == 'human_readable' else dt.strftime("%Y%m%d%H%M%S") result = dt.isoformat().replace('+00:00', '') if date_format == 'human_readable' else dt.strftime("%Y%m%d%H%M%S")
elif date_format == 'timestamp': elif date_format == 'timestamp':
td = dt - _ALWAYS td = dt - _ALWAYS
result = int((td.microseconds + (td.seconds + td.days * 24 * 3600) * 10 ** 6) / 10 ** 6) result = int((td.microseconds + (td.seconds + td.days * 24 * 3600) * 10 ** 6) / 10 ** 6)
@ -167,7 +178,10 @@ class OpensshCertificateTimeParameters(object):
result = _FOREVER result = _FOREVER
else: else:
try: try:
result = datetime.utcfromtimestamp(timestamp) if _USE_TIMEZONE:
result = datetime.fromtimestamp(timestamp, tz=_datetime.timezone.utc)
else:
result = datetime.utcfromtimestamp(timestamp)
except OverflowError as e: except OverflowError as e:
raise ValueError raise ValueError
return result return result
@ -180,11 +194,11 @@ class OpensshCertificateTimeParameters(object):
elif time_string == 'forever': elif time_string == 'forever':
result = _FOREVER result = _FOREVER
elif is_relative_time_string(time_string): elif is_relative_time_string(time_string):
result = convert_relative_to_datetime(time_string) result = convert_relative_to_datetime(time_string, with_timezone=_USE_TIMEZONE)
else: else:
for time_format in ("%Y-%m-%d", "%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S"): for time_format in ("%Y-%m-%d", "%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S"):
try: try:
result = datetime.strptime(time_string, time_format) result = _ensure_utc_timezone_if_use_timezone(datetime.strptime(time_string, time_format))
except ValueError: except ValueError:
pass pass
if result is None: if result is None:

View File

@ -165,6 +165,16 @@ from ansible_collections.community.crypto.plugins.module_utils.acme.io import (
read_file, read_file,
) )
from ansible_collections.community.crypto.plugins.module_utils.crypto.support import (
get_now_datetime,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import (
CRYPTOGRAPHY_TIMEZONE,
set_not_valid_after,
set_not_valid_before,
)
CRYPTOGRAPHY_IMP_ERR = None CRYPTOGRAPHY_IMP_ERR = None
try: try:
import cryptography import cryptography
@ -244,8 +254,9 @@ def main():
domain = to_text(challenge_data['resource']) domain = to_text(challenge_data['resource'])
identifier_type, identifier = to_text(challenge_data.get('resource_original', 'dns:' + challenge_data['resource'])).split(':', 1) identifier_type, identifier = to_text(challenge_data.get('resource_original', 'dns:' + challenge_data['resource'])).split(':', 1)
subject = issuer = cryptography.x509.Name([]) subject = issuer = cryptography.x509.Name([])
not_valid_before = datetime.datetime.utcnow() now = get_now_datetime(with_timezone=CRYPTOGRAPHY_TIMEZONE)
not_valid_after = datetime.datetime.utcnow() + datetime.timedelta(days=10) not_valid_before = now
not_valid_after = now + datetime.timedelta(days=10)
if identifier_type == 'dns': if identifier_type == 'dns':
san = cryptography.x509.DNSName(identifier) san = cryptography.x509.DNSName(identifier)
elif identifier_type == 'ip': elif identifier_type == 'ip':
@ -254,7 +265,7 @@ def main():
raise ModuleFailException('Unsupported identifier type "{0}"'.format(identifier_type)) raise ModuleFailException('Unsupported identifier type "{0}"'.format(identifier_type))
# Generate regular self-signed certificate # Generate regular self-signed certificate
regular_certificate = cryptography.x509.CertificateBuilder().subject_name( cert_builder = cryptography.x509.CertificateBuilder().subject_name(
subject subject
).issuer_name( ).issuer_name(
issuer issuer
@ -262,14 +273,13 @@ def main():
private_key.public_key() private_key.public_key()
).serial_number( ).serial_number(
cryptography.x509.random_serial_number() cryptography.x509.random_serial_number()
).not_valid_before(
not_valid_before
).not_valid_after(
not_valid_after
).add_extension( ).add_extension(
cryptography.x509.SubjectAlternativeName([san]), cryptography.x509.SubjectAlternativeName([san]),
critical=False, critical=False,
).sign( )
cert_builder = set_not_valid_before(cert_builder, not_valid_before)
cert_builder = set_not_valid_after(cert_builder, not_valid_after)
regular_certificate = cert_builder.sign(
private_key, private_key,
cryptography.hazmat.primitives.hashes.SHA256(), cryptography.hazmat.primitives.hashes.SHA256(),
_cryptography_backend _cryptography_backend
@ -278,7 +288,7 @@ def main():
# Process challenge # Process challenge
if challenge == 'tls-alpn-01': if challenge == 'tls-alpn-01':
value = base64.b64decode(challenge_data['resource_value']) value = base64.b64decode(challenge_data['resource_value'])
challenge_certificate = cryptography.x509.CertificateBuilder().subject_name( cert_builder = cryptography.x509.CertificateBuilder().subject_name(
subject subject
).issuer_name( ).issuer_name(
issuer issuer
@ -286,10 +296,6 @@ def main():
private_key.public_key() private_key.public_key()
).serial_number( ).serial_number(
cryptography.x509.random_serial_number() cryptography.x509.random_serial_number()
).not_valid_before(
not_valid_before
).not_valid_after(
not_valid_after
).add_extension( ).add_extension(
cryptography.x509.SubjectAlternativeName([san]), cryptography.x509.SubjectAlternativeName([san]),
critical=False, critical=False,
@ -299,7 +305,10 @@ def main():
encode_octet_string(value), encode_octet_string(value),
), ),
critical=True, critical=True,
).sign( )
cert_builder = set_not_valid_before(cert_builder, not_valid_before)
cert_builder = set_not_valid_after(cert_builder, not_valid_after)
challenge_certificate = cert_builder.sign(
private_key, private_key,
cryptography.hazmat.primitives.hashes.SHA256(), cryptography.hazmat.primitives.hashes.SHA256(),
_cryptography_backend _cryptography_backend

View File

@ -209,7 +209,6 @@ EXAMPLES = '''
import atexit import atexit
import base64 import base64
import datetime
import traceback import traceback
from os.path import isfile from os.path import isfile
@ -221,9 +220,16 @@ 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.version import LooseVersion
from ansible_collections.community.crypto.plugins.module_utils.crypto.support import (
get_now_datetime,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import ( from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import (
CRYPTOGRAPHY_TIMEZONE,
cryptography_oid_to_name, cryptography_oid_to_name,
cryptography_get_extensions_from_cert, cryptography_get_extensions_from_cert,
get_not_valid_after,
get_not_valid_before,
) )
MINIMAL_CRYPTOGRAPHY_VERSION = '1.6' MINIMAL_CRYPTOGRAPHY_VERSION = '1.6'
@ -392,7 +398,7 @@ def main():
for attribute in x509.subject: for attribute in x509.subject:
result['subject'][cryptography_oid_to_name(attribute.oid, short=True)] = attribute.value result['subject'][cryptography_oid_to_name(attribute.oid, short=True)] = attribute.value
result['expired'] = x509.not_valid_after < datetime.datetime.utcnow() result['expired'] = get_not_valid_after(x509) < get_now_datetime(with_timezone=CRYPTOGRAPHY_TIMEZONE)
result['extensions'] = [] result['extensions'] = []
for dotted_number, entry in cryptography_get_extensions_from_cert(x509).items(): for dotted_number, entry in cryptography_get_extensions_from_cert(x509).items():
@ -410,8 +416,8 @@ def main():
for attribute in x509.issuer: for attribute in x509.issuer:
result['issuer'][cryptography_oid_to_name(attribute.oid, short=True)] = attribute.value result['issuer'][cryptography_oid_to_name(attribute.oid, short=True)] = attribute.value
result['not_after'] = x509.not_valid_after.strftime('%Y%m%d%H%M%SZ') result['not_after'] = get_not_valid_after(x509).strftime('%Y%m%d%H%M%SZ')
result['not_before'] = x509.not_valid_before.strftime('%Y%m%d%H%M%SZ') result['not_before'] = get_not_valid_before(x509).strftime('%Y%m%d%H%M%SZ')
result['serial_number'] = x509.serial_number result['serial_number'] = x509.serial_number
result['signature_algorithm'] = cryptography_oid_to_name(x509.signature_algorithm_oid) result['signature_algorithm'] = cryptography_oid_to_name(x509.signature_algorithm_oid)

View File

@ -410,6 +410,10 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.support im
get_relative_time_option, get_relative_time_option,
) )
from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import (
CRYPTOGRAPHY_TIMEZONE,
)
from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.certificate_info import ( from ansible_collections.community.crypto.plugins.module_utils.crypto.module_backends.certificate_info import (
select_backend, select_backend,
) )
@ -451,7 +455,7 @@ def main():
module.fail_json( module.fail_json(
msg='The value for valid_at.{0} must be of type string (got {1})'.format(k, type(v)) msg='The value for valid_at.{0} must be of type string (got {1})'.format(k, type(v))
) )
valid_at[k] = get_relative_time_option(v, 'valid_at.{0}'.format(k)) valid_at[k] = get_relative_time_option(v, 'valid_at.{0}'.format(k), with_timezone=CRYPTOGRAPHY_TIMEZONE)
try: try:
result = module_backend.get_info(der_support_enabled=module.params['content'] is None) result = module_backend.get_info(der_support_enabled=module.params['content'] is None)

View File

@ -475,6 +475,7 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.support im
) )
from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import ( from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_support import (
CRYPTOGRAPHY_TIMEZONE,
cryptography_decode_name, cryptography_decode_name,
cryptography_get_name, cryptography_get_name,
cryptography_key_needs_digest_for_signing, cryptography_key_needs_digest_for_signing,
@ -484,11 +485,17 @@ from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptograp
) )
from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_crl import ( from ansible_collections.community.crypto.plugins.module_utils.crypto.cryptography_crl import (
CRYPTOGRAPHY_TIMEZONE_INVALIDITY_DATE,
REVOCATION_REASON_MAP, REVOCATION_REASON_MAP,
TIMESTAMP_FORMAT, TIMESTAMP_FORMAT,
cryptography_decode_revoked_certificate, cryptography_decode_revoked_certificate,
cryptography_dump_revoked, cryptography_dump_revoked,
cryptography_get_signature_algorithm_oid_from_crl, cryptography_get_signature_algorithm_oid_from_crl,
get_next_update,
get_last_update,
set_next_update,
set_last_update,
set_revocation_date,
) )
from ansible_collections.community.crypto.plugins.module_utils.crypto.pem import ( from ansible_collections.community.crypto.plugins.module_utils.crypto.pem import (
@ -560,8 +567,8 @@ class CRL(OpenSSLObject):
except (TypeError, ValueError) as exc: except (TypeError, ValueError) as exc:
module.fail_json(msg=to_native(exc)) module.fail_json(msg=to_native(exc))
self.last_update = get_relative_time_option(module.params['last_update'], 'last_update') self.last_update = get_relative_time_option(module.params['last_update'], 'last_update', with_timezone=CRYPTOGRAPHY_TIMEZONE)
self.next_update = get_relative_time_option(module.params['next_update'], 'next_update') self.next_update = get_relative_time_option(module.params['next_update'], 'next_update', with_timezone=CRYPTOGRAPHY_TIMEZONE)
self.digest = select_message_digest(module.params['digest']) self.digest = select_message_digest(module.params['digest'])
if self.digest is None: if self.digest is None:
@ -607,7 +614,8 @@ class CRL(OpenSSLObject):
result['issuer_critical'] = rc['issuer_critical'] result['issuer_critical'] = rc['issuer_critical']
result['revocation_date'] = get_relative_time_option( result['revocation_date'] = get_relative_time_option(
rc['revocation_date'], rc['revocation_date'],
path_prefix + 'revocation_date' path_prefix + 'revocation_date',
with_timezone=CRYPTOGRAPHY_TIMEZONE,
) )
if rc['reason']: if rc['reason']:
result['reason'] = REVOCATION_REASON_MAP[rc['reason']] result['reason'] = REVOCATION_REASON_MAP[rc['reason']]
@ -615,7 +623,8 @@ class CRL(OpenSSLObject):
if rc['invalidity_date']: if rc['invalidity_date']:
result['invalidity_date'] = get_relative_time_option( result['invalidity_date'] = get_relative_time_option(
rc['invalidity_date'], rc['invalidity_date'],
path_prefix + 'invalidity_date' path_prefix + 'invalidity_date',
with_timezone=CRYPTOGRAPHY_TIMEZONE_INVALIDITY_DATE,
) )
result['invalidity_date_critical'] = rc['invalidity_date_critical'] result['invalidity_date_critical'] = rc['invalidity_date_critical']
self.revoked_certificates.append(result) self.revoked_certificates.append(result)
@ -731,9 +740,9 @@ class CRL(OpenSSLObject):
if self.crl is None: if self.crl is None:
return False return False
if self.last_update != self.crl.last_update and not self.ignore_timestamps: if self.last_update != get_last_update(self.crl) and not self.ignore_timestamps:
return False return False
if self.next_update != self.crl.next_update and not self.ignore_timestamps: if self.next_update != get_next_update(self.crl) and not self.ignore_timestamps:
return False return False
if cryptography_key_needs_digest_for_signing(self.privatekey): if cryptography_key_needs_digest_for_signing(self.privatekey):
if self.crl.signature_hash_algorithm is None or self.digest.name != self.crl.signature_hash_algorithm.name: if self.crl.signature_hash_algorithm is None or self.digest.name != self.crl.signature_hash_algorithm.name:
@ -780,8 +789,8 @@ class CRL(OpenSSLObject):
except ValueError as e: except ValueError as e:
raise CRLError(e) raise CRLError(e)
crl = crl.last_update(self.last_update) crl = set_last_update(crl, self.last_update)
crl = crl.next_update(self.next_update) crl = set_next_update(crl, self.next_update)
if self.update and self.crl: if self.update and self.crl:
new_entries = set([self._compress_entry(entry) for entry in self.revoked_certificates]) new_entries = set([self._compress_entry(entry) for entry in self.revoked_certificates])
@ -792,7 +801,7 @@ class CRL(OpenSSLObject):
for entry in self.revoked_certificates: for entry in self.revoked_certificates:
revoked_cert = RevokedCertificateBuilder() revoked_cert = RevokedCertificateBuilder()
revoked_cert = revoked_cert.serial_number(entry['serial_number']) revoked_cert = revoked_cert.serial_number(entry['serial_number'])
revoked_cert = revoked_cert.revocation_date(entry['revocation_date']) revoked_cert = set_revocation_date(revoked_cert, entry['revocation_date'])
if entry['issuer'] is not None: if entry['issuer'] is not None:
revoked_cert = revoked_cert.add_extension( revoked_cert = revoked_cert.add_extension(
x509.CertificateIssuer(entry['issuer']), x509.CertificateIssuer(entry['issuer']),
@ -876,8 +885,8 @@ class CRL(OpenSSLObject):
for entry in self.revoked_certificates: for entry in self.revoked_certificates:
result['revoked_certificates'].append(cryptography_dump_revoked(entry, idn_rewrite=self.name_encoding)) result['revoked_certificates'].append(cryptography_dump_revoked(entry, idn_rewrite=self.name_encoding))
elif self.crl: elif self.crl:
result['last_update'] = self.crl.last_update.strftime(TIMESTAMP_FORMAT) result['last_update'] = get_last_update(self.crl).strftime(TIMESTAMP_FORMAT)
result['next_update'] = self.crl.next_update.strftime(TIMESTAMP_FORMAT) result['next_update'] = get_next_update(self.crl).strftime(TIMESTAMP_FORMAT)
result['digest'] = cryptography_oid_to_name(cryptography_get_signature_algorithm_oid_from_crl(self.crl)) result['digest'] = cryptography_oid_to_name(cryptography_get_signature_algorithm_oid_from_crl(self.crl))
issuer = [] issuer = []
for attribute in self.crl.issuer: for attribute in self.crl.issuer: