[Storage] Add source login authentication for synchronous blob copy

src/azure-cli/azure/cli/command_modules/storage/_help.py

@@ -786,6 +786,16 @@
 type: command
 short-summary: Copy a blob asynchronously. Use `az storage blob show` to check the status of the blobs.
 parameters:
+  - name: --source-auth-mode
+    type: string
+    short-summary: Use Entra login for a named source blob without generating a SAS.
+    long-summary: >
+        Requires --auth-mode login and --requires-sync true. Specify the source using
+        --source-account-name, --source-container, and --source-blob. If the source account
+        name is omitted, the destination account is used. Source URI, source key/SAS,
+        file sources, and destination blob type conversion are not supported in this mode.
+        The signed-in identity needs blob read access to the source and write access to
+        the destination. Synchronous copy is subject to the service's size limits.
   - name: --source-uri -u
     type: string
     short-summary: >
@@ -835,6 +845,9 @@
         Specify this conditional header to copy the blob only if the source blob
         has not been modified since the specified date/time.
 examples:
+  - name: Copy a blob synchronously between accounts using Entra login without a SAS.
+    text: |
+        az storage blob copy start --account-name destinationaccount --destination-container destinationcontainer --destination-blob example.txt --source-account-name sourceaccount --source-container sourcecontainer --source-blob example.txt --auth-mode login --source-auth-mode login --requires-sync true
   - name: Copy a blob asynchronously. Use `az storage blob show` to check the status of the blobs.
     text: |
         az storage blob copy start --account-key 00000000 --account-name MyAccount --destination-blob MyDestinationBlob --destination-container MyDestinationContainer --source-uri https://storage.blob.core.windows.net/photos

src/azure-cli/azure/cli/command_modules/storage/_params.py

@@ -1397,6 +1397,11 @@ def load_arguments(self, _):  # pylint: disable=too-many-locals, too-many-statem
         c.register_precondition_options(prefix='source_')
         c.register_source_uri_arguments(validator=validate_source_url)
         c.argument('source_is_validated_same_account', ignore_type)
+        c.argument('source_token_credential', ignore_type)
+        c.extra('source_auth_mode', arg_type=get_enum_type(['login']), arg_group='Copy Source',
+            help='Use Entra login for a named source blob without generating a SAS. Requires '
+                 '--auth-mode login and --requires-sync true. Not supported with --source-uri '
+                 'or --destination-blob-type other than Detect.')
 
         c.ignore('incremental_copy')
         c.argument('if_match', options_list=['--destination-if-match'])

src/azure-cli/azure/cli/command_modules/storage/_validators.py

@@ -447,6 +447,20 @@ def validate_source_url(cmd, namespace):  # pylint: disable=too-many-statements,
     source_sas = ns.pop('source_sas', None)
     token_credential = ns.get('token_credential')
     is_oauth = token_credential is not None
+    source_login = ns.get('source_auth_mode') == 'login'
+    if source_login:
+        if not is_oauth or not ns.get('requires_sync'):
+            raise InvalidArgumentValueError(
+                '--source-auth-mode login requires --auth-mode login and --requires-sync true.')
+        if (not container or not blob or share or path or file_snapshot or ns.get('source_url') or
+                source_account_key or source_sas or
+                ns.get('destination_blob_type') not in (None, 'Detect')):
+            raise InvalidArgumentValueError(
+                '--source-auth-mode login requires a named source blob without source key/SAS '
+                'or blob type conversion.')
+        if not _is_valid_storage_account_name(source_account_name or ns.get('account_name')):
+            raise InvalidArgumentValueError('--source-auth-mode login requires a valid source storage account name.')
+        ns['source_token_credential'] = token_credential
 
     # source in the form of an uri
     uri = ns.get('source_url', None)
@@ -500,7 +514,7 @@ def validate_source_url(cmd, namespace):  # pylint: disable=too-many-statements,
 
     # if oauth, use user delegation key to generate sas
     source_user_delegation_key = None
-    if is_oauth:
+    if is_oauth and not source_login:
         client_kwargs = {'account_name': source_account_name,
                          'token_credential': token_credential}
         if valid_blob_source:
@@ -520,7 +534,7 @@ def validate_source_url(cmd, namespace):  # pylint: disable=too-many-statements,
                 dir_name = None
             source_sas = create_short_lived_file_sas_v2(cmd, source_account_name, source_account_key, share,
                                                         dir_name, file_name)
