From 1bf335336a7bf64974cef0fde6a22330ffc60b4b Mon Sep 17 00:00:00 2001 From: Yash Date: Tue, 25 Aug 2026 10:42:02 +1000 Subject: [PATCH 1/3] Add Validation Logic --- .../keyvault/_client_factory.py | 7 +- .../command_modules/keyvault/_completers.py | 6 +- .../command_modules/keyvault/_validators.py | 68 +++++++++ .../keyvault/tests/latest/test_validators.py | 131 ++++++++++++++++++ 4 files changed, 208 insertions(+), 4 deletions(-) create mode 100644 src/azure-cli/azure/cli/command_modules/keyvault/tests/latest/test_validators.py 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..78817443a03 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,74 @@ 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() + 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..60e124a90bd --- /dev/null +++ b/src/azure-cli/azure/cli/command_modules/keyvault/tests/latest/test_validators.py @@ -0,0 +1,131 @@ +# -------------------------------------------------------------------------------------------- +# 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_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() From 1f67cb50dc48933d4c33dee1cc89c0583f19507e Mon Sep 17 00:00:00 2001 From: Yash Date: Tue, 25 Aug 2026 10:46:51 +1000 Subject: [PATCH 2/3] Add backslash patch --- .../command_modules/keyvault/_validators.py | 5 +++ .../keyvault/tests/latest/test_validators.py | 36 +++++++++++++++++++ 2 files changed, 41 insertions(+) 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 78817443a03..3cc725cd880 100644 --- a/src/azure-cli/azure/cli/command_modules/keyvault/_validators.py +++ b/src/azure-cli/azure/cli/command_modules/keyvault/_validators.py @@ -594,6 +594,11 @@ def _invalid(reason): # 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( 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 index 60e124a90bd..a34d7e0eafd 100644 --- 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 @@ -76,6 +76,42 @@ def test_rejects_vault_name_template_escape(self): 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']: From 4d1ec36d545ee71f8be3a3dfcee1eb509be4e63b Mon Sep 17 00:00:00 2001 From: Yash Date: Mon, 31 Aug 2026 13:11:54 +1000 Subject: [PATCH 3/3] Added test validation --- .../command_modules/keyvault/_validators.py | 21 +++-- .../keyvault/tests/latest/test_validators.py | 86 +++++++++++++++++++ 2 files changed, 99 insertions(+), 8 deletions(-) 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 3cc725cd880..8117d088c11 100644 --- a/src/azure-cli/azure/cli/command_modules/keyvault/_validators.py +++ b/src/azure-cli/azure/cli/command_modules/keyvault/_validators.py @@ -570,9 +570,12 @@ def validate_vault_uri(cli_ctx, uri): """ from urllib.parse import urlparse + # Never echo credentials that were embedded in the URI back to the terminal or logs. + display = re.sub(r'(?<=//)[^/@]*@', '***@', uri, count=1) if isinstance(uri, str) else uri + def _invalid(reason): return InvalidArgumentValueError( - "'{}' is not a valid Key Vault or Managed HSM URI: {}.".format(uri, reason)) + "'{}' is not a valid Key Vault or Managed HSM URI: {}.".format(display, reason)) if not uri or not isinstance(uri, str): raise _invalid('a value is required') @@ -580,7 +583,7 @@ def _invalid(reason): try: parsed = urlparse(uri) hostname = parsed.hostname - parsed.port # pylint: disable=pointless-statement # raises ValueError on a malformed port + port = parsed.port # raises ValueError on a malformed port except ValueError: raise _invalid('it is not a well-formed absolute URI') # pylint: disable=raise-missing-from @@ -588,12 +591,13 @@ def _invalid(reason): 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: + if parsed.username is not None or parsed.password is not None: 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() + hostname = hostname[:-1] if hostname.endswith('.') else hostname # drop a single DNS root terminator + hostname = hostname.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. @@ -603,12 +607,13 @@ def _invalid(reason): 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 " + "send an access token to it. Expected a host ending in: {}. For a private or disconnected " + "deployment, pass the full vault URI with --id and register its suffix with " "'az config set keyvault.allowed_dns_suffixes=[,]'.".format( - uri, cli_ctx.cloud.name, ', '.join(allowed) or '')) + display, cli_ctx.cloud.name, ', '.join(allowed) or '')) - return 'https://{}'.format(parsed.netloc) + # Rebuild from the validated host so that nothing else in the authority survives into the URL. + return 'https://{}{}'.format(hostname, ':{}'.format(port) if port else '') def _construct_vnet(cmd, resource_group_name, vnet_name, subnet_name): 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 index a34d7e0eafd..19e88a8c5dc 100644 --- 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 @@ -4,6 +4,7 @@ # -------------------------------------------------------------------------------------------- import unittest +from unittest import mock from azure.cli.core.azclierror import InvalidArgumentValueError from azure.cli.core.cloud import ( @@ -13,6 +14,7 @@ Cloud, CloudSuffixes, ) +from azure.cli.core.profiles import ResourceType from azure.cli.command_modules.keyvault._validators import validate_vault_uri @@ -31,6 +33,7 @@ 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) + self.data = {} class VaultUriValidationTest(unittest.TestCase): @@ -123,6 +126,44 @@ def test_rejects_credentials_in_uri(self): with self.assertRaises(InvalidArgumentValueError): validate_vault_uri(cli_ctx, 'https://user:pass@myvault.vault.azure.net') + def test_rejects_empty_userinfo(self): + # urlparse reports empty userinfo as '' rather than None, so a truthiness check misses these. + cli_ctx = _CliCtx() + for uri in [ + 'https://@myvault.vault.azure.net', + 'https://:@myvault.vault.azure.net', + 'https://@attacker.example.vault.azure.net', + ]: + with self.assertRaises(InvalidArgumentValueError): + validate_vault_uri(cli_ctx, uri) + + def test_does_not_echo_credentials_in_error(self): + cli_ctx = _CliCtx() + with self.assertRaises(InvalidArgumentValueError) as ctx: + validate_vault_uri(cli_ctx, 'https://user:sup3rs3cret@attacker.example') + self.assertNotIn('sup3rs3cret', str(ctx.exception)) + self.assertIn('***@', str(ctx.exception)) + + def test_trailing_dots(self): + cli_ctx = _CliCtx() + # A single DNS root terminator is legitimate and is normalized away. + self.assertEqual( + validate_vault_uri(cli_ctx, 'https://myvault.vault.azure.net.'), + 'https://myvault.vault.azure.net') + # Anything beyond that leaves an empty label and is not a valid authority. + for uri in ['https://myvault.vault.azure.net..', 'https://myvault.vault.azure.net...']: + with self.assertRaises(InvalidArgumentValueError): + validate_vault_uri(cli_ctx, uri) + + def test_returned_origin_is_normalized(self): + cli_ctx = _CliCtx() + self.assertEqual( + validate_vault_uri(cli_ctx, 'https://MyVault.Vault.Azure.Net/secrets/s'), + 'https://myvault.vault.azure.net') + self.assertEqual( + validate_vault_uri(cli_ctx, 'https://myvault.vault.azure.net:8443/secrets/s'), + 'https://myvault.vault.azure.net:8443') + def test_rejects_malformed(self): cli_ctx = _CliCtx() for uri in [None, '', 'not-a-uri', 'https://', 'https://myvault.vault.azure.net:notaport']: @@ -163,5 +204,50 @@ def test_cloud_without_suffixes_rejects_everything(self): validate_vault_uri(cli_ctx, 'https://myvault.vault.azure.net') +class DataPlaneClientCredentialGuardTest(unittest.TestCase): + """An untrusted host must be rejected before any Azure AD token is acquired.""" + + @staticmethod + def _prepare(cli_ctx, command_args): + from azure.cli.command_modules.keyvault import _client_factory + + with mock.patch.object(_client_factory, 'Profile') as profile_cls, \ + mock.patch.object(_client_factory, 'get_api_version', return_value='7.4'): + profile_cls.return_value.get_login_credentials.return_value = ('credential', None, None) + raised = None + try: + _client_factory._prepare_data_plane_azure_keyvault_client( # pylint: disable=protected-access + cli_ctx, dict(command_args), ResourceType.DATA_KEYVAULT_SECRETS) + except InvalidArgumentValueError as ex: + raised = ex + return raised, profile_cls + + def test_no_token_acquired_for_untrusted_host(self): + cli_ctx = _CliCtx() + for command_args in [ + {'identifier': 'https://attacker.example/secrets/leak'}, + {'identifier': 'https://127.0.0.1:8443/secrets/leak'}, + {'identifier': 'https://maliciousvault.azure.net/secrets/leak'}, + {'vault_base_url': 'https://attacker.example'}, + {'vault_base_url': 'https://attacker.example#.vault.azure.net'}, + {'vault_base_url': 'https://attacker.example\\.vault.azure.net'}, + {'hsm_name': 'https://attacker.example'}, + ]: + raised, profile_cls = self._prepare(cli_ctx, command_args) + self.assertIsNotNone(raised, 'expected rejection for {}'.format(command_args)) + profile_cls.assert_not_called() + + def test_token_acquired_for_trusted_host(self): + cli_ctx = _CliCtx() + for command_args in [ + {'identifier': 'https://myvault.vault.azure.net/secrets/s'}, + {'vault_base_url': 'https://myvault.vault.azure.net'}, + {'hsm_name': 'https://myhsm.managedhsm.azure.net'}, + ]: + raised, profile_cls = self._prepare(cli_ctx, command_args) + self.assertIsNone(raised) + profile_cls.assert_called_once() + + if __name__ == '__main__': unittest.main()