feat: adds new external account authorized user credentials

google/auth/external_account_authorized_user.py

@@ -13,9 +13,13 @@
 # limitations under the License.
 
 """External Account Authorized User Credentials.
-This module provides credentials to access Google Cloud resources from on-prem
-or non-Google Cloud platforms which support external credentials (e.g. OIDC ID
-tokens) as part of a web-based 3-legged OAuth flow.
+This module provides credentials based on OAuth 2.0 access and refresh tokens.
+These credentials usually access resources on behalf of a user (resource
+owner).
+
+Specifically, these are sourced using external identities via Workforce Identity Federation.
+
+Obtaining the initial access and refresh token can be done through the Google Cloud CLI.
 
 Example credential:
 {
@@ -38,16 +42,23 @@
 from google.oauth2 import sts
 from google.oauth2 import utils
 
-
 _EXTERNAL_ACCOUNT_AUTHORIZED_USER_JSON_TYPE = "external_account_authorized_user"
 
 
-class Credentials(credentials.CredentialsWithQuotaProject, credentials.ReadOnlyScoped):
+class Credentials(
+    credentials.CredentialsWithQuotaProject,
+    credentials.ReadOnlyScoped,
+    credentials.CredentialsWithTokenUri,
+):
     """Credentials for External Account Authorized Users.
 
     This is used to instantiate Credentials for exchanging refresh tokens from
     authorized users for Google access token and authorizing requests to Google
     APIs.
+
+    The credentials are considered immutable. If you want to modify the
+    quota project, use `with_quota_project` and if you want to modify the token
+    uri, use `with_token_uri`
     """
 
     def __init__(
@@ -88,20 +99,10 @@ def info(self):
                 useful for serializing the current credentials so it can deserialized
                 later.
         """
-        config_info = {
-            "type": _HEADFUL_JSON_TYPE,
-            "audience": self._audience,
-            "refresh_token": self._refresh_token,
-            "token_url": self._token_url,
-            "token_info_url": self._token_info_url,
-            "client_id": self._client_id,
-            "client_secret": self._client_secret,
-            "revoke_url": self._revoke_url,
-            "quota_project_id": self._quota_project_id,
-        }
+        config_info = self.constructor_args()
+        config_info.update(type=_EXTERNAL_ACCOUNT_AUTHORIZED_USER_JSON_TYPE)
         return {key: value for key, value in config_info.items() if value is not None}
 
