Repository: GoogleCloudPlatform/cloud-sql-python-connector (python)

Issue:
Redundant instance refreshes (with IAM auth)
### Bug Description

Hi, after #890, number of refreshes dropped and it is much better. However, I noticed that a lot of refreshes still occur just before token expiration. It happens every 30 second (rate limiter in `Instance`) for those 4 minutes ([refresh-buffer](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/blob/2a955322251a22e8c8b9d2494209b6db94b0d14d/google/cloud/sql/connector/refresh_utils.py#L36)) just before IAM token expiration.

The problem is that [here](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/blob/2a955322251a22e8c8b9d2494209b6db94b0d14d/google/cloud/sql/connector/instance.py#L337), expiration is set to lower value of ephemeral cert or IAM credentials. But IAM credentials are refreshed only when they are invalid [here](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/blob/2a955322251a22e8c8b9d2494209b6db94b0d14d/google/cloud/sql/connector/refresh_utils.py#L91) (or in get-ephemeral, but metadata are refreshed first).

Ephemeral cert expiration is usually 1 hour so let's say that we request instance refresh 30 minutes before IAM token expiration.
Expiration value sent to `_seconds_until_refresh()` computation will always be set to IAM token expiration because now it is just 30  minutes (and fewer) and its lower than ephemeral cert 1h. When it's less than 4 minutes before expiration, [this condition](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/blob/main/google/cloud/sql/connector/refresh_utils.py#L228) applies and seconds-until refresh starts returning `0` - meaning, refresh is performed instantly (well, with 30 seconds rate limiter). And this repeats every 30 seconds until IAM token is refreshed.

I don't know why expiration is set to IAM token expiration, can't we remove [this block](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/blob/2a955322251a22e8c8b9d2494209b6db94b0d14d/google/cloud/sql/connector/instance.py#L333-L337)? Ephemeral cert should be valid even when IAM token expired, shouldn't it?

I checked the golang implementation and it refreshes IAM token [always](https://github.com/GoogleCloudPlatform/cloud-sql-go-connector/blob/8a1e5218b018ee1ef1134a6b3ee3f5671a106885/internal/cloudsql/refresh.go#L164), not only when it is invalid.

The point is, 4 minutes before IAM token expiration, next refresh is computed based on IAM token, but IAM token is not actually refreshed, only ephemeral cert is refreshed but it's expiration time is not considered when scheduling next refresh, so it keeps refreshing ephemeral cert but schedules next refresh based on IAM token again and again until IAM expires.

I would create a PR but I was not sure if my first suggestion (don't set expiration to IAM token) is OK, it worked just fine in my project, but maybe I missed something about tokens and their validity. I would probably go with the second suggestion (always refresh IAM) as it is already used in go and also [cloud-sql-proxy](https://github.com/GoogleCloudPlatform/cloud-sql-proxy) which I used as well and saw no problem with it (but Go refreshes token only in get-ephemeral, not in get-metadata).

(For sake of completeness of my ideas, the third option is to refresh IAM token in [_get_metadata](https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/blob/main/google/cloud/sql/connector/refresh_utils.py#L91) (and `_get_ephemeral`) when it is just about to expire, i.e. `credentials.expiry - utcnow() < _refresh_buffer`, if you don't want to refresh it always, but note that "always" means just once an hour (considering default expiration is 1h) so I think "refresh always" is simpler in this case and number of refreshes is same).


### Steps to reproduce?

You can use my code from #890. You have to find out when your IAM token expires and then run the script at least for that long. Refresh logs should appear every 30 seconds for 4 minutes before IAM token expiration.

```py
import logging
import time

import google.cloud.sql.connector
import sqlalchemy

instance_logger = logging.getLogger(name='google.cloud.sql.connector.instance')
instance_logger.setLevel(logging.DEBUG)

logging.basicConfig(format='%(asctime)s [%(levelname)s]: %(message)s')
logging.debug('test start')

connector = google.cloud.sql.connector.Connector()
def getconn():
    return connector.connect(
        "<project>:<region>:<db-name>",
        "pg8000",
        user="<username>",
        db="<db-name>",
        enable_iam_auth=True,
        ip_type=google.cloud.sql.connector.IPTypes.PRIVATE,
    )

pool = sqlalchemy.create_engine("postgresql+pg8000://", creator=getconn)

while True:
    with pool.connect() as db_conn:
        db_conn.execute(sqlalchemy.text("SELECT 1"))
    time.sleep(5)
```

### Environment

1. OS type and version: Linux, but this should not matter :)
2. Python version: 3.10
3. Cloud SQL Python Connector version:  current `main`, with #890 merged.


Pull request description:
fix: use IAM login creds in expiration logic
Alternative to #894 

There are two credential objects managed in the Python Connector. 
1) Credential object used for authenticating the SQL Admin API calls
2) Credential object used for automatic IAM authentication that has login scopes and is embedded into the ephemeral cert to be later used as the "password" for an IAM login

When automatic IAM authentication is used, the expiration of the ephemeral cert and the IAM credentials object OAuth2 token should be compared when calculating the time until the next refresh operation occurs. This is because the cert and the OAuth2 token must both be valid in order for the successful connection.

Previously we were incorrectly using the expiry of credentials object for the SQL Admin APIs to calculate the next refresh when using IAM authentication. This PR fixes that by instead using the proper IAM credentials object.

Fixes #892

Patch:
diff --git a/google/cloud/sql/connector/instance.py b/google/cloud/sql/connector/instance.py
index 711346a..ebb9e08 100644
--- a/google/cloud/sql/connector/instance.py
+++ b/google/cloud/sql/connector/instance.py
@@ -29,8 +29,6 @@ from typing import (
 )
 
 import aiohttp
-from cryptography.hazmat.backends import default_backend
-from cryptography.x509 import load_pem_x509_certificate
 
 from google.auth.credentials import Credentials
 from google.cloud.sql.connector.exceptions import (
@@ -323,18 +321,7 @@ class Instance:
                 ephemeral_task.cancel()
                 raise
 
-            ephemeral_cert = await ephemeral_task
-
-            x509 = load_pem_x509_certificate(
-                ephemeral_cert.encode("UTF-8"), default_backend()
-            )
-            expiration = x509.not_valid_after
-
-            if self._enable_iam_auth:
-                if self._credentials is not None:
-                    token_expiration: datetime.datetime = self._credentials.expiry
-                if expiration > token_expiration:
-                    expiration = token_expiration
+            ephemeral_cert, expiration = await ephemeral_task
 
         except aiohttp.ClientResponseError as e:
             logger.debug(
diff --git a/google/cloud/sql/connector/refresh_utils.py b/google/cloud/sql/connector/refresh_utils.py
index 3798d31..6db1667 100644
--- a/google/cloud/sql/connector/refresh_utils.py
+++ b/google/cloud/sql/connector/refresh_utils.py
@@ -19,7 +19,10 @@ import asyncio
 import copy
 import datetime
 import logging
-from typing import Any, Dict, List, TYPE_CHECKING
+from typing import Any, Dict, List, Tuple, TYPE_CHECKING
+
+from cryptography.hazmat.backends import default_backend
+from cryptography.x509 import load_pem_x509_certificate
 
 from google.auth.credentials import Credentials, Scoped
 import google.auth.transport.requests
@@ -133,7 +136,7 @@ async def _get_ephemeral(
     instance: str,
     pub_key: str,
     enable_iam_auth: bool = False,
-) -> str:
+) -> Tuple[str, datetime.datetime]:
     """Asynchronously requests an ephemeral certificate from the Cloud SQL Instance.
 
     :type sqladmin_api_endpoint: str
@@ -204,7 +207,18 @@ async def _get_ephemeral(
 
     ret_dict = await resp.json()
 
-    return ret_dict["ephemeralCert"]["cert"]
+    ephemeral_cert: str = ret_dict["ephemeralCert"]["cert"]
+
+    # decode cert to read expiration
+    x509 = load_pem_x509_certificate(ephemeral_cert.encode("UTF-8"), default_backend())
+    expiration = x509.not_valid_after
+    # for IAM authentication OAuth2 token is embedded in cert so it
+    # must still be valid for successful connection
+    if enable_iam_auth:
+        token_expiration: datetime.datetime = login_creds.expiry
+        if expiration > token_expiration:
+            expiration = token_expiration
+    return ephemeral_cert, expiration
 
 
 def _seconds_until_refresh(
@@ -274,7 +288,6 @@ def _downscope_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)
+    request = google.auth.transport.requests.Request()
+    scoped_creds.refresh(request)
     return scoped_creds


Test patch:
diff --git a/tests/unit/test_instance.py b/tests/unit/test_instance.py
index 97fea15..00fe59d 100644
--- a/tests/unit/test_instance.py
+++ b/tests/unit/test_instance.py
@@ -244,7 +244,7 @@ async def test_perform_refresh(
 
 @pytest.mark.asyncio
 async def test_perform_refresh_expiration(
-    instance: Instance,
+    instance: Instance, fake_credentials: Credentials
 ) -> None:
     """
     Test that _perform_refresh returns ConnectionInfo with proper expiration.
@@ -254,10 +254,13 @@ async def test_perform_refresh_expiration(
     """
     # set credentials expiration to 1 minute from now
     expiration = datetime.datetime.utcnow() + datetime.timedelta(minutes=1)
-    setattr(instance._credentials, "expiry", expiration)
     setattr(instance, "_enable_iam_auth", True)
-    # set all credentials to valid so downscoped credential does not refresh
-    with patch.object(Credentials, "valid", True):
+    # set downscoped credential to mock
+    with patch(
+        "google.cloud.sql.connector.refresh_utils._downscope_credentials"
+    ) as mock_auth:
+        setattr(fake_credentials, "expiry", expiration)
+        mock_auth.return_value = fake_credentials
         instance_metadata = await instance._perform_refresh()
 
     # verify instance metadata object is returned
diff --git a/tests/unit/test_refresh_utils.py b/tests/unit/test_refresh_utils.py
index 3753067..5688d89 100644
--- a/tests/unit/test_refresh_utils.py
+++ b/tests/unit/test_refresh_utils.py
@@ -28,6 +28,7 @@ from mocks import (  # type: ignore
 )
 import pytest  # noqa F401 Needed to run the tests
 
+import google.auth
 from google.auth.credentials import Credentials
 from google.cloud.sql.connector.refresh_utils import (
     _downscope_credentials,
@@ -76,12 +77,13 @@ async def test_get_ephemeral(
                 instance,
                 pub_key,
             )
-    result = result.strip()  # remove any trailing whitespace
-    result = result.split("\n")
+    cert, _ = result
+    cert = cert.strip()  # remove any trailing whitespace
+    cert = cert.split("\n")
 
     assert (
-        result[0] == "-----BEGIN CERTIFICATE-----"
-        and result[len(result) - 1] == "-----END CERTIFICATE-----"
+        cert[0] == "-----BEGIN CERTIFICATE-----"
+        and cert[len(cert) - 1] == "-----END CERTIFICATE-----"
     )
 
 
@@ -287,19 +289,20 @@ async def test_is_valid_with_expired_metadata() -> None:
     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
+# TODO: https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/issues/901
+# 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.
+#     """
+#     # override actual refresh URI
+#     setattr(fake_credentials, "with_scopes", google.auth.credentials.Credentials(scopes=["https://www.googleapis.com/auth/sqlservice.login"]))
+#     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:
@@ -308,8 +311,10 @@ def test_downscope_credentials_user() -> None:
     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):
+    # override actual refresh URI
+    with patch.object(
+        google.oauth2.credentials.Credentials, "refresh", lambda *args: None
+    ):
         credentials = _downscope_credentials(creds)
     # verify default credential scopes have not been altered
     assert creds.scopes == SCOPES