[v2] Port/Update usage of datetime.utcnow for DeprecationWarnings

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/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/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).replace(tzinfo=None) + 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).replace(tzinfo=None) + timedelta(
             minutes=TOKEN_EXPIRATION_MINS
         )
         return token_expiration.strftime('%Y-%m-%dT%H:%M:%SZ')

awscli/customizations/logs/tail.py

@@ -14,7 +14,7 @@
 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
@@ -251,6 +251,11 @@ def _should_use_color(self, parsed_globals):
         return is_a_tty()
 
 
+def _utcnow():
+    """Return the current UTC time as a timezone-aware datetime."""
+    return datetime.now(timezone.utc)
+
+
 class TimestampUtils:
     _RELATIVE_TIMESTAMP_REGEX = re.compile(
         r"(?P<amount>\d+)(?P<unit>s|m|h|d|w)$"
@@ -266,7 +271,7 @@ class TimestampUtils:
     def __init__(self, now=None):
         self._now = now
         if now is None:
-            self._now = datetime.utcnow
+            self._now = _utcnow
 
     def to_epoch_millis(self, timestamp):
         re_match = self._RELATIVE_TIMESTAMP_REGEX.match(timestamp)

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
+            _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_s3express.py

@@ -36,7 +36,7 @@
 S3EXPRESS_BUCKET = "mytestbucket--usw2-az5--x-s3"
 
 
-DATE = datetime.datetime(2023, 11, 26, 0, 0, 0, tzinfo=tzutc())
+DATE = datetime.datetime(2023, 11, 26, 0, 0, 0, tzinfo=datetime.timezone.utc)
 
 
 CREATE_SESSION_RESPONSE = (
@@ -108,7 +108,6 @@ def test_s3_express_auth_headers(self):
 class TestS3ExpressIdentityCache:
     def test_default_s3_express_cache(self, default_s3_client, mock_datetime):
         mock_datetime.now.return_value = DATE
-        mock_datetime.utcnow.return_value = DATE
 
         identity_cache = S3ExpressIdentityCache(
             default_s3_client,
@@ -126,7 +125,6 @@ def test_s3_express_cache_one_network_call(
         self, default_s3_client, mock_datetime
     ):
         mock_datetime.now.return_value = DATE
-        mock_datetime.utcnow.return_value = DATE
         bucket = 'my_bucket'
 
         identity_cache = S3ExpressIdentityCache(
@@ -151,7 +149,6 @@ def test_s3_express_cache_multiple_buckets(
         self, default_s3_client, mock_datetime
     ):
         mock_datetime.now.return_value = DATE
-        mock_datetime.utcnow.return_value = DATE
         bucket = 'my_bucket'
         other_bucket = 'other_bucket'
 
@@ -204,7 +201,7 @@ def _call_get_object(self, client):
         )
 
     def test_create_bucket(self, default_s3_client, mock_datetime):
-        mock_datetime.utcnow.return_value = DATE
+        mock_datetime.now.return_value = DATE
 
         with ClientHTTPStubber(default_s3_client) as stubber:
             stubber.add_response()
@@ -228,7 +225,6 @@ def test_create_bucket(self, default_s3_client, mock_datetime):
         self._assert_standard_sigv4_signature(stubber.requests[0].headers)
 
     def test_get_object(self, default_s3_client, mock_datetime):
-        mock_datetime.utcnow.return_value = DATE
         mock_datetime.now.return_value = DATE
 
         with ClientHTTPStubber(default_s3_client) as stubber:
@@ -250,7 +246,6 @@ def test_get_object(self, default_s3_client, mock_datetime):
     def test_cache_with_multiple_requests(
         self, default_s3_client, mock_datetime
     ):
-        mock_datetime.utcnow.return_value = DATE
         mock_datetime.now.return_value = DATE
 
         with ClientHTTPStubber(default_s3_client) as stubber:
@@ -275,7 +270,6 @@ def test_cache_with_multiple_requests(
     def test_delete_objects_injects_correct_checksum(
         self, default_s3_client, mock_datetime
     ):
-        mock_datetime.utcnow.return_value = DATE
         mock_datetime.now.return_value = DATE
 
         with ClientHTTPStubber(default_s3_client) as stubber:

tests/functional/botocore/test_sts.py

@@ -11,7 +11,7 @@
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
 import re
-from datetime import datetime
+import datetime
 
 from botocore.config import Config
 from botocore.stub import Stubber
@@ -32,9 +32,9 @@ def setUp(self):
         self.stubber.activate()
 
     def test_presigned_url_contains_no_content_type(self):
-        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
+            _datetime.datetime.now.return_value = timestamp
             url = self.client.generate_presigned_url('get_caller_identity', {})
 
         # There should be no 'content-type' in x-amz-signedheaders

tests/functional/dsql/test_generate_db_auth_token.py

@@ -50,7 +50,7 @@ def test_generate_simple_token(self):
         clock = datetime.datetime(2024, 11, 7, 17, 39, 33, tzinfo=tzutc())
 
         with mock.patch('datetime.datetime') as dt:
-            dt.utcnow.return_value = clock
+            dt.now.return_value = clock
             stdout, _, _ = self.run_cmd(command, expected_rc=0)
 
         # Expected hashes are always the same as session variables come from the BaseAwsCommandParamsTest class
@@ -79,7 +79,7 @@ def test_missing_hostname_raises_exception(self):
         clock = datetime.datetime(2024, 11, 7, 17, 39, 33, tzinfo=tzutc())
 
         with mock.patch('datetime.datetime') as dt:
-            dt.utcnow.return_value = clock
+            dt.now.return_value = clock
             stdout, _, _ = self.run_cmd(command, expected_rc=252)
 
 
@@ -92,7 +92,7 @@ def test_generate_simple_token(self):
         clock = datetime.datetime(2024, 11, 7, 17, 39, 33, tzinfo=tzutc())
 
         with mock.patch('datetime.datetime') as dt:
-            dt.utcnow.return_value = clock
+            dt.now.return_value = clock
             stdout, _, _ = self.run_cmd(command, expected_rc=0)
 
         # Expected hashes are always the same as session variables come from the BaseAwsCommandParamsTest class

tests/functional/ec2/test_bundle_instance.py

@@ -42,7 +42,7 @@ def setUp(self):
             mock.Mock(wraps=datetime.datetime),
         )
         mocked_datetime = self.datetime_patcher.start()
-        mocked_datetime.utcnow.return_value = datetime.datetime(2013, 8, 9)
+        mocked_datetime.now.return_value = datetime.datetime(2013, 8, 9)
 
     def tearDown(self):
         super(TestBundleInstance, self).tearDown()

tests/functional/ec2instanceconnect/test_opentunnel.py

@@ -409,7 +409,7 @@ def request_params_for_describe_eice():
 def datetime_utcnow_patch():
     clock = datetime.datetime(2020, 1, 1, 1, 1, 1, tzinfo=tzutc())
     with mock.patch('datetime.datetime') as dt:
-        dt.utcnow.return_value = clock
+        dt.now.return_value = clock
         yield dt
 
 

tests/functional/eks/test_get_token.py

@@ -78,7 +78,7 @@ def set_kubernetes_exec_info(self, api_version):
 
     @mock.patch('awscli.customizations.eks.get_token.datetime')
     def test_get_token(self, mock_datetime):
-        mock_datetime.utcnow.return_value = datetime(2019, 10, 23, 23, 0, 0, 0)
+        mock_datetime.now.return_value = datetime(2019, 10, 23, 23, 0, 0, 0)
         cmd = 'eks get-token --cluster-name %s' % self.cluster_name
         response = self.run_get_token(cmd)
         self.assertEqual(
@@ -96,7 +96,7 @@ def test_get_token(self, mock_datetime):
 
     @mock.patch('awscli.customizations.eks.get_token.datetime')
     def test_query_nested_object(self, mock_datetime):
-        mock_datetime.utcnow.return_value = datetime(2019, 10, 23, 23, 0, 0, 0)
+        mock_datetime.now.return_value = datetime(2019, 10, 23, 23, 0, 0, 0)
         cmd = 'eks get-token --cluster-name %s' % self.cluster_name
         cmd += ' --query status'
         response = self.run_get_token(cmd)
@@ -119,7 +119,7 @@ def test_query_value(self):
 
     @mock.patch('awscli.customizations.eks.get_token.datetime')
     def test_output_text(self, mock_datetime):
-        mock_datetime.utcnow.return_value = datetime(2019, 10, 23, 23, 0, 0, 0)
+        mock_datetime.now.return_value = datetime(2019, 10, 23, 23, 0, 0, 0)
         cmd = 'eks get-token --cluster-name %s' % self.cluster_name
         cmd += ' --output text'
         stdout, _, _ = self.run_cmd(cmd)
@@ -129,7 +129,7 @@ def test_output_text(self, mock_datetime):
 
     @mock.patch('awscli.customizations.eks.get_token.datetime')
     def test_output_table(self, mock_datetime):
-        mock_datetime.utcnow.return_value = datetime(2019, 10, 23, 23, 0, 0, 0)
+        mock_datetime.now.return_value = datetime(2019, 10, 23, 23, 0, 0, 0)
         cmd = 'eks get-token --cluster-name %s' % self.cluster_name
         cmd += ' --output table'
         stdout, _, _ = self.run_cmd(cmd)