Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/msal/authority.py: 61%

1import json 

2try: 

3 from urllib.parse import urlparse 

4except ImportError: # Fall back to Python 2 

5 from urlparse import urlparse 

6import logging 

7 

8 

9logger = logging.getLogger(__name__) 

10 

11# Endpoints were copied from here 

12# https://docs.microsoft.com/en-us/azure/active-directory/develop/authentication-national-cloud#azure-ad-authentication-endpoints 

13AZURE_US_GOVERNMENT = "login.microsoftonline.us" 

14AZURE_CHINA = "login.chinacloudapi.cn" 

15AZURE_PUBLIC = "login.microsoftonline.com" 

16 

17WORLD_WIDE = 'login.microsoftonline.com' # There was an alias login.windows.net 

18WELL_KNOWN_AUTHORITY_HOSTS = set([ 

19 WORLD_WIDE, 

20 AZURE_CHINA, 

21 'login-us.microsoftonline.com', 

22 AZURE_US_GOVERNMENT, 

23 ]) 

24WELL_KNOWN_B2C_HOSTS = [ 

25 "b2clogin.com", 

26 "b2clogin.cn", 

27 "b2clogin.us", 

28 "b2clogin.de", 

29 "ciamlogin.com", 

30 ] 

31_CIAM_DOMAIN_SUFFIX = ".ciamlogin.com" 

32 

33 

34class AuthorityBuilder(object): 

35 def __init__(self, instance, tenant): 

36 """A helper to save caller from doing string concatenation. 

37 

38 Usage is documented in :func:`application.ClientApplication.__init__`. 

39 """ 

40 self._instance = instance.rstrip("/") 

41 self._tenant = tenant.strip("/") 

42 

43 def __str__(self): 

44 return "https://{}/{}".format(self._instance, self._tenant) 

45 

46 

47class Authority(object): 

48 """This class represents an (already-validated) authority. 

49 

50 Once constructed, it contains members named "*_endpoint" for this instance. 

51 TODO: It will also cache the previously-validated authority instances. 

52 """ 

53 _domains_without_user_realm_discovery = set([]) 

54 

55 def __init__( 

56 self, authority_url, http_client, 

57 validate_authority=True, 

58 instance_discovery=None, 

59 ): 

60 """Creates an authority instance, and also validates it. 

61 

62 :param validate_authority: 

63 The Authority validation process actually checks two parts: 

64 instance (a.k.a. host) and tenant. We always do a tenant discovery. 

65 This parameter only controls whether an instance discovery will be 

66 performed. 

67 """ 

68 # :param instance_discovery: 

69 # By default, the known-to-Microsoft validation will use an 

70 # instance discovery endpoint located at ``login.microsoftonline.com``. 

71 # You can customize the endpoint by providing a url as a string. 

72 # Or you can turn this behavior off by passing in a False here. 

73 self._http_client = http_client 

74 if isinstance(authority_url, AuthorityBuilder): 

75 authority_url = str(authority_url) 

76 authority, self.instance, tenant = canonicalize(authority_url) 

77 is_ciam = self.instance.endswith(_CIAM_DOMAIN_SUFFIX) 

78 self.is_adfs = tenant.lower() == 'adfs' and not is_ciam 

79 parts = authority.path.split('/') 

80 self._is_b2c = any( 

81 self.instance.endswith("." + d) for d in WELL_KNOWN_B2C_HOSTS 

82 ) or (len(parts) == 3 and parts[2].lower().startswith("b2c_")) 

83 self._is_known_to_developer = self.is_adfs or self._is_b2c or not validate_authority 

84 is_known_to_microsoft = self.instance in WELL_KNOWN_AUTHORITY_HOSTS 

85 instance_discovery_endpoint = 'https://{}/common/discovery/instance'.format( # Note: This URL seemingly returns V1 endpoint only 

86 WORLD_WIDE # Historically using WORLD_WIDE. Could use self.instance too 

87 # See https://github.com/AzureAD/microsoft-authentication-library-for-dotnet/blob/4.0.0/src/Microsoft.Identity.Client/Instance/AadInstanceDiscovery.cs#L101-L103 

88 # and https://github.com/AzureAD/microsoft-authentication-library-for-dotnet/blob/4.0.0/src/Microsoft.Identity.Client/Instance/AadAuthority.cs#L19-L33 

89 ) if instance_discovery in (None, True) else instance_discovery 

90 if instance_discovery_endpoint and not ( 

91 is_known_to_microsoft or self._is_known_to_developer): 

92 payload = _instance_discovery( 

93 "https://{}{}/oauth2/v2.0/authorize".format( 

94 self.instance, authority.path), 

95 self._http_client, 

96 instance_discovery_endpoint) 

97 if payload.get("error") == "invalid_instance": 

98 raise ValueError( 

99 "invalid_instance: " 

100 "The authority you provided, %s, is not whitelisted. " 

101 "If it is indeed your legit customized domain name, " 

102 "you can turn off this check by passing in " 

103 "validate_authority=False" 

104 % authority_url) 

