feat: downscope token used for IAM DB AuthN (#488)

Author: jackwotherspoon

google/cloud/sql/connector/instance.py

@@ -261,10 +261,7 @@ def _auth_init(self, credentials: Optional[Credentials]) -> None:
             Credentials object used to authenticate connections to Cloud SQL server.
             If not specified, Application Default Credentials are used.
         """
-        scopes = [
-            "https://www.googleapis.com/auth/sqlservice.admin",
-            "https://www.googleapis.com/auth/cloud-platform",
-        ]
+        scopes = ["https://www.googleapis.com/auth/sqlservice.admin"]
         # if Credentials object is passed in, use for authentication
         if isinstance(credentials, Credentials):
             credentials = with_scopes_if_required(credentials, scopes=scopes)

google/cloud/sql/connector/refresh_utils.py

@@ -17,10 +17,11 @@
 
 import aiohttp
 import google.auth
-from google.auth.credentials import Credentials
+from google.auth.credentials import Credentials, Scoped
 import google.auth.transport.requests
-from typing import Any, Dict
+from typing import Any, Dict, List
 import datetime
+import copy
 import asyncio
 import logging
 
@@ -166,7 +167,7 @@ async def _get_ephemeral(
     elif not isinstance(pub_key, str):
         raise TypeError(f"pub_key must be of type str, got {type(pub_key)}")
 
-    if not credentials.valid or enable_iam_auth:
+    if not credentials.valid:
         request = google.auth.transport.requests.Request()
         credentials.refresh(request)
 
@@ -181,7 +182,9 @@ async def _get_ephemeral(
     data = {"public_key": pub_key}
 
     if enable_iam_auth:
-        data["access_token"] = credentials.token
+        # down-scope credentials with only IAM login scope (refreshes them too)
+        login_creds = _downscope_credentials(credentials)
+        data["access_token"] = login_creds.token
 
     resp = await client_session.post(
         url, headers=headers, json=data, raise_for_status=True
@@ -229,3 +232,38 @@ async def _is_valid(task: asyncio.Task) -> bool:
         # supress any errors from task
         logger.debug("Current instance metadata is invalid.")
     return False
+
+
+def _downscope_credentials(
+    credentials: Credentials,
+    scopes: List[str] = ["https://www.googleapis.com/auth/sqlservice.login"],
+) -> Credentials:
+    """Generate a down-scoped credential.
+
+    :type credentials: google.auth.credentials.Credentials
+    :param credentials
+        Credentials object used to generate down-scoped credentials.
+
+    :type scopes: List[str]
+    :param scopes
+        List of Google scopes to include in down-scoped credentials object.
+
+    :rtype: google.auth.credentials.Credentials
+    :returns: Down-scoped credentials object.
+    """
+    # credentials sourced from a service account or metadata are children of
+    # Scoped class and are capable of being re-scoped
+    if isinstance(credentials, Scoped):
+        scoped_creds = credentials.with_scopes(scopes=scopes)
+    # authenticated user credentials can not be re-scoped
+    else:
+        # create shallow copy to not overwrite scopes on default credentials
+        scoped_creds = copy.copy(credentials)
+        # overwrite '_scopes' to down-scope user credentials
+        # Cloud SDK reference: https://github.com/google-cloud-sdk-unofficial/google-cloud-sdk/blob/93920ccb6d2cce0fe6d1ce841e9e33410551d66b/lib/googlecloudsdk/command_lib/sql/generate_login_token_util.py#L116
+        scoped_creds._scopes = scopes
+    # down-scoped credentials require refresh, are invalid after being re-scoped
+    if not scoped_creds.valid:
+        request = google.auth.transport.requests.Request()
+        scoped_creds.refresh(request)
+    return scoped_creds

tests/conftest.py

@@ -30,10 +30,7 @@
 from google.cloud.sql.connector.instance import Instance
 from google.cloud.sql.connector.utils import generate_keys
 
-SCOPES = [
-    "https://www.googleapis.com/auth/sqlservice.admin",
-    "https://www.googleapis.com/auth/cloud-platform",
-]
+SCOPES = ["https://www.googleapis.com/auth/sqlservice.admin"]
 
 
 def pytest_addoption(parser: Any) -> None:

tests/unit/test_instance.py

@@ -294,7 +294,9 @@ async def test_perform_refresh_expiration(
     expiration = datetime.datetime.utcnow() + datetime.timedelta(minutes=1)
     setattr(instance._credentials, "expiry", expiration)
     setattr(instance, "_enable_iam_auth", True)
-    instance_metadata = await instance._perform_refresh()
+    # set all credentials to valid so downscoped credential does not refresh
+    with patch.object(Credentials, "valid", True):
+        instance_metadata = await instance._perform_refresh()
 
     # verify instance metadata object is returned
     assert isinstance(instance_metadata, InstanceMetadata)

tests/unit/test_refresh_utils.py

@@ -17,15 +17,17 @@
 
 import aiohttp
 from google.auth.credentials import Credentials
+import google.oauth2.credentials
 import pytest  # noqa F401 Needed to run the tests
-from mock import Mock
+from mock import Mock, patch
 from aioresponses import aioresponses
 import asyncio
 
 from google.cloud.sql.connector.refresh_utils import (
     _get_ephemeral,
     _get_metadata,
     _is_valid,
+    _downscope_credentials,
 )
 from google.cloud.sql.connector.utils import generate_keys
 
@@ -35,6 +37,7 @@
     instance_metadata_expired,
     FakeCSQLInstance,
 )
+from tests.conftest import SCOPES  # type: ignore
 
 
 @pytest.fixture
@@ -231,3 +234,34 @@ async def test_is_valid_with_expired_metadata() -> None:
     # task that returns class with expiration 10 mins in past
     task = asyncio.create_task(instance_metadata_expired())
     assert not await _is_valid(task)
+
+
+def test_downscope_credentials_service_account(fake_credentials: Credentials) -> None:
+    """
+    Test _downscope_credentials with google.oauth2.service_account.Credentials
+    which mimics an authenticated service account.
+    """
+    # set all credentials to valid to skip refreshing credentials
+    with patch.object(Credentials, "valid", True):
+        credentials = _downscope_credentials(fake_credentials)
+    # verify default credential scopes have not been altered
+    assert fake_credentials.scopes == SCOPES
+    # verify downscoped credentials have new scope
+    assert credentials.scopes == ["https://www.googleapis.com/auth/sqlservice.login"]
+    assert credentials != fake_credentials
+
+
+def test_downscope_credentials_user() -> None:
+    """
+    Test _downscope_credentials with google.oauth2.credentials.Credentials
+    which mimics an authenticated user.
+    """
+    creds = google.oauth2.credentials.Credentials("token", scopes=SCOPES)
+    # set all credentials to valid to skip refreshing credentials
+    with patch.object(Credentials, "valid", True):
+        credentials = _downscope_credentials(creds)
+    # verify default credential scopes have not been altered
+    assert creds.scopes == SCOPES
+    # verify downscoped credentials have new scope
+    assert credentials.scopes == ["https://www.googleapis.com/auth/sqlservice.login"]
+    assert credentials != creds