[SnapshotSAS]Add Snapshot SAS

sdk/storage/azure-storage-blob/azure/storage/blob/_shared/constants.py

@@ -7,7 +7,7 @@
 import sys
 
 
-X_MS_VERSION = '2018-03-28'
+X_MS_VERSION = '2018-06-17'
 
 # Socket timeout in seconds
 DEFAULT_SOCKET_TIMEOUT = 20

sdk/storage/azure-storage-blob/azure/storage/blob/blob_client.py

@@ -21,8 +21,8 @@
 
 from ._shared import encode_base64
 from ._shared.base_client import StorageAccountHostsMixin, parse_connection_str, parse_query
-from ._shared.shared_access_signature import BlobSharedAccessSignature
 from ._shared.encryption import generate_blob_encryption_data
+from ._shared.shared_access_signature import ResourceSharedAccessSignature
 from ._shared.uploads import IterStreamer
 from ._shared.downloads import StorageStreamDownloader
 from ._shared.request_handlers import (
@@ -294,12 +294,12 @@ def generate_shared_access_signature(
         """
         if not hasattr(self.credential, 'account_key') or not self.credential.account_key:
             raise ValueError("No account SAS key available.")
-        sas = BlobSharedAccessSignature(self.credential.account_name, self.credential.account_key)
+        sas = ResourceSharedAccessSignature(self.credential.account_name, self.credential.account_key)
         return sas.generate_blob(
             self.container_name,
             self.blob_name,
-            permission,
-            expiry,
+            permission=permission,
+            expiry=expiry,
             start=start,
             policy_id=policy_id,
             ip=ip,

sdk/storage/azure-storage-blob/azure/storage/blob/container_client.py

@@ -21,7 +21,7 @@
 
 import six
 
-from ._shared.shared_access_signature import BlobSharedAccessSignature
+from ._shared.shared_access_signature import ResourceSharedAccessSignature
 from ._shared.base_client import StorageAccountHostsMixin, parse_connection_str, parse_query
 from ._shared.request_handlers import add_metadata_headers, serialize_iso
 from ._shared.response_handlers import (
@@ -261,7 +261,7 @@ def generate_shared_access_signature(
         """
         if not hasattr(self.credential, 'account_key') and not self.credential.account_key:
             raise ValueError("No account SAS key available.")
-        sas = BlobSharedAccessSignature(self.credential.account_name, self.credential.account_key)
+        sas = ResourceSharedAccessSignature(self.credential.account_name, self.credential.account_key)
         return sas.generate_container(
             self.container_name,
             permission=permission,

sdk/storage/azure-storage-blob/tests/test_common_blob.py

@@ -1175,6 +1175,37 @@ def test_sas_access_blob(self):
         # Assert
         self.assertEqual(self.byte_data, content)
 
+    def test_sas_access_blob_snapshot(self):
+        # SAS URL is calculated from storage key, so this test runs live only
+        if TestMode.need_recording_file(self.test_mode):
+            return
+
+        # Arrange
+        blob_name = self._create_block_blob()
+        blob_client = self.bsc.get_blob_client(self.container_name, blob_name)
+        blob_snapshot = blob_client.create_snapshot()
+        blob_snapshot_client = self.bsc.get_blob_client(self.container_name, blob_name, snapshot=blob_snapshot)
+
+        token = blob_snapshot_client.generate_shared_access_signature(
+            permission=BlobPermissions.READ + BlobPermissions.DELETE,
+            expiry=datetime.utcnow() + timedelta(hours=1),
+        )
+
+        service = BlobClient(blob_snapshot_client.url, credential=token)
+
+        # Act
+        snapshot_content = service.download_blob().content_as_bytes()
+
+        # Assert
+        self.assertEqual(self.byte_data, snapshot_content)
+
+        # Act
+        service.delete_blob()
+
+        # Assert
+        with self.assertRaises(ResourceNotFoundError):
+            service.get_blob_properties()
+
     @record
     def test_sas_signed_identifier(self):
         # SAS URL is calculated from storage key, so this test runs live only

sdk/storage/azure-storage-file/azure/storage/file/file_client.py

@@ -26,13 +26,13 @@
 from ._generated.models import StorageErrorException, FileHTTPHeaders
 from ._shared.uploads import IterStreamer, FileChunkUploader, upload_data_chunks
 from ._shared.downloads import StorageStreamDownloader
-from ._shared.shared_access_signature import FileSharedAccessSignature
 from ._shared.base_client import StorageAccountHostsMixin, parse_connection_str, parse_query
 from ._shared.request_handlers import add_metadata_headers, get_length
 from ._shared.response_handlers import return_response_headers, process_storage_error
 from ._deserialize import deserialize_file_properties, deserialize_file_stream
 from ._polling import CloseHandles
 from .models import HandlesPaged
+from ._shared.shared_access_signature import SharedAccessSignature
 
 if TYPE_CHECKING:
     from datetime import datetime
@@ -294,7 +294,7 @@ def generate_shared_access_signature(
         """
         if not hasattr(self.credential, 'account_key') or not self.credential.account_key:
             raise ValueError("No account SAS key available.")
-        sas = FileSharedAccessSignature(self.credential.account_name, self.credential.account_key)
+        sas = SharedAccessSignature(self.credential.account_name, self.credential.account_key)
         if len(self.file_path) > 1:
             file_path = '/'.join(self.file_path[:-1])
         else:

sdk/storage/azure-storage-file/azure/storage/file/share_client.py

@@ -15,13 +15,13 @@
 
 import six
 from azure.core.tracing.decorator import distributed_trace
-from ._shared.shared_access_signature import FileSharedAccessSignature
 from ._shared.base_client import StorageAccountHostsMixin, parse_connection_str, parse_query
 from ._shared.request_handlers import add_metadata_headers, serialize_iso
 from ._shared.response_handlers import (
     return_response_headers,
     process_storage_error,
     return_headers_and_deserialized)
+from ._shared.shared_access_signature import SharedAccessSignature
 from ._generated import AzureFileStorage
 from ._generated.version import VERSION
 from ._generated.models import (
@@ -237,7 +237,7 @@ def generate_shared_access_signature(
         """
         if not hasattr(self.credential, 'account_key') or not self.credential.account_key:
             raise ValueError("No account SAS key available.")
-        sas = FileSharedAccessSignature(self.credential.account_name, self.credential.account_key)
+        sas = SharedAccessSignature(self.credential.account_name, self.credential.account_key)
         return sas.generate_share(
             self.share_name,
             permission,