From c608ae00186140d5073eba94cc6420d2a5820c93 Mon Sep 17 00:00:00 2001
From: Yang An <yangan@microsoft.com>
Date: Fri, 16 Oct 2026 11:25:12 +1100
Subject: [PATCH 1/5] [Storage] Validate account endpoints before attaching SAS
 credentials

Validate parsed endpoint origins against account and trusted connection-string endpoints before SAS generation or attachment. Preserve external URL passthrough without adding storage credentials and cover copy and filesystem validators.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 src/azure-cli/HISTORY.rst                     |   1 +
 .../command_modules/storage/_validators.py    | 116 +++++++++--
 .../latest/test_storage_azcopy_scenarios.py   |  32 ++-
 .../tests/latest/test_storage_validators.py   | 190 +++++++++++++++++-
 4 files changed, 318 insertions(+), 21 deletions(-)

diff --git a/src/azure-cli/HISTORY.rst b/src/azure-cli/HISTORY.rst
index 47e64c36723..fcd7198fb52 100644
--- a/src/azure-cli/HISTORY.rst
+++ b/src/azure-cli/HISTORY.rst
@@ -96,6 +96,7 @@ Release History
 
 **Storage**
 
+* `az storage copy`, `az storage fs directory upload/download`: Validate configured account endpoints before attaching SAS credentials
 * `az storage blob download-batch`: Ensure downloaded blobs stay within the destination directory (#34126)
 
 2.90.0
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 0caee9ed3d5..620a9df02b2 100644
--- a/src/azure-cli/azure/cli/command_modules/storage/_validators.py
+++ b/src/azure-cli/azure/cli/command_modules/storage/_validators.py
@@ -1804,6 +1804,61 @@ def validate_or_policy(cmd, namespace):
             name=namespace.destination_account)
 
 
+def _storage_url_origin(url):
+    from urllib.parse import urlsplit
+    from azure.cli.core.azclierror import InvalidArgumentValueError
+
+    try:
+        parsed = urlsplit(url)
+        if parsed.scheme not in ('https', 'http') or not parsed.hostname or parsed.fragment:
+            raise ValueError()
+        if (parsed.username is not None or parsed.password is not None or
+                '\\' in url or any(ord(char) <= 32 for char in url)):
+            raise ValueError()
+        port = parsed.port
+        return parsed.scheme, parsed.hostname, port if port is not None else (443 if parsed.scheme == 'https' else 80)
+    except ValueError:
+        raise InvalidArgumentValueError(
+            'Invalid storage endpoint URL. Use an account endpoint without user information or a fragment.') from None
+
+
+def _storage_account_urls(cmd, account_name, service, connection_string=None):
+    from azure.cli.core.azclierror import InvalidArgumentValueError
+
+    services = ('blob', 'dfs') if service == 'blob' else ('file',)
+    if connection_string:
+        try:
+            settings = {key.upper(): value for key, value in validate_key_value_pairs(connection_string).items()}
+        except ValueError:
+            raise InvalidArgumentValueError('Invalid storage connection string.') from None
+        client_types = {
+            'blob': (ResourceType.DATA_STORAGE_BLOB, '_blob_service_client#BlobServiceClient'),
+            'dfs': (ResourceType.DATA_STORAGE_FILEDATALAKE, '_data_lake_service_client#DataLakeServiceClient'),
+            'file': (ResourceType.DATA_STORAGE_FILESHARE, '_share_service_client#ShareServiceClient')
+        }
+        urls = []
+        for endpoint_service in services:
+            endpoint_key = 'BLOBENDPOINT' if endpoint_service == 'dfs' else endpoint_service.upper() + 'ENDPOINT'
+            if not any(settings.get(key) for key in ('ACCOUNTNAME', 'USEDEVELOPMENTSTORAGE', endpoint_key)):
+                continue
+            client_type = get_sdk(cmd.cli_ctx, *client_types[endpoint_service])
+            try:
+                with client_type.from_connection_string(connection_string) as client:
+                    urls.append(client.primary_endpoint)
+                    if client.secondary_hostname:
+                        urls.append(client.secondary_endpoint)
+            except ValueError:
+                raise InvalidArgumentValueError('Invalid storage connection string.') from None
+        return urls
+
+    if not _is_valid_storage_account_name(account_name):
+        raise InvalidArgumentValueError('Provide a valid storage account name for the storage credentials.')
+    suffix = cmd.cli_ctx.cloud.suffixes.storage_endpoint
+    return ['https://{}.{}.{}'.format(account, endpoint_service, suffix)
+            for endpoint_service in services
+            for account in (account_name, account_name + '-secondary')]
+
+
 def get_url_with_sas(cmd, namespace, url=None, container=None, blob=None, share=None, file_path=None):
     import re
 
