diff --git a/storages/backends/azure_storage.py b/storages/backends/azure_storage.py index df27994c6..ce9c0e2c6 100644 --- a/storages/backends/azure_storage.py +++ b/storages/backends/azure_storage.py @@ -1,6 +1,6 @@ +import datetime import mimetypes import warnings -from datetime import datetime from datetime import timedelta from tempfile import SpooledTemporaryFile from urllib.parse import urlparse @@ -123,7 +123,7 @@ def __init__(self, **settings): self._service_client = None self._client = None self._user_delegation_key = None - self._user_delegation_key_expiry = datetime.utcnow() + self._user_delegation_key_expiry = datetime.datetime.now(datetime.timezone.utc) if self.connection_string and (not self.account_name or not self.account_key): parsed = parse_connection_string( self.connection_string, case_sensitive_keys=True @@ -212,7 +212,7 @@ def get_user_delegation_key(self, expiry): self._user_delegation_key is None or expiry > self._user_delegation_key_expiry ): - now = datetime.utcnow() + now = datetime.datetime.now(datetime.timezone.utc) key_expiry_time = now + timedelta(days=7) self._user_delegation_key = self.service_client.get_user_delegation_key( key_start_time=now, key_expiry_time=key_expiry_time @@ -291,7 +291,7 @@ def _save(self, name, content): def _expire_at(self, expire): # azure expects time in UTC - return datetime.utcnow() + timedelta(seconds=expire) + return datetime.datetime.now(datetime.timezone.utc) + timedelta(seconds=expire) def url(self, name, expire=None, parameters=None, mode="r"): name = self._get_valid_path(name) diff --git a/storages/backends/s3.py b/storages/backends/s3.py index fa2d39f16..252a93ab4 100644 --- a/storages/backends/s3.py +++ b/storages/backends/s3.py @@ -7,6 +7,7 @@ import warnings from datetime import datetime from datetime import timedelta +from datetime import timezone from urllib.parse import urlencode from django.contrib.staticfiles.storage import ManifestFilesMixin @@ -681,7 +682,7 @@ def url(self, name, parameters=None, expire=None, http_method=None): ) if self.querystring_auth and self.cloudfront_signer: - expiration = datetime.utcnow() + timedelta(seconds=expire) + expiration = datetime.now(timezone.utc) + timedelta(seconds=expire) return self.cloudfront_signer.generate_presigned_url( url, date_less_than=expiration ) diff --git a/tests/test_azure.py b/tests/test_azure.py index 059cb2eec..fb0fbaf1d 100644 --- a/tests/test_azure.py +++ b/tests/test_azure.py @@ -159,9 +159,9 @@ def test_url_expire(self, generate_blob_sas_mocked): datetime.datetime(2016, 11, 6, 4), datetime.timezone.utc ) - with mock.patch("storages.backends.azure_storage.datetime") as d_mocked: + with mock.patch("storages.backends.azure_storage.datetime") as datetime_mocked: # Implicit read permission - d_mocked.utcnow.return_value = fixed_time + datetime_mocked.datetime.now.return_value = fixed_time self.assertEqual( self.storage.url("some blob", 100), "https://ret_foo.blob.core.windows.net/test/some%20blob", @@ -179,7 +179,7 @@ def test_url_expire(self, generate_blob_sas_mocked): self.assertEqual(str(called_kwargs["permission"]), "r") # Explicit write permission - d_mocked.utcnow.return_value = fixed_time + datetime_mocked.datetime.now.return_value = fixed_time self.assertEqual( self.storage.url("some blob", expire=100, mode="w"), "https://ret_foo.blob.core.windows.net/test/some%20blob", @@ -218,8 +218,8 @@ def test_url_expire_user_delegation_key(self, generate_blob_sas_mocked): fixed_time = make_aware( datetime.datetime(2016, 11, 6, 4), datetime.timezone.utc ) - with mock.patch("storages.backends.azure_storage.datetime") as d_mocked: - d_mocked.utcnow.return_value = fixed_time + with mock.patch("storages.backends.azure_storage.datetime") as datetime_mocked: + datetime_mocked.datetime.now.return_value = fixed_time service_client.get_user_delegation_key.return_value = "user delegation key" self.assertEqual( self.storage.url("some blob", 100), @@ -435,3 +435,13 @@ def test_client_settings(self, bsc): bsc.assert_called_once_with( "https://test.blob.core.windows.net", credential=None, api_version="1.3" ) + + def test_user_delegation_key_expiry_is_aware(self): + """ + Ensure _user_delegation_key_expiry is a UTC-aware datetime. + """ + storage = azure_storage.AzureStorage() + self.assertIsNotNone(storage._user_delegation_key_expiry.tzinfo) + self.assertEqual( + storage._user_delegation_key_expiry.utcoffset().total_seconds(), 0 + ) diff --git a/tests/test_s3.py b/tests/test_s3.py index e324baf05..1e3312964 100644 --- a/tests/test_s3.py +++ b/tests/test_s3.py @@ -728,7 +728,7 @@ def test_url_protocol(self): self.assertEqual(storage.url_protocol, "https:") @mock.patch("storages.backends.s3.datetime") - def test_storage_url_custom_domain_signed_urls(self, dt): + def test_storage_url_custom_domain_signed_urls(self, datetime_mocked): key_id = "test-key" filename = "file.txt" pem = dedent( @@ -766,7 +766,9 @@ def test_storage_url_custom_domain_signed_urls(self, dt): self.assertEqual(self.storage.url(filename), url) self.storage.querystring_auth = True - dt.utcnow.return_value = datetime.datetime.utcfromtimestamp(0) + datetime_mocked.now.return_value = datetime.datetime.fromtimestamp( + 0, datetime.timezone.utc + ) self.assertEqual(self.storage.url(filename), signed_url) def test_generated_url_is_encoded(self):