Device Flow

Author: rayluo

msal/application.py

@@ -1,4 +1,9 @@
 import time
+try:  # Python 2
+    from urlparse import urljoin
+except:  # Python 3
+    from urllib.parse import urljoin
+import logging
 
 from oauth2cli import Client
 from .authority import Authority
@@ -61,7 +66,11 @@ def __init__(
         default_body["client_info"] = 1
         self.client = Client(
             self.client_id,
-            configuration={"token_endpoint": self.authority.token_endpoint},
+            configuration={
+                "token_endpoint": self.authority.token_endpoint,
+                "device_authorization_endpoint": urljoin(
+                    self.authority.token_endpoint, "devicecode"),
+                },
             default_body=default_body,
             on_obtaining_tokens=self.token_cache.add,
             on_removing_rt=self.token_cache.remove_rt,
@@ -219,7 +228,32 @@ def acquire_token_silent(
                 scope=decorate_scope(scope, self.client_id))
             if "error" not in response:
                 return response
-
+            logging.debug(
+                "Refresh failed. {error}: {error_description}".format(**response))
+
+    def initiate_device_flow(self, scope=None, **kwargs):
+        return self.client.initiate_device_flow(
+            scope=decorate_scope(scope, self.client_id) if scope else None,
+            **kwargs)
+
+    def acquire_token_by_device_flow(
+            self, flow, exit_condition=lambda: True, **kwargs):
+        """Obtain token by a device flow object, with optional polling effect.
+
+        Args:
+            flow (dict):
+                An object previously generated by initiate_device_flow(...).
+            exit_condition (Callable):
+                This method implements a loop to provide polling effect.
+                The loop's exit condition is calculated by this callback.
+                The default callback makes the loop run only once, i.e. no polling.
+        """
+        return self.client.obtain_token_by_device_flow(
+                flow, exit_condition=exit_condition,
+                data={"code": flow["device_code"]},  # 2018-10-4 Hack:
+                    # during transition period,
+                    # service seemingly need both device_code and code parameter.
+                **kwargs)
 
 class PublicClientApplication(ClientApplication):  # browser app or mobile app
 
@@ -254,8 +288,6 @@ def acquire_token(
         # It will handle the TWO round trips of Authorization Code Grant flow.
         raise NotImplemented()
 
-    # TODO: Support Device Code flow
-
 
 class ConfidentialClientApplication(ClientApplication):  # server-side web app
 

tests/test_application.py

@@ -13,6 +13,28 @@
     with open(CONFIG_FILE) as conf:
         CONFIG = json.load(conf)
 
+logger = logging.getLogger(__file__)
+logging.basicConfig(level=logging.DEBUG)
+
+
+class Oauth2TestCase(unittest.TestCase):
+
+    def assertLoosely(self, response, assertion=None,
+            skippable_errors=("invalid_grant", "interaction_required")):
+        if response.get("error") in skippable_errors:
+            logger.debug("Response = %s", response)
+            # Some of these errors are configuration issues, not library issues
+            raise unittest.SkipTest(response.get("error_description"))
+        else:
+            if assertion is None:
+                assertion = lambda: self.assertIn(
+                    "access_token", response,
+                    "{error}: {error_description}".format(
+                        # Do explicit response.get(...) rather than **response
+                        error=response.get("error"),
+                        error_description=response.get("error_description")))
+            assertion()
+
 
 @unittest.skipUnless("client_id" in CONFIG, "client_id missing")
 class TestConfidentialClientApplication(unittest.TestCase):
@@ -58,23 +80,14 @@ def test_username_password(self):
 
 
 @unittest.skipUnless("client_id" in CONFIG, "client_id missing")
-class TestClientApplication(unittest.TestCase):
+class TestClientApplication(Oauth2TestCase):
 
     @classmethod
     def setUpClass(cls):
         cls.app = ClientApplication(
             CONFIG["client_id"], client_credential=CONFIG.get("client_secret"),
             authority=CONFIG.get("authority"))
 
-    def assertLoosely(self, result):
-        if "error" in result:
-            # Some of these errors are configuration issues, not library issues
-            if result["error"] == "invalid_grant":
-                raise unittest.SkipTest(result.get("error_description"))
-            self.assertEqual(result["error"], "interaction_required")
-        else:
-            self.assertIn('access_token', result)
-
     @unittest.skipUnless("scope" in CONFIG, "Missing scope")
     def test_auth_code(self):
         from oauth2cli.authcode import obtain_auth_code
@@ -88,8 +101,18 @@ def test_auth_code(self):
         result = self.app.acquire_token_with_authorization_code(
             ac, CONFIG["scope"], redirect_uri=redirect_uri)
         logging.debug("cache = %s", json.dumps(self.app.token_cache._cache, indent=4))
-        self.assertIn("access_token", result, "We should receive AT by auth code")
+        self.assertIn(
+            "access_token", result,
+            "{error}: {error_description}".format(
+                # Note: No interpolation here, cause error won't always present
+                error=result.get("error"),
+                error_description=result.get("error_description")))
+
+        self.assertCacheWorks(result)
 
+
+    def assertCacheWorks(self, result_from_wire):
+        result = result_from_wire
         # Going to test acquire_token_silent(...) to locate an AT from cache
         # In practice, you may want to filter based on its "username" field
         accounts = self.app.get_accounts()
@@ -109,3 +132,20 @@ def test_auth_code(self):
         self.assertNotEqual(result['access_token'], result_from_cache['access_token'],
                 "We should get a fresh AT (via RT)")
 
+    def test_device_flow(self):
+        flow = self.app.initiate_device_flow(scope=CONFIG.get("scope"))
+        logging.warn(flow["message"])
+
+        duration = 30
+        logging.warn("We will wait up to %d seconds for you to sign in" % duration)
+        result = self.app.acquire_token_by_device_flow(
+            flow,
+            exit_condition=lambda end=time.time() + duration: time.time() > end)
+        self.assertLoosely(
+                result,
+                assertion=lambda: self.assertIn('access_token', result),
+                skippable_errors=self.app.client.DEVICE_FLOW_RETRIABLE_ERRORS)
+
+        if "access_token" in result:
+            self.assertCacheWorks(result)
+