diff --git a/src/azure-cli/azure/cli/command_modules/storage/_params.py b/src/azure-cli/azure/cli/command_modules/storage/_params.py
index 6176b6b9e22..00638e62189 100644
--- a/src/azure-cli/azure/cli/command_modules/storage/_params.py
+++ b/src/azure-cli/azure/cli/command_modules/storage/_params.py
@@ -1399,6 +1399,7 @@ def load_arguments(self, _):  # pylint: disable=too-many-locals, too-many-statem
         c.register_precondition_options()
         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.ignore('incremental_copy')
         c.argument('if_match', options_list=['--destination-if-match'])
diff --git a/src/azure-cli/azure/cli/command_modules/storage/_validators.py b/src/azure-cli/azure/cli/command_modules/storage/_validators.py
index c4ced3e18e0..420f34385db 100644
--- a/src/azure-cli/azure/cli/command_modules/storage/_validators.py
+++ b/src/azure-cli/azure/cli/command_modules/storage/_validators.py
@@ -43,6 +43,11 @@ def _query_account_key(cli_ctx, account_name):
     return scf.storage_accounts.list_keys(rg, account_name, logging_enable=False).keys_property[0].value  # pylint: disable=no-member
 
 
+def _is_valid_storage_account_name(account_name):
+    import re
+    return bool(account_name and re.fullmatch(r'[a-z0-9]{3,24}', account_name))
+
+
 def _query_account_rg(cli_ctx, account_name):
     """Query the storage account's resource group, which the mgmt sdk requires."""
     scf = storage_client_factory(cli_ctx)
@@ -422,6 +427,7 @@ def validate_source_url(cmd, namespace):  # pylint: disable=too-many-statements,
         '\n\tOR --source-share --source-path [--source-account-name & key]'
 
     ns = vars(namespace)
+    ns['source_is_validated_same_account'] = False
 
     # source as blob
     container = ns.pop('source_container', None)
@@ -487,6 +493,9 @@ def validate_source_url(cmd, namespace):  # pylint: disable=too-many-statements,
             except ValueError:
                 raise RequiredArgumentMissingError('Source storage account {} not found.'.format(source_account_name))
 
+    ns['source_is_validated_same_account'] = bool(
+        valid_blob_source and same_account and _is_valid_storage_account_name(source_account_name))
+
     # if oauth, use user delegation key to generate sas
     source_user_delegation_key = None
     if is_oauth:
