[Storage] Validate account endpoints before attaching SAS credentials

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

@@ -1804,6 +1804,66 @@ 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
+    from .storage_url_helpers import normalize_url_origin
+
+    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()
+        if url.startswith(' ') or ' ' in parsed.netloc:
+            raise ValueError()
+        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
+
+
+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:
+        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.')
+    return [get_account_url(cmd.cli_ctx, account, endpoint_service)
+            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 +1884,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 +1964,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 +2011,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 +2035,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 +2071,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 +2093,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
 

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

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):

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()