aboutsummaryrefslogtreecommitdiff
path: root/.venv/lib/python3.12/site-packages/azure/ai/ml/_azure_environments.py
blob: 51673b141ad13dcb55a1c7e40912c376429962b9 (about) (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
# ---------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# ---------------------------------------------------------

"""Metadata to interact with different Azure clouds."""

import logging
import os
from typing import Dict, List, Optional

from azure.ai.ml._utils.utils import _get_mfe_url_override
from azure.ai.ml.constants._common import AZUREML_CLOUD_ENV_NAME, ArmConstants
from azure.ai.ml.exceptions import MlException
from azure.core.rest import HttpRequest
from azure.mgmt.core import ARMPipelineClient

module_logger = logging.getLogger(__name__)


class AzureEnvironments:
    ENV_DEFAULT = "AzureCloud"
    ENV_US_GOVERNMENT = "AzureUSGovernment"
    ENV_CHINA = "AzureChinaCloud"


class EndpointURLS:  # pylint: disable=too-few-public-methods
    AZURE_PORTAL_ENDPOINT = "azure_portal"
    RESOURCE_MANAGER_ENDPOINT = "resource_manager"
    ACTIVE_DIRECTORY_ENDPOINT = "active_directory"
    AML_RESOURCE_ID = "aml_resource_id"
    STORAGE_ENDPOINT = "storage_endpoint"
    REGISTRY_DISCOVERY_ENDPOINT = "registry_discovery_endpoint"


class CloudArgumentKeys:
    CLOUD_METADATA = "cloud_metadata"


_environments = {
    AzureEnvironments.ENV_DEFAULT: {
        EndpointURLS.AZURE_PORTAL_ENDPOINT: "https://portal.azure.com/",
        EndpointURLS.RESOURCE_MANAGER_ENDPOINT: "https://management.azure.com/",
        EndpointURLS.ACTIVE_DIRECTORY_ENDPOINT: "https://login.microsoftonline.com/",
        EndpointURLS.AML_RESOURCE_ID: "https://ml.azure.com/",
        EndpointURLS.STORAGE_ENDPOINT: "core.windows.net",
        EndpointURLS.REGISTRY_DISCOVERY_ENDPOINT: "https://eastus.api.azureml.ms/",
    },
    AzureEnvironments.ENV_CHINA: {
        EndpointURLS.AZURE_PORTAL_ENDPOINT: "https://portal.azure.cn/",
        EndpointURLS.RESOURCE_MANAGER_ENDPOINT: "https://management.chinacloudapi.cn/",
        EndpointURLS.ACTIVE_DIRECTORY_ENDPOINT: "https://login.chinacloudapi.cn/",
        EndpointURLS.AML_RESOURCE_ID: "https://ml.azure.cn/",
        EndpointURLS.STORAGE_ENDPOINT: "core.chinacloudapi.cn",
        EndpointURLS.REGISTRY_DISCOVERY_ENDPOINT: "https://chinaeast2.api.ml.azure.cn/",
    },
    AzureEnvironments.ENV_US_GOVERNMENT: {
        EndpointURLS.AZURE_PORTAL_ENDPOINT: "https://portal.azure.us/",
        EndpointURLS.RESOURCE_MANAGER_ENDPOINT: "https://management.usgovcloudapi.net/",
        EndpointURLS.ACTIVE_DIRECTORY_ENDPOINT: "https://login.microsoftonline.us/",
        EndpointURLS.AML_RESOURCE_ID: "https://ml.azure.us/",
        EndpointURLS.STORAGE_ENDPOINT: "core.usgovcloudapi.net",
        EndpointURLS.REGISTRY_DISCOVERY_ENDPOINT: "https://usgovarizona.api.ml.azure.us/",
    },
}


def _get_cloud(cloud: str) -> Dict[str, str]:
    if cloud in _environments:
        return _environments[cloud]
    arm_url = os.environ.get(ArmConstants.METADATA_URL_ENV_NAME, ArmConstants.DEFAULT_URL)
    arm_clouds = _get_clouds_by_metadata_url(arm_url)
    try:
        new_cloud = arm_clouds[cloud]
        _environments.update(new_cloud)  # type: ignore[arg-type]
        return new_cloud
    except KeyError as e:
        msg = 'Unknown cloud environment "{0}".'.format(cloud)
        raise MlException(message=msg, no_personal_data_message=msg) from e


def _get_default_cloud_name() -> str:
    """
    :return: Configured cloud, defaults to 'AzureCloud' if
    AZUREML_CURRENT_CLOUD and ARM_CLOUD_METADATA_URL are not set to dynamically retrieve cloud info.
    AZUREML_CURRENT_CLOUD is also set by the SDK based on ARM_CLOUD_METADATA_URL.
    :rtype: str
    """
    current_cloud_env = os.getenv(AZUREML_CLOUD_ENV_NAME)
    if current_cloud_env is not None:
        return current_cloud_env
    arm_metadata_url = os.getenv(ArmConstants.METADATA_URL_ENV_NAME)
    if arm_metadata_url is not None:
        clouds = _get_clouds_by_metadata_url(arm_metadata_url)  # prefer ARM metadata url when set
        for cloud_name in clouds:  # pylint: disable=consider-using-dict-items
            if clouds[cloud_name][EndpointURLS.RESOURCE_MANAGER_ENDPOINT] in arm_metadata_url:
                _set_cloud(cloud_name)
                return cloud_name
    return AzureEnvironments.ENV_DEFAULT


def _get_cloud_details(cloud_name: Optional[str] = None) -> Dict[str, str]:
    """Returns a Cloud endpoints object for the specified Azure Cloud.

    :param cloud_name: cloud name
    :type cloud_name: str
    :return: azure environment endpoint.
    :rtype: Dict[str, str]
    """
    if cloud_name is None:
        cloud_name = _get_default_cloud_name()
        module_logger.debug(
            "Using the default cloud configuration: '%s'.",
            cloud_name,
        )
    return _get_cloud(cloud_name)


def _set_cloud(cloud_name: Optional[str] = None):
    """Sets the current cloud.

    :param cloud_name: cloud name
    :type cloud_name: str
    """
    if cloud_name is not None:
        try:
            _get_cloud(cloud_name)
        except Exception as e:
            msg = 'Unknown cloud environment supplied: "{0}".'.format(cloud_name)
            raise MlException(message=msg, no_personal_data_message=msg) from e
    else:
        cloud_name = _get_default_cloud_name()
    os.environ[AZUREML_CLOUD_ENV_NAME] = cloud_name


def _get_base_url_from_metadata(cloud_name: Optional[str] = None, is_local_mfe: bool = False) -> str:
    """Retrieve the base url for a cloud from the metadata in SDK.

    :param cloud_name: cloud name
    :type cloud_name: Optional[str]
    :param is_local_mfe: Whether is local Management Front End. Defaults to False.
    :type is_local_mfe: bool
    :return: base url for a cloud
    :rtype: str
    """
    base_url = None
    if is_local_mfe:
        base_url = _get_mfe_url_override()
    if base_url is None:
        cloud_details = _get_cloud_details(cloud_name)
        base_url = str(cloud_details.get(EndpointURLS.RESOURCE_MANAGER_ENDPOINT)).strip("/")
    return base_url


def _get_aml_resource_id_from_metadata(cloud_name: Optional[str] = None) -> str:
    """Retrieve the aml_resource_id for a cloud from the metadata in SDK.

    :param cloud_name: cloud name
    :type cloud_name: str
    :return: aml_resource_id for a cloud
    :rtype: str
    """
    cloud_details = _get_cloud_details(cloud_name)
    aml_resource_id = str(cloud_details.get(EndpointURLS.AML_RESOURCE_ID)).strip("/")
    return aml_resource_id


def _get_active_directory_url_from_metadata(cloud_name: Optional[str] = None) -> str:
    """Retrieve the active_directory_url for a cloud from the metadata in SDK.

    :param cloud_name: cloud name
    :type cloud_name: str
    :return: active_directory for a cloud
    :rtype: str
    """
    cloud_details = _get_cloud_details(cloud_name)
    active_directory_url = str(cloud_details.get(EndpointURLS.ACTIVE_DIRECTORY_ENDPOINT)).strip("/")
    return active_directory_url


def _get_storage_endpoint_from_metadata(cloud_name: Optional[str] = None) -> str:
    """Retrieve the storage_endpoint for a cloud from the metadata in SDK.

    :param cloud_name: cloud name
    :type cloud_name: str
    :return: storage_endpoint for a cloud
    :rtype: str
    """
    cloud_details = _get_cloud_details(cloud_name)
    storage_endpoint = cloud_details.get(EndpointURLS.STORAGE_ENDPOINT)
    return str(storage_endpoint)


def _get_azure_portal_id_from_metadata(cloud_name: Optional[str] = None) -> str:
    """Retrieve the azure_portal_id for a cloud from the metadata in SDK.

    :param cloud_name: cloud name
    :type cloud_name: str
    :return: azure_portal_id for a cloud
    :rtype: str
    """
    cloud_details = _get_cloud_details(cloud_name)
    azure_portal_id = cloud_details.get(EndpointURLS.AZURE_PORTAL_ENDPOINT)
    return str(azure_portal_id)


def _get_cloud_information_from_metadata(cloud_name: Optional[str] = None, **kwargs) -> Dict:
    """Retrieve the cloud information from the metadata in SDK.

    :param cloud_name: cloud name
    :type cloud_name: str
    :return: A dictionary of additional configuration parameters required for passing in cloud information.
    :rtype: Dict
    """
    cloud_details = _get_cloud_details(cloud_name)
    credential_scopes = _resource_to_scopes(
        cloud_details.get(EndpointURLS.RESOURCE_MANAGER_ENDPOINT).strip("/")  # type: ignore[union-attr]
    )

    # Update the kwargs with the cloud information
    client_kwargs: Dict = {"cloud": cloud_name}
    if credential_scopes is not None:
        client_kwargs["credential_scopes"] = credential_scopes
    kwargs.update(client_kwargs)
    return kwargs


def _get_registry_discovery_endpoint_from_metadata(cloud_name: Optional[str] = None) -> str:
    """Retrieve the registry_discovery_endpoint for a cloud from the metadata in SDK.

    :param cloud_name: cloud name
    :type cloud_name: str
    :return: registry_discovery_endpoint for a cloud
    :rtype: str
    """
    cloud_details = _get_cloud_details(cloud_name)
    registry_discovery_endpoint = cloud_details.get(EndpointURLS.REGISTRY_DISCOVERY_ENDPOINT)
    return str(registry_discovery_endpoint)


def _resource_to_scopes(resource: str) -> List[str]:
    """Convert the resource ID to scopes by appending the /.default suffix and return a list. For example:
    'https://management.core.windows.net/' ->

    ['https://management.core.windows.net//.default']

    :param resource: The resource ID
    :type resource: str
    :return: A list of scopes
    :rtype: List[str]
    """
    scope = resource + "/.default"
    return [scope]


def _get_registry_discovery_url(cloud: dict, cloud_suffix: str = "") -> str:
    """Get or generate the registry discovery url.

    :param cloud: configuration of the cloud to get the registry_discovery_url from
    :type cloud: dict
    :param cloud_suffix: the suffix to use for the cloud, in the case that the registry_discovery_url
        must be generated
    :type cloud_suffix: str
    :return: string of discovery url
    :rtype: str
    """
    cloud_name = cloud["name"]
    if cloud_name in _environments:
        return _environments[cloud_name][EndpointURLS.REGISTRY_DISCOVERY_ENDPOINT]
    registry_discovery_from_env = os.getenv(ArmConstants.REGISTRY_ENV_URL)
    if registry_discovery_from_env is not None:
        return registry_discovery_from_env
    registry_discovery_region = os.environ.get(
        ArmConstants.REGISTRY_DISCOVERY_REGION_ENV_NAME,
        ArmConstants.REGISTRY_DISCOVERY_DEFAULT_REGION,
    )
    return f"https://{cloud_name.lower()}{registry_discovery_region}.api.ml.azure.{cloud_suffix}/"


def _get_clouds_by_metadata_url(metadata_url: str) -> Dict[str, Dict[str, str]]:
    """Get all the clouds by the specified metadata url.

    :param metadata_url: The metadata url
    :type metadata_url: str
    :return: A dictionary of cloud name to various relevant endpoints/uris
    :rtype: Dict[str, Dict[str, str]]
    """
    try:
        module_logger.debug("Start : Loading cloud metadata from the url specified by %s", metadata_url)
        client = ARMPipelineClient(base_url=metadata_url, policies=[])
        HttpRequest("GET", metadata_url)
        with client.send_request(HttpRequest("GET", metadata_url)) as meta_response:
            arm_cloud_dict = meta_response.json()
            cli_cloud_dict = _convert_arm_to_cli(arm_cloud_dict)
            module_logger.debug(
                "Finish : Loading cloud metadata from the url specified by %s",
                metadata_url,
            )
            return cli_cloud_dict
    except Exception as ex:  # pylint: disable=broad-except
        module_logger.warning(
            "Error: Azure ML was unable to load cloud metadata from the url specified by %s. %s. "
            "This may be due to a misconfiguration of networking controls. Azure Machine Learning Python "
            "SDK requires outbound access to Azure Resource Manager. Please contact your networking team "
            "to configure outbound access to Azure Resource Manager on both Network Security Group and "
            "Firewall. For more details on required configurations, see "
            "https://learn.microsoft.com/azure/machine-learning/how-to-access-azureml-behind-firewall.",
            metadata_url,
            ex,
        )
        return {}


def _convert_arm_to_cli(arm_cloud_metadata) -> Dict[str, Dict[str, str]]:
    cli_cloud_metadata_dict = {}
    if isinstance(arm_cloud_metadata, dict):
        arm_cloud_metadata = [arm_cloud_metadata]

    for cloud in arm_cloud_metadata:
        try:
            cloud_name = cloud["name"]
            portal_endpoint = cloud["portal"]
            cloud_suffix = ".".join(portal_endpoint.split(".")[2:]).replace("/", "")
            registry_discovery_url = _get_registry_discovery_url(cloud, cloud_suffix)
            cli_cloud_metadata_dict[cloud_name] = {
                EndpointURLS.AZURE_PORTAL_ENDPOINT: cloud["portal"],
                EndpointURLS.RESOURCE_MANAGER_ENDPOINT: cloud["resourceManager"],
                EndpointURLS.ACTIVE_DIRECTORY_ENDPOINT: cloud["authentication"]["loginEndpoint"],
                EndpointURLS.AML_RESOURCE_ID: "https://ml.azure.{}".format(cloud_suffix),
                EndpointURLS.STORAGE_ENDPOINT: cloud["suffixes"]["storage"],
                EndpointURLS.REGISTRY_DISCOVERY_ENDPOINT: registry_discovery_url,
            }
        except KeyError as ex:
            module_logger.warning("Property on cloud not found in arm cloud metadata: %s", ex)
            continue
    return cli_cloud_metadata_dict


def _add_cloud_to_environments(kwargs):
    cloud_name = kwargs["cloud"]
    if cloud_name in _environments:
        raise AttributeError(f"Cannot overwrite existing cloud: {cloud_name}")
    cloud_metadata = kwargs[CloudArgumentKeys.CLOUD_METADATA]
    if cloud_metadata is None:
        raise LookupError(f"{CloudArgumentKeys.CLOUD_METADATA} not present in kwargs, no environment to add!")
    _environments[kwargs["cloud"]] = {
        EndpointURLS.AZURE_PORTAL_ENDPOINT: cloud_metadata[EndpointURLS.AZURE_PORTAL_ENDPOINT],
        EndpointURLS.RESOURCE_MANAGER_ENDPOINT: cloud_metadata[EndpointURLS.RESOURCE_MANAGER_ENDPOINT],
        EndpointURLS.ACTIVE_DIRECTORY_ENDPOINT: cloud_metadata[EndpointURLS.ACTIVE_DIRECTORY_ENDPOINT],
        EndpointURLS.AML_RESOURCE_ID: cloud_metadata[EndpointURLS.AML_RESOURCE_ID],
        EndpointURLS.STORAGE_ENDPOINT: cloud_metadata[EndpointURLS.STORAGE_ENDPOINT],
        EndpointURLS.REGISTRY_DISCOVERY_ENDPOINT: cloud_metadata[EndpointURLS.REGISTRY_DISCOVERY_ENDPOINT],
    }