Bypass device authorization flow, for real

Author: rayluo

msal/throttled_http_client.py

@@ -100,16 +100,17 @@ def __init__(self, http_client, http_cache):
                     # acquire_token_silent(..., force_refresh=True) pattern.
                     str(kwargs.get("params")) + str(kwargs.get("data"))),
                 ),
-            expires_in=lambda result=None, data=None, **ignored:
+            expires_in=lambda result=None, kwargs=None, **ignored:
                 60
                 if result.status_code == 400
                     # Here we choose to cache exact HTTP 400 errors only (rather than 4xx)
                     # because they are the ones defined in OAuth2
                     # (https://datatracker.ietf.org/doc/html/rfc6749#section-5.2)
                     # Other 4xx errors might have different requirements e.g.
                     # "407 Proxy auth required" would need a key including http headers.
-                and not(  # Exclude Device Flow cause its retry is expected and regulated
-                    isinstance(data, dict) and data.get("grant_type") == DEVICE_AUTH_GRANT
+                and not(  # Exclude Device Flow whose retry is expected and regulated
+                    isinstance(kwargs.get("data"), dict)
+                    and kwargs["data"].get("grant_type") == DEVICE_AUTH_GRANT
                     )
                 and "retry-after" not in set(  # Leave it to the Retry-After decorator
                     h.lower() for h in getattr(result, "headers", {}).keys())

tests/test_throttled_http_client.py

@@ -163,9 +163,9 @@ def test_device_flow_retry_should_not_be_cached(self):
         http_cache = {}
         http_client = DummyHttpClient(status_code=400)
         http_client = ThrottledHttpClient(http_client, http_cache)
-        resp1 = http_client.get(
+        resp1 = http_client.post(
             "https://example.com", data={"grant_type": DEVICE_AUTH_GRANT})
-        resp2 = http_client.get(
+        resp2 = http_client.post(
             "https://example.com", data={"grant_type": DEVICE_AUTH_GRANT})
         logger.debug(http_cache)
         self.assertNotEqual(resp1.text, resp2.text, "Should return a new response")