fix: Resolve Python 3.12 datetime.utcnow() DeprecationWarning

awscli/botocore/auth.py

@@ -26,6 +26,7 @@
     HTTPHeaders,
     encodebytes,
     ensure_unicode,
+    get_current_datetime,
     json,
     parse_qs,
     quote,
@@ -426,7 +427,7 @@ def signature(self, string_to_sign, request):
     def add_auth(self, request):
         if self.credentials is None:
             raise NoCredentialsError()
-        datetime_now = datetime.datetime.utcnow()
+        datetime_now = get_current_datetime()
         request.context['timestamp'] = datetime_now.strftime(SIGV4_TIMESTAMP)
         # This could be a retry.  Make sure the previous
         # authorization header is removed first.
@@ -564,7 +565,7 @@ class S3ExpressPostAuth(S3ExpressAuth):
     REQUIRES_IDENTITY_CACHE = True
 
     def add_auth(self, request):
-        datetime_now = datetime.datetime.utcnow()
+        datetime_now = get_current_datetime()
         request.context['timestamp'] = datetime_now.strftime(SIGV4_TIMESTAMP)
 
         fields = {}
@@ -825,7 +826,7 @@ class S3SigV4PostAuth(SigV4Auth):
     """
 
     def add_auth(self, request):
-        datetime_now = datetime.datetime.utcnow()
+        datetime_now = get_current_datetime()
         request.context['timestamp'] = datetime_now.strftime(SIGV4_TIMESTAMP)
 
         fields = {}

awscli/botocore/compat.py

@@ -289,6 +289,14 @@ def get_tzinfo_options():
         return (tzlocal,)
 
 
+def get_current_datetime(remove_tzinfo=True):
+    """Retrieve the current timezone in UTC, with or without an explicit timezone."""
+    datetime_now = datetime.datetime.now(datetime.timezone.utc)
+    if remove_tzinfo:
+        datetime_now = datetime_now.replace(tzinfo=None)
+    return datetime_now
+
+
 ########################################################
 #              urllib3 compat backports                #
 ########################################################

awscli/botocore/crt/auth.py

@@ -11,7 +11,6 @@
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
 
-import datetime
 from io import BytesIO
 
 import awscrt.auth
@@ -23,7 +22,13 @@
     _get_body_as_dict,
     _host_from_url,
 )
-from botocore.compat import HTTPHeaders, parse_qs, urlsplit, urlunsplit
+from botocore.compat import (
+    HTTPHeaders,
+    get_current_datetime,
+    parse_qs,
+    urlsplit,
+    urlunsplit,
+)
 from botocore.exceptions import NoCredentialsError
 from botocore.useragent import register_feature_id
 from botocore.utils import percent_encode_sequence
@@ -56,11 +61,7 @@ def add_auth(self, request):
         if self.credentials is None:
             raise NoCredentialsError()
 
-        # Use utcnow() because that's what gets mocked by tests, but set
-        # timezone because CRT assumes naive datetime is local time.
-        datetime_now = datetime.datetime.utcnow().replace(
-            tzinfo=datetime.timezone.utc
-        )
+        datetime_now = get_current_datetime(remove_tzinfo=False)
 
         # Use existing 'X-Amz-Content-SHA256' header if able
         existing_sha256 = self._get_existing_sha256(request)
@@ -254,11 +255,7 @@ def add_auth(self, request):
         if self.credentials is None:
             raise NoCredentialsError()
 
-        # Use utcnow() because that's what gets mocked by tests, but set
-        # timezone because CRT assumes naive datetime is local time.
-        datetime_now = datetime.datetime.utcnow().replace(
-            tzinfo=datetime.timezone.utc
-        )
+        datetime_now = get_current_datetime(remove_tzinfo=False)
 
         # Use existing 'X-Amz-Content-SHA256' header if able
         existing_sha256 = self._get_existing_sha256(request)

awscli/botocore/endpoint.py

@@ -21,6 +21,7 @@
 
 from botocore import parsers
 from botocore.awsrequest import create_request_object
+from botocore.compat import get_current_datetime
 from botocore.exceptions import HTTPClientError
 from botocore.history import get_global_history_recorder
 from botocore.hooks import first_non_none_response
@@ -147,7 +148,7 @@ def prepare_request(self, request):
     def _calculate_ttl(
         self, response_received_timestamp, date_header, read_timeout
     ):
-        local_timestamp = datetime.datetime.utcnow()
+        local_timestamp = get_current_datetime()
         date_conversion = datetime.datetime.strptime(
             date_header, "%a, %d %b %Y %H:%M:%S %Z"
         )
@@ -164,7 +165,7 @@ def _set_ttl(self, retries_context, read_timeout, success_response):
         has_streaming_input = retries_context.get('has_streaming_input')
         if response_date_header and not has_streaming_input:
             try:
-                response_received_timestamp = datetime.datetime.utcnow()
+                response_received_timestamp = get_current_datetime()
                 retries_context['ttl'] = self._calculate_ttl(
                     response_received_timestamp,
                     response_date_header,

awscli/botocore/signers.py

@@ -18,7 +18,7 @@
 import botocore
 import botocore.auth
 from botocore.awsrequest import create_request_object, prepare_request_dict
-from botocore.compat import OrderedDict
+from botocore.compat import OrderedDict, get_current_datetime
 from botocore.exceptions import (
     ParamValidationError,
     UnknownClientMethodError,
@@ -717,7 +717,7 @@ def generate_presigned_post(
         policy = {}
 
         # Create an expiration date for the policy
-        datetime_now = datetime.datetime.utcnow()
+        datetime_now = get_current_datetime()
         expire_date = datetime_now + datetime.timedelta(seconds=expires_in)
         policy['expiration'] = expire_date.strftime(botocore.auth.ISO8601)
 

awscli/botocore/utils.py

@@ -42,6 +42,7 @@
 from awscrt.crypto import EC
 from botocore.compat import (
     MD5_AVAILABLE,
+    get_current_datetime,
     get_md5,
     get_tzinfo_options,
     json,
@@ -633,7 +634,7 @@ def _evaluate_expiration(self, credentials):
             refresh_interval_with_jitter = refresh_interval + random.randint(
                 120, 600
             )
-            current_time = datetime.datetime.utcnow()
+            current_time = get_current_datetime()
             refresh_offset = datetime.timedelta(
                 seconds=refresh_interval_with_jitter
             )

awscli/customizations/cloudformation/deployer.py

@@ -15,7 +15,7 @@
 import logging
 import sys
 import time
-from datetime import datetime
+from datetime import datetime, timezone
 
 import botocore
 
@@ -98,7 +98,7 @@ def create_changeset(
         :return:
         """
 
