[STG78] Copy Source OAuth Token Support (#18998)

* OAuth Copy Source Support

* added tokens to blob client

* added support in file client

* Added tests

* regenerated

* Added sync tests and renamed param

* added some async tests

* removed import and fixed desc and validation

* added more tests

* remove import

* fixed ci

Author: tasherif-msft

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

@@ -426,6 +426,7 @@ def _upload_blob_from_url_options(self, source_url, **kwargs):
         tier = kwargs.pop('standard_blob_tier', None)
         overwrite = kwargs.pop('overwrite', False)
         content_settings = kwargs.pop('content_settings', None)
+        source_authorization = kwargs.pop('source_authorization', None)
         if content_settings:
             kwargs['blob_http_headers'] = BlobHTTPHeaders(
                 blob_cache_control=content_settings.cache_control,
@@ -444,6 +445,7 @@ def _upload_blob_from_url_options(self, source_url, **kwargs):
                                encryption_algorithm=cpk.algorithm)
 
         options = {
+            'copy_source_authorization': source_authorization,
             'content_length': 0,
             'copy_source_blob_properties': kwargs.pop('include_source_blob_properties', True),
             'source_content_md5': kwargs.pop('source_content_md5', None),
@@ -552,6 +554,9 @@ def upload_blob_from_url(self, source_url, **kwargs):
         :keyword ~azure.storage.blob.StandardBlobTier standard_blob_tier:
             A standard blob tier value to set the blob to. For this version of the library,
             this is only applicable to block blobs on standard storage accounts.
+        :keyword str source_authorization:
+            Authenticate as a service principal using a client secret to access a source blob. Ensure "bearer " is
+            the prefix of the source_authorization string.
         """
         options = self._upload_blob_from_url_options(
             source_url=self._encode_source_url(source_url),
@@ -1756,10 +1761,17 @@ def _start_copy_from_url_options(self, source_url, metadata=None, incremental_co
                 headers['x-ms-source-lease-id'] = source_lease
 
         tier = kwargs.pop('premium_page_blob_tier', None) or kwargs.pop('standard_blob_tier', None)
-
-        if kwargs.get('requires_sync'):
-            headers['x-ms-requires-sync'] = str(kwargs.pop('requires_sync'))
-
+        requires_sync = kwargs.pop('requires_sync', None)
+        source_authorization = kwargs.pop('source_authorization', None)
+        if source_authorization and incremental_copy:
+            raise ValueError("Source authorization tokens are not applicable for incremental copying.")
+        if requires_sync is True:
+            headers['x-ms-requires-sync'] = str(requires_sync)
+            if source_authorization:
+                headers['x-ms-copy-source-authorization'] = source_authorization
+        else:
+            if source_authorization:
+                raise ValueError("Source authorization tokens are only applicable for synchronous copy operations.")
         timeout = kwargs.pop('timeout', None)
         dest_mod_conditions = get_modify_conditions(kwargs)
         blob_tags_string = serialize_blob_tags_header(kwargs.pop('tags', None))
@@ -1916,6 +1928,10 @@ def start_copy_from_url(self, source_url, metadata=None, incremental_copy=False,
 
         :keyword bool requires_sync:
             Enforces that the service will not return a response until the copy is complete.
+        :keyword str source_authorization:
+            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.
         :returns: A dictionary of copy properties (etag, last_modified, copy_id, copy_status).
         :rtype: dict[str, str or ~datetime.datetime]
 
@@ -2210,6 +2226,7 @@ def _stage_block_from_url_options(
             **kwargs
         ):
         # type: (...) -> Dict[str, Any]
+        source_authorization = kwargs.pop('source_authorization', None)
         if source_length is not None and source_offset is None:
             raise ValueError("Source offset value must not be None if length is set.")
         if source_length is not None:
@@ -2229,6 +2246,7 @@ def _stage_block_from_url_options(
             cpk_info = CpkInfo(encryption_key=cpk.key_value, encryption_key_sha256=cpk.key_hash,
                                encryption_algorithm=cpk.algorithm)
         options = {
+            'copy_source_authorization': source_authorization,
             'block_id': block_id,
             'content_length': 0,
             'source_url': source_url,
@@ -2245,7 +2263,7 @@ def _stage_block_from_url_options(
 
     @distributed_trace
     def stage_block_from_url(
-            self, block_id,  # type: str
+            self, block_id,  # type: Union[str, int]
             source_url,  # type: str
             source_offset=None,  # type: Optional[int]
             source_length=None,  # type: Optional[int]
@@ -2286,6 +2304,9 @@ def stage_block_from_url(
 
         :keyword int timeout:
             The timeout parameter is expressed in seconds.
+        :keyword str source_authorization:
+            Authenticate as a service principal using a client secret to access a source blob. Ensure "bearer " is
+            the prefix of the source_authorization string.
         :returns: Blob property dict.
         :rtype: dict[str, Any]
         """
@@ -3151,6 +3172,7 @@ def _upload_pages_from_url_options(  # type: ignore
             if_sequence_number_less_than=kwargs.pop('if_sequence_number_lt', None),
             if_sequence_number_equal_to=kwargs.pop('if_sequence_number_eq', None)
         )
+        source_authorization = kwargs.pop('source_authorization', None)
         access_conditions = get_access_conditions(kwargs.pop('lease', None))
         mod_conditions = get_modify_conditions(kwargs)
         source_mod_conditions = get_source_conditions(kwargs)
@@ -3165,6 +3187,7 @@ def _upload_pages_from_url_options(  # type: ignore
                                encryption_algorithm=cpk.algorithm)
 
         options = {
+            'copy_source_authorization': source_authorization,
             'source_url': source_url,
             'content_length': 0,
             'source_range': source_range,
@@ -3279,6 +3302,9 @@ def upload_pages_from_url(self, source_url,  # type: str
 
         :keyword int timeout:
             The timeout parameter is expressed in seconds.
+        :keyword str source_authorization:
+            Authenticate as a service principal using a client secret to access a source blob. Ensure "bearer " is
+            the prefix of the source_authorization string.
         """
         options = self._upload_pages_from_url_options(
             source_url=self._encode_source_url(source_url),
@@ -3571,6 +3597,7 @@ def _append_block_from_url_options(  # type: ignore
                 max_size=maxsize_condition,
                 append_position=appendpos_condition
             )
+        source_authorization = kwargs.pop('source_authorization', None)
         access_conditions = get_access_conditions(kwargs.pop('lease', None))
         mod_conditions = get_modify_conditions(kwargs)
         source_mod_conditions = get_source_conditions(kwargs)
@@ -3584,6 +3611,7 @@ def _append_block_from_url_options(  # type: ignore
                                encryption_algorithm=cpk.algorithm)
 
         options = {
+            'copy_source_authorization': source_authorization,
             'source_url': copy_source_url,
             'content_length': 0,
             'source_range': source_range,
@@ -3690,6 +3718,9 @@ def append_block_from_url(self, copy_source_url,  # type: str
 
         :keyword int timeout:
             The timeout parameter is expressed in seconds.
+        :keyword str source_authorization:
+            Authenticate as a service principal using a client secret to access a source blob. Ensure "bearer " is
+            the prefix of the source_authorization string.
         """
         options = self._append_block_from_url_options(
             copy_source_url=self._encode_source_url(copy_source_url),

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

@@ -229,6 +229,9 @@ async def upload_blob_from_url(self, source_url, **kwargs):
         :keyword ~azure.storage.blob.StandardBlobTier standard_blob_tier:
             A standard blob tier value to set the blob to. For this version of the library,
             this is only applicable to block blobs on standard storage accounts.
+        :keyword str source_authorization:
+            Authenticate as a service principal using a client secret to access a source blob. Ensure "bearer " is
+            the prefix of the source_authorization string.
         """
         options = self._upload_blob_from_url_options(
             source_url=self._encode_source_url(source_url),
@@ -1161,6 +1164,10 @@ async def start_copy_from_url(self, source_url, metadata=None, incremental_copy=
 
         :keyword bool requires_sync:
             Enforces that the service will not return a response until the copy is complete.
+        :keyword str source_authorization:
+            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.
         :returns: A dictionary of copy properties (etag, last_modified, copy_id, copy_status).
         :rtype: dict[str, str or ~datetime.datetime]
 
@@ -1377,7 +1384,7 @@ async def stage_block(
 
     @distributed_trace_async
     async def stage_block_from_url(
-            self, block_id,  # type: str
+            self, block_id,  # type: Union[str, int]
             source_url,  # type: str
             source_offset=None,  # type: Optional[int]
             source_length=None,  # type: Optional[int]
@@ -1418,6 +1425,9 @@ async def stage_block_from_url(
 
         :keyword int timeout:
             The timeout parameter is expressed in seconds.
+        :keyword str source_authorization:
+            Authenticate as a service principal using a client secret to access a source blob. Ensure "bearer " is
+            the prefix of the source_authorization string.
         :rtype: None
         """
         options = self._stage_block_from_url_options(
@@ -2131,6 +2141,9 @@ async def upload_pages_from_url(self, source_url,  # type: str
 
         :keyword int timeout:
             The timeout parameter is expressed in seconds.
+        :keyword str source_authorization:
+            Authenticate as a service principal using a client secret to access a source blob. Ensure "bearer " is
+            the prefix of the source_authorization string.
         """
 
         options = self._upload_pages_from_url_options(
@@ -2391,6 +2404,9 @@ async def append_block_from_url(self, copy_source_url,  # type: str
 
         :keyword int timeout:
             The timeout parameter is expressed in seconds.
+        :keyword str source_authorization:
+            Authenticate as a service principal using a client secret to access a source blob. Ensure "bearer " is
+            the prefix of the source_authorization string.
         """
         options = self._append_block_from_url_options(
             copy_source_url=self._encode_source_url(copy_source_url),