diff --git a/libcloud/storage/drivers/azure_blobs.py b/libcloud/storage/drivers/azure_blobs.py index d7e14f1db2..f56b21b4ef 100644 --- a/libcloud/storage/drivers/azure_blobs.py +++ b/libcloud/storage/drivers/azure_blobs.py @@ -19,6 +19,7 @@ import base64 import hashlib import binascii +from typing import Literal from datetime import datetime, timedelta from libcloud.utils.py3 import ET, b, httplib, tostring, urlquote, urlencode @@ -603,7 +604,12 @@ def get_object(self, container_name, object_name): raise ObjectDoesNotExistError(value=None, driver=self, object_name=object_name) - def get_object_cdn_url(self, obj, ex_expiry=AZURE_STORAGE_CDN_URL_EXPIRY_HOURS): + def get_object_cdn_url( + self, + obj, + ex_expiry=AZURE_STORAGE_CDN_URL_EXPIRY_HOURS, + ex_method: Literal["GET", "PUT", "DELETE"] = "GET", + ): """ Return a SAS URL that enables reading the given object. @@ -623,10 +629,17 @@ def get_object_cdn_url(self, obj, ex_expiry=AZURE_STORAGE_CDN_URL_EXPIRY_HOURS): start = now - timedelta(minutes=AZURE_STORAGE_CDN_URL_START_MINUTES) expiry = now + timedelta(hours=ex_expiry) + if ex_method == "PUT": + sp = "wc" + elif ex_method == "DELETE": + sp = "d" + else: + sp = "r" + params = { "st": start.strftime(AZURE_STORAGE_CDN_URL_DATE_FORMAT), "se": expiry.strftime(AZURE_STORAGE_CDN_URL_DATE_FORMAT), - "sp": "r", + "sp": sp, "spr": "https" if self.secure else "http,https", "sv": self.connectionCls.API_VERSION, "sr": "b", diff --git a/libcloud/storage/drivers/ovh.py b/libcloud/storage/drivers/ovh.py index fc6dd63a87..98397dd7db 100644 --- a/libcloud/storage/drivers/ovh.py +++ b/libcloud/storage/drivers/ovh.py @@ -83,7 +83,9 @@ def __init__( def list_regions(self): return REGION_TO_HOST_MAP.keys() - def get_object_cdn_url(self, obj, ex_expiry=S3_CDN_URL_EXPIRY_HOURS): + def get_object_cdn_url(self, obj, ex_expiry=S3_CDN_URL_EXPIRY_HOURS, ex_method: str = "GET"): # In order to download (private) objects we need to be able to generate a valid CDN URL, # hence shamefully just use the working code from the S3StorageDriver. - return S3StorageDriver.get_object_cdn_url(self, obj, ex_expiry=ex_expiry) + return S3StorageDriver.get_object_cdn_url( + self, obj, ex_expiry=ex_expiry, ex_method=ex_method + ) diff --git a/libcloud/storage/drivers/s3.py b/libcloud/storage/drivers/s3.py index fe1beb2996..eda9c7b9f7 100644 --- a/libcloud/storage/drivers/s3.py +++ b/libcloud/storage/drivers/s3.py @@ -17,7 +17,7 @@ import hmac import time import base64 -from typing import Dict, Optional +from typing import Dict, Literal, Optional from hashlib import sha1 from datetime import datetime @@ -1275,7 +1275,12 @@ def __init__( def list_regions(self): return REGION_TO_HOST_MAP.keys() - def get_object_cdn_url(self, obj, ex_expiry=S3_CDN_URL_EXPIRY_HOURS): + def get_object_cdn_url( + self, + obj, + ex_expiry=S3_CDN_URL_EXPIRY_HOURS, + ex_method: Literal["GET", "PUT", "DELETE"] = "GET", + ): """ Return a "presigned URL" for read-only access to object @@ -1320,7 +1325,7 @@ def get_object_cdn_url(self, obj, ex_expiry=S3_CDN_URL_EXPIRY_HOURS): params=params_to_sign, headers=headers_to_sign, dt=now, - method="GET", + method=ex_method, path=object_path, data=UnsignedPayloadSentinel, ) diff --git a/libcloud/storage/drivers/scaleway.py b/libcloud/storage/drivers/scaleway.py index a599d0c21b..fdadbe9b2b 100644 --- a/libcloud/storage/drivers/scaleway.py +++ b/libcloud/storage/drivers/scaleway.py @@ -85,7 +85,9 @@ def __init__( def list_regions(self): return REGION_TO_HOST_MAP.keys() - def get_object_cdn_url(self, obj, ex_expiry=S3_CDN_URL_EXPIRY_HOURS): + def get_object_cdn_url(self, obj, ex_expiry=S3_CDN_URL_EXPIRY_HOURS, ex_method: str = "GET"): # In order to download (private) objects we need to be able to generate a valid CDN URL, # hence shamefully just use the working code from the S3StorageDriver. - return S3StorageDriver.get_object_cdn_url(self, obj, ex_expiry=ex_expiry) + return S3StorageDriver.get_object_cdn_url( + self, obj, ex_expiry=ex_expiry, ex_method=ex_method + ) diff --git a/libcloud/test/storage/test_aurora.py b/libcloud/test/storage/test_aurora.py index 682d747ec7..14e8a5a013 100644 --- a/libcloud/test/storage/test_aurora.py +++ b/libcloud/test/storage/test_aurora.py @@ -38,6 +38,13 @@ def test_get_object_cdn_url(self): with self.assertRaises(LibcloudError): self.driver.get_object_cdn_url(obj) + def test_get_object_cdn_url_put(self): + self.mock_response_klass.type = "get_object" + obj = self.driver.get_object(container_name="test2", object_name="test") + + with self.assertRaises(LibcloudError): + self.driver.get_object_cdn_url(obj) + if __name__ == "__main__": sys.exit(unittest.main()) diff --git a/libcloud/test/storage/test_azure_blobs.py b/libcloud/test/storage/test_azure_blobs.py index a67e7dd79d..cf8c06614d 100644 --- a/libcloud/test/storage/test_azure_blobs.py +++ b/libcloud/test/storage/test_azure_blobs.py @@ -511,6 +511,15 @@ def test_get_object_cdn_url(self): self.assertEqual(len(query["sig"]), 1) self.assertGreater(len(query["sig"][0]), 0) + def test_get_object_cdn_url_put(self): + obj = self.driver.get_object(container_name="test_container200", object_name="test") + + url = urlparse.urlparse(self.driver.get_object_cdn_url(obj, ex_method="PUT")) + query = urlparse.parse_qs(url.query) + + self.assertEqual(len(query["sig"]), 1) + self.assertGreater(len(query["sig"][0]), 0) + def test_get_object_container_doesnt_exist(self): # This method makes two requests which makes mocking the response a bit # trickier diff --git a/libcloud/test/storage/test_s3.py b/libcloud/test/storage/test_s3.py index d844da31e9..be70898c84 100644 --- a/libcloud/test/storage/test_s3.py +++ b/libcloud/test/storage/test_s3.py @@ -530,6 +530,23 @@ def test_get_object_cdn_url(self): with self.assertRaises(NotImplementedError): self.driver.get_object_cdn_url(obj) + def test_get_object_cdn_url_put(self): + self.mock_response_klass.type = "get_object" + obj = self.driver.get_object(container_name="test2", object_name="test") + + # cdn urls can only be generated using a V4 connection + if issubclass(self.driver.connectionCls, S3SignatureV4Connection): + cdn_url = self.driver.get_object_cdn_url(obj, ex_method="PUT", ex_expiry=12) + url = urlparse.urlparse(cdn_url) + query = urlparse.parse_qs(url.query) + + self.assertEqual(len(query["X-Amz-Signature"]), 1) + self.assertGreater(len(query["X-Amz-Signature"][0]), 0) + self.assertEqual(query["X-Amz-Expires"], ["43200"]) + else: + with self.assertRaises(NotImplementedError): + self.driver.get_object_cdn_url(obj) + def test_get_object_container_doesnt_exist(self): # This method makes two requests which makes mocking the response a bit # trickier