signed upload method on azure and s3

Author: Polandia94

libcloud/storage/drivers/azure_blobs.py

@@ -20,6 +20,7 @@
 import hashlib
 import binascii
 from datetime import datetime, timedelta
+from typing import Literal
 
 from libcloud.utils.py3 import ET, b, httplib, tostring, urlquote, urlencode
 from libcloud.utils.xml import fixxpath
@@ -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/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
 
@@ -1274,7 +1274,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
 
@@ -1319,7 +1324,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,
         )