Skip to content
Merged
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
21 changes: 13 additions & 8 deletions src/azure-cli-core/azure/cli/core/_identity.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,8 @@ class Identity:
"""Class to interact with Azure Identity.
"""
def __init__(self, authority, tenant_id):
if not authority:
raise ValueError("Unexpected CLI error: authority cannot be none.")
self.authority = authority
self.tenant_id = tenant_id

Expand Down Expand Up @@ -140,17 +142,20 @@ def get_user_credential(self, home_account_id, username):
auth_profile = AuthProfile(self.authority, home_account_id, self.tenant_id, username)
return InteractiveBrowserCredential(profile=auth_profile, silent_auth_only=True)

def get_service_principal_credential(self, client_id):
credCache = ServicePrincipalCredentialCache()
credCache.retrieve_secret_of_service_principal(client_id)
# TODO
def get_service_principal_credential(self, client_id, use_cert_sn_issuer):
cred_cache = ServicePrincipalCredentialCache()
client_secret, certificate_path = cred_cache.retrieve_secret_of_service_principal(client_id, self.tenant_id)
# TODO: support use_cert_sn_issuer in CertificateCredential
if client_secret:
return ClientSecretCredential(self.tenant_id, client_id, client_secret, use_cert_sn_issuer=use_cert_sn_issuer)
return ClientSecretCredential(self.tenant_id, client_id, client_secret)
if certificate_path:
return CertificateCredential(self.tenant_id, client_id, certificate_path)
raise CLIError("Secret of service principle {} not found. Please run 'az login'".format(client_id))

def get_msi_credential(self):
pass
@staticmethod
def get_msi_credential(client_id=None):
# TODO: support object_id and msi_res_id
return ManagedIdentityCredential(client_id=client_id)


class ServicePrincipalCredentialCache:
Expand Down Expand Up @@ -194,7 +199,7 @@ def retrieve_secret_of_service_principal(self, sp_id, tenant):
"Trying credential under tenant %s, assuming that is an app credential.",
sp_id, tenant, matched[0][_SERVICE_PRINCIPAL_TENANT])
cred = matched[0]
return cred.get(_ACCESS_TOKEN, None)
return cred.get(_ACCESS_TOKEN, None), cred.get(_SERVICE_PRINCIPAL_CERT_FILE, None)

def save_service_principal_cred(self, sp_entry):
self.load_service_principal_creds()
Expand Down
14 changes: 7 additions & 7 deletions src/azure-cli-core/azure/cli/core/_profile.py
Original file line number Diff line number Diff line change
Expand Up @@ -511,8 +511,8 @@ def _try_parse_msi_account_name(account):
def _create_identity_credential(self, account, aux_tenant_id=None):
user_type = account[_USER_ENTITY][_USER_TYPE]
username_or_sp_id = account[_USER_ENTITY][_USER_NAME]
home_account_id = account[_USER_ENTITY][_USER_HOME_ACCOUNT_ID]
identity_type, _ = Profile._try_parse_msi_account_name(account)
home_account_id = account[_USER_ENTITY].get(_USER_HOME_ACCOUNT_ID)
identity_type, identity_id = Profile._try_parse_msi_account_name(account)
tenant_id = aux_tenant_id if aux_tenant_id else account[_TENANT_ID]

authority = self.cli_ctx.cloud.endpoints.active_directory.replace('https://', '')
Expand All @@ -522,22 +522,22 @@ def _create_identity_credential(self, account, aux_tenant_id=None):
if in_cloud_console() and account[_USER_ENTITY].get(_CLOUD_SHELL_ID):
if aux_tenant_id:
raise CLIError("Tenant shouldn't be specified for Cloud Shell account")
return ManagedIdentity()
return Identity.get_msi_credential()

# User
if user_type == _USER:
if not home_account_id:
raise CLIError("CLI authentication is migrated to AADv2.0, please run 'az login' to re-login")
return identity.get_user_credential(home_account_id, username_or_sp_id)

# Service Principal
use_cert_sn_issuer = account[_USER_ENTITY].get(_SERVICE_PRINCIPAL_CERT_SN_ISSUER_AUTH)
# todo: get service principle key
return identity.get_service_principal_credential()
return identity.get_service_principal_credential(username_or_sp_id, use_cert_sn_issuer)

# MSI
# todo: MSI identity_id
if aux_tenant_id:
raise CLIError("Tenant shouldn't be specified for MSI account")
return get_msi_credential()
return Identity.get_msi_credential(identity_id)

def get_login_credentials(self, resource=None, subscription_id=None, aux_subscriptions=None, aux_tenants=None):
if aux_tenants and aux_subscriptions:
Expand Down