Fix azurerm MSI authentication with other Azure Cloud

[tu-doan]

SUMMARY

Fix azurerm MSI authentication with other Azure cloud environment.
Currently, this function only support Azure Public Cloud because of missing attribute when calling msrestazure.azure_active_directory.MSIAuthentication.

ISSUE TYPE

• Bugfix Pull Request

COMPONENT NAME

_get_msi_credentials() in plugins/module_utils/azure_rm_common.py.

Tested with AzureUSGovernment and Azure Public Cloud

ADDITIONAL INFORMATION

Issue was posted here : #893

[locmai]

Suggested change

	def _get_msi_credentials(self, subscription_id=None, client_id=None, _cloud_environment=None, **kwargs):
	if not _cloud_environment:
	cloud_environment = azure_cloud.AZURE_PUBLIC_CLOUD
	else:
	# try to look up "well-known" values via the name attribute on azure_cloud members
	all_clouds = [x[1] for x in inspect.getmembers(azure_cloud) if isinstance(x[1], azure_cloud.Cloud)]
	matched_clouds = [x for x in all_clouds if x.name == _cloud_environment]
	if len(matched_clouds) == 1:
	cloud_environment = matched_clouds[0]
	elif len(matched_clouds) > 1:
	self.fail("Azure SDK failure: more than one cloud matched for cloud_environment name '{0}'".format(cloud_environment))
	else:
	if not urlparse.urlparse(cloud_environment).scheme:
	self.fail("cloud_environment must be an endpoint discovery URL or one of {0}".format([x.name for x in all_clouds]))
	try:
	self._cloud_environment = azure_cloud.get_cloud_from_metadata_endpoint(cloud_environment)
	except Exception as e:
	self.fail("cloud_environment {0} could not be resolved: {1}".format(cloud_environment, e.message), exception=traceback.format_exc())
	credentials = MSIAuthentication(client_id=client_id, cloud_environment=cloud_environment)
	credential = MSIAuthenticationWrapper(client_id=client_id, cloud_environment=cloud_environment)
	def _get_msi_credentials(self, subscription_id=None, client_id=None, _cloud_environment="AzureCloud", **kwargs):
	try:
	all_clouds = {x[1].name:x[1] for x in inspect.getmembers(azure_cloud) if isinstance(x[1], azure_cloud.Cloud)}
	if _cloud_environment not in all_clouds:
	self.fail(f"Azure SDK failure: no cloud_environment name matched {_cloud_environment}")
	cloud_environment = all_clouds[_cloud_environment]
	self._cloud_environment = azure_cloud.get_cloud_from_metadata_endpoint(cloud_environment.endpoints. resource_manager)
	except Exception as e:
	self.fail(f"cloud_environment {_cloud_environment} could not be resolved: {e.message}", exception=traceback.format_exc())
	credentials = MSIAuthentication(client_id=client_id, cloud_environment=cloud_environment)
	credential = MSIAuthenticationWrapper(client_id=client_id, cloud_environment=cloud_environment)

I believe this would be a bit cleaner.

[locmai]

overall I think this is the right fix we need for #893

[locmai]

one more note:
I think we could add a _get_azure_cloud function since the _get_azure_cli_credentials could benefit from that as well:

def _get_azure_cloud(self, cloud_environment: str) -> azure_cloud.Cloud:
    all_clouds = {x[1].name:x[1] for x in inspect.getmembers(azure_cloud) if isinstance(x[1], azure_cloud.Cloud)}
    if _cloud_environment not in all_clouds:
        self.fail(f"Azure SDK failure: no cloud_environment name matched {_cloud_environment}")
    return all_clouds[_cloud_environment]

[Fred-sun]

@doanminhtu “cloud_environment” Without definition before use, can you refine your changes?

Suggested change

	if not urlparse.urlparse(cloud_environment).scheme:
	if not urlparse.urlparse(cloud_environment).scheme:

[tu-doan]

hi @Fred-sun . Many thanks to your review and suggestion. My code was not correct. Then I have just fixed the issue and new code pushed.
Could you please take a look to review it ?
Best regards,

[Fred-sun]

ready_for_review

[xuzhang3]

@doanminhtu LGTM 🚢