Add google.oauth2.credentials.Credentials.from_authorized_user_file (#226)

Author: hiranya911

packages/google-auth/google/auth/_cloud_sdk.py

@@ -18,13 +18,9 @@
 import os
 import subprocess
 
-import six
-
 from google.auth import environment_vars
 import google.oauth2.credentials
 
-# The Google OAuth 2.0 token endpoint. Used for authorized user credentials.
-_GOOGLE_OAUTH2_TOKEN_ENDPOINT = 'https://accounts.google.com/o/oauth2/token'
 
 # The ~/.config subdirectory containing gcloud credentials.
 _CONFIG_DIRECTORY = 'gcloud'
@@ -94,20 +90,8 @@ def load_authorized_user_credentials(info):
     Raises:
         ValueError: if the info is in the wrong format or missing data.
     """
-    keys_needed = set(('refresh_token', 'client_id', 'client_secret'))
-    missing = keys_needed.difference(six.iterkeys(info))
-
-    if missing:
-        raise ValueError(
-            'Authorized user info was not in the expected format, missing '
-            'fields {}.'.format(', '.join(missing)))
-
-    return google.oauth2.credentials.Credentials(
-        None,  # No access token, must be refreshed.
-        refresh_token=info['refresh_token'],
-        token_uri=_GOOGLE_OAUTH2_TOKEN_ENDPOINT,
-        client_id=info['client_id'],
-        client_secret=info['client_secret'])
+    return google.oauth2.credentials.Credentials.from_authorized_user_info(
+        info)
 
 
 def get_project_id():

packages/google-auth/google/oauth2/credentials.py

@@ -31,11 +31,20 @@
 .. _rfc6749 section 4.1: https://tools.ietf.org/html/rfc6749#section-4.1
 """
 
+import io
+import json
+
+import six
+
 from google.auth import _helpers
 from google.auth import credentials
 from google.oauth2 import _client
 
 
+# The Google OAuth 2.0 token endpoint. Used for authorized user credentials.
+_GOOGLE_OAUTH2_TOKEN_ENDPOINT = 'https://accounts.google.com/o/oauth2/token'
+
+
 class Credentials(credentials.ReadOnlyScoped, credentials.Credentials):
     """Credentials using OAuth 2.0 access and refresh tokens."""
 
@@ -120,3 +129,56 @@ def refresh(self, request):
         self.expiry = expiry
         self._refresh_token = refresh_token
         self._id_token = grant_response.get('id_token')
+
+    @classmethod
+    def from_authorized_user_info(cls, info, scopes=None):
+        """Creates a Credentials instance from parsed authorized user info.
+
+        Args:
+            info (Mapping[str, str]): The authorized user info in Google
+                format.
+            scopes (Sequence[str]): Optional list of scopes to include in the
+                credentials.
+
+        Returns:
+            google.oauth2.credentials.Credentials: The constructed
+                credentials.
+
+        Raises:
+            ValueError: If the info is not in the expected format.
+        """
+        keys_needed = set(('refresh_token', 'client_id', 'client_secret'))
+        missing = keys_needed.difference(six.iterkeys(info))
+
+        if missing:
+            raise ValueError(
+                'Authorized user info was not in the expected format, missing '
+                'fields {}.'.format(', '.join(missing)))
+
+        return Credentials(
+            None,  # No access token, must be refreshed.
+            refresh_token=info['refresh_token'],
+            token_uri=_GOOGLE_OAUTH2_TOKEN_ENDPOINT,
+            scopes=scopes,
+            client_id=info['client_id'],
+            client_secret=info['client_secret'])
+
+    @classmethod
+    def from_authorized_user_file(cls, filename, scopes=None):
+        """Creates a Credentials instance from an authorized user json file.
+
+        Args:
+            filename (str): The path to the authorized user json file.
+            scopes (Sequence[str]): Optional list of scopes to include in the
+                credentials.
+
+        Returns:
+            google.oauth2.credentials.Credentials: The constructed
+                credentials.
+
+        Raises:
+            ValueError: If the file is not in the expected format.
+        """
+        with io.open(filename, 'r', encoding='utf-8') as json_file:
+            data = json.load(json_file)
+            return cls.from_authorized_user_info(data, scopes)

packages/google-auth/tests/oauth2/test_credentials.py

@@ -13,6 +13,8 @@
 # limitations under the License.
 
 import datetime
+import json
+import os
 
 import mock
 
@@ -21,6 +23,14 @@
 from google.oauth2 import credentials
 
 
+DATA_DIR = os.path.join(os.path.dirname(__file__), '..', 'data')
+
+AUTH_USER_JSON_FILE = os.path.join(DATA_DIR, 'authorized_user.json')
+
+with open(AUTH_USER_JSON_FILE, 'r') as fh:
+    AUTH_USER_INFO = json.load(fh)
+
+
 class TestCredentials(object):
     TOKEN_URI = 'https://example.com/oauth2/token'
     REFRESH_TOKEN = 'refresh_token'
@@ -84,3 +94,42 @@ def test_refresh_success(self, unused_utcnow, refresh_grant):
         # Check that the credentials are valid (have a token and are not
         # expired)
         assert credentials.valid
+
+    def test_from_authorized_user_info(self):
+        info = AUTH_USER_INFO.copy()
+
+        creds = credentials.Credentials.from_authorized_user_info(info)
+        assert creds.client_secret == info['client_secret']
+        assert creds.client_id == info['client_id']
+        assert creds.refresh_token == info['refresh_token']
+        assert creds.token_uri == credentials._GOOGLE_OAUTH2_TOKEN_ENDPOINT
+        assert creds.scopes is None
+
+        scopes = ['email', 'profile']
+        creds = credentials.Credentials.from_authorized_user_info(
+            info, scopes)
+        assert creds.client_secret == info['client_secret']
+        assert creds.client_id == info['client_id']
+        assert creds.refresh_token == info['refresh_token']
+        assert creds.token_uri == credentials._GOOGLE_OAUTH2_TOKEN_ENDPOINT
+        assert creds.scopes == scopes
+
+    def test_from_authorized_user_file(self):
+        info = AUTH_USER_INFO.copy()
+
+        creds = credentials.Credentials.from_authorized_user_file(
+            AUTH_USER_JSON_FILE)
+        assert creds.client_secret == info['client_secret']
+        assert creds.client_id == info['client_id']
+        assert creds.refresh_token == info['refresh_token']
+        assert creds.token_uri == credentials._GOOGLE_OAUTH2_TOKEN_ENDPOINT
+        assert creds.scopes is None
+
+        scopes = ['email', 'profile']
+        creds = credentials.Credentials.from_authorized_user_file(
+            AUTH_USER_JSON_FILE, scopes)
+        assert creds.client_secret == info['client_secret']
+        assert creds.client_id == info['client_id']
+        assert creds.refresh_token == info['refresh_token']
+        assert creds.token_uri == credentials._GOOGLE_OAUTH2_TOKEN_ENDPOINT
+        assert creds.scopes == scopes

packages/google-auth/tests/test__cloud_sdk.py

@@ -145,7 +145,8 @@ def test_load_authorized_user_credentials():
     assert credentials._client_id == AUTHORIZED_USER_FILE_DATA['client_id']
     assert (credentials._client_secret ==
             AUTHORIZED_USER_FILE_DATA['client_secret'])
-    assert credentials._token_uri == _cloud_sdk._GOOGLE_OAUTH2_TOKEN_ENDPOINT
+    assert (credentials._token_uri ==
+            google.oauth2.credentials._GOOGLE_OAUTH2_TOKEN_ENDPOINT)
 
 
 def test_load_authorized_user_credentials_bad_format():