Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/adal/authentication_context.py: 30%
89 statements
« prev ^ index » next coverage.py v7.2.7, created at 2023-06-07 06:23 +0000
1#------------------------------------------------------------------------------
2#
3# Copyright (c) Microsoft Corporation.
4# All rights reserved.
5#
6# This code is licensed under the MIT License.
7#
8# Permission is hereby granted, free of charge, to any person obtaining a copy
9# of this software and associated documentation files(the "Software"), to deal
10# in the Software without restriction, including without limitation the rights
11# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
12# copies of the Software, and to permit persons to whom the Software is
13# furnished to do so, subject to the following conditions :
14#
15# The above copyright notice and this permission notice shall be included in
16# all copies or substantial portions of the Software.
17#
18# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
19# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
20# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
21# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
22# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
23# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
24# THE SOFTWARE.
25#
26#------------------------------------------------------------------------------
27import os
28import threading
29import warnings
31from .authority import Authority
32from . import argument
33from .code_request import CodeRequest
34from .token_request import TokenRequest
35from .token_cache import TokenCache
36from . import log
37from .constants import OAuth2DeviceCodeResponseParameters
39GLOBAL_ADAL_OPTIONS = {}
41class AuthenticationContext(object):
42 '''Retrieves authentication tokens from Azure Active Directory.
44 For usages, check out the "sample" folder at:
45 https://github.com/AzureAD/azure-activedirectory-library-for-python
46 '''
48 def __init__(
49 self, authority, validate_authority=None, cache=None,
50 api_version=None, timeout=None, enable_pii=False, verify_ssl=None, proxies=None):
51 '''Creates a new AuthenticationContext object.
53 By default the authority will be checked against a list of known Azure
54 Active Directory authorities. If the authority is not recognized as
55 one of these well known authorities then token acquisition will fail.
56 This behavior can be turned off via the validate_authority parameter
57 below.
59 :param str authority: A URL that identifies a token authority. It should be of the
60 format https://login.microsoftonline.com/your_tenant
61 :param bool validate_authority: (optional) Turns authority validation
62 on or off. This parameter default to true.
63 :param TokenCache cache: (optional) Sets the token cache used by this
64 AuthenticationContext instance. If this parameter is not set, then
65 a default is used. Cache instances is only used by that instance of
66 the AuthenticationContext and are not shared unless it has been
67 manually passed during the construction of other
68 AuthenticationContexts.
69 :param api_version: (optional) Specifies API version using on the wire.
70 Historically it has a hardcoded default value as "1.0".
71 Developers have been encouraged to set it as None explicitly,
72 which means the underlying API version will be automatically chosen.
73 Starting from ADAL Python 1.0, this default value becomes None.
74 :param timeout: (optional) requests timeout. How long to wait for the server to send
75 data before giving up, as a float, or a `(connect timeout,
76 read timeout) <timeouts>` tuple.
77 :param enable_pii: (optional) Unless this is set to True,
78 there will be no Personally Identifiable Information (PII) written in log.
79 :param verify_ssl: (optional) requests verify. Either a boolean, in which case it
80 controls whether we verify the server's TLS certificate, or a string, in which
81 case it must be a path to a CA bundle to use. If this value is not provided, and
82 ADAL_PYTHON_SSL_NO_VERIFY env variable is set, behavior is equivalent to
83 verify_ssl=False.
84 :param proxies: (optional) requests proxies. Dictionary mapping protocol to the URL
85 of the proxy. See http://docs.python-requests.org/en/master/user/advanced/#proxies
86 for details.
87 '''
88 self.authority = Authority(authority, validate_authority is None or validate_authority)
89 self._oauth2client = None
90 self.correlation_id = None
91 env_verify = 'ADAL_PYTHON_SSL_NO_VERIFY' not in os.environ
92 verify = verify_ssl if verify_ssl is not None else env_verify
93 if api_version is not None:
94 warnings.warn(
95 """The default behavior of including api-version=1.0 on the wire
96 is now deprecated.
97 Future version of ADAL will change the default value to None.
99 To ensure a smooth transition, you are recommended to explicitly
100 set it to None in your code now, and test out the new behavior.
102 context = AuthenticationContext(..., api_version=None)
103 """, DeprecationWarning)
104 self._call_context = {
105 'options': GLOBAL_ADAL_OPTIONS,
106 'api_version': api_version,
107 'verify_ssl': verify,
108 'proxies':proxies,
109 'timeout':timeout,
110 "enable_pii": enable_pii,
111 }
112 self._token_requests_with_user_code = {}
113 self.cache = cache or TokenCache()
114 self._lock = threading.RLock()
116 @property
117 def options(self):
118 return self._call_context['options']
120 @options.setter
121 def options(self, val):
122 self._call_context['options'] = val
124 def _acquire_token(self, token_func, correlation_id=None):
125 self._call_context['log_context'] = log.create_log_context(
126 correlation_id or self.correlation_id, self._call_context.get('enable_pii', False))
127 self.authority.validate(self._call_context)
128 return token_func(self)
130 def acquire_token(self, resource, user_id, client_id):
131 '''Gets a token for a given resource via cached tokens.
133 :param str resource: A URI that identifies the resource for which the
134 token is valid.
135 :param str user_id: The username of the user on behalf this application
136 is authenticating.
137 :param str client_id: The OAuth client id of the calling application.
138 :returns: dic with several keys, include "accessToken" and
139 "refreshToken".
140 '''
141 def token_func(self):
142 token_request = TokenRequest(self._call_context, self, client_id, resource)
143 return token_request.get_token_from_cache_with_refresh(user_id)
145 return self._acquire_token(token_func)
147 def acquire_token_with_username_password(self, resource, username, password, client_id):
148 '''Gets a token for a given resource via user credentails.
150 :param str resource: A URI that identifies the resource for which the
151 token is valid.
152 :param str username: The username of the user on behalf this
153 application is authenticating.
154 :param str password: The password of the user named in the username
155 parameter.
156 :param str client_id: The OAuth client id of the calling application.
157 :returns: dict with several keys, include "accessToken" and
158 "refreshToken".
159 '''
160 def token_func(self):
161 token_request = TokenRequest(self._call_context, self, client_id, resource)
162 return token_request.get_token_with_username_password(username, password)
164 return self._acquire_token(token_func)
166 def acquire_token_with_client_credentials(self, resource, client_id, client_secret):
167 '''Gets a token for a given resource via client credentials.
169 :param str resource: A URI that identifies the resource for which the
170 token is valid.
171 :param str client_id: The OAuth client id of the calling application.
172 :param str client_secret: The OAuth client secret of the calling application.
173 :returns: dict with several keys, include "accessToken".
174 '''
175 def token_func(self):
176 token_request = TokenRequest(self._call_context, self, client_id, resource)
177 return token_request.get_token_with_client_credentials(client_secret)
179 return self._acquire_token(token_func)
181 def acquire_token_with_authorization_code(self, authorization_code,
182 redirect_uri, resource,
183 client_id, client_secret=None, code_verifier=None):
184 '''Gets a token for a given resource via authorization code for a
185 server app.
187 :param str authorization_code: An authorization code returned from a
188 client.
189 :param str redirect_uri: the redirect uri that was used in the
190 authorize call.
191 :param str resource: A URI that identifies the resource for which the
192 token is valid.
193 :param str client_id: The OAuth client id of the calling application.
194 :param str client_secret: (only for confidential clients)The OAuth
195 client secret of the calling application. This parameter if not set,
196 defaults to None
197 :param str code_verifier: (optional)The code verifier that was used to
198 obtain authorization code if PKCE was used in the authorization
199 code grant request.(usually used by public clients) This parameter if not set,
200 defaults to None
201 :returns: dict with several keys, include "accessToken" and
202 "refreshToken".
203 '''
204 def token_func(self):
205 token_request = TokenRequest(
206 self._call_context,
207 self,
208 client_id,
209 resource,
210 redirect_uri)
211 return token_request.get_token_with_authorization_code(
212 authorization_code,
213 client_secret, code_verifier)
215 return self._acquire_token(token_func)
217 def acquire_token_with_refresh_token(self, refresh_token, client_id,
218 resource, client_secret=None):
219 '''Gets a token for a given resource via refresh tokens
221 :param str refresh_token: A refresh token returned in a tokne response
222 from a previous invocation of acquireToken.
223 :param str client_id: The OAuth client id of the calling application.
224 :param str resource: A URI that identifies the resource for which the
225 token is valid.
226 :param str client_secret: (optional)The OAuth client secret of the
227 calling application.
228 :returns: dict with several keys, include "accessToken" and
229 "refreshToken".
230 '''
231 def token_func(self):
232 token_request = TokenRequest(self._call_context, self, client_id, resource)
233 return token_request.get_token_with_refresh_token(refresh_token, client_secret)
235 return self._acquire_token(token_func)
237 def acquire_token_with_client_certificate(self, resource, client_id,
238 certificate, thumbprint, public_certificate=None):
239 '''Gets a token for a given resource via certificate credentials
241 :param str resource: A URI that identifies the resource for which the
242 token is valid.
243 :param str client_id: The OAuth client id of the calling application.
244 :param str certificate: A PEM encoded certificate private key.
245 :param str thumbprint: hex encoded thumbprint of the certificate.
246 :param str public_certificate(optional): if not None, it will be sent to the service for subject name
247 and issuer based authentication, which is to support cert auto rolls. The value must match the
248 certificate private key parameter.
250 Per `specs <https://tools.ietf.org/html/rfc7515#section-4.1.6>`_,
251 "the certificate containing
252 the public key corresponding to the key used to digitally sign the
253 JWS MUST be the first certificate. This MAY be followed by
254 additional certificates, with each subsequent certificate being the
255 one used to certify the previous one."
256 However, your certificate's issuer may use a different order.
257 So, if your attempt ends up with an error AADSTS700027 -
258 "The provided signature value did not match the expected signature value",
259 you may try use only the leaf cert (in PEM/str format) instead.
261 :returns: dict with several keys, include "accessToken".
262 '''
263 def token_func(self):
264 token_request = TokenRequest(self._call_context, self, client_id, resource)
265 return token_request.get_token_with_certificate(certificate, thumbprint, public_certificate)
267 return self._acquire_token(token_func)
269 def acquire_user_code(self, resource, client_id, language=None):
270 '''Gets the user code info which contains user_code, device_code for
271 authenticating user on device.
273 :param str resource: A URI that identifies the resource for which the
274 device_code and user_code is valid for.
275 :param str client_id: The OAuth client id of the calling application.
276 :param str language: The language code specifying how the message
277 should be localized to.
278 :returns: dict contains code and uri for users to login through browser.
279 '''
280 self._call_context['log_context'] = log.create_log_context(
281 self.correlation_id, self._call_context.get('enable_pii', False))
282 self.authority.validate(self._call_context)
283 code_request = CodeRequest(self._call_context, self, client_id, resource)
284 return code_request.get_user_code_info(language)
286 def acquire_token_with_device_code(self, resource, user_code_info, client_id):
287 '''Gets a new access token using via a device code.
289 :param str resource: A URI that identifies the resource for which the
290 token is valid.
291 :param dict user_code_info: The code info from the invocation of
292 "acquire_user_code"
293 :param str client_id: The OAuth client id of the calling application.
294 :returns: dict with several keys, include "accessToken" and
295 "refreshToken".
296 '''
297 def token_func(self):
298 token_request = TokenRequest(self._call_context, self, client_id, resource)
300 key = user_code_info[OAuth2DeviceCodeResponseParameters.DEVICE_CODE]
301 with self._lock:
302 self._token_requests_with_user_code[key] = token_request
304 token = token_request.get_token_with_device_code(user_code_info)
306 with self._lock:
307 self._token_requests_with_user_code.pop(key, None)
309 return token
311 return self._acquire_token(token_func, user_code_info.get('correlation_id', None))
313 def cancel_request_to_get_token_with_device_code(self, user_code_info):
314 '''Cancels the polling request to get token with device code.
316 :param dict user_code_info: The code info from the invocation of
317 "acquire_user_code"
318 :returns: None
319 '''
320 argument.validate_user_code_info(user_code_info)
322 key = user_code_info[OAuth2DeviceCodeResponseParameters.DEVICE_CODE]
323 with self._lock:
324 request = self._token_requests_with_user_code.get(key)
326 if not request:
327 raise ValueError('No acquire_token_with_device_code existed to be cancelled')
329 request.cancel_token_request_with_device_code()
330 self._token_requests_with_user_code.pop(key, None)