@@ -1824,20 +1879,28 @@ def get_url_with_sas(cmd, namespace, url=None, container=None, blob=None, share=
     storage_endpoint = cmd.cli_ctx.cloud.suffixes.storage_endpoint
     service = None
     if url is not None:
-        # validate source is uri or local path
-        storage_pattern = re.compile(r'https://(.*?)\.(blob|dfs|file).%s' % storage_endpoint)
-        result = re.findall(storage_pattern, url)
-        if result:  # source is URL
-            storage_info = result[0]
-            namespace.account_name = storage_info[0]
-            if storage_info[1] in ['blob', 'dfs']:
-                service = 'blob'
-            elif storage_info[1] in ['file']:
-                service = 'file'
-            else:
-                raise ValueError('{} is not valid storage endpoint.'.format(url))
-        else:
-            logger.info("%s is not Azure storage url.", url)
+        if '://' not in url:
+            return service, url
+        origin = _storage_url_origin(url)
+        storage_info = re.fullmatch(
+            r'([a-z0-9]{3,24})(-secondary)?\.(blob|dfs|file)\.' + re.escape(storage_endpoint.lower()),
+            origin[1])
+        if storage_info:
+            if not namespace.account_name:
+                namespace.account_name = get_config_value(cmd, 'storage', 'account', None) or storage_info[1]
+            service = 'file' if storage_info[3] == 'file' else 'blob'
+        connection_string = namespace.connection_string
+        if not connection_string and not namespace.account_key and not namespace.sas_token:
+            connection_string = get_config_value(cmd, 'storage', 'connection_string', None)
+        if connection_string:
+            for candidate_service in ('blob', 'file'):
+                if origin in [_storage_url_origin(endpoint) for endpoint in
+                              _storage_account_urls(cmd, namespace.account_name, candidate_service, connection_string)]:
+                    service = candidate_service
+                    namespace.connection_string = connection_string
+                    break
+        if service is None:
+            logger.info("The copy URL is not a configured Azure storage endpoint; no storage SAS will be attached.")
             return service, url
     # validate credential
     if container:
@@ -1896,8 +1959,19 @@ def _is_valid_uri(uri):
     return False
 
 
-def _add_sas_for_url(cmd, url, account_name, account_key, sas_token, service, resource_types, permissions):
+def _add_sas_for_url(cmd, url, account_name, account_key, sas_token, service, resource_types, permissions,
+                     connection_string=None):
     from azure.cli.command_modules.storage.azcopy.util import _generate_sas_token
+    from azure.cli.core.azclierror import InvalidArgumentValueError
+
+    if service is None:
+        return url
+    origin = _storage_url_origin(url)
+    endpoints = _storage_account_urls(cmd, account_name, service, connection_string)
+    if origin not in [_storage_url_origin(endpoint) for endpoint in endpoints]:
+        raise InvalidArgumentValueError(
+            'Cannot attach storage credentials to a URL outside the configured account endpoints.',
+            recommendation='Use the account endpoint or configure the endpoint in a trusted connection string.')
 
     if sas_token:
         sas_token = sas_token.lstrip('?')
@@ -1932,7 +2006,8 @@ def validate_azcopy_credential(cmd, namespace):
             namespace.destination = _add_sas_for_url(cmd, url=namespace.destination,
                                                      account_name=namespace.account_name,
                                                      account_key=namespace.account_key, sas_token=namespace.sas_token,
-                                                     service=service, resource_types='co', permissions='wac')
+                                                     service=service, resource_types='co', permissions='wac',
+                                                     connection_string=namespace.connection_string)
 
     if not _is_valid_uri(namespace.source):
         # determine if source account is same with destination
@@ -1955,7 +2030,8 @@ def validate_azcopy_credential(cmd, namespace):
         if not auth_mode_login:
             namespace.source = _add_sas_for_url(cmd, url=namespace.source, account_name=namespace.account_name,
                                                 account_key=namespace.account_key, sas_token=namespace.sas_token,
-                                                service=service, resource_types='sco', permissions='rl')
+                                                service=service, resource_types='sco', permissions='rl',
+                                                connection_string=namespace.connection_string)
 
 
 def is_directory(props):
@@ -1990,7 +2066,8 @@ def validate_fs_directory_upload_destination_url(cmd, namespace):
     else:
         namespace.destination = _add_sas_for_url(cmd, url=url, account_name=namespace.account_name,
                                                  account_key=namespace.account_key, sas_token=namespace.sas_token,
-                                                 service='blob', resource_types='co', permissions='rwdlac')
+                                                 service='blob', resource_types='co', permissions='rwdlac',
+                                                 connection_string=namespace.connection_string)
     del namespace.destination_fs
     del namespace.destination_path
 
