sync copy from url with encryption scope

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

@@ -1869,9 +1869,21 @@ def _start_copy_from_url_options(self, source_url, metadata=None, incremental_co
 
         tier = kwargs.pop('premium_page_blob_tier', None) or kwargs.pop('standard_blob_tier', None)
         requires_sync = kwargs.pop('requires_sync', None)
+        encryption_scope_str = kwargs.pop('encryption_scope', None)
         source_authorization = kwargs.pop('source_authorization', None)
+
+        if not requires_sync and encryption_scope_str:
+            raise ValueError("Encryption_scope is only supported for sync copy, please specify requires_sync=True")
         if source_authorization and incremental_copy:
             raise ValueError("Source authorization tokens are not applicable for incremental copying.")
+        #
+        # TODO: refactor start_copy_from_url api in _blob_client.py. Call _generated/_blob_operations.py copy_from_url
+        #  when requires_sync=True is set.
+        #  Currently both sync copy and async copy are calling _generated/_blob_operations.py start_copy_from_url.
+        #  As sync copy diverges more from async copy, more problem will surface.
+        if encryption_scope_str:
+            headers.update({'x-ms-encryption-scope': encryption_scope_str})
+
         if requires_sync is True:
             headers['x-ms-requires-sync'] = str(requires_sync)
             if source_authorization:
@@ -2056,6 +2068,17 @@ def start_copy_from_url(self, source_url, metadata=None, incremental_copy=False,
             Authenticate as a service principal using a client secret to access a source blob. Ensure "bearer " is
             the prefix of the source_authorization string. This option is only available when `incremental_copy` is
             set to False and `requires_sync` is set to True.
+
+            .. versionadded:: 12.9.0
+
+        :keyword str encryption_scope:
+            A predefined encryption scope used to encrypt the data on the sync copied blob. An encryption
+            scope can be created using the Management API and referenced here by name. If a default
+            encryption scope has been defined at the container, this value will override it if the
+            container-level scope is configured to allow overrides. Otherwise an error will be raised.
+
+            .. versionadded:: 12.10.0
+
         :returns: A dictionary of copy properties (etag, last_modified, copy_id, copy_status).
         :rtype: dict[str, Union[str, ~datetime.datetime]]
 

sdk/storage/azure-storage-blob/azure/storage/blob/aio/_blob_client_async.py

@@ -1275,6 +1275,17 @@ async def start_copy_from_url(self, source_url, metadata=None, incremental_copy=
             Authenticate as a service principal using a client secret to access a source blob. Ensure "bearer " is
             the prefix of the source_authorization string. This option is only available when `incremental_copy` is
             set to False and `requires_sync` is set to True.
+
+            .. versionadded:: 12.9.0
+
+        :keyword str encryption_scope:
+            A predefined encryption scope used to encrypt the data on the sync copied blob. An encryption
+            scope can be created using the Management API and referenced here by name. If a default
+            encryption scope has been defined at the container, this value will override it if the
+            container-level scope is configured to allow overrides. Otherwise an error will be raised.
+
+            .. versionadded:: 12.10.0
+
         :returns: A dictionary of copy properties (etag, last_modified, copy_id, copy_status).
         :rtype: dict[str, Union[str, ~datetime.datetime]]
 

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

@@ -864,7 +864,7 @@ def test_copy_with_account_encryption_scope_sas(self, storage_account_name, stor
             max_page_size=1024)
 
         self._setup(bsc_with_sas_credential)
-        # blob is encrypted using TEST_SAS_ENCRYPTION_SCOPE
+        # blob is encrypted using TEST_SAS_ENCRYPTION_SCOPE_2
         blob_client, _ = self._create_block_blob(bsc_with_sas_credential, blob_name="blockblob", data=b'AAABBBCCC', overwrite=True)
 
         #
@@ -897,6 +897,51 @@ def test_copy_with_account_encryption_scope_sas(self, storage_account_name, stor
 
         self._teardown(bsc_with_sas_credential)
 
+    @pytest.mark.live_test_only
+    @BlobPreparer()
+    def test_copy_blob_from_url_with_ecryption_scope(self, storage_account_name, storage_account_key):
+        # Arrange
+
+        # create sas for source blob
+        sas_token = generate_account_sas(
+            storage_account_name,
+            account_key=storage_account_key,
+            resource_types=ResourceTypes(object=True, container=True),
+            permission=AccountSasPermissions(read=True, write=True, delete=True, list=True),
+            expiry=datetime.utcnow() + timedelta(hours=1),
+        )
+        bsc_with_sas_credential = BlobServiceClient(
+            self.account_url(storage_account_name, "blob"),
+            credential=sas_token,
+            connection_data_block_size=1024,
+            max_single_put_size=1024,
+            min_large_block_upload_threshold=1024,
+            max_block_size=1024,
+            max_page_size=1024)
+
+        self._setup(bsc_with_sas_credential)
+        blob_client, _ = self._create_block_blob(bsc_with_sas_credential, blob_name="blockblob", data=b'AAABBBCCC', overwrite=True)
+
+        bsc = BlobServiceClient(
+            self.account_url(storage_account_name, "blob"),
+            credential=storage_account_key,
+            connection_data_block_size=1024,
+            max_single_put_size=1024,
+            min_large_block_upload_threshold=1024,
+            max_block_size=1024,
+            max_page_size=1024)
+        copied_blob = self.get_resource_name('copiedblob')
+        copied_blob_client = bsc.get_blob_client(self.container_name, copied_blob)
+
+        copied_blob_client.start_copy_from_url(blob_client.url, requires_sync=True,
+                                               encryption_scope=TEST_SAS_ENCRYPTION_SCOPE)
+
+        props = copied_blob_client.get_blob_properties()
+
+        self.assertEqual(props.encryption_scope, TEST_SAS_ENCRYPTION_SCOPE)
+
+        self._teardown(bsc_with_sas_credential)
+
     @pytest.mark.live_test_only
     @BlobPreparer()
     def test_copy_with_user_delegation_encryption_scope_sas(self, storage_account_name, storage_account_key):

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

@@ -888,7 +888,7 @@ async def test_copy_with_account_encryption_scope_sas(self, storage_account_name
             max_page_size=1024)
 
         await self._setup(bsc_with_sas_credential)
-        # blob is encrypted using TEST_SAS_ENCRYPTION_SCOPE
+        # blob is encrypted using TEST_SAS_ENCRYPTION_SCOPE_2
         blob_client, _ = await self._create_block_blob(bsc_with_sas_credential, blob_name="blockblob", data=b'AAABBBCCC', overwrite=True)
 
         #
@@ -921,6 +921,51 @@ async def test_copy_with_account_encryption_scope_sas(self, storage_account_name
 
         self._teardown(bsc_with_sas_credential)
 
+    @pytest.mark.live_test_only
+    @BlobPreparer()
+    async def test_copy_blob_from_url_with_ecryption_scope(self, storage_account_name, storage_account_key):
+        # Arrange
+
+        # create sas for source blob
+        sas_token = generate_account_sas(
+            storage_account_name,
+            account_key=storage_account_key,
+            resource_types=ResourceTypes(object=True, container=True),
+            permission=AccountSasPermissions(read=True, write=True, delete=True, list=True),
+            expiry=datetime.utcnow() + timedelta(hours=1),
+        )
+        bsc_with_sas_credential = BlobServiceClient(
+            self.account_url(storage_account_name, "blob"),
+            credential=sas_token,
+            connection_data_block_size=1024,
+            max_single_put_size=1024,
+            min_large_block_upload_threshold=1024,
+            max_block_size=1024,
+            max_page_size=1024)
+
+        await self._setup(bsc_with_sas_credential)
+        blob_client, _ = await self._create_block_blob(bsc_with_sas_credential, blob_name="blockblob", data=b'AAABBBCCC', overwrite=True)
+
+        bsc = BlobServiceClient(
+            self.account_url(storage_account_name, "blob"),
+            credential=storage_account_key,
+            connection_data_block_size=1024,
+            max_single_put_size=1024,
+            min_large_block_upload_threshold=1024,
+            max_block_size=1024,
+            max_page_size=1024)
+        copied_blob = self.get_resource_name('copiedblob')
+        copied_blob_client = bsc.get_blob_client(self.container_name, copied_blob)
+
+        await copied_blob_client.start_copy_from_url(blob_client.url, requires_sync=True,
+                                               encryption_scope=TEST_SAS_ENCRYPTION_SCOPE)
+
+        props = await copied_blob_client.get_blob_properties()
+
+        self.assertEqual(props.encryption_scope, TEST_SAS_ENCRYPTION_SCOPE)
+
+        self._teardown(bsc_with_sas_credential)
+
     @pytest.mark.live_test_only
     @BlobPreparer()
     async def test_copy_with_user_delegation_encryption_scope_sas(self, storage_account_name, storage_account_key):