diff --git a/src/azure-cli/azure/cli/command_modules/keyvault/_client_factory.py b/src/azure-cli/azure/cli/command_modules/keyvault/_client_factory.py index 6d40452430a..5029080bea4 100644 --- a/src/azure-cli/azure/cli/command_modules/keyvault/_client_factory.py +++ b/src/azure-cli/azure/cli/command_modules/keyvault/_client_factory.py @@ -265,13 +265,16 @@ def data_plane_azure_keyvault_ekm_client(cli_ctx, command_args): def _prepare_data_plane_azure_keyvault_client(cli_ctx, command_args, resource_type): + from azure.cli.command_modules.keyvault._validators import validate_vault_uri + version = str(get_api_version(cli_ctx, resource_type)) - profile = Profile(cli_ctx=cli_ctx) - credential, _, _ = profile.get_login_credentials(subscription_id=cli_ctx.data.get('subscription_id')) vault_url = \ command_args.get('hsm_name', None) or \ command_args.get('vault_base_url', None) or \ command_args.get('identifier', None) if not vault_url: raise RequiredArgumentMissingError('Please specify --hsm-name or --id') + vault_url = validate_vault_uri(cli_ctx, vault_url) + profile = Profile(cli_ctx=cli_ctx) + credential, _, _ = profile.get_login_credentials(subscription_id=cli_ctx.data.get('subscription_id')) return vault_url, credential, version diff --git a/src/azure-cli/azure/cli/command_modules/keyvault/_completers.py b/src/azure-cli/azure/cli/command_modules/keyvault/_completers.py index 510c8523022..df28b0cfdfe 100644 --- a/src/azure-cli/azure/cli/command_modules/keyvault/_completers.py +++ b/src/azure-cli/azure/cli/command_modules/keyvault/_completers.py @@ -6,13 +6,15 @@ from azure.cli.core.decorators import Completer from azure.cli.core._profile import Profile +from ._validators import validate_vault_uri + def get_keyvault_name_completion_list(resource_name): @Completer def completer(cmd, prefix, namespace, **kwargs): # pylint: disable=unused-argument func_name = 'list_properties_of_{}s'.format(resource_name) - vault = namespace.vault_base_url + vault = validate_vault_uri(cmd.cli_ctx, namespace.vault_base_url) profile = Profile(cli_ctx=cmd.cli_ctx) credential, _, _ = profile.get_login_credentials(subscription_id=cmd.cli_ctx.data.get('subscription_id')) if resource_name == 'key': @@ -40,7 +42,7 @@ def get_keyvault_version_completion_list(resource_name): @Completer def completer(cmd, prefix, namespace, **kwargs): # pylint: disable=unused-argument func_name = 'list_properties_of_{}_versions'.format(resource_name) - vault = namespace.vault_base_url + vault = validate_vault_uri(cmd.cli_ctx, namespace.vault_base_url) profile = Profile(cli_ctx=cmd.cli_ctx) credential, _, _ = profile.get_login_credentials(subscription_id=cmd.cli_ctx.data.get('subscription_id')) if resource_name == 'key': diff --git a/src/azure-cli/azure/cli/command_modules/keyvault/_validators.py b/src/azure-cli/azure/cli/command_modules/keyvault/_validators.py index 668b83ae9a0..3cc725cd880 100644 --- a/src/azure-cli/azure/cli/command_modules/keyvault/_validators.py +++ b/src/azure-cli/azure/cli/command_modules/keyvault/_validators.py @@ -538,6 +538,79 @@ def get_hsm_base_url_type(cli_ctx): return _get_base_url_type(cli_ctx, service='hsm') +def _get_allowed_vault_dns_suffixes(cli_ctx): + from azure.cli.core.cloud import CloudSuffixNotSetException + + suffixes = [] + for suffix_name in ('keyvault_dns', 'mhsm_dns'): + try: + suffixes.append(getattr(cli_ctx.cloud.suffixes, suffix_name)) + except CloudSuffixNotSetException: # Not every cloud publishes both suffixes + pass + + # Escape hatch for private/disconnected footprints whose suffixes aren't in the cloud metadata. + configured = cli_ctx.config.get('keyvault', 'allowed_dns_suffixes', None) + if configured: + suffixes.extend(configured.split(',')) + + normalized = [] + for suffix in suffixes: + suffix = (suffix or '').strip().rstrip('.').lower() + if suffix: + normalized.append(suffix if suffix.startswith('.') else '.' + suffix) + return normalized + + +def validate_vault_uri(cli_ctx, uri): + """Validate a vault URI before it is used as an authentication target, and return its origin. + + Azure CLI builds its data-plane clients with `verify_challenge_resource=False`, so the SDK will + mint a token for whatever resource the contacted host asks for. Validating the host here is the + compensating control required by https://aka.ms/azsdk/blog/vault-uri. + """ + from urllib.parse import urlparse + + def _invalid(reason): + return InvalidArgumentValueError( + "'{}' is not a valid Key Vault or Managed HSM URI: {}.".format(uri, reason)) + + if not uri or not isinstance(uri, str): + raise _invalid('a value is required') + + try: + parsed = urlparse(uri) + hostname = parsed.hostname + parsed.port # pylint: disable=pointless-statement # raises ValueError on a malformed port + except ValueError: + raise _invalid('it is not a well-formed absolute URI') # pylint: disable=raise-missing-from + + if parsed.scheme.lower() != 'https': + raise _invalid('the scheme must be https') + if not hostname: + raise _invalid('it is not a well-formed absolute URI') + if parsed.username or parsed.password: + raise _invalid('it must not contain credentials') + + # Compare against the suffixes with a leading '.' so that look-alikes such as + # 'maliciousvault.azure.net' don't match '.vault.azure.net'. + hostname = hostname.rstrip('.').lower() + # urlparse and the HTTP transport disagree on characters such as '\', which urlparse keeps in the + # host but the transport treats as a path separator. Requiring well-formed DNS labels keeps the + # name validated here identical to the one actually dialled. + if not all(re.fullmatch(r'[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?', label) for label in hostname.split('.')): + raise _invalid('the host is not a valid DNS name') + allowed = _get_allowed_vault_dns_suffixes(cli_ctx) + if not any(len(hostname) > len(suffix) and hostname.endswith(suffix) for suffix in allowed): + raise InvalidArgumentValueError( + "'{}' is not a recognized Key Vault or Managed HSM host in cloud '{}'. Azure CLI will not " + "send an access token to it. Expected a host ending in: {}. If this is a private or " + "disconnected deployment, register its suffixes with " + "'az config set keyvault.allowed_dns_suffixes=[,]'.".format( + uri, cli_ctx.cloud.name, ', '.join(allowed) or '')) + + return 'https://{}'.format(parsed.netloc) + + def _construct_vnet(cmd, resource_group_name, vnet_name, subnet_name): from azure.mgmt.core.tools import resource_id from azure.cli.core.commands.client_factory import get_subscription_id diff --git a/src/azure-cli/azure/cli/command_modules/keyvault/tests/latest/test_validators.py b/src/azure-cli/azure/cli/command_modules/keyvault/tests/latest/test_validators.py new file mode 100644 index 00000000000..a34d7e0eafd --- /dev/null +++ b/src/azure-cli/azure/cli/command_modules/keyvault/tests/latest/test_validators.py @@ -0,0 +1,167 @@ +# -------------------------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for license information. +# -------------------------------------------------------------------------------------------- + +import unittest + +from azure.cli.core.azclierror import InvalidArgumentValueError +from azure.cli.core.cloud import ( + AZURE_CHINA_CLOUD, + AZURE_PUBLIC_CLOUD, + AZURE_US_GOV_CLOUD, + Cloud, + CloudSuffixes, +) + +from azure.cli.command_modules.keyvault._validators import validate_vault_uri + + +class _Config: + def __init__(self, allowed_dns_suffixes=None): + self._allowed_dns_suffixes = allowed_dns_suffixes + + def get(self, section, option, fallback=None): + if section == 'keyvault' and option == 'allowed_dns_suffixes': + return self._allowed_dns_suffixes or fallback + return fallback + + +class _CliCtx: # pylint: disable=too-few-public-methods + def __init__(self, cloud=AZURE_PUBLIC_CLOUD, allowed_dns_suffixes=None): + self.cloud = cloud + self.config = _Config(allowed_dns_suffixes) + + +class VaultUriValidationTest(unittest.TestCase): + + def test_accepts_key_vault_and_mhsm_hosts(self): + cli_ctx = _CliCtx() + for uri in [ + 'https://myvault.vault.azure.net', + 'https://myvault.vault.azure.net/', + 'https://myhsm.managedhsm.azure.net', + # Managed HSM may use multi-level names for region support. + 'https://myhsm.eastus.managedhsm.azure.net', + ]: + self.assertTrue(validate_vault_uri(cli_ctx, uri).startswith('https://')) + + def test_normalizes_to_origin(self): + cli_ctx = _CliCtx() + self.assertEqual( + validate_vault_uri(cli_ctx, 'https://myvault.vault.azure.net/secrets/s/version'), + 'https://myvault.vault.azure.net') + + def test_rejects_foreign_host(self): + cli_ctx = _CliCtx() + for uri in [ + 'https://attacker.example/secrets/leak', + 'https://127.0.0.1:8443/secrets/leak', + 'https://vault.azure.net.attacker.example/secrets/leak', + ]: + with self.assertRaises(InvalidArgumentValueError): + validate_vault_uri(cli_ctx, uri) + + def test_rejects_suffix_look_alike(self): + # Must not match '.vault.azure.net' without the separating dot. + cli_ctx = _CliCtx() + with self.assertRaises(InvalidArgumentValueError): + validate_vault_uri(cli_ctx, 'https://maliciousvault.azure.net/secrets/s') + + def test_rejects_vault_name_template_escape(self): + # '--vault-name ' is concatenated as 'https://{name}.vault.azure.net'; a name + # carrying URL syntax must not be able to select a different host. + cli_ctx = _CliCtx() + for name in ['attacker.example#', 'attacker.example?', 'attacker.example/', 'user@attacker.example#']: + with self.assertRaises(InvalidArgumentValueError): + validate_vault_uri(cli_ctx, 'https://{}.vault.azure.net'.format(name)) + + def test_rejects_backslash_host_confusion(self): + # urlparse keeps '\' in the host, but the HTTP transport treats it as a path separator and + # would dial 'attacker.example'. The two parsers must not be allowed to disagree. + cli_ctx = _CliCtx() + for uri in [ + 'https://attacker.example\\.vault.azure.net', + 'https://attacker.example\\.managedhsm.azure.net', + 'https://attacker.example\\@.vault.azure.net', + ]: + with self.assertRaises(InvalidArgumentValueError): + validate_vault_uri(cli_ctx, uri) + + def test_rejects_invalid_dns_labels(self): + cli_ctx = _CliCtx() + for uri in [ + 'https://.attacker.example.vault.azure.net', # empty leading label + 'https://a..b.vault.azure.net', # empty inner label + 'https://-bad.vault.azure.net', # label may not start with '-' + 'https://bad-.vault.azure.net', # label may not end with '-' + 'https://attacker.example;.vault.azure.net', + 'https://attacker.example,.vault.azure.net', + 'https://attacker.example%2f.vault.azure.net', + 'https://{}.vault.azure.net'.format('a' * 64), # label longer than 63 chars + ]: + with self.assertRaises(InvalidArgumentValueError): + validate_vault_uri(cli_ctx, uri) + + def test_accepts_valid_dns_labels(self): + cli_ctx = _CliCtx() + for uri in [ + 'https://a.vault.azure.net', + 'https://my-vault-01.vault.azure.net', + 'https://{}.vault.azure.net'.format('a' * 63), + ]: + self.assertEqual(validate_vault_uri(cli_ctx, uri), uri) + + def test_rejects_non_https(self): + cli_ctx = _CliCtx() + for uri in ['http://myvault.vault.azure.net', 'ftp://myvault.vault.azure.net']: + with self.assertRaises(InvalidArgumentValueError): + validate_vault_uri(cli_ctx, uri) + + def test_rejects_credentials_in_uri(self): + cli_ctx = _CliCtx() + with self.assertRaises(InvalidArgumentValueError): + validate_vault_uri(cli_ctx, 'https://user:pass@myvault.vault.azure.net') + + def test_rejects_malformed(self): + cli_ctx = _CliCtx() + for uri in [None, '', 'not-a-uri', 'https://', 'https://myvault.vault.azure.net:notaport']: + with self.assertRaises(InvalidArgumentValueError): + validate_vault_uri(cli_ctx, uri) + + def test_sovereign_clouds(self): + for cloud, uri in [ + (AZURE_CHINA_CLOUD, 'https://myvault.vault.azure.cn'), + (AZURE_CHINA_CLOUD, 'https://myhsm.managedhsm.azure.cn'), + (AZURE_US_GOV_CLOUD, 'https://myvault.vault.usgovcloudapi.net'), + (AZURE_US_GOV_CLOUD, 'https://myhsm.managedhsm.usgovcloudapi.net'), + ]: + cli_ctx = _CliCtx(cloud=cloud) + self.assertEqual(validate_vault_uri(cli_ctx, uri), uri) + + def test_rejects_other_clouds_suffix(self): + cli_ctx = _CliCtx(cloud=AZURE_CHINA_CLOUD) + with self.assertRaises(InvalidArgumentValueError): + validate_vault_uri(cli_ctx, 'https://myvault.vault.azure.net') + + def test_configured_suffix_allow_list(self): + cloud = Cloud('PrivateCloud', suffixes=CloudSuffixes()) + cli_ctx = _CliCtx(cloud=cloud, allowed_dns_suffixes='.vault.contoso.local,managedhsm.contoso.local') + self.assertEqual( + validate_vault_uri(cli_ctx, 'https://myvault.vault.contoso.local'), + 'https://myvault.vault.contoso.local') + self.assertEqual( + validate_vault_uri(cli_ctx, 'https://myhsm.managedhsm.contoso.local'), + 'https://myhsm.managedhsm.contoso.local') + with self.assertRaises(InvalidArgumentValueError): + validate_vault_uri(cli_ctx, 'https://attacker.example') + + def test_cloud_without_suffixes_rejects_everything(self): + cloud = Cloud('PrivateCloud', suffixes=CloudSuffixes()) + cli_ctx = _CliCtx(cloud=cloud) + with self.assertRaises(InvalidArgumentValueError): + validate_vault_uri(cli_ctx, 'https://myvault.vault.azure.net') + + +if __name__ == '__main__': + unittest.main()