34 lines
1.3 KiB
Python
34 lines
1.3 KiB
Python
|
"""Util to generate S3 authorization headers for object storage access control"""
|
||
|
|
|
||
|
|
from django.core.files.storage import default_storage
|
||
|
|
|
||
|
|
import botocore
|
||
|
|
|
||
|
|
|
||
|
|
def generate_s3_authorization_headers(key):
|
||
|
|
"""
|
||
|
|
Generate authorization headers for an s3 object.
|
||
|
|
These headers can be used as an alternative to signed urls with many benefits:
|
||
|
|
- the urls of our files never expire and can be stored in our documents' content
|
||
|
|
- we don't leak authorized urls that could be shared (file access can only be done
|
||
|
|
with cookies)
|
||
|
|
- access control is truly realtime
|
||
|
|
- the object storage service does not need to be exposed on internet
|
||
|
|
"""
|
||
|
|
url = default_storage.unsigned_connection.meta.client.generate_presigned_url(
|
||
|
|
"get_object",
|
||
|
|
ExpiresIn=0,
|
||
|
|
Params={"Bucket": default_storage.bucket_name, "Key": key},
|
||
|
|
)
|
||
|
|
request = botocore.awsrequest.AWSRequest(method="get", url=url)
|
||
|
|
|
||
|
|
s3_client = default_storage.connection.meta.client
|
||
|
|
# pylint: disable=protected-access
|
||
|
|
credentials = s3_client._request_signer._credentials # noqa: SLF001
|
||
|
|
frozen_credentials = credentials.get_frozen_credentials()
|
||
|
|
region = s3_client.meta.region_name
|
||
|
|
auth = botocore.auth.S3SigV4Auth(frozen_credentials, "s3", region)
|
||
|
|
auth.add_auth(request)
|
||
|
|
|
||
|
|
return request
|