diff --git a/src/azure-cli/azure/cli/command_modules/storage/operations/blob.py b/src/azure-cli/azure/cli/command_modules/storage/operations/blob.py
index 55c9f206108..93e26180e5c 100644
--- a/src/azure-cli/azure/cli/command_modules/storage/operations/blob.py
+++ b/src/azure-cli/azure/cli/command_modules/storage/operations/blob.py
@@ -1085,14 +1085,59 @@ def query_blob(client, query_expression, input_config=None, output_config=None,
     return reader.readall().decode("utf-8")
 
 
+def _normalize_url_origin(url):
+    from ipaddress import ip_address
+    from urllib.parse import urlsplit
+    import idna
+
+    try:
+        parsed = urlsplit(url)
+        scheme = parsed.scheme.lower()
+        hostname = parsed.hostname
+        if scheme not in ('http', 'https') or not hostname:
+            return None
+
+        hostname = hostname.rstrip('.')
+        if not hostname:
+            return None
+        try:
+            hostname = ip_address(hostname).compressed
+        except ValueError:
+            hostname = idna.encode(hostname, uts46=True).decode('ascii').lower().rstrip('.')
+            if not hostname:
+                return None
+
+        port = parsed.port
+        if port is None:
+            port = 443 if scheme == 'https' else 80
+        return scheme, hostname, port
+    except (UnicodeError, ValueError):
+        return None
+
+
+def _same_url_origin(first_url, second_url):
+    first_origin = _normalize_url_origin(first_url)
+    return first_origin is not None and first_origin == _normalize_url_origin(second_url)
+
+
+def _account_names_match_if_available(first_client, second_client):
+    first_account = first_client.account_name
+    second_account = second_client.account_name
+    return not first_account or not second_account or first_account.lower() == second_account.lower()
+
+
 def copy_blob(cmd, client, source_url, metadata=None, **kwargs):
     if not kwargs['requires_sync']:
         kwargs.pop('requires_sync')
     blob_type = kwargs.pop('destination_blob_type', None)
+    source_is_validated_same_account = kwargs.pop('source_is_validated_same_account', False)
     src_client = kwargs.pop('source_client', None)
     if src_client is None:
         src_client = client.from_blob_url(source_url)
-        if src_client.account_name == client.account_name:
+        source_matches_destination = _same_url_origin(source_url, client.url) and \
+            _account_names_match_if_available(src_client, client)
+        can_reuse_destination_credential = source_is_validated_same_account or source_matches_destination
+        if can_reuse_destination_credential:
             src_client = client.from_blob_url(source_url, credential=client.credential)
     StandardBlobTier = cmd.get_models('_models#StandardBlobTier')
     if blob_type is not None and blob_type != 'Detect':
diff --git a/src/azure-cli/azure/cli/command_modules/storage/tests/latest/test_storage_blob_copy_scenarios.py b/src/azure-cli/azure/cli/command_modules/storage/tests/latest/test_storage_blob_copy_scenarios.py
index 77590974d6b..6ad1fd6ea5f 100644
--- a/src/azure-cli/azure/cli/command_modules/storage/tests/latest/test_storage_blob_copy_scenarios.py
+++ b/src/azure-cli/azure/cli/command_modules/storage/tests/latest/test_storage_blob_copy_scenarios.py
@@ -3,10 +3,169 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 
+from unittest import TestCase, mock
+
 from azure.cli.testsdk import LiveScenarioTest, ResourceGroupPreparer, StorageAccountPreparer, JMESPathCheck
+from ...operations import blob as blob_operations
 from ..storage_test_util import StorageScenarioMixin
 
 
+class StorageBlobCopySecurityTests(TestCase):
+    def test_storage_blob_copy_source_origin_normalization(self):
+        equivalent_origins = [
+            ('https://account.blob.core.windows.net/container/source',
+             'https://ACCOUNT.blob.core.windows.net:443/container/destination'),
+            ('https://account.blob.core.usgovcloudapi.net./container/source',
+             'https://account.blob.core.usgovcloudapi.net/container/destination'),
+            ('https://account.blob.core.windows.net。/container/source',
+             'https://account.blob.core.windows.net/container/destination'),
+            ('https://xn--bcher-kva.example/container/source',
+             'https://bücher.example:443/container/destination'),
+            ('http://127.0.0.1:10000/account/container/source',
+             'http://127.0.0.1:10000/account/container/destination'),
+            ('http://[::1]:10000/account/container/source',
+             'http://[0:0:0:0:0:0:0:1]:10000/account/container/destination'),
+        ]
+        for source_url, destination_url in equivalent_origins:
+            with self.subTest(source_url=source_url, destination_url=destination_url):
+                self.assertTrue(blob_operations._same_url_origin(source_url, destination_url))
+
+        different_origins = [
+            ('https://account.blob.core.windows.net.attacker.example/container/source',
+             'https://account.blob.core.windows.net/container/destination'),
+            ('https://account.blob.core.windows.net@attacker.example/container/source',
+             'https://account.blob.core.windows.net/container/destination'),
+            ('http://account.blob.core.windows.net/container/source',
+             'https://account.blob.core.windows.net/container/destination'),
+            ('https://account.blob.core.windows.net:444/container/source',
+             'https://account.blob.core.windows.net/container/destination'),
+            ('https://faß.example/container/source',
+             'https://fass.example/container/destination'),
+            ('https://account.blob.core.windows.net/container/source',
+             'https://custom.example/container/destination'),
+            ('not-a-url', 'https://account.blob.core.windows.net/container/destination'),
+        ]
+        for source_url, destination_url in different_origins:
+            with self.subTest(source_url=source_url, destination_url=destination_url):
+                self.assertFalse(blob_operations._same_url_origin(source_url, destination_url))
+
+    @mock.patch.object(blob_operations, 'generate_sas_blob_uri')
+    def test_storage_blob_copy_does_not_reuse_credentials_for_untrusted_source(self, generate_sas):
+        source_url = 'https://account.blob.core.windows.net.attacker.example/container/source'
+
+        for blob_type in ('BlockBlob', 'AppendBlob', 'PageBlob'):
+            with self.subTest(blob_type=blob_type):
+                destination_client = mock.MagicMock()
+                destination_client.account_name = 'account'
+                destination_client.url = 'https://account.blob.core.windows.net/container/destination'
+                destination_client.credential = mock.sentinel.destination_credential
+
+                source_client = mock.MagicMock()
+                source_client.account_name = 'account'
+                source_client.get_blob_properties.return_value.size = 512
+                source_service_client = source_client._get_container_client.return_value. \
+                    _get_blob_service_client.return_value
+                source_service_client.credential = None
+                destination_client.from_blob_url.return_value = source_client
+
+                cmd = mock.MagicMock()
+                blob_operations.copy_blob(cmd, destination_client, source_url,
+                                          requires_sync=False, destination_blob_type=blob_type)
+
+                self.assertTrue(all('credential' not in call.kwargs
+                                    for call in destination_client.from_blob_url.call_args_list))
+                generate_sas.assert_not_called()
+                generate_sas.reset_mock()
+
+    @mock.patch.object(blob_operations, 'generate_sas_blob_uri')
+    def test_storage_blob_copy_reuses_credentials_for_same_source_origin(self, generate_sas):
+        source_url = 'https://ACCOUNT.blob.core.windows.net:443/container/source'
+        signed_source_url = source_url + '?sig=generated'
+        generate_sas.return_value = signed_source_url
+
+        destination_client = mock.MagicMock()
+        destination_client.account_name = 'account'
+        destination_client.url = 'https://account.blob.core.windows.net/container/destination'
+        destination_client.credential = mock.MagicMock(account_key='key')
+
+        anonymous_source_client = mock.MagicMock()
+        anonymous_source_client.account_name = 'ACCOUNT'
+        authenticated_source_client = mock.MagicMock()
+        authenticated_source_service = authenticated_source_client._get_container_client.return_value. \
+            _get_blob_service_client.return_value
+        authenticated_source_service.credential = destination_client.credential
+        destination_client.from_blob_url.side_effect = [anonymous_source_client, authenticated_source_client]
+
+        cmd = mock.MagicMock()
+        blob_operations.copy_blob(cmd, destination_client, source_url,
+                                  requires_sync=False, destination_blob_type='BlockBlob')
+
+        destination_client.from_blob_url.assert_has_calls([
+            mock.call(source_url),
+            mock.call(source_url, credential=destination_client.credential)
+        ])
+        generate_sas.assert_called_once()
+        destination_client.upload_blob_from_url.assert_called_once_with(
+            source_url=signed_source_url, overwrite=True, tags=None, destination_lease=None,
+            standard_blob_tier=None, source_if_modified_since=None, source_if_unmodified_since=None,
+            if_modified_since=None, if_unmodified_since=None, timeout=None)
+
+    @mock.patch.object(blob_operations, 'generate_sas_blob_uri')
+    def test_storage_blob_copy_reuses_credentials_for_validated_same_account_source(self, generate_sas):
+        source_url = 'https://account.blob.core.windows.net/src/input'
+        signed_source_url = source_url + '?sig=generated'
+        generate_sas.return_value = signed_source_url
+
+        destination_client = mock.MagicMock()
+        destination_client.url = 'https://storage.internal.example/dst/output'
+        destination_client.credential = mock.MagicMock(account_key='key')
+
+        anonymous_source_client = mock.MagicMock()
+        authenticated_source_client = mock.MagicMock()
+        source_service_client = authenticated_source_client._get_container_client.return_value. \
+            _get_blob_service_client.return_value
+        source_service_client.credential = destination_client.credential
+        destination_client.from_blob_url.side_effect = [anonymous_source_client, authenticated_source_client]
+
+        cmd = mock.MagicMock()
+        blob_operations.copy_blob(
+            cmd, destination_client, source_url, source_is_validated_same_account=True,
+            requires_sync=False, destination_blob_type='BlockBlob')
+
+        destination_client.from_blob_url.assert_has_calls([
+            mock.call(source_url),
+            mock.call(source_url, credential=destination_client.credential)
+        ])
+        generate_sas.assert_called_once()
+        destination_client.upload_blob_from_url.assert_called_once_with(
+            source_url=signed_source_url, overwrite=True, tags=None, destination_lease=None,
+            standard_blob_tier=None, source_if_modified_since=None, source_if_unmodified_since=None,
+            if_modified_since=None, if_unmodified_since=None, timeout=None)
+
+    @mock.patch.object(blob_operations, 'generate_sas_blob_uri')
+    def test_storage_blob_copy_does_not_reuse_credentials_across_path_style_accounts(self, generate_sas):
+        source_url = 'http://127.0.0.1:10000/account2/src/input'
+
+        destination_client = mock.MagicMock()
+        destination_client.account_name = 'account1'
+        destination_client.url = 'http://127.0.0.1:10000/account1/dst/output'
+        destination_client.credential = mock.sentinel.destination_credential
+
+        source_client = mock.MagicMock()
+        source_client.account_name = 'account2'
+        source_service_client = source_client._get_container_client.return_value. \
+            _get_blob_service_client.return_value
+        source_service_client.credential = None
+        destination_client.from_blob_url.return_value = source_client
+
+        cmd = mock.MagicMock()
+        blob_operations.copy_blob(
+            cmd, destination_client, source_url, requires_sync=False, destination_blob_type='BlockBlob')
+
+        destination_client.from_blob_url.assert_called_once_with(source_url)
+        generate_sas.assert_not_called()
+
+
 class StorageBlobCopyTests(StorageScenarioMixin, LiveScenarioTest):
     @ResourceGroupPreparer()
     @StorageAccountPreparer(parameter_name='source_account')
diff --git a/src/azure-cli/azure/cli/command_modules/storage/tests/latest/test_storage_validators.py b/src/azure-cli/azure/cli/command_modules/storage/tests/latest/test_storage_validators.py
index 98b73a150dc..935a32f8baf 100644
--- a/src/azure-cli/azure/cli/command_modules/storage/tests/latest/test_storage_validators.py
+++ b/src/azure-cli/azure/cli/command_modules/storage/tests/latest/test_storage_validators.py
@@ -19,6 +19,7 @@
                                                            process_blob_source_uri, get_char_options_validator,
                                                            get_source_file_or_blob_service_client_track2,
                                                            validate_encryption_source, validate_source_uri,
+                                                           validate_source_url,
                                                            validate_encryption_services, as_user_validator,
                                                            get_not_none_validator, validate_upload_blob)
 
