Add signed upload to azure and s3

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",

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
+        )

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,
         )

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
+        )

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())

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

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