-
Notifications
You must be signed in to change notification settings - Fork 202
/
Copy pathapplication.py
2327 lines (2090 loc) · 117 KB
/
application.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
import functools
import json
import time
import logging
import sys
import warnings
from threading import Lock
import os
from .oauth2cli import Client, JwtAssertionCreator
from .oauth2cli.oidc import decode_part
from .authority import Authority, WORLD_WIDE
from .mex import send_request as mex_send_request
from .wstrust_request import send_request as wst_send_request
from .wstrust_response import *
from .token_cache import TokenCache, _get_username, _GRANT_TYPE_BROKER
import msal.telemetry
from .region import _detect_region
from .throttled_http_client import ThrottledHttpClient
from .cloudshell import _is_running_in_cloud_shell
# The __init__.py will import this. Not the other way around.
__version__ = "1.28.1" # When releasing, also check and bump our dependencies's versions if needed
logger = logging.getLogger(__name__)
_AUTHORITY_TYPE_CLOUDSHELL = "CLOUDSHELL"
def extract_certs(public_cert_content):
# Parses raw public certificate file contents and returns a list of strings
# Usage: headers = {"x5c": extract_certs(open("my_cert.pem").read())}
public_certificates = re.findall(
r'-----BEGIN CERTIFICATE-----(?P<cert_value>[^-]+)-----END CERTIFICATE-----',
public_cert_content, re.I)
if public_certificates:
return [cert.strip() for cert in public_certificates]
# The public cert tags are not found in the input,
# let's make best effort to exclude a private key pem file.
if "PRIVATE KEY" in public_cert_content:
raise ValueError(
"We expect your public key but detect a private key instead")
return [public_cert_content.strip()]
def _merge_claims_challenge_and_capabilities(capabilities, claims_challenge):
# Represent capabilities as {"access_token": {"xms_cc": {"values": capabilities}}}
# and then merge/add it into incoming claims
if not capabilities:
return claims_challenge
claims_dict = json.loads(claims_challenge) if claims_challenge else {}
for key in ["access_token"]: # We could add "id_token" if we'd decide to
claims_dict.setdefault(key, {}).update(xms_cc={"values": capabilities})
return json.dumps(claims_dict)
def _str2bytes(raw):
# A conversion based on duck-typing rather than six.text_type
try:
return raw.encode(encoding="utf-8")
except:
return raw
def _load_private_key_from_pfx_path(pfx_path, passphrase_bytes):
# Cert concepts https://security.stackexchange.com/a/226758/125264
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.serialization import pkcs12
with open(pfx_path, 'rb') as f:
private_key, cert, _ = pkcs12.load_key_and_certificates( # cryptography 2.5+
# https://cryptography.io/en/latest/hazmat/primitives/asymmetric/serialization/#cryptography.hazmat.primitives.serialization.pkcs12.load_key_and_certificates
f.read(), passphrase_bytes)
sha1_thumbprint = cert.fingerprint(hashes.SHA1()).hex() # cryptography 0.7+
# https://cryptography.io/en/latest/x509/reference/#x-509-certificate-object
return private_key, sha1_thumbprint
def _load_private_key_from_pem_str(private_key_pem_str, passphrase_bytes):
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend
return serialization.load_pem_private_key( # cryptography 0.6+
_str2bytes(private_key_pem_str),
passphrase_bytes,
backend=default_backend(), # It was a required param until 2020
)
def _pii_less_home_account_id(home_account_id):
parts = home_account_id.split(".") # It could contain one or two parts
parts[0] = "********"
return ".".join(parts)
def _clean_up(result):
if isinstance(result, dict):
if "_msalruntime_telemetry" in result or "_msal_python_telemetry" in result:
result["msal_telemetry"] = json.dumps({ # Telemetry as an opaque string
"msalruntime_telemetry": result.get("_msalruntime_telemetry"),
"msal_python_telemetry": result.get("_msal_python_telemetry"),
}, separators=(",", ":"))
return {
k: result[k] for k in result
if k != "refresh_in" # MSAL handled refresh_in, customers need not
and not k.startswith('_') # Skim internal properties
}
return result # It could be None
def _preferred_browser():
"""Register Edge and return a name suitable for subsequent webbrowser.get(...)
when appropriate. Otherwise return None.
"""
# On Linux, only Edge will provide device-based Conditional Access support
if sys.platform != "linux": # On other platforms, we have no browser preference
return None
browser_path = "/usr/bin/microsoft-edge" # Use a full path owned by sys admin
# Note: /usr/bin/microsoft-edge, /usr/bin/microsoft-edge-stable, etc.
# are symlinks that point to the actual binaries which are found under
# /opt/microsoft/msedge/msedge or /opt/microsoft/msedge-beta/msedge.
# Either method can be used to detect an Edge installation.
user_has_no_preference = "BROWSER" not in os.environ
user_wont_mind_edge = "microsoft-edge" in os.environ.get("BROWSER", "") # Note:
# BROWSER could contain "microsoft-edge" or "/path/to/microsoft-edge".
# Python documentation (https://docs.python.org/3/library/webbrowser.html)
# does not document the name being implicitly register,
# so there is no public API to know whether the ENV VAR browser would work.
# Therefore, we would not bother examine the env var browser's type.
# We would just register our own Edge instance.
if (user_has_no_preference or user_wont_mind_edge) and os.path.exists(browser_path):
try:
import webbrowser # Lazy import. Some distro may not have this.
browser_name = "msal-edge" # Avoid popular name "microsoft-edge"
# otherwise `BROWSER="microsoft-edge"; webbrowser.get("microsoft-edge")`
# would return a GenericBrowser instance which won't work.
try:
registration_available = isinstance(
webbrowser.get(browser_name), webbrowser.BackgroundBrowser)
except webbrowser.Error:
registration_available = False
if not registration_available:
logger.debug("Register %s with %s", browser_name, browser_path)
# By registering our own browser instance with our own name,
# rather than populating a process-wide BROWSER enn var,
# this approach does not have side effect on non-MSAL code path.
webbrowser.register( # Even double-register happens to work fine
browser_name, None, webbrowser.BackgroundBrowser(browser_path))
return browser_name
except ImportError:
pass # We may still proceed
return None
class _ClientWithCcsRoutingInfo(Client):
def initiate_auth_code_flow(self, **kwargs):
if kwargs.get("login_hint"): # eSTS could have utilized this as-is, but nope
kwargs["X-AnchorMailbox"] = "UPN:%s" % kwargs["login_hint"]
return super(_ClientWithCcsRoutingInfo, self).initiate_auth_code_flow(
client_info=1, # To be used as CSS Routing info
**kwargs)
def obtain_token_by_auth_code_flow(
self, auth_code_flow, auth_response, **kwargs):
# Note: the obtain_token_by_browser() is also covered by this
assert isinstance(auth_code_flow, dict) and isinstance(auth_response, dict)
headers = kwargs.pop("headers", {})
client_info = json.loads(
decode_part(auth_response["client_info"])
) if auth_response.get("client_info") else {}
if "uid" in client_info and "utid" in client_info:
# Note: The value of X-AnchorMailbox is also case-insensitive
headers["X-AnchorMailbox"] = "Oid:{uid}@{utid}".format(**client_info)
return super(_ClientWithCcsRoutingInfo, self).obtain_token_by_auth_code_flow(
auth_code_flow, auth_response, headers=headers, **kwargs)
def obtain_token_by_username_password(self, username, password, **kwargs):
headers = kwargs.pop("headers", {})
headers["X-AnchorMailbox"] = "upn:{}".format(username)
return super(_ClientWithCcsRoutingInfo, self).obtain_token_by_username_password(
username, password, headers=headers, **kwargs)
class ClientApplication(object):
"""You do not usually directly use this class. Use its subclasses instead:
:class:`PublicClientApplication` and :class:`ConfidentialClientApplication`.
"""
ACQUIRE_TOKEN_SILENT_ID = "84"
ACQUIRE_TOKEN_BY_REFRESH_TOKEN = "85"
ACQUIRE_TOKEN_BY_USERNAME_PASSWORD_ID = "301"
ACQUIRE_TOKEN_ON_BEHALF_OF_ID = "523"
ACQUIRE_TOKEN_BY_DEVICE_FLOW_ID = "622"
ACQUIRE_TOKEN_FOR_CLIENT_ID = "730"
ACQUIRE_TOKEN_BY_AUTHORIZATION_CODE_ID = "832"
ACQUIRE_TOKEN_INTERACTIVE = "169"
GET_ACCOUNTS_ID = "902"
REMOVE_ACCOUNT_ID = "903"
ATTEMPT_REGION_DISCOVERY = True # "TryAutoDetect"
_TOKEN_SOURCE = "token_source"
_TOKEN_SOURCE_IDP = "identity_provider"
_TOKEN_SOURCE_CACHE = "cache"
_TOKEN_SOURCE_BROKER = "broker"
_enable_broker = False
_AUTH_SCHEME_UNSUPPORTED = (
"auth_scheme is currently only available from broker. "
"You can enable broker by following these instructions. "
"https://msal-python.readthedocs.io/en/latest/#publicclientapplication")
def __init__(
self, client_id,
client_credential=None, authority=None, validate_authority=True,
token_cache=None,
http_client=None,
verify=True, proxies=None, timeout=None,
client_claims=None, app_name=None, app_version=None,
client_capabilities=None,
azure_region=None, # Note: We choose to add this param in this base class,
# despite it is currently only needed by ConfidentialClientApplication.
# This way, it holds the same positional param place for PCA,
# when we would eventually want to add this feature to PCA in future.
exclude_scopes=None,
http_cache=None,
instance_discovery=None,
allow_broker=None,
enable_pii_log=None,
oidc_authority=None,
):
"""Create an instance of application.
:param str client_id: Your app has a client_id after you register it on Microsoft Entra admin center.
:param client_credential:
For :class:`PublicClientApplication`, you use `None` here.
For :class:`ConfidentialClientApplication`,
it can be a string containing client secret,
or an X509 certificate container in this form::
{
"private_key": "...-----BEGIN PRIVATE KEY-----... in PEM format",
"thumbprint": "A1B2C3D4E5F6...",
"public_certificate": "...-----BEGIN CERTIFICATE-----... (Optional. See below.)",
"passphrase": "Passphrase if the private_key is encrypted (Optional. Added in version 1.6.0)",
}
MSAL Python requires a "private_key" in PEM format.
If your cert is in a PKCS12 (.pfx) format, you can also
`convert it to PEM and get the thumbprint <https://github.com/Azure/azure-sdk-for-python/blob/07d10639d7e47f4852eaeb74aef5d569db499d6e/sdk/identity/azure-identity/azure/identity/_credentials/certificate.py#L101-L123>`_.
The thumbprint is available in your app's registration in Azure Portal.
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>`_.
*Added in version 0.5.0*:
public_certificate (optional) is public key certificate
which will be sent through 'x5c' JWT header only for
subject name and issuer authentication to support cert auto rolls.
Per `specs <https://tools.ietf.org/html/rfc7515#section-4.1.6>`_,
"the certificate containing
the public key corresponding to the key used to digitally sign the
JWS MUST be the first certificate. This MAY be followed by
additional certificates, with each subsequent certificate being the
one used to certify the previous one."
However, your certificate's issuer may use a different order.
So, if your attempt ends up with an error AADSTS700027 -
"The provided signature value did not match the expected signature value",
you may try use only the leaf cert (in PEM/str format) instead.
*Added in version 1.13.0*:
It can also be a completely pre-signed assertion that you've assembled yourself.
Simply pass a container containing only the key "client_assertion", like this::
{
"client_assertion": "...a JWT with claims aud, exp, iss, jti, nbf, and sub..."
}
.. admonition:: Supporting reading client cerficates from PFX files
*Added in version 1.29.0*:
Feed in a dictionary containing the path to a PFX file::
{
"private_key_pfx_path": "/path/to/your.pfx",
"passphrase": "Passphrase if the private_key is encrypted (Optional. Added in version 1.6.0)",
}
The following command will generate a .pfx file from your .key and .pem file::
openssl pkcs12 -export -out certificate.pfx -inkey privateKey.key -in certificate.pem
:type client_credential: Union[dict, str]
:param dict client_claims:
*Added in version 0.5.0*:
It is a dictionary of extra claims that would be signed by
by this :class:`ConfidentialClientApplication` 's private key.
For example, you can use {"client_ip": "x.x.x.x"}.
You may also override any of the following default claims::
{
"aud": the_token_endpoint,
"iss": self.client_id,
"sub": same_as_issuer,
"exp": now + 10_min,
"iat": now,
"jti": a_random_uuid
}
:param str authority:
A URL that identifies a token authority. It should be of the format
``https://login.microsoftonline.com/your_tenant``
By default, we will use ``https://login.microsoftonline.com/common``
*Changed in version 1.17*: you can also use predefined constant
and a builder like this::
from msal.authority import (
AuthorityBuilder,
AZURE_US_GOVERNMENT, AZURE_CHINA, AZURE_PUBLIC)
my_authority = AuthorityBuilder(AZURE_PUBLIC, "contoso.onmicrosoft.com")
# Now you get an equivalent of
# "https://login.microsoftonline.com/contoso.onmicrosoft.com"
# You can feed such an authority to msal's ClientApplication
from msal import PublicClientApplication
app = PublicClientApplication("my_client_id", authority=my_authority, ...)
:param bool validate_authority: (optional) Turns authority validation
on or off. This parameter default to true.
:param TokenCache token_cache:
Sets the token cache used by this ClientApplication instance.
By default, an in-memory cache will be created and used.
:param http_client: (optional)
Your implementation of abstract class HttpClient <msal.oauth2cli.http.http_client>
Defaults to a requests session instance.
Since MSAL 1.11.0, the default session would be configured
to attempt one retry on connection error.
If you are providing your own http_client,
it will be your http_client's duty to decide whether to perform retry.
:param verify: (optional)
It will be passed to the
`verify parameter in the underlying requests library
<http://docs.python-requests.org/en/v2.9.1/user/advanced/#ssl-cert-verification>`_
This does not apply if you have chosen to pass your own Http client
:param proxies: (optional)
It will be passed to the
`proxies parameter in the underlying requests library
<http://docs.python-requests.org/en/v2.9.1/user/advanced/#proxies>`_
This does not apply if you have chosen to pass your own Http client
:param timeout: (optional)
It will be passed to the
`timeout parameter in the underlying requests library
<http://docs.python-requests.org/en/v2.9.1/user/advanced/#timeouts>`_
This does not apply if you have chosen to pass your own Http client
:param app_name: (optional)
You can provide your application name for Microsoft telemetry purposes.
Default value is None, means it will not be passed to Microsoft.
:param app_version: (optional)
You can provide your application version for Microsoft telemetry purposes.
Default value is None, means it will not be passed to Microsoft.
:param list[str] client_capabilities: (optional)
Allows configuration of one or more client capabilities, e.g. ["CP1"].
Client capability is meant to inform the Microsoft identity platform
(STS) what this client is capable for,
so STS can decide to turn on certain features.
For example, if client is capable to handle *claims challenge*,
STS can then issue CAE access tokens to resources
knowing when the resource emits *claims challenge*
the client will be capable to handle.
Implementation details:
Client capability is implemented using "claims" parameter on the wire,
for now.
MSAL will combine them into
`claims parameter <https://openid.net/specs/openid-connect-core-1_0-final.html#ClaimsParameter>`_
which you will later provide via one of the acquire-token request.
:param str azure_region: (optional)
Instructs MSAL to use the Entra regional token service. This legacy feature is only available to
first-party applications. Only ``acquire_token_for_client()`` is supported.
Supports 3 values:
``azure_region=None`` - meaning no region is used. This is the default value.
``azure_region="some_region"`` - meaning the specified region is used.
``azure_region=True`` - meaning MSAL will try to auto-detect the region. This is not recommended.
.. note::
Region auto-discovery has been tested on VMs and on Azure Functions. It is unreliable.
Applications using this option should configure a short timeout.
For more details and for the values of the region string
see https://learn.microsoft.com/entra/msal/dotnet/resources/region-discovery-troubleshooting
New in version 1.12.0.
:param list[str] exclude_scopes: (optional)
Historically MSAL hardcodes `offline_access` scope,
which would allow your app to have prolonged access to user's data.
If that is unnecessary or undesirable for your app,
now you can use this parameter to supply an exclusion list of scopes,
such as ``exclude_scopes = ["offline_access"]``.
:param dict http_cache:
MSAL has long been caching tokens in the ``token_cache``.
Recently, MSAL also introduced a concept of ``http_cache``,
by automatically caching some finite amount of non-token http responses,
so that *long-lived*
``PublicClientApplication`` and ``ConfidentialClientApplication``
would be more performant and responsive in some situations.
This ``http_cache`` parameter accepts any dict-like object.
If not provided, MSAL will use an in-memory dict.
If your app is a command-line app (CLI),
you would want to persist your http_cache across different CLI runs.
The following recipe shows a way to do so::
# Just add the following lines at the beginning of your CLI script
import sys, atexit, pickle
http_cache_filename = sys.argv[0] + ".http_cache"
try:
with open(http_cache_filename, "rb") as f:
persisted_http_cache = pickle.load(f) # Take a snapshot
except (
FileNotFoundError, # Or IOError in Python 2
pickle.UnpicklingError, # A corrupted http cache file
):
persisted_http_cache = {} # Recover by starting afresh
atexit.register(lambda: pickle.dump(
# When exit, flush it back to the file.
# It may occasionally overwrite another process's concurrent write,
# but that is fine. Subsequent runs will reach eventual consistency.
persisted_http_cache, open(http_cache_file, "wb")))
# And then you can implement your app as you normally would
app = msal.PublicClientApplication(
"your_client_id",
...,
http_cache=persisted_http_cache, # Utilize persisted_http_cache
...,
#token_cache=..., # You may combine the old token_cache trick
# Please refer to token_cache recipe at
# https://msal-python.readthedocs.io/en/latest/#msal.SerializableTokenCache
)
app.acquire_token_interactive(["your", "scope"], ...)
Content inside ``http_cache`` are cheap to obtain.
There is no need to share them among different apps.
Content inside ``http_cache`` will contain no tokens nor
Personally Identifiable Information (PII). Encryption is unnecessary.
New in version 1.16.0.
:param boolean instance_discovery:
Historically, MSAL would connect to a central endpoint located at
``https://login.microsoftonline.com`` to acquire some metadata,
especially when using an unfamiliar authority.
This behavior is known as Instance Discovery.
This parameter defaults to None, which enables the Instance Discovery.
If you know some authorities which you allow MSAL to operate with as-is,
without involving any Instance Discovery, the recommended pattern is::
known_authorities = frozenset([ # Treat your known authorities as const
"https://contoso.com/adfs", "https://login.azs/foo"])
...
authority = "https://contoso.com/adfs" # Assuming your app will use this
app1 = PublicClientApplication(
"client_id",
authority=authority,
# Conditionally disable Instance Discovery for known authorities
instance_discovery=authority not in known_authorities,
)
If you do not know some authorities beforehand,
yet still want MSAL to accept any authority that you will provide,
you can use a ``False`` to unconditionally disable Instance Discovery.
New in version 1.19.0.
:param boolean allow_broker:
Deprecated. Please use ``enable_broker_on_windows`` instead.
:param boolean enable_pii_log:
When enabled, logs may include PII (Personal Identifiable Information).
This can be useful in troubleshooting broker behaviors.
The default behavior is False.
New in version 1.24.0.
:param str oidc_authority:
*Added in version 1.28.0*:
It is a URL that identifies an OpenID Connect (OIDC) authority of
the format ``https://contoso.com/tenant``.
MSAL will append ".well-known/openid-configuration" to the authority
and retrieve the OIDC metadata from there, to figure out the endpoints.
Note: Broker will NOT be used for OIDC authority.
"""
self.client_id = client_id
self.client_credential = client_credential
self.client_claims = client_claims
self._client_capabilities = client_capabilities
self._instance_discovery = instance_discovery
if exclude_scopes and not isinstance(exclude_scopes, list):
raise ValueError(
"Invalid exclude_scopes={}. It need to be a list of strings.".format(
repr(exclude_scopes)))
self._exclude_scopes = frozenset(exclude_scopes or [])
if "openid" in self._exclude_scopes:
raise ValueError(
'Invalid exclude_scopes={}. You can not opt out "openid" scope'.format(
repr(exclude_scopes)))
if http_client:
self.http_client = http_client
else:
import requests # Lazy load
self.http_client = requests.Session()
self.http_client.verify = verify
self.http_client.proxies = proxies
# Requests, does not support session - wide timeout
# But you can patch that (https://github.com/psf/requests/issues/3341):
self.http_client.request = functools.partial(
self.http_client.request, timeout=timeout)
# Enable a minimal retry. Better than nothing.
# https://github.com/psf/requests/blob/v2.25.1/requests/adapters.py#L94-L108
a = requests.adapters.HTTPAdapter(max_retries=1)
self.http_client.mount("http://", a)
self.http_client.mount("https://", a)
self.http_client = ThrottledHttpClient(
self.http_client,
{} if http_cache is None else http_cache, # Default to an in-memory dict
)
self.app_name = app_name
self.app_version = app_version
# Here the self.authority will not be the same type as authority in input
if oidc_authority and authority:
raise ValueError("You can not provide both authority and oidc_authority")
try:
authority_to_use = authority or "https://{}/common/".format(WORLD_WIDE)
self.authority = Authority(
authority_to_use,
self.http_client,
validate_authority=validate_authority,
instance_discovery=self._instance_discovery,
oidc_authority_url=oidc_authority,
)
except ValueError: # Those are explicit authority validation errors
raise
except Exception: # The rest are typically connection errors
if validate_authority and azure_region and not oidc_authority:
# Since caller opts in to use region, here we tolerate connection
# errors happened during authority validation at non-region endpoint
self.authority = Authority(
authority_to_use,
self.http_client,
instance_discovery=False,
)
else:
raise
self._decide_broker(allow_broker, enable_pii_log)
self.token_cache = token_cache or TokenCache()
self._region_configured = azure_region
self._region_detected = None
self.client, self._regional_client = self._build_client(
client_credential, self.authority)
self.authority_groups = None
self._telemetry_buffer = {}
self._telemetry_lock = Lock()
def _decide_broker(self, allow_broker, enable_pii_log):
is_confidential_app = self.client_credential or isinstance(
self, ConfidentialClientApplication)
if is_confidential_app and allow_broker:
raise ValueError("allow_broker=True is only supported in PublicClientApplication")
# Historically, we chose to support ClientApplication("client_id", allow_broker=True)
if allow_broker:
warnings.warn(
"allow_broker is deprecated. "
"Please use PublicClientApplication(..., enable_broker_on_windows=True)",
DeprecationWarning)
self._enable_broker = self._enable_broker or (
# When we started the broker project on Windows platform,
# the allow_broker was meant to be cross-platform. Now we realize
# that other platforms have different redirect_uri requirements,
# so the old allow_broker is deprecated and will only for Windows.
allow_broker and sys.platform == "win32")
if (self._enable_broker and not is_confidential_app
and not self.authority.is_adfs and not self.authority._is_b2c):
try:
from . import broker # Trigger Broker's initialization
if enable_pii_log:
broker._enable_pii_log()
except RuntimeError:
self._enable_broker = False
logger.exception(
"Broker is unavailable on this platform. "
"We will fallback to non-broker.")
logger.debug("Broker enabled? %s", self._enable_broker)
def is_pop_supported(self):
"""Returns True if this client supports Proof-of-Possession Access Token."""
return self._enable_broker
def _decorate_scope(
self, scopes,
reserved_scope=frozenset(['openid', 'profile', 'offline_access'])):
if not isinstance(scopes, (list, set, tuple)):
raise ValueError("The input scopes should be a list, tuple, or set")
scope_set = set(scopes) # Input scopes is typically a list. Copy it to a set.
if scope_set & reserved_scope:
# These scopes are reserved for the API to provide good experience.
# We could make the developer pass these and then if they do they will
# come back asking why they don't see refresh token or user information.
raise ValueError(
"API does not accept {} value as user-provided scopes".format(
reserved_scope))
# client_id can also be used as a scope in B2C
decorated = scope_set | reserved_scope
decorated -= self._exclude_scopes
return list(decorated)
def _build_telemetry_context(
self, api_id, correlation_id=None, refresh_reason=None):
return msal.telemetry._TelemetryContext(
self._telemetry_buffer, self._telemetry_lock, api_id,
correlation_id=correlation_id, refresh_reason=refresh_reason)
def _get_regional_authority(self, central_authority):
if not self._region_configured: # User did not opt-in to ESTS-R
return None # Short circuit to completely bypass region detection
self._region_detected = self._region_detected or _detect_region(
self.http_client if self._region_configured is not None else None)
if (self._region_configured != self.ATTEMPT_REGION_DISCOVERY
and self._region_configured != self._region_detected):
logger.warning('Region configured ({}) != region detected ({})'.format(
repr(self._region_configured), repr(self._region_detected)))
region_to_use = (
self._region_detected
if self._region_configured == self.ATTEMPT_REGION_DISCOVERY
else self._region_configured) # It will retain the None i.e. opted out
logger.debug('Region to be used: {}'.format(repr(region_to_use)))
if region_to_use:
regional_host = ("{}.login.microsoft.com".format(region_to_use)
if central_authority.instance in (
# The list came from point 3 of the algorithm section in this internal doc
# https://identitydivision.visualstudio.com/DevEx/_git/AuthLibrariesApiReview?path=/PinAuthToRegion/AAD%20SDK%20Proposal%20to%20Pin%20Auth%20to%20region.md&anchor=algorithm&_a=preview
"login.microsoftonline.com",
"login.microsoft.com",
"login.windows.net",
"sts.windows.net",
)
else "{}.{}".format(region_to_use, central_authority.instance))
return Authority( # The central_authority has already been validated
"https://{}/{}".format(regional_host, central_authority.tenant),
self.http_client,
instance_discovery=False,
)
return None
def _build_client(self, client_credential, authority, skip_regional_client=False):
client_assertion = None
client_assertion_type = None
default_headers = {
"x-client-sku": "MSAL.Python", "x-client-ver": __version__,
"x-client-os": sys.platform,
"x-ms-lib-capability": "retry-after, h429",
}
if self.app_name:
default_headers['x-app-name'] = self.app_name
if self.app_version:
default_headers['x-app-ver'] = self.app_version
default_body = {"client_info": 1}
if isinstance(client_credential, dict):
client_assertion_type = Client.CLIENT_ASSERTION_TYPE_JWT
# Use client_credential.get("...") rather than "..." in client_credential
# so that we can ignore an empty string came from an empty ENV VAR.
if client_credential.get("client_assertion"):
client_assertion = client_credential['client_assertion']
else:
headers = {}
if client_credential.get('public_certificate'):
headers["x5c"] = extract_certs(client_credential['public_certificate'])
passphrase_bytes = _str2bytes(
client_credential["passphrase"]
) if client_credential.get("passphrase") else None
if client_credential.get("private_key_pfx_path"):
private_key, sha1_thumbprint = _load_private_key_from_pfx_path(
client_credential["private_key_pfx_path"], passphrase_bytes)
elif (
client_credential.get("private_key") # PEM blob
and client_credential.get("thumbprint")):
sha1_thumbprint = client_credential["thumbprint"]
if passphrase_bytes:
private_key = _load_private_key_from_pem_str(
client_credential['private_key'], passphrase_bytes)
else: # PEM without passphrase
private_key = client_credential['private_key']
else:
raise ValueError(
"client_credential needs to follow this format "
"https://msal-python.readthedocs.io/en/latest/#msal.ClientApplication.params.client_credential")
assertion = JwtAssertionCreator(
private_key, algorithm="RS256",
sha1_thumbprint=sha1_thumbprint, headers=headers)
client_assertion = assertion.create_regenerative_assertion(
audience=authority.token_endpoint, issuer=self.client_id,
additional_claims=self.client_claims or {})
else:
default_body['client_secret'] = client_credential
central_configuration = {
"authorization_endpoint": authority.authorization_endpoint,
"token_endpoint": authority.token_endpoint,
"device_authorization_endpoint": authority.device_authorization_endpoint,
}
central_client = _ClientWithCcsRoutingInfo(
central_configuration,
self.client_id,
http_client=self.http_client,
default_headers=default_headers,
default_body=default_body,
client_assertion=client_assertion,
client_assertion_type=client_assertion_type,
on_obtaining_tokens=lambda event: self.token_cache.add(dict(
event, environment=authority.instance)),
on_removing_rt=self.token_cache.remove_rt,
on_updating_rt=self.token_cache.update_rt)
regional_client = None
if (client_credential # Currently regional endpoint only serves some CCA flows
and not skip_regional_client):
regional_authority = self._get_regional_authority(authority)
if regional_authority:
regional_configuration = {
"authorization_endpoint": regional_authority.authorization_endpoint,
"token_endpoint": regional_authority.token_endpoint,
"device_authorization_endpoint":
regional_authority.device_authorization_endpoint,
}
regional_client = _ClientWithCcsRoutingInfo(
regional_configuration,
self.client_id,
http_client=self.http_client,
default_headers=default_headers,
default_body=default_body,
client_assertion=client_assertion,
client_assertion_type=client_assertion_type,
on_obtaining_tokens=lambda event: self.token_cache.add(dict(
event, environment=authority.instance)),
on_removing_rt=self.token_cache.remove_rt,
on_updating_rt=self.token_cache.update_rt)
return central_client, regional_client
def initiate_auth_code_flow(
self,
scopes, # type: list[str]
redirect_uri=None,
state=None, # Recommended by OAuth2 for CSRF protection
prompt=None,
login_hint=None, # type: Optional[str]
domain_hint=None, # type: Optional[str]
claims_challenge=None,
max_age=None,
response_mode=None, # type: Optional[str]
):
"""Initiate an auth code flow.
Later when the response reaches your redirect_uri,
you can use :func:`~acquire_token_by_auth_code_flow()`
to complete the authentication/authorization.
:param list scopes:
It is a list of case-sensitive strings.
:param str redirect_uri:
Optional. If not specified, server will use the pre-registered one.
:param str state:
An opaque value used by the client to
maintain state between the request and callback.
If absent, this library will automatically generate one internally.
:param str prompt:
By default, no prompt value will be sent, not even string ``"none"``.
You will have to specify a value explicitly.
Its valid values are the constants defined in
:class:`Prompt <msal.Prompt>`.
:param str login_hint:
Optional. Identifier of the user. Generally a User Principal Name (UPN).
:param domain_hint:
Can be one of "consumers" or "organizations" or your tenant domain "contoso.com".
If included, it will skip the email-based discovery process that user goes
through on the sign-in page, leading to a slightly more streamlined user experience.
More information on possible values available in
`Auth Code Flow doc <https://docs.microsoft.com/en-us/azure/active-directory/develop/v2-oauth2-auth-code-flow#request-an-authorization-code>`_ and
`domain_hint doc <https://docs.microsoft.com/en-us/openspecs/windows_protocols/ms-oapx/86fb452d-e34a-494e-ac61-e526e263b6d8>`_.
:param int max_age:
OPTIONAL. Maximum Authentication Age.
Specifies the allowable elapsed time in seconds
since the last time the End-User was actively authenticated.
If the elapsed time is greater than this value,
Microsoft identity platform will actively re-authenticate the End-User.
MSAL Python will also automatically validate the auth_time in ID token.
New in version 1.15.
:param str response_mode:
OPTIONAL. Specifies the method with which response parameters should be returned.
The default value is equivalent to ``query``, which is still secure enough in MSAL Python
(because MSAL Python does not transfer tokens via query parameter in the first place).
For even better security, we recommend using the value ``form_post``.
In "form_post" mode, response parameters
will be encoded as HTML form values that are transmitted via the HTTP POST method and
encoded in the body using the application/x-www-form-urlencoded format.
Valid values can be either "form_post" for HTTP POST to callback URI or
"query" (the default) for HTTP GET with parameters encoded in query string.
More information on possible values
`here <https://openid.net/specs/oauth-v2-multiple-response-types-1_0.html#ResponseModes>`
and `here <https://openid.net/specs/oauth-v2-form-post-response-mode-1_0.html#FormPostResponseMode>`
:return:
The auth code flow. It is a dict in this form::
{
"auth_uri": "https://...", // Guide user to visit this
"state": "...", // You may choose to verify it by yourself,
// or just let acquire_token_by_auth_code_flow()
// do that for you.
"...": "...", // Everything else are reserved and internal
}
The caller is expected to:
1. somehow store this content, typically inside the current session,
2. guide the end user (i.e. resource owner) to visit that auth_uri,
3. and then relay this dict and subsequent auth response to
:func:`~acquire_token_by_auth_code_flow()`.
"""
client = _ClientWithCcsRoutingInfo(
{"authorization_endpoint": self.authority.authorization_endpoint},
self.client_id,
http_client=self.http_client)
flow = client.initiate_auth_code_flow(
redirect_uri=redirect_uri, state=state, login_hint=login_hint,
prompt=prompt,
scope=self._decorate_scope(scopes),
domain_hint=domain_hint,
claims=_merge_claims_challenge_and_capabilities(
self._client_capabilities, claims_challenge),
max_age=max_age,
response_mode=response_mode,
)
flow["claims_challenge"] = claims_challenge
return flow
def get_authorization_request_url(
self,
scopes, # type: list[str]
login_hint=None, # type: Optional[str]
state=None, # Recommended by OAuth2 for CSRF protection
redirect_uri=None,
response_type="code", # Could be "token" if you use Implicit Grant
prompt=None,
nonce=None,
domain_hint=None, # type: Optional[str]
claims_challenge=None,
**kwargs):
"""Constructs a URL for you to start a Authorization Code Grant.
:param list[str] scopes: (Required)
Scopes requested to access a protected API (a resource).
:param str state: Recommended by OAuth2 for CSRF protection.
:param str login_hint:
Identifier of the user. Generally a User Principal Name (UPN).
:param str redirect_uri:
Address to return to upon receiving a response from the authority.
:param str response_type:
Default value is "code" for an OAuth2 Authorization Code grant.
You could use other content such as "id_token" or "token",
which would trigger an Implicit Grant, but that is
`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>`_.
:param str prompt:
By default, no prompt value will be sent, not even string ``"none"``.
You will have to specify a value explicitly.
Its valid values are the constants defined in
:class:`Prompt <msal.Prompt>`.
:param nonce:
A cryptographically random value used to mitigate replay attacks. See also
`OIDC specs <https://openid.net/specs/openid-connect-core-1_0.html#AuthRequest>`_.
:param domain_hint:
Can be one of "consumers" or "organizations" or your tenant domain "contoso.com".
If included, it will skip the email-based discovery process that user goes
through on the sign-in page, leading to a slightly more streamlined user experience.
More information on possible values available in
`Auth Code Flow doc <https://docs.microsoft.com/en-us/azure/active-directory/develop/v2-oauth2-auth-code-flow#request-an-authorization-code>`_ and
`domain_hint doc <https://docs.microsoft.com/en-us/openspecs/windows_protocols/ms-oapx/86fb452d-e34a-494e-ac61-e526e263b6d8>`_.
:param claims_challenge:
The claims_challenge parameter requests specific claims requested by the resource provider
in the form of a claims_challenge directive in the www-authenticate header to be
returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
It is a string of a JSON object which contains lists of claims being requested from these locations.
:return: The authorization url as a string.
"""
authority = kwargs.pop("authority", None) # Historically we support this
if authority:
warnings.warn(
"We haven't decided if this method will accept authority parameter")
# The previous implementation is, it will use self.authority by default.
# Multi-tenant app can use new authority on demand
the_authority = Authority(
authority,
self.http_client,
instance_discovery=self._instance_discovery,
) if authority else self.authority
client = _ClientWithCcsRoutingInfo(
{"authorization_endpoint": the_authority.authorization_endpoint},
self.client_id,
http_client=self.http_client)
warnings.warn(
"Change your get_authorization_request_url() "
"to initiate_auth_code_flow()", DeprecationWarning)
with warnings.catch_warnings(record=True):
return client.build_auth_request_uri(
response_type=response_type,
redirect_uri=redirect_uri, state=state, login_hint=login_hint,
prompt=prompt,
scope=self._decorate_scope(scopes),
nonce=nonce,
domain_hint=domain_hint,
claims=_merge_claims_challenge_and_capabilities(
self._client_capabilities, claims_challenge),
)
def acquire_token_by_auth_code_flow(
self, auth_code_flow, auth_response, scopes=None, **kwargs):
"""Validate the auth response being redirected back, and obtain tokens.
It automatically provides nonce protection.
:param dict auth_code_flow:
The same dict returned by :func:`~initiate_auth_code_flow()`.
:param dict auth_response:
A dict of the query string received from auth server.
:param list[str] scopes:
Scopes requested to access a protected API (a resource).
Most of the time, you can leave it empty.
If you requested user consent for multiple resources, here you will
need to provide a subset of what you required in
:func:`~initiate_auth_code_flow()`.
OAuth2 was designed mostly for singleton services,
where tokens are always meant for the same resource and the only
changes are in the scopes.
In Microsoft Entra, tokens can be issued for multiple 3rd party resources.
You can ask authorization code for multiple resources,
but when you redeem it, the token is for only one intended
recipient, called audience.
So the developer need to specify a scope so that we can restrict the
token to be issued for the corresponding audience.
:return:
* A dict containing "access_token" and/or "id_token", among others,
depends on what scope was used.
(See https://tools.ietf.org/html/rfc6749#section-5.1)
* A dict containing "error", optionally "error_description", "error_uri".
(It is either `this <https://tools.ietf.org/html/rfc6749#section-4.1.2.1>`_
or `that <https://tools.ietf.org/html/rfc6749#section-5.2>`_)
* Most client-side data error would result in ValueError exception.
So the usage pattern could be without any protocol details::
def authorize(): # A controller in a web app
try:
result = msal_app.acquire_token_by_auth_code_flow(
session.get("flow", {}), request.args)
if "error" in result:
return render_template("error.html", result)
use(result) # Token(s) are available in result and cache
except ValueError: # Usually caused by CSRF
pass # Simply ignore them
return redirect(url_for("index"))
"""
self._validate_ssh_cert_input_data(kwargs.get("data", {}))