#-------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
#--------------------------------------------------------------------------
import io
import json
import os
import sys
try:
from inspect import getfullargspec as get_arg_spec
except ImportError:
from inspect import getargspec as get_arg_spec
import adal
from msrestazure.azure_active_directory import AdalAuthentication
from .credentials import get_azure_cli_credentials
from .cloud import get_cli_active_cloud
def _instantiate_client(client_class, **kwargs):
"""Instantiate a client from kwargs, removing the subscription_id/tenant_id argument if unsupported.
"""
args = get_arg_spec(client_class.__init__).args
for key in ['subscription_id', 'tenant_id']:
if key not in kwargs:
continue
if key not in args:
del kwargs[key]
elif sys.version_info < (3, 0) and isinstance(kwargs[key], unicode):
kwargs[key] = kwargs[key].encode('utf-8')
return client_class(**kwargs)
[docs]def get_client_from_cli_profile(client_class, **kwargs):
"""Return a SDK client initialized with current CLI credentials, CLI default subscription and CLI default cloud.
This method will fill automatically the following client parameters:
- credentials
- subscription_id
- base_url
Parameters provided in kwargs will override CLI parameters and be passed directly to the client.
:Example:
.. code:: python
from azure.common.client_factory import get_client_from_cli_profile
from azure.mgmt.compute import ComputeManagementClient
client = get_client_from_cli_profile(ComputeManagementClient)
.. versionadded:: 1.1.6
:param client_class: A SDK client class
:return: An instantiated client
:raises: ImportError if azure-cli-core package is not available
"""
is_graphrbac = client_class.__name__ == 'GraphRbacManagementClient'
cloud = get_cli_active_cloud()
parameters = {}
if 'credentials' not in kwargs or 'subscription_id' not in kwargs:
if is_graphrbac:
resource = cloud.endpoints.active_directory_graph_resource_id
else:
resource = None
credentials, subscription_id, tenant_id = get_azure_cli_credentials(resource=resource, with_tenant=True)
parameters.update({
'credentials': kwargs.get('credentials', credentials),
'subscription_id': kwargs.get('subscription_id', subscription_id)
})
args = get_arg_spec(client_class.__init__).args
if 'adla_job_dns_suffix' in args and 'adla_job_dns_suffix' not in kwargs: # Datalake
# Let it raise here with AttributeError at worst, this would mean this cloud does not define
# ADL endpoint and no manual suffix was given
parameters['adla_job_dns_suffix'] = cloud.suffixes.azure_datalake_analytics_catalog_and_job_endpoint
elif 'base_url' in args and 'base_url' not in kwargs:
if is_graphrbac:
parameters['base_url'] = cloud.endpoints.active_directory_graph_resource_id
else:
parameters['base_url'] = cloud.endpoints.resource_manager
if 'tenant_id' in args and 'tenant_id' not in kwargs:
parameters['tenant_id'] = tenant_id
parameters.update(kwargs)
return _instantiate_client(client_class, **parameters)
[docs]def get_client_from_json_dict(client_class, config_dict, **kwargs):
"""Return a SDK client initialized with a JSON auth dict.
The easiest way to obtain this content is to call the following CLI commands:
.. code:: bash
az ad sp create-for-rbac --sdk-auth
This method will fill automatically the following client parameters:
- credentials
- subscription_id
- base_url
- tenant_id
Parameters provided in kwargs will override parameters and be passed directly to the client.
:Example:
.. code:: python
from azure.common.client_factory import get_client_from_auth_file
from azure.mgmt.compute import ComputeManagementClient
config_dict = {
"clientId": "ad735158-65ca-11e7-ba4d-ecb1d756380e",
"clientSecret": "b70bb224-65ca-11e7-810c-ecb1d756380e",
"subscriptionId": "bfc42d3a-65ca-11e7-95cf-ecb1d756380e",
"tenantId": "c81da1d8-65ca-11e7-b1d1-ecb1d756380e",
"activeDirectoryEndpointUrl": "https://login.microsoftonline.com",
"resourceManagerEndpointUrl": "https://management.azure.com/",
"activeDirectoryGraphResourceId": "https://graph.windows.net/",
"sqlManagementEndpointUrl": "https://management.core.windows.net:8443/",
"galleryEndpointUrl": "https://gallery.azure.com/",
"managementEndpointUrl": "https://management.core.windows.net/"
}
client = get_client_from_json_dict(ComputeManagementClient, config_dict)
.. versionadded:: 1.1.7
:param client_class: A SDK client class
:param dict config_dict: A config dict.
:return: An instantiated client
"""
is_graphrbac = client_class.__name__ == 'GraphRbacManagementClient'
parameters = {
'subscription_id': config_dict.get('subscriptionId'),
'base_url': config_dict.get('resourceManagerEndpointUrl'),
'tenant_id': config_dict.get('tenantId') # GraphRbac
}
if is_graphrbac:
parameters['base_url'] = config_dict['activeDirectoryGraphResourceId']
if 'credentials' not in kwargs:
# Get the right resource for Credentials
if is_graphrbac:
resource = config_dict['activeDirectoryGraphResourceId']
else:
resource = config_dict['resourceManagerEndpointUrl']
authority_url = (config_dict['activeDirectoryEndpointUrl'] + '/' +
config_dict['tenantId'])
context = adal.AuthenticationContext(authority_url, api_version=None)
parameters['credentials'] = AdalAuthentication(
context.acquire_token_with_client_credentials,
resource,
config_dict['clientId'],
config_dict['clientSecret']
)
parameters.update(kwargs)
return _instantiate_client(client_class, **parameters)
[docs]def get_client_from_auth_file(client_class, auth_path=None, **kwargs):
"""Return a SDK client initialized with auth file.
The easiest way to obtain this file is to call the following CLI commands:
.. code:: bash
az ad sp create-for-rbac --sdk-auth
You can specific the file path directly, or fill the environment variable AZURE_AUTH_LOCATION.
File must be UTF-8.
This method will fill automatically the following client parameters:
- credentials
- subscription_id
- base_url
Parameters provided in kwargs will override parameters and be passed directly to the client.
:Example:
.. code:: python
from azure.common.client_factory import get_client_from_auth_file
from azure.mgmt.compute import ComputeManagementClient
client = get_client_from_auth_file(ComputeManagementClient)
Example of file:
.. code:: json
{
"clientId": "ad735158-65ca-11e7-ba4d-ecb1d756380e",
"clientSecret": "b70bb224-65ca-11e7-810c-ecb1d756380e",
"subscriptionId": "bfc42d3a-65ca-11e7-95cf-ecb1d756380e",
"tenantId": "c81da1d8-65ca-11e7-b1d1-ecb1d756380e",
"activeDirectoryEndpointUrl": "https://login.microsoftonline.com",
"resourceManagerEndpointUrl": "https://management.azure.com/",
"activeDirectoryGraphResourceId": "https://graph.windows.net/",
"sqlManagementEndpointUrl": "https://management.core.windows.net:8443/",
"galleryEndpointUrl": "https://gallery.azure.com/",
"managementEndpointUrl": "https://management.core.windows.net/"
}
.. versionadded:: 1.1.7
:param client_class: A SDK client class
:param str auth_path: Path to the file.
:return: An instantiated client
:raises: KeyError if AZURE_AUTH_LOCATION is not an environment variable and no path is provided
:raises: FileNotFoundError if provided file path does not exists
:raises: json.JSONDecodeError if provided file is not JSON valid
:raises: UnicodeDecodeError if file is not UTF8 compliant
"""
auth_path = auth_path or os.environ['AZURE_AUTH_LOCATION']
with io.open(auth_path, 'r', encoding='utf-8-sig') as auth_fd:
config_dict = json.load(auth_fd)
return get_client_from_json_dict(client_class, config_dict, **kwargs)