@@ -162,6 +163,62 @@ def test_validate_source_uri(self):
         validate_source_uri(MockCmd(self.cli), ns)
         self.assertEqual(ns.copy_source, 'https://other_name.file.core.windows.net/share2?some_sas_token')
 
+    @mock.patch('azure.cli.command_modules.storage._validators.validate_client_parameters')
+    def test_validate_source_url_marks_structured_same_account_blob(self, _):
+        ns = Namespace(
+            source_url=None, source_sas=None, source_container='src', source_blob='input',
+            source_snapshot=None, source_share=None, source_path=None, file_snapshot=None,
+            source_account_name='account', source_account_key=None, token_credential=None,
+            account_name='account', account_key='key', account_url='https://storage.internal.example',
+            connection_string=None, sas_token=None, location_mode=None, connection_timeout=None,
+            container_name='dst', share_name=None)
+
+        validate_source_url(MockCmd(self.cli), ns)
+
+        self.assertTrue(ns.source_is_validated_same_account)
+        self.assertEqual(ns.source_url, 'https://account.blob.core.windows.net/src/input')
+
+    def test_validate_source_url_does_not_mark_arbitrary_uri_as_same_account(self):
+        source_url = 'https://account.blob.core.windows.net.attacker.example/src/input'
+        ns = Namespace(
+            source_url=source_url, source_sas=None, source_container=None, source_blob=None,
+            source_snapshot=None, source_share=None, source_path=None, file_snapshot=None,
+            source_account_name=None, source_account_key=None)
+        validate_source_url(MockCmd(self.cli), ns)
+        validate_source_url(MockCmd(self.cli), ns)
+
+        self.assertFalse(ns.source_is_validated_same_account)
+        self.assertEqual(ns.source_url, source_url)
+
+    @mock.patch('azure.cli.command_modules.storage._validators.validate_client_parameters')
+    @mock.patch('azure.cli.command_modules.storage.util.create_short_lived_file_sas_v2', return_value='sas')
+    def test_validate_source_url_does_not_mark_structured_file_as_same_account(self, _, __):
+        ns = Namespace(
+            source_url=None, source_sas=None, source_container=None, source_blob=None,
+            source_snapshot=None, source_share='src', source_path='input', file_snapshot=None,
+            source_account_name='account', source_account_key=None, token_credential=None,
+            account_name='account', account_key='key', sas_token=None,
+            container_name='dst', share_name=None)
+
+        validate_source_url(MockCmd(self.cli), ns)
+
+        self.assertFalse(ns.source_is_validated_same_account)
+
+    @mock.patch('azure.cli.command_modules.storage._validators.validate_client_parameters')
+    def test_validate_source_url_requires_valid_account_name_for_same_account(self, _):
+        for account_name in (None, 'same@attacker.example:443/'):
+            with self.subTest(account_name=account_name):
+                ns = Namespace(
+                    source_url=None, source_sas=None, source_container='src', source_blob='input',
+                    source_snapshot=None, source_share=None, source_path=None, file_snapshot=None,
+                    source_account_name=account_name, source_account_key=None, token_credential=None,
+                    account_name=account_name, account_key='key', sas_token=None,
+                    container_name='dst', share_name=None)
+
+                validate_source_url(MockCmd(self.cli), ns)
+
+                self.assertFalse(ns.source_is_validated_same_account)
+
     def test_get_not_none_validator(self):
         from azure.cli.core.azclierror import InvalidArgumentValueError
         from knack.arguments import CLICommandArgument, CLIArgumentType