105 tenant_discovery_endpoint = payload['tenant_discovery_endpoint'] 

106 else: 

107 tenant_discovery_endpoint = authority._replace( 

108 path="{prefix}{version}/.well-known/openid-configuration".format( 

109 prefix=tenant if is_ciam and len(authority.path) <= 1 # Path-less CIAM 

110 else authority.path, # In B2C, it is "/tenant/policy" 

111 version="" if self.is_adfs else "/v2.0", 

112 ) 

113 ).geturl() # Keeping original port and query. Query is useful for test. 

114 try: 

115 openid_config = tenant_discovery( 

116 tenant_discovery_endpoint, 

117 self._http_client) 

118 except ValueError: 

119 raise ValueError( 

120 "Unable to get authority configuration for {}. " 

121 "Authority would typically be in a format of " 

122 "https://login.microsoftonline.com/your_tenant " 

123 "Also please double check your tenant name or GUID is correct.".format( 

124 authority_url)) 

125 logger.debug("openid_config = %s", openid_config) 

126 self.authorization_endpoint = openid_config['authorization_endpoint'] 

127 self.token_endpoint = openid_config['token_endpoint'] 

128 self.device_authorization_endpoint = openid_config.get('device_authorization_endpoint') 

129 _, _, self.tenant = canonicalize(self.token_endpoint) # Usually a GUID 

130 

131 def user_realm_discovery(self, username, correlation_id=None, response=None): 

132 # It will typically return a dict containing "ver", "account_type", 

133 # "federation_protocol", "cloud_audience_urn", 

134 # "federation_metadata_url", "federation_active_auth_url", etc. 

135 if self.instance not in self.__class__._domains_without_user_realm_discovery: 

136 resp = response or self._http_client.get( 

137 "https://{netloc}/common/userrealm/{username}?api-version=1.0".format( 

138 netloc=self.instance, username=username), 

139 headers={'Accept': 'application/json', 

140 'client-request-id': correlation_id},) 

141 if resp.status_code != 404: 

142 resp.raise_for_status() 

143 return json.loads(resp.text) 

144 self.__class__._domains_without_user_realm_discovery.add(self.instance) 

145 return {} # This can guide the caller to fall back normal ROPC flow 

146 

147 

148def canonicalize(authority_or_auth_endpoint): 

149 # Returns (url_parsed_result, hostname_in_lowercase, tenant) 

150 authority = urlparse(authority_or_auth_endpoint) 

151 if authority.scheme == "https": 

152 parts = authority.path.split("/") 

153 first_part = parts[1] if len(parts) >= 2 and parts[1] else None 

154 if authority.hostname.endswith(_CIAM_DOMAIN_SUFFIX): # CIAM 

155 # Use path in CIAM authority. It will be validated by OIDC Discovery soon 

156 tenant = first_part if first_part else "{}.onmicrosoft.com".format( 

157 # Fallback to sub domain name. This variation may not be advertised 

158 authority.hostname.rsplit(_CIAM_DOMAIN_SUFFIX, 1)[0]) 

159 return authority, authority.hostname, tenant 

160 # AAD 

161 if len(parts) >= 2 and parts[1]: 

162 return authority, authority.hostname, parts[1] 

163 raise ValueError( 

164 "Your given address (%s) should consist of " 

165 "an https url with a minimum of one segment in a path: e.g. " 

166 "https://login.microsoftonline.com/<tenant> " 

167 "or https://<tenant_name>.ciamlogin.com/<tenant> " 

168 "or https://<tenant_name>.b2clogin.com/<tenant_name>.onmicrosoft.com/policy" 

169 % authority_or_auth_endpoint) 

170 

171def _instance_discovery(url, http_client, instance_discovery_endpoint, **kwargs): 

172 resp = http_client.get( 

173 instance_discovery_endpoint, 

174 params={'authorization_endpoint': url, 'api-version': '1.0'}, 

175 **kwargs) 

176 return json.loads(resp.text) 

177 

178def tenant_discovery(tenant_discovery_endpoint, http_client, **kwargs): 

179 # Returns Openid Configuration 

180 resp = http_client.get(tenant_discovery_endpoint, **kwargs) 

181 if resp.status_code == 200: 

182 return json.loads(resp.text) # It could raise ValueError 

183 if 400 <= resp.status_code < 500: 

184 # Nonexist tenant would hit this path 

185 # e.g. https://login.microsoftonline.com/nonexist_tenant/v2.0/.well-known/openid-configuration 

186 raise ValueError("OIDC Discovery failed on {}. HTTP status: {}, Error: {}".format( 

187 tenant_discovery_endpoint, 

188 resp.status_code, 

189 resp.text, # Expose it as-is b/c OIDC defines no error response format 

190 )) 

191 # Transient network error would hit this path 

192 resp.raise_for_status() 

193 raise RuntimeError( # A fallback here, in case resp.raise_for_status() is no-op 

194 "Unable to complete OIDC Discovery: %d, %s" % (resp.status_code, resp.text)) 

195