Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 28 additions & 19 deletions src/azure-cli/azure/cli/command_modules/appservice/custom.py
Original file line number Diff line number Diff line change
Expand Up @@ -6923,6 +6923,29 @@ def _validate_flex_ssl_params(is_flex, load_to_code, enable_using_msi, webapp, r
'Ensure the app\'s managed identity has the permission on the Key Vault.')


def _get_app_service_certificate_kv_secret_name(cmd, subscription_id, key_vault_certificate_name):
# App Service Certificate orders are exposed by the Microsoft.CertificateRegistration resource provider,
# which the WebSiteManagementClient SDK no longer surfaces, so query the ARM REST API directly.
management_hostname = cmd.cli_ctx.cloud.endpoints.resource_manager
request_url = "{}/subscriptions/{}/providers/Microsoft.CertificateRegistration/certificateOrders" \
"?api-version={}".format(management_hostname.strip('/'), subscription_id, VERSION_2022_09_01)
try:
while request_url:
response = send_raw_request(cmd.cli_ctx, method='get', url=request_url).json()
for asc in response.get('value', []):
if asc.get('name') == key_vault_certificate_name:
certificates = asc.get('properties', {}).get('certificates') or {}
cert = certificates.get(key_vault_certificate_name)
if cert:
return cert.get('keyVaultSecretName')
request_url = response.get('nextLink')
except Exception as ex: # pylint: disable=broad-except
logger.warning("Unable to check App Service Certificate orders (%s). If '%s' is an App Service "
"Certificate, the import may not resolve the correct Key Vault secret name.",
str(ex), key_vault_certificate_name)
return None


def import_ssl_cert(cmd, resource_group_name, key_vault, key_vault_certificate_name, name=None, certificate_name=None,
load_to_code=None, enable_using_msi=None):
Certificate = cmd.get_models('Certificate')
Expand Down Expand Up @@ -6965,28 +6988,14 @@ def import_ssl_cert(cmd, resource_group_name, key_vault, key_vault_certificate_n
kv_resource_group_name = kv_id_parts['resource_group']
kv_subscription = kv_id_parts['subscription']

# If in the public cloud, check if certificate is an app service certificate, in the same or a diferent
# subscription
# If in the public cloud, check if certificate is an app service certificate in the same or a different
# subscription. App Service Certificate orders belong to the Microsoft.CertificateRegistration provider,
# which is no longer exposed by the WebSiteManagementClient SDK, so the ARM REST API is queried directly.
kv_secret_name = None
cloud_type = cmd.cli_ctx.cloud.name
from azure.cli.core.commands.client_factory import get_subscription_id
subscription_id = get_subscription_id(cmd.cli_ctx)
if cloud_type.lower() == PUBLIC_CLOUD.lower():
if kv_subscription.lower() != subscription_id.lower():
cert_orders_client = get_mgmt_service_client(cmd.cli_ctx, ResourceType.MGMT_APPSERVICE,
subscription_id=kv_subscription)
else:
cert_orders_client = client

if hasattr(cert_orders_client, 'app_service_certificate_orders'):
ascs = cert_orders_client.app_service_certificate_orders.list(api_version=VERSION_2022_09_01)
for asc in ascs:
if asc.name == key_vault_certificate_name:
kv_secret_name = asc.certificates[key_vault_certificate_name].key_vault_secret_name
else:
logger.warning("Unable to check App Service Certificate orders. "
"If '%s' is an App Service Certificate, the import may not resolve "
"the correct Key Vault secret name.", key_vault_certificate_name)
kv_secret_name = _get_app_service_certificate_kv_secret_name(cmd, kv_subscription,
key_vault_certificate_name)

# if kv_secret_name is not populated, it is not an appservice certificate, proceed for KV certificates
kv_secret_name = kv_secret_name or key_vault_certificate_name
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,88 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------

"""
Unit tests for App Service Certificate order resolution used by `az webapp config ssl import`.

These validate that the Key Vault secret name is resolved via the ARM REST API for the
Microsoft.CertificateRegistration provider (which the WebSiteManagementClient SDK no longer
exposes), without requiring Azure connectivity.
"""

import unittest
from unittest.mock import MagicMock, patch

from azure.cli.command_modules.appservice.custom import _get_app_service_certificate_kv_secret_name

_CUSTOM_MOD = "azure.cli.command_modules.appservice.custom"


def _make_cmd():
cmd = MagicMock()
cmd.cli_ctx.cloud.endpoints.resource_manager = "https://management.azure.com"
return cmd


def _make_response(payload):
response = MagicMock()
response.json.return_value = payload
return response


class TestAppServiceCertificateSecretResolution(unittest.TestCase):
def test_resolves_secret_name_for_matching_order(self):
cmd = _make_cmd()
payload = {
"value": [
{
"name": "my-cert",
"properties": {
"certificates": {
"my-cert": {"keyVaultSecretName": "the-secret-name"}
}
},
}
]
}
with patch(f"{_CUSTOM_MOD}.send_raw_request", return_value=_make_response(payload)) as mock_req:
result = _get_app_service_certificate_kv_secret_name(cmd, "sub-id", "my-cert")

self.assertEqual(result, "the-secret-name")
called_url = mock_req.call_args.kwargs.get("url") or mock_req.call_args.args[-1]
self.assertIn("/subscriptions/sub-id/providers/Microsoft.CertificateRegistration/certificateOrders", called_url)

def test_returns_none_when_no_matching_order(self):
cmd = _make_cmd()
payload = {"value": [{"name": "other-cert", "properties": {"certificates": {}}}]}
with patch(f"{_CUSTOM_MOD}.send_raw_request", return_value=_make_response(payload)):
result = _get_app_service_certificate_kv_secret_name(cmd, "sub-id", "my-cert")

self.assertIsNone(result)

def test_follows_pagination_next_link(self):
cmd = _make_cmd()
page1 = {"value": [{"name": "other", "properties": {"certificates": {}}}], "nextLink": "https://next"}
page2 = {
"value": [
{"name": "my-cert", "properties": {"certificates": {"my-cert": {"keyVaultSecretName": "secret2"}}}}
]
}
with patch(f"{_CUSTOM_MOD}.send_raw_request",
side_effect=[_make_response(page1), _make_response(page2)]) as mock_req:
result = _get_app_service_certificate_kv_secret_name(cmd, "sub-id", "my-cert")

self.assertEqual(result, "secret2")
self.assertEqual(mock_req.call_count, 2)

def test_swallows_request_errors_and_returns_none(self):
cmd = _make_cmd()
with patch(f"{_CUSTOM_MOD}.send_raw_request", side_effect=Exception("boom")):
result = _get_app_service_certificate_kv_secret_name(cmd, "sub-id", "my-cert")

self.assertIsNone(result)


if __name__ == "__main__":
unittest.main()
Loading