-        elif valid_blob_source and (ns.get('share_name', None) or not same_account):
+        elif valid_blob_source and not source_login and (ns.get('share_name', None) or not same_account):
             source_sas = create_short_lived_blob_sas_v2(cmd, source_account_name, container, blob,
                                                         account_key=source_account_key,
                                                         user_delegation_key=source_user_delegation_key)

src/azure-cli/azure/cli/command_modules/storage/operations/blob.py

@@ -1144,7 +1144,11 @@ def _account_names_match(first_client, second_client):
     return bool(first_account and second_account and first_account.lower() == second_account.lower())
 
 
-def copy_blob(cmd, client, source_url, metadata=None, **kwargs):
+def copy_blob(cmd, client, source_url, metadata=None, source_token_credential=None, **kwargs):
+    source_auth_mode = kwargs.pop('source_auth_mode', None)
+    if source_auth_mode == 'login':
+        kwargs['source_authorization'] = 'Bearer ' + source_token_credential.get_token(
+            'https://storage.azure.com/.default').token
     if not kwargs['requires_sync']:
         kwargs.pop('requires_sync')
     blob_type = kwargs.pop('destination_blob_type', None)

src/azure-cli/azure/cli/command_modules/storage/tests/latest/test_storage_blob_source_oauth.py

