Closed andyp05 closed 3 months ago
Forgot to add the exception: ValueError("Either user_delegation_key or account_key must be provided.")
It seems once I switched to DefaultAzureCredential() and disabled Access Tokens in Azure, tag use of {{ imgfield.url }} stopped working and threw an exception. The workaround was to create a template tag to add the sas myself.
{{ f.doc.value|create_az_url }} <--works
{{ f.doc.url }} <-- stopped working
@tonybaloney do you have any ideas, you contributed a bunch of docs recently? In general, for Azure I'm relying on the community as I've never used it.
I think it's because the code assumes you use SAS tokens and if you have an expiry it'll try and write a SAS token back to the connection object. Also the get_user_delegation_key
doesn't return anything if you don't use a token credential.
https://github.com/jschneier/django-storages/blob/master/storages/backends/azure_storage.py#L319-L331 and https://github.com/jschneier/django-storages/blob/master/storages/backends/azure_storage.py#L220-L223
Both of these should be fixed.
I've got on my todo list this coming week to convert a Django template app to System Assigned Managed Identity using DefaultAzureCredential so I'll add this property (.url
) to the test as well and reproduce it so I can contribute a patch from a working demo.
@jschneier please feel free to tag me in any Azure issues like this.
@andyp05 hope you don't mind waiting a couple of days for a fix, but I can't see an obvious workaround without modifying the code.
Also @andyp05 just to check, what I'm assuming you're trying to do is:
Yes. I am using a managed identity in production and an App in development. Both use DefaultAzureCredential() to authenticate. I am having the same issues in prod and development. Creating a new blob works. Reading a blob using a form and trying to display an image in a template does not.
Generate a valid url with SAS when an image is pulled from a model using a Form. {{image.url}} used to work in a tag when using token authorization. It stopped working when I switched to DefaultAzureCredential().
I switched to DefaultAzureCredential (and Managed Identity) to turn off 'Allow storage account key access' to improve security.
Hope that is clear and helpful.
Can you share the full settings? That would help me verify we're checking the same scenario
Also, are your storage containers set to Private, Anonymous (Blob) or Anonymous (Container)?
Have you generated a SAS token and configured it in settings.py
?
Which role have you assigned to the Managed Identity?
User Delegation Key generation is a special role "Storage Blob Delegator" (db58b8e5-c6ad-4a2a-8342-4190687cbf4a
), but this control is also available in "Storage Blob Data Contributor"
I'm looking at the error you're getting and it seems to be because your Managed Identity doesn't have that role (that's the most likely cause)
I do have that role. Storing images to an ImageField works. The problem is when I try to view the image in a template using a form, the <img src="{{ f.image.url }}">
throws an exception. It never did.
I created a template tag to create the SAS and that works fine as a workaround. The SAS is generating fine, so it is not an authorization issue.
The relevant code looks like:
blob_service_client = BlobServiceClient(settings.AZURE_DOMAIN_FOR_SAS, credential=DefaultAzureCredential())
user_delegation_key = blob_service_client.get_user_delegation_key(
key_start_time=now,
key_expiry_time=now + datetime.timedelta(seconds=key_time_expiration + 60)
)
blob_sas = generate_blob_sas(account_name=settings.AZURE_ACCOUNT_NAME,
container_name=settings.AZURE_CONTAINER,
blob_name=filename,
user_delegation_key=user_delegation_key,
permission=BlobSasPermissions(read=True, create=create),
expiry=expire)
{{ f.doc.value | create_az_url }} <--works
{{ f.doc.url }} <-- stopped working
Ok, gotcha. I can't reproduce this but I'm going to configure a standalone template to try and reproduce it in another tenant.
Here's the demo app I've written to test it https://github.com/tonybaloney/django-storages/blob/demo_stire/demo/app/templates/home.html#L9
I did notice that setting an expiry value is really important to the code flow and forcing it down the path of fetching a user delegation key: https://github.com/tonybaloney/django-storages/blob/demo_stire/demo/demo/settings.py#L135
Here are all the storage account settings I've used https://github.com/tonybaloney/django-storages/blob/demo_stire/docs/backends/_resources/azure.bicep
I'm testing this locally by using az login
first then DefaultAzureCredential
will pick up the Azure CLI Extension credential session when running in a debugger like VS code.
If you spot anything else different in those settings lmk
In a sample template to allow user upload of an image, using the following:
<div class="col-md-9">{{ form.logo }}</div>
I see the input widget and if I add a file, it will upload correctly.
If I add the ".url" to display an existing image, it returns nothing.
<img src="{{ form.logo.url }}">
If in the view I construct the URL and SAS using the code snippet I showed above, it displays fine.
I did not have "expiration_secs": 3600
, in settings. I added it, but it did not fix the issue.
I've tried a template with an edit form, a custom model with an image field and I still can't reproduce this error with the .url
property. I tried creating a model with an ImageField and editing it via the admin portal and again via a custom form view.
I've got a full E2E demo app that's compatible with AZD (Azure Developer CLI) here that has managed identity and the same settings for the storage containers we discussed. Everything works fine.
https://github.com/tonybaloney/django-on-azure
Without seeing specifics of your setup I'm stuck trying to figure out a proper fix or documentation on how to avoid this. I'd also like to add a test to the library to make sure nobody else hits this issue.
I have office hours if you can email me at anthonyshaw
at microsoft
dot com
I think I found the issue. I am using a custom domain and have AZURE_CUSTOM_DOMAIN set in settings.py. If I remove this setting, the full url is created with the SAS token.
I found the existing issues opened on this. I had not put together the custom domain and the exception when I first reported the problem. As a workaround of the custom domain and proxy issue, I have implemented another settings variable called AZURE_URL_FOR_SAS for the purpose of getting the get_user_delegation_key and sas. However, I am returning the full URL with AZURE_CUSTOM_DOMAIN. This way, the user sees the custom domain and it is proxied by the CDN. Is this something that could be added to this package or am I missing something.
I found the existing issues opened on this. I had not put together the custom domain and the exception when I first reported the problem. As a workaround of the custom domain and proxy issue, I have implemented another settings variable called AZURE_URL_FOR_SAS for the purpose of getting the get_user_delegation_key and sas. However, I am returning the full URL with AZURE_CUSTOM_DOMAIN. This way, the user sees the custom domain and it is proxied by the CDN. Is this something that could be added to this package or am I missing something.
Are you using Azure Front Door, Azure CDN or something else that requires the custom domain
Was using Cloudflare as a proxy since all my domains are there. I tried turning off the proxy service (and with it all caching) and doing straight DNS and the problem remained.
I am using a custom domain for consistency of the user interface when a user clicks on an image, I want them to see my url. Another benefit is I get caching and DDOS protections,
I saw an issue where DNS providers convert HEAD calls to GET calls at the edge. Some say this is the problem. But turning off all Cloudflare performance and caching by not using the proxy should have disabled all conversions. Still did not work.
Could a solution be that in _get_service_client()
use the base url and for the actual https://.... url formation use the custom domain.
Two settings:
AZURE_SAS_DOMAIN
AZURE_CUSTOM_DOMAIN
If using a custom domain, both settings are required.
I am doing this manually using template tags and it seems to work fine. Thanks
We already have AZURE_ACCOUNT_NAME
and methods for getting clients that use that, surely there is no need for additional configuration AZURE_SAS_DOMAIN
?
I must be missing something. I have AZURE_ACCOUNT_NAME
setup with the storage account name. Everything worked fine until I setup a custom domain and defined AZURE_CUSTOM_DOMAIN
It seems the code recognizes the custom domain and uses it in the call to _get_service_client()
You mention you already have methods to use AZURE_ACCOUNT_NAME
. How do I get django_storages to use them.
get_user_delegation_key
uses the custom service client, isn't what you want that it simply doesn't do that?
Tested with a custom domain by deploying an Azure Frontdoor CDN profile:
STORAGES = {
"default": {
"BACKEND": "storages.backends.azure_storage.AzureStorage",
"OPTIONS": {
"token_credential": DefaultAzureCredential(),
"account_name": "xxxxx",
"azure_container": "media",
"expiration_secs": 3600,
"custom_domain": "antdjangocdntest-xxx.b02.azurefd.net"
}
},
"staticfiles": {
"BACKEND": "storages.backends.azure_storage.AzureStorage",
"OPTIONS": {
"token_credential": DefaultAzureCredential(),
"account_name": "xxxxx",
"azure_container": "static",
"expiration_secs": 3600,
"custom_domain": "antdjangocdntest-xxx.b02.azurefd.net"
},
},
}
Everything works as expected, including the .url
attributes on image fields and a custom upload. Still not able to reproduce this bug, I wonder if SAS domains detect Azure FD URLs/Azure CDN since those are the only configurations I've ever tried.
Hi folks did we ever figure out what is going on here?
Never got form.field.url
to work. Always returned blank.
I just generate a SAS myself and send the url in as a context field.
Using a custom domain:
>>>logo
<ImageFieldFile: contacts/ca-36d864c5.jpg>
>>>logo.file.name
'contacts/ca-36d864c5.jpg'
>>>logo.url
Unexpected return type <class 'str'> from ContentDecodePolicy.deserialize_from_http_generics.
Traceback (most recent call last):
File "C:\Program Files\JetBrains\PyCharm 241.14494.19\plugins\python\helpers-pro\pydevd_asyncio\pydevd_asyncio_utils.py", line 117, in _exec_async_code
result = func()
^^^^^^
File "<input>", line 1, in <module>
File "E:\Projects\VT\.venv\Lib\site-packages\django\db\models\fields\files.py", line 67, in url
return self.storage.url(self.name)
^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "E:\Projects\VT\.venv\Lib\site-packages\storages\backends\azure_storage.py", line 319, in url
user_delegation_key = self.get_user_delegation_key(expiry)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "E:\Projects\VT\.venv\Lib\site-packages\storages\backends\azure_storage.py", line 232, in get_user_delegation_key
self.custom_service_client.get_user_delegation_key(
File "E:\Projects\VT\.venv\Lib\site-packages\azure\core\tracing\decorator.py", line 78, in wrapper_use_tracer
return func(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^
File "E:\Projects\VT\.venv\Lib\site-packages\azure\storage\blob\_blob_service_client.py", line 230, in get_user_delegation_key
process_storage_error(error)
File "E:\Projects\VT\.venv\Lib\site-packages\azure\storage\blob\_shared\response_handlers.py", line 182, in process_storage_error
exec("raise error from None") # pylint: disable=exec-used # nosec
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "<string>", line 1, in <module>
azure.core.exceptions.HttpResponseError: <!DOCTYPE html><html lang="en-US"><head><title>Just a moment...</title><meta http-equiv="Content-Type" content="text/html; charset=UTF-8"><meta http-equiv="X-UA-Compatible" content="IE=Edge"><meta name="robots" content="noindex,nofollow"><meta name="viewport" content="width=device-width,initial-scale=1"><style>*{box-sizing:border-box;margin:0;padding:0}html{line-height:1.15;-webkit-text-size-adjust:100%;color:#313131}button,html{font-family:system-ui,-apple-system,BlinkMacSystemFont,Segoe UI,Roboto,Helvetica Neue,Arial,Noto Sans,sans-serif,Apple Color Emoji,Segoe UI Emoji,Segoe UI Symbol,Noto Color Emoji}@media (prefers-color-scheme:dark){body{background-color:#222;color:#d9d9d9}body a{color:#fff}body a:hover{color:#ee730a;text-decoration:underline}body .lds-ring div{border-color:#999 transparent transparent}body .font-red{color:#b20f03}body .pow-button{background-color:#4693ff;color:#1d1d1d}body #challenge-success-text{background-image:url(data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHdpZHRoPSIzMiIgaGVpZ2h0PSIzMiIgZmlsbD0ibm9uZSIgdmlld0JveD0iMCAwIDI2IDI2Ij48cGF0aCBmaWxsPSIjZDlkOWQ5IiBkPSJNMTMgMGExMyAxMyAwIDEgMCAwIDI2IDEzIDEzIDAgMCAwIDAtMjZtMCAyNGExMSAxMSAwIDEgMSAwLTIyIDExIDExIDAgMCAxIDAgMjIiLz48cGF0aCBmaWxsPSIjZDlkOWQ5IiBkPSJtMTAuOTU1IDE2LjA1NS0zLjk1LTQuMTI1LTEuNDQ1IDEuMzg1IDUuMzcgNS42MSA5LjQ5NS05LjYtMS40Mi0xLjQwNXoiLz48L3N2Zz4=)}body #challenge-error-text{background-image:url(data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHdpZHRoPSIzMiIgaGVpZ2h0PSIzMiIgZmlsbD0ibm9uZSI+PHBhdGggZmlsbD0iI0IyMEYwMyIgZD0iTTE2IDNhMTMgMTMgMCAxIDAgMTMgMTNBMTMuMDE1IDEzLjAxNSAwIDAgMCAxNiAzbTAgMjRhMTEgMTEgMCAxIDEgMTEtMTEgMTEuMDEgMTEuMDEgMCAwIDEtMTEgMTEiLz48cGF0aCBmaWxsPSIjQjIwRjAzIiBkPSJNMTcuMDM4IDE4LjYxNUgxNC44N0wxNC41NjMgOS41aDIuNzgzem0tMS4wODQgMS40MjdxLjY2IDAgMS4wNTcuMzg4LjQwNy4zODkuNDA3Ljk5NCAwIC41OTYtLjQwNy45ODQtLjM5Ny4zOS0xLjA1Ny4zODktLjY1IDAtMS4wNTYtLjM4OS0uMzk4LS4zODktLjM5OC0uOTg0IDAtLjU5Ny4zOTgtLjk4NS40MDYtLjM5NyAxLjA1Ni0uMzk3Ii8+PC9zdmc+)}}body{display:flex;flex-d
I think the problem may be on the azure side and not in this package since I cannot manually generate a SAS with the custom domain. If I generate with the standard blob.core.windows.net
domain it works fine. I can then replace the domain with the custom domain and that still works.
Just to add info: If I run a ModelSerializer using DRF on a model with an ImageField and AZURE_CUSTOM_DOMAIN is set, an exception is thrown. If I comment out that line in settings, it works fine.
The following issue states that a SAS token cannot be generated using a custom URL. It is an old issue, but my experience is that it is still true. (https://github.com/Azure/azure-storage-node/issues/437)
If this is the case, we would need to store both URLs. The 'normal' https://xxx.blob.core.windows.net/
and the custom domain.
It would be nice if this package could generate SAS token using the windows.net domain, get the token and append to custom domain. I am doing this manually now for normal use, but DRF is blowing up.
Thanks
per my earlier comments, we already have AZURE_ACCOUNT_NAME
from which the 'normal' url is derived.
So if what you are saying is right then simply update get_user_delegation_key
so that it uses the "service client" instead of the "custom service client"
per my earlier comments, we already have
AZURE_ACCOUNT_NAME
from which the 'normal' url is derived.So if what you are saying is right then simply update
get_user_delegation_key
so that it uses the "service client" instead of the "custom service client"
@andyp05 can you verify if this would fix the issue for you. i think the patch we're looking at is
diff --git a/storages/backends/azure_storage.py b/storages/backends/azure_storage.py
index 69b0213..c62c8f0 100644
--- a/storages/backends/azure_storage.py
+++ b/storages/backends/azure_storage.py
@@ -229,7 +229,7 @@ class AzureStorage(BaseStorage):
now = datetime.utcnow()
key_expiry_time = now + timedelta(days=7)
self._user_delegation_key = (
- self.custom_service_client.get_user_delegation_key(
+ self.service_client.get_user_delegation_key(
key_start_time=now, key_expiry_time=key_expiry_time
)
)
Took a very quick look at this and I think it solves the get_user_delegation_key() exception problem. However, the custom domain is not being returned in the url. I will look at it more closely over the weekend.
My quick thought is that the custom_domain is only needed for the formation of the url and not to get a custom client.
in def url()
...
if self.custom_domain:
parsed_url = urlparse(container_blob_url)
new_netloc = self.custom_domain
# Reconstruct the URL with the new domain
container_blob_url = urlunparse(parsed_url._replace(netloc=new_netloc))
Yes, makes sense. Please test out and patch / report back. I am aiming to produce a release on Monday.
I removed all references to custom client and added the code to reconstruct the url if a custom domain is set. This seems to resolve the issues I was experiencing, but I have not run unit tests on it. I am not that familiar with the code. I may have missed something big or small.
I can save a new image, pull it back out, and get the URL in a template all with the custom domain used in all user facing html.
import mimetypes
from datetime import datetime
from datetime import timedelta
from tempfile import SpooledTemporaryFile
from urllib.parse import urlparse, urlunparse
from azure.core.exceptions import ResourceNotFoundError
from azure.core.utils import parse_connection_string
from azure.storage.blob import BlobClient
from azure.storage.blob import BlobSasPermissions
from azure.storage.blob import BlobServiceClient
from azure.storage.blob import ContentSettings
from azure.storage.blob import generate_blob_sas
from django.core.exceptions import SuspiciousOperation
from django.core.files.base import File
from django.utils import timezone
from django.utils.deconstruct import deconstructible
from storages.base import BaseStorage
from storages.utils import clean_name
from storages.utils import get_available_overwrite_name
from storages.utils import safe_join
from storages.utils import setting
from storages.utils import to_bytes
@deconstructible
class AzureStorageFile(File):
def __init__(self, name, mode, storage):
self.name = name
self._mode = mode
self._storage = storage
self._is_dirty = False
self._file = None
self._path = storage._get_valid_path(name)
def _get_file(self):
if self._file is not None:
return self._file
file = SpooledTemporaryFile(
max_size=self._storage.max_memory_size,
suffix=".AzureStorageFile",
dir=setting("FILE_UPLOAD_TEMP_DIR", None),
)
if "r" in self._mode or "a" in self._mode:
download_stream = self._storage.client.download_blob(
self._path, timeout=self._storage.timeout
)
download_stream.readinto(file)
if "r" in self._mode:
file.seek(0)
self._file = file
return self._file
def _set_file(self, value):
self._file = value
file = property(_get_file, _set_file)
def read(self, *args, **kwargs):
if "r" not in self._mode and "a" not in self._mode:
raise AttributeError("File was not opened in read mode.")
return super().read(*args, **kwargs)
def write(self, content):
if "w" not in self._mode and "+" not in self._mode and "a" not in self._mode:
raise AttributeError("File was not opened in write mode.")
self._is_dirty = True
return super().write(to_bytes(content))
def close(self):
if self._file is None:
return
if self._is_dirty:
self._file.seek(0)
self._storage._save(self.name, self._file)
self._is_dirty = False
self._file.close()
self._file = None
def _content_type(content):
try:
return content.file.content_type
except AttributeError:
pass
try:
return content.content_type
except AttributeError:
pass
return None
def _get_valid_path(s):
# A blob name:
# * must not end with dot or slash
# * can contain any character
# * must escape URL reserved characters
# (not needed here since the azure client will do that)
s = s.strip("./")
if len(s) > _AZURE_NAME_MAX_LEN:
raise ValueError("File name max len is %d" % _AZURE_NAME_MAX_LEN)
if not len(s):
raise ValueError("File name must contain one or more printable characters")
if s.count("/") > 256:
raise ValueError("File name must not contain more than 256 slashes")
return s
# Max len according to azure's docs
_AZURE_NAME_MAX_LEN = 1024
@deconstructible
class AzureStorage(BaseStorage):
def __init__(self, **settings):
super().__init__(**settings)
self._service_client = None
self._client = None
self._user_delegation_key = None
self._user_delegation_key_expiry = datetime.utcnow()
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
)
if not self.account_name and "AccountName" in parsed:
self.account_name = parsed["AccountName"]
if not self.account_key and "AccountKey" in parsed:
self.account_key = parsed["AccountKey"]
def get_default_settings(self):
return {
"account_name": setting("AZURE_ACCOUNT_NAME"),
"account_key": setting("AZURE_ACCOUNT_KEY"),
"object_parameters": setting("AZURE_OBJECT_PARAMETERS", {}),
"azure_container": setting("AZURE_CONTAINER"),
"azure_ssl": setting("AZURE_SSL", True),
"upload_max_conn": setting("AZURE_UPLOAD_MAX_CONN", 2),
"timeout": setting("AZURE_CONNECTION_TIMEOUT_SECS", 20),
"max_memory_size": setting("AZURE_BLOB_MAX_MEMORY_SIZE", 2 * 1024 * 1024),
"expiration_secs": setting("AZURE_URL_EXPIRATION_SECS"),
"overwrite_files": setting("AZURE_OVERWRITE_FILES", False),
"location": setting("AZURE_LOCATION", ""),
"default_content_type": "application/octet-stream",
"cache_control": setting("AZURE_CACHE_CONTROL"),
"sas_token": setting("AZURE_SAS_TOKEN"),
"endpoint_suffix": setting("AZURE_ENDPOINT_SUFFIX", "core.windows.net"),
"custom_domain": setting("AZURE_CUSTOM_DOMAIN"),
"connection_string": setting("AZURE_CONNECTION_STRING"),
"token_credential": setting("AZURE_TOKEN_CREDENTIAL"),
"api_version": setting("AZURE_API_VERSION", None),
}
def _get_service_client(self):
if self.connection_string is not None:
return BlobServiceClient.from_connection_string(self.connection_string)
account_domain = (
"{}.blob.{}".format(
self.account_name,
self.endpoint_suffix,
)
)
account_url = "{}://{}".format(self.azure_protocol, account_domain)
credential = None
if self.account_key:
credential = {
"account_name": self.account_name,
"account_key": self.account_key,
}
elif self.sas_token:
credential = self.sas_token
elif self.token_credential:
credential = self.token_credential
options = {}
if self.api_version:
options["api_version"] = self.api_version
return BlobServiceClient(account_url, credential=credential, **options)
@property
def service_client(self):
if self._service_client is None:
self._service_client = self._get_service_client()
return self._service_client
@property
def client(self):
print('client')
if self._client is None:
self._client = self.service_client.get_container_client(
self.azure_container
)
return self._client
def get_user_delegation_key(self, expiry):
# We'll only be able to get a user delegation key if we've authenticated with a
# token credential.
if self.token_credential is None:
return None
# Get a new key if we don't already have one, or if the one we have expires too
# soon.
if (
self._user_delegation_key is None
or expiry > self._user_delegation_key_expiry
):
now = datetime.utcnow()
key_expiry_time = now + timedelta(days=7)
self._user_delegation_key = (
self.service_client.get_user_delegation_key(
# self.custom_service_client.get_user_delegation_key(
key_start_time=now, key_expiry_time=key_expiry_time
)
)
self._user_delegation_key_expiry = key_expiry_time
return self._user_delegation_key
@property
def azure_protocol(self):
if self.azure_ssl:
return "https"
else:
return "http"
def _normalize_name(self, name):
try:
return safe_join(self.location, name)
except ValueError:
raise SuspiciousOperation("Attempted access to '%s' denied." % name)
def _get_valid_path(self, name):
# Must be idempotent
return _get_valid_path(self._normalize_name(clean_name(name)))
def _open(self, name, mode="rb"):
return AzureStorageFile(name, mode, self)
def get_available_name(self, name, max_length=_AZURE_NAME_MAX_LEN):
"""
Returns a filename that's free on the target storage system, and
available for new content to be written to.
"""
name = clean_name(name)
if self.overwrite_files:
return get_available_overwrite_name(name, max_length)
return super().get_available_name(name, max_length)
def exists(self, name):
blob_client = self.client.get_blob_client(self._get_valid_path(name))
return blob_client.exists()
def delete(self, name):
try:
self.client.delete_blob(self._get_valid_path(name), timeout=self.timeout)
except ResourceNotFoundError:
pass
def size(self, name):
blob_client = self.client.get_blob_client(self._get_valid_path(name))
properties = blob_client.get_blob_properties(timeout=self.timeout)
return properties.size
def _save(self, name, content):
cleaned_name = clean_name(name)
name = self._get_valid_path(name)
params = self._get_content_settings_parameters(name, content)
# Unwrap django file (wrapped by parent's save call)
if isinstance(content, File):
content = content.file
content.seek(0)
self.client.upload_blob(
name,
content,
content_settings=ContentSettings(**params),
max_concurrency=self.upload_max_conn,
timeout=self.timeout,
overwrite=self.overwrite_files,
)
return cleaned_name
def _expire_at(self, expire):
# azure expects time in UTC
return datetime.utcnow() + timedelta(seconds=expire)
def url(self, name, expire=None, parameters=None):
name = self._get_valid_path(name)
params = parameters or {}
if expire is None:
expire = self.expiration_secs
credential = None
if expire:
expiry = self._expire_at(expire)
user_delegation_key = self.get_user_delegation_key(expiry)
sas_token = generate_blob_sas(
self.account_name,
self.azure_container,
name,
account_key=self.account_key,
user_delegation_key=user_delegation_key,
permission=BlobSasPermissions(read=True),
expiry=expiry,
**params,
)
credential = sas_token
container_blob_url = self.client.get_blob_client(name).url
if self.custom_domain:
# if custom domain, replace the account name with the custom domain
parsed_url = urlparse(container_blob_url)
new_netloc = self.custom_domain
# Reconstruct the URL with the new domain
container_blob_url = urlunparse(parsed_url._replace(netloc=new_netloc))
return BlobClient.from_blob_url(container_blob_url, credential=credential).url
def _get_content_settings_parameters(self, name, content=None):
params = {}
guessed_type, content_encoding = mimetypes.guess_type(name)
content_type = (
_content_type(content) or guessed_type or self.default_content_type
)
params["cache_control"] = self.cache_control
params["content_type"] = content_type
params["content_encoding"] = content_encoding
params.update(self.get_object_parameters(name))
return params
def get_object_parameters(self, name):
"""
Returns a dictionary that is passed to content settings. Override this
method to adjust this on a per-object basis to set e.g ContentDisposition.
By default, returns the value of AZURE_OBJECT_PARAMETERS.
"""
return self.object_parameters.copy()
def get_modified_time(self, name):
"""
Returns an (aware) datetime object containing the last modified time if
USE_TZ is True, otherwise returns a naive datetime in the local timezone.
"""
blob_client = self.client.get_blob_client(self._get_valid_path(name))
properties = blob_client.get_blob_properties(timeout=self.timeout)
if not setting("USE_TZ", False):
return timezone.make_naive(properties.last_modified)
tz = timezone.get_current_timezone()
if timezone.is_naive(properties.last_modified):
return timezone.make_aware(properties.last_modified, tz)
# `last_modified` is in UTC time_zone, we
# must convert it to settings time_zone
return properties.last_modified.astimezone(tz)
def list_all(self, path=""):
"""Return all files for a given path"""
if path:
path = self._get_valid_path(path)
if path and not path.endswith("/"):
path += "/"
# XXX make generator, add start, end
return [
blob.name
for blob in self.client.list_blobs(
name_starts_with=path, timeout=self.timeout
)
]
def listdir(self, path=""):
"""
Return directories and files for a given path.
Leave the path empty to list the root.
Order of dirs and files is undefined.
"""
files = []
dirs = set()
for name in self.list_all(path):
n = name[len(path) :]
if "/" in n:
dirs.add(n.split("/", 1)[0])
else:
files.append(n)
return list(dirs), files
@andyp05 thanks I've cleaned up and fixed the test suite from the above in #1418, can you please verify it looks okay from your end.
Ran some tests. Adding, editing, and deleting blobs seems to work fine. All links use the custom domain when displaying images in a template. No errors encountered. Thanks.
If you are putting out a new release, You may want to replace utcnow() which is depreciated. Also, the datetime returned does not have a timezone set so it is naive.
The code is already importing from django.utils import timezone
So maybe something like from datetime import timezone as dttz
and datetime.now(dttz.utc)
.
Running Django 5.02, python 3.12.2 on Windows Server and IIS (hypercorn) and django-storages 1.14.2.
Have managed identity setup and working fine using DefaultAzureCredential() to get secrets in settings.py Azure Storage using DefaultAzureCredential() manually to upload files not using django models is working fine.
However, django-storages is throwing an exception in get_user_delegation_key() when reading an ImageField from a model.
AZURE_CLIENT_ID environment variable is set to client id of managed identity. Also tried putting this in call to DefaultAzureCredential()
In settings.py:
The following IS working for Key Vault access:
It is not an DefaultAzureCredential(), Access Control, or Firewall issue since I can upload and read files from the app not using django models from another section of my code. Code snippet that works:
Any ideas would be appreciated.
Thanks