@@ -2011,7 +2088,8 @@ def validate_fs_directory_download_source_url(cmd, namespace):
     else:
         namespace.source = _add_sas_for_url(cmd, url=url, account_name=namespace.account_name,
                                             account_key=namespace.account_key, sas_token=namespace.sas_token,
-                                            service='blob', resource_types='co', permissions='rl')
+                                            service='blob', resource_types='co', permissions='rl',
+                                            connection_string=namespace.connection_string)
     del namespace.source_fs
     del namespace.source_path
 
diff --git a/src/azure-cli/azure/cli/command_modules/storage/tests/latest/test_storage_azcopy_scenarios.py b/src/azure-cli/azure/cli/command_modules/storage/tests/latest/test_storage_azcopy_scenarios.py
index 95480cbbba3..e6b34cf2cd0 100644
--- a/src/azure-cli/azure/cli/command_modules/storage/tests/latest/test_storage_azcopy_scenarios.py
+++ b/src/azure-cli/azure/cli/command_modules/storage/tests/latest/test_storage_azcopy_scenarios.py
@@ -5,11 +5,41 @@
 import os
 import shutil
 from datetime import datetime, timedelta
-from azure.cli.testsdk import (StorageAccountPreparer, LiveScenarioTest, JMESPathCheck, ResourceGroupPreparer)
+from unittest import mock
+from azure.cli.testsdk import (
+    StorageAccountPreparer, LiveScenarioTest, ScenarioTest, JMESPathCheck, ResourceGroupPreparer)
+from azure.cli.core.azclierror import InvalidArgumentValueError
 from ..storage_test_util import StorageScenarioMixin, StorageTestFilesPreparer
 from knack.util import CLIError
 
 
+class StorageAzcopyEndpointTests(ScenarioTest):
+    def test_storage_copy_credential_endpoint_validation(self):
+        with mock.patch('azure.cli.command_modules.storage.operations.azcopy.AzCopy') as azcopy, \
+                mock.patch('azure.cli.command_modules.storage.azcopy.util._generate_sas_token',
+                           return_value='sig=test-signature') as generate:
+            with self.assertRaisesRegex(InvalidArgumentValueError, 'Cannot attach storage credentials'):
+                self.cmd(
+                    'storage copy --source ./source --destination https://another.blob.core.windows.net/c '
+                    '--account-name victim --account-key a2V5')
+            azcopy.assert_not_called()
+            generate.assert_not_called()
+
+            self.cmd('storage copy --source ./source --destination https://victim.blob.core.windows.net/c '
+                     '--account-name victim --account-key a2V5')
+            self.assertEqual(azcopy.return_value.copy.call_args[0],
+                             ('./source', 'https://victim.blob.core.windows.net/c?sig=test-signature'))
+
+            generate.reset_mock()
+            self.cmd(
+                'storage copy --source ./source '
+                '--destination https://victim.blob.core.windows.net.attacker.example/c '
+                '--account-name victim --account-key a2V5')
+            self.assertEqual(azcopy.return_value.copy.call_args[0],
+                             ('./source', 'https://victim.blob.core.windows.net.attacker.example/c'))
+            generate.assert_not_called()
+
+
 class StorageAzcopyTests(StorageScenarioMixin, LiveScenarioTest):
 
     @ResourceGroupPreparer()
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 fba10572ff3..b51304d643f 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
@@ -453,11 +453,199 @@ def _create_namespace(self, **kwargs):
         return ns
 
 