@@ -0,0 +1,129 @@
+import unittest
+from argparse import Namespace
+from unittest import mock
+
+from azure.cli.core.azclierror import InvalidArgumentValueError
+from azure.cli.command_modules.storage import _validators
+from azure.cli.command_modules.storage.operations import blob
+
+
+class StorageBlobSourceOAuthTests(unittest.TestCase):
+    def setUp(self):
+        self.cmd = mock.MagicMock()
+        self.cmd.cli_ctx.cloud.suffixes.storage_endpoint = 'core.windows.net'
+        self.namespace = Namespace(
+            account_name='destination', source_account_name='source',
+            source_container='states', source_blob='folder/state file.tfstate',
+            token_credential=mock.Mock(), source_auth_mode='login', requires_sync=True)
+
+    @mock.patch.object(_validators, 'validate_client_parameters')
+    @mock.patch.object(_validators, 'cf_blob_service')
+    @mock.patch.object(_validators, '_query_account_key')
+    def test_named_source_uses_no_account_key_or_delegation_sas(self, query_key, source_client, validate_client):
+        with mock.patch('azure.cli.command_modules.storage.util.create_short_lived_blob_sas_v2') as sas:
+            _validators.validate_source_url(self.cmd, self.namespace)
+        self.assertEqual(self.namespace.source_url,
+                         'https://source.blob.core.windows.net/states/folder/state%20file.tfstate')
+        query_key.assert_not_called()
+        source_client.assert_not_called()
+        sas.assert_not_called()
+        validate_client.assert_called_once()
+        self.assertIs(self.namespace.source_token_credential, self.namespace.token_credential)
+
+    def test_source_login_rejects_unsupported_combinations(self):
+        for changes in (
+                {'requires_sync': False}, {'token_credential': None},
+                {'source_url': 'https://untrusted.example/blob'}, {'source_sas': 'sig=value'},
+                {'source_account_key': 'key'}, {'destination_blob_type': 'BlockBlob'},
+                {'source_account_name': 'source.blob.core.windows.net@untrusted.example'},
+                {'source_share': 'share', 'source_path': 'file'}):
+            with self.subTest(changes=changes):
+                namespace = Namespace(**{**vars(self.namespace), **changes})
+                with self.assertRaises(InvalidArgumentValueError):
+                    _validators.validate_source_url(self.cmd, namespace)
+
+    @mock.patch.object(_validators, 'validate_client_parameters')
+    @mock.patch.object(_validators, 'cf_blob_service')
+    def test_default_source_login_preserves_delegation_sas(self, source_client, validate_client):
+        self.namespace.source_auth_mode = None
+        with mock.patch('azure.cli.command_modules.storage.util.create_short_lived_blob_sas_v2',
+                        return_value='sig=test') as sas:
+            _validators.validate_source_url(self.cmd, self.namespace)
+        source_client.return_value.get_user_delegation_key.assert_called_once()
+        sas.assert_called_once()
+        self.assertTrue(self.namespace.source_url.endswith('?sig=test'))
+
+    @mock.patch.object(_validators, 'validate_client_parameters')
+    def test_source_login_uses_cloud_endpoint_and_same_account_default(self, validate_client):
+        self.namespace.source_account_name = None
+        self.namespace.source_snapshot = 'snapshot-value'
+        self.cmd.cli_ctx.cloud.suffixes.storage_endpoint = 'core.usgovcloudapi.net'
+        _validators.validate_source_url(self.cmd, self.namespace)
+        self.assertEqual(self.namespace.source_url,
+                         'https://destination.blob.core.usgovcloudapi.net/states/'
+                         'folder/state%20file.tfstate?snapshot=snapshot-value')
+
+    def test_synchronous_copy_passes_source_bearer_token(self):
+        client = mock.MagicMock()
+        client.url = 'https://destination.blob.core.windows.net/backups/state.tfstate'
+        client.credential.get_token.return_value.token = 'test-token'
+        source_url = 'https://source.blob.core.windows.net/states/state.tfstate'
+        blob.copy_blob(self.cmd, client, source_url, source_auth_mode='login', requires_sync=True,
+                   source_token_credential=client.credential)
+        client.credential.get_token.assert_called_once_with('https://storage.azure.com/.default')
+        client.start_copy_from_url.assert_called_once_with(
+            source_url=source_url, metadata=None, incremental_copy=False,
+            requires_sync=True, source_authorization='Bearer test-token')
+
+    def test_source_login_uses_sdk_storage_scope_with_real_cloud(self):
+        from azure.cli.core.cloud import AZURE_PUBLIC_CLOUD
+        self.cmd.cli_ctx.cloud = AZURE_PUBLIC_CLOUD
+        client = mock.MagicMock()
+        client.url = 'https://destination.blob.core.windows.net/container/example.txt'
+        client.credential.get_token.return_value.token = 'test-token'
+        blob.copy_blob(self.cmd, client,
+                       'https://source.blob.core.windows.net/container/example.txt',
+                       source_auth_mode='login', requires_sync=True,
+                       source_token_credential=client.credential)
+        client.credential.get_token.assert_called_once_with('https://storage.azure.com/.default')
+
+    @mock.patch.object(_validators, 'validate_client_parameters')
+    def test_source_login_is_independent_of_destination_credentials(self, validate_client):
+        import base64
+        from azure.storage.blob import BlobClient
+        account_key = base64.b64encode(b'synthetic-unit-test-key').decode()
+        destination_url = 'https://destination.blob.core.windows.net/container/example.txt'
+        connection_string = ('DefaultEndpointsProtocol=https;AccountName=destination;'
+                             'AccountKey=' + account_key + ';EndpointSuffix=core.windows.net')
+        clients = {
+            'account-key': BlobClient.from_blob_url(destination_url, credential=account_key),
+            'sas': BlobClient.from_blob_url(destination_url, credential='sig=synthetic'),
+            'connection-string': BlobClient.from_connection_string(
+                connection_string, container_name='container', blob_name='example.txt'),
+        }
+        for auth, client in clients.items():
+            with self.subTest(destination_auth=auth):
+                namespace = Namespace(**vars(self.namespace))
+                namespace.token_credential = mock.Mock()
+                namespace.token_credential.get_token.return_value.token = 'synthetic-unit-test-token'
+                _validators.validate_source_url(self.cmd, namespace)
+                with mock.patch.object(client, 'start_copy_from_url') as copy:
+                    blob.copy_blob(self.cmd, client, namespace.source_url, source_auth_mode='login',
+                                   requires_sync=True, source_token_credential=namespace.source_token_credential)
+                namespace.token_credential.get_token.assert_called_once_with('https://storage.azure.com/.default')
+                copy.assert_called_once_with(
+                    source_url=namespace.source_url, metadata=None, incremental_copy=False,
+                    requires_sync=True, source_authorization='Bearer synthetic-unit-test-token')
+
+    def test_default_copy_does_not_request_source_bearer_token(self):
+        client = mock.MagicMock()
+        client.url = 'https://destination.blob.core.windows.net/container/example.txt'
+        source_url = 'https://source.blob.core.windows.net/container/example.txt?sig=test'
+        blob.copy_blob(self.cmd, client, source_url, source_auth_mode=None, requires_sync=False)
+        client.credential.get_token.assert_not_called()
+        client.start_copy_from_url.assert_called_once_with(
+            source_url=source_url, metadata=None, incremental_copy=False)
+
+
+if __name__ == '__main__':
+    unittest.main()