[3006.x] Migrate to TZ-aware datetime objects in x509_v2

changelog/65837.fixed.md

@@ -0,0 +1 @@
+Corrected x509_v2 CRL creation `last_update` and `next_update` values when system timezone is not UTC

salt/modules/x509_v2.py

@@ -135,12 +135,12 @@
 
 import base64
 import copy
-import datetime
 import glob
 import logging
 import os.path
 import re
 import sys
+from datetime import datetime, timedelta, timezone
 
 try:
     import cryptography.x509 as cx509
@@ -1376,10 +1376,12 @@ def expires(certificate, days=0):
         Defaults to ``0``, which checks for the current time.
     """
     cert = x509util.load_cert(certificate)
-    # dates are encoded in UTC/GMT, they are returned as a naive datetime object
-    return cert.not_valid_after <= datetime.datetime.utcnow() + datetime.timedelta(
-        days=days
-    )
+    try:
+        not_after = cert.not_valid_after_utc
+    except AttributeError:
+        # naive datetime object, release <42 (it's always UTC)
+        not_after = cert.not_valid_after.replace(tzinfo=timezone.utc)
+    return not_after <= datetime.now(tz=timezone.utc) + timedelta(days=days)
 
 
 def expired(certificate):
@@ -1659,6 +1661,13 @@ def read_certificate(certificate):
     cert = x509util.load_cert(certificate)
     key_type = x509util.get_key_type(cert.public_key(), as_string=True)
 
+    try:
+        not_before = cert.not_valid_before_utc
+        not_after = cert.not_valid_after_utc
+    except AttributeError:
+        # naive datetime object, release <42 (it's always UTC)
+        not_before = cert.not_valid_before.replace(tzinfo=timezone.utc)
+        not_after = cert.not_valid_after.replace(tzinfo=timezone.utc)
     ret = {
         "version": cert.version.value + 1,  # 0-indexed
         "key_size": cert.public_key().key_size if key_type in ["ec", "rsa"] else None,
@@ -1674,8 +1683,8 @@ def read_certificate(certificate):
         "issuer": _parse_dn(cert.issuer),
         "issuer_hash": x509util.pretty_hex(_get_name_hash(cert.issuer)),
         "issuer_str": cert.issuer.rfc4514_string(),
-        "not_before": cert.not_valid_before.strftime(x509util.TIME_FMT),
-        "not_after": cert.not_valid_after.strftime(x509util.TIME_FMT),
+        "not_before": not_before.strftime(x509util.TIME_FMT),
+        "not_after": not_after.strftime(x509util.TIME_FMT),
         "public_key": get_public_key(cert),
         "extensions": _parse_extensions(cert.extensions),
     }
@@ -1741,10 +1750,16 @@ def read_crl(crl):
         The certificate revocation list to read.
     """
     crl = x509util.load_crl(crl)
+    try:
+        last_update = crl.last_update_utc
+        next_update = crl.next_update_utc
+    except AttributeError:
+        last_update = crl.last_update.replace(tzinfo=timezone.utc)
+        next_update = crl.next_update.replace(tzinfo=timezone.utc)
     ret = {
         "issuer": _parse_dn(crl.issuer),
-        "last_update": crl.last_update.strftime(x509util.TIME_FMT),
-        "next_update": crl.next_update.strftime(x509util.TIME_FMT),
+        "last_update": last_update.strftime(x509util.TIME_FMT),
+        "next_update": next_update.strftime(x509util.TIME_FMT),
         "revoked_certificates": {},
         "extensions": _parse_extensions(crl.extensions),
     }
@@ -1764,12 +1779,15 @@ def read_crl(crl):
         ret["signature_algorithm"] = crl.signature_algorithm_oid.dotted_string
 
     for revoked in crl:
+        try:
+            revocation_date = revoked.revocation_date_utc
+        except AttributeError:
+            # naive datetime object, release <42 (it's always UTC)
+            revocation_date = revoked.revocation_date.replace(tzinfo=timezone.utc)
         ret["revoked_certificates"].update(
             {
                 x509util.dec2hex(revoked.serial_number).replace(":", ""): {
-                    "revocation_date": revoked.revocation_date.strftime(
-                        x509util.TIME_FMT
-                    ),
+                    "revocation_date": revocation_date.strftime(x509util.TIME_FMT),
                     "extensions": _parse_crl_entry_extensions(revoked.extensions),
                 }
             }

salt/states/x509_v2.py

@@ -183,9 +183,9 @@
 
 import base64
 import copy
-import datetime
 import logging
 import os.path
+from datetime import datetime, timedelta, timezone
 
 import salt.utils.files
 from salt.exceptions import CommandExecutionError, SaltInvocationError
@@ -487,11 +487,16 @@ def certificate_managed(
                     else None
                 ):
                     changes["pkcs12_friendlyname"] = pkcs12_friendlyname
+                try:
+                    curr_not_after = current.not_valid_after_utc
+                except AttributeError:
+                    # naive datetime object, release <42 (it's always UTC)
+                    curr_not_after = current.not_valid_after.replace(
+                        tzinfo=timezone.utc
+                    )
 
-                if (
-                    current.not_valid_after
-                    < datetime.datetime.utcnow()
-                    + datetime.timedelta(days=days_remaining)
+                if curr_not_after < datetime.now(tz=timezone.utc) + timedelta(
+                    days=days_remaining
                 ):
                     changes["expiration"] = True
 
@@ -896,10 +901,14 @@ def crl_managed(
 
                 if encoding != current_encoding:
                     changes["encoding"] = encoding
+                try:
+                    curr_next_update = current.next_update_utc
+                except AttributeError:
+                    # naive datetime object, release <42 (it's always UTC)
+                    curr_next_update = current.next_update.replace(tzinfo=timezone.utc)
                 if days_remaining and (
-                    current.next_update
-                    < datetime.datetime.utcnow()
-                    + datetime.timedelta(days=days_remaining)
+                    curr_next_update
+                    < datetime.now(tz=timezone.utc) + timedelta(days=days_remaining)
                 ):
                     changes["expiration"] = True
 

salt/utils/x509.py

@@ -1,10 +1,10 @@
 import base64
 import copy
-import datetime
 import ipaddress
 import logging
 import os.path
 import re
+from datetime import datetime, timedelta, timezone
 from enum import Enum
 from urllib.parse import urlparse, urlunparse
 
@@ -313,14 +313,14 @@ def build_crt(
     )
 
     not_before = (
-        datetime.datetime.strptime(not_before, TIME_FMT)
+        datetime.strptime(not_before, TIME_FMT).replace(tzinfo=timezone.utc)
         if not_before
-        else datetime.datetime.utcnow()
+        else datetime.now(tz=timezone.utc)
     )
     not_after = (
-        datetime.datetime.strptime(not_after, TIME_FMT)
+        datetime.strptime(not_after, TIME_FMT).replace(tzinfo=timezone.utc)
         if not_after
-        else datetime.datetime.utcnow() + datetime.timedelta(days=days_valid)
+        else datetime.now(tz=timezone.utc) + timedelta(days=days_valid)
     )
     builder = builder.not_valid_before(not_before).not_valid_after(not_after)
 
@@ -422,32 +422,38 @@ def build_crl(
     builder = cx509.CertificateRevocationListBuilder()
     if signing_cert:
         builder = builder.issuer_name(signing_cert.subject)
-    builder = builder.last_update(datetime.datetime.today())
+    builder = builder.last_update(datetime.now(tz=timezone.utc))
     builder = builder.next_update(
-        datetime.datetime.today() + datetime.timedelta(days=days_valid)
+        datetime.now(tz=timezone.utc) + timedelta(days=days_valid)
     )
     for rev in revoked:
         serial_number = not_after = revocation_date = None
         if "not_after" in rev:
-            not_after = datetime.datetime.strptime(rev["not_after"], TIME_FMT)
+            not_after = datetime.strptime(rev["not_after"], TIME_FMT).replace(
+                tzinfo=timezone.utc
+            )
         if "serial_number" in rev:
             serial_number = rev["serial_number"]
         if "certificate" in rev:
             rev_cert = load_cert(rev["certificate"])
             serial_number = rev_cert.serial_number
-            not_after = rev_cert.not_valid_after
+            try:
+                not_after = rev_cert.not_valid_after_utc
+            except AttributeError:
+                # naive datetime object, release <42 (it's always UTC)
+                not_after = rev_cert.not_valid_after.replace(tzinfo=timezone.utc)
         if not serial_number:
             raise SaltInvocationError("Need serial_number or certificate")
         serial_number = _get_serial_number(serial_number)
         if not_after and not include_expired:
-            if datetime.datetime.utcnow() > not_after:
+            if datetime.now(tz=timezone.utc) > not_after:
                 continue
         if "revocation_date" in rev:
-            revocation_date = datetime.datetime.strptime(
+            revocation_date = datetime.strptime(
                 rev["revocation_date"], TIME_FMT
-            )
+            ).replace(tzinfo=timezone.utc)
         else:
-            revocation_date = datetime.datetime.utcnow()
+            revocation_date = datetime.now(tz=timezone.utc)
 
         revoked_cert = cx509.RevokedCertificateBuilder(
             serial_number=serial_number, revocation_date=revocation_date
@@ -1624,8 +1630,9 @@ def _create_invalidity_date(val, **kwargs):
     if critical:
         val = val.split(" ", maxsplit=1)[1]
     try:
+        # InvalidityDate deals in naive datetime objects only currently
         return (
-            cx509.InvalidityDate(datetime.datetime.strptime(val, TIME_FMT)),
+            cx509.InvalidityDate(datetime.strptime(val, TIME_FMT)),
             critical,
         )
     except ValueError as err:

tests/pytests/unit/utils/test_x509.py

@@ -1,5 +1,5 @@
-import datetime
 import ipaddress
+from datetime import datetime, timedelta, timezone
 
 import pytest
 
@@ -11,6 +11,9 @@
     "cryptography", reason="Needs cryptography library", minversion="37.0"
 )
 cx509 = pytest.importorskip("cryptography.x509", reason="Needs cryptography library")
+cprim = pytest.importorskip(
+    "cryptography.hazmat.primitives", reason="Needs cryptography library"
+)
 
 
 @pytest.fixture
@@ -1019,12 +1022,12 @@ def test_create_crl_reason(self, val, expected, critical):
         [
             (
                 "critical, 2022-10-11 13:37:42",
-                datetime.datetime.strptime("2022-10-11 13:37:42", "%Y-%m-%d %H:%M:%S"),
+                datetime.strptime("2022-10-11 13:37:42", "%Y-%m-%d %H:%M:%S"),
                 True,
             ),
             (
                 "2022-10-11 13:37:42",
-                datetime.datetime.strptime("2022-10-11 13:37:42", "%Y-%m-%d %H:%M:%S"),
+                datetime.strptime("2022-10-11 13:37:42", "%Y-%m-%d %H:%M:%S"),
                 False,
             ),
         ],
@@ -1875,9 +1878,7 @@ def test_get_dn(inpt, expected):
             cx509.Extension(
                 cx509.InvalidityDate.oid,
                 value=cx509.InvalidityDate(
-                    datetime.datetime.strptime(
-                        "2022-10-11 13:37:42", "%Y-%m-%d %H:%M:%S"
-                    )
+                    datetime.strptime("2022-10-11 13:37:42", "%Y-%m-%d %H:%M:%S")
                 ),
                 critical=False,
             ),
@@ -1888,3 +1889,86 @@ def test_get_dn(inpt, expected):
 def test_render_extension(inpt, expected):
     ret = x509.render_extension(inpt)
     assert ret == expected
+
+
+@pytest.fixture
+def ca_cert():
+    return """\
+-----BEGIN CERTIFICATE-----
+MIIDODCCAiCgAwIBAgIIbfpgqP0VGPgwDQYJKoZIhvcNAQELBQAwKzELMAkGA1UE
+BhMCVVMxDTALBgNVBAMMBFRlc3QxDTALBgNVBAoMBFNhbHQwHhcNMjIxMTE1MTQw
+NDMzWhcNMzIxMTEyMTQwNDMzWjArMQswCQYDVQQGEwJVUzENMAsGA1UEAwwEVGVz
+dDENMAsGA1UECgwEU2FsdDCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEB
+AOGTScvrjcEt6vsJcG9RUp6fKaDNDWZnJET0omanK9ZwaoGpJPp8UDYe/8ADeI7N
+10wdyB4oDM9gRDjInBtdQO/PsrmKZF6LzqVFgLMxu2up+PHMi9z6B2P4esIAzMu9
+PYxc9zH4HzLImHqscVD2HCabsjp9X134Af7hVY5NN/W/4qTP7uOM20wSG2TPI6+B
+tA9VyPbEPMPRzXzrqc45rVYe6kb2bT84GE93Vcu/e5JZ/k2AKD8Hoa2cxLPsTLq5
+igl+D+k+dfUtiABiKPvVQiYBsD1fyHDn2m7B6pCgvrGqHjsoAKufgFnXy6PJRg7n
+vQfaxSiusM5s+VS+fjlvgwsCAwEAAaNgMF4wDwYDVR0TBAgwBgEB/wIBATALBgNV
+HQ8EBAMCAQYwHQYDVR0OBBYEFFzy8fRTKSOe7kBakqO0Ki71potnMB8GA1UdIwQY
+MBaAFFzy8fRTKSOe7kBakqO0Ki71potnMA0GCSqGSIb3DQEBCwUAA4IBAQBZS4MP
+fXYPoGZ66seM+0eikScZHirbRe8vHxHkujnTBUjQITKm86WeQgeBCD2pobgBGZtt
+5YFozM4cERqY7/1BdemUxFvPmMFFznt0TM5w+DfGWVK8un6SYwHnmBbnkWgX4Srm
+GsL0HHWxVXkGnFGFk6Sbo3vnN7CpkpQTWFqeQQ5rHOw91pt7KnNZwc6I3ZjrCUHJ
++UmKKrga16a4Q+8FBpYdphQU609npo/0zuaE6FyiJYlW3tG+mlbbNgzY/+eUaxt2
+9Bp9mtA+Hkox551Mfpq45Oi+ehwMt0xjZCjuFCM78oiUdHCGO+EmcT7ogiYALiOF
+LN1w5sybsYwIw6QN
+-----END CERTIFICATE-----
+"""
+
+
+@pytest.fixture
+def ca_key():
+    return """\
+-----BEGIN RSA PRIVATE KEY-----
+MIIEowIBAAKCAQEA4ZNJy+uNwS3q+wlwb1FSnp8poM0NZmckRPSiZqcr1nBqgakk
++nxQNh7/wAN4js3XTB3IHigMz2BEOMicG11A78+yuYpkXovOpUWAszG7a6n48cyL
+3PoHY/h6wgDMy709jFz3MfgfMsiYeqxxUPYcJpuyOn1fXfgB/uFVjk039b/ipM/u
+44zbTBIbZM8jr4G0D1XI9sQ8w9HNfOupzjmtVh7qRvZtPzgYT3dVy797kln+TYAo
+PwehrZzEs+xMurmKCX4P6T519S2IAGIo+9VCJgGwPV/IcOfabsHqkKC+saoeOygA
+q5+AWdfLo8lGDue9B9rFKK6wzmz5VL5+OW+DCwIDAQABAoIBAFfImc9hu6iR1gAb
+jEXFwAE6r1iEc9KGEPdEvG52X/jzhn8u89UGy7BEIAL5VtE8Caz1agtSSqnpLKNs
+blO31q18hnDuCmFAxwpKIeuaTvV3EAoJL+Su6HFfIWaeKRSgcHNPOmOXy4xXw/75
+XJ/FJu9fZ9ybLaHEAgLObh0Sr9RSPQbZ72ZawPP8+5WCbR+2w90RApHXQL0piSbW
+lIx1NE6o5wQb3vik8z/k5FqLCY2a8++WNyfvS+WWFY5WXGI7ZiDDQk46gnslquH2
+Lon5CEn3JlTGQFhxaaa2ivssscf2lA2Rvm2E8o1rdZJS2OpSE0ai4TXY9XnyjZj1
+5usWIwECgYEA+3Mwu03A7PyLEBksS/u3MSo/176S9lF/uXcecQNdhAIalUZ8AgV3
+7HP2yI9ZC0ekA809ZzFjGFostXm9VfUOEZ549jLOMzvBtCdaI0aBUE8icu52fX4r
+fT2NY6hYgz5/fxD8sq1XH/fqNNexABwtViH6YAly/9A1/8M3BOWt72UCgYEA5ag8
+sIfiBUoWd1sS6qHDuugWlpx4ZWYC/59XEJyCN2wioP8qFji/aNZxF1wLfyQe/zaa
+YBFusjsBnSfBU1p4UKCRHWQ9/CnC0DzqTkyKC4Fv8GuxgywNm5W9gPKk7idHP7mw
+e+7Uvf1pOQccqEPh7yltpW+Xw27gfsC2DMAIGa8CgYByv/q5P56PiCCeVB6W/mR3
+l2RTPLEsn7y+EtJdmL+QgrVG8kedVImJ6tHwbRqhvyvmYD9pXGxwrJZCqy/wjkjB
+WaSyFjVrxBV99Yd5Ga/hyntaH+ELHA0UtoZTuHvMSTU9866ei+R6vlSvkM9B0ZoO
++KqeMTG99HLwKVJudbKO0QKBgQCd33U49XBOqoufKSBr4yAmUH2Ws6GgMuxExUiY
+xr5NUyzK+B36gLA0ZZYAtOnCURZt4x9kgxdRtnZ5jma74ilrY7XeOpbRzfN6KyX3
+BW6wUh6da6rvvUztc5Z+Gk9+18mG6SOFTr04jgfTiCwPD/s06YnSfFAbrRDukZOU
+WD45SQKBgBvjSwl3AbPoJnRjZjGuCUMKQKrLm30xCeorxasu+di/4YV5Yd8VUjaO
+mYyqXW6bQndKLuXT+AXtCd/Xt2sI96z8mc0G5fImDUxQjMUuS3RyQK357cEOu8Zy
+HdI7Pfaf/l0HozAw/Al+LXbpmSBdfmz0U/EGAKRqXMW5+vQ7XHXD
+-----END RSA PRIVATE KEY-----"""
+
+
+def test_build_crl_accounts_for_local_time_zone(ca_key, ca_cert):
+    curr_time = datetime.now(tz=timezone(timedelta(hours=1)))
+    curr_time_naive = curr_time.replace(tzinfo=None)
+
+    def dtn(tz=None):
+        if tz is None:
+            return curr_time_naive
+        return curr_time
+
+    curr_time_utc = curr_time.astimezone(timezone.utc).replace(microsecond=0)
+    curr_time_utc_naive = curr_time_utc.replace(tzinfo=None)
+    privkey = cprim.serialization.load_pem_private_key(ca_key.encode(), password=None)
+    cert = cx509.load_pem_x509_certificate(ca_cert.encode())
+    with patch("salt.utils.x509.datetime") as fakedate:
+        fakedate.today.return_value = curr_time_naive
+        fakedate.now.side_effect = dtn
+        fakedate.utcnow.return_value = curr_time_utc_naive
+        builder, _ = x509.build_crl(privkey, [], signing_cert=cert)
+        crl = builder.sign(privkey, algorithm=cprim.hashes.SHA256())
+    try:
+        assert crl.last_update_utc == curr_time_utc
+    except AttributeError:
+        assert crl.last_update == curr_time_utc_naive