+class TestAzcopyEndpointValidation(unittest.TestCase):
+    def setUp(self):
+        self.cli = MockCLI()
+        self.cmd = MockCmd(self.cli)
+        patcher = mock.patch(
+            'azure.cli.command_modules.storage.azcopy.util._generate_sas_token',
+            return_value='sig=test-signature')
+        self.generate = patcher.start()
+        self.addCleanup(patcher.stop)
+        config = mock.patch('azure.cli.command_modules.storage._validators.get_config_value',
+                            side_effect=lambda cmd, section, key, default: default)
+        config.start()
+        self.addCleanup(config.stop)
+
+    def _add_sas(self, url, **kwargs):
+        from azure.cli.command_modules.storage._validators import _add_sas_for_url
+        args = dict(account_name='victim', account_key='a2V5', sas_token=None,
+                    service='blob', resource_types='co', permissions='wac')
+        args.update(kwargs)
+        return _add_sas_for_url(self.cmd, url, **args)
+
+    def test_reject_untrusted_sas_destinations(self):
+        from azure.cli.core.azclierror import InvalidArgumentValueError
+        urls = [
+            'https://victim.blob.core.windows.net.attacker.example/c',
+            'https://victim.blob.core.windows.net@attacker.example/c',
+            'https://attacker.example/https://victim.blob.core.windows.net/c',
+            'https://another.blob.core.windows.net/c',
+            'https://victim.blob.core.windows.net:8443/c',
+            'https://victim.blob.core.windows.net:0/c',
+            'http://victim.blob.core.windows.net/c',
+            'https://user@victim.blob.core.windows.net/c',
+            'https://victim.blob.core.windows.net\\@attacker.example/c',
+            'https://victim.blob.core.windows.net\n.attacker.example/c',
+            'https://victim.blob.core.windows.net/c#fragment',
+            'https://victim.blob.core.windows.net:invalid/c',
+        ]
+        for url in urls:
+            for sas in (None, 'sig=existing-secret'):
+                with self.subTest(url=url, supplied_sas=bool(sas)):
+                    with self.assertRaises(InvalidArgumentValueError) as error:
+                        self._add_sas(url, sas_token=sas)
+                    self.assertNotIn('existing-secret', str(error.exception))
+        self.generate.assert_not_called()
+
+    def test_accept_account_endpoints(self):
+        for service in ('blob', 'dfs', 'file'):
+            for account in ('victim', 'victim-secondary'):
+                url = 'https://{}.{}.core.windows.net/c'.format(account, service)
+                with self.subTest(url=url):
+                    result = self._add_sas(url, service='file' if service == 'file' else 'blob')
+                    self.assertEqual(result, url + '?sig=test-signature')
+        url = 'https://VICTIM.BLOB.CORE.WINDOWS.NET:443/c'
+        self.assertEqual(self._add_sas(url, sas_token='?sig=provided'), url + '?sig=provided')
+
+    def test_accept_active_cloud_endpoint(self):
+        self.cli.cloud = Namespace(suffixes=Namespace(storage_endpoint='storage.example.test'))
+        url = 'https://victim.blob.storage.example.test/c'
+        self.assertEqual(self._add_sas(url), url + '?sig=test-signature')
+
+    def test_custom_connection_endpoint(self):
+        from azure.cli.core.azclierror import InvalidArgumentValueError
+        connection = ('DefaultEndpointsProtocol=https;AccountName=victim;AccountKey=a2V5;'
+                      'BlobEndpoint=https://configured.example.test')
+        url = 'https://configured.example.test/c'
+        self.assertEqual(self._add_sas(url, connection_string=connection), url + '?sig=test-signature')
+        with self.assertRaises(InvalidArgumentValueError):
+            self._add_sas('https://configured.example.test.attacker.example/c',
+                          connection_string=connection)
+
+    def test_connection_string_suffix_and_explicit_http_endpoint(self):
+        connection = ('DefaultEndpointsProtocol=https;AccountName=victim;AccountKey=a2V5;'
+                      'EndpointSuffix=storage.example.test')
+        url = 'https://victim.blob.storage.example.test/c'
+        self.assertEqual(self._add_sas(url, connection_string=connection), url + '?sig=test-signature')
+        connection = ('AccountName=victim;AccountKey=a2V5;'
+                      'BlobEndpoint=http://127.0.0.1:10000/victim')
+        url = 'http://127.0.0.1:10000/victim/c'
+        self.assertEqual(self._add_sas(url, connection_string=connection), url + '?sig=test-signature')
+
+    def _copy_namespace(self, **kwargs):
+        args = dict(source='./source', destination='./destination',
+                    destination_container=None, destination_blob=None,
+                    destination_share=None, destination_file_path=None,
+                    source_container=None, source_blob=None, source_share=None, source_file_path=None,
+                    source_account_name=None, source_account_key=None, source_connection_string=None,
+                    source_sas=None, account_name=None, account_key=None, connection_string=None,
+                    sas_token=None, auth_mode=None, _cmd=self.cmd)
+        args.update(kwargs)
+        return Namespace(**args)
+
+    def test_copy_untrusted_urls_never_receive_sas(self):
+        from azure.cli.command_modules.storage._validators import validate_azcopy_credential
+        for url in ('https://victim.blob.core.windows.net.attacker.example/c',
+                    'https://attacker.example/https://victim.blob.core.windows.net/c',
+                    'https://bucket.s3.amazonaws.com/c'):
+            for side in ('source', 'destination'):
+                with self.subTest(url=url, side=side):
+                    ns = self._copy_namespace(
+                        **{side: url}, account_name='victim', account_key='a2V5', sas_token='sig=secret',
+                        source_account_name='victim', source_account_key='a2V5', source_sas='sig=secret')
+                    with mock.patch(
+                            'azure.cli.command_modules.storage._validators.validate_client_parameters') as validate:
+                        validate_azcopy_credential(self.cmd, ns)
+                    self.assertEqual(getattr(ns, side), url)
+                    validate.assert_not_called()
+        self.generate.assert_not_called()
+
+    def test_copy_keeps_explicit_credential_account(self):
+        from azure.cli.command_modules.storage._validators import validate_azcopy_credential
+        from azure.cli.core.azclierror import InvalidArgumentValueError
+        for side in ('source', 'destination'):
+            with self.subTest(side=side):
+                ns = self._copy_namespace(
+                    **{side: 'https://another.blob.core.windows.net/c'},
+                    account_name='victim', account_key='a2V5',
+                    source_account_name='victim', source_account_key='a2V5')
+                with mock.patch('azure.cli.command_modules.storage._validators.validate_client_parameters'):
+                    with self.assertRaises(InvalidArgumentValueError):
+                        validate_azcopy_credential(self.cmd, ns)
+                self.assertEqual(ns.account_name, 'victim')
+        self.generate.assert_not_called()
+
+    def test_copy_configured_endpoint(self):
+        from azure.cli.command_modules.storage._validators import validate_azcopy_credential
+        connection = ('AccountName=victim;AccountKey=a2V5;'
+                      'BlobEndpoint=https://configured.example.test')
+        for side in ('source', 'destination'):
+            with self.subTest(side=side):
+                ns = self._copy_namespace(
+                    **{side: 'https://configured.example.test/c'}, connection_string=connection,
+                    source_connection_string=connection)
+                validate_azcopy_credential(self.cmd, ns)
+                self.assertEqual(getattr(ns, side), 'https://configured.example.test/c?sig=test-signature')
+
+    def test_copy_sas_only_connection_strings(self):
+        from azure.cli.command_modules.storage._validators import validate_azcopy_credential
+        for endpoint_key, host in (('BlobEndpoint', 'blob.example.test'), ('FileEndpoint', 'file.example.test')):
+            connection = '{}=https://{};SharedAccessSignature=sig=provided'.format(endpoint_key, host)
+            for side in ('source', 'destination'):
+                for url, expected in (
+                        ('https://{}/c'.format(host), 'https://{}/c?sig=provided'.format(host)),
+                        ('https://bucket.s3.amazonaws.com/c', 'https://bucket.s3.amazonaws.com/c')):
+                    with self.subTest(endpoint_key=endpoint_key, side=side, url=url):
+                        ns = self._copy_namespace(**{side: url}, connection_string=connection,
+                                                  source_connection_string=connection)
+                        validate_azcopy_credential(self.cmd, ns)
+                        self.assertEqual(getattr(ns, side), expected)
+        self.generate.assert_not_called()
+
+    def test_fs_directory_callers_validate_endpoint(self):
+        from azure.cli.command_modules.storage._validators import (
+            validate_fs_directory_upload_destination_url, validate_fs_directory_download_source_url)
+        from azure.cli.core.azclierror import InvalidArgumentValueError
+        connection = ('AccountName=victim;AccountKey=a2V5;'
+                      'BlobEndpoint=https://configured.example.test')
+        for validator, side, resource_types, permissions in (
+                (validate_fs_directory_upload_destination_url, 'destination', 'co', 'rwdlac'),
+                (validate_fs_directory_download_source_url, 'source', 'co', 'rl')):
+            for endpoint, valid in (('https://configured.example.test/fs', True),
+                                    ('https://configured.example.test.attacker.example/fs', False)):
+                with self.subTest(side=side, endpoint=endpoint):
+                    ns = Namespace(account_name='victim', account_key='a2V5', sas_token=None,
+                                   connection_string=connection,
+                                   **{side + '_fs': 'fs', side + '_path': None})
+                    with mock.patch('azure.cli.command_modules.storage._validators.cf_adls_file_system',
+                                    return_value=Namespace(url=endpoint)):
+                        self.generate.reset_mock()
+                        if valid:
+                            validator(self.cmd, ns)
+                            self.assertEqual(getattr(ns, side), endpoint + '?sig=test-signature')
+                            self.generate.assert_called_once_with(
+                                self.cmd, 'victim', 'a2V5', 'blob',
+                                resource_types=resource_types, permissions=permissions)
+                        else:
+                            with self.assertRaises(InvalidArgumentValueError):
+                                validator(self.cmd, ns)
+                            self.generate.assert_not_called()
+
+    def test_external_urls_do_not_receive_credentials(self):
+        from azure.cli.command_modules.storage._validators import _add_sas_for_url
+        url = 'https://bucket.s3.amazonaws.com/c'
+        result = _add_sas_for_url(self.cmd, url, 'victim', 'a2V5', 'sig=secret', None, 'co', 'wac')
+        self.assertEqual(result, url)
+        self.generate.assert_not_called()
+
+
 class TestAzcopyValidator(unittest.TestCase):
     def setUp(self):
         self.cli = MockCLI()
 
