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