-    @property
     def constructor_args(self):
         return {
             "audience": self._audience,
@@ -120,9 +121,6 @@ def requires_scopes(self):
         the initial token is requested and can not be changed."""
         return False
 
-    def _make_sts_request(self, request):
-        return self._sts_client.refresh_token(request, self._refresh_token)
-
     def get_project_id(self):
         return None
 
@@ -133,11 +131,20 @@ def refresh(self, request):
         lifetime = datetime.timedelta(seconds=response_data.get("expires_in"))
         self.expiry = now + lifetime
 
+    def _make_sts_request(self, request):
+        return self._sts_client.refresh_token(request, self._refresh_token)
+
     @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
     def with_quota_project(self, quota_project_id):
-        return self.__class__(
-            quota_project_id=quota_project_id, **self.constructor_args
-        )
+        kwargs = self.constructor_args()
+        kwargs.update(quota_project_id=quota_project_id)
+        return self.__class__(**kwargs)
+
+    @_helpers.copy_docstring(credentials.CredentialsWithTokenUri)
+    def with_token_uri(self, token_uri):
+        kwargs = self.constructor_args()
+        kwargs.update(token_url=token_uri)
+        return self.__class__(**kwargs)
 
     @classmethod
     def from_info(cls, info, **kwargs):
@@ -162,6 +169,8 @@ def from_info(cls, info, **kwargs):
             token_info_url=info.get("token_info_url"),
             client_id=info.get("client_id"),
             client_secret=info.get("client_secret"),
+            revoke_url=info.get("revoke_url"),
+            quota_project_id=info.get("quota_project_id"),
             **kwargs
         )
 

tests/data/external_account_authorized_user.json

@@ -0,0 +1,9 @@
+{
+    "type": "external_account_authorized_user",
+    "audience": "//iam.googleapis.com/locations/global/workforcePools/$WORKFORCE_POOL_ID/providers/$PROVIDER_ID",
+    "refresh_token": "refreshToken",
+    "token_url": "https://sts.googleapis.com/v1/oauth/token",
+    "token_info_url": "https://sts.googleapis.com/v1/instrospect",
+    "client_id": "clientId",
+    "client_secret": "clientSecret"
+}

tests/oauth2/test_sts.py

@@ -443,7 +443,7 @@ def test__make_request_success(self):
         self.assert_request_kwargs(request.call_args[1], headers, request_data)
         assert response == self.SUCCESS_RESPONSE
 
-    def test_refresh_token_failure(self):
+    def test_make_request_failure(self):
         """Test refresh token with failure response."""
         client = self.make_client(self.CLIENT_AUTH_BASIC)
         request = self.make_mock_request(

tests/test__default.py

@@ -152,6 +152,9 @@
     DATA_DIR, "impersonated_service_account_service_account_source.json"
 )
 
+EXTERNAL_ACCOUNT_AUTHORIZED_USER_FILE = os.path.join(
+    DATA_DIR, "external_account_authorized_user.json"
+)
 
 MOCK_CREDENTIALS = mock.Mock(spec=credentials.CredentialsWithQuotaProject)
 MOCK_CREDENTIALS.with_quota_project.return_value = MOCK_CREDENTIALS
@@ -541,23 +544,9 @@ def test__get_explicit_environ_credentials_no_env():
     assert _default._get_explicit_environ_credentials() == (None, None)
 
 
-def test_load_credentials_from_file_external_account_authorized_user(tmpdir):
-    config_file = tmpdir.join("config.json")
-    config_file.write(
-        json.dumps(
-            {
-                "type": "external_account_authorized_user",
-                "audience": "//iam.googleapis.com/locations/global/workforcePools/$WORKFORCE_POOL_ID/providers/$PROVIDER_ID",
-                "refresh_token": "refreshToken",
-                "token_url": "https://sts.googleapis.com/v1/oauth/token",
-                "token_info_url": "https://sts.googleapis.com/v1/instrospect",
-                "client_id": "clientId",
-                "client_secret": "clientSecret",
-            }
-        )
-    )
+def test_load_credentials_from_file_external_account_authorized_user():
     credentials, project_id = _default.load_credentials_from_file(
-        str(config_file), request=mock.sentinel.request
+        EXTERNAL_ACCOUNT_AUTHORIZED_USER_FILE, request=mock.sentinel.request
     )
 
     assert isinstance(credentials, external_account_authorized_user.Credentials)

tests/test_external_account_authorized_user.py

@@ -1,3 +1,17 @@
+# Copyright 2022 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 import datetime
 import json
 
@@ -15,14 +29,15 @@ class TestCredentials(object):
     TOKEN_INFO_URL = "https://sts.googleapis.com/v1/introspect"
     REVOKE_URL = "https://sts.googleapis.com/v1/revoke"
     PROJECT_NUMBER = "123456"
+    QUOTA_PROJECT_ID = "654321"
     POOL_ID = "POOL_ID"
     PROVIDER_ID = "PROVIDER_ID"
     AUDIENCE = (
         "//iam.googleapis.com/projects/{}"
         "/locations/global/workloadIdentityPools/{}"
         "/providers/{}"
     ).format(PROJECT_NUMBER, POOL_ID, PROVIDER_ID)
-    REFRESH_TOKEN = "refreshtoken"
+    REFRESH_TOKEN = "REFRESH_TOKEN"
     ACCESS_TOKEN = "ACCESS_TOKEN"
     CLIENT_ID = "username"
     CLIENT_SECRET = "password"
@@ -103,13 +118,22 @@ def test_refresh_auth_success(self):
 
     def test_refresh_auth_failure(self):
         request = self.make_mock_request(
-            status=http_client.BAD_REQUEST, data={"error": "XXXXXX"}
+            status=http_client.BAD_REQUEST,
+            data={
+                "error": "invalid_request",
+                "error_description": "Invalid subject token",
+                "error_uri": "https://tools.ietf.org/html/rfc6749",
+            },
         )
         creds = self.make_credentials()
 
-        with pytest.raises(exceptions.OAuthError):
+        with pytest.raises(exceptions.OAuthError) as excinfo:
             creds.refresh(request)
 
+        assert excinfo.match(
+            r"Error code invalid_request: Invalid subject token - https://tools.ietf.org/html/rfc6749"
+        )
+
         assert not creds.expiry
         assert not creds.expired
         assert not creds.token
@@ -127,3 +151,107 @@ def test_refresh_auth_failure(self):
                 "grant_type=refresh_token&refresh_token=" + self.REFRESH_TOKEN, "UTF-8"
             ),
         )
+
+    def test_info(self):
+        creds = self.make_credentials()
+        info = creds.info
+
+        assert info["audience"] == self.AUDIENCE
+        assert info["refresh_token"] == self.REFRESH_TOKEN
+        assert info["token_url"] == self.TOKEN_URL
+        assert info["token_info_url"] == self.TOKEN_INFO_URL
+        assert info["client_id"] == self.CLIENT_ID
+        assert info["client_secret"] == self.CLIENT_SECRET
+        assert "revoke_url" not in info
+        assert "quota_project_id" not in info
+
+    def test_info_full(self):
+        creds = self.make_credentials(
+            revoke_url=self.REVOKE_URL, quota_project_id=self.QUOTA_PROJECT_ID
+        )
+        info = creds.info
+
+        assert info["audience"] == self.AUDIENCE
+        assert info["refresh_token"] == self.REFRESH_TOKEN
+        assert info["token_url"] == self.TOKEN_URL
+        assert info["token_info_url"] == self.TOKEN_INFO_URL
+        assert info["client_id"] == self.CLIENT_ID
+        assert info["client_secret"] == self.CLIENT_SECRET
+        assert info["revoke_url"] == self.REVOKE_URL
+        assert info["quota_project_id"] == self.QUOTA_PROJECT_ID
+
+    def test_get_project_id(self):
+        creds = self.make_credentials()
+        assert creds.get_project_id() is None
+
+    def test_with_quota_project(self):
+        creds = self.make_credentials()
+        new_creds = creds.with_quota_project(self.QUOTA_PROJECT_ID)
+        assert new_creds._audience == creds._audience
+        assert new_creds._refresh_token == creds._refresh_token
+        assert new_creds._token_url == creds._token_url
+        assert new_creds._token_info_url == creds._token_info_url
+        assert new_creds._client_id == creds._client_id
+        assert new_creds._client_secret == creds._client_secret
+        assert new_creds._revoke_url == creds._revoke_url
+        assert new_creds._quota_project_id == self.QUOTA_PROJECT_ID
+
+    def test_with_token_uri(self):
+        creds = self.make_credentials()
+        new_creds = creds.with_token_uri("https://google.com")
+        assert new_creds._audience == creds._audience
+        assert new_creds._refresh_token == creds._refresh_token
+        assert new_creds._token_url == "https://google.com"
+        assert new_creds._token_info_url == creds._token_info_url
+        assert new_creds._client_id == creds._client_id
+        assert new_creds._client_secret == creds._client_secret
+        assert new_creds._revoke_url == creds._revoke_url
+        assert new_creds._quota_project_id == creds._quota_project_id
+
+    def test_from_file_required_options_only(self, tmpdir):
+        info = {
+            "audience": self.AUDIENCE,
+            "refresh_token": self.REFRESH_TOKEN,
+            "token_url": self.TOKEN_URL,
+            "token_info_url": self.TOKEN_INFO_URL,
+            "client_id": self.CLIENT_ID,
+            "client_secret": self.CLIENT_SECRET,
+        }
+        config_file = tmpdir.join("config.json")
+        config_file.write(json.dumps(info))
+        creds = external_account_authorized_user.Credentials.from_file(str(config_file))
+
+        assert isinstance(creds, external_account_authorized_user.Credentials)
+        creds._audience == self.AUDIENCE
+        creds._refresh_token == self.REFRESH_TOKEN
+        creds._token_url == self.TOKEN_URL
+        creds._token_info_url == self.TOKEN_INFO_URL
+        creds._client_id == self.CLIENT_ID
+        creds._client_secret == self.CLIENT_SECRET
+        creds._revoke_url is None
+        creds._quota_project_id is None
+
+    def test_from_file_full_options(self, tmpdir):
+        info = {
+            "audience": self.AUDIENCE,
+            "refresh_token": self.REFRESH_TOKEN,
+            "token_url": self.TOKEN_URL,
+            "token_info_url": self.TOKEN_INFO_URL,
+            "client_id": self.CLIENT_ID,
+            "client_secret": self.CLIENT_SECRET,
+            "revoke_url": self.REVOKE_URL,
+            "quota_project_id": self.QUOTA_PROJECT_ID,
+        }
+        config_file = tmpdir.join("config.json")
+        config_file.write(json.dumps(info))
+        creds = external_account_authorized_user.Credentials.from_file(str(config_file))
+
+        assert isinstance(creds, external_account_authorized_user.Credentials)
+        creds._audience == self.AUDIENCE
+        creds._refresh_token == self.REFRESH_TOKEN
+        creds._token_url == self.TOKEN_URL
+        creds._token_info_url == self.TOKEN_INFO_URL
+        creds._client_id == self.CLIENT_ID
+        creds._client_secret == self.CLIENT_SECRET
+        creds._revoke_url == self.REVOKE_URL
+        creds._quota_project_id == self.QUOTA_PROJECT_ID