-    def test_validate_azcopy_credential(self):
+    @mock.patch('azure.cli.command_modules.storage._validators._query_account_key', return_value=None)
+    def test_validate_azcopy_credential(self, _query_key):
         from azure.cli.command_modules.storage._validators import validate_azcopy_credential
         ns = Namespace(source='https://testaccount1.s3.amazonaws.com/test/dir',
                        destination='https://testaccount2.blob.core.windows.net/test/blob',

From 5a46e4e9b07827946f93d4451f43339e559079f9 Mon Sep 17 00:00:00 2001
From: Yang An <yangan@microsoft.com>
Date: Fri, 16 Oct 2026 11:36:01 +1100
Subject: [PATCH 2/5] {Storage} Leave history generation to the release
 pipeline

Remove the manual changelog entry; retain the draft PR History Notes for automated release history generation.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 src/azure-cli/HISTORY.rst | 1 -
 1 file changed, 1 deletion(-)

diff --git a/src/azure-cli/HISTORY.rst b/src/azure-cli/HISTORY.rst
index fcd7198fb52..47e64c36723 100644
--- a/src/azure-cli/HISTORY.rst
+++ b/src/azure-cli/HISTORY.rst
@@ -96,7 +96,6 @@ Release History
 
 **Storage**
 
-* `az storage copy`, `az storage fs directory upload/download`: Validate configured account endpoints before attaching SAS credentials
 * `az storage blob download-batch`: Ensure downloaded blobs stay within the destination directory (#34126)
 
 2.90.0

From ac7662d097480085845dd703cc567a788ae032fd Mon Sep 17 00:00:00 2001
From: Yang An <yangan@microsoft.com>
Date: Fri, 16 Oct 2026 11:49:18 +1100
Subject: [PATCH 3/5] {Storage} Reuse account endpoint URL builder

Use the existing get_account_url helper for primary and secondary standard endpoints without changing SAS destination validation.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 .../azure/cli/command_modules/storage/_validators.py          | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

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 620a9df02b2..cde82be4135 100644
--- a/src/azure-cli/azure/cli/command_modules/storage/_validators.py
+++ b/src/azure-cli/azure/cli/command_modules/storage/_validators.py
@@ -1824,6 +1824,7 @@ def _storage_url_origin(url):
 
 def _storage_account_urls(cmd, account_name, service, connection_string=None):
     from azure.cli.core.azclierror import InvalidArgumentValueError
+    from ._client_factory import get_account_url
 
     services = ('blob', 'dfs') if service == 'blob' else ('file',)
     if connection_string:
@@ -1853,8 +1854,7 @@ def _storage_account_urls(cmd, account_name, service, connection_string=None):
 
     if not _is_valid_storage_account_name(account_name):
         raise InvalidArgumentValueError('Provide a valid storage account name for the storage credentials.')
-    suffix = cmd.cli_ctx.cloud.suffixes.storage_endpoint
-    return ['https://{}.{}.{}'.format(account, endpoint_service, suffix)
+    return [get_account_url(cmd.cli_ctx, account, endpoint_service)
             for endpoint_service in services
             for account in (account_name, account_name + '-secondary')]
 

From fed087f0d04c94f25ce27986412fcaa4a38f47c3 Mon Sep 17 00:00:00 2001
From: Yang An <yangan@microsoft.com>
Date: Fri, 16 Oct 2026 12:07:47 +1100
Subject: [PATCH 4/5] {Storage} Preserve spaces in SAS wildcard resource paths

Restrict space rejection to URL authorities and leading whitespace while continuing to reject control characters. Cover Blob and File wildcard paths in both copy directions with generated and supplied SAS credentials.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 .../command_modules/storage/_validators.py    |  4 ++-
 .../tests/latest/test_storage_validators.py   | 30 +++++++++++++++++++
 2 files changed, 33 insertions(+), 1 deletion(-)

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 cde82be4135..bc5b0df99fa 100644
--- a/src/azure-cli/azure/cli/command_modules/storage/_validators.py
+++ b/src/azure-cli/azure/cli/command_modules/storage/_validators.py
@@ -1813,7 +1813,9 @@ def _storage_url_origin(url):
         if parsed.scheme not in ('https', 'http') or not parsed.hostname or parsed.fragment:
             raise ValueError()
         if (parsed.username is not None or parsed.password is not None or
-                '\\' in url or any(ord(char) <= 32 for char in url)):
+                '\\' in url or any(ord(char) < 32 for char in url)):
+            raise ValueError()
+        if url.startswith(' ') or ' ' in parsed.netloc:
             raise ValueError()
         port = parsed.port
         return parsed.scheme, parsed.hostname, port if port is not None else (443 if parsed.scheme == 'https' else 80)
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 b51304d643f..1e8a18d7e05 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
@@ -489,6 +489,11 @@ def test_reject_untrusted_sas_destinations(self):
             'https://victim.blob.core.windows.net\n.attacker.example/c',
             'https://victim.blob.core.windows.net/c#fragment',
             'https://victim.blob.core.windows.net:invalid/c',
+            ' https://victim.blob.core.windows.net/c',
+            'https://victim.blob.core.windows.net /c',
+            'https://victim.blob.core.windows.net/my\tfolder/*',
+            'https://victim.blob.core.windows.net/my\nfolder/*',
+            'https://victim.blob.core.windows.net/my\rfolder/*',
         ]
         for url in urls:
             for sas in (None, 'sig=existing-secret'):
@@ -544,6 +549,31 @@ def _copy_namespace(self, **kwargs):
         args.update(kwargs)
         return Namespace(**args)
 
+    def test_copy_wildcard_paths_with_spaces(self):
+        from azure.cli.command_modules.storage._validators import validate_azcopy_credential
+        for side in ('source', 'destination'):
+            for service, resource, path in (
+                    ('blob', 'container', 'blob'), ('file', 'share', 'file_path')):
+                for sas in (None, 'sig=provided'):
+                    with self.subTest(side=side, service=service, supplied_sas=bool(sas)):
+                        ns = self._copy_namespace(
+                            **{side: None, side + '_' + resource: 'container',
+                               side + '_' + path: 'my folder/*'},
+                            account_name='victim', account_key='a2V5', sas_token=sas,
+                            source_account_name='victim', source_account_key='a2V5', source_sas=sas)
+                        self.generate.reset_mock()
+                        validate_azcopy_credential(self.cmd, ns)
+                        expected = 'https://victim.{}.core.windows.net/container/my folder/*?{}'.format(
+                            service, sas or 'sig=test-signature')
+                        self.assertEqual(getattr(ns, side), expected)
+                        if sas:
+                            self.generate.assert_not_called()
+                        else:
+                            self.generate.assert_called_once_with(
+                                self.cmd, 'victim', 'a2V5', service,
+                                resource_types='sco' if side == 'source' else 'co',
+                                permissions='rl' if side == 'source' else 'wac')
+
     def test_copy_untrusted_urls_never_receive_sas(self):
         from azure.cli.command_modules.storage._validators import validate_azcopy_credential
         for url in ('https://victim.blob.core.windows.net.attacker.example/c',

From f2cc635751913cf56ebcd70da022fd65d85fd79e Mon Sep 17 00:00:00 2001
From: Yang An <yangan@microsoft.com>
Date: Fri, 16 Oct 2026 12:55:21 +1100
Subject: [PATCH 5/5] {Storage} Share proven URL origin normalization

Move existing blob-copy origin normalization and comparison into storage_url_helpers. Reuse normalization in the AzCopy SAS trust gate while preserving credential-boundary validation and existing blob-copy semantics. Cover equivalent DNS, IDNA and IPv6 origins.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 .../command_modules/storage/_validators.py    |  7 ++--
 .../storage/operations/blob.py                | 36 +------------------
 .../storage/storage_url_helpers.py            | 36 +++++++++++++++++++
 .../tests/latest/test_storage_validators.py   | 29 +++++++++++++++
 4 files changed, 71 insertions(+), 37 deletions(-)

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 bc5b0df99fa..f3eb16a1084 100644
--- a/src/azure-cli/azure/cli/command_modules/storage/_validators.py
+++ b/src/azure-cli/azure/cli/command_modules/storage/_validators.py
@@ -1807,6 +1807,7 @@ def validate_or_policy(cmd, namespace):
 def _storage_url_origin(url):
     from urllib.parse import urlsplit
     from azure.cli.core.azclierror import InvalidArgumentValueError
+    from .storage_url_helpers import normalize_url_origin
 
     try:
         parsed = urlsplit(url)
@@ -1817,8 +1818,10 @@ def _storage_url_origin(url):
             raise ValueError()
         if url.startswith(' ') or ' ' in parsed.netloc:
             raise ValueError()
-        port = parsed.port
-        return parsed.scheme, parsed.hostname, port if port is not None else (443 if parsed.scheme == 'https' else 80)
+        origin = normalize_url_origin(url)
+        if origin is None:
+            raise ValueError()
+        return origin
     except ValueError:
         raise InvalidArgumentValueError(
             'Invalid storage endpoint URL. Use an account endpoint without user information or a fragment.') from None
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 fef4f163baf..003920cf719 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
@@ -20,6 +20,7 @@
 from knack.util import CLIError
 from .._transformers import transform_response_with_bytearray
 from ..util import get_datetime_from_string
+from ..storage_url_helpers import same_url_origin as _same_url_origin
 
 logger = get_logger(__name__)
 
@@ -1103,41 +1104,6 @@ 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(first_client, second_client):
     first_account = first_client.account_name
     second_account = second_client.account_name
diff --git a/src/azure-cli/azure/cli/command_modules/storage/storage_url_helpers.py b/src/azure-cli/azure/cli/command_modules/storage/storage_url_helpers.py
index 8cf7c4f0fe7..cc2f0cab264 100644
--- a/src/azure-cli/azure/cli/command_modules/storage/storage_url_helpers.py
+++ b/src/azure-cli/azure/cli/command_modules/storage/storage_url_helpers.py
@@ -4,11 +4,47 @@
 # --------------------------------------------------------------------------------------------
 
 """
+Help functions to normalize and compare storage URL origins.
 Help functions to extract storage account, container name, blob name and other information from
 container and/or blob URL.
 """
 
 
+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)
+
+
 # pylint: disable=too-few-public-methods, too-many-instance-attributes
 class StorageResourceIdentifier:
     def __init__(self, cloud, moniker):
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 1e8a18d7e05..53a1bc586bf 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
@@ -518,6 +518,23 @@ def test_accept_active_cloud_endpoint(self):
         url = 'https://victim.blob.storage.example.test/c'
         self.assertEqual(self._add_sas(url), url + '?sig=test-signature')
 
+    def test_accept_normalized_account_origins(self):
+        for url in (
+                'https://VICTIM.blob.core.windows.net:443/c',
+                'https://victim.blob.core.windows.net./c',
+                'https://victim.blob.core.windows.net\u3002/c'):
+            with self.subTest(url=url):
+                self.assertEqual(self._add_sas(url), url + '?sig=test-signature')
+
+    def test_accept_normalized_connection_origins(self):
+        for endpoint, url in (
+                ('https://xn--bcher-kva.example', 'https://b\u00fccher.example:443/c'),
+                ('http://[::1]:10000/account', 'http://[0:0:0:0:0:0:0:1]:10000/account/c'),
+                ('https://configured.example.test.', 'https://configured.example.test/c')):
+            with self.subTest(endpoint=endpoint, url=url):
+                connection = 'AccountName=victim;AccountKey=a2V5;BlobEndpoint=' + endpoint
+                self.assertEqual(self._add_sas(url, connection_string=connection), url + '?sig=test-signature')
+
     def test_custom_connection_endpoint(self):
         from azure.cli.core.azclierror import InvalidArgumentValueError
         connection = ('DefaultEndpointsProtocol=https;AccountName=victim;AccountKey=a2V5;'
@@ -591,6 +608,18 @@ def test_copy_untrusted_urls_never_receive_sas(self):
                     validate.assert_not_called()
         self.generate.assert_not_called()
 
+    def test_copy_normalized_account_origins(self):
+        from azure.cli.command_modules.storage._validators import validate_azcopy_credential
+        for side in ('source', 'destination'):
+            for url in ('https://VICTIM.blob.core.windows.net:443/c',
+                        'https://victim.blob.core.windows.net./c',
+                        'https://victim.blob.core.windows.net\u3002/c'):
+                with self.subTest(side=side, url=url):
+                    ns = self._copy_namespace(
+                        **{side: url}, account_key='a2V5', source_account_key='a2V5')
+                    validate_azcopy_credential(self.cmd, ns)
+                    self.assertEqual(getattr(ns, side), url + '?sig=test-signature')
+
     def test_copy_keeps_explicit_credential_account(self):
         from azure.cli.command_modules.storage._validators import validate_azcopy_credential
         from azure.cli.core.azclierror import InvalidArgumentValueError