Script 'mail_helper' called by obssrc
Hello community,
here is the log from the commit of package python-msal for openSUSE:Factory
checked in at 2024-08-01 22:04:21
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Comparing /work/SRC/openSUSE:Factory/python-msal (Old)
and /work/SRC/openSUSE:Factory/.python-msal.new.7232 (New)
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Package is "python-msal"
Thu Aug 1 22:04:21 2024 rev:23 rq:1190666 version:1.30.0
Changes:
--------
--- /work/SRC/openSUSE:Factory/python-msal/python-msal.changes 2024-07-09
20:06:26.246047802 +0200
+++ /work/SRC/openSUSE:Factory/.python-msal.new.7232/python-msal.changes
2024-08-01 22:04:46.329314831 +0200
@@ -1,0 +2,15 @@
+Wed Jul 31 12:50:17 UTC 2024 - John Paul Adrian Glaubitz
<adrian.glaub...@suse.com>
+
+- Update to version 1.30.0
+ * New feature: Support Subject Name/Issuer authentication when using
+ .pfx certificate file. Documentation available in one of the recent
+ purple boxes here. (#718)
+ * New feature: Automatically use SHA256 and PSS padding when using
+ .pfx certificate on non-ADFS, non-OIDC authorities. (#722)
+ * New feature: Expose refresh_on (if any) to fresh or cached response,
+ so that caller may choose to proactively call acquire_token_silent()
+ early. (#723)
+ * Bugfix for token cache search. MSAL 1.27+ customers please upgrade
+ to MSAL 1.30+. (#717)
+
+-------------------------------------------------------------------
Old:
----
msal-1.29.0.tar.gz
New:
----
msal-1.30.0.tar.gz
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Other differences:
------------------
++++++ python-msal.spec ++++++
--- /var/tmp/diff_new_pack.7UdU9n/_old 2024-08-01 22:04:47.069345353 +0200
+++ /var/tmp/diff_new_pack.7UdU9n/_new 2024-08-01 22:04:47.073345518 +0200
@@ -18,7 +18,7 @@
%{?sle15_python_module_pythons}
Name: python-msal
-Version: 1.29.0
+Version: 1.30.0
Release: 0
Summary: Microsoft Authentication Library (MSAL) for Python
License: MIT
++++++ msal-1.29.0.tar.gz -> msal-1.30.0.tar.gz ++++++
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/msal-1.29.0/PKG-INFO new/msal-1.30.0/PKG-INFO
--- old/msal-1.29.0/PKG-INFO 2024-06-22 04:14:01.844720100 +0200
+++ new/msal-1.30.0/PKG-INFO 2024-07-17 06:01:39.363946200 +0200
@@ -1,6 +1,6 @@
Metadata-Version: 2.1
Name: msal
-Version: 1.29.0
+Version: 1.30.0
Summary: The Microsoft Authentication Library (MSAL) for Python library
enables your app to access the Microsoft Cloud by supporting authentication of
users with Microsoft Azure Active Directory accounts (AAD) and Microsoft
Accounts (MSA) using industry standard OAuth2 and OpenID Connect.
Home-page:
https://github.com/AzureAD/microsoft-authentication-library-for-python
Author: Microsoft Corporation
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/msal-1.29.0/msal/application.py
new/msal-1.30.0/msal/application.py
--- old/msal-1.29.0/msal/application.py 2024-06-22 04:13:56.000000000 +0200
+++ new/msal-1.30.0/msal/application.py 2024-07-17 06:01:34.000000000 +0200
@@ -21,7 +21,7 @@
# The __init__.py will import this. Not the other way around.
-__version__ = "1.29.0" # When releasing, also check and bump our
dependencies's versions if needed
+__version__ = "1.30.0" # When releasing, also check and bump our
dependencies's versions if needed
logger = logging.getLogger(__name__)
_AUTHORITY_TYPE_CLOUDSHELL = "CLOUDSHELL"
@@ -61,17 +61,24 @@
return raw
-def _load_private_key_from_pfx_path(pfx_path, passphrase_bytes):
+def _parse_pfx(pfx_path, passphrase_bytes):
# Cert concepts https://security.stackexchange.com/a/226758/125264
- from cryptography.hazmat.primitives import hashes
+ from cryptography.hazmat.primitives import hashes, serialization
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)
+ if not (private_key and cert):
+ raise ValueError("Your PFX file shall contain both private key and
cert")
+ cert_pem = cert.public_bytes(encoding=serialization.Encoding.PEM).decode()
# cryptography 1.0+
+ x5c = [
+ '\n'.join(cert_pem.splitlines()[1:-1]) # Strip the "--- header ---"
and "--- footer ---"
+ ]
+ sha256_thumbprint = cert.fingerprint(hashes.SHA256()).hex() #
cryptography 0.7+
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
+ return private_key, sha256_thumbprint, sha1_thumbprint, x5c
def _load_private_key_from_pem_str(private_key_pem_str, passphrase_bytes):
@@ -97,11 +104,14 @@
"msalruntime_telemetry": result.get("_msalruntime_telemetry"),
"msal_python_telemetry": result.get("_msal_python_telemetry"),
}, separators=(",", ":"))
- return {
+ return_value = {
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
}
+ if "refresh_in" in result: # To encourage proactive refresh
+ return_value["refresh_on"] = int(time.time() +
result["refresh_in"])
+ return return_value
return result # It could be None
@@ -231,47 +241,71 @@
: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::
+ it supports many different input formats for different scenarios.
- {
- "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)",
- }
+ .. admonition:: Support using a client secret.
- 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>`_.
+ Just feed in a string, such as ``"your client secret"``.
- 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>`_.
+ .. admonition:: Support using a certificate in X.509 (.pem) format
- *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::
+ Feed in a dict in this form::
- {
- "client_assertion": "...a JWT with claims aud, exp, iss,
jti, nbf, and sub..."
- }
+ {
+ "private_key": "...-----BEGIN PRIVATE KEY-----... in
PEM format",
+ "thumbprint": "A1B2C3D4E5F6...",
+ "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 PKCS12 (.pfx) format,
+ you can convert it to X.509 (.pem) format,
+ by ``openssl pkcs12 -in file.pfx -out file.pem -nodes``.
+
+ 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>`_.
+
+ .. admonition:: Support Subject Name/Issuer Auth with a cert in
.pem
+
+ `Subject Name/Issuer Auth
+
<https://github.com/AzureAD/microsoft-authentication-library-for-python/issues/60>`_
+ is an approach to allow easier certificate rotation.
+
+ *Added in version 0.5.0*::
+
+ {
+ "private_key": "...-----BEGIN PRIVATE KEY-----... in
PEM format",
+ "thumbprint": "A1B2C3D4E5F6...",
+ "public_certificate": "...-----BEGIN
CERTIFICATE-----...",
+ "passphrase": "Passphrase if the private_key is
encrypted (Optional. Added in version 1.6.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.
+
+ .. admonition:: Supporting raw assertion obtained from elsewhere
+
+ *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
@@ -280,14 +314,26 @@
{
"private_key_pfx_path": "/path/to/your.pfx",
- "passphrase": "Passphrase if the private_key is
encrypted (Optional. Added in version 1.6.0)",
+ "passphrase": "Passphrase if the private_key is
encrypted (Optional)",
}
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]
+ .. admonition:: Support Subject Name/Issuer Auth with a cert in
.pfx
+
+ *Added in version 1.30.0*:
+ If your .pfx file contains both the private key and public
cert,
+ you can opt in for Subject Name/Issuer Auth like this::
+
+ {
+ "private_key_pfx_path": "/path/to/your.pfx",
+ "public_certificate": True,
+ "passphrase": "Passphrase if the private_key is
encrypted (Optional)",
+ }
+
+ :type client_credential: Union[dict, str, None]
:param dict client_claims:
*Added in version 0.5.0*:
@@ -699,14 +745,16 @@
client_assertion = client_credential['client_assertion']
else:
headers = {}
- if client_credential.get('public_certificate'):
- headers["x5c"] =
extract_certs(client_credential['public_certificate'])
+ sha1_thumbprint = sha256_thumbprint = None
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)
+ private_key, sha256_thumbprint, sha1_thumbprint, x5c =
_parse_pfx(
+ client_credential["private_key_pfx_path"],
+ passphrase_bytes)
+ if client_credential.get("public_certificate") is True and
x5c:
+ headers["x5c"] = x5c
elif (
client_credential.get("private_key") # PEM blob
and client_credential.get("thumbprint")):
@@ -720,9 +768,22 @@
raise ValueError(
"client_credential needs to follow this format "
"https://msal-python.readthedocs.io/en/latest/#msal.ClientApplication.params.client_credential";)
+ if ("x5c" not in headers # So the .pfx file contains no
certificate
+ and
isinstance(client_credential.get('public_certificate'), str)
+ ): # Then we treat the public_certificate value as PEM content
+ headers["x5c"] =
extract_certs(client_credential['public_certificate'])
+ if sha256_thumbprint and not authority.is_adfs:
+ assertion_params = {
+ "algorithm": "PS256", "sha256_thumbprint":
sha256_thumbprint,
+ }
+ else: # Fall back
+ if not sha1_thumbprint:
+ raise ValueError("You shall provide a thumbprint in
SHA1.")
+ assertion_params = {
+ "algorithm": "RS256", "sha1_thumbprint":
sha1_thumbprint,
+ }
assertion = JwtAssertionCreator(
- private_key, algorithm="RS256",
- sha1_thumbprint=sha1_thumbprint, headers=headers)
+ private_key, headers=headers, **assertion_params)
client_assertion = assertion.create_regenerative_assertion(
audience=authority.token_endpoint, issuer=self.client_id,
additional_claims=self.client_claims or {})
@@ -1449,9 +1510,11 @@
"expires_in": int(expires_in), # OAuth2 specs defines it
as int
self._TOKEN_SOURCE: self._TOKEN_SOURCE_CACHE,
}
- if "refresh_on" in entry and int(entry["refresh_on"]) < now:
# aging
- refresh_reason = msal.telemetry.AT_AGING
- break # With a fallback in hand, we break here to go
refresh
+ if "refresh_on" in entry:
+ access_token_from_cache["refresh_on"] =
int(entry["refresh_on"])
+ if int(entry["refresh_on"]) < now: # aging
+ refresh_reason = msal.telemetry.AT_AGING
+ break # With a fallback in hand, we break here to go
refresh
self._build_telemetry_context(-1).hit_an_access_token()
return access_token_from_cache # It is still good as new
else:
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/msal-1.29.0/msal/authority.py
new/msal-1.30.0/msal/authority.py
--- old/msal-1.29.0/msal/authority.py 2024-06-22 04:13:56.000000000 +0200
+++ new/msal-1.30.0/msal/authority.py 2024-07-17 06:01:34.000000000 +0200
@@ -68,11 +68,11 @@
"""
self._http_client = http_client
if oidc_authority_url:
- logger.info("Initializing with OIDC authority: %s",
oidc_authority_url)
+ logger.debug("Initializing with OIDC authority: %s",
oidc_authority_url)
tenant_discovery_endpoint = self._initialize_oidc_authority(
oidc_authority_url)
else:
- logger.info("Initializing with Entra authority: %s", authority_url)
+ logger.debug("Initializing with Entra authority: %s",
authority_url)
tenant_discovery_endpoint = self._initialize_entra_authority(
authority_url, validate_authority, instance_discovery)
try:
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/msal-1.29.0/msal/managed_identity.py
new/msal-1.30.0/msal/managed_identity.py
--- old/msal-1.29.0/msal/managed_identity.py 2024-06-22 04:13:56.000000000
+0200
+++ new/msal-1.30.0/msal/managed_identity.py 2024-07-17 06:01:34.000000000
+0200
@@ -273,8 +273,10 @@
"token_type": entry.get("token_type", "Bearer"),
"expires_in": int(expires_in), # OAuth2 specs defines it
as int
}
- if "refresh_on" in entry and int(entry["refresh_on"]) < now:
# aging
- break # With a fallback in hand, we break here to go
refresh
+ if "refresh_on" in entry:
+ access_token_from_cache["refresh_on"] =
int(entry["refresh_on"])
+ if int(entry["refresh_on"]) < now: # aging
+ break # With a fallback in hand, we break here to go
refresh
return access_token_from_cache # It is still good as new
try:
result = _obtain_token(self._http_client, self._managed_identity,
resource)
@@ -290,6 +292,8 @@
params={},
data={},
))
+ if "refresh_in" in result:
+ result["refresh_on"] = int(now + result["refresh_in"])
if (result and "error" not in result) or (not
access_token_from_cache):
return result
except: # The exact HTTP exception is transportation-layer dependent
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/msal-1.29.0/msal/oauth2cli/assertion.py
new/msal-1.30.0/msal/oauth2cli/assertion.py
--- old/msal-1.29.0/msal/oauth2cli/assertion.py 2024-06-22 04:13:56.000000000
+0200
+++ new/msal-1.30.0/msal/oauth2cli/assertion.py 2024-07-17 06:01:34.000000000
+0200
@@ -15,6 +15,8 @@
except: # Otherwise we treat it as bytes and return it as-is
return raw
+def _encode_thumbprint(thumbprint):
+ return base64.urlsafe_b64encode(binascii.a2b_hex(thumbprint)).decode()
class AssertionCreator(object):
def create_normal_assertion(
@@ -65,7 +67,11 @@
class JwtAssertionCreator(AssertionCreator):
- def __init__(self, key, algorithm, sha1_thumbprint=None, headers=None):
+ def __init__(
+ self, key, algorithm, sha1_thumbprint=None, headers=None,
+ *,
+ sha256_thumbprint=None,
+ ):
"""Construct a Jwt assertion creator.
Args:
@@ -80,13 +86,15 @@
RSA and ECDSA algorithms require "pip install cryptography".
sha1_thumbprint (str): The x5t aka X.509 certificate SHA-1
thumbprint.
headers (dict): Additional headers, e.g. "kid" or "x5c" etc.
+ sha256_thumbprint (str): The x5t#S256 aka X.509 certificate
SHA-256 thumbprint.
"""
self.key = key
self.algorithm = algorithm
self.headers = headers or {}
+ if sha256_thumbprint: #
https://datatracker.ietf.org/doc/html/rfc7515#section-4.1.8
+ self.headers["x5t#S256"] = _encode_thumbprint(sha256_thumbprint)
if sha1_thumbprint: #
https://tools.ietf.org/html/rfc7515#section-4.1.7
- self.headers["x5t"] = base64.urlsafe_b64encode(
- binascii.a2b_hex(sha1_thumbprint)).decode()
+ self.headers["x5t"] = _encode_thumbprint(sha1_thumbprint)
def create_normal_assertion(
self, audience, issuer, subject=None, expires_at=None,
expires_in=600,
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/msal-1.29.0/msal/token_cache.py
new/msal-1.30.0/msal/token_cache.py
--- old/msal-1.29.0/msal/token_cache.py 2024-06-22 04:13:56.000000000 +0200
+++ new/msal-1.30.0/msal/token_cache.py 2024-07-17 06:01:34.000000000 +0200
@@ -118,6 +118,12 @@
with self._lock:
return self._cache.get(credential_type, {}).get(key, default)
+ @staticmethod
+ def _is_matching(entry: dict, query: dict, target_set: set = None) -> bool:
+ return is_subdict_of(query or {}, entry) and (
+ target_set <= set(entry.get("target", "").split())
+ if target_set else True)
+
def search(self, credential_type, target=None, query=None): # O(n)
generator
"""Returns a generator of matching entries.
@@ -136,7 +142,10 @@
preferred_result = self._get_access_token(
query["home_account_id"], query["environment"],
query["client_id"], query["realm"], target)
- if preferred_result:
+ if preferred_result and self._is_matching(
+ preferred_result, query,
+ # Needs no target_set here because it is satisfied by dict key
+ ):
yield preferred_result
target_set = set(target)
@@ -145,11 +154,10 @@
# there is no point to attempt an O(1) key-value search here.
# So we always do an O(n) in-memory search.
for entry in self._cache.get(credential_type, {}).values():
- if is_subdict_of(query or {}, entry) and (
- target_set <= set(entry.get("target", "").split())
- if target else True):
- if entry != preferred_result: # Avoid yielding the same
entry twice
- yield entry
+ if (entry != preferred_result # Avoid yielding the same entry
twice
+ and self._is_matching(entry, query, target_set=target_set)
+ ):
+ yield entry
def find(self, credential_type, target=None, query=None):
"""Equivalent to list(search(...))."""
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/msal-1.29.0/msal.egg-info/PKG-INFO
new/msal-1.30.0/msal.egg-info/PKG-INFO
--- old/msal-1.29.0/msal.egg-info/PKG-INFO 2024-06-22 04:14:01.000000000
+0200
+++ new/msal-1.30.0/msal.egg-info/PKG-INFO 2024-07-17 06:01:39.000000000
+0200
@@ -1,6 +1,6 @@
Metadata-Version: 2.1
Name: msal
-Version: 1.29.0
+Version: 1.30.0
Summary: The Microsoft Authentication Library (MSAL) for Python library
enables your app to access the Microsoft Cloud by supporting authentication of
users with Microsoft Azure Active Directory accounts (AAD) and Microsoft
Accounts (MSA) using industry standard OAuth2 and OpenID Connect.
Home-page:
https://github.com/AzureAD/microsoft-authentication-library-for-python
Author: Microsoft Corporation
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/msal-1.29.0/tests/test_account_source.py
new/msal-1.30.0/tests/test_account_source.py
--- old/msal-1.29.0/tests/test_account_source.py 2024-06-22
04:13:56.000000000 +0200
+++ new/msal-1.30.0/tests/test_account_source.py 2024-07-17
06:01:34.000000000 +0200
@@ -46,20 +46,19 @@
mocked_broker_ats.assert_not_called()
self.assertEqual(result["token_source"], "identity_provider")
- def test_ropc_flow_and_its_silent_call_should_bypass_broker(self, _,
mocked_broker_ats):
+ def test_ropc_flow_and_its_silent_call_should_invoke_broker(self, _,
mocked_broker_ats):
app = msal.PublicClientApplication("client_id",
enable_broker_on_windows=True)
- with patch.object(app.authority, "user_realm_discovery",
return_value={}):
+ with patch("msal.broker._signin_silently",
return_value=dict(TOKEN_RESPONSE, _account_id="placeholder")):
result = app.acquire_token_by_username_password(
"username", "placeholder", [SCOPE], post=_mock_post)
- self.assertEqual(result["token_source"], "identity_provider")
+ self.assertEqual(result["token_source"], "broker")
account = app.get_accounts()[0]
- self.assertEqual(account["account_source"], "password")
+ self.assertEqual(account["account_source"], "broker")
result = app.acquire_token_silent_with_error(
[SCOPE], account, force_refresh=True, post=_mock_post)
- mocked_broker_ats.assert_not_called()
- self.assertEqual(result["token_source"], "identity_provider")
+ self.assertEqual(result["token_source"], "broker")
def test_interactive_flow_and_its_silent_call_should_invoke_broker(self,
_, mocked_broker_ats):
app = msal.PublicClientApplication("client_id",
enable_broker_on_windows=True)
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/msal-1.29.0/tests/test_application.py
new/msal-1.30.0/tests/test_application.py
--- old/msal-1.29.0/tests/test_application.py 2024-06-22 04:13:56.000000000
+0200
+++ new/msal-1.30.0/tests/test_application.py 2024-07-17 06:01:34.000000000
+0200
@@ -1,6 +1,7 @@
# Note: Since Aug 2019 we move all e2e tests into test_e2e.py,
# so this test_application file contains only unit tests without dependency.
import sys
+import time
from msal.application import *
from msal.application import _str2bytes
import msal
@@ -353,10 +354,18 @@
uid=self.uid, utid=self.utid, refresh_token=self.rt),
})
+ def assertRefreshOn(self, result, refresh_in):
+ refresh_on = int(time.time() + refresh_in)
+ self.assertTrue(
+ refresh_on - 1 < result.get("refresh_on", 0) < refresh_on + 1,
+ "refresh_on should be set properly")
+
def test_fresh_token_should_be_returned_from_cache(self):
# a.k.a. Return unexpired token that is not above token refresh
expiration threshold
+ refresh_in = 450
access_token = "An access token prepopulated into cache"
- self.populate_cache(access_token=access_token, expires_in=900,
refresh_in=450)
+ self.populate_cache(
+ access_token=access_token, expires_in=900, refresh_in=refresh_in)
result = self.app.acquire_token_silent(
['s1'], self.account,
post=lambda url, *args, **kwargs: # Utilize the undocumented test
feature
@@ -365,32 +374,38 @@
self.assertEqual(result[self.app._TOKEN_SOURCE],
self.app._TOKEN_SOURCE_CACHE)
self.assertEqual(access_token, result.get("access_token"))
self.assertNotIn("refresh_in", result, "Customers need not know
refresh_in")
+ self.assertRefreshOn(result, refresh_in)
def test_aging_token_and_available_aad_should_return_new_token(self):
# a.k.a. Attempt to refresh unexpired token when AAD available
self.populate_cache(access_token="old AT", expires_in=3599,
refresh_in=-1)
new_access_token = "new AT"
+ new_refresh_in = 123
def mock_post(url, headers=None, *args, **kwargs):
self.assertEqual("4|84,4|", (headers or
{}).get(CLIENT_CURRENT_TELEMETRY))
return MinimalResponse(status_code=200, text=json.dumps({
"access_token": new_access_token,
- "refresh_in": 123,
+ "refresh_in": new_refresh_in,
}))
result = self.app.acquire_token_silent(['s1'], self.account,
post=mock_post)
self.assertEqual(result[self.app._TOKEN_SOURCE],
self.app._TOKEN_SOURCE_IDP)
self.assertEqual(new_access_token, result.get("access_token"))
self.assertNotIn("refresh_in", result, "Customers need not know
refresh_in")
+ self.assertRefreshOn(result, new_refresh_in)
def test_aging_token_and_unavailable_aad_should_return_old_token(self):
# a.k.a. Attempt refresh unexpired token when AAD unavailable
+ refresh_in = -1
old_at = "old AT"
- self.populate_cache(access_token=old_at, expires_in=3599,
refresh_in=-1)
+ self.populate_cache(
+ access_token=old_at, expires_in=3599, refresh_in=refresh_in)
def mock_post(url, headers=None, *args, **kwargs):
self.assertEqual("4|84,4|", (headers or
{}).get(CLIENT_CURRENT_TELEMETRY))
return MinimalResponse(status_code=400, text=json.dumps({"error":
"foo"}))
result = self.app.acquire_token_silent(['s1'], self.account,
post=mock_post)
self.assertEqual(result[self.app._TOKEN_SOURCE],
self.app._TOKEN_SOURCE_CACHE)
self.assertEqual(old_at, result.get("access_token"))
+ self.assertRefreshOn(result, refresh_in)
def test_expired_token_and_unavailable_aad_should_return_error(self):
# a.k.a. Attempt refresh expired token when AAD unavailable
@@ -407,16 +422,18 @@
# a.k.a. Attempt refresh expired token when AAD available
self.populate_cache(access_token="expired at", expires_in=-1,
refresh_in=-900)
new_access_token = "new AT"
+ new_refresh_in = 123
def mock_post(url, headers=None, *args, **kwargs):
self.assertEqual("4|84,3|", (headers or
{}).get(CLIENT_CURRENT_TELEMETRY))
return MinimalResponse(status_code=200, text=json.dumps({
"access_token": new_access_token,
- "refresh_in": 123,
+ "refresh_in": new_refresh_in,
}))
result = self.app.acquire_token_silent(['s1'], self.account,
post=mock_post)
self.assertEqual(result[self.app._TOKEN_SOURCE],
self.app._TOKEN_SOURCE_IDP)
self.assertEqual(new_access_token, result.get("access_token"))
self.assertNotIn("refresh_in", result, "Customers need not know
refresh_in")
+ self.assertRefreshOn(result, new_refresh_in)
class TestTelemetryMaintainingOfflineState(unittest.TestCase):
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/msal-1.29.0/tests/test_broker.py
new/msal-1.30.0/tests/test_broker.py
--- old/msal-1.29.0/tests/test_broker.py 2024-06-22 04:13:56.000000000
+0200
+++ new/msal-1.30.0/tests/test_broker.py 2024-07-17 06:01:34.000000000
+0200
@@ -41,10 +41,15 @@
self.assertIn("Status_AccountUnusable",
result.get("error_description", ""))
def test_unconfigured_app_should_raise_exception(self):
- app_without_needed_redirect_uri =
"289a413d-284b-4303-9c79-94380abe5d22"
+ self.skipTest(
+ "After PyMsalRuntime 0.13.2, "
+ "AADSTS error codes were removed from error_context; "
+ "it is not in telemetry either.")
+ app_without_needed_redirect_uri =
"f62c5ae3-bf3a-4af5-afa8-a68b800396e9" # This is the lab app. We repurpose it
to be used here
with self.assertRaises(RedirectUriError):
- _signin_interactively(
+ result = _signin_interactively(
self._authority, app_without_needed_redirect_uri,
self._scopes, None)
+ print(result)
# Note: _acquire_token_silently() would raise same exception,
# we skip its test here due to the lack of a valid account_id
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/msal-1.29.0/tests/test_cryptography.py
new/msal-1.30.0/tests/test_cryptography.py
--- old/msal-1.29.0/tests/test_cryptography.py 2024-06-22 04:13:56.000000000
+0200
+++ new/msal-1.30.0/tests/test_cryptography.py 2024-07-17 06:01:34.000000000
+0200
@@ -8,7 +8,7 @@
import requests
from msal.application import (
- _str2bytes, _load_private_key_from_pem_str,
_load_private_key_from_pfx_path)
+ _str2bytes, _load_private_key_from_pem_str, _parse_pfx)
latest_cryptography_version = ET.fromstring(
@@ -48,7 +48,7 @@
_load_private_key_from_pem_str(f.read(), passphrase_bytes)
pfx = sibling("certificate-with-password.pfx") # Created by:
# openssl pkcs12 -export -inkey
test/certificate-with-password.pem -in tests/certificate-with-password.pem -out
tests/certificate-with-password.pfx
- _load_private_key_from_pfx_path(pfx, passphrase_bytes)
+ _parse_pfx(pfx, passphrase_bytes)
self.assertEqual(0, len(encountered_warnings),
"Did cryptography deprecate the functions that we used?")
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/msal-1.29.0/tests/test_e2e.py
new/msal-1.30.0/tests/test_e2e.py
--- old/msal-1.29.0/tests/test_e2e.py 2024-06-22 04:13:56.000000000 +0200
+++ new/msal-1.30.0/tests/test_e2e.py 2024-07-17 06:01:34.000000000 +0200
@@ -80,7 +80,7 @@
else "the upn from {}".format(_render(
username_uri, description="here" if html_mode else None)),
lab=_render(
- "https://aka.ms/GetLabUserSecret?Secret="; + (lab_name or
"msidlabXYZ"),
+ "https://aka.ms/GetLabSecret?Secret="; + (lab_name or "msidlabXYZ"),
description="this password api" if html_mode else None,
),
)
@@ -463,7 +463,10 @@
# id came from
https://docs.msidlab.com/accounts/confidentialclient.html
client_id = os.getenv(env_client_id)
# Cert came from
https://ms.portal.azure.com/#@microsoft.onmicrosoft.com/asset/Microsoft_Azure_KeyVault/Certificate/https://msidlabs.vault.azure.net/certificates/LabVaultAccessCert
- client_credential = {"private_key_pfx_path":
os.getenv(env_client_cert_path)}
+ client_credential = {
+ "private_key_pfx_path": os.getenv(env_client_cert_path),
+ "public_certificate": True, # Opt in for SNI
+ }
elif os.getenv(env_client_id) and os.getenv(env_name2):
# Data came from here
# https://docs.msidlab.com/accounts/confidentialclient.html
@@ -529,7 +532,7 @@
lab_name = lab_name.lower()
if lab_name not in cls._secrets:
logger.info("Querying lab user password for %s", lab_name)
- url = "https://msidlab.com/api/LabUserSecret?secret=%s"; % lab_name
+ url = "https://msidlab.com/api/LabSecret?secret=%s"; % lab_name
resp = cls.session.get(url)
cls._secrets[lab_name] = resp.json()["value"]
return cls._secrets[lab_name]
@@ -689,11 +692,28 @@
class PopWithExternalKeyTestCase(LabBasedTestCase):
def _test_service_principal(self):
- # Any SP can obtain an ssh-cert. Here we use the lab app.
- result = get_lab_app().acquire_token_for_client(self.SCOPE,
data=self.DATA1)
+ app = get_lab_app() # Any SP can obtain an ssh-cert. Here we use the
lab app.
+ result = app.acquire_token_for_client(self.SCOPE, data=self.DATA1)
self.assertIsNotNone(result.get("access_token"), "Encountered {}:
{}".format(
result.get("error"), result.get("error_description")))
self.assertEqual(self.EXPECTED_TOKEN_TYPE, result["token_type"])
+ self.assertEqual(result["token_source"], "identity_provider")
+
+ # Test cache hit
+ cached_result = app.acquire_token_for_client(self.SCOPE,
data=self.DATA1)
+ self.assertIsNotNone(
+ cached_result.get("access_token"), "Encountered {}: {}".format(
+ cached_result.get("error"),
cached_result.get("error_description")))
+ self.assertEqual(self.EXPECTED_TOKEN_TYPE, cached_result["token_type"])
+ self.assertEqual(cached_result["token_source"], "cache")
+
+ # refresh_token grant can fetch an ssh-cert bound to a different key
+ refreshed_result = app.acquire_token_for_client(self.SCOPE,
data=self.DATA2)
+ self.assertIsNotNone(
+ refreshed_result.get("access_token"), "Encountered {}: {}".format(
+ refreshed_result.get("error"),
refreshed_result.get("error_description")))
+ self.assertEqual(self.EXPECTED_TOKEN_TYPE,
refreshed_result["token_type"])
+ self.assertEqual(refreshed_result["token_source"], "identity_provider")
def _test_user_account(self):
lab_user = self.get_lab_user(usertype="cloud")
@@ -711,16 +731,30 @@
self.assertIsNotNone(result.get("access_token"), "Encountered {}:
{}".format(
result.get("error"), result.get("error_description")))
self.assertEqual(self.EXPECTED_TOKEN_TYPE, result["token_type"])
+ self.assertEqual(result["token_source"], "identity_provider")
logger.debug("%s.cache = %s",
self.id(), json.dumps(self.app.token_cache._cache, indent=4))
+ # refresh_token grant can hit an ssh-cert bound to the same key
+ account = self.app.get_accounts()[0]
+ cached_result = self.app.acquire_token_silent(
+ self.SCOPE, account=account, data=self.DATA1)
+ self.assertIsNotNone(cached_result)
+ self.assertEqual(self.EXPECTED_TOKEN_TYPE, cached_result["token_type"])
+ ## Actually, the self._test_acquire_token_interactive() already
contained
+ ## a built-in refresh test, so the token in cache has been refreshed
already.
+ ## Therefore, the following line won't pass, which is expected.
+ #self.assertEqual(result["access_token"],
cached_result['access_token'])
+ self.assertEqual(cached_result["token_source"], "cache")
+
# refresh_token grant can fetch an ssh-cert bound to a different key
account = self.app.get_accounts()[0]
- refreshed_ssh_cert = self.app.acquire_token_silent(
+ refreshed_result = self.app.acquire_token_silent(
self.SCOPE, account=account, data=self.DATA2)
- self.assertIsNotNone(refreshed_ssh_cert)
- self.assertEqual(self.EXPECTED_TOKEN_TYPE,
refreshed_ssh_cert["token_type"])
- self.assertNotEqual(result["access_token"],
refreshed_ssh_cert['access_token'])
+ self.assertIsNotNone(refreshed_result)
+ self.assertEqual(self.EXPECTED_TOKEN_TYPE,
refreshed_result["token_type"])
+ self.assertNotEqual(result["access_token"],
refreshed_result['access_token'])
+ self.assertEqual(refreshed_result["token_source"], "identity_provider")
class SshCertTestCase(PopWithExternalKeyTestCase):
@@ -829,7 +863,7 @@
#
https://msidlab.com/api/user?usertype=onprem&federationprovider=ADFSv2019
username = "..." # The upn from the link above
- password="***" # From
https://aka.ms/GetLabUserSecret?Secret=msidlabXYZ
+ password="***" # From https://aka.ms/GetLabSecret?Secret=msidlabXYZ
"""
config = self.get_lab_user(usertype="onprem",
federationProvider="ADFSv2019")
config["authority"] = "https://fs.%s.com/adfs"; % config["lab_name"]
@@ -922,7 +956,7 @@
username="b2clo...@msidlabb2c.onmicrosoft.com"
# This won't work https://msidlab.com/api/user?usertype=b2c
- password="***" # From
https://aka.ms/GetLabUserSecret?Secret=msidlabb2c
+ password="***" # From
https://aka.ms/GetLabSecret?Secret=msidlabb2c
"""
config = self.get_lab_app_object(azureenvironment="azureb2ccloud")
self._test_acquire_token_by_auth_code(
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/msal-1.29.0/tests/test_mi.py
new/msal-1.30.0/tests/test_mi.py
--- old/msal-1.29.0/tests/test_mi.py 2024-06-22 04:13:56.000000000 +0200
+++ new/msal-1.30.0/tests/test_mi.py 2024-07-17 06:01:34.000000000 +0200
@@ -26,6 +26,7 @@
SERVICE_FABRIC,
DEFAULT_TO_VM,
)
+from msal.token_cache import is_subdict_of
class ManagedIdentityTestCase(unittest.TestCase):
@@ -60,7 +61,7 @@
http_client=requests.Session(),
)
- def _test_token_cache(self, app):
+ def assertCacheStatus(self, app):
cache = app._token_cache._cache
self.assertEqual(1, len(cache.get("AccessToken", [])), "Should have 1
AT")
at = list(cache["AccessToken"].values())[0]
@@ -70,30 +71,55 @@
"Should have expected client_id")
self.assertEqual("managed_identity", at["realm"], "Should have
expected realm")
- def _test_happy_path(self, app, mocked_http):
- result = app.acquire_token_for_client(resource="R")
+ def _test_happy_path(self, app, mocked_http, expires_in, resource="R"):
+ result = app.acquire_token_for_client(resource=resource)
mocked_http.assert_called()
- self.assertEqual({
+ call_count = mocked_http.call_count
+ expected_result = {
"access_token": "AT",
- "expires_in": 1234,
- "resource": "R",
"token_type": "Bearer",
- }, result, "Should obtain a token response")
+ }
+ self.assertTrue(
+ is_subdict_of(expected_result, result), # We will test refresh_on
later
+ "Should obtain a token response")
+ self.assertEqual(expires_in, result["expires_in"], "Should have
expected expires_in")
+ if expires_in >= 7200:
+ expected_refresh_on = int(time.time() + expires_in / 2)
+ self.assertTrue(
+ expected_refresh_on - 1 <= result["refresh_on"] <=
expected_refresh_on + 1,
+ "Should have a refresh_on time around the middle of the
token's life")
self.assertEqual(
result["access_token"],
- app.acquire_token_for_client(resource="R").get("access_token"),
+
app.acquire_token_for_client(resource=resource).get("access_token"),
"Should hit the same token from cache")
- self._test_token_cache(app)
+
+ self.assertCacheStatus(app)
+
+ result = app.acquire_token_for_client(resource=resource)
+ self.assertEqual(
+ call_count, mocked_http.call_count,
+ "No new call to the mocked http should be made for a cache hit")
+ self.assertTrue(
+ is_subdict_of(expected_result, result), # We will test refresh_on
later
+ "Should obtain a token response")
+ self.assertTrue(
+ expires_in - 5 < result["expires_in"] <= expires_in,
+ "Should have similar expires_in")
+ if expires_in >= 7200:
+ self.assertTrue(
+ expected_refresh_on - 5 < result["refresh_on"] <=
expected_refresh_on,
+ "Should have a refresh_on time around the middle of the
token's life")
class VmTestCase(ClientTestCase):
def test_happy_path(self):
+ expires_in = 7890 # We test a bigger than 7200 value here
with patch.object(self.app._http_client, "get",
return_value=MinimalResponse(
status_code=200,
- text='{"access_token": "AT", "expires_in": "1234", "resource":
"R"}',
+ text='{"access_token": "AT", "expires_in": "%s", "resource": "R"}'
% expires_in,
)) as mocked_method:
- self._test_happy_path(self.app, mocked_method)
+ self._test_happy_path(self.app, mocked_method, expires_in)
def test_vm_error_should_be_returned_as_is(self):
raw_error = '{"raw": "error format is undefined"}'
@@ -110,12 +136,13 @@
class AppServiceTestCase(ClientTestCase):
def test_happy_path(self):
+ expires_in = 1234
with patch.object(self.app._http_client, "get",
return_value=MinimalResponse(
status_code=200,
text='{"access_token": "AT", "expires_on": "%s", "resource": "R"}'
% (
- int(time.time()) + 1234),
+ int(time.time()) + expires_in),
)) as mocked_method:
- self._test_happy_path(self.app, mocked_method)
+ self._test_happy_path(self.app, mocked_method, expires_in)
def test_app_service_error_should_be_normalized(self):
raw_error = '{"statusCode": 500, "message": "error content is
undefined"}'
@@ -134,12 +161,13 @@
class MachineLearningTestCase(ClientTestCase):
def test_happy_path(self):
+ expires_in = 1234
with patch.object(self.app._http_client, "get",
return_value=MinimalResponse(
status_code=200,
text='{"access_token": "AT", "expires_on": "%s", "resource": "R"}'
% (
- int(time.time()) + 1234),
+ int(time.time()) + expires_in),
)) as mocked_method:
- self._test_happy_path(self.app, mocked_method)
+ self._test_happy_path(self.app, mocked_method, expires_in)
def test_machine_learning_error_should_be_normalized(self):
raw_error = '{"error": "placeholder", "message": "placeholder"}'
@@ -162,12 +190,14 @@
class ServiceFabricTestCase(ClientTestCase):
def _test_happy_path(self, app):
+ expires_in = 1234
with patch.object(app._http_client, "get",
return_value=MinimalResponse(
status_code=200,
text='{"access_token": "AT", "expires_on": %s, "resource": "R",
"token_type": "Bearer"}' % (
- int(time.time()) + 1234),
+ int(time.time()) + expires_in),
)) as mocked_method:
- super(ServiceFabricTestCase, self)._test_happy_path(app,
mocked_method)
+ super(ServiceFabricTestCase, self)._test_happy_path(
+ app, mocked_method, expires_in)
def test_happy_path(self):
self._test_happy_path(self.app)
@@ -212,15 +242,16 @@
})
def test_happy_path(self, mocked_stat):
+ expires_in = 1234
with patch.object(self.app._http_client, "get", side_effect=[
self.challenge,
MinimalResponse(
status_code=200,
- text='{"access_token": "AT", "expires_in": "1234", "resource":
"R"}',
+ text='{"access_token": "AT", "expires_in": "%s", "resource":
"R"}' % expires_in,
),
]) as mocked_method:
try:
- super(ArcTestCase, self)._test_happy_path(self.app,
mocked_method)
+ self._test_happy_path(self.app, mocked_method, expires_in)
mocked_stat.assert_called_with(os.path.join(
_supported_arc_platforms_and_their_prefixes[sys.platform],
"foo.key"))