-        now = datetime.utcnow().isoformat()
+        now = datetime.now(timezone.utc).isoformat()
         description = f"Created by AWS CLI at {now} UTC"
 
         # Each changeset will get a unique name based on time

awscli/customizations/cloudtrail/validation.py

@@ -18,7 +18,7 @@
 import re
 import sys
 import zlib
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from zlib import error as ZLibError
 
 from awscrt.crypto import RSA, RSASignatureAlgorithm
@@ -570,7 +570,7 @@ def traverse_digests(self, start_date, end_date=None, is_backfill=False):
         :param is_backfill: Flag indicating whether to process backfill digests only.
         """
         if end_date is None:
-            end_date = datetime.utcnow()
+            end_date = datetime.now(timezone.utc)
         end_date = normalize_date(end_date)
         start_date = normalize_date(start_date)
         bucket = self.starting_bucket
@@ -1026,7 +1026,7 @@ def handle_args(self, args):
         if args.end_time:
             self.end_time = normalize_date(parse_date(args.end_time))
         else:
-            self.end_time = normalize_date(datetime.utcnow())
+            self.end_time = normalize_date(datetime.now(timezone.utc))
         if self.start_time > self.end_time:
             raise ValueError(
                 'Invalid time range specified: start-time must '

awscli/customizations/codecommit.py

@@ -159,7 +159,7 @@ def sign_request(self, region, url_to_sign):
         request = AWSRequest()
         request.url = url_to_sign
         request.method = 'GIT'
-        now = datetime.datetime.utcnow()
+        now = datetime.datetime.now(datetime.timezone.utc)
         request.context['timestamp'] = now.strftime('%Y%m%dT%H%M%S')
         split = urlsplit(request.url)
         # we don't want to include the port number in the signature

awscli/customizations/codedeploy/push.py

@@ -16,7 +16,7 @@
 import sys
 import tempfile
 import zipfile
-from datetime import datetime
+from datetime import datetime, timezone
 
 from botocore.exceptions import ClientError
 
@@ -131,7 +131,7 @@ def _validate_args(self, parsed_args):
             )
         if not parsed_args.description:
             parsed_args.description = (
-                f'Uploaded by AWS CLI {datetime.utcnow().isoformat()} UTC'
+                f'Uploaded by AWS CLI {datetime.now(timezone.utc).isoformat()} UTC'
             )
 
     def _push(self, params):

awscli/customizations/datapipeline/__init__.py

@@ -12,7 +12,7 @@
 # language governing permissions and limitations under the License.
 
 import json
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 
 from awscli.arguments import CustomArgument
 from awscli.customizations.commands import BasicCommand
@@ -197,7 +197,7 @@ class QueryArgBuilder:
 
     def __init__(self, current_time=None):
         if current_time is None:
-            current_time = datetime.utcnow()
+            current_time = datetime.now(timezone.utc)
         self.current_time = current_time
 
     def build_query(self, parsed_args):

awscli/customizations/ec2/bundleinstance.py

@@ -129,7 +129,7 @@ def _generate_policy(params):
     # Called if there is no policy supplied by the user.
     # Creates a policy that provides access for 24 hours.
     delta = datetime.timedelta(hours=24)
-    expires = datetime.datetime.utcnow() + delta
+    expires = datetime.datetime.now(datetime.timezone.utc) + delta
     expires_iso = expires.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
     policy = POLICY.format(
         expires=expires_iso, bucket=params['Bucket'], prefix=params['Prefix']

awscli/customizations/eks/get_token.py

@@ -14,7 +14,7 @@
 import json
 import os
 import sys
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 
 import botocore
 from botocore.model import ServiceId
@@ -105,7 +105,7 @@ class GetTokenCommand(BasicCommand):
     ]
 
     def get_expiration_time(self):
-        token_expiration = datetime.utcnow() + timedelta(
+        token_expiration = datetime.now(timezone.utc) + timedelta(
             minutes=TOKEN_EXPIRATION_MINS
         )
         return token_expiration.strftime('%Y-%m-%dT%H:%M:%SZ')

awscli/customizations/logs/tail.py

@@ -10,11 +10,12 @@
 # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
+import functools
 import json
 import re
 import time
 from collections import defaultdict
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 
 import colorama
 from botocore.utils import datetime2timestamp, parse_timestamp
@@ -266,7 +267,7 @@ class TimestampUtils:
     def __init__(self, now=None):
         self._now = now
         if now is None:
-            self._now = datetime.utcnow
+            self._now = functools.partial(datetime.now, timezone.utc)
 
     def to_epoch_millis(self, timestamp):
         re_match = self._RELATIVE_TIMESTAMP_REGEX.match(timestamp)

tests/functional/botocore/test_disable_s3_express_auth.py

@@ -32,7 +32,7 @@ class TestDisableS3ExpressAuth:
     def mock_datetime(self):
         with mock.patch('datetime.datetime', spec=True) as mock_dt:
             mock_dt.now.return_value = self.DATE
-            mock_dt.utcnow.return_value = self.DATE
+            mock_dt.now.return_value = self.DATE
             yield mock_dt
 
     def test_disable_s3_express_auth_enabled(

tests/functional/botocore/test_ec2.py

@@ -92,14 +92,14 @@ def setUp(self):
             '<snapshotId>%s</snapshotId>\n'
             '</CopySnapshotResponse>\n'
         )
-        self.now = datetime.datetime(2011, 9, 9, 23, 36)
+        self.now = datetime.datetime(2011, 9, 9, 23, 36, tzinfo=datetime.timezone.utc)
         self.datetime_patch = mock.patch.object(
             botocore.auth.datetime,
             'datetime',
             mock.Mock(wraps=datetime.datetime),
         )
         self.mocked_datetime = self.datetime_patch.start()
-        self.mocked_datetime.utcnow.return_value = self.now
+        self.mocked_datetime.now.return_value = self.now
 
     def tearDown(self):
         super().tearDown()

tests/functional/botocore/test_lex.py

@@ -10,7 +10,7 @@
 # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
-from datetime import datetime
+import datetime
 
 from tests import BaseSessionTest, ClientHTTPStubber, mock
 
@@ -31,10 +31,10 @@ def test_unsigned_payload(self):
             'inputStream': b'',
         }
 
-        timestamp = datetime(2017, 3, 22, 0, 0)
+        timestamp = datetime.datetime(2017, 3, 22, 0, 0, tzinfo=datetime.timezone.utc)
 
-        with mock.patch('botocore.auth.datetime') as _datetime:
-            _datetime.datetime.utcnow.return_value = timestamp
+        with mock.patch('botocore.auth.datetime.datetime') as _datetime:
+            _datetime.now.return_value = timestamp
             self.http_stubber.add_response(body=b'{}')
             with self.http_stubber:
                 self.client.post_content(**params)

tests/functional/botocore/test_retry.py

@@ -263,7 +263,7 @@ def _test_amz_sdk_request_header_with_test_case(
             mock.Mock(wraps=datetime.datetime),
         )
         mocked_datetime = datetime_patcher.start()
-        mocked_datetime.utcnow.side_effect = utcnow_side_effects
+        mocked_datetime.now.side_effect = utcnow_side_effects
 
         client = self.session.create_client(
             'dynamodb', self.region, config=client_config

tests/functional/botocore/test_s3.py

@@ -38,7 +38,7 @@
 )
 from tests.utils.botocore import get_checksum_cls
 
-DATE = datetime.datetime(2021, 8, 27, 0, 0, 0)
+DATE = datetime.datetime(2021, 8, 27, 0, 0, 0, tzinfo=datetime.timezone.utc)
 
 
 class TestS3BucketValidation(unittest.TestCase):