[Key Vault] az keyvault: Add host validation for the resolved vault URI

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

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

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=<suffix>[,<suffix>]'.".format(
+                uri, cli_ctx.cloud.name, ', '.join(allowed) or '<none configured>'))
+
+    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

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