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..f3eb16a1084 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,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 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_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..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 @@ -453,11 +453,258 @@ 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', + ' 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'): + 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_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;' + '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_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', + '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_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 + 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',