fix: only add IAM scope to credentials that can change scopes (#451)

Author: busunkim96

CONTRIBUTING.rst

@@ -43,37 +43,69 @@ To run a single session, specify it with ``nox -s``::
 
     $ nox -f system_tests/noxfile.py -s service_account
 
+
+Project and Credentials Setup
+-------------------------------
+
+Enable the IAM Service Account Credentials API on the project.
+
 To run system tests locally, you will need to set up a data directory ::
 
     $ mkdir system_tests/data
 
-Add a service account file and authorized user file to the data directory.
-Your directory should look like this ::
+Your directory should look like this. Follow the instructions below for creating each file. ::
 
   system_tests/
       data/
-        service_account.json
         authorized_user.json
+        impersonated_service_account.json
+        service_account.json
 
-The files must be named exactly ``service_account.json``
-and ``authorized_user.json``. See `Creating and Managing Service Account Keys`_ for how to
-obtain a service account. 
+
+``authorized_user.json``
+~~~~~~~~~~~~~~~~~~~~~~~~
 
 Use the `gcloud CLI`_ to get an authorized user file ::
 
     $ gcloud auth application-default login --scopes=https://www.googleapis.com/auth/userinfo.email,https://www.googleapis.com/auth/cloud-platform,openid
 
 You will see something like::
 
-    Credentials saved to file: [/usr/local/home/.config/gcloud/application_default_credentials.json]```
+    Credentials saved to file: [/usr/local/home/.config/gcloud/application_default_credentials.json]
 
 Copy the contents of the file to ``authorized_user.json``.
 
-.. _Creating and Managing Service Account Keys: https://cloud.google.com/iam/docs/creating-managing-service-account-keys
+Open the IAM page of the Google Cloud Console. Grant the user the `Service Account Token Creator Role`.
+This will allow the user to impersonate service accounts on the project.
+
 .. _gcloud CLI: https://cloud.google.com/sdk/gcloud/
 
+
+``service_account.json``
+~~~~~~~~~~~~~~~~~~~~~~~~
+
+Follow `Creating and Managing Service Account Keys`_ to create a service account. 
+
+Copy the credentials file to ``service_account.json``.
+
+Grant the account associated with ``service_account.json`` the following roles.
+
+- App Engine Admin (for App Engine tests)
+- Service Account Token Creator (for impersonated credentials tests)
+- Pub/Sub Viewer (for gRPC tests)
+- Storage Object Viewer (for impersonated credentials tests)
+
+``impersonated_service_account.json``
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Follow `Creating and Managing Service Account Keys`_ to create a service account. 
+
+Copy the credentials file to ``impersonated_service_account.json``.
+
+.. _Creating and Managing Service Account Keys: https://cloud.google.com/iam/docs/creating-managing-service-account-keys
+
 App Engine System Tests
-^^^^^^^^^^^^^^^^^^^^^^^
+~~~~~~~~~~~~~~~~~~~~~~~~
 
 To run the App Engine tests, you wil need to deploy a default App Engine service.
 If you already have a default service associated with your project, you can skip this step.

google/auth/impersonated_credentials.py

@@ -205,7 +205,11 @@ def __init__(
         super(Credentials, self).__init__()
 
         self._source_credentials = copy.copy(source_credentials)
-        self._source_credentials._scopes = _IAM_SCOPE
+        # Service account source credentials must have the _IAM_SCOPE
+        # added to refresh correctly. User credentials cannot have
+        # their original scopes modified.
+        if isinstance(self._source_credentials, credentials.Scoped):
+            self._source_credentials = self._source_credentials.with_scopes(_IAM_SCOPE)
         self._target_principal = target_principal
         self._target_scopes = target_scopes
         self._delegates = delegates

system_tests/conftest.py

@@ -25,6 +25,9 @@
 
 HERE = os.path.dirname(__file__)
 DATA_DIR = os.path.join(HERE, "data")
+IMPERSONATED_SERVICE_ACCOUNT_FILE = os.path.join(
+    DATA_DIR, "impersonated_service_account.json"
+)
 SERVICE_ACCOUNT_FILE = os.path.join(DATA_DIR, "service_account.json")
 AUTHORIZED_USER_FILE = os.path.join(DATA_DIR, "authorized_user.json")
 URLLIB3_HTTP = urllib3.PoolManager(retries=False)
@@ -39,6 +42,12 @@ def service_account_file():
     yield SERVICE_ACCOUNT_FILE
 
 
+@pytest.fixture
+def impersonated_service_account_file():
+    """The full path to a valid service account key file."""
+    yield IMPERSONATED_SERVICE_ACCOUNT_FILE
+
+
 @pytest.fixture
 def authorized_user_file():
     """The full path to a valid authorized user file."""

system_tests/noxfile.py

@@ -170,7 +170,8 @@ def configure_cloud_sdk(session, application_default_credentials, project=False)
 # Test sesssions
 
 TEST_DEPENDENCIES = ["pytest", "requests"]
-PYTHON_VERSIONS=['2.7', '3.7']
+PYTHON_VERSIONS = ["2.7", "3.7"]
+
 
 @nox.session(python=PYTHON_VERSIONS)
 def service_account(session):
@@ -186,6 +187,13 @@ def oauth2_credentials(session):
     session.run("pytest", "test_oauth2_credentials.py")
 
 
+@nox.session(python=PYTHON_VERSIONS)
+def impersonated_credentials(session):
+    session.install(*TEST_DEPENDENCIES)
+    session.install(LIBRARY_DIR)
+    session.run("pytest", "test_impersonated_credentials.py")
+
+
 @nox.session(python=PYTHON_VERSIONS)
 def default_explicit_service_account(session):
     session.env[EXPLICIT_CREDENTIALS_ENV] = SERVICE_ACCOUNT_FILE

system_tests/secrets.tar.enc

@@ -0,0 +1,99 @@
+# Copyright 2020 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 json
+import pytest
+
+import google.oauth2.credentials
+from google.oauth2 import service_account
+import google.auth.impersonated_credentials
+from google.auth import _helpers
+
+
+GOOGLE_OAUTH2_TOKEN_ENDPOINT = "https://oauth2.googleapis.com/token"
+
+
+@pytest.fixture
+def service_account_credentials(service_account_file):
+    yield service_account.Credentials.from_service_account_file(service_account_file)
+
+
+@pytest.fixture
+def impersonated_service_account_credentials(impersonated_service_account_file):
+    yield service_account.Credentials.from_service_account_file(
+        impersonated_service_account_file
+    )
+
+
+def test_refresh_with_user_credentials_as_source(
+    authorized_user_file,
+    impersonated_service_account_credentials,
+    http_request,
+    token_info,
+):
+    with open(authorized_user_file, "r") as fh:
+        info = json.load(fh)
+
+    source_credentials = google.oauth2.credentials.Credentials(
+        None,
+        refresh_token=info["refresh_token"],
+        token_uri=GOOGLE_OAUTH2_TOKEN_ENDPOINT,
+        client_id=info["client_id"],
+        client_secret=info["client_secret"],
+        # The source credential needs this scope for the generateAccessToken request
+        # The user must also have `Service Account Token Creator` on the project
+        # that owns the impersonated service account.
+        # See https://cloud.google.com/iam/docs/creating-short-lived-service-account-credentials
+        scopes=["https://www.googleapis.com/auth/cloud-platform"],
+    )
+
+    source_credentials.refresh(http_request)
+
+    target_scopes = [
+        "https://www.googleapis.com/auth/devstorage.read_only",
+        "https://www.googleapis.com/auth/analytics",
+    ]
+    target_credentials = google.auth.impersonated_credentials.Credentials(
+        source_credentials=source_credentials,
+        target_principal=impersonated_service_account_credentials.service_account_email,
+        target_scopes=target_scopes,
+        lifetime=100,
+    )
+
+    target_credentials.refresh(http_request)
+    assert target_credentials.token
+
+
+def test_refresh_with_service_account_credentials_as_source(
+    http_request,
+    service_account_credentials,
+    impersonated_service_account_credentials,
+    token_info,
+):
+    source_credentials = service_account_credentials.with_scopes(["email"])
+    source_credentials.refresh(http_request)
+    assert source_credentials.token
+
+    target_scopes = [
+        "https://www.googleapis.com/auth/devstorage.read_only",
+        "https://www.googleapis.com/auth/analytics",
+    ]
+    target_credentials = google.auth.impersonated_credentials.Credentials(
+        source_credentials=source_credentials,
+        target_principal=impersonated_service_account_credentials.service_account_email,
+        target_scopes=target_scopes,
+    )
+
+    target_credentials.refresh(http_request)
+    assert target_credentials.token

system_tests/test_impersonated_credentials.py

@@ -26,6 +26,7 @@
 from google.auth import impersonated_credentials
 from google.auth import transport
 from google.auth.impersonated_credentials import Credentials
+from google.oauth2 import credentials
 from google.oauth2 import service_account
 
 DATA_DIR = os.path.join(os.path.dirname(__file__), "", "data")
@@ -102,17 +103,30 @@ class TestImpersonatedCredentials(object):
     SOURCE_CREDENTIALS = service_account.Credentials(
         SIGNER, SERVICE_ACCOUNT_EMAIL, TOKEN_URI
     )
+    USER_SOURCE_CREDENTIALS = credentials.Credentials(token="ABCDE")
 
-    def make_credentials(self, lifetime=LIFETIME, target_principal=TARGET_PRINCIPAL):
+    def make_credentials(
+        self,
+        source_credentials=SOURCE_CREDENTIALS,
+        lifetime=LIFETIME,
+        target_principal=TARGET_PRINCIPAL,
+    ):
 
         return Credentials(
-            source_credentials=self.SOURCE_CREDENTIALS,
+            source_credentials=source_credentials,
             target_principal=target_principal,
             target_scopes=self.TARGET_SCOPES,
             delegates=self.DELEGATES,
             lifetime=lifetime,
         )
 
+    def test_make_from_user_credentials(self):
+        credentials = self.make_credentials(
+            source_credentials=self.USER_SOURCE_CREDENTIALS
+        )
+        assert not credentials.valid
+        assert credentials.expired
+
     def test_default_state(self):
         credentials = self.make_credentials()
         assert not credentials.valid