Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.10/site-packages/msal/application.py: 15%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import functools
2import json
3import time
4import logging
5import platform
6import sys
7import warnings
8from threading import Lock
9from typing import Optional # Needed in Python 3.7 & 3.8
10from urllib.parse import urlparse
11import os
13from .oauth2cli import Client, JwtAssertionCreator
14from .oauth2cli.oidc import decode_part
15from .authority import (
16 Authority,
17 WORLD_WIDE,
18 WELL_KNOWN_AUTHORITY_HOSTS,
19 _get_instance_discovery_endpoint,
20 _get_instance_discovery_host,
21)
22from .mex import send_request as mex_send_request
23from .wstrust_request import send_request as wst_send_request
24from .wstrust_response import *
25from .token_cache import TokenCache, _get_username, _GRANT_TYPE_BROKER, _compute_ext_cache_key, _parse_claims_or_raise, _merge_claims
26import msal.telemetry
27from .region import _detect_region, _validate_region
28from .throttled_http_client import ThrottledHttpClient
29from .cloudshell import _is_running_in_cloud_shell
30from .sku import SKU, __version__
31from .oauth2cli.authcode import is_wsl
34logger = logging.getLogger(__name__)
35_AUTHORITY_TYPE_CLOUDSHELL = "CLOUDSHELL"
38def _validate_explicit_region(region, name):
39 if not _validate_region(region, source=name):
40 raise ValueError(
41 "Invalid {}={!r}. It must be a lowercase DNS label that starts "
42 "with a letter, contains only letters, digits, and internal "
43 "hyphens, and is at most 63 characters.".format(name, region))
44 return region
47def _init_broker(enable_pii_log): # Make it a function to allow mocking
48 from . import broker # Trigger Broker's initialization, lazily
49 if enable_pii_log:
50 broker._enable_pii_log()
52def extract_certs(public_cert_content):
53 # Parses raw public certificate file contents and returns a list of strings
54 # Usage: headers = {"x5c": extract_certs(open("my_cert.pem").read())}
55 public_certificates = re.findall(
56 r'-----BEGIN CERTIFICATE-----(?P<cert_value>[^-]+)-----END CERTIFICATE-----',
57 public_cert_content, re.I)
58 if public_certificates:
59 return [cert.strip() for cert in public_certificates]
60 # The public cert tags are not found in the input,
61 # let's make best effort to exclude a private key pem file.
62 if "PRIVATE KEY" in public_cert_content:
63 raise ValueError(
64 "We expect your public key but detect a private key instead")
65 return [public_cert_content.strip()]
68def _merge_claims_challenge_and_capabilities(capabilities, claims_challenge):
69 # Represent capabilities as {"access_token": {"xms_cc": {"values": capabilities}}}
70 # and then merge/add it into incoming claims
71 if not capabilities:
72 return claims_challenge
73 claims_dict = json.loads(claims_challenge) if claims_challenge else {}
74 for key in ["access_token"]: # We could add "id_token" if we'd decide to
75 claims_dict.setdefault(key, {}).update(xms_cc={"values": capabilities})
76 return json.dumps(claims_dict)
79def _stash_client_claims(forwarded_client_claims, data):
80 """Validate ``forwarded_client_claims`` and stash it into the request ``data``.
82 ``forwarded_client_claims`` carries *client-originated* claims supplied by
83 the caller. The raw value is stored in ``data`` (under the internal
84 ``client_claims`` key) so that it
85 (a) contributes to the extended cache key -- isolating cache entries by
86 claims value -- and (b) is stripped from the request body by the oauth2
87 layer (it reaches the wire only after being merged into the standard OAuth
88 ``claims`` parameter). ``data`` is mutated in place.
90 Unlike ``claims_challenge`` (server-issued, which bypasses the cache),
91 ``forwarded_client_claims`` tokens are cached and keyed on the claims value.
92 A no-op when ``forwarded_client_claims`` is ``None``.
93 """
94 if forwarded_client_claims is None:
95 return
96 if not isinstance(forwarded_client_claims, str):
97 raise ValueError(
98 "forwarded_client_claims must be a string, got {}".format(
99 type(forwarded_client_claims).__name__))
100 _parse_claims_or_raise(forwarded_client_claims) # Fail fast on malformed JSON
101 data["client_claims"] = forwarded_client_claims
104def _str2bytes(raw):
105 # A conversion based on duck-typing rather than six.text_type
106 try:
107 return raw.encode(encoding="utf-8")
108 except:
109 return raw
111def _extract_cert_and_thumbprints(cert):
112 # Cert concepts https://security.stackexchange.com/a/226758/125264
113 from cryptography.hazmat.primitives import hashes, serialization
114 cert_pem = cert.public_bytes( # Requires cryptography 1.0+
115 encoding=serialization.Encoding.PEM).decode()
116 x5c = [
117 '\n'.join(
118 cert_pem.splitlines()
119 [1:-1] # Strip the "--- header ---" and "--- footer ---"
120 )
121 ]
122 # https://cryptography.io/en/latest/x509/reference/#x-509-certificate-object - Requires cryptography 0.7+
123 sha256_thumbprint = cert.fingerprint(hashes.SHA256()).hex()
124 sha1_thumbprint = cert.fingerprint(hashes.SHA1()).hex() # CodeQL [SM02167] for legacy support such as ADFS
125 return sha256_thumbprint, sha1_thumbprint, x5c
127def _parse_pfx(pfx_path, passphrase_bytes):
128 # Cert concepts https://security.stackexchange.com/a/226758/125264
129 from cryptography.hazmat.primitives.serialization import pkcs12
130 with open(pfx_path, 'rb') as f:
131 private_key, cert, _ = pkcs12.load_key_and_certificates( # cryptography 2.5+
132 # https://cryptography.io/en/latest/hazmat/primitives/asymmetric/serialization/#cryptography.hazmat.primitives.serialization.pkcs12.load_key_and_certificates
133 f.read(), passphrase_bytes)
134 if not (private_key and cert):
135 raise ValueError("Your PFX file shall contain both private key and cert")
136 sha256_thumbprint, sha1_thumbprint, x5c = _extract_cert_and_thumbprints(cert)
137 return private_key, sha256_thumbprint, sha1_thumbprint, x5c
140def _load_private_key_from_pem_str(private_key_pem_str, passphrase_bytes):
141 from cryptography.hazmat.primitives import serialization
142 from cryptography.hazmat.backends import default_backend
143 return serialization.load_pem_private_key( # cryptography 0.6+
144 _str2bytes(private_key_pem_str),
145 passphrase_bytes,
146 backend=default_backend(), # It was a required param until 2020
147 )
150def _pii_less_home_account_id(home_account_id):
151 parts = home_account_id.split(".") # It could contain one or two parts
152 parts[0] = "********"
153 return ".".join(parts)
156def _clean_up(result):
157 if isinstance(result, dict):
158 if "_msalruntime_telemetry" in result or "_msal_python_telemetry" in result:
159 result["msal_telemetry"] = json.dumps({ # Telemetry as an opaque string
160 "msalruntime_telemetry": result.get("_msalruntime_telemetry"),
161 "msal_python_telemetry": result.get("_msal_python_telemetry"),
162 }, separators=(",", ":"))
163 return_value = {
164 k: result[k] for k in result
165 if k != "refresh_in" # MSAL handled refresh_in, customers need not
166 and not k.startswith('_') # Skim internal properties
167 }
168 if "refresh_in" in result: # To encourage proactive refresh
169 return_value["refresh_on"] = int(time.time() + result["refresh_in"])
170 return return_value
171 return result # It could be None
174def _preferred_browser():
175 """Register Edge and return a name suitable for subsequent webbrowser.get(...)
176 when appropriate. Otherwise return None.
177 """
178 # On Linux, only Edge will provide device-based Conditional Access support
179 if sys.platform != "linux": # On other platforms, we have no browser preference
180 return None
181 browser_path = "/usr/bin/microsoft-edge" # Use a full path owned by sys admin
182 # Note: /usr/bin/microsoft-edge, /usr/bin/microsoft-edge-stable, etc.
183 # are symlinks that point to the actual binaries which are found under
184 # /opt/microsoft/msedge/msedge or /opt/microsoft/msedge-beta/msedge.
185 # Either method can be used to detect an Edge installation.
186 user_has_no_preference = "BROWSER" not in os.environ
187 user_wont_mind_edge = "microsoft-edge" in os.environ.get("BROWSER", "") # Note:
188 # BROWSER could contain "microsoft-edge" or "/path/to/microsoft-edge".
189 # Python documentation (https://docs.python.org/3/library/webbrowser.html)
190 # does not document the name being implicitly register,
191 # so there is no public API to know whether the ENV VAR browser would work.
192 # Therefore, we would not bother examine the env var browser's type.
193 # We would just register our own Edge instance.
194 if (user_has_no_preference or user_wont_mind_edge) and os.path.exists(browser_path):
195 try:
196 import webbrowser # Lazy import. Some distro may not have this.
197 browser_name = "msal-edge" # Avoid popular name "microsoft-edge"
198 # otherwise `BROWSER="microsoft-edge"; webbrowser.get("microsoft-edge")`
199 # would return a GenericBrowser instance which won't work.
200 try:
201 registration_available = isinstance(
202 webbrowser.get(browser_name), webbrowser.BackgroundBrowser)
203 except webbrowser.Error:
204 registration_available = False
205 if not registration_available:
206 logger.debug("Register %s with %s", browser_name, browser_path)
207 # By registering our own browser instance with our own name,
208 # rather than populating a process-wide BROWSER enn var,
209 # this approach does not have side effect on non-MSAL code path.
210 webbrowser.register( # Even double-register happens to work fine
211 browser_name, None, webbrowser.BackgroundBrowser(browser_path))
212 return browser_name
213 except ImportError:
214 pass # We may still proceed
215 return None
217def _is_ssh_cert_or_pop_request(token_type, auth_scheme) -> bool:
218 return token_type == "ssh-cert" or token_type == "pop" or isinstance(auth_scheme, msal.auth_scheme.PopAuthScheme)
220class _ClientWithCcsRoutingInfo(Client):
222 def initiate_auth_code_flow(self, **kwargs):
223 if kwargs.get("login_hint"): # eSTS could have utilized this as-is, but nope
224 kwargs["X-AnchorMailbox"] = "UPN:%s" % kwargs["login_hint"]
225 return super(_ClientWithCcsRoutingInfo, self).initiate_auth_code_flow(
226 client_info=1, # To be used as CSS Routing info
227 **kwargs)
229 def obtain_token_by_auth_code_flow(
230 self, auth_code_flow, auth_response, **kwargs):
231 # Note: the obtain_token_by_browser() is also covered by this
232 assert isinstance(auth_code_flow, dict) and isinstance(auth_response, dict)
233 headers = kwargs.pop("headers", {})
234 client_info = json.loads(
235 decode_part(auth_response["client_info"])
236 ) if auth_response.get("client_info") else {}
237 if "uid" in client_info and "utid" in client_info:
238 # Note: The value of X-AnchorMailbox is also case-insensitive
239 headers["X-AnchorMailbox"] = "Oid:{uid}@{utid}".format(**client_info)
240 return super(_ClientWithCcsRoutingInfo, self).obtain_token_by_auth_code_flow(
241 auth_code_flow, auth_response, headers=headers, **kwargs)
243 def obtain_token_by_username_password(self, username, password, **kwargs):
244 headers = kwargs.pop("headers", {})
245 headers["X-AnchorMailbox"] = "upn:{}".format(username)
246 return super(_ClientWithCcsRoutingInfo, self).obtain_token_by_username_password(
247 username, password, headers=headers, **kwargs)
250def _msal_extension_check():
251 # Can't run this in module or class level otherwise you'll get circular import error
252 try:
253 from msal_extensions import __version__ as v
254 major, minor, _ = v.split(".", maxsplit=3)
255 if not (int(major) >= 1 and int(minor) >= 2):
256 warnings.warn(
257 "Please upgrade msal-extensions. "
258 "Only msal-extensions 1.2+ can work with msal 1.30+")
259 except ImportError:
260 pass # The optional msal_extensions is not installed. Business as usual.
261 except ValueError:
262 logger.exception(f"msal_extensions version {v} not in major.minor.patch format")
263 except:
264 logger.exception(
265 "Unable to import msal_extensions during an optional check. "
266 "This exception can be safely ignored."
267 )
270class ClientApplication(object):
271 """You do not usually directly use this class. Use its subclasses instead:
272 :class:`PublicClientApplication` and :class:`ConfidentialClientApplication`.
273 """
274 ACQUIRE_TOKEN_SILENT_ID = "84"
275 ACQUIRE_TOKEN_BY_REFRESH_TOKEN = "85"
276 ACQUIRE_TOKEN_BY_USERNAME_PASSWORD_ID = "301"
277 ACQUIRE_TOKEN_ON_BEHALF_OF_ID = "523"
278 ACQUIRE_TOKEN_BY_DEVICE_FLOW_ID = "622"
279 ACQUIRE_TOKEN_FOR_CLIENT_ID = "730"
280 ACQUIRE_TOKEN_BY_AUTHORIZATION_CODE_ID = "832"
281 ACQUIRE_TOKEN_INTERACTIVE = "169"
282 ACQUIRE_TOKEN_BY_USER_FIC_ID = "950"
283 GET_ACCOUNTS_ID = "902"
284 REMOVE_ACCOUNT_ID = "903"
286 ATTEMPT_REGION_DISCOVERY = True # "TryAutoDetect"
287 DISABLE_MSAL_FORCE_REGION = False # Used in azure_region to disable MSAL_FORCE_REGION behavior
288 _TOKEN_SOURCE = "token_source"
289 _TOKEN_SOURCE_IDP = "identity_provider"
290 _TOKEN_SOURCE_CACHE = "cache"
291 _TOKEN_SOURCE_BROKER = "broker"
293 _enable_broker = False
294 _AUTH_SCHEME_UNSUPPORTED = (
295 "auth_scheme is currently only available from broker. "
296 "You can enable broker by following these instructions. "
297 "https://msal-python.readthedocs.io/en/latest/#publicclientapplication")
299 def __init__(
300 self, client_id,
301 client_credential=None, authority=None, validate_authority=True,
302 token_cache=None,
303 http_client=None,
304 verify=True, proxies=None, timeout=None,
305 client_claims=None, app_name=None, app_version=None,
306 client_capabilities=None,
307 azure_region=None, # Note: We choose to add this param in this base class,
308 # despite it is currently only needed by ConfidentialClientApplication.
309 # This way, it holds the same positional param place for PCA,
310 # when we would eventually want to add this feature to PCA in future.
311 exclude_scopes=None,
312 http_cache=None,
313 instance_discovery=None,
314 allow_broker=None,
315 enable_pii_log=None,
316 oidc_authority=None,
317 ):
318 """Create an instance of application.
320 :param str client_id: Your app has a client_id after you register it on Microsoft Entra admin center.
322 :param client_credential:
323 For :class:`PublicClientApplication`, you use `None` here.
325 For :class:`ConfidentialClientApplication`,
326 it supports many different input formats for different scenarios.
328 .. admonition:: Support using a client secret.
330 Just feed in a string, such as ``"your client secret"``.
332 .. admonition:: Support using a certificate in X.509 (.pem) format
334 Deprecated because it uses SHA-1 thumbprint,
335 unless you are still using ADFS which supports SHA-1 thumbprint only.
336 Please use the .pfx option documented later in this page.
338 Feed in a dict in this form::
340 {
341 "private_key": "...-----BEGIN PRIVATE KEY-----... in PEM format",
342 "thumbprint": "An SHA-1 thumbprint such as A1B2C3D4E5F6..."
343 "Changed in version 1.35.0, if thumbprint is absent"
344 "and a public_certificate is present, MSAL will"
345 "automatically calculate an SHA-256 thumbprint instead.",
346 "passphrase": "Needed if the private_key is encrypted (Added in version 1.6.0)",
347 "public_certificate": "...-----BEGIN CERTIFICATE-----...", # Needed if you use Subject Name/Issuer auth. Added in version 0.5.0.
348 }
350 MSAL Python requires a "private_key" in PEM format.
351 If your cert is in PKCS12 (.pfx) format,
352 you can convert it to X.509 (.pem) format,
353 by ``openssl pkcs12 -in file.pfx -out file.pem -nodes``.
355 The thumbprint is available in your app's registration in Azure Portal.
356 Alternatively, you can `calculate the thumbprint <https://github.com/Azure/azure-sdk-for-python/blob/07d10639d7e47f4852eaeb74aef5d569db499d6e/sdk/identity/azure-identity/azure/identity/_credentials/certificate.py#L94-L97>`_.
358 ``public_certificate`` (optional) is public key certificate
359 which will be sent through 'x5c' JWT header.
360 This is useful when you use `Subject Name/Issuer Authentication
361 <https://github.com/AzureAD/microsoft-authentication-library-for-python/issues/60>`_
362 which is an approach to allow easier certificate rotation.
363 Per `specs <https://tools.ietf.org/html/rfc7515#section-4.1.6>`_,
364 "the certificate containing
365 the public key corresponding to the key used to digitally sign the
366 JWS MUST be the first certificate. This MAY be followed by
367 additional certificates, with each subsequent certificate being the
368 one used to certify the previous one."
369 However, your certificate's issuer may use a different order.
370 So, if your attempt ends up with an error AADSTS700027 -
371 "The provided signature value did not match the expected signature value",
372 you may try use only the leaf cert (in PEM/str format) instead.
374 .. admonition:: Supporting raw assertion obtained from elsewhere
376 *Added in version 1.13.0*:
377 It can also be a completely pre-signed assertion that you've assembled yourself.
378 Simply pass a container containing only the key "client_assertion", like this::
380 {
381 "client_assertion": "...a JWT with claims aud, exp, iss, jti, nbf, and sub..."
382 }
384 .. note::
386 A pre-signed JWT string has a fixed expiration. Long-running
387 confidential client applications (for example, workloads using
388 AKS workload identity federation, or any other dynamic
389 credential source) should instead pass a **callable** which
390 MSAL will invoke on demand to obtain a fresh assertion::
392 def get_client_assertion():
393 # e.g. read the projected service-account token from disk
394 with open("/var/run/secrets/azure/tokens/azure-identity-token") as f:
395 return f.read()
397 app = ConfidentialClientApplication(
398 "client_id",
399 client_credential={"client_assertion": get_client_assertion},
400 ...,
401 )
403 The callable is only invoked when MSAL needs to send a token
404 request on the wire (the in-memory token cache transparently
405 avoids unnecessary calls).
407 If your callback is itself expensive (for example it calls
408 out to a key vault), wrap it in :class:`msal.AutoRefresher`
409 to memoize the assertion for its lifetime::
411 from msal import AutoRefresher
412 smart_callback = AutoRefresher(get_client_assertion, expires_in=3600)
413 app = ConfidentialClientApplication(
414 "client_id",
415 client_credential={"client_assertion": smart_callback},
416 ...,
417 )
419 Passing a plain ``str`` / ``bytes`` ``client_assertion`` is
420 still supported for backward compatibility but is discouraged
421 because the assertion will eventually expire.
423 .. admonition:: Supporting reading client certificates from PFX files
425 This usage will automatically use SHA-256 thumbprint of the certificate.
427 *Added in version 1.29.0*:
428 Feed in a dictionary containing the path to a PFX file::
430 {
431 "private_key_pfx_path": "/path/to/your.pfx", # Added in version 1.29.0
432 "public_certificate": True, # Only needed if you use Subject Name/Issuer auth. Added in version 1.30.0
433 "passphrase": "Passphrase if the private_key is encrypted (Optional)",
434 }
436 The following command will generate a .pfx file from your .key and .pem file::
438 openssl pkcs12 -export -out certificate.pfx -inkey privateKey.key -in certificate.pem
440 `Subject Name/Issuer Auth
441 <https://github.com/AzureAD/microsoft-authentication-library-for-python/issues/60>`_
442 is an approach to allow easier certificate rotation.
443 If your .pfx file contains both the private key and public cert,
444 you can opt in for Subject Name/Issuer Auth by setting "public_certificate" to ``True``.
446 :type client_credential: Union[dict, str, None]
448 :param dict client_claims:
449 *Added in version 0.5.0*:
450 It is a dictionary of extra claims that would be signed by
451 by this :class:`ConfidentialClientApplication` 's private key.
452 For example, you can use {"client_ip": "x.x.x.x"}.
453 You may also override any of the following default claims::
455 {
456 "aud": the_token_endpoint,
457 "iss": self.client_id,
458 "sub": same_as_issuer,
459 "exp": now + 10_min,
460 "iat": now,
461 "jti": a_random_uuid
462 }
464 .. note::
466 This *constructor* ``client_claims`` (a ``dict`` signed into the
467 client-assertion JWT) is distinct from the per-request
468 ``forwarded_client_claims`` parameter (a JSON string of
469 client-originated claims forwarded in the token request) accepted
470 by the token-acquisition methods.
472 :param str authority:
473 A URL that identifies a token authority. It should be of the format
474 ``https://login.microsoftonline.com/your_tenant``
475 By default, we will use ``https://login.microsoftonline.com/common``
477 *Changed in version 1.17*: you can also use predefined constant
478 and a builder like this::
480 from msal.authority import (
481 AuthorityBuilder,
482 AZURE_US_GOVERNMENT, AZURE_CHINA, AZURE_PUBLIC)
483 my_authority = AuthorityBuilder(AZURE_PUBLIC, "contoso.onmicrosoft.com")
484 # Now you get an equivalent of
485 # "https://login.microsoftonline.com/contoso.onmicrosoft.com"
487 # You can feed such an authority to msal's ClientApplication
488 from msal import PublicClientApplication
489 app = PublicClientApplication("my_client_id", authority=my_authority, ...)
491 :param bool validate_authority: (optional) Turns authority validation
492 on or off. This parameter default to true.
493 :param TokenCache token_cache:
494 Sets the token cache used by this ClientApplication instance.
495 By default, an in-memory cache will be created and used.
496 :param http_client: (optional)
497 Your implementation of abstract class HttpClient <msal.oauth2cli.http.http_client>
498 Defaults to a requests session instance.
499 Since MSAL 1.11.0, the default session would be configured
500 to attempt one retry on connection error.
501 If you are providing your own http_client,
502 it will be your http_client's duty to decide whether to perform retry.
504 :param verify: (optional)
505 It will be passed to the
506 `verify parameter in the underlying requests library
507 <http://docs.python-requests.org/en/v2.9.1/user/advanced/#ssl-cert-verification>`_
508 This does not apply if you have chosen to pass your own Http client
509 :param proxies: (optional)
510 It will be passed to the
511 `proxies parameter in the underlying requests library
512 <http://docs.python-requests.org/en/v2.9.1/user/advanced/#proxies>`_
513 This does not apply if you have chosen to pass your own Http client
514 :param timeout: (optional)
515 It will be passed to the
516 `timeout parameter in the underlying requests library
517 <http://docs.python-requests.org/en/v2.9.1/user/advanced/#timeouts>`_
518 This does not apply if you have chosen to pass your own Http client
519 :param app_name: (optional)
520 You can provide your application name for Microsoft telemetry purposes.
521 Default value is None, means it will not be passed to Microsoft.
522 :param app_version: (optional)
523 You can provide your application version for Microsoft telemetry purposes.
524 Default value is None, means it will not be passed to Microsoft.
525 :param list[str] client_capabilities: (optional)
526 Allows configuration of one or more client capabilities, e.g. ["CP1"].
528 Client capability is meant to inform the Microsoft identity platform
529 (STS) what this client is capable for,
530 so STS can decide to turn on certain features.
531 For example, if client is capable to handle *claims challenge*,
532 STS may issue
533 `Continuous Access Evaluation (CAE) <https://learn.microsoft.com/entra/identity/conditional-access/concept-continuous-access-evaluation>`_
534 access tokens to resources,
535 knowing that when the resource emits a *claims challenge*
536 the client will be able to handle those challenges.
538 Implementation details:
539 Client capability is implemented using "claims" parameter on the wire,
540 for now.
541 MSAL will combine them into
542 `claims parameter <https://openid.net/specs/openid-connect-core-1_0-final.html#ClaimsParameter>`_
543 which you will later provide via one of the acquire-token request.
545 :param str azure_region: (optional)
546 Instructs MSAL to use the Entra regional token service. This legacy feature is only available to
547 first-party applications. Only ``acquire_token_for_client()`` is supported.
549 Supports 4 values:
551 1. ``azure_region=None`` - This default value means no region is configured.
552 MSAL will use the region defined in env var ``MSAL_FORCE_REGION``.
553 2. ``azure_region="some_region"`` - meaning the specified region is used.
554 3. ``azure_region=True`` - meaning
555 MSAL will try to auto-detect the region. This is not recommended.
556 4. ``azure_region=False`` - meaning MSAL will use no region.
558 .. note::
559 Region auto-discovery has been tested on VMs and on Azure Functions. It is unreliable.
560 Applications using this option should configure a short timeout.
562 For more details and for the values of the region string
563 see https://learn.microsoft.com/entra/msal/dotnet/resources/region-discovery-troubleshooting
565 New in version 1.12.0.
567 :param list[str] exclude_scopes: (optional)
568 Historically MSAL hardcodes `offline_access` scope,
569 which would allow your app to have prolonged access to user's data.
570 If that is unnecessary or undesirable for your app,
571 now you can use this parameter to supply an exclusion list of scopes,
572 such as ``exclude_scopes = ["offline_access"]``.
574 :param dict http_cache:
575 MSAL has long been caching tokens in the ``token_cache``.
576 Recently, MSAL also introduced a concept of ``http_cache``,
577 by automatically caching some finite amount of non-token http responses,
578 so that *long-lived*
579 ``PublicClientApplication`` and ``ConfidentialClientApplication``
580 would be more performant and responsive in some situations.
582 This ``http_cache`` parameter accepts any dict-like object.
583 If not provided, MSAL will use an in-memory dict.
585 If your app is a command-line app (CLI),
586 you would want to persist your http_cache across different CLI runs.
587 The persisted file's format may change due to, but not limited to,
588 `unstable protocol <https://docs.python.org/3/library/pickle.html#data-stream-format>`_,
589 so your implementation shall tolerate unexpected loading errors.
590 The following recipe shows a way to do so::
592 # Just add the following lines at the beginning of your CLI script
593 import sys, atexit, pickle, logging
594 http_cache_filename = sys.argv[0] + ".http_cache"
595 try:
596 with open(http_cache_filename, "rb") as f:
597 persisted_http_cache = pickle.load(f) # Take a snapshot
598 except (
599 FileNotFoundError, # Or IOError in Python 2
600 pickle.UnpicklingError, # A corrupted http cache file
601 AttributeError, # Cache created by a different version of MSAL
602 ):
603 persisted_http_cache = {} # Recover by starting afresh
604 except: # Unexpected exceptions
605 logging.exception("You may want to debug this")
606 persisted_http_cache = {} # Recover by starting afresh
607 atexit.register(lambda: pickle.dump(
608 # When exit, flush it back to the file.
609 # It may occasionally overwrite another process's concurrent write,
610 # but that is fine. Subsequent runs will reach eventual consistency.
611 persisted_http_cache, open(http_cache_file, "wb")))
613 # And then you can implement your app as you normally would
614 app = msal.PublicClientApplication(
615 "your_client_id",
616 ...,
617 http_cache=persisted_http_cache, # Utilize persisted_http_cache
618 ...,
619 #token_cache=..., # You may combine the old token_cache trick
620 # Please refer to token_cache recipe at
621 # https://msal-python.readthedocs.io/en/latest/#msal.SerializableTokenCache
622 )
623 app.acquire_token_interactive(["your", "scope"], ...)
625 Content inside ``http_cache`` are cheap to obtain.
626 There is no need to share them among different apps.
628 Content inside ``http_cache`` will contain no tokens nor
629 Personally Identifiable Information (PII). Encryption is unnecessary.
631 New in version 1.16.0.
633 :param boolean instance_discovery:
634 Historically, MSAL would connect to a central endpoint located at
635 ``https://login.microsoftonline.com`` to acquire some metadata,
636 especially when using an unfamiliar authority.
637 This behavior is known as Instance Discovery.
639 This parameter defaults to None, which enables the Instance Discovery.
641 If you know some authorities which you allow MSAL to operate with as-is,
642 without involving any Instance Discovery, the recommended pattern is::
644 known_authorities = frozenset([ # Treat your known authorities as const
645 "https://contoso.com/adfs", "https://login.azs/foo"])
646 ...
647 authority = "https://contoso.com/adfs" # Assuming your app will use this
648 app1 = PublicClientApplication(
649 "client_id",
650 authority=authority,
651 # Conditionally disable Instance Discovery for known authorities
652 instance_discovery=authority not in known_authorities,
653 )
655 If you do not know some authorities beforehand,
656 yet still want MSAL to accept any authority that you will provide,
657 you can use a ``False`` to unconditionally disable Instance Discovery.
659 New in version 1.19.0.
661 :param boolean allow_broker:
662 Deprecated. Please use ``enable_broker_on_windows`` instead.
664 :param boolean enable_pii_log:
665 When enabled, logs may include PII (Personal Identifiable Information).
666 This can be useful in troubleshooting broker behaviors.
667 The default behavior is False.
669 New in version 1.24.0.
671 :param str oidc_authority:
672 *Added in version 1.28.0*:
673 It is a URL that identifies an OpenID Connect (OIDC) authority of
674 the format ``https://contoso.com/tenant``.
675 MSAL will append ".well-known/openid-configuration" to the authority
676 and retrieve the OIDC metadata from there, to figure out the endpoints.
678 Note: Broker will NOT be used for OIDC authority.
679 """
680 self.client_id = client_id
681 self.client_credential = client_credential
682 self.client_claims = client_claims
683 self._client_capabilities = client_capabilities
684 self._instance_discovery = instance_discovery
685 if isinstance(azure_region, str):
686 _validate_explicit_region(azure_region, "azure_region")
687 force_region = None
688 if azure_region is None and client_credential:
689 force_region = os.getenv("MSAL_FORCE_REGION")
690 if force_region is not None:
691 _validate_explicit_region(force_region, "MSAL_FORCE_REGION")
693 if exclude_scopes and not isinstance(exclude_scopes, list):
694 raise ValueError(
695 "Invalid exclude_scopes={}. It need to be a list of strings.".format(
696 repr(exclude_scopes)))
697 self._exclude_scopes = frozenset(exclude_scopes or [])
698 if "openid" in self._exclude_scopes:
699 raise ValueError(
700 'Invalid exclude_scopes={}. You can not opt out "openid" scope'.format(
701 repr(exclude_scopes)))
703 if http_client:
704 self.http_client = http_client
705 else:
706 import requests # Lazy load
708 self.http_client = requests.Session()
709 self.http_client.verify = verify
710 self.http_client.proxies = proxies
711 # Requests, does not support session - wide timeout
712 # But you can patch that (https://github.com/psf/requests/issues/3341):
713 self.http_client.request = functools.partial(
714 self.http_client.request, timeout=timeout)
716 # Enable a minimal retry. Better than nothing.
717 # https://github.com/psf/requests/blob/v2.25.1/requests/adapters.py#L94-L108
718 a = requests.adapters.HTTPAdapter(max_retries=1)
719 self.http_client.mount("http://", a)
720 self.http_client.mount("https://", a)
721 self.http_client = ThrottledHttpClient(
722 self.http_client,
723 http_cache=http_cache,
724 default_throttle_time=60
725 # The default value 60 was recommended mainly for PCA at the end of
726 # https://identitydivision.visualstudio.com/devex/_git/AuthLibrariesApiReview?version=GBdev&path=%2FService%20protection%2FIntial%20set%20of%20protection%20measures.md&_a=preview
727 if isinstance(self, PublicClientApplication) else 5,
728 )
730 self.app_name = app_name
731 self.app_version = app_version
733 # Here the self.authority will not be the same type as authority in input
734 if oidc_authority and authority:
735 raise ValueError("You can not provide both authority and oidc_authority")
736 if isinstance(authority, str) and urlparse(authority).path.startswith(
737 "/dstsv2"): # dSTS authority's path always starts with "/dstsv2"
738 oidc_authority = authority # So we treat it as if an oidc_authority
739 authority_to_use = authority or "https://{}/common/".format(WORLD_WIDE)
740 try:
741 self.authority = Authority(
742 authority_to_use,
743 self.http_client,
744 validate_authority=validate_authority,
745 instance_discovery=self._instance_discovery,
746 oidc_authority_url=oidc_authority,
747 )
748 except OSError:
749 authority_host = urlparse(str(authority_to_use)).hostname
750 if not (
751 validate_authority
752 and not oidc_authority
753 and authority_host in WELL_KNOWN_AUTHORITY_HOSTS
754 and (
755 azure_region
756 or force_region is not None
757 )):
758 raise
759 self.authority = Authority(
760 authority_to_use,
761 self.http_client,
762 validate_authority=True,
763 instance_discovery=False,
764 )
766 self._decide_broker(allow_broker, enable_pii_log)
767 self.token_cache = token_cache or TokenCache()
768 self._region_configured = azure_region
769 self._region_detected = None
770 self.client, self._regional_client = self._build_client(
771 client_credential, self.authority)
772 # Warn if using a static string/bytes client_assertion (discouraged for long-running apps)
773 if isinstance(client_credential, dict) and isinstance(
774 client_credential.get("client_assertion"), (str, bytes)):
775 warnings.warn(
776 "Passing a static string/bytes 'client_assertion' is "
777 "discouraged because the JWT will eventually expire. "
778 "Pass a no-arg callable instead (optionally wrapped in "
779 "msal.AutoRefresher) so MSAL can obtain a fresh "
780 "assertion on demand. "
781 "See https://github.com/AzureAD/microsoft-authentication-library-for-python/issues/746",
782 DeprecationWarning, stacklevel=2)
784 self.authority_groups = {}
785 self._telemetry_buffer = {}
786 self._telemetry_lock = Lock()
787 _msal_extension_check()
790 def _decide_broker(self, allow_broker, enable_pii_log):
791 is_confidential_app = self.client_credential or isinstance(
792 self, ConfidentialClientApplication)
793 if is_confidential_app and allow_broker:
794 raise ValueError("allow_broker=True is only supported in PublicClientApplication")
795 # Historically, we chose to support ClientApplication("client_id", allow_broker=True)
796 if allow_broker:
797 warnings.warn(
798 "allow_broker is deprecated. "
799 "Please use PublicClientApplication(..., "
800 "enable_broker_on_windows=True, "
801 # No need to mention non-Windows platforms, because allow_broker is only for Windows
802 "...)",
803 DeprecationWarning)
804 opted_in_for_broker = (
805 self._enable_broker # True means Opted-in from PCA
806 or (
807 # When we started the broker project on Windows platform,
808 # the allow_broker was meant to be cross-platform. Now we realize
809 # that other platforms have different redirect_uri requirements,
810 # so the old allow_broker is deprecated and will only for Windows.
811 allow_broker and sys.platform == "win32")
812 )
813 self._enable_broker = ( # This same variable will also store the state
814 opted_in_for_broker
815 and not is_confidential_app
816 and not self.authority.is_adfs
817 and not self.authority._is_b2c
818 )
819 if (
820 self._enable_broker
821 and sys.platform == "darwin"
822 and platform.machine() != "arm64"
823 ):
824 # Broker on macOS is supported only on Apple Silicon (arm64).
825 # Anything else on darwin -- Intel Macs, and an x86_64 Python
826 # running under Rosetta -- is excluded by product policy,
827 # regardless of whether a broker is installed on the device.
828 # This is an allowlist so that an unrecognized architecture
829 # errs on the side of not using the broker.
830 self._enable_broker = False
831 logger.warning(
832 "Broker on macOS is supported only on Apple Silicon (arm64). "
833 "We will fallback to non-broker.")
834 if self._enable_broker:
835 try:
836 _init_broker(enable_pii_log)
837 except RuntimeError:
838 self._enable_broker = False
839 logger.warning( # It is common on Mac and Linux where broker is not built-in
840 "Broker is unavailable on this platform. "
841 "We will fallback to non-broker.")
842 logger.debug("Broker enabled? %s", self._enable_broker)
844 def is_pop_supported(self):
845 """Returns True if this client supports Proof-of-Possession Access Token."""
846 return self._enable_broker and sys.platform in ("win32", "darwin")
848 def _decorate_scope(
849 self, scopes,
850 reserved_scope=frozenset(['openid', 'profile', 'offline_access'])):
851 if not isinstance(scopes, (list, set, tuple)):
852 raise ValueError("The input scopes should be a list, tuple, or set")
853 scope_set = set(scopes) # Input scopes is typically a list. Copy it to a set.
854 if scope_set & reserved_scope:
855 # These scopes are reserved for the API to provide good experience.
856 # We could make the developer pass these and then if they do they will
857 # come back asking why they don't see refresh token or user information.
858 raise ValueError(
859 """You cannot use any scope value that is reserved.
860Your input: {}
861The reserved list: {}""".format(list(scope_set), list(reserved_scope)))
862 raise ValueError(
863 "You cannot use any scope value that is in this reserved list: {}".format(
864 list(reserved_scope)))
866 # client_id can also be used as a scope in B2C
867 decorated = scope_set | reserved_scope
868 decorated -= self._exclude_scopes
869 return list(decorated)
871 def _build_telemetry_context(
872 self, api_id, correlation_id=None, refresh_reason=None):
873 return msal.telemetry._TelemetryContext(
874 self._telemetry_buffer, self._telemetry_lock, api_id,
875 correlation_id=correlation_id, refresh_reason=refresh_reason)
877 def _get_regional_authority(self, central_authority) -> Optional[Authority]:
878 if self._region_configured is False: # User opts out of ESTS-R
879 return None # Short circuit to completely bypass region detection
880 if self._region_configured is None: # User did not make an ESTS-R choice
881 force_region = os.getenv("MSAL_FORCE_REGION")
882 self._region_configured = (
883 _validate_explicit_region(force_region, "MSAL_FORCE_REGION")
884 if force_region is not None else None)
885 self._region_detected = self._region_detected or _detect_region(
886 self.http_client if self._region_configured is not None else None)
887 if (self._region_configured != self.ATTEMPT_REGION_DISCOVERY
888 and self._region_configured != self._region_detected):
889 logger.warning('Region configured ({}) != region detected ({})'.format(
890 repr(self._region_configured), repr(self._region_detected)))
891 region_to_use = (
892 self._region_detected
893 if self._region_configured == self.ATTEMPT_REGION_DISCOVERY
894 else self._region_configured) # It will retain the None i.e. opted out
895 if isinstance(region_to_use, str):
896 region_to_use = _validate_region(
897 region_to_use, source="azure_region parameter")
898 logger.debug('Region to be used: {}'.format(repr(region_to_use)))
899 if region_to_use:
900 regional_host = ("{}.login.microsoft.com".format(region_to_use)
901 if central_authority.instance in (
902 # The list came from point 3 of the algorithm section in this internal doc
903 # https://identitydivision.visualstudio.com/DevEx/_git/AuthLibrariesApiReview?path=/PinAuthToRegion/AAD%20SDK%20Proposal%20to%20Pin%20Auth%20to%20region.md&anchor=algorithm&_a=preview
904 "login.microsoftonline.com",
905 "login.microsoft.com",
906 "login.windows.net",
907 "sts.windows.net",
908 )
909 else "{}.{}".format(region_to_use, central_authority.instance))
910 return Authority( # The central_authority has already been validated
911 "https://{}/{}".format(regional_host, central_authority.tenant),
912 self.http_client,
913 instance_discovery=False,
914 )
915 return None
917 def _build_client(self, client_credential, authority, skip_regional_client=False):
918 client_assertion = None
919 client_assertion_type = None
920 default_headers = {
921 "x-client-sku": SKU, "x-client-ver": __version__,
922 "x-client-os": sys.platform,
923 "x-ms-lib-capability": "retry-after, h429",
924 }
925 if self.app_name:
926 default_headers['x-app-name'] = self.app_name
927 if self.app_version:
928 default_headers['x-app-ver'] = self.app_version
929 default_body = {"client_info": 1}
930 if isinstance(client_credential, dict):
931 client_assertion_type = Client.CLIENT_ASSERTION_TYPE_JWT
932 # Use client_credential.get("...") rather than "..." in client_credential
933 # so that we can ignore an empty string came from an empty ENV VAR.
934 if client_credential.get("client_assertion"):
935 client_assertion = client_credential['client_assertion']
936 else:
937 headers = {}
938 sha1_thumbprint = sha256_thumbprint = None
939 passphrase_bytes = _str2bytes(
940 client_credential["passphrase"]
941 ) if client_credential.get("passphrase") else None
942 if client_credential.get("private_key_pfx_path"):
943 private_key, sha256_thumbprint, sha1_thumbprint, x5c = _parse_pfx(
944 client_credential["private_key_pfx_path"],
945 passphrase_bytes)
946 if client_credential.get("public_certificate") is True and x5c:
947 headers["x5c"] = x5c
948 elif client_credential.get("private_key"): # PEM blob
949 private_key = ( # handles both encrypted and unencrypted
950 _load_private_key_from_pem_str(
951 client_credential['private_key'], passphrase_bytes)
952 if passphrase_bytes
953 else client_credential['private_key']
954 )
956 # Determine thumbprints based on what's provided
957 if client_credential.get("thumbprint"):
958 # User provided a thumbprint - use it as SHA-1 (legacy/manual approach)
959 sha1_thumbprint = client_credential["thumbprint"]
960 sha256_thumbprint = None
961 elif isinstance(client_credential.get('public_certificate'), str):
962 # No thumbprint provided, but we have a certificate to calculate thumbprints
963 from cryptography import x509
964 cert = x509.load_pem_x509_certificate(
965 _str2bytes(client_credential['public_certificate']))
966 sha256_thumbprint, sha1_thumbprint, headers["x5c"] = (
967 _extract_cert_and_thumbprints(cert))
968 else:
969 raise ValueError(
970 "You must provide either 'thumbprint' or 'public_certificate' "
971 "from which the thumbprint can be calculated.")
972 else:
973 raise ValueError(
974 "client_credential needs to follow this format "
975 "https://msal-python.readthedocs.io/en/latest/#msal.ClientApplication.params.client_credential")
976 if ("x5c" not in headers # So the .pfx file contains no certificate
977 and isinstance(client_credential.get('public_certificate'), str)
978 ): # Then we treat the public_certificate value as PEM content
979 headers["x5c"] = extract_certs(client_credential['public_certificate'])
980 if sha256_thumbprint and not authority.is_adfs:
981 assertion_params = {
982 "algorithm": "PS256", "sha256_thumbprint": sha256_thumbprint,
983 }
984 else: # Fall back
985 if not sha1_thumbprint:
986 raise ValueError("You shall provide a thumbprint in SHA1.")
987 assertion_params = {
988 "algorithm": "RS256", "sha1_thumbprint": sha1_thumbprint,
989 }
990 assertion = JwtAssertionCreator(
991 private_key, headers=headers, **assertion_params)
992 client_assertion = assertion.create_regenerative_assertion(
993 audience=authority.token_endpoint, issuer=self.client_id,
994 additional_claims=self.client_claims or {})
995 else:
996 default_body['client_secret'] = client_credential
997 central_configuration = {
998 "authorization_endpoint": authority.authorization_endpoint,
999 "token_endpoint": authority.token_endpoint,
1000 "device_authorization_endpoint": authority.device_authorization_endpoint,
1001 }
1002 central_client = _ClientWithCcsRoutingInfo(
1003 central_configuration,
1004 self.client_id,
1005 http_client=self.http_client,
1006 default_headers=default_headers,
1007 default_body=default_body,
1008 client_assertion=client_assertion,
1009 client_assertion_type=client_assertion_type,
1010 on_obtaining_tokens=lambda event: self.token_cache.add(dict(
1011 event, environment=authority.instance)),
1012 on_removing_rt=self.token_cache.remove_rt,
1013 on_updating_rt=self.token_cache.update_rt)
1015 regional_client = None
1016 if (client_credential # Currently regional endpoint only serves some CCA flows
1017 and not skip_regional_client):
1018 regional_authority = self._get_regional_authority(authority)
1019 if regional_authority:
1020 regional_configuration = {
1021 "authorization_endpoint": regional_authority.authorization_endpoint,
1022 "token_endpoint": regional_authority.token_endpoint,
1023 "device_authorization_endpoint":
1024 regional_authority.device_authorization_endpoint,
1025 }
1026 regional_client = _ClientWithCcsRoutingInfo(
1027 regional_configuration,
1028 self.client_id,
1029 http_client=self.http_client,
1030 default_headers=default_headers,
1031 default_body=default_body,
1032 client_assertion=client_assertion,
1033 client_assertion_type=client_assertion_type,
1034 on_obtaining_tokens=lambda event: self.token_cache.add(dict(
1035 event, environment=authority.instance)),
1036 on_removing_rt=self.token_cache.remove_rt,
1037 on_updating_rt=self.token_cache.update_rt)
1038 return central_client, regional_client
1040 def initiate_auth_code_flow(
1041 self,
1042 scopes, # type: list[str]
1043 redirect_uri=None,
1044 state=None, # Recommended by OAuth2 for CSRF protection
1045 prompt=None,
1046 login_hint=None, # type: Optional[str]
1047 domain_hint=None, # type: Optional[str]
1048 claims_challenge=None,
1049 max_age=None,
1050 response_mode=None, # type: Optional[str]
1051 ):
1052 """Initiate an auth code flow.
1054 Later when the response reaches your redirect_uri,
1055 you can use :func:`~acquire_token_by_auth_code_flow()`
1056 to complete the authentication/authorization.
1058 :param list scopes:
1059 It is a list of case-sensitive strings.
1060 :param str redirect_uri:
1061 Optional. If not specified, server will use the pre-registered one.
1062 :param str state:
1063 An opaque value used by the client to
1064 maintain state between the request and callback.
1065 If absent, this library will automatically generate one internally.
1066 :param str prompt:
1067 By default, no prompt value will be sent, not even string ``"none"``.
1068 You will have to specify a value explicitly.
1069 Its valid values are the constants defined in
1070 :class:`Prompt <msal.Prompt>`.
1072 :param str login_hint:
1073 Optional. Identifier of the user. Generally a User Principal Name (UPN).
1074 :param domain_hint:
1075 Can be one of "consumers" or "organizations" or your tenant domain "contoso.com".
1076 If included, it will skip the email-based discovery process that user goes
1077 through on the sign-in page, leading to a slightly more streamlined user experience.
1078 More information on possible values available in
1079 `Auth Code Flow doc <https://docs.microsoft.com/en-us/azure/active-directory/develop/v2-oauth2-auth-code-flow#request-an-authorization-code>`_ and
1080 `domain_hint doc <https://docs.microsoft.com/en-us/openspecs/windows_protocols/ms-oapx/86fb452d-e34a-494e-ac61-e526e263b6d8>`_.
1082 :param int max_age:
1083 OPTIONAL. Maximum Authentication Age.
1084 Specifies the allowable elapsed time in seconds
1085 since the last time the End-User was actively authenticated.
1086 If the elapsed time is greater than this value,
1087 Microsoft identity platform will actively re-authenticate the End-User.
1089 MSAL Python will also automatically validate the auth_time in ID token.
1091 New in version 1.15.
1093 :param str response_mode:
1094 OPTIONAL. Specifies the method with which response parameters should be returned.
1095 The default value is equivalent to ``query``, which was still secure enough in MSAL Python
1096 (because MSAL Python does not transfer tokens via query parameter in the first place).
1097 For even better security, we recommend using the value ``form_post``.
1098 In "form_post" mode, response parameters
1099 will be encoded as HTML form values that are transmitted via the HTTP POST method and
1100 encoded in the body using the application/x-www-form-urlencoded format.
1101 Valid values can be either "form_post" for HTTP POST to callback URI or
1102 "query" (the default) for HTTP GET with parameters encoded in query string.
1103 More information on possible values
1104 `here <https://openid.net/specs/oauth-v2-multiple-response-types-1_0.html#ResponseModes>`
1105 and `here <https://openid.net/specs/oauth-v2-form-post-response-mode-1_0.html#FormPostResponseMode>`
1107 .. note::
1108 You should configure your web framework to accept form_post responses instead of query responses.
1109 While this parameter still works, it will be removed in a future version.
1110 Using query-based response modes is less secure and should be avoided.
1112 :return:
1113 The auth code flow. It is a dict in this form::
1115 {
1116 "auth_uri": "https://...", // Guide user to visit this
1117 "state": "...", // You may choose to verify it by yourself,
1118 // or just let acquire_token_by_auth_code_flow()
1119 // do that for you.
1120 "...": "...", // Everything else are reserved and internal
1121 }
1123 The caller is expected to:
1125 1. somehow store this content, typically inside the current session,
1126 2. guide the end user (i.e. resource owner) to visit that auth_uri,
1127 3. and then relay this dict and subsequent auth response to
1128 :func:`~acquire_token_by_auth_code_flow()`.
1129 """
1130 # Note to maintainers: Do not emit warning for the use of response_mode here,
1131 # because response_mode=form_post is still the recommended usage for MSAL Python 1.x.
1132 # App developers making the right call shall not be disturbed by unactionable warnings.
1133 client = _ClientWithCcsRoutingInfo(
1134 {"authorization_endpoint": self.authority.authorization_endpoint},
1135 self.client_id,
1136 http_client=self.http_client)
1137 flow = client.initiate_auth_code_flow(
1138 redirect_uri=redirect_uri, state=state, login_hint=login_hint,
1139 prompt=prompt,
1140 scope=self._decorate_scope(scopes),
1141 domain_hint=domain_hint,
1142 claims=_merge_claims_challenge_and_capabilities(
1143 self._client_capabilities, claims_challenge),
1144 max_age=max_age,
1145 response_mode=response_mode,
1146 )
1147 flow["claims_challenge"] = claims_challenge
1148 return flow
1150 def get_authorization_request_url(
1151 self,
1152 scopes, # type: list[str]
1153 login_hint=None, # type: Optional[str]
1154 state=None, # Recommended by OAuth2 for CSRF protection
1155 redirect_uri=None,
1156 response_type="code", # Could be "token" if you use Implicit Grant
1157 prompt=None,
1158 nonce=None,
1159 domain_hint=None, # type: Optional[str]
1160 claims_challenge=None,
1161 **kwargs):
1162 """Constructs a URL for you to start a Authorization Code Grant.
1164 :param list[str] scopes: (Required)
1165 Scopes requested to access a protected API (a resource).
1166 :param str state: Recommended by OAuth2 for CSRF protection.
1167 :param str login_hint:
1168 Identifier of the user. Generally a User Principal Name (UPN).
1169 :param str redirect_uri:
1170 Address to return to upon receiving a response from the authority.
1171 :param str response_type:
1172 Default value is "code" for an OAuth2 Authorization Code grant.
1174 You could use other content such as "id_token" or "token",
1175 which would trigger an Implicit Grant, but that is
1176 `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>`_.
1178 :param str prompt:
1179 By default, no prompt value will be sent, not even string ``"none"``.
1180 You will have to specify a value explicitly.
1181 Its valid values are the constants defined in
1182 :class:`Prompt <msal.Prompt>`.
1183 :param nonce:
1184 A cryptographically random value used to mitigate replay attacks. See also
1185 `OIDC specs <https://openid.net/specs/openid-connect-core-1_0.html#AuthRequest>`_.
1186 :param domain_hint:
1187 Can be one of "consumers" or "organizations" or your tenant domain "contoso.com".
1188 If included, it will skip the email-based discovery process that user goes
1189 through on the sign-in page, leading to a slightly more streamlined user experience.
1190 More information on possible values available in
1191 `Auth Code Flow doc <https://docs.microsoft.com/en-us/azure/active-directory/develop/v2-oauth2-auth-code-flow#request-an-authorization-code>`_ and
1192 `domain_hint doc <https://docs.microsoft.com/en-us/openspecs/windows_protocols/ms-oapx/86fb452d-e34a-494e-ac61-e526e263b6d8>`_.
1193 :param claims_challenge:
1194 The claims_challenge parameter requests specific claims requested by the resource provider
1195 in the form of a claims_challenge directive in the www-authenticate header to be
1196 returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
1197 It is a string of a JSON object which contains lists of claims being requested from these locations.
1199 :return: The authorization url as a string.
1200 """
1201 authority = kwargs.pop("authority", None) # Historically we support this
1202 if authority:
1203 warnings.warn(
1204 "We haven't decided if this method will accept authority parameter")
1205 # The previous implementation is, it will use self.authority by default.
1206 # Multi-tenant app can use new authority on demand
1207 the_authority = Authority(
1208 authority,
1209 self.http_client,
1210 instance_discovery=self._instance_discovery,
1211 ) if authority else self.authority
1213 client = _ClientWithCcsRoutingInfo(
1214 {"authorization_endpoint": the_authority.authorization_endpoint},
1215 self.client_id,
1216 http_client=self.http_client)
1217 warnings.warn(
1218 "Change your get_authorization_request_url() "
1219 "to initiate_auth_code_flow()", DeprecationWarning)
1220 with warnings.catch_warnings(record=True):
1221 return client.build_auth_request_uri(
1222 response_type=response_type,
1223 redirect_uri=redirect_uri, state=state, login_hint=login_hint,
1224 prompt=prompt,
1225 scope=self._decorate_scope(scopes),
1226 nonce=nonce,
1227 domain_hint=domain_hint,
1228 claims=_merge_claims_challenge_and_capabilities(
1229 self._client_capabilities, claims_challenge),
1230 )
1232 def acquire_token_by_auth_code_flow(
1233 self, auth_code_flow, auth_response, scopes=None, **kwargs):
1234 """Validate the auth response being redirected back, and obtain tokens.
1236 It automatically provides nonce protection.
1238 :param dict auth_code_flow:
1239 The same dict returned by :func:`~initiate_auth_code_flow()`.
1240 :param dict auth_response:
1241 A dict of the query string received from auth server.
1242 :param list[str] scopes:
1243 Scopes requested to access a protected API (a resource).
1245 Most of the time, you can leave it empty.
1247 If you requested user consent for multiple resources, here you will
1248 need to provide a subset of what you required in
1249 :func:`~initiate_auth_code_flow()`.
1251 OAuth2 was designed mostly for singleton services,
1252 where tokens are always meant for the same resource and the only
1253 changes are in the scopes.
1254 In Microsoft Entra, tokens can be issued for multiple 3rd party resources.
1255 You can ask authorization code for multiple resources,
1256 but when you redeem it, the token is for only one intended
1257 recipient, called audience.
1258 So the developer need to specify a scope so that we can restrict the
1259 token to be issued for the corresponding audience.
1261 :return:
1262 * A dict containing "access_token" and/or "id_token", among others,
1263 depends on what scope was used.
1264 (See https://tools.ietf.org/html/rfc6749#section-5.1)
1265 * A dict containing "error", optionally "error_description", "error_uri".
1266 (It is either `this <https://tools.ietf.org/html/rfc6749#section-4.1.2.1>`_
1267 or `that <https://tools.ietf.org/html/rfc6749#section-5.2>`_)
1268 * Most client-side data error would result in ValueError exception.
1269 So the usage pattern could be without any protocol details::
1271 def authorize(): # A controller in a web app
1272 try:
1273 result = msal_app.acquire_token_by_auth_code_flow(
1274 session.get("flow", {}), request.args)
1275 if "error" in result:
1276 return render_template("error.html", result)
1277 use(result) # Token(s) are available in result and cache
1278 except ValueError: # Usually caused by CSRF
1279 pass # Simply ignore them
1280 return redirect(url_for("index"))
1281 """
1282 self._validate_ssh_cert_input_data(kwargs.get("data", {}))
1283 telemetry_context = self._build_telemetry_context(
1284 self.ACQUIRE_TOKEN_BY_AUTHORIZATION_CODE_ID)
1285 response = _clean_up(self.client.obtain_token_by_auth_code_flow(
1286 auth_code_flow,
1287 auth_response,
1288 scope=self._decorate_scope(scopes) if scopes else None,
1289 headers=telemetry_context.generate_headers(),
1290 data=dict(
1291 kwargs.pop("data", {}),
1292 claims=_merge_claims_challenge_and_capabilities(
1293 self._client_capabilities,
1294 auth_code_flow.pop("claims_challenge", None))),
1295 **kwargs))
1296 if "access_token" in response:
1297 response[self._TOKEN_SOURCE] = self._TOKEN_SOURCE_IDP
1298 telemetry_context.update_telemetry(response)
1299 return response
1301 def acquire_token_by_authorization_code(
1302 self,
1303 code,
1304 scopes, # Syntactically required. STS accepts empty value though.
1305 redirect_uri=None,
1306 # REQUIRED, if the "redirect_uri" parameter was included in the
1307 # authorization request as described in Section 4.1.1, and their
1308 # values MUST be identical.
1309 nonce=None,
1310 claims_challenge=None,
1311 forwarded_client_claims=None,
1312 **kwargs):
1313 """The second half of the Authorization Code Grant.
1315 :param code: The authorization code returned from Authorization Server.
1316 :param list[str] scopes: (Required)
1317 Scopes requested to access a protected API (a resource).
1319 If you requested user consent for multiple resources, here you will
1320 typically want to provide a subset of what you required in AuthCode.
1322 OAuth2 was designed mostly for singleton services,
1323 where tokens are always meant for the same resource and the only
1324 changes are in the scopes.
1325 In Microsoft Entra, tokens can be issued for multiple 3rd party resources.
1326 You can ask authorization code for multiple resources,
1327 but when you redeem it, the token is for only one intended
1328 recipient, called audience.
1329 So the developer need to specify a scope so that we can restrict the
1330 token to be issued for the corresponding audience.
1332 :param nonce:
1333 If you provided a nonce when calling :func:`get_authorization_request_url`,
1334 same nonce should also be provided here, so that we'll validate it.
1335 An exception will be raised if the nonce in id token mismatches.
1337 :param claims_challenge:
1338 The claims_challenge parameter requests specific claims requested by the resource provider
1339 in the form of a claims_challenge directive in the www-authenticate header to be
1340 returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
1341 It is a string of a JSON object which contains lists of claims being requested from these locations.
1342 :param str forwarded_client_claims:
1343 Optional. A JSON string of *client-originated* claims to include in
1344 the token request. Unlike ``claims_challenge`` (server-issued, which
1345 bypasses the cache), tokens acquired with ``forwarded_client_claims``
1346 **are cached** and keyed on the claims value. Send the *same* value on
1347 every request that should share the cached token; omitting or changing
1348 it routes to a different cache entry (a cache miss), so use stable,
1349 non-dynamic values. The value is merged into the standard OAuth
1350 ``claims`` request parameter sent on the wire.
1352 Not to be confused with the constructor ``client_claims`` parameter
1353 (a ``dict`` of extra claims signed into the client-assertion JWT).
1355 :return: A dict representing the json response from Microsoft Entra:
1357 - A successful response would contain "access_token" key,
1358 - an error response would contain "error" and usually "error_description".
1359 """
1360 # If scope is absent on the wire, STS will give you a token associated
1361 # to the FIRST scope sent during the authorization request.
1362 # So in theory, you can omit scope here when you were working with only
1363 # one scope. But, MSAL decorates your scope anyway, so they are never
1364 # really empty.
1365 assert isinstance(scopes, list), "Invalid parameter type"
1366 self._validate_ssh_cert_input_data(kwargs.get("data", {}))
1367 warnings.warn(
1368 "Change your acquire_token_by_authorization_code() "
1369 "to acquire_token_by_auth_code_flow()", DeprecationWarning)
1370 with warnings.catch_warnings(record=True):
1371 telemetry_context = self._build_telemetry_context(
1372 self.ACQUIRE_TOKEN_BY_AUTHORIZATION_CODE_ID)
1373 _data = kwargs.pop("data", {})
1374 _stash_client_claims(forwarded_client_claims, _data)
1375 response = _clean_up(self.client.obtain_token_by_authorization_code(
1376 code, redirect_uri=redirect_uri,
1377 scope=self._decorate_scope(scopes),
1378 headers=telemetry_context.generate_headers(),
1379 data=dict(
1380 _data,
1381 claims=_merge_claims(
1382 _merge_claims_challenge_and_capabilities(
1383 self._client_capabilities, claims_challenge),
1384 _data.get("client_claims"))),
1385 nonce=nonce,
1386 **kwargs))
1387 if "access_token" in response:
1388 response[self._TOKEN_SOURCE] = self._TOKEN_SOURCE_IDP
1389 telemetry_context.update_telemetry(response)
1390 return response
1392 def get_accounts(self, username=None):
1393 """Get a list of accounts which previously signed in, i.e. exists in cache.
1395 An account can later be used in :func:`~acquire_token_silent`
1396 to find its tokens.
1398 :param username:
1399 Filter accounts with this username only. Case insensitive.
1400 :return: A list of account objects.
1401 Each account is a dict. For now, we only document its "username" field.
1402 Your app can choose to display those information to end user,
1403 and allow user to choose one of his/her accounts to proceed.
1404 """
1405 accounts = self._find_msal_accounts(environment=self.authority.instance)
1406 if not accounts: # Now try other aliases of this authority instance
1407 for alias in self._get_authority_aliases(self.authority.instance):
1408 accounts = self._find_msal_accounts(environment=alias)
1409 if accounts:
1410 break
1411 if username:
1412 # Federated account["username"] from AAD could contain mixed case
1413 lowercase_username = username.lower()
1414 accounts = [a for a in accounts
1415 if a["username"].lower() == lowercase_username]
1416 if not accounts:
1417 logger.debug(( # This would also happen when the cache is empty
1418 "get_accounts(username='{}') finds no account. "
1419 "If tokens were acquired without 'profile' scope, "
1420 "they would contain no username for filtering. "
1421 "Consider calling get_accounts(username=None) instead."
1422 ).format(username))
1423 # Does not further filter by existing RTs here. It probably won't matter.
1424 # Because in most cases Accounts and RTs co-exist.
1425 # Even in the rare case when an RT is revoked and then removed,
1426 # acquire_token_silent() would then yield no result,
1427 # apps would fall back to other acquire methods. This is the standard pattern.
1428 return accounts
1430 def _find_msal_accounts(self, environment):
1431 interested_authority_types = [
1432 TokenCache.AuthorityType.ADFS, TokenCache.AuthorityType.MSSTS]
1433 if _is_running_in_cloud_shell():
1434 interested_authority_types.append(_AUTHORITY_TYPE_CLOUDSHELL)
1435 grouped_accounts = {
1436 a.get("home_account_id"): # Grouped by home tenant's id
1437 { # These are minimal amount of non-tenant-specific account info
1438 "home_account_id": a.get("home_account_id"),
1439 "environment": a.get("environment"),
1440 "username": a.get("username"),
1441 "account_source": a.get("account_source"),
1443 # The following fields for backward compatibility, for now
1444 "authority_type": a.get("authority_type"),
1445 "local_account_id": a.get("local_account_id"), # Tenant-specific
1446 "realm": a.get("realm"), # Tenant-specific
1447 }
1448 for a in self.token_cache.search(
1449 TokenCache.CredentialType.ACCOUNT,
1450 query={"environment": environment})
1451 if a["authority_type"] in interested_authority_types
1452 }
1453 return list(grouped_accounts.values())
1455 def _get_instance_metadata(self, instance): # This exists so it can be mocked in unit test
1456 instance_discovery_host = _get_instance_discovery_host(instance)
1457 resp = self.http_client.get(
1458 _get_instance_discovery_endpoint(instance),
1459 params={
1460 'api-version': '1.1',
1461 'authorization_endpoint': (
1462 "https://{}/common/oauth2/authorize".format(instance_discovery_host)
1463 ),
1464 },
1465 headers={'Accept': 'application/json'})
1466 resp.raise_for_status()
1467 return json.loads(resp.text)['metadata']
1469 def _get_authority_aliases(self, instance):
1470 if self._instance_discovery is False:
1471 return []
1472 if self.authority._is_known_to_developer:
1473 # Then it is an ADFS/B2C/known_authority_hosts situation
1474 # which may not reach the central endpoint, so we skip it.
1475 return []
1476 if instance not in self.authority_groups:
1477 self.authority_groups[instance] = [
1478 set(group['aliases']) for group in self._get_instance_metadata(instance)]
1479 for group in self.authority_groups[instance]:
1480 if instance in group:
1481 return [alias for alias in group if alias != instance]
1482 return []
1484 def remove_account(self, account):
1485 """Sign me out and forget me from token cache"""
1486 if self._enable_broker:
1487 from .broker import _signout_silently
1488 error = _signout_silently(self.client_id, account["local_account_id"])
1489 if error:
1490 logger.debug("_signout_silently() returns error: %s", error)
1491 # Broker sign-out has been attempted, even if the _forget_me() below throws.
1492 self._forget_me(account)
1494 def _sign_out(self, home_account):
1495 # Remove all relevant RTs and ATs from token cache
1496 owned_by_home_account = {
1497 "environment": home_account["environment"],
1498 "home_account_id": home_account["home_account_id"],} # realm-independent
1499 app_metadata = self._get_app_metadata(home_account["environment"])
1500 # Remove RTs/FRTs, and they are realm-independent
1501 for rt in [ # Remove RTs from a static list (rather than from a dynamic generator),
1502 # to avoid changing self.token_cache while it is being iterated
1503 rt for rt in self.token_cache.search(
1504 TokenCache.CredentialType.REFRESH_TOKEN, query=owned_by_home_account)
1505 # Do RT's app ownership check as a precaution, in case family apps
1506 # and 3rd-party apps share same token cache, although they should not.
1507 if rt["client_id"] == self.client_id or (
1508 app_metadata.get("family_id") # Now let's settle family business
1509 and rt.get("family_id") == app_metadata["family_id"])
1510 ]:
1511 self.token_cache.remove_rt(rt)
1512 for at in list(self.token_cache.search( # Remove ATs from a static list,
1513 # to avoid changing self.token_cache while it is being iterated
1514 TokenCache.CredentialType.ACCESS_TOKEN, query=owned_by_home_account,
1515 # Regardless of realm, b/c we've removed realm-independent RTs anyway
1516 )):
1517 # To avoid the complexity of locating sibling family app's AT,
1518 # we skip AT's app ownership check.
1519 # It means ATs for other apps will also be removed, it is OK because:
1520 # * non-family apps are not supposed to share token cache to begin with;
1521 # * Even if it happens, we keep other app's RT already, so SSO still works
1522 self.token_cache.remove_at(at)
1524 def _forget_me(self, home_account):
1525 # It implies signout, and then also remove all relevant accounts and IDTs
1526 self._sign_out(home_account)
1527 owned_by_home_account = {
1528 "environment": home_account["environment"],
1529 "home_account_id": home_account["home_account_id"],} # realm-independent
1530 for idt in list(self.token_cache.search( # Remove IDTs from a static list,
1531 # to avoid changing self.token_cache while it is being iterated
1532 TokenCache.CredentialType.ID_TOKEN, query=owned_by_home_account, # regardless of realm
1533 )):
1534 self.token_cache.remove_idt(idt)
1535 for a in list(self.token_cache.search( # Remove Accounts from a static list,
1536 # to avoid changing self.token_cache while it is being iterated
1537 TokenCache.CredentialType.ACCOUNT, query=owned_by_home_account, # regardless of realm
1538 )):
1539 self.token_cache.remove_account(a)
1541 def _acquire_token_by_cloud_shell(self, scopes, data=None):
1542 from .cloudshell import _obtain_token
1543 response = _obtain_token(
1544 self.http_client, scopes, client_id=self.client_id, data=data)
1545 if "error" not in response:
1546 self.token_cache.add(dict(
1547 client_id=self.client_id,
1548 scope=response["scope"].split() if "scope" in response else scopes,
1549 token_endpoint=self.authority.token_endpoint,
1550 response=response,
1551 data=data or {},
1552 authority_type=_AUTHORITY_TYPE_CLOUDSHELL,
1553 ))
1554 if "access_token" in response:
1555 response[self._TOKEN_SOURCE] = self._TOKEN_SOURCE_BROKER
1556 return response
1558 def acquire_token_silent(
1559 self,
1560 scopes, # type: List[str]
1561 account, # type: Optional[Account]
1562 authority=None, # See get_authorization_request_url()
1563 force_refresh=False, # type: Optional[boolean]
1564 claims_challenge=None,
1565 forwarded_client_claims=None,
1566 auth_scheme=None,
1567 **kwargs):
1568 """Acquire an access token for given account, without user interaction.
1570 It has same parameters as the :func:`~acquire_token_silent_with_error`.
1571 The difference is the behavior of the return value.
1572 This method will combine the cache empty and refresh error
1573 into one return value, `None`.
1574 If your app does not care about the exact token refresh error during
1575 token cache look-up, then this method is easier and recommended.
1577 :return:
1578 - A dict containing no "error" key,
1579 and typically contains an "access_token" key,
1580 if cache lookup succeeded.
1581 - None when cache lookup does not yield a token.
1582 """
1583 if not account:
1584 return None # A backward-compatible NO-OP to drop the account=None usage
1585 if forwarded_client_claims is not None:
1586 kwargs["data"] = kwargs.get("data", {})
1587 _stash_client_claims(forwarded_client_claims, kwargs["data"])
1588 result = _clean_up(self._acquire_token_silent_with_error(
1589 scopes, account, authority=authority, force_refresh=force_refresh,
1590 claims_challenge=claims_challenge, auth_scheme=auth_scheme, **kwargs))
1591 return result if result and "error" not in result else None
1593 def acquire_token_silent_with_error(
1594 self,
1595 scopes, # type: List[str]
1596 account, # type: Optional[Account]
1597 authority=None, # See get_authorization_request_url()
1598 force_refresh=False, # type: Optional[boolean]
1599 claims_challenge=None,
1600 forwarded_client_claims=None,
1601 auth_scheme=None,
1602 **kwargs):
1603 """Acquire an access token for given account, without user interaction.
1605 It is done either by finding a valid access token from cache,
1606 or by finding a valid refresh token from cache and then automatically
1607 use it to redeem a new access token.
1609 This method will differentiate cache empty from token refresh error.
1610 If your app cares the exact token refresh error during
1611 token cache look-up, then this method is suitable.
1612 Otherwise, the other method :func:`~acquire_token_silent` is recommended.
1614 :param list[str] scopes: (Required)
1615 Scopes requested to access a protected API (a resource).
1616 :param account: (Required)
1617 One of the account object returned by :func:`~get_accounts`.
1618 Starting from MSAL Python 1.23,
1619 a ``None`` input will become a NO-OP and always return ``None``.
1620 :param force_refresh:
1621 If True, it will skip Access Token look-up,
1622 and try to find a Refresh Token to obtain a new Access Token.
1623 :param claims_challenge:
1624 The claims_challenge parameter requests specific claims requested by the resource provider
1625 in the form of a claims_challenge directive in the www-authenticate header to be
1626 returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
1627 It is a string of a JSON object which contains lists of claims being requested from these locations.
1628 :param str forwarded_client_claims:
1629 Optional. A JSON string of *client-originated* claims, applied only
1630 when no cached token is found and a network request is made. Unlike
1631 ``claims_challenge`` (server-issued, which bypasses the cache), tokens
1632 acquired with ``forwarded_client_claims`` **are cached** and keyed on
1633 the claims value. Send the *same* value on every call that should
1634 reuse the cached token; different or omitted values route to separate
1635 cache entries, so use stable, non-dynamic values.
1637 Not to be confused with the constructor ``client_claims`` parameter
1638 (a ``dict`` of extra claims signed into the client-assertion JWT).
1639 :param object auth_scheme:
1640 You can provide an ``msal.auth_scheme.PopAuthScheme`` object
1641 so that MSAL will get a Proof-of-Possession (POP) token for you.
1643 New in version 1.26.0.
1645 :return:
1646 - A dict containing no "error" key,
1647 and typically contains an "access_token" key,
1648 if cache lookup succeeded.
1649 - None when there is simply no token in the cache.
1650 - A dict containing an "error" key, when token refresh failed.
1651 """
1652 if not account:
1653 return None # A backward-compatible NO-OP to drop the account=None usage
1654 if forwarded_client_claims is not None:
1655 kwargs["data"] = kwargs.get("data", {})
1656 _stash_client_claims(forwarded_client_claims, kwargs["data"])
1657 return _clean_up(self._acquire_token_silent_with_error(
1658 scopes, account, authority=authority, force_refresh=force_refresh,
1659 claims_challenge=claims_challenge, auth_scheme=auth_scheme, **kwargs))
1661 def _acquire_token_silent_with_error(
1662 self,
1663 scopes, # type: List[str]
1664 account, # type: Optional[Account]
1665 authority=None, # See get_authorization_request_url()
1666 force_refresh=False, # type: Optional[boolean]
1667 claims_challenge=None,
1668 auth_scheme=None,
1669 **kwargs):
1670 assert isinstance(scopes, list), "Invalid parameter type"
1671 self._validate_ssh_cert_input_data(kwargs.get("data", {}))
1672 correlation_id = msal.telemetry._get_new_correlation_id()
1673 if authority:
1674 warnings.warn("We haven't decided how/if this method will accept authority parameter")
1675 # the_authority = Authority(
1676 # authority,
1677 # self.http_client,
1678 # instance_discovery=self._instance_discovery,
1679 # ) if authority else self.authority
1680 result = self._acquire_token_silent_from_cache_and_possibly_refresh_it(
1681 scopes, account, self.authority, force_refresh=force_refresh,
1682 claims_challenge=claims_challenge,
1683 correlation_id=correlation_id,
1684 auth_scheme=auth_scheme,
1685 **kwargs)
1686 if result and "error" not in result:
1687 return result
1688 final_result = result
1689 for alias in self._get_authority_aliases(self.authority.instance):
1690 if not list(self.token_cache.search( # Need a list to test emptiness
1691 self.token_cache.CredentialType.REFRESH_TOKEN,
1692 # target=scopes, # MUST NOT filter by scopes, because:
1693 # 1. AAD RTs are scope-independent;
1694 # 2. therefore target is optional per schema;
1695 query={"environment": alias})):
1696 # Skip heavy weight logic when RT for this alias doesn't exist
1697 continue
1698 the_authority = Authority(
1699 "https://" + alias + "/" + self.authority.tenant,
1700 self.http_client,
1701 instance_discovery=False,
1702 )
1703 result = self._acquire_token_silent_from_cache_and_possibly_refresh_it(
1704 scopes, account, the_authority, force_refresh=force_refresh,
1705 claims_challenge=claims_challenge,
1706 correlation_id=correlation_id,
1707 auth_scheme=auth_scheme,
1708 **kwargs)
1709 if result:
1710 if "error" not in result:
1711 return result
1712 final_result = result
1713 if final_result and final_result.get("suberror"):
1714 final_result["classification"] = { # Suppress these suberrors, per #57
1715 "bad_token": "",
1716 "token_expired": "",
1717 "protection_policy_required": "",
1718 "client_mismatch": "",
1719 "device_authentication_failed": "",
1720 }.get(final_result["suberror"], final_result["suberror"])
1721 return final_result
1723 def _acquire_token_silent_from_cache_and_possibly_refresh_it(
1724 self,
1725 scopes, # type: List[str]
1726 account, # type: Optional[Account]
1727 authority, # This can be different than self.authority
1728 force_refresh=False, # type: Optional[boolean]
1729 claims_challenge=None,
1730 correlation_id=None,
1731 http_exceptions=None,
1732 auth_scheme=None,
1733 **kwargs):
1734 # This internal method has two calling patterns:
1735 # it accepts a non-empty account to find token for a user,
1736 # and accepts account=None to find a token for the current app.
1737 access_token_from_cache = None
1738 if not (force_refresh or claims_challenge or auth_scheme): # Then attempt AT cache
1739 query={
1740 "client_id": self.client_id,
1741 "environment": authority.instance,
1742 "realm": authority.tenant,
1743 "home_account_id": (account or {}).get("home_account_id"),
1744 }
1745 key_id = kwargs.get("data", {}).get("key_id")
1746 if key_id: # Some token types (SSH-certs, POP) are bound to a key
1747 query["key_id"] = key_id
1748 ext_cache_key = _compute_ext_cache_key(kwargs.get("data", {}))
1749 if ext_cache_key: # FMI tokens need cache isolation by path
1750 query["ext_cache_key"] = ext_cache_key
1751 now = time.time()
1752 refresh_reason = msal.telemetry.AT_ABSENT
1753 for entry in self.token_cache.search( # A generator allows us to
1754 # break early in cache-hit without finding a full list
1755 self.token_cache.CredentialType.ACCESS_TOKEN,
1756 target=scopes,
1757 query=query,
1758 ): # This loop is about token search, not about token deletion.
1759 # Note that search() holds a lock during this loop;
1760 # that is fine because this loop is fast
1761 expires_in = int(entry["expires_on"]) - now
1762 if expires_in < 5*60: # Then consider it expired
1763 refresh_reason = msal.telemetry.AT_EXPIRED
1764 continue # Removal is not necessary, it will be overwritten
1765 logger.debug("Cache hit an AT")
1766 access_token_from_cache = { # Mimic a real response
1767 "access_token": entry["secret"],
1768 "token_type": entry.get("token_type", "Bearer"),
1769 "expires_in": int(expires_in), # OAuth2 specs defines it as int
1770 self._TOKEN_SOURCE: self._TOKEN_SOURCE_CACHE,
1771 }
1772 if "refresh_on" in entry:
1773 access_token_from_cache["refresh_on"] = int(entry["refresh_on"])
1774 if int(entry["refresh_on"]) < now: # aging
1775 refresh_reason = msal.telemetry.AT_AGING
1776 break # With a fallback in hand, we break here to go refresh
1777 self._build_telemetry_context(-1).hit_an_access_token()
1778 return access_token_from_cache # It is still good as new
1779 else:
1780 refresh_reason = msal.telemetry.FORCE_REFRESH # TODO: It could also mean claims_challenge
1781 assert refresh_reason, "It should have been established at this point"
1782 if not http_exceptions: # It can be a tuple of exceptions
1783 # The exact HTTP exceptions are transportation-layer dependent
1784 from requests.exceptions import RequestException # Lazy load
1785 http_exceptions = (RequestException,)
1786 try:
1787 data = kwargs.get("data", {})
1788 if account and account.get("authority_type") == _AUTHORITY_TYPE_CLOUDSHELL:
1789 if auth_scheme:
1790 raise ValueError("auth_scheme is not supported in Cloud Shell")
1791 return self._acquire_token_by_cloud_shell(scopes, data=data)
1793 is_ssh_cert_or_pop_request = _is_ssh_cert_or_pop_request(data.get("token_type"), auth_scheme)
1795 if self._enable_broker and account and account.get("account_source") in (
1796 _GRANT_TYPE_BROKER, # Broker successfully established this account previously.
1797 None, # Unknown data from older MSAL. Broker might still work.
1798 ) and (sys.platform in ("win32", "darwin") or not is_ssh_cert_or_pop_request):
1799 from .broker import _acquire_token_silently
1800 response = _acquire_token_silently(
1801 "https://{}/{}".format(self.authority.instance, self.authority.tenant),
1802 self.client_id,
1803 account["local_account_id"],
1804 scopes,
1805 claims=_merge_claims_challenge_and_capabilities(
1806 self._client_capabilities, claims_challenge),
1807 correlation_id=correlation_id,
1808 auth_scheme=auth_scheme,
1809 **data)
1810 if response: # Broker provides a decisive outcome
1811 account_was_established_by_broker = account.get(
1812 "account_source") == _GRANT_TYPE_BROKER
1813 broker_attempt_succeeded_just_now = "error" not in response
1814 if account_was_established_by_broker or broker_attempt_succeeded_just_now:
1815 return self._process_broker_response(response, scopes, data)
1817 if auth_scheme:
1818 raise ValueError(self._AUTH_SCHEME_UNSUPPORTED)
1819 if account:
1820 result = self._acquire_token_silent_by_finding_rt_belongs_to_me_or_my_family(
1821 authority, self._decorate_scope(scopes), account,
1822 refresh_reason=refresh_reason, claims_challenge=claims_challenge,
1823 correlation_id=correlation_id,
1824 **kwargs)
1825 else: # The caller is acquire_token_for_client()
1826 result = self._acquire_token_for_client(
1827 scopes, refresh_reason, claims_challenge=claims_challenge,
1828 **kwargs)
1829 if result and "access_token" in result:
1830 result[self._TOKEN_SOURCE] = self._TOKEN_SOURCE_IDP
1831 if (result and "error" not in result) or (not access_token_from_cache):
1832 return result
1833 except http_exceptions:
1834 # Typically network error. Potential AAD outage?
1835 if not access_token_from_cache: # It means there is no fall back option
1836 raise # We choose to bubble up the exception
1837 return access_token_from_cache
1839 def _process_broker_response(self, response, scopes, data):
1840 if "error" not in response:
1841 self.token_cache.add(dict(
1842 client_id=self.client_id,
1843 scope=response["scope"].split() if "scope" in response else scopes,
1844 token_endpoint=self.authority.token_endpoint,
1845 response=response,
1846 data=data,
1847 _account_id=response["_account_id"],
1848 environment=self.authority.instance, # Be consistent with non-broker flows
1849 grant_type=_GRANT_TYPE_BROKER, # A pseudo grant type for TokenCache to mark account_source as broker
1850 ))
1851 response[self._TOKEN_SOURCE] = self._TOKEN_SOURCE_BROKER
1852 return _clean_up(response)
1854 def _acquire_token_silent_by_finding_rt_belongs_to_me_or_my_family(
1855 self, authority, scopes, account, **kwargs):
1856 query = {
1857 "environment": authority.instance,
1858 "home_account_id": (account or {}).get("home_account_id"),
1859 # "realm": authority.tenant, # AAD RTs are tenant-independent
1860 }
1861 app_metadata = self._get_app_metadata(authority.instance)
1862 if not app_metadata: # Meaning this app is now used for the first time.
1863 # When/if we have a way to directly detect current app's family,
1864 # we'll rewrite this block, to support multiple families.
1865 # For now, we try existing RTs (*). If it works, we are in that family.
1866 # (*) RTs of a different app/family are not supposed to be
1867 # shared with or accessible by us in the first place.
1868 at = self._acquire_token_silent_by_finding_specific_refresh_token(
1869 authority, scopes,
1870 dict(query, family_id="1"), # A hack, we have only 1 family for now
1871 rt_remover=lambda rt_item: None, # NO-OP b/c RTs are likely not mine
1872 break_condition=lambda response: # Break loop when app not in family
1873 # Based on an AAD-only behavior mentioned in internal doc here
1874 # https://msazure.visualstudio.com/One/_git/ESTS-Docs/pullrequest/1138595
1875 "client_mismatch" in response.get("error_additional_info", []),
1876 **kwargs)
1877 if at and "error" not in at:
1878 return at
1879 last_resp = None
1880 if app_metadata.get("family_id"): # Meaning this app belongs to this family
1881 last_resp = at = self._acquire_token_silent_by_finding_specific_refresh_token(
1882 authority, scopes, dict(query, family_id=app_metadata["family_id"]),
1883 **kwargs)
1884 if at and "error" not in at:
1885 return at
1886 # Either this app is an orphan, so we will naturally use its own RT;
1887 # or all attempts above have failed, so we fall back to non-foci behavior.
1888 return self._acquire_token_silent_by_finding_specific_refresh_token(
1889 authority, scopes, dict(query, client_id=self.client_id),
1890 **kwargs) or last_resp
1892 def _get_app_metadata(self, environment):
1893 return self.token_cache._get_app_metadata(
1894 environment=environment, client_id=self.client_id, default={})
1896 def _acquire_token_silent_by_finding_specific_refresh_token(
1897 self, authority, scopes, query,
1898 rt_remover=None, break_condition=lambda response: False,
1899 refresh_reason=None, correlation_id=None, claims_challenge=None,
1900 **kwargs):
1901 matches = list(self.token_cache.search( # We want a list to test emptiness
1902 self.token_cache.CredentialType.REFRESH_TOKEN,
1903 # target=scopes, # AAD RTs are scope-independent
1904 query=query))
1905 logger.debug("Found %d RTs matching %s", len(matches), {
1906 k: _pii_less_home_account_id(v) if k == "home_account_id" and v else v
1907 for k, v in query.items()
1908 })
1910 response = None # A distinguishable value to mean cache is empty
1911 if not matches: # Then exit early to avoid expensive operations
1912 return response
1913 client, _ = self._build_client(
1914 # Potentially expensive if building regional client
1915 self.client_credential, authority, skip_regional_client=True)
1916 telemetry_context = self._build_telemetry_context(
1917 self.ACQUIRE_TOKEN_SILENT_ID,
1918 correlation_id=correlation_id, refresh_reason=refresh_reason)
1919 # Pop "data" once (rather than per-iteration) so client_claims and any
1920 # other data fields apply consistently across all candidate RTs.
1921 _data = kwargs.pop("data", {})
1922 for entry in sorted( # Since unfit RTs would not be aggressively removed,
1923 # we start from newer RTs which are more likely fit.
1924 matches,
1925 key=lambda e: int(e.get("last_modification_time", "0")),
1926 reverse=True):
1927 logger.debug("Cache attempts an RT")
1928 headers = telemetry_context.generate_headers()
1929 if query.get("home_account_id"): # Then use it as CCS Routing info
1930 headers["X-AnchorMailbox"] = "Oid:{}".format( # case-insensitive value
1931 query["home_account_id"].replace(".", "@"))
1932 response = client.obtain_token_by_refresh_token(
1933 entry, rt_getter=lambda token_item: token_item["secret"],
1934 on_removing_rt=lambda rt_item: None, # Disable RT removal,
1935 # because an invalid_grant could be caused by new MFA policy,
1936 # the RT could still be useful for other MFA-less scope or tenant
1937 on_obtaining_tokens=lambda event: self.token_cache.add(dict(
1938 event,
1939 environment=authority.instance,
1940 skip_account_creation=True, # To honor a concurrent remove_account()
1941 )),
1942 scope=scopes,
1943 headers=headers,
1944 data=dict(
1945 _data,
1946 claims=_merge_claims(
1947 _merge_claims_challenge_and_capabilities(
1948 self._client_capabilities, claims_challenge),
1949 _data.get("client_claims"))),
1950 **kwargs)
1951 telemetry_context.update_telemetry(response)
1952 if "error" not in response:
1953 return response
1954 logger.debug("Refresh failed. {error}: {error_description}".format(
1955 error=response.get("error"),
1956 error_description=response.get("error_description"),
1957 ))
1958 if break_condition(response):
1959 break
1960 return response # Returns the latest error (if any), or just None
1962 def _validate_ssh_cert_input_data(self, data):
1963 if data.get("token_type") == "ssh-cert":
1964 if not data.get("req_cnf"):
1965 raise ValueError(
1966 "When requesting an SSH certificate, "
1967 "you must include a string parameter named 'req_cnf' "
1968 "containing the public key in JWK format "
1969 "(https://tools.ietf.org/html/rfc7517).")
1970 if not data.get("key_id"):
1971 raise ValueError(
1972 "When requesting an SSH certificate, "
1973 "you must include a string parameter named 'key_id' "
1974 "which identifies the key in the 'req_cnf' argument.")
1976 def acquire_token_by_refresh_token(self, refresh_token, scopes, **kwargs):
1977 """Acquire token(s) based on a refresh token (RT) obtained from elsewhere.
1979 You use this method only when you have old RTs from elsewhere,
1980 and now you want to migrate them into MSAL.
1981 Calling this method results in new tokens automatically storing into MSAL.
1983 You do NOT need to use this method if you are already using MSAL.
1984 MSAL maintains RT automatically inside its token cache,
1985 and an access token can be retrieved
1986 when you call :func:`~acquire_token_silent`.
1988 :param str refresh_token: The old refresh token, as a string.
1990 :param list scopes:
1991 The scopes associate with this old RT.
1992 Each scope needs to be in the Microsoft identity platform (v2) format.
1993 See `Scopes not resources <https://docs.microsoft.com/en-us/azure/active-directory/develop/migrate-python-adal-msal#scopes-not-resources>`_.
1995 :return:
1996 * A dict contains "error" and some other keys, when error happened.
1997 * A dict contains no "error" key means migration was successful.
1998 """
1999 self._validate_ssh_cert_input_data(kwargs.get("data", {}))
2000 telemetry_context = self._build_telemetry_context(
2001 self.ACQUIRE_TOKEN_BY_REFRESH_TOKEN,
2002 refresh_reason=msal.telemetry.FORCE_REFRESH)
2003 response = _clean_up(self.client.obtain_token_by_refresh_token(
2004 refresh_token,
2005 scope=self._decorate_scope(scopes),
2006 headers=telemetry_context.generate_headers(),
2007 rt_getter=lambda rt: rt,
2008 on_updating_rt=False,
2009 on_removing_rt=lambda rt_item: None, # No OP
2010 **kwargs))
2011 if "access_token" in response:
2012 response[self._TOKEN_SOURCE] = self._TOKEN_SOURCE_IDP
2013 telemetry_context.update_telemetry(response)
2014 return response
2016 def acquire_token_by_username_password(
2017 self, username, password, scopes, claims_challenge=None,
2018 # Note: We shouldn't need to surface enable_msa_passthrough,
2019 # because this ROPC won't work with MSA account anyway.
2020 auth_scheme=None,
2021 **kwargs):
2022 """Gets a token for a given resource via user credentials.
2024 See this page for constraints of Username Password Flow.
2025 https://github.com/AzureAD/microsoft-authentication-library-for-python/wiki/Username-Password-Authentication
2027 :param str username: Typically a UPN in the form of an email address.
2028 :param str password: The password.
2029 :param list[str] scopes:
2030 Scopes requested to access a protected API (a resource).
2031 :param claims_challenge:
2032 The claims_challenge parameter requests specific claims requested by the resource provider
2033 in the form of a claims_challenge directive in the www-authenticate header to be
2034 returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
2035 It is a string of a JSON object which contains lists of claims being requested from these locations.
2037 :param object auth_scheme:
2038 You can provide an ``msal.auth_scheme.PopAuthScheme`` object
2039 so that MSAL will get a Proof-of-Possession (POP) token for you.
2041 New in version 1.26.0.
2043 :return: A dict representing the json response from Microsoft Entra:
2045 - A successful response would contain "access_token" key,
2046 - an error response would contain "error" and usually "error_description".
2048 [Deprecated] This API is deprecated for public client flows and will be
2049 removed in a future release. Use a more secure flow instead.
2050 Migration guide: https://aka.ms/msal-ropc-migration
2052 """
2053 is_confidential_app = self.client_credential or isinstance(
2054 self, ConfidentialClientApplication)
2055 if not is_confidential_app:
2056 warnings.warn("""This API has been deprecated for public client flows, please use a more secure flow.
2057 See https://aka.ms/msal-ropc-migration for migration guidance""", DeprecationWarning)
2058 claims = _merge_claims_challenge_and_capabilities(
2059 self._client_capabilities, claims_challenge)
2060 if self._enable_broker and sys.platform in ("win32", "darwin"):
2061 from .broker import _signin_silently
2062 response = _signin_silently(
2063 "https://{}/{}".format(self.authority.instance, self.authority.tenant),
2064 self.client_id,
2065 scopes, # Decorated scopes won't work due to offline_access
2066 MSALRuntime_Username=username,
2067 MSALRuntime_Password=password,
2068 validateAuthority="no" if (
2069 self.authority._is_known_to_developer
2070 or self._instance_discovery is False) else None,
2071 claims=claims,
2072 auth_scheme=auth_scheme,
2073 )
2074 return self._process_broker_response(response, scopes, kwargs.get("data", {}))
2076 if auth_scheme:
2077 raise ValueError(self._AUTH_SCHEME_UNSUPPORTED)
2078 scopes = self._decorate_scope(scopes)
2079 telemetry_context = self._build_telemetry_context(
2080 self.ACQUIRE_TOKEN_BY_USERNAME_PASSWORD_ID)
2081 headers = telemetry_context.generate_headers()
2082 data = dict(kwargs.pop("data", {}), claims=claims)
2083 response = None
2084 if not self.authority.is_adfs:
2085 user_realm_result = self.authority.user_realm_discovery(
2086 username, correlation_id=headers[msal.telemetry.CLIENT_REQUEST_ID])
2087 if user_realm_result.get("account_type") == "Federated":
2088 response = _clean_up(self._acquire_token_by_username_password_federated(
2089 user_realm_result, username, password, scopes=scopes,
2090 data=data,
2091 headers=headers, **kwargs))
2092 if response is None: # Either ADFS or not federated
2093 response = _clean_up(self.client.obtain_token_by_username_password(
2094 username, password, scope=scopes,
2095 headers=headers,
2096 data=data,
2097 **kwargs))
2098 if "access_token" in response:
2099 response[self._TOKEN_SOURCE] = self._TOKEN_SOURCE_IDP
2100 telemetry_context.update_telemetry(response)
2101 return response
2103 def _acquire_token_by_username_password_federated(
2104 self, user_realm_result, username, password, scopes=None, **kwargs):
2105 wstrust_endpoint = {}
2106 if user_realm_result.get("federation_metadata_url"):
2107 wstrust_endpoint = mex_send_request(
2108 user_realm_result["federation_metadata_url"],
2109 self.http_client)
2110 if wstrust_endpoint is None:
2111 raise ValueError("Unable to find wstrust endpoint from MEX. "
2112 "This typically happens when attempting MSA accounts. "
2113 "More details available here. "
2114 "https://github.com/AzureAD/microsoft-authentication-library-for-python/wiki/Username-Password-Authentication")
2115 logger.debug("wstrust_endpoint = %s", wstrust_endpoint)
2116 wstrust_result = wst_send_request(
2117 username, password,
2118 user_realm_result.get("cloud_audience_urn", "urn:federation:MicrosoftOnline"),
2119 wstrust_endpoint.get("address",
2120 # Fallback to an AAD supplied endpoint
2121 user_realm_result.get("federation_active_auth_url")),
2122 wstrust_endpoint.get("action"), self.http_client)
2123 if not ("token" in wstrust_result and "type" in wstrust_result):
2124 raise RuntimeError("Unsuccessful RSTR. %s" % wstrust_result)
2125 GRANT_TYPE_SAML1_1 = 'urn:ietf:params:oauth:grant-type:saml1_1-bearer'
2126 grant_type = {
2127 SAML_TOKEN_TYPE_V1: GRANT_TYPE_SAML1_1,
2128 SAML_TOKEN_TYPE_V2: self.client.GRANT_TYPE_SAML2,
2129 WSS_SAML_TOKEN_PROFILE_V1_1: GRANT_TYPE_SAML1_1,
2130 WSS_SAML_TOKEN_PROFILE_V2: self.client.GRANT_TYPE_SAML2
2131 }.get(wstrust_result.get("type"))
2132 if not grant_type:
2133 raise RuntimeError(
2134 "RSTR returned unknown token type: %s", wstrust_result.get("type"))
2135 self.client.grant_assertion_encoders.setdefault( # Register a non-standard type
2136 grant_type, self.client.encode_saml_assertion)
2137 return self.client.obtain_token_by_assertion(
2138 wstrust_result["token"], grant_type, scope=scopes,
2139 on_obtaining_tokens=lambda event: self.token_cache.add(dict(
2140 event,
2141 environment=self.authority.instance,
2142 username=username, # Useful in case IDT contains no such info
2143 )),
2144 **kwargs)
2147class PublicClientApplication(ClientApplication): # browser app or mobile app
2149 DEVICE_FLOW_CORRELATION_ID = "_correlation_id"
2150 CONSOLE_WINDOW_HANDLE = object()
2152 def __init__(
2153 self, client_id, client_credential=None,
2154 *,
2155 enable_broker_on_windows=None,
2156 enable_broker_on_mac=None,
2157 enable_broker_on_linux=None,
2158 enable_broker_on_wsl=None,
2159 **kwargs):
2160 """Same as :func:`ClientApplication.__init__`,
2161 except that ``client_credential`` parameter shall remain ``None``.
2163 .. note::
2165 **What is a broker, and why use it?**
2167 A broker is a component installed on your device.
2168 Broker implicitly gives your device an identity. By using a broker,
2169 your device becomes a factor that can satisfy MFA (Multi-factor authentication).
2170 This factor would become mandatory
2171 if a tenant's admin enables a corresponding Conditional Access (CA) policy.
2172 The broker's presence allows Microsoft identity platform
2173 to have higher confidence that the tokens are being issued to your device,
2174 and that is more secure.
2176 An additional benefit of broker is,
2177 it runs as a long-lived process with your device's OS,
2178 and maintains its own cache,
2179 so that your broker-enabled apps (even a CLI)
2180 could automatically SSO from a previously established signed-in session.
2182 **How to opt in to use broker?**
2184 1. You can set any combination of the following opt-in parameters to true:
2186 +--------------------------+-------------------------------------------------+------------------------------------------------------------------------------------+
2187 | Opt-in flag | If app will run on | App has registered this as a Desktop platform redirect URI in Azure Portal |
2188 +==========================+=================================================+====================================================================================+
2189 | enable_broker_on_windows | Windows 10+ | ms-appx-web://Microsoft.AAD.BrokerPlugin/your_client_id |
2190 +--------------------------+-------------------------------------------------+------------------------------------------------------------------------------------+
2191 | enable_broker_on_wsl | WSL | ms-appx-web://Microsoft.AAD.BrokerPlugin/your_client_id |
2192 +--------------------------+-------------------------------------------------+------------------------------------------------------------------------------------+
2193 | enable_broker_on_mac | Apple Silicon Mac with Company Portal installed | msauth.com.msauth.unsignedapp://auth |
2194 +--------------------------+-------------------------------------------------+------------------------------------------------------------------------------------+
2195 | enable_broker_on_linux | Linux with Intune installed | ``https://login.microsoftonline.com/common/oauth2/nativeclient`` (MUST be enabled) |
2196 +--------------------------+-------------------------------------------------+------------------------------------------------------------------------------------+
2198 2. Install broker dependency,
2199 e.g. ``pip install msal[broker]>=1.33,<2``.
2201 3. Test with ``acquire_token_interactive()`` and ``acquire_token_silent()``.
2203 **The fallback behaviors of MSAL Python's broker support**
2205 MSAL will either error out, or silently fallback to non-broker flows.
2207 1. MSAL will ignore the `enable_broker_...` and bypass broker
2208 on those auth flows that are known to be NOT supported by broker.
2209 This includes ADFS, B2C, etc..
2210 For other "could-use-broker" scenarios, please see below.
2211 2. MSAL errors out when app developer opted-in to use broker
2212 but a direct dependency "mid-tier" package is not installed.
2213 Error message guides app developer to declare the correct dependency
2214 ``msal[broker]``.
2215 We error out here because the error is actionable to app developers.
2216 3. MSAL silently "deactivates" the broker and fallback to non-broker,
2217 when opted-in, dependency installed yet failed to initialize.
2218 We anticipate this would happen on a device whose OS is too old
2219 or the underlying broker component is somehow unavailable.
2220 There is not much an app developer or the end user can do here.
2221 Eventually, the conditional access policy shall
2222 force the user to switch to a different device.
2223 4. MSAL errors out when broker is opted in, installed, initialized,
2224 but subsequent token request(s) failed.
2226 :param boolean enable_broker_on_windows:
2227 This setting is only effective if your app is running on Windows 10+.
2228 This parameter defaults to None, which means MSAL will not utilize a broker.
2230 New in MSAL Python 1.25.0.
2232 :param boolean enable_broker_on_mac:
2233 This setting is only effective if your app is running on
2234 an Apple Silicon (arm64) Mac.
2235 Broker is not supported on Intel-based Macs, where this setting
2236 is ignored and MSAL will fall back to non-broker.
2237 This parameter defaults to None, which means MSAL will not utilize a broker.
2239 New in MSAL Python 1.31.0.
2241 :param boolean enable_broker_on_linux:
2242 This setting is only effective if your app is running on Linux, including WSL.
2243 This parameter defaults to None, which means MSAL will not utilize a broker.
2245 New in MSAL Python 1.33.0.
2247 :param boolean enable_broker_on_wsl:
2248 This setting is only effective if your app is running on WSL.
2249 This parameter defaults to None, which means MSAL will not utilize a broker.
2251 New in MSAL Python 1.33.0.
2252 """
2253 if client_credential is not None:
2254 raise ValueError("Public Client should not possess credentials")
2256 self._enable_broker = bool(
2257 enable_broker_on_windows and sys.platform == "win32"
2258 or enable_broker_on_mac and sys.platform == "darwin"
2259 or enable_broker_on_linux and sys.platform == "linux"
2260 or enable_broker_on_wsl and is_wsl()
2261 )
2263 super(PublicClientApplication, self).__init__(
2264 client_id, client_credential=None, **kwargs)
2266 def acquire_token_interactive(
2267 self,
2268 scopes, # type: list[str]
2269 prompt=None,
2270 login_hint=None, # type: Optional[str]
2271 domain_hint=None, # type: Optional[str]
2272 claims_challenge=None,
2273 timeout=None,
2274 port=None,
2275 extra_scopes_to_consent=None,
2276 max_age=None,
2277 parent_window_handle=None,
2278 on_before_launching_ui=None,
2279 auth_scheme=None,
2280 **kwargs):
2281 """Acquire token interactively i.e. via a local browser.
2283 Prerequisite: In Azure Portal, configure the Redirect URI of your
2284 "Mobile and Desktop application" as ``http://localhost``.
2285 If you opts in to use broker during ``PublicClientApplication`` creation,
2286 your app also need this Redirect URI:
2287 ``ms-appx-web://Microsoft.AAD.BrokerPlugin/YOUR_CLIENT_ID``
2289 :param list scopes:
2290 It is a list of case-sensitive strings.
2291 :param str prompt:
2292 By default, no prompt value will be sent, not even string ``"none"``.
2293 You will have to specify a value explicitly.
2294 Its valid values are the constants defined in
2295 :class:`Prompt <msal.Prompt>`.
2296 :param str login_hint:
2297 Optional. Identifier of the user. Generally a User Principal Name (UPN).
2298 :param domain_hint:
2299 Can be one of "consumers" or "organizations" or your tenant domain "contoso.com".
2300 If included, it will skip the email-based discovery process that user goes
2301 through on the sign-in page, leading to a slightly more streamlined user experience.
2302 More information on possible values available in
2303 `Auth Code Flow doc <https://docs.microsoft.com/en-us/azure/active-directory/develop/v2-oauth2-auth-code-flow#request-an-authorization-code>`_ and
2304 `domain_hint doc <https://docs.microsoft.com/en-us/openspecs/windows_protocols/ms-oapx/86fb452d-e34a-494e-ac61-e526e263b6d8>`_.
2306 :param claims_challenge:
2307 The claims_challenge parameter requests specific claims requested by the resource provider
2308 in the form of a claims_challenge directive in the www-authenticate header to be
2309 returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
2310 It is a string of a JSON object which contains lists of claims being requested from these locations.
2312 :param int timeout:
2313 This method will block the current thread.
2314 This parameter specifies the timeout value in seconds.
2315 Default value ``None`` means wait indefinitely.
2317 :param int port:
2318 The port to be used to listen to an incoming auth response.
2319 By default we will use a system-allocated port.
2320 (The rest of the redirect_uri is hard coded as ``http://localhost``.)
2322 :param list extra_scopes_to_consent:
2323 "Extra scopes to consent" is a concept only available in Microsoft Entra.
2324 It refers to other resources you might want to prompt to consent for,
2325 in the same interaction, but for which you won't get back a
2326 token for in this particular operation.
2328 :param int max_age:
2329 OPTIONAL. Maximum Authentication Age.
2330 Specifies the allowable elapsed time in seconds
2331 since the last time the End-User was actively authenticated.
2332 If the elapsed time is greater than this value,
2333 Microsoft identity platform will actively re-authenticate the End-User.
2335 MSAL Python will also automatically validate the auth_time in ID token.
2337 New in version 1.15.
2339 :param int parent_window_handle:
2340 OPTIONAL.
2342 * If your app does not opt in to use broker,
2343 you do not need to provide a ``parent_window_handle`` here.
2345 * If your app opts in to use broker,
2346 ``parent_window_handle`` is required.
2348 - If your app is a GUI app running on Windows or Mac system,
2349 you are required to also provide its window handle,
2350 so that the sign-in window will pop up on top of your window.
2351 - If your app is a console app running on Windows or Mac system,
2352 you can use a placeholder
2353 ``PublicClientApplication.CONSOLE_WINDOW_HANDLE``.
2355 Most Python scripts are console apps.
2357 New in version 1.20.0.
2359 :param function on_before_launching_ui:
2360 A callback with the form of
2361 ``lambda ui="xyz", **kwargs: print("A {} will be launched".format(ui))``,
2362 where ``ui`` will be either "browser" or "broker".
2363 You can use it to inform your end user to expect a pop-up window.
2365 New in version 1.20.0.
2367 :param object auth_scheme:
2368 You can provide an ``msal.auth_scheme.PopAuthScheme`` object
2369 so that MSAL will get a Proof-of-Possession (POP) token for you.
2371 New in version 1.26.0.
2373 :return:
2374 - A dict containing no "error" key,
2375 and typically contains an "access_token" key.
2376 - A dict containing an "error" key, when token refresh failed.
2377 """
2378 data = kwargs.pop("data", {})
2379 enable_msa_passthrough = kwargs.pop( # MUST remove it from kwargs
2380 "enable_msa_passthrough", # Keep it as a hidden param, for now.
2381 # OPTIONAL. MSA-Passthrough is a legacy configuration,
2382 # needed by a small amount of Microsoft first-party apps,
2383 # which would login MSA accounts via ".../organizations" authority.
2384 # If you app belongs to this category, AND you are enabling broker,
2385 # you would want to enable this flag. Default value is False.
2386 # More background of MSA-PT is available from this internal docs:
2387 # https://microsoft.sharepoint.com/:w:/t/Identity-DevEx/EatIUauX3c9Ctw1l7AQ6iM8B5CeBZxc58eoQCE0IuZ0VFw?e=tgc3jP&CID=39c853be-76ea-79d7-ee73-f1b2706ede05
2388 False
2389 ) and data.get("token_type") != "ssh-cert" # Work around a known issue as of PyMsalRuntime 0.8
2390 self._validate_ssh_cert_input_data(data)
2391 is_ssh_cert_or_pop_request = _is_ssh_cert_or_pop_request(data.get("token_type"), auth_scheme)
2393 if not on_before_launching_ui:
2394 on_before_launching_ui = lambda **kwargs: None
2395 if _is_running_in_cloud_shell() and prompt == "none":
2396 # Note: _acquire_token_by_cloud_shell() is always silent,
2397 # so we would not fire on_before_launching_ui()
2398 return self._acquire_token_by_cloud_shell(scopes, data=data)
2399 claims = _merge_claims_challenge_and_capabilities(
2400 self._client_capabilities, claims_challenge)
2401 if self._enable_broker and (sys.platform in ("win32", "darwin") or not is_ssh_cert_or_pop_request):
2402 if parent_window_handle is None:
2403 raise ValueError(
2404 "parent_window_handle is required when you opted into using broker. "
2405 "You need to provide the window handle of your GUI application, "
2406 "or use msal.PublicClientApplication.CONSOLE_WINDOW_HANDLE "
2407 "when and only when your application is a console app.")
2408 if extra_scopes_to_consent:
2409 logger.warning(
2410 "Ignoring parameter extra_scopes_to_consent, "
2411 "which is not supported by broker")
2412 response = self._acquire_token_interactive_via_broker(
2413 scopes,
2414 parent_window_handle,
2415 enable_msa_passthrough,
2416 claims,
2417 data,
2418 on_before_launching_ui,
2419 auth_scheme,
2420 prompt=prompt,
2421 login_hint=login_hint,
2422 max_age=max_age,
2423 )
2424 return self._process_broker_response(response, scopes, data)
2426 if isinstance(auth_scheme, msal.auth_scheme.PopAuthScheme) and sys.platform == "linux":
2427 raise ValueError("POP is not supported on Linux")
2428 elif auth_scheme:
2429 raise ValueError(self._AUTH_SCHEME_UNSUPPORTED)
2430 on_before_launching_ui(ui="browser")
2431 telemetry_context = self._build_telemetry_context(
2432 self.ACQUIRE_TOKEN_INTERACTIVE)
2433 response = _clean_up(self.client.obtain_token_by_browser(
2434 scope=self._decorate_scope(scopes) if scopes else None,
2435 extra_scope_to_consent=extra_scopes_to_consent,
2436 redirect_uri="http://localhost:{port}".format(
2437 # Hardcode the host, for now. AAD portal rejects 127.0.0.1 anyway
2438 port=port or 0),
2439 prompt=prompt,
2440 login_hint=login_hint,
2441 max_age=max_age,
2442 timeout=timeout,
2443 auth_params={
2444 "claims": claims,
2445 "domain_hint": domain_hint,
2446 },
2447 data=dict(data, claims=claims),
2448 headers=telemetry_context.generate_headers(),
2449 browser_name=_preferred_browser(),
2450 **kwargs))
2451 if "access_token" in response:
2452 response[self._TOKEN_SOURCE] = self._TOKEN_SOURCE_IDP
2453 telemetry_context.update_telemetry(response)
2454 return response
2456 def _acquire_token_interactive_via_broker(
2457 self,
2458 scopes, # type: list[str]
2459 parent_window_handle, # type: int
2460 enable_msa_passthrough, # type: boolean
2461 claims, # type: str
2462 data, # type: dict
2463 on_before_launching_ui, # type: callable
2464 auth_scheme, # type: object
2465 prompt=None,
2466 login_hint=None, # type: Optional[str]
2467 max_age=None,
2468 **kwargs):
2469 from .broker import _signin_interactively, _signin_silently, _acquire_token_silently
2470 if "welcome_template" in kwargs:
2471 logger.debug(kwargs["welcome_template"]) # Experimental
2472 authority = "https://{}/{}".format(
2473 self.authority.instance, self.authority.tenant)
2474 validate_authority = "no" if (
2475 self.authority._is_known_to_developer
2476 or self._instance_discovery is False) else None
2477 # Calls different broker methods to mimic the OIDC behaviors
2478 if login_hint and prompt != "select_account": # OIDC prompts when the user did not sign in
2479 accounts = self.get_accounts(username=login_hint)
2480 if len(accounts) == 1: # Unambiguously proceed with this account
2481 logger.debug("Calling broker._acquire_token_silently()")
2482 response = _acquire_token_silently( # When it works, it bypasses prompt
2483 authority,
2484 self.client_id,
2485 accounts[0]["local_account_id"],
2486 scopes,
2487 claims=claims,
2488 auth_scheme=auth_scheme,
2489 **data)
2490 if response and "error" not in response:
2491 return response
2492 # login_hint undecisive or not exists
2493 if prompt == "none" or not prompt: # Must/Can attempt _signin_silently()
2494 logger.debug("Calling broker._signin_silently()")
2495 response = _signin_silently( # Unlike OIDC, it doesn't honor login_hint
2496 authority, self.client_id, scopes,
2497 validateAuthority=validate_authority,
2498 claims=claims,
2499 max_age=max_age,
2500 enable_msa_pt=enable_msa_passthrough,
2501 auth_scheme=auth_scheme,
2502 **data)
2503 is_wrong_account = bool(
2504 # _signin_silently() only gets tokens for default account,
2505 # but this seems to have been fixed in PyMsalRuntime 0.11.2
2506 "access_token" in response and login_hint
2507 and login_hint != response.get(
2508 "id_token_claims", {}).get("preferred_username"))
2509 wrong_account_error_message = (
2510 'prompt="none" will not work for login_hint="non-default-user"')
2511 if is_wrong_account:
2512 logger.debug(wrong_account_error_message)
2513 if prompt == "none":
2514 return response if not is_wrong_account else {
2515 "error": "broker_error",
2516 "error_description": wrong_account_error_message,
2517 }
2518 else:
2519 assert bool(prompt) is False
2520 from pymsalruntime import Response_Status
2521 recoverable_errors = frozenset([
2522 Response_Status.Status_AccountUnusable,
2523 Response_Status.Status_InteractionRequired,
2524 ])
2525 if is_wrong_account or "error" in response and response.get(
2526 "_broker_status") in recoverable_errors:
2527 pass # It will fall back to the _signin_interactively()
2528 else:
2529 return response
2531 logger.debug("Falls back to broker._signin_interactively()")
2532 on_before_launching_ui(ui="broker")
2533 return _signin_interactively(
2534 authority, self.client_id, scopes,
2535 None if parent_window_handle is self.CONSOLE_WINDOW_HANDLE
2536 else parent_window_handle,
2537 validateAuthority=validate_authority,
2538 login_hint=login_hint,
2539 prompt=prompt,
2540 claims=claims,
2541 max_age=max_age,
2542 enable_msa_pt=enable_msa_passthrough,
2543 auth_scheme=auth_scheme,
2544 **data)
2546 def initiate_device_flow(self, scopes=None, *, claims_challenge=None, **kwargs):
2547 """Initiate a Device Flow instance,
2548 which will be used in :func:`~acquire_token_by_device_flow`.
2550 :param list[str] scopes:
2551 Scopes requested to access a protected API (a resource).
2552 :return: A dict representing a newly created Device Flow object.
2554 - A successful response would contain "user_code" key, among others
2555 - an error response would contain some other readable key/value pairs.
2556 """
2557 correlation_id = msal.telemetry._get_new_correlation_id()
2558 flow = self.client.initiate_device_flow(
2559 scope=self._decorate_scope(scopes or []),
2560 headers={msal.telemetry.CLIENT_REQUEST_ID: correlation_id},
2561 data={"claims": _merge_claims_challenge_and_capabilities(
2562 self._client_capabilities, claims_challenge)},
2563 **kwargs)
2564 flow[self.DEVICE_FLOW_CORRELATION_ID] = correlation_id
2565 return flow
2567 def acquire_token_by_device_flow(self, flow, claims_challenge=None, **kwargs):
2568 """Obtain token by a device flow object, with customizable polling effect.
2570 :param dict flow:
2571 A dict previously generated by :func:`~initiate_device_flow`.
2572 By default, this method's polling effect will block current thread.
2573 You can abort the polling loop at any time,
2574 by changing the value of the flow's "expires_at" key to 0.
2575 :param claims_challenge:
2576 The claims_challenge parameter requests specific claims requested by the resource provider
2577 in the form of a claims_challenge directive in the www-authenticate header to be
2578 returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
2579 It is a string of a JSON object which contains lists of claims being requested from these locations.
2581 :return: A dict representing the json response from Microsoft Entra:
2583 - A successful response would contain "access_token" key,
2584 - an error response would contain "error" and usually "error_description".
2585 """
2586 telemetry_context = self._build_telemetry_context(
2587 self.ACQUIRE_TOKEN_BY_DEVICE_FLOW_ID,
2588 correlation_id=flow.get(self.DEVICE_FLOW_CORRELATION_ID))
2589 response = _clean_up(self.client.obtain_token_by_device_flow(
2590 flow,
2591 data=dict(
2592 kwargs.pop("data", {}),
2593 code=flow["device_code"], # 2018-10-4 Hack:
2594 # during transition period,
2595 # service seemingly need both device_code and code parameter.
2596 claims=_merge_claims_challenge_and_capabilities(
2597 self._client_capabilities, claims_challenge),
2598 ),
2599 headers=telemetry_context.generate_headers(),
2600 **kwargs))
2601 if "access_token" in response:
2602 response[self._TOKEN_SOURCE] = self._TOKEN_SOURCE_IDP
2603 telemetry_context.update_telemetry(response)
2604 return response
2607class ConfidentialClientApplication(ClientApplication): # server-side web app
2608 """Same as :func:`ClientApplication.__init__`,
2609 except that ``allow_broker`` parameter shall remain ``None``.
2610 """
2612 def acquire_token_for_client(self, scopes, claims_challenge=None, fmi_path=None, forwarded_client_claims=None, **kwargs):
2613 """Acquires token for the current confidential client, not for an end user.
2615 Since MSAL Python 1.23, it will automatically look for token from cache,
2616 and only send request to Identity Provider when cache misses.
2618 :param list[str] scopes: (Required)
2619 Scopes requested to access a protected API (a resource).
2620 :param claims_challenge:
2621 The claims_challenge parameter requests specific claims requested by the resource provider
2622 in the form of a claims_challenge directive in the www-authenticate header to be
2623 returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
2624 It is a string of a JSON object which contains lists of claims being requested from these locations.
2625 :param str fmi_path:
2626 Optional. The Federated Managed Identity (FMI) credential path.
2627 When provided, it is sent as the ``fmi_path`` parameter in the
2628 token request body, and the resulting token is cached separately
2629 so that different FMI paths do not share cached tokens.
2630 Example usage::
2632 result = cca.acquire_token_for_client(
2633 scopes=["api://resource/.default"],
2634 fmi_path="SomeFmiPath/FmiCredentialPath",
2635 )
2636 :param str forwarded_client_claims:
2637 Optional. A JSON string of *client-originated* claims to include in
2638 the token request. Unlike ``claims_challenge`` (server-issued, which
2639 bypasses the cache), tokens acquired with ``forwarded_client_claims``
2640 **are cached** and keyed on the claims value. Send the *same* value on
2641 every request that should share the cached token; omitting or changing
2642 it routes to a different cache entry (a cache miss), so use stable,
2643 non-dynamic values. The value is merged into the standard OAuth
2644 ``claims`` request parameter sent on the wire.
2646 Not to be confused with the constructor ``client_claims`` parameter
2647 (a ``dict`` of extra claims signed into the client-assertion JWT).
2648 :return: A dict representing the json response from Microsoft Entra:
2650 - A successful response would contain "access_token" key,
2651 - an error response would contain "error" and usually "error_description".
2652 """
2653 if kwargs.get("force_refresh"):
2654 raise ValueError( # We choose to disallow force_refresh
2655 "Historically, this method does not support force_refresh behavior. "
2656 )
2657 if fmi_path is not None:
2658 if not isinstance(fmi_path, str):
2659 raise ValueError(
2660 "fmi_path must be a string, got {}".format(type(fmi_path).__name__))
2661 kwargs["data"] = kwargs.get("data", {})
2662 kwargs["data"]["fmi_path"] = fmi_path
2663 if forwarded_client_claims is not None:
2664 # Carry it in the request data so it contributes to the extended
2665 # cache key (different claims => separate cache entries). It is
2666 # merged into the "claims" body parameter in _acquire_token_for_client
2667 # and stripped from the wire body by the oauth2 layer.
2668 kwargs["data"] = kwargs.get("data", {})
2669 _stash_client_claims(forwarded_client_claims, kwargs["data"])
2670 return _clean_up(self._acquire_token_silent_with_error(
2671 scopes, None, claims_challenge=claims_challenge, **kwargs))
2673 def _acquire_token_for_client(
2674 self,
2675 scopes,
2676 refresh_reason,
2677 claims_challenge=None,
2678 **kwargs
2679 ):
2680 if self.authority.tenant.lower() in ["common", "organizations"]:
2681 warnings.warn(
2682 "Using /common or /organizations authority "
2683 "in acquire_token_for_client() is unreliable. "
2684 "Please use a specific tenant instead.", DeprecationWarning)
2685 self._validate_ssh_cert_input_data(kwargs.get("data", {}))
2686 telemetry_context = self._build_telemetry_context(
2687 self.ACQUIRE_TOKEN_FOR_CLIENT_ID, refresh_reason=refresh_reason)
2688 client = self._regional_client or self.client
2689 request_data = kwargs.pop("data", {})
2690 claims = _merge_claims_challenge_and_capabilities(
2691 self._client_capabilities, claims_challenge)
2692 # Client-originated claims (set via forwarded_client_claims=) are merged into the
2693 # same OAuth "claims" parameter and sent on the wire. The raw
2694 # "client_claims" entry stays in request_data so it keys the cache; the
2695 # oauth2 layer drops it from the actual request body.
2696 client_claims = request_data.get("client_claims")
2697 if client_claims:
2698 claims = _merge_claims(claims, client_claims)
2699 response = client.obtain_token_for_client(
2700 scope=scopes, # This grant flow requires no scope decoration
2701 headers=telemetry_context.generate_headers(),
2702 data=dict(request_data, claims=claims),
2703 **kwargs)
2704 telemetry_context.update_telemetry(response)
2705 return response
2707 def remove_tokens_for_client(self):
2708 """Remove all tokens that were previously acquired via
2709 :func:`~acquire_token_for_client()` for the current client."""
2710 for env in [self.authority.instance] + self._get_authority_aliases(
2711 self.authority.instance):
2712 for at in list(self.token_cache.search( # Remove ATs from a snapshot
2713 TokenCache.CredentialType.ACCESS_TOKEN, query={
2714 "client_id": self.client_id,
2715 "environment": env,
2716 "home_account_id": None, # These are mostly app-only tokens
2717 })):
2718 self.token_cache.remove_at(at)
2719 # acquire_token_for_client() obtains no RTs, so we have no RT to remove
2721 def acquire_token_on_behalf_of(self, user_assertion, scopes, claims_challenge=None, forwarded_client_claims=None, **kwargs):
2722 """Acquires token using on-behalf-of (OBO) flow.
2724 The current app is a middle-tier service which was called with a token
2725 representing an end user.
2726 The current app can use such token (a.k.a. a user assertion) to request
2727 another token to access downstream web API, on behalf of that user.
2728 See `detail docs here <https://docs.microsoft.com/en-us/azure/active-directory/develop/v2-oauth2-on-behalf-of-flow>`_ .
2730 The current middle-tier app has no user interaction to obtain consent.
2731 See how to gain consent upfront for your middle-tier app from this article.
2732 https://docs.microsoft.com/en-us/azure/active-directory/develop/v2-oauth2-on-behalf-of-flow#gaining-consent-for-the-middle-tier-application
2734 :param str user_assertion: The incoming token already received by this app
2735 :param list[str] scopes: Scopes required by downstream API (a resource).
2736 :param claims_challenge:
2737 The claims_challenge parameter requests specific claims requested by the resource provider
2738 in the form of a claims_challenge directive in the www-authenticate header to be
2739 returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
2740 It is a string of a JSON object which contains lists of claims being requested from these locations.
2741 :param str forwarded_client_claims:
2742 Optional. A JSON string of *client-originated* claims to include in
2743 the token request. Unlike ``claims_challenge`` (server-issued, which
2744 bypasses the cache), tokens acquired with ``forwarded_client_claims``
2745 **are cached** and keyed on the claims value. Send the *same* value on
2746 every request that should share the cached token; omitting or changing
2747 it routes to a different cache entry (a cache miss), so use stable,
2748 non-dynamic values. The value is merged into the standard OAuth
2749 ``claims`` request parameter sent on the wire.
2751 Not to be confused with the constructor ``client_claims`` parameter
2752 (a ``dict`` of extra claims signed into the client-assertion JWT).
2754 :return: A dict representing the json response from Microsoft Entra:
2756 - A successful response would contain "access_token" key,
2757 - an error response would contain "error" and usually "error_description".
2758 """
2759 telemetry_context = self._build_telemetry_context(
2760 self.ACQUIRE_TOKEN_ON_BEHALF_OF_ID)
2761 _data = kwargs.pop("data", {})
2762 _stash_client_claims(forwarded_client_claims, _data)
2763 # The implementation is NOT based on Token Exchange (RFC 8693)
2764 response = _clean_up(self.client.obtain_token_by_assertion( # bases on assertion RFC 7521
2765 user_assertion,
2766 self.client.GRANT_TYPE_JWT, # IDTs and AAD ATs are all JWTs
2767 scope=self._decorate_scope(scopes), # Decoration is used for:
2768 # 1. Explicitly requesting an RT, without relying on AAD default
2769 # behavior, even though it currently still issues an RT.
2770 # 2. Requesting an IDT (which would otherwise be unavailable)
2771 # so that the calling app could use id_token_claims to implement
2772 # their own cache mapping, which is likely needed in web apps.
2773 data=dict(
2774 _data,
2775 requested_token_use="on_behalf_of",
2776 claims=_merge_claims(
2777 _merge_claims_challenge_and_capabilities(
2778 self._client_capabilities, claims_challenge),
2779 _data.get("client_claims"))),
2780 headers=telemetry_context.generate_headers(),
2781 # TBD: Expose a login_hint (or ccs_routing_hint) param for web app
2782 **kwargs))
2783 if "access_token" in response:
2784 response[self._TOKEN_SOURCE] = self._TOKEN_SOURCE_IDP
2785 telemetry_context.update_telemetry(response)
2786 return response
2788 def acquire_token_by_user_federated_identity_credential(
2789 self, scopes, assertion, username=None, user_object_id=None,
2790 claims_challenge=None, forwarded_client_claims=None, **kwargs):
2791 """Acquires a user-scoped token using the ``user_fic`` grant type.
2793 This method exchanges a federated identity credential (typically an
2794 agent instance token from Leg 2 of the agent identity protocol) for
2795 a user-scoped access token, enabling an agent to act on behalf of
2796 a specific user.
2798 :param list[str] scopes: Scopes required by downstream API (a resource).
2799 :param str assertion:
2800 The federated identity credential token (e.g. the instance token
2801 obtained from Leg 2 of the agent identity flow).
2802 :param str username:
2803 The target user's UPN (User Principal Name).
2804 Mutually exclusive with ``user_object_id``.
2805 :param str user_object_id:
2806 The target user's Object ID.
2807 Mutually exclusive with ``username``.
2808 :param claims_challenge:
2809 The claims_challenge parameter requests specific claims requested by the resource provider
2810 in the form of a claims_challenge directive in the www-authenticate header to be
2811 returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
2812 It is a string of a JSON object which contains lists of claims being requested from these locations.
2813 :param str forwarded_client_claims:
2814 Optional. A JSON string of *client-originated* claims to include in
2815 the token request. Unlike ``claims_challenge`` (server-issued, which
2816 bypasses the cache), tokens acquired with ``forwarded_client_claims``
2817 **are cached** and keyed on the claims value. Send the *same* value on
2818 every request that should share the cached token; omitting or changing
2819 it routes to a different cache entry (a cache miss), so use stable,
2820 non-dynamic values. The value is merged into the standard OAuth
2821 ``claims`` request parameter sent on the wire.
2823 Not to be confused with the constructor ``client_claims`` parameter
2824 (a ``dict`` of extra claims signed into the client-assertion JWT).
2826 :return: A dict representing the json response from Microsoft Entra:
2828 - A successful response would contain "access_token" key,
2829 - an error response would contain "error" and usually "error_description".
2830 """
2831 # Input validation
2832 if not assertion:
2833 raise ValueError("assertion is required and must be non-empty")
2834 if not username and not user_object_id:
2835 raise ValueError(
2836 "Either username or user_object_id must be provided")
2837 if username and user_object_id:
2838 raise ValueError(
2839 "username and user_object_id are mutually exclusive")
2841 telemetry_context = self._build_telemetry_context(
2842 self.ACQUIRE_TOKEN_BY_USER_FIC_ID)
2843 headers = telemetry_context.generate_headers()
2844 if username:
2845 headers["X-AnchorMailbox"] = "upn:{}".format(username)
2846 elif user_object_id:
2847 headers["X-AnchorMailbox"] = "Oid:{}@{}".format(
2848 user_object_id, self.authority.tenant)
2849 _data = kwargs.pop("data", {})
2850 _stash_client_claims(forwarded_client_claims, _data)
2851 response = _clean_up(self.client.obtain_token_by_user_fic(
2852 scope=self._decorate_scope(scopes),
2853 assertion=assertion,
2854 username=username,
2855 user_object_id=user_object_id,
2856 headers=headers,
2857 data=dict(
2858 _data,
2859 claims=_merge_claims(
2860 _merge_claims_challenge_and_capabilities(
2861 self._client_capabilities, claims_challenge),
2862 _data.get("client_claims"))),
2863 **kwargs))
2864 if "access_token" in response:
2865 response[self._TOKEN_SOURCE] = self._TOKEN_SOURCE_IDP
2866 telemetry_context.update_telemetry(response)
2867 return response