Add access_token_sha256_to_refresh

Author: rayluo

tests/test_application.py

@@ -1,5 +1,6 @@
 # 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 hashlib
 import json
 import logging
 import sys
@@ -56,6 +57,35 @@ def test_bytes_to_bytes(self):
         self.assertEqual(type(_str2bytes(b"some bytes")), type(b"bytes"))
 
 
+def fake_token_getter(
+    *,
+    access_token: str = "an access token",
+    status_code: int = 200,
+    expires_in: int = 3600,
+    token_type: str = "Bearer",
+    payload: dict = None,
+    headers: dict = None,
+):
+    """A helper to create a fake token getter,
+    which will be consumed by ClientApplication's acquire methods' post parameter.
+
+    Generic mock.patch() is inconvenient because:
+    1. If you patch it at or above oauth2.py _obtain_token(), token cache is not populated.
+    2. If you patch it at request.post(), your test cases become fragile because
+       more http round-trips may be added for future flows,
+       then your existing test case would break until you mock new round-trips.
+    """
+    return lambda url, *args, **kwargs: MinimalResponse(
+        status_code=status_code,
+        text=json.dumps(payload or {
+            "access_token": access_token,
+            "expires_in": expires_in,
+            "token_type": token_type,
+            }),
+        headers=headers,
+    )
+
+
 class TestClientApplicationAcquireTokenSilentErrorBehaviors(unittest.TestCase):
 
     def setUp(self):
@@ -856,3 +886,45 @@ def test_app_did_not_register_redirect_uri_should_error_out(self):
                 )
             self.assertEqual(result.get("error"), "broker_error")
 
+
+@patch("msal.authority.tenant_discovery", new=Mock(return_value={
+    "authorization_endpoint": "https://contoso.com/placeholder",
+    "token_endpoint": "https://contoso.com/placeholder",
+    }))
+class AccessTokenToRefreshTestCase(unittest.TestCase):
+    def test_mismatching_hash_should_not_trigger_refresh(self):
+        scopes = ["scope"]
+        token1 = "AT one"
+        token1_hash = hashlib.sha256(token1.encode()).hexdigest()
+        token2 = "AT two"
+        app = msal.ConfidentialClientApplication("foo", client_credential="bar")
+
+        # Prepopulate cache
+        app.acquire_token_for_client(scopes, post=fake_token_getter(access_token=token1))
+        self.assertNotEqual(app.token_cache._cache, {}, "Cache should have been populated")
+
+        # Test mismatching hash should not trigger refresh
+        result = app.acquire_token_for_client(
+            scopes,
+            access_token_sha256_to_refresh="mismatching hash",
+            post=fake_token_getter(access_token=token2))
+        self.assertEqual(result.get("access_token"), token1, "Should hit old token")
+        self.assertEqual(result.get("token_source"), app._TOKEN_SOURCE_CACHE)
+
+        # Test matching hash should trigger refresh
+        result = app.acquire_token_for_client(
+            scopes,
+            access_token_sha256_to_refresh=token1_hash,
+            post=fake_token_getter(access_token=token2))
+        self.assertEqual(result.get("access_token"), token2, "Should obtain new token")
+        self.assertEqual(result.get("token_source"), app._TOKEN_SOURCE_IDP)
+
+        # A client using old token1, even with claims challenge,
+        # should not trigger refresh, because we have token2 in cache.
+        result = app.acquire_token_for_client(
+            scopes,
+            access_token_sha256_to_refresh=token1_hash,
+            claims_challenge='{"access_token": {"xms_cc": {"values": ["challenge for token1"]}}}',
+            post=fake_token_getter(access_token="AT three"))
+        self.assertEqual(result.get("access_token"), token2, "Token 2 should be returned")
+        self.assertEqual(result.get("token_source"), app._TOKEN_SOURCE_CACHE)