Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/msal/application.py: 16%
531 statements
« prev ^ index » next coverage.py v7.2.2, created at 2023-03-26 07:13 +0000
1import functools
2import json
3import time
4try: # Python 2
5 from urlparse import urljoin
6except: # Python 3
7 from urllib.parse import urljoin
8import logging
9import sys
10import warnings
11from threading import Lock
12import os
14from .oauth2cli import Client, JwtAssertionCreator
15from .oauth2cli.oidc import decode_part
16from .authority import Authority, WORLD_WIDE
17from .mex import send_request as mex_send_request
18from .wstrust_request import send_request as wst_send_request
19from .wstrust_response import *
20from .token_cache import TokenCache, _get_username
21import msal.telemetry
22from .region import _detect_region
23from .throttled_http_client import ThrottledHttpClient
24from .cloudshell import _is_running_in_cloud_shell
27# The __init__.py will import this. Not the other way around.
28__version__ = "1.21.0" # When releasing, also check and bump our dependencies's versions if needed
30logger = logging.getLogger(__name__)
31_AUTHORITY_TYPE_CLOUDSHELL = "CLOUDSHELL"
33def extract_certs(public_cert_content):
34 # Parses raw public certificate file contents and returns a list of strings
35 # Usage: headers = {"x5c": extract_certs(open("my_cert.pem").read())}
36 public_certificates = re.findall(
37 r'-----BEGIN CERTIFICATE-----(?P<cert_value>[^-]+)-----END CERTIFICATE-----',
38 public_cert_content, re.I)
39 if public_certificates:
40 return [cert.strip() for cert in public_certificates]
41 # The public cert tags are not found in the input,
42 # let's make best effort to exclude a private key pem file.
43 if "PRIVATE KEY" in public_cert_content:
44 raise ValueError(
45 "We expect your public key but detect a private key instead")
46 return [public_cert_content.strip()]
49def _merge_claims_challenge_and_capabilities(capabilities, claims_challenge):
50 # Represent capabilities as {"access_token": {"xms_cc": {"values": capabilities}}}
51 # and then merge/add it into incoming claims
52 if not capabilities:
53 return claims_challenge
54 claims_dict = json.loads(claims_challenge) if claims_challenge else {}
55 for key in ["access_token"]: # We could add "id_token" if we'd decide to
56 claims_dict.setdefault(key, {}).update(xms_cc={"values": capabilities})
57 return json.dumps(claims_dict)
60def _str2bytes(raw):
61 # A conversion based on duck-typing rather than six.text_type
62 try:
63 return raw.encode(encoding="utf-8")
64 except:
65 return raw
68def _clean_up(result):
69 if isinstance(result, dict):
70 return {
71 k: result[k] for k in result
72 if k != "refresh_in" # MSAL handled refresh_in, customers need not
73 and not k.startswith('_') # Skim internal properties
74 }
75 return result # It could be None
78def _preferred_browser():
79 """Register Edge and return a name suitable for subsequent webbrowser.get(...)
80 when appropriate. Otherwise return None.
81 """
82 # On Linux, only Edge will provide device-based Conditional Access support
83 if sys.platform != "linux": # On other platforms, we have no browser preference
84 return None
85 browser_path = "/usr/bin/microsoft-edge" # Use a full path owned by sys admin
86 # Note: /usr/bin/microsoft-edge, /usr/bin/microsoft-edge-stable, etc.
87 # are symlinks that point to the actual binaries which are found under
88 # /opt/microsoft/msedge/msedge or /opt/microsoft/msedge-beta/msedge.
89 # Either method can be used to detect an Edge installation.
90 user_has_no_preference = "BROWSER" not in os.environ
91 user_wont_mind_edge = "microsoft-edge" in os.environ.get("BROWSER", "") # Note:
92 # BROWSER could contain "microsoft-edge" or "/path/to/microsoft-edge".
93 # Python documentation (https://docs.python.org/3/library/webbrowser.html)
94 # does not document the name being implicitly register,
95 # so there is no public API to know whether the ENV VAR browser would work.
96 # Therefore, we would not bother examine the env var browser's type.
97 # We would just register our own Edge instance.
98 if (user_has_no_preference or user_wont_mind_edge) and os.path.exists(browser_path):
99 try:
100 import webbrowser # Lazy import. Some distro may not have this.
101 browser_name = "msal-edge" # Avoid popular name "microsoft-edge"
102 # otherwise `BROWSER="microsoft-edge"; webbrowser.get("microsoft-edge")`
103 # would return a GenericBrowser instance which won't work.
104 try:
105 registration_available = isinstance(
106 webbrowser.get(browser_name), webbrowser.BackgroundBrowser)
107 except webbrowser.Error:
108 registration_available = False
109 if not registration_available:
110 logger.debug("Register %s with %s", browser_name, browser_path)
111 # By registering our own browser instance with our own name,
112 # rather than populating a process-wide BROWSER enn var,
113 # this approach does not have side effect on non-MSAL code path.
114 webbrowser.register( # Even double-register happens to work fine
115 browser_name, None, webbrowser.BackgroundBrowser(browser_path))
116 return browser_name
117 except ImportError:
118 pass # We may still proceed
119 return None
122class _ClientWithCcsRoutingInfo(Client):
124 def initiate_auth_code_flow(self, **kwargs):
125 if kwargs.get("login_hint"): # eSTS could have utilized this as-is, but nope
126 kwargs["X-AnchorMailbox"] = "UPN:%s" % kwargs["login_hint"]
127 return super(_ClientWithCcsRoutingInfo, self).initiate_auth_code_flow(
128 client_info=1, # To be used as CSS Routing info
129 **kwargs)
131 def obtain_token_by_auth_code_flow(
132 self, auth_code_flow, auth_response, **kwargs):
133 # Note: the obtain_token_by_browser() is also covered by this
134 assert isinstance(auth_code_flow, dict) and isinstance(auth_response, dict)
135 headers = kwargs.pop("headers", {})
136 client_info = json.loads(
137 decode_part(auth_response["client_info"])
138 ) if auth_response.get("client_info") else {}
139 if "uid" in client_info and "utid" in client_info:
140 # Note: The value of X-AnchorMailbox is also case-insensitive
141 headers["X-AnchorMailbox"] = "Oid:{uid}@{utid}".format(**client_info)
142 return super(_ClientWithCcsRoutingInfo, self).obtain_token_by_auth_code_flow(
143 auth_code_flow, auth_response, headers=headers, **kwargs)
145 def obtain_token_by_username_password(self, username, password, **kwargs):
146 headers = kwargs.pop("headers", {})
147 headers["X-AnchorMailbox"] = "upn:{}".format(username)
148 return super(_ClientWithCcsRoutingInfo, self).obtain_token_by_username_password(
149 username, password, headers=headers, **kwargs)
152class ClientApplication(object):
153 ACQUIRE_TOKEN_SILENT_ID = "84"
154 ACQUIRE_TOKEN_BY_REFRESH_TOKEN = "85"
155 ACQUIRE_TOKEN_BY_USERNAME_PASSWORD_ID = "301"
156 ACQUIRE_TOKEN_ON_BEHALF_OF_ID = "523"
157 ACQUIRE_TOKEN_BY_DEVICE_FLOW_ID = "622"
158 ACQUIRE_TOKEN_FOR_CLIENT_ID = "730"
159 ACQUIRE_TOKEN_BY_AUTHORIZATION_CODE_ID = "832"
160 ACQUIRE_TOKEN_INTERACTIVE = "169"
161 GET_ACCOUNTS_ID = "902"
162 REMOVE_ACCOUNT_ID = "903"
164 ATTEMPT_REGION_DISCOVERY = True # "TryAutoDetect"
166 def __init__(
167 self, client_id,
168 client_credential=None, authority=None, validate_authority=True,
169 token_cache=None,
170 http_client=None,
171 verify=True, proxies=None, timeout=None,
172 client_claims=None, app_name=None, app_version=None,
173 client_capabilities=None,
174 azure_region=None, # Note: We choose to add this param in this base class,
175 # despite it is currently only needed by ConfidentialClientApplication.
176 # This way, it holds the same positional param place for PCA,
177 # when we would eventually want to add this feature to PCA in future.
178 exclude_scopes=None,
179 http_cache=None,
180 instance_discovery=None,
181 allow_broker=None,
182 ):
183 """Create an instance of application.
185 :param str client_id: Your app has a client_id after you register it on AAD.
187 :param Union[str, dict] client_credential:
188 For :class:`PublicClientApplication`, you simply use `None` here.
189 For :class:`ConfidentialClientApplication`,
190 it can be a string containing client secret,
191 or an X509 certificate container in this form::
193 {
194 "private_key": "...-----BEGIN PRIVATE KEY-----...",
195 "thumbprint": "A1B2C3D4E5F6...",
196 "public_certificate": "...-----BEGIN CERTIFICATE-----... (Optional. See below.)",
197 "passphrase": "Passphrase if the private_key is encrypted (Optional. Added in version 1.6.0)",
198 }
200 *Added in version 0.5.0*:
201 public_certificate (optional) is public key certificate
202 which will be sent through 'x5c' JWT header only for
203 subject name and issuer authentication to support cert auto rolls.
205 Per `specs <https://tools.ietf.org/html/rfc7515#section-4.1.6>`_,
206 "the certificate containing
207 the public key corresponding to the key used to digitally sign the
208 JWS MUST be the first certificate. This MAY be followed by
209 additional certificates, with each subsequent certificate being the
210 one used to certify the previous one."
211 However, your certificate's issuer may use a different order.
212 So, if your attempt ends up with an error AADSTS700027 -
213 "The provided signature value did not match the expected signature value",
214 you may try use only the leaf cert (in PEM/str format) instead.
216 *Added in version 1.13.0*:
217 It can also be a completely pre-signed assertion that you've assembled yourself.
218 Simply pass a container containing only the key "client_assertion", like this::
220 {
221 "client_assertion": "...a JWT with claims aud, exp, iss, jti, nbf, and sub..."
222 }
224 :param dict client_claims:
225 *Added in version 0.5.0*:
226 It is a dictionary of extra claims that would be signed by
227 by this :class:`ConfidentialClientApplication` 's private key.
228 For example, you can use {"client_ip": "x.x.x.x"}.
229 You may also override any of the following default claims::
231 {
232 "aud": the_token_endpoint,
233 "iss": self.client_id,
234 "sub": same_as_issuer,
235 "exp": now + 10_min,
236 "iat": now,
237 "jti": a_random_uuid
238 }
240 :param str authority:
241 A URL that identifies a token authority. It should be of the format
242 ``https://login.microsoftonline.com/your_tenant``
243 By default, we will use ``https://login.microsoftonline.com/common``
245 *Changed in version 1.17*: you can also use predefined constant
246 and a builder like this::
248 from msal.authority import (
249 AuthorityBuilder,
250 AZURE_US_GOVERNMENT, AZURE_CHINA, AZURE_PUBLIC)
251 my_authority = AuthorityBuilder(AZURE_PUBLIC, "contoso.onmicrosoft.com")
252 # Now you get an equivalent of
253 # "https://login.microsoftonline.com/contoso.onmicrosoft.com"
255 # You can feed such an authority to msal's ClientApplication
256 from msal import PublicClientApplication
257 app = PublicClientApplication("my_client_id", authority=my_authority, ...)
259 :param bool validate_authority: (optional) Turns authority validation
260 on or off. This parameter default to true.
261 :param TokenCache cache:
262 Sets the token cache used by this ClientApplication instance.
263 By default, an in-memory cache will be created and used.
264 :param http_client: (optional)
265 Your implementation of abstract class HttpClient <msal.oauth2cli.http.http_client>
266 Defaults to a requests session instance.
267 Since MSAL 1.11.0, the default session would be configured
268 to attempt one retry on connection error.
269 If you are providing your own http_client,
270 it will be your http_client's duty to decide whether to perform retry.
272 :param verify: (optional)
273 It will be passed to the
274 `verify parameter in the underlying requests library
275 <http://docs.python-requests.org/en/v2.9.1/user/advanced/#ssl-cert-verification>`_
276 This does not apply if you have chosen to pass your own Http client
277 :param proxies: (optional)
278 It will be passed to the
279 `proxies parameter in the underlying requests library
280 <http://docs.python-requests.org/en/v2.9.1/user/advanced/#proxies>`_
281 This does not apply if you have chosen to pass your own Http client
282 :param timeout: (optional)
283 It will be passed to the
284 `timeout parameter in the underlying requests library
285 <http://docs.python-requests.org/en/v2.9.1/user/advanced/#timeouts>`_
286 This does not apply if you have chosen to pass your own Http client
287 :param app_name: (optional)
288 You can provide your application name for Microsoft telemetry purposes.
289 Default value is None, means it will not be passed to Microsoft.
290 :param app_version: (optional)
291 You can provide your application version for Microsoft telemetry purposes.
292 Default value is None, means it will not be passed to Microsoft.
293 :param list[str] client_capabilities: (optional)
294 Allows configuration of one or more client capabilities, e.g. ["CP1"].
296 Client capability is meant to inform the Microsoft identity platform
297 (STS) what this client is capable for,
298 so STS can decide to turn on certain features.
299 For example, if client is capable to handle *claims challenge*,
300 STS can then issue CAE access tokens to resources
301 knowing when the resource emits *claims challenge*
302 the client will be capable to handle.
304 Implementation details:
305 Client capability is implemented using "claims" parameter on the wire,
306 for now.
307 MSAL will combine them into
308 `claims parameter <https://openid.net/specs/openid-connect-core-1_0-final.html#ClaimsParameter>`_
309 which you will later provide via one of the acquire-token request.
311 :param str azure_region:
312 AAD provides regional endpoints for apps to opt in
313 to keep their traffic remain inside that region.
315 As of 2021 May, regional service is only available for
316 ``acquire_token_for_client()`` sent by any of the following scenarios::
318 1. An app powered by a capable MSAL
319 (MSAL Python 1.12+ will be provisioned)
321 2. An app with managed identity, which is formerly known as MSI.
322 (However MSAL Python does not support managed identity,
323 so this one does not apply.)
325 3. An app authenticated by
326 `Subject Name/Issuer (SNI) <https://github.com/AzureAD/microsoft-authentication-library-for-python/issues/60>`_.
328 4. An app which already onboard to the region's allow-list.
330 This parameter defaults to None, which means region behavior remains off.
332 App developer can opt in to a regional endpoint,
333 by provide its region name, such as "westus", "eastus2".
334 You can find a full list of regions by running
335 ``az account list-locations -o table``, or referencing to
336 `this doc <https://docs.microsoft.com/en-us/dotnet/api/microsoft.azure.management.resourcemanager.fluent.core.region?view=azure-dotnet>`_.
338 An app running inside Azure Functions and Azure VM can use a special keyword
339 ``ClientApplication.ATTEMPT_REGION_DISCOVERY`` to auto-detect region.
341 .. note::
343 Setting ``azure_region`` to non-``None`` for an app running
344 outside of Azure Function/VM could hang indefinitely.
346 You should consider opting in/out region behavior on-demand,
347 by loading ``azure_region=None`` or ``azure_region="westus"``
348 or ``azure_region=True`` (which means opt-in and auto-detect)
349 from your per-deployment configuration, and then do
350 ``app = ConfidentialClientApplication(..., azure_region=azure_region)``.
352 Alternatively, you can configure a short timeout,
353 or provide a custom http_client which has a short timeout.
354 That way, the latency would be under your control,
355 but still less performant than opting out of region feature.
357 New in version 1.12.0.
359 :param list[str] exclude_scopes: (optional)
360 Historically MSAL hardcodes `offline_access` scope,
361 which would allow your app to have prolonged access to user's data.
362 If that is unnecessary or undesirable for your app,
363 now you can use this parameter to supply an exclusion list of scopes,
364 such as ``exclude_scopes = ["offline_access"]``.
366 :param dict http_cache:
367 MSAL has long been caching tokens in the ``token_cache``.
368 Recently, MSAL also introduced a concept of ``http_cache``,
369 by automatically caching some finite amount of non-token http responses,
370 so that *long-lived*
371 ``PublicClientApplication`` and ``ConfidentialClientApplication``
372 would be more performant and responsive in some situations.
374 This ``http_cache`` parameter accepts any dict-like object.
375 If not provided, MSAL will use an in-memory dict.
377 If your app is a command-line app (CLI),
378 you would want to persist your http_cache across different CLI runs.
379 The following recipe shows a way to do so::
381 # Just add the following lines at the beginning of your CLI script
382 import sys, atexit, pickle
383 http_cache_filename = sys.argv[0] + ".http_cache"
384 try:
385 with open(http_cache_filename, "rb") as f:
386 persisted_http_cache = pickle.load(f) # Take a snapshot
387 except (
388 FileNotFoundError, # Or IOError in Python 2
389 pickle.UnpicklingError, # A corrupted http cache file
390 ):
391 persisted_http_cache = {} # Recover by starting afresh
392 atexit.register(lambda: pickle.dump(
393 # When exit, flush it back to the file.
394 # It may occasionally overwrite another process's concurrent write,
395 # but that is fine. Subsequent runs will reach eventual consistency.
396 persisted_http_cache, open(http_cache_file, "wb")))
398 # And then you can implement your app as you normally would
399 app = msal.PublicClientApplication(
400 "your_client_id",
401 ...,
402 http_cache=persisted_http_cache, # Utilize persisted_http_cache
403 ...,
404 #token_cache=..., # You may combine the old token_cache trick
405 # Please refer to token_cache recipe at
406 # https://msal-python.readthedocs.io/en/latest/#msal.SerializableTokenCache
407 )
408 app.acquire_token_interactive(["your", "scope"], ...)
410 Content inside ``http_cache`` are cheap to obtain.
411 There is no need to share them among different apps.
413 Content inside ``http_cache`` will contain no tokens nor
414 Personally Identifiable Information (PII). Encryption is unnecessary.
416 New in version 1.16.0.
418 :param boolean instance_discovery:
419 Historically, MSAL would connect to a central endpoint located at
420 ``https://login.microsoftonline.com`` to acquire some metadata,
421 especially when using an unfamiliar authority.
422 This behavior is known as Instance Discovery.
424 This parameter defaults to None, which enables the Instance Discovery.
426 If you know some authorities which you allow MSAL to operate with as-is,
427 without involving any Instance Discovery, the recommended pattern is::
429 known_authorities = frozenset([ # Treat your known authorities as const
430 "https://contoso.com/adfs", "https://login.azs/foo"])
431 ...
432 authority = "https://contoso.com/adfs" # Assuming your app will use this
433 app1 = PublicClientApplication(
434 "client_id",
435 authority=authority,
436 # Conditionally disable Instance Discovery for known authorities
437 instance_discovery=authority not in known_authorities,
438 )
440 If you do not know some authorities beforehand,
441 yet still want MSAL to accept any authority that you will provide,
442 you can use a ``False`` to unconditionally disable Instance Discovery.
444 New in version 1.19.0.
446 :param boolean allow_broker:
447 A broker is a component installed on your device.
448 Broker implicitly gives your device an identity. By using a broker,
449 your device becomes a factor that can satisfy MFA (Multi-factor authentication).
450 This factor would become mandatory
451 if a tenant's admin enables a corresponding Conditional Access (CA) policy.
452 The broker's presence allows Microsoft identity platform
453 to have higher confidence that the tokens are being issued to your device,
454 and that is more secure.
456 An additional benefit of broker is,
457 it runs as a long-lived process with your device's OS,
458 and maintains its own cache,
459 so that your broker-enabled apps (even a CLI)
460 could automatically SSO from a previously established signed-in session.
462 This parameter defaults to None, which means MSAL will not utilize a broker.
463 If this parameter is set to True,
464 MSAL will use the broker whenever possible,
465 and automatically fall back to non-broker behavior.
466 That also means your app does not need to enable broker conditionally,
467 you can always set allow_broker to True,
468 as long as your app meets the following prerequisite:
470 * Installed optional dependency, e.g. ``pip install msal[broker]>=1.20,<2``.
471 (Note that broker is currently only available on Windows 10+)
473 * Register a new redirect_uri for your desktop app as:
474 ``ms-appx-web://Microsoft.AAD.BrokerPlugin/your_client_id``
476 * Tested your app in following scenarios:
478 * Windows 10+
480 * PublicClientApplication's following methods::
481 acquire_token_interactive(), acquire_token_by_username_password(),
482 acquire_token_silent() (or acquire_token_silent_with_error()).
484 * AAD and MSA accounts (i.e. Non-ADFS, non-B2C)
486 New in version 1.20.0.
487 """
488 self.client_id = client_id
489 self.client_credential = client_credential
490 self.client_claims = client_claims
491 self._client_capabilities = client_capabilities
492 self._instance_discovery = instance_discovery
494 if exclude_scopes and not isinstance(exclude_scopes, list):
495 raise ValueError(
496 "Invalid exclude_scopes={}. It need to be a list of strings.".format(
497 repr(exclude_scopes)))
498 self._exclude_scopes = frozenset(exclude_scopes or [])
499 if "openid" in self._exclude_scopes:
500 raise ValueError(
501 'Invalid exclude_scopes={}. You can not opt out "openid" scope'.format(
502 repr(exclude_scopes)))
504 if http_client:
505 self.http_client = http_client
506 else:
507 import requests # Lazy load
509 self.http_client = requests.Session()
510 self.http_client.verify = verify
511 self.http_client.proxies = proxies
512 # Requests, does not support session - wide timeout
513 # But you can patch that (https://github.com/psf/requests/issues/3341):
514 self.http_client.request = functools.partial(
515 self.http_client.request, timeout=timeout)
517 # Enable a minimal retry. Better than nothing.
518 # https://github.com/psf/requests/blob/v2.25.1/requests/adapters.py#L94-L108
519 a = requests.adapters.HTTPAdapter(max_retries=1)
520 self.http_client.mount("http://", a)
521 self.http_client.mount("https://", a)
522 self.http_client = ThrottledHttpClient(
523 self.http_client,
524 {} if http_cache is None else http_cache, # Default to an in-memory dict
525 )
527 self.app_name = app_name
528 self.app_version = app_version
530 # Here the self.authority will not be the same type as authority in input
531 try:
532 authority_to_use = authority or "https://{}/common/".format(WORLD_WIDE)
533 self.authority = Authority(
534 authority_to_use,
535 self.http_client,
536 validate_authority=validate_authority,
537 instance_discovery=self._instance_discovery,
538 )
539 except ValueError: # Those are explicit authority validation errors
540 raise
541 except Exception: # The rest are typically connection errors
542 if validate_authority and azure_region:
543 # Since caller opts in to use region, here we tolerate connection
544 # errors happened during authority validation at non-region endpoint
545 self.authority = Authority(
546 authority_to_use,
547 self.http_client,
548 instance_discovery=False,
549 )
550 else:
551 raise
552 is_confidential_app = bool(
553 isinstance(self, ConfidentialClientApplication) or self.client_credential)
554 if is_confidential_app and allow_broker:
555 raise ValueError("allow_broker=True is only supported in PublicClientApplication")
556 self._enable_broker = False
557 if (allow_broker and not is_confidential_app
558 and sys.platform == "win32"
559 and not self.authority.is_adfs and not self.authority._is_b2c):
560 try:
561 from . import broker # Trigger Broker's initialization
562 self._enable_broker = True
563 except RuntimeError:
564 logger.exception(
565 "Broker is unavailable on this platform. "
566 "We will fallback to non-broker.")
567 logger.debug("Broker enabled? %s", self._enable_broker)
569 self.token_cache = token_cache or TokenCache()
570 self._region_configured = azure_region
571 self._region_detected = None
572 self.client, self._regional_client = self._build_client(
573 client_credential, self.authority)
574 self.authority_groups = None
575 self._telemetry_buffer = {}
576 self._telemetry_lock = Lock()
578 def _decorate_scope(
579 self, scopes,
580 reserved_scope=frozenset(['openid', 'profile', 'offline_access'])):
581 if not isinstance(scopes, (list, set, tuple)):
582 raise ValueError("The input scopes should be a list, tuple, or set")
583 scope_set = set(scopes) # Input scopes is typically a list. Copy it to a set.
584 if scope_set & reserved_scope:
585 # These scopes are reserved for the API to provide good experience.
586 # We could make the developer pass these and then if they do they will
587 # come back asking why they don't see refresh token or user information.
588 raise ValueError(
589 "API does not accept {} value as user-provided scopes".format(
590 reserved_scope))
592 # client_id can also be used as a scope in B2C
593 decorated = scope_set | reserved_scope
594 decorated -= self._exclude_scopes
595 return list(decorated)
597 def _build_telemetry_context(
598 self, api_id, correlation_id=None, refresh_reason=None):
599 return msal.telemetry._TelemetryContext(
600 self._telemetry_buffer, self._telemetry_lock, api_id,
601 correlation_id=correlation_id, refresh_reason=refresh_reason)
603 def _get_regional_authority(self, central_authority):
604 self._region_detected = self._region_detected or _detect_region(
605 self.http_client if self._region_configured is not None else None)
606 if (self._region_configured != self.ATTEMPT_REGION_DISCOVERY
607 and self._region_configured != self._region_detected):
608 logger.warning('Region configured ({}) != region detected ({})'.format(
609 repr(self._region_configured), repr(self._region_detected)))
610 region_to_use = (
611 self._region_detected
612 if self._region_configured == self.ATTEMPT_REGION_DISCOVERY
613 else self._region_configured) # It will retain the None i.e. opted out
614 logger.debug('Region to be used: {}'.format(repr(region_to_use)))
615 if region_to_use:
616 regional_host = ("{}.login.microsoft.com".format(region_to_use)
617 if central_authority.instance in (
618 # The list came from point 3 of the algorithm section in this internal doc
619 # https://identitydivision.visualstudio.com/DevEx/_git/AuthLibrariesApiReview?path=/PinAuthToRegion/AAD%20SDK%20Proposal%20to%20Pin%20Auth%20to%20region.md&anchor=algorithm&_a=preview
620 "login.microsoftonline.com",
621 "login.microsoft.com",
622 "login.windows.net",
623 "sts.windows.net",
624 )
625 else "{}.{}".format(region_to_use, central_authority.instance))
626 return Authority( # The central_authority has already been validated
627 "https://{}/{}".format(regional_host, central_authority.tenant),
628 self.http_client,
629 instance_discovery=False,
630 )
631 return None
633 def _build_client(self, client_credential, authority, skip_regional_client=False):
634 client_assertion = None
635 client_assertion_type = None
636 default_headers = {
637 "x-client-sku": "MSAL.Python", "x-client-ver": __version__,
638 "x-client-os": sys.platform,
639 "x-client-cpu": "x64" if sys.maxsize > 2 ** 32 else "x86",
640 "x-ms-lib-capability": "retry-after, h429",
641 }
642 if self.app_name:
643 default_headers['x-app-name'] = self.app_name
644 if self.app_version:
645 default_headers['x-app-ver'] = self.app_version
646 default_body = {"client_info": 1}
647 if isinstance(client_credential, dict):
648 assert (("private_key" in client_credential
649 and "thumbprint" in client_credential) or
650 "client_assertion" in client_credential)
651 client_assertion_type = Client.CLIENT_ASSERTION_TYPE_JWT
652 if 'client_assertion' in client_credential:
653 client_assertion = client_credential['client_assertion']
654 else:
655 headers = {}
656 if 'public_certificate' in client_credential:
657 headers["x5c"] = extract_certs(client_credential['public_certificate'])
658 if not client_credential.get("passphrase"):
659 unencrypted_private_key = client_credential['private_key']
660 else:
661 from cryptography.hazmat.primitives import serialization
662 from cryptography.hazmat.backends import default_backend
663 unencrypted_private_key = serialization.load_pem_private_key(
664 _str2bytes(client_credential["private_key"]),
665 _str2bytes(client_credential["passphrase"]),
666 backend=default_backend(), # It was a required param until 2020
667 )
668 assertion = JwtAssertionCreator(
669 unencrypted_private_key, algorithm="RS256",
670 sha1_thumbprint=client_credential.get("thumbprint"), headers=headers)
671 client_assertion = assertion.create_regenerative_assertion(
672 audience=authority.token_endpoint, issuer=self.client_id,
673 additional_claims=self.client_claims or {})
674 else:
675 default_body['client_secret'] = client_credential
676 central_configuration = {
677 "authorization_endpoint": authority.authorization_endpoint,
678 "token_endpoint": authority.token_endpoint,
679 "device_authorization_endpoint":
680 authority.device_authorization_endpoint or
681 urljoin(authority.token_endpoint, "devicecode"),
682 }
683 central_client = _ClientWithCcsRoutingInfo(
684 central_configuration,
685 self.client_id,
686 http_client=self.http_client,
687 default_headers=default_headers,
688 default_body=default_body,
689 client_assertion=client_assertion,
690 client_assertion_type=client_assertion_type,
691 on_obtaining_tokens=lambda event: self.token_cache.add(dict(
692 event, environment=authority.instance)),
693 on_removing_rt=self.token_cache.remove_rt,
694 on_updating_rt=self.token_cache.update_rt)
696 regional_client = None
697 if (client_credential # Currently regional endpoint only serves some CCA flows
698 and not skip_regional_client):
699 regional_authority = self._get_regional_authority(authority)
700 if regional_authority:
701 regional_configuration = {
702 "authorization_endpoint": regional_authority.authorization_endpoint,
703 "token_endpoint": regional_authority.token_endpoint,
704 "device_authorization_endpoint":
705 regional_authority.device_authorization_endpoint or
706 urljoin(regional_authority.token_endpoint, "devicecode"),
707 }
708 regional_client = _ClientWithCcsRoutingInfo(
709 regional_configuration,
710 self.client_id,
711 http_client=self.http_client,
712 default_headers=default_headers,
713 default_body=default_body,
714 client_assertion=client_assertion,
715 client_assertion_type=client_assertion_type,
716 on_obtaining_tokens=lambda event: self.token_cache.add(dict(
717 event, environment=authority.instance)),
718 on_removing_rt=self.token_cache.remove_rt,
719 on_updating_rt=self.token_cache.update_rt)
720 return central_client, regional_client
722 def initiate_auth_code_flow(
723 self,
724 scopes, # type: list[str]
725 redirect_uri=None,
726 state=None, # Recommended by OAuth2 for CSRF protection
727 prompt=None,
728 login_hint=None, # type: Optional[str]
729 domain_hint=None, # type: Optional[str]
730 claims_challenge=None,
731 max_age=None,
732 response_mode=None, # type: Optional[str]
733 ):
734 """Initiate an auth code flow.
736 Later when the response reaches your redirect_uri,
737 you can use :func:`~acquire_token_by_auth_code_flow()`
738 to complete the authentication/authorization.
740 :param list scopes:
741 It is a list of case-sensitive strings.
742 :param str redirect_uri:
743 Optional. If not specified, server will use the pre-registered one.
744 :param str state:
745 An opaque value used by the client to
746 maintain state between the request and callback.
747 If absent, this library will automatically generate one internally.
748 :param str prompt:
749 By default, no prompt value will be sent, not even "none".
750 You will have to specify a value explicitly.
751 Its valid values are defined in Open ID Connect specs
752 https://openid.net/specs/openid-connect-core-1_0.html#AuthRequest
753 :param str login_hint:
754 Optional. Identifier of the user. Generally a User Principal Name (UPN).
755 :param domain_hint:
756 Can be one of "consumers" or "organizations" or your tenant domain "contoso.com".
757 If included, it will skip the email-based discovery process that user goes
758 through on the sign-in page, leading to a slightly more streamlined user experience.
759 More information on possible values
760 `here <https://docs.microsoft.com/en-us/azure/active-directory/develop/v2-oauth2-auth-code-flow#request-an-authorization-code>`_ and
761 `here <https://docs.microsoft.com/en-us/openspecs/windows_protocols/ms-oapx/86fb452d-e34a-494e-ac61-e526e263b6d8>`_.
763 :param int max_age:
764 OPTIONAL. Maximum Authentication Age.
765 Specifies the allowable elapsed time in seconds
766 since the last time the End-User was actively authenticated.
767 If the elapsed time is greater than this value,
768 Microsoft identity platform will actively re-authenticate the End-User.
770 MSAL Python will also automatically validate the auth_time in ID token.
772 New in version 1.15.
774 :param str response_mode:
775 OPTIONAL. Specifies the method with which response parameters should be returned.
776 The default value is equivalent to ``query``, which is still secure enough in MSAL Python
777 (because MSAL Python does not transfer tokens via query parameter in the first place).
778 For even better security, we recommend using the value ``form_post``.
779 In "form_post" mode, response parameters
780 will be encoded as HTML form values that are transmitted via the HTTP POST method and
781 encoded in the body using the application/x-www-form-urlencoded format.
782 Valid values can be either "form_post" for HTTP POST to callback URI or
783 "query" (the default) for HTTP GET with parameters encoded in query string.
784 More information on possible values
785 `here <https://openid.net/specs/oauth-v2-multiple-response-types-1_0.html#ResponseModes>`
786 and `here <https://openid.net/specs/oauth-v2-form-post-response-mode-1_0.html#FormPostResponseMode>`
788 :return:
789 The auth code flow. It is a dict in this form::
791 {
792 "auth_uri": "https://...", // Guide user to visit this
793 "state": "...", // You may choose to verify it by yourself,
794 // or just let acquire_token_by_auth_code_flow()
795 // do that for you.
796 "...": "...", // Everything else are reserved and internal
797 }
799 The caller is expected to::
801 1. somehow store this content, typically inside the current session,
802 2. guide the end user (i.e. resource owner) to visit that auth_uri,
803 3. and then relay this dict and subsequent auth response to
804 :func:`~acquire_token_by_auth_code_flow()`.
805 """
806 client = _ClientWithCcsRoutingInfo(
807 {"authorization_endpoint": self.authority.authorization_endpoint},
808 self.client_id,
809 http_client=self.http_client)
810 flow = client.initiate_auth_code_flow(
811 redirect_uri=redirect_uri, state=state, login_hint=login_hint,
812 prompt=prompt,
813 scope=self._decorate_scope(scopes),
814 domain_hint=domain_hint,
815 claims=_merge_claims_challenge_and_capabilities(
816 self._client_capabilities, claims_challenge),
817 max_age=max_age,
818 response_mode=response_mode,
819 )
820 flow["claims_challenge"] = claims_challenge
821 return flow
823 def get_authorization_request_url(
824 self,
825 scopes, # type: list[str]
826 login_hint=None, # type: Optional[str]
827 state=None, # Recommended by OAuth2 for CSRF protection
828 redirect_uri=None,
829 response_type="code", # Could be "token" if you use Implicit Grant
830 prompt=None,
831 nonce=None,
832 domain_hint=None, # type: Optional[str]
833 claims_challenge=None,
834 **kwargs):
835 """Constructs a URL for you to start a Authorization Code Grant.
837 :param list[str] scopes: (Required)
838 Scopes requested to access a protected API (a resource).
839 :param str state: Recommended by OAuth2 for CSRF protection.
840 :param str login_hint:
841 Identifier of the user. Generally a User Principal Name (UPN).
842 :param str redirect_uri:
843 Address to return to upon receiving a response from the authority.
844 :param str response_type:
845 Default value is "code" for an OAuth2 Authorization Code grant.
847 You could use other content such as "id_token" or "token",
848 which would trigger an Implicit Grant, but that is
849 `not recommended <https://docs.microsoft.com/en-us/azure/active-directory/develop/v2-oauth2-implicit-grant-flow#is-the-implicit-grant-suitable-for-my-app>`_.
851 :param str prompt:
852 By default, no prompt value will be sent, not even "none".
853 You will have to specify a value explicitly.
854 Its valid values are defined in Open ID Connect specs
855 https://openid.net/specs/openid-connect-core-1_0.html#AuthRequest
856 :param nonce:
857 A cryptographically random value used to mitigate replay attacks. See also
858 `OIDC specs <https://openid.net/specs/openid-connect-core-1_0.html#AuthRequest>`_.
859 :param domain_hint:
860 Can be one of "consumers" or "organizations" or your tenant domain "contoso.com".
861 If included, it will skip the email-based discovery process that user goes
862 through on the sign-in page, leading to a slightly more streamlined user experience.
863 More information on possible values
864 `here <https://docs.microsoft.com/en-us/azure/active-directory/develop/v2-oauth2-auth-code-flow#request-an-authorization-code>`_ and
865 `here <https://docs.microsoft.com/en-us/openspecs/windows_protocols/ms-oapx/86fb452d-e34a-494e-ac61-e526e263b6d8>`_.
866 :param claims_challenge:
867 The claims_challenge parameter requests specific claims requested by the resource provider
868 in the form of a claims_challenge directive in the www-authenticate header to be
869 returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
870 It is a string of a JSON object which contains lists of claims being requested from these locations.
872 :return: The authorization url as a string.
873 """
874 authority = kwargs.pop("authority", None) # Historically we support this
875 if authority:
876 warnings.warn(
877 "We haven't decided if this method will accept authority parameter")
878 # The previous implementation is, it will use self.authority by default.
879 # Multi-tenant app can use new authority on demand
880 the_authority = Authority(
881 authority,
882 self.http_client,
883 instance_discovery=self._instance_discovery,
884 ) if authority else self.authority
886 client = _ClientWithCcsRoutingInfo(
887 {"authorization_endpoint": the_authority.authorization_endpoint},
888 self.client_id,
889 http_client=self.http_client)
890 warnings.warn(
891 "Change your get_authorization_request_url() "
892 "to initiate_auth_code_flow()", DeprecationWarning)
893 with warnings.catch_warnings(record=True):
894 return client.build_auth_request_uri(
895 response_type=response_type,
896 redirect_uri=redirect_uri, state=state, login_hint=login_hint,
897 prompt=prompt,
898 scope=self._decorate_scope(scopes),
899 nonce=nonce,
900 domain_hint=domain_hint,
901 claims=_merge_claims_challenge_and_capabilities(
902 self._client_capabilities, claims_challenge),
903 )
905 def acquire_token_by_auth_code_flow(
906 self, auth_code_flow, auth_response, scopes=None, **kwargs):
907 """Validate the auth response being redirected back, and obtain tokens.
909 It automatically provides nonce protection.
911 :param dict auth_code_flow:
912 The same dict returned by :func:`~initiate_auth_code_flow()`.
913 :param dict auth_response:
914 A dict of the query string received from auth server.
915 :param list[str] scopes:
916 Scopes requested to access a protected API (a resource).
918 Most of the time, you can leave it empty.
920 If you requested user consent for multiple resources, here you will
921 need to provide a subset of what you required in
922 :func:`~initiate_auth_code_flow()`.
924 OAuth2 was designed mostly for singleton services,
925 where tokens are always meant for the same resource and the only
926 changes are in the scopes.
927 In AAD, tokens can be issued for multiple 3rd party resources.
928 You can ask authorization code for multiple resources,
929 but when you redeem it, the token is for only one intended
930 recipient, called audience.
931 So the developer need to specify a scope so that we can restrict the
932 token to be issued for the corresponding audience.
934 :return:
935 * A dict containing "access_token" and/or "id_token", among others,
936 depends on what scope was used.
937 (See https://tools.ietf.org/html/rfc6749#section-5.1)
938 * A dict containing "error", optionally "error_description", "error_uri".
939 (It is either `this <https://tools.ietf.org/html/rfc6749#section-4.1.2.1>`_
940 or `that <https://tools.ietf.org/html/rfc6749#section-5.2>`_)
941 * Most client-side data error would result in ValueError exception.
942 So the usage pattern could be without any protocol details::
944 def authorize(): # A controller in a web app
945 try:
946 result = msal_app.acquire_token_by_auth_code_flow(
947 session.get("flow", {}), request.args)
948 if "error" in result:
949 return render_template("error.html", result)
950 use(result) # Token(s) are available in result and cache
951 except ValueError: # Usually caused by CSRF
952 pass # Simply ignore them
953 return redirect(url_for("index"))
954 """
955 self._validate_ssh_cert_input_data(kwargs.get("data", {}))
956 telemetry_context = self._build_telemetry_context(
957 self.ACQUIRE_TOKEN_BY_AUTHORIZATION_CODE_ID)
958 response =_clean_up(self.client.obtain_token_by_auth_code_flow(
959 auth_code_flow,
960 auth_response,
961 scope=self._decorate_scope(scopes) if scopes else None,
962 headers=telemetry_context.generate_headers(),
963 data=dict(
964 kwargs.pop("data", {}),
965 claims=_merge_claims_challenge_and_capabilities(
966 self._client_capabilities,
967 auth_code_flow.pop("claims_challenge", None))),
968 **kwargs))
969 telemetry_context.update_telemetry(response)
970 return response
972 def acquire_token_by_authorization_code(
973 self,
974 code,
975 scopes, # Syntactically required. STS accepts empty value though.
976 redirect_uri=None,
977 # REQUIRED, if the "redirect_uri" parameter was included in the
978 # authorization request as described in Section 4.1.1, and their
979 # values MUST be identical.
980 nonce=None,
981 claims_challenge=None,
982 **kwargs):
983 """The second half of the Authorization Code Grant.
985 :param code: The authorization code returned from Authorization Server.
986 :param list[str] scopes: (Required)
987 Scopes requested to access a protected API (a resource).
989 If you requested user consent for multiple resources, here you will
990 typically want to provide a subset of what you required in AuthCode.
992 OAuth2 was designed mostly for singleton services,
993 where tokens are always meant for the same resource and the only
994 changes are in the scopes.
995 In AAD, tokens can be issued for multiple 3rd party resources.
996 You can ask authorization code for multiple resources,
997 but when you redeem it, the token is for only one intended
998 recipient, called audience.
999 So the developer need to specify a scope so that we can restrict the
1000 token to be issued for the corresponding audience.
1002 :param nonce:
1003 If you provided a nonce when calling :func:`get_authorization_request_url`,
1004 same nonce should also be provided here, so that we'll validate it.
1005 An exception will be raised if the nonce in id token mismatches.
1007 :param claims_challenge:
1008 The claims_challenge parameter requests specific claims requested by the resource provider
1009 in the form of a claims_challenge directive in the www-authenticate header to be
1010 returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
1011 It is a string of a JSON object which contains lists of claims being requested from these locations.
1013 :return: A dict representing the json response from AAD:
1015 - A successful response would contain "access_token" key,
1016 - an error response would contain "error" and usually "error_description".
1017 """
1018 # If scope is absent on the wire, STS will give you a token associated
1019 # to the FIRST scope sent during the authorization request.
1020 # So in theory, you can omit scope here when you were working with only
1021 # one scope. But, MSAL decorates your scope anyway, so they are never
1022 # really empty.
1023 assert isinstance(scopes, list), "Invalid parameter type"
1024 self._validate_ssh_cert_input_data(kwargs.get("data", {}))
1025 warnings.warn(
1026 "Change your acquire_token_by_authorization_code() "
1027 "to acquire_token_by_auth_code_flow()", DeprecationWarning)
1028 with warnings.catch_warnings(record=True):
1029 telemetry_context = self._build_telemetry_context(
1030 self.ACQUIRE_TOKEN_BY_AUTHORIZATION_CODE_ID)
1031 response = _clean_up(self.client.obtain_token_by_authorization_code(
1032 code, redirect_uri=redirect_uri,
1033 scope=self._decorate_scope(scopes),
1034 headers=telemetry_context.generate_headers(),
1035 data=dict(
1036 kwargs.pop("data", {}),
1037 claims=_merge_claims_challenge_and_capabilities(
1038 self._client_capabilities, claims_challenge)),
1039 nonce=nonce,
1040 **kwargs))
1041 telemetry_context.update_telemetry(response)
1042 return response
1044 def get_accounts(self, username=None):
1045 """Get a list of accounts which previously signed in, i.e. exists in cache.
1047 An account can later be used in :func:`~acquire_token_silent`
1048 to find its tokens.
1050 :param username:
1051 Filter accounts with this username only. Case insensitive.
1052 :return: A list of account objects.
1053 Each account is a dict. For now, we only document its "username" field.
1054 Your app can choose to display those information to end user,
1055 and allow user to choose one of his/her accounts to proceed.
1056 """
1057 accounts = self._find_msal_accounts(environment=self.authority.instance)
1058 if not accounts: # Now try other aliases of this authority instance
1059 for alias in self._get_authority_aliases(self.authority.instance):
1060 accounts = self._find_msal_accounts(environment=alias)
1061 if accounts:
1062 break
1063 if username:
1064 # Federated account["username"] from AAD could contain mixed case
1065 lowercase_username = username.lower()
1066 accounts = [a for a in accounts
1067 if a["username"].lower() == lowercase_username]
1068 if not accounts:
1069 logger.debug(( # This would also happen when the cache is empty
1070 "get_accounts(username='{}') finds no account. "
1071 "If tokens were acquired without 'profile' scope, "
1072 "they would contain no username for filtering. "
1073 "Consider calling get_accounts(username=None) instead."
1074 ).format(username))
1075 # Does not further filter by existing RTs here. It probably won't matter.
1076 # Because in most cases Accounts and RTs co-exist.
1077 # Even in the rare case when an RT is revoked and then removed,
1078 # acquire_token_silent() would then yield no result,
1079 # apps would fall back to other acquire methods. This is the standard pattern.
1080 return accounts
1082 def _find_msal_accounts(self, environment):
1083 interested_authority_types = [
1084 TokenCache.AuthorityType.ADFS, TokenCache.AuthorityType.MSSTS]
1085 if _is_running_in_cloud_shell():
1086 interested_authority_types.append(_AUTHORITY_TYPE_CLOUDSHELL)
1087 grouped_accounts = {
1088 a.get("home_account_id"): # Grouped by home tenant's id
1089 { # These are minimal amount of non-tenant-specific account info
1090 "home_account_id": a.get("home_account_id"),
1091 "environment": a.get("environment"),
1092 "username": a.get("username"),
1094 # The following fields for backward compatibility, for now
1095 "authority_type": a.get("authority_type"),
1096 "local_account_id": a.get("local_account_id"), # Tenant-specific
1097 "realm": a.get("realm"), # Tenant-specific
1098 }
1099 for a in self.token_cache.find(
1100 TokenCache.CredentialType.ACCOUNT,
1101 query={"environment": environment})
1102 if a["authority_type"] in interested_authority_types
1103 }
1104 return list(grouped_accounts.values())
1106 def _get_instance_metadata(self): # This exists so it can be mocked in unit test
1107 resp = self.http_client.get(
1108 "https://login.microsoftonline.com/common/discovery/instance?api-version=1.1&authorization_endpoint=https://login.microsoftonline.com/common/oauth2/authorize", # TBD: We may extend this to use self._instance_discovery endpoint
1109 headers={'Accept': 'application/json'})
1110 resp.raise_for_status()
1111 return json.loads(resp.text)['metadata']
1113 def _get_authority_aliases(self, instance):
1114 if self._instance_discovery is False:
1115 return []
1116 if self.authority._is_known_to_developer:
1117 # Then it is an ADFS/B2C/known_authority_hosts situation
1118 # which may not reach the central endpoint, so we skip it.
1119 return []
1120 if not self.authority_groups:
1121 self.authority_groups = [
1122 set(group['aliases']) for group in self._get_instance_metadata()]
1123 for group in self.authority_groups:
1124 if instance in group:
1125 return [alias for alias in group if alias != instance]
1126 return []
1128 def remove_account(self, account):
1129 """Sign me out and forget me from token cache"""
1130 self._forget_me(account)
1131 if self._enable_broker:
1132 from .broker import _signout_silently
1133 error = _signout_silently(self.client_id, account["local_account_id"])
1134 if error:
1135 logger.debug("_signout_silently() returns error: %s", error)
1137 def _sign_out(self, home_account):
1138 # Remove all relevant RTs and ATs from token cache
1139 owned_by_home_account = {
1140 "environment": home_account["environment"],
1141 "home_account_id": home_account["home_account_id"],} # realm-independent
1142 app_metadata = self._get_app_metadata(home_account["environment"])
1143 # Remove RTs/FRTs, and they are realm-independent
1144 for rt in [rt for rt in self.token_cache.find(
1145 TokenCache.CredentialType.REFRESH_TOKEN, query=owned_by_home_account)
1146 # Do RT's app ownership check as a precaution, in case family apps
1147 # and 3rd-party apps share same token cache, although they should not.
1148 if rt["client_id"] == self.client_id or (
1149 app_metadata.get("family_id") # Now let's settle family business
1150 and rt.get("family_id") == app_metadata["family_id"])
1151 ]:
1152 self.token_cache.remove_rt(rt)
1153 for at in self.token_cache.find( # Remove ATs
1154 # Regardless of realm, b/c we've removed realm-independent RTs anyway
1155 TokenCache.CredentialType.ACCESS_TOKEN, query=owned_by_home_account):
1156 # To avoid the complexity of locating sibling family app's AT,
1157 # we skip AT's app ownership check.
1158 # It means ATs for other apps will also be removed, it is OK because:
1159 # * non-family apps are not supposed to share token cache to begin with;
1160 # * Even if it happens, we keep other app's RT already, so SSO still works
1161 self.token_cache.remove_at(at)
1163 def _forget_me(self, home_account):
1164 # It implies signout, and then also remove all relevant accounts and IDTs
1165 self._sign_out(home_account)
1166 owned_by_home_account = {
1167 "environment": home_account["environment"],
1168 "home_account_id": home_account["home_account_id"],} # realm-independent
1169 for idt in self.token_cache.find( # Remove IDTs, regardless of realm
1170 TokenCache.CredentialType.ID_TOKEN, query=owned_by_home_account):
1171 self.token_cache.remove_idt(idt)
1172 for a in self.token_cache.find( # Remove Accounts, regardless of realm
1173 TokenCache.CredentialType.ACCOUNT, query=owned_by_home_account):
1174 self.token_cache.remove_account(a)
1176 def _acquire_token_by_cloud_shell(self, scopes, data=None):
1177 from .cloudshell import _obtain_token
1178 response = _obtain_token(
1179 self.http_client, scopes, client_id=self.client_id, data=data)
1180 if "error" not in response:
1181 self.token_cache.add(dict(
1182 client_id=self.client_id,
1183 scope=response["scope"].split() if "scope" in response else scopes,
1184 token_endpoint=self.authority.token_endpoint,
1185 response=response,
1186 data=data or {},
1187 authority_type=_AUTHORITY_TYPE_CLOUDSHELL,
1188 ))
1189 return response
1191 def acquire_token_silent(
1192 self,
1193 scopes, # type: List[str]
1194 account, # type: Optional[Account]
1195 authority=None, # See get_authorization_request_url()
1196 force_refresh=False, # type: Optional[boolean]
1197 claims_challenge=None,
1198 **kwargs):
1199 """Acquire an access token for given account, without user interaction.
1201 It is done either by finding a valid access token from cache,
1202 or by finding a valid refresh token from cache and then automatically
1203 use it to redeem a new access token.
1205 This method will combine the cache empty and refresh error
1206 into one return value, `None`.
1207 If your app does not care about the exact token refresh error during
1208 token cache look-up, then this method is easier and recommended.
1210 Internally, this method calls :func:`~acquire_token_silent_with_error`.
1212 :param claims_challenge:
1213 The claims_challenge parameter requests specific claims requested by the resource provider
1214 in the form of a claims_challenge directive in the www-authenticate header to be
1215 returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
1216 It is a string of a JSON object which contains lists of claims being requested from these locations.
1218 :return:
1219 - A dict containing no "error" key,
1220 and typically contains an "access_token" key,
1221 if cache lookup succeeded.
1222 - None when cache lookup does not yield a token.
1223 """
1224 result = self.acquire_token_silent_with_error(
1225 scopes, account, authority=authority, force_refresh=force_refresh,
1226 claims_challenge=claims_challenge, **kwargs)
1227 return result if result and "error" not in result else None
1229 def acquire_token_silent_with_error(
1230 self,
1231 scopes, # type: List[str]
1232 account, # type: Optional[Account]
1233 authority=None, # See get_authorization_request_url()
1234 force_refresh=False, # type: Optional[boolean]
1235 claims_challenge=None,
1236 **kwargs):
1237 """Acquire an access token for given account, without user interaction.
1239 It is done either by finding a valid access token from cache,
1240 or by finding a valid refresh token from cache and then automatically
1241 use it to redeem a new access token.
1243 This method will differentiate cache empty from token refresh error.
1244 If your app cares the exact token refresh error during
1245 token cache look-up, then this method is suitable.
1246 Otherwise, the other method :func:`~acquire_token_silent` is recommended.
1248 :param list[str] scopes: (Required)
1249 Scopes requested to access a protected API (a resource).
1250 :param account:
1251 one of the account object returned by :func:`~get_accounts`,
1252 or use None when you want to find an access token for this client.
1253 :param force_refresh:
1254 If True, it will skip Access Token look-up,
1255 and try to find a Refresh Token to obtain a new Access Token.
1256 :param claims_challenge:
1257 The claims_challenge parameter requests specific claims requested by the resource provider
1258 in the form of a claims_challenge directive in the www-authenticate header to be
1259 returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
1260 It is a string of a JSON object which contains lists of claims being requested from these locations.
1261 :return:
1262 - A dict containing no "error" key,
1263 and typically contains an "access_token" key,
1264 if cache lookup succeeded.
1265 - None when there is simply no token in the cache.
1266 - A dict containing an "error" key, when token refresh failed.
1267 """
1268 assert isinstance(scopes, list), "Invalid parameter type"
1269 self._validate_ssh_cert_input_data(kwargs.get("data", {}))
1270 correlation_id = msal.telemetry._get_new_correlation_id()
1271 if authority:
1272 warnings.warn("We haven't decided how/if this method will accept authority parameter")
1273 # the_authority = Authority(
1274 # authority,
1275 # self.http_client,
1276 # instance_discovery=self._instance_discovery,
1277 # ) if authority else self.authority
1278 result = self._acquire_token_silent_from_cache_and_possibly_refresh_it(
1279 scopes, account, self.authority, force_refresh=force_refresh,
1280 claims_challenge=claims_challenge,
1281 correlation_id=correlation_id,
1282 **kwargs)
1283 if result and "error" not in result:
1284 return result
1285 final_result = result
1286 for alias in self._get_authority_aliases(self.authority.instance):
1287 if not self.token_cache.find(
1288 self.token_cache.CredentialType.REFRESH_TOKEN,
1289 # target=scopes, # MUST NOT filter by scopes, because:
1290 # 1. AAD RTs are scope-independent;
1291 # 2. therefore target is optional per schema;
1292 query={"environment": alias}):
1293 # Skip heavy weight logic when RT for this alias doesn't exist
1294 continue
1295 the_authority = Authority(
1296 "https://" + alias + "/" + self.authority.tenant,
1297 self.http_client,
1298 instance_discovery=False,
1299 )
1300 result = self._acquire_token_silent_from_cache_and_possibly_refresh_it(
1301 scopes, account, the_authority, force_refresh=force_refresh,
1302 claims_challenge=claims_challenge,
1303 correlation_id=correlation_id,
1304 **kwargs)
1305 if result:
1306 if "error" not in result:
1307 return result
1308 final_result = result
1309 if final_result and final_result.get("suberror"):
1310 final_result["classification"] = { # Suppress these suberrors, per #57
1311 "bad_token": "",
1312 "token_expired": "",
1313 "protection_policy_required": "",
1314 "client_mismatch": "",
1315 "device_authentication_failed": "",
1316 }.get(final_result["suberror"], final_result["suberror"])
1317 return final_result
1319 def _acquire_token_silent_from_cache_and_possibly_refresh_it(
1320 self,
1321 scopes, # type: List[str]
1322 account, # type: Optional[Account]
1323 authority, # This can be different than self.authority
1324 force_refresh=False, # type: Optional[boolean]
1325 claims_challenge=None,
1326 correlation_id=None,
1327 **kwargs):
1328 access_token_from_cache = None
1329 if not (force_refresh or claims_challenge): # Bypass AT when desired or using claims
1330 query={
1331 "client_id": self.client_id,
1332 "environment": authority.instance,
1333 "realm": authority.tenant,
1334 "home_account_id": (account or {}).get("home_account_id"),
1335 }
1336 key_id = kwargs.get("data", {}).get("key_id")
1337 if key_id: # Some token types (SSH-certs, POP) are bound to a key
1338 query["key_id"] = key_id
1339 matches = self.token_cache.find(
1340 self.token_cache.CredentialType.ACCESS_TOKEN,
1341 target=scopes,
1342 query=query)
1343 now = time.time()
1344 refresh_reason = msal.telemetry.AT_ABSENT
1345 for entry in matches:
1346 expires_in = int(entry["expires_on"]) - now
1347 if expires_in < 5*60: # Then consider it expired
1348 refresh_reason = msal.telemetry.AT_EXPIRED
1349 continue # Removal is not necessary, it will be overwritten
1350 logger.debug("Cache hit an AT")
1351 access_token_from_cache = { # Mimic a real response
1352 "access_token": entry["secret"],
1353 "token_type": entry.get("token_type", "Bearer"),
1354 "expires_in": int(expires_in), # OAuth2 specs defines it as int
1355 }
1356 if "refresh_on" in entry and int(entry["refresh_on"]) < now: # aging
1357 refresh_reason = msal.telemetry.AT_AGING
1358 break # With a fallback in hand, we break here to go refresh
1359 self._build_telemetry_context(-1).hit_an_access_token()
1360 return access_token_from_cache # It is still good as new
1361 else:
1362 refresh_reason = msal.telemetry.FORCE_REFRESH # TODO: It could also mean claims_challenge
1363 assert refresh_reason, "It should have been established at this point"
1364 try:
1365 data = kwargs.get("data", {})
1366 if account and account.get("authority_type") == _AUTHORITY_TYPE_CLOUDSHELL:
1367 return self._acquire_token_by_cloud_shell(scopes, data=data)
1369 if self._enable_broker and account is not None:
1370 from .broker import _acquire_token_silently
1371 response = _acquire_token_silently(
1372 "https://{}/{}".format(self.authority.instance, self.authority.tenant),
1373 self.client_id,
1374 account["local_account_id"],
1375 scopes,
1376 claims=_merge_claims_challenge_and_capabilities(
1377 self._client_capabilities, claims_challenge),
1378 correlation_id=correlation_id,
1379 **data)
1380 if response: # The broker provided a decisive outcome, so we use it
1381 return self._process_broker_response(response, scopes, data)
1383 result = _clean_up(self._acquire_token_silent_by_finding_rt_belongs_to_me_or_my_family(
1384 authority, self._decorate_scope(scopes), account,
1385 refresh_reason=refresh_reason, claims_challenge=claims_challenge,
1386 correlation_id=correlation_id,
1387 **kwargs))
1388 if (result and "error" not in result) or (not access_token_from_cache):
1389 return result
1390 except: # The exact HTTP exception is transportation-layer dependent
1391 # Typically network error. Potential AAD outage?
1392 if not access_token_from_cache: # It means there is no fall back option
1393 raise # We choose to bubble up the exception
1394 return access_token_from_cache
1396 def _process_broker_response(self, response, scopes, data):
1397 if "error" not in response:
1398 self.token_cache.add(dict(
1399 client_id=self.client_id,
1400 scope=response["scope"].split() if "scope" in response else scopes,
1401 token_endpoint=self.authority.token_endpoint,
1402 response=response,
1403 data=data,
1404 _account_id=response["_account_id"],
1405 ))
1406 return _clean_up(response)
1408 def _acquire_token_silent_by_finding_rt_belongs_to_me_or_my_family(
1409 self, authority, scopes, account, **kwargs):
1410 query = {
1411 "environment": authority.instance,
1412 "home_account_id": (account or {}).get("home_account_id"),
1413 # "realm": authority.tenant, # AAD RTs are tenant-independent
1414 }
1415 app_metadata = self._get_app_metadata(authority.instance)
1416 if not app_metadata: # Meaning this app is now used for the first time.
1417 # When/if we have a way to directly detect current app's family,
1418 # we'll rewrite this block, to support multiple families.
1419 # For now, we try existing RTs (*). If it works, we are in that family.
1420 # (*) RTs of a different app/family are not supposed to be
1421 # shared with or accessible by us in the first place.
1422 at = self._acquire_token_silent_by_finding_specific_refresh_token(
1423 authority, scopes,
1424 dict(query, family_id="1"), # A hack, we have only 1 family for now
1425 rt_remover=lambda rt_item: None, # NO-OP b/c RTs are likely not mine
1426 break_condition=lambda response: # Break loop when app not in family
1427 # Based on an AAD-only behavior mentioned in internal doc here
1428 # https://msazure.visualstudio.com/One/_git/ESTS-Docs/pullrequest/1138595
1429 "client_mismatch" in response.get("error_additional_info", []),
1430 **kwargs)
1431 if at and "error" not in at:
1432 return at
1433 last_resp = None
1434 if app_metadata.get("family_id"): # Meaning this app belongs to this family
1435 last_resp = at = self._acquire_token_silent_by_finding_specific_refresh_token(
1436 authority, scopes, dict(query, family_id=app_metadata["family_id"]),
1437 **kwargs)
1438 if at and "error" not in at:
1439 return at
1440 # Either this app is an orphan, so we will naturally use its own RT;
1441 # or all attempts above have failed, so we fall back to non-foci behavior.
1442 return self._acquire_token_silent_by_finding_specific_refresh_token(
1443 authority, scopes, dict(query, client_id=self.client_id),
1444 **kwargs) or last_resp
1446 def _get_app_metadata(self, environment):
1447 apps = self.token_cache.find( # Use find(), rather than token_cache.get(...)
1448 TokenCache.CredentialType.APP_METADATA, query={
1449 "environment": environment, "client_id": self.client_id})
1450 return apps[0] if apps else {}
1452 def _acquire_token_silent_by_finding_specific_refresh_token(
1453 self, authority, scopes, query,
1454 rt_remover=None, break_condition=lambda response: False,
1455 refresh_reason=None, correlation_id=None, claims_challenge=None,
1456 **kwargs):
1457 matches = self.token_cache.find(
1458 self.token_cache.CredentialType.REFRESH_TOKEN,
1459 # target=scopes, # AAD RTs are scope-independent
1460 query=query)
1461 logger.debug("Found %d RTs matching %s", len(matches), query)
1463 response = None # A distinguishable value to mean cache is empty
1464 if not matches: # Then exit early to avoid expensive operations
1465 return response
1466 client, _ = self._build_client(
1467 # Potentially expensive if building regional client
1468 self.client_credential, authority, skip_regional_client=True)
1469 telemetry_context = self._build_telemetry_context(
1470 self.ACQUIRE_TOKEN_SILENT_ID,
1471 correlation_id=correlation_id, refresh_reason=refresh_reason)
1472 for entry in sorted( # Since unfit RTs would not be aggressively removed,
1473 # we start from newer RTs which are more likely fit.
1474 matches,
1475 key=lambda e: int(e.get("last_modification_time", "0")),
1476 reverse=True):
1477 logger.debug("Cache attempts an RT")
1478 headers = telemetry_context.generate_headers()
1479 if query.get("home_account_id"): # Then use it as CCS Routing info
1480 headers["X-AnchorMailbox"] = "Oid:{}".format( # case-insensitive value
1481 query["home_account_id"].replace(".", "@"))
1482 response = client.obtain_token_by_refresh_token(
1483 entry, rt_getter=lambda token_item: token_item["secret"],
1484 on_removing_rt=lambda rt_item: None, # Disable RT removal,
1485 # because an invalid_grant could be caused by new MFA policy,
1486 # the RT could still be useful for other MFA-less scope or tenant
1487 on_obtaining_tokens=lambda event: self.token_cache.add(dict(
1488 event,
1489 environment=authority.instance,
1490 skip_account_creation=True, # To honor a concurrent remove_account()
1491 )),
1492 scope=scopes,
1493 headers=headers,
1494 data=dict(
1495 kwargs.pop("data", {}),
1496 claims=_merge_claims_challenge_and_capabilities(
1497 self._client_capabilities, claims_challenge)),
1498 **kwargs)
1499 telemetry_context.update_telemetry(response)
1500 if "error" not in response:
1501 return response
1502 logger.debug("Refresh failed. {error}: {error_description}".format(
1503 error=response.get("error"),
1504 error_description=response.get("error_description"),
1505 ))
1506 if break_condition(response):
1507 break
1508 return response # Returns the latest error (if any), or just None
1510 def _validate_ssh_cert_input_data(self, data):
1511 if data.get("token_type") == "ssh-cert":
1512 if not data.get("req_cnf"):
1513 raise ValueError(
1514 "When requesting an SSH certificate, "
1515 "you must include a string parameter named 'req_cnf' "
1516 "containing the public key in JWK format "
1517 "(https://tools.ietf.org/html/rfc7517).")
1518 if not data.get("key_id"):
1519 raise ValueError(
1520 "When requesting an SSH certificate, "
1521 "you must include a string parameter named 'key_id' "
1522 "which identifies the key in the 'req_cnf' argument.")
1524 def acquire_token_by_refresh_token(self, refresh_token, scopes, **kwargs):
1525 """Acquire token(s) based on a refresh token (RT) obtained from elsewhere.
1527 You use this method only when you have old RTs from elsewhere,
1528 and now you want to migrate them into MSAL.
1529 Calling this method results in new tokens automatically storing into MSAL.
1531 You do NOT need to use this method if you are already using MSAL.
1532 MSAL maintains RT automatically inside its token cache,
1533 and an access token can be retrieved
1534 when you call :func:`~acquire_token_silent`.
1536 :param str refresh_token: The old refresh token, as a string.
1538 :param list scopes:
1539 The scopes associate with this old RT.
1540 Each scope needs to be in the Microsoft identity platform (v2) format.
1541 See `Scopes not resources <https://docs.microsoft.com/en-us/azure/active-directory/develop/migrate-python-adal-msal#scopes-not-resources>`_.
1543 :return:
1544 * A dict contains "error" and some other keys, when error happened.
1545 * A dict contains no "error" key means migration was successful.
1546 """
1547 self._validate_ssh_cert_input_data(kwargs.get("data", {}))
1548 telemetry_context = self._build_telemetry_context(
1549 self.ACQUIRE_TOKEN_BY_REFRESH_TOKEN,
1550 refresh_reason=msal.telemetry.FORCE_REFRESH)
1551 response = _clean_up(self.client.obtain_token_by_refresh_token(
1552 refresh_token,
1553 scope=self._decorate_scope(scopes),
1554 headers=telemetry_context.generate_headers(),
1555 rt_getter=lambda rt: rt,
1556 on_updating_rt=False,
1557 on_removing_rt=lambda rt_item: None, # No OP
1558 **kwargs))
1559 telemetry_context.update_telemetry(response)
1560 return response
1562 def acquire_token_by_username_password(
1563 self, username, password, scopes, claims_challenge=None, **kwargs):
1564 """Gets a token for a given resource via user credentials.
1566 See this page for constraints of Username Password Flow.
1567 https://github.com/AzureAD/microsoft-authentication-library-for-python/wiki/Username-Password-Authentication
1569 :param str username: Typically a UPN in the form of an email address.
1570 :param str password: The password.
1571 :param list[str] scopes:
1572 Scopes requested to access a protected API (a resource).
1573 :param claims_challenge:
1574 The claims_challenge parameter requests specific claims requested by the resource provider
1575 in the form of a claims_challenge directive in the www-authenticate header to be
1576 returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
1577 It is a string of a JSON object which contains lists of claims being requested from these locations.
1579 :return: A dict representing the json response from AAD:
1581 - A successful response would contain "access_token" key,
1582 - an error response would contain "error" and usually "error_description".
1583 """
1584 claims = _merge_claims_challenge_and_capabilities(
1585 self._client_capabilities, claims_challenge)
1586 if self._enable_broker:
1587 from .broker import _signin_silently
1588 response = _signin_silently(
1589 "https://{}/{}".format(self.authority.instance, self.authority.tenant),
1590 self.client_id,
1591 scopes, # Decorated scopes won't work due to offline_access
1592 MSALRuntime_Username=username,
1593 MSALRuntime_Password=password,
1594 validateAuthority="no" if (
1595 self.authority._is_known_to_developer
1596 or self._instance_discovery is False) else None,
1597 claims=claims,
1598 )
1599 return self._process_broker_response(response, scopes, kwargs.get("data", {}))
1601 scopes = self._decorate_scope(scopes)
1602 telemetry_context = self._build_telemetry_context(
1603 self.ACQUIRE_TOKEN_BY_USERNAME_PASSWORD_ID)
1604 headers = telemetry_context.generate_headers()
1605 data = dict(kwargs.pop("data", {}), claims=claims)
1606 if not self.authority.is_adfs:
1607 user_realm_result = self.authority.user_realm_discovery(
1608 username, correlation_id=headers[msal.telemetry.CLIENT_REQUEST_ID])
1609 if user_realm_result.get("account_type") == "Federated":
1610 response = _clean_up(self._acquire_token_by_username_password_federated(
1611 user_realm_result, username, password, scopes=scopes,
1612 data=data,
1613 headers=headers, **kwargs))
1614 telemetry_context.update_telemetry(response)
1615 return response
1616 response = _clean_up(self.client.obtain_token_by_username_password(
1617 username, password, scope=scopes,
1618 headers=headers,
1619 data=data,
1620 **kwargs))
1621 telemetry_context.update_telemetry(response)
1622 return response
1624 def _acquire_token_by_username_password_federated(
1625 self, user_realm_result, username, password, scopes=None, **kwargs):
1626 wstrust_endpoint = {}
1627 if user_realm_result.get("federation_metadata_url"):
1628 wstrust_endpoint = mex_send_request(
1629 user_realm_result["federation_metadata_url"],
1630 self.http_client)
1631 if wstrust_endpoint is None:
1632 raise ValueError("Unable to find wstrust endpoint from MEX. "
1633 "This typically happens when attempting MSA accounts. "
1634 "More details available here. "
1635 "https://github.com/AzureAD/microsoft-authentication-library-for-python/wiki/Username-Password-Authentication")
1636 logger.debug("wstrust_endpoint = %s", wstrust_endpoint)
1637 wstrust_result = wst_send_request(
1638 username, password,
1639 user_realm_result.get("cloud_audience_urn", "urn:federation:MicrosoftOnline"),
1640 wstrust_endpoint.get("address",
1641 # Fallback to an AAD supplied endpoint
1642 user_realm_result.get("federation_active_auth_url")),
1643 wstrust_endpoint.get("action"), self.http_client)
1644 if not ("token" in wstrust_result and "type" in wstrust_result):
1645 raise RuntimeError("Unsuccessful RSTR. %s" % wstrust_result)
1646 GRANT_TYPE_SAML1_1 = 'urn:ietf:params:oauth:grant-type:saml1_1-bearer'
1647 grant_type = {
1648 SAML_TOKEN_TYPE_V1: GRANT_TYPE_SAML1_1,
1649 SAML_TOKEN_TYPE_V2: self.client.GRANT_TYPE_SAML2,
1650 WSS_SAML_TOKEN_PROFILE_V1_1: GRANT_TYPE_SAML1_1,
1651 WSS_SAML_TOKEN_PROFILE_V2: self.client.GRANT_TYPE_SAML2
1652 }.get(wstrust_result.get("type"))
1653 if not grant_type:
1654 raise RuntimeError(
1655 "RSTR returned unknown token type: %s", wstrust_result.get("type"))
1656 self.client.grant_assertion_encoders.setdefault( # Register a non-standard type
1657 grant_type, self.client.encode_saml_assertion)
1658 return self.client.obtain_token_by_assertion(
1659 wstrust_result["token"], grant_type, scope=scopes,
1660 on_obtaining_tokens=lambda event: self.token_cache.add(dict(
1661 event,
1662 environment=self.authority.instance,
1663 username=username, # Useful in case IDT contains no such info
1664 )),
1665 **kwargs)
1668class PublicClientApplication(ClientApplication): # browser app or mobile app
1670 DEVICE_FLOW_CORRELATION_ID = "_correlation_id"
1671 CONSOLE_WINDOW_HANDLE = object()
1673 def __init__(self, client_id, client_credential=None, **kwargs):
1674 if client_credential is not None:
1675 raise ValueError("Public Client should not possess credentials")
1676 super(PublicClientApplication, self).__init__(
1677 client_id, client_credential=None, **kwargs)
1679 def acquire_token_interactive(
1680 self,
1681 scopes, # type: list[str]
1682 prompt=None,
1683 login_hint=None, # type: Optional[str]
1684 domain_hint=None, # type: Optional[str]
1685 claims_challenge=None,
1686 timeout=None,
1687 port=None,
1688 extra_scopes_to_consent=None,
1689 max_age=None,
1690 parent_window_handle=None,
1691 on_before_launching_ui=None,
1692 **kwargs):
1693 """Acquire token interactively i.e. via a local browser.
1695 Prerequisite: In Azure Portal, configure the Redirect URI of your
1696 "Mobile and Desktop application" as ``http://localhost``.
1697 If you opts in to use broker during ``PublicClientApplication`` creation,
1698 your app also need this Redirect URI:
1699 ``ms-appx-web://Microsoft.AAD.BrokerPlugin/YOUR_CLIENT_ID``
1701 :param list scopes:
1702 It is a list of case-sensitive strings.
1703 :param str prompt:
1704 By default, no prompt value will be sent, not even "none".
1705 You will have to specify a value explicitly.
1706 Its valid values are defined in Open ID Connect specs
1707 https://openid.net/specs/openid-connect-core-1_0.html#AuthRequest
1708 :param str login_hint:
1709 Optional. Identifier of the user. Generally a User Principal Name (UPN).
1710 :param domain_hint:
1711 Can be one of "consumers" or "organizations" or your tenant domain "contoso.com".
1712 If included, it will skip the email-based discovery process that user goes
1713 through on the sign-in page, leading to a slightly more streamlined user experience.
1714 More information on possible values
1715 `here <https://docs.microsoft.com/en-us/azure/active-directory/develop/v2-oauth2-auth-code-flow#request-an-authorization-code>`_ and
1716 `here <https://docs.microsoft.com/en-us/openspecs/windows_protocols/ms-oapx/86fb452d-e34a-494e-ac61-e526e263b6d8>`_.
1718 :param claims_challenge:
1719 The claims_challenge parameter requests specific claims requested by the resource provider
1720 in the form of a claims_challenge directive in the www-authenticate header to be
1721 returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
1722 It is a string of a JSON object which contains lists of claims being requested from these locations.
1724 :param int timeout:
1725 This method will block the current thread.
1726 This parameter specifies the timeout value in seconds.
1727 Default value ``None`` means wait indefinitely.
1729 :param int port:
1730 The port to be used to listen to an incoming auth response.
1731 By default we will use a system-allocated port.
1732 (The rest of the redirect_uri is hard coded as ``http://localhost``.)
1734 :param list extra_scopes_to_consent:
1735 "Extra scopes to consent" is a concept only available in AAD.
1736 It refers to other resources you might want to prompt to consent for,
1737 in the same interaction, but for which you won't get back a
1738 token for in this particular operation.
1740 :param int max_age:
1741 OPTIONAL. Maximum Authentication Age.
1742 Specifies the allowable elapsed time in seconds
1743 since the last time the End-User was actively authenticated.
1744 If the elapsed time is greater than this value,
1745 Microsoft identity platform will actively re-authenticate the End-User.
1747 MSAL Python will also automatically validate the auth_time in ID token.
1749 New in version 1.15.
1751 :param int parent_window_handle:
1752 OPTIONAL. If your app is a GUI app running on modern Windows system,
1753 and your app opts in to use broker,
1754 you are recommended to also provide its window handle,
1755 so that the sign in UI window will properly pop up on top of your window.
1757 New in version 1.20.0.
1759 :param function on_before_launching_ui:
1760 A callback with the form of
1761 ``lambda ui="xyz", **kwargs: print("A {} will be launched".format(ui))``,
1762 where ``ui`` will be either "browser" or "broker".
1763 You can use it to inform your end user to expect a pop-up window.
1765 New in version 1.20.0.
1767 :return:
1768 - A dict containing no "error" key,
1769 and typically contains an "access_token" key.
1770 - A dict containing an "error" key, when token refresh failed.
1771 """
1772 data = kwargs.pop("data", {})
1773 enable_msa_passthrough = kwargs.pop( # MUST remove it from kwargs
1774 "enable_msa_passthrough", # Keep it as a hidden param, for now.
1775 # OPTIONAL. MSA-Passthrough is a legacy configuration,
1776 # needed by a small amount of Microsoft first-party apps,
1777 # which would login MSA accounts via ".../organizations" authority.
1778 # If you app belongs to this category, AND you are enabling broker,
1779 # you would want to enable this flag. Default value is False.
1780 # More background of MSA-PT is available from this internal docs:
1781 # https://microsoft.sharepoint.com/:w:/t/Identity-DevEx/EatIUauX3c9Ctw1l7AQ6iM8B5CeBZxc58eoQCE0IuZ0VFw?e=tgc3jP&CID=39c853be-76ea-79d7-ee73-f1b2706ede05
1782 False
1783 ) and data.get("token_type") != "ssh-cert" # Work around a known issue as of PyMsalRuntime 0.8
1784 self._validate_ssh_cert_input_data(data)
1785 if not on_before_launching_ui:
1786 on_before_launching_ui = lambda **kwargs: None
1787 if _is_running_in_cloud_shell() and prompt == "none":
1788 # Note: _acquire_token_by_cloud_shell() is always silent,
1789 # so we would not fire on_before_launching_ui()
1790 return self._acquire_token_by_cloud_shell(scopes, data=data)
1791 claims = _merge_claims_challenge_and_capabilities(
1792 self._client_capabilities, claims_challenge)
1793 if self._enable_broker:
1794 if parent_window_handle is None:
1795 raise ValueError(
1796 "parent_window_handle is required when you opted into using broker. "
1797 "You need to provide the window handle of your GUI application, "
1798 "or use msal.PublicClientApplication.CONSOLE_WINDOW_HANDLE "
1799 "when and only when your application is a console app.")
1800 if extra_scopes_to_consent:
1801 logger.warning(
1802 "Ignoring parameter extra_scopes_to_consent, "
1803 "which is not supported by broker")
1804 return self._acquire_token_interactive_via_broker(
1805 scopes,
1806 parent_window_handle,
1807 enable_msa_passthrough,
1808 claims,
1809 data,
1810 on_before_launching_ui,
1811 prompt=prompt,
1812 login_hint=login_hint,
1813 max_age=max_age,
1814 )
1816 on_before_launching_ui(ui="browser")
1817 telemetry_context = self._build_telemetry_context(
1818 self.ACQUIRE_TOKEN_INTERACTIVE)
1819 response = _clean_up(self.client.obtain_token_by_browser(
1820 scope=self._decorate_scope(scopes) if scopes else None,
1821 extra_scope_to_consent=extra_scopes_to_consent,
1822 redirect_uri="http://localhost:{port}".format(
1823 # Hardcode the host, for now. AAD portal rejects 127.0.0.1 anyway
1824 port=port or 0),
1825 prompt=prompt,
1826 login_hint=login_hint,
1827 max_age=max_age,
1828 timeout=timeout,
1829 auth_params={
1830 "claims": claims,
1831 "domain_hint": domain_hint,
1832 },
1833 data=dict(data, claims=claims),
1834 headers=telemetry_context.generate_headers(),
1835 browser_name=_preferred_browser(),
1836 **kwargs))
1837 telemetry_context.update_telemetry(response)
1838 return response
1840 def _acquire_token_interactive_via_broker(
1841 self,
1842 scopes, # type: list[str]
1843 parent_window_handle, # type: int
1844 enable_msa_passthrough, # type: boolean
1845 claims, # type: str
1846 data, # type: dict
1847 on_before_launching_ui, # type: callable
1848 prompt=None,
1849 login_hint=None, # type: Optional[str]
1850 max_age=None,
1851 **kwargs):
1852 from .broker import _signin_interactively, _signin_silently, _acquire_token_silently
1853 if "welcome_template" in kwargs:
1854 logger.debug(kwargs["welcome_template"]) # Experimental
1855 authority = "https://{}/{}".format(
1856 self.authority.instance, self.authority.tenant)
1857 validate_authority = "no" if (
1858 self.authority._is_known_to_developer
1859 or self._instance_discovery is False) else None
1860 # Calls different broker methods to mimic the OIDC behaviors
1861 if login_hint and prompt != "select_account": # OIDC prompts when the user did not sign in
1862 accounts = self.get_accounts(username=login_hint)
1863 if len(accounts) == 1: # Unambiguously proceed with this account
1864 logger.debug("Calling broker._acquire_token_silently()")
1865 response = _acquire_token_silently( # When it works, it bypasses prompt
1866 authority,
1867 self.client_id,
1868 accounts[0]["local_account_id"],
1869 scopes,
1870 claims=claims,
1871 **data)
1872 if response and "error" not in response:
1873 return self._process_broker_response(response, scopes, data)
1874 # login_hint undecisive or not exists
1875 if prompt == "none" or not prompt: # Must/Can attempt _signin_silently()
1876 logger.debug("Calling broker._signin_silently()")
1877 response = _signin_silently( # Unlike OIDC, it doesn't honor login_hint
1878 authority, self.client_id, scopes,
1879 validateAuthority=validate_authority,
1880 claims=claims,
1881 max_age=max_age,
1882 enable_msa_pt=enable_msa_passthrough,
1883 **data)
1884 is_wrong_account = bool(
1885 # _signin_silently() only gets tokens for default account,
1886 # but this seems to have been fixed in PyMsalRuntime 0.11.2
1887 "access_token" in response and login_hint
1888 and response.get("id_token_claims", {}) != login_hint)
1889 wrong_account_error_message = (
1890 'prompt="none" will not work for login_hint="non-default-user"')
1891 if is_wrong_account:
1892 logger.debug(wrong_account_error_message)
1893 if prompt == "none":
1894 return self._process_broker_response( # It is either token or error
1895 response, scopes, data
1896 ) if not is_wrong_account else {
1897 "error": "broker_error",
1898 "error_description": wrong_account_error_message,
1899 }
1900 else:
1901 assert bool(prompt) is False
1902 from pymsalruntime import Response_Status
1903 recoverable_errors = frozenset([
1904 Response_Status.Status_AccountUnusable,
1905 Response_Status.Status_InteractionRequired,
1906 ])
1907 if is_wrong_account or "error" in response and response.get(
1908 "_broker_status") in recoverable_errors:
1909 pass # It will fall back to the _signin_interactively()
1910 else:
1911 return self._process_broker_response(response, scopes, data)
1913 logger.debug("Falls back to broker._signin_interactively()")
1914 on_before_launching_ui(ui="broker")
1915 response = _signin_interactively(
1916 authority, self.client_id, scopes,
1917 None if parent_window_handle is self.CONSOLE_WINDOW_HANDLE
1918 else parent_window_handle,
1919 validateAuthority=validate_authority,
1920 login_hint=login_hint,
1921 prompt=prompt,
1922 claims=claims,
1923 max_age=max_age,
1924 enable_msa_pt=enable_msa_passthrough,
1925 **data)
1926 return self._process_broker_response(response, scopes, data)
1928 def initiate_device_flow(self, scopes=None, **kwargs):
1929 """Initiate a Device Flow instance,
1930 which will be used in :func:`~acquire_token_by_device_flow`.
1932 :param list[str] scopes:
1933 Scopes requested to access a protected API (a resource).
1934 :return: A dict representing a newly created Device Flow object.
1936 - A successful response would contain "user_code" key, among others
1937 - an error response would contain some other readable key/value pairs.
1938 """
1939 correlation_id = msal.telemetry._get_new_correlation_id()
1940 flow = self.client.initiate_device_flow(
1941 scope=self._decorate_scope(scopes or []),
1942 headers={msal.telemetry.CLIENT_REQUEST_ID: correlation_id},
1943 **kwargs)
1944 flow[self.DEVICE_FLOW_CORRELATION_ID] = correlation_id
1945 return flow
1947 def acquire_token_by_device_flow(self, flow, claims_challenge=None, **kwargs):
1948 """Obtain token by a device flow object, with customizable polling effect.
1950 :param dict flow:
1951 A dict previously generated by :func:`~initiate_device_flow`.
1952 By default, this method's polling effect will block current thread.
1953 You can abort the polling loop at any time,
1954 by changing the value of the flow's "expires_at" key to 0.
1955 :param claims_challenge:
1956 The claims_challenge parameter requests specific claims requested by the resource provider
1957 in the form of a claims_challenge directive in the www-authenticate header to be
1958 returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
1959 It is a string of a JSON object which contains lists of claims being requested from these locations.
1961 :return: A dict representing the json response from AAD:
1963 - A successful response would contain "access_token" key,
1964 - an error response would contain "error" and usually "error_description".
1965 """
1966 telemetry_context = self._build_telemetry_context(
1967 self.ACQUIRE_TOKEN_BY_DEVICE_FLOW_ID,
1968 correlation_id=flow.get(self.DEVICE_FLOW_CORRELATION_ID))
1969 response = _clean_up(self.client.obtain_token_by_device_flow(
1970 flow,
1971 data=dict(
1972 kwargs.pop("data", {}),
1973 code=flow["device_code"], # 2018-10-4 Hack:
1974 # during transition period,
1975 # service seemingly need both device_code and code parameter.
1976 claims=_merge_claims_challenge_and_capabilities(
1977 self._client_capabilities, claims_challenge),
1978 ),
1979 headers=telemetry_context.generate_headers(),
1980 **kwargs))
1981 telemetry_context.update_telemetry(response)
1982 return response
1985class ConfidentialClientApplication(ClientApplication): # server-side web app
1987 def acquire_token_for_client(self, scopes, claims_challenge=None, **kwargs):
1988 """Acquires token for the current confidential client, not for an end user.
1990 :param list[str] scopes: (Required)
1991 Scopes requested to access a protected API (a resource).
1992 :param claims_challenge:
1993 The claims_challenge parameter requests specific claims requested by the resource provider
1994 in the form of a claims_challenge directive in the www-authenticate header to be
1995 returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
1996 It is a string of a JSON object which contains lists of claims being requested from these locations.
1998 :return: A dict representing the json response from AAD:
2000 - A successful response would contain "access_token" key,
2001 - an error response would contain "error" and usually "error_description".
2002 """
2003 # TBD: force_refresh behavior
2004 if self.authority.tenant.lower() in ["common", "organizations"]:
2005 warnings.warn(
2006 "Using /common or /organizations authority "
2007 "in acquire_token_for_client() is unreliable. "
2008 "Please use a specific tenant instead.", DeprecationWarning)
2009 self._validate_ssh_cert_input_data(kwargs.get("data", {}))
2010 telemetry_context = self._build_telemetry_context(
2011 self.ACQUIRE_TOKEN_FOR_CLIENT_ID)
2012 client = self._regional_client or self.client
2013 response = _clean_up(client.obtain_token_for_client(
2014 scope=scopes, # This grant flow requires no scope decoration
2015 headers=telemetry_context.generate_headers(),
2016 data=dict(
2017 kwargs.pop("data", {}),
2018 claims=_merge_claims_challenge_and_capabilities(
2019 self._client_capabilities, claims_challenge)),
2020 **kwargs))
2021 telemetry_context.update_telemetry(response)
2022 return response
2024 def acquire_token_on_behalf_of(self, user_assertion, scopes, claims_challenge=None, **kwargs):
2025 """Acquires token using on-behalf-of (OBO) flow.
2027 The current app is a middle-tier service which was called with a token
2028 representing an end user.
2029 The current app can use such token (a.k.a. a user assertion) to request
2030 another token to access downstream web API, on behalf of that user.
2031 See `detail docs here <https://docs.microsoft.com/en-us/azure/active-directory/develop/v2-oauth2-on-behalf-of-flow>`_ .
2033 The current middle-tier app has no user interaction to obtain consent.
2034 See how to gain consent upfront for your middle-tier app from this article.
2035 https://docs.microsoft.com/en-us/azure/active-directory/develop/v2-oauth2-on-behalf-of-flow#gaining-consent-for-the-middle-tier-application
2037 :param str user_assertion: The incoming token already received by this app
2038 :param list[str] scopes: Scopes required by downstream API (a resource).
2039 :param claims_challenge:
2040 The claims_challenge parameter requests specific claims requested by the resource provider
2041 in the form of a claims_challenge directive in the www-authenticate header to be
2042 returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
2043 It is a string of a JSON object which contains lists of claims being requested from these locations.
2045 :return: A dict representing the json response from AAD:
2047 - A successful response would contain "access_token" key,
2048 - an error response would contain "error" and usually "error_description".
2049 """
2050 telemetry_context = self._build_telemetry_context(
2051 self.ACQUIRE_TOKEN_ON_BEHALF_OF_ID)
2052 # The implementation is NOT based on Token Exchange
2053 # https://tools.ietf.org/html/draft-ietf-oauth-token-exchange-16
2054 response = _clean_up(self.client.obtain_token_by_assertion( # bases on assertion RFC 7521
2055 user_assertion,
2056 self.client.GRANT_TYPE_JWT, # IDTs and AAD ATs are all JWTs
2057 scope=self._decorate_scope(scopes), # Decoration is used for:
2058 # 1. Explicitly requesting an RT, without relying on AAD default
2059 # behavior, even though it currently still issues an RT.
2060 # 2. Requesting an IDT (which would otherwise be unavailable)
2061 # so that the calling app could use id_token_claims to implement
2062 # their own cache mapping, which is likely needed in web apps.
2063 data=dict(
2064 kwargs.pop("data", {}),
2065 requested_token_use="on_behalf_of",
2066 claims=_merge_claims_challenge_and_capabilities(
2067 self._client_capabilities, claims_challenge)),
2068 headers=telemetry_context.generate_headers(),
2069 # TBD: Expose a login_hint (or ccs_routing_hint) param for web app
2070 **kwargs))
2071 telemetry_context.update_telemetry(response)
2072 return response