Merge branch 'main' into warn_rsa

Author: daniel-sanche

google/auth/_default.py

@@ -330,31 +330,31 @@ def _get_explicit_environ_credentials(quota_project_id=None):
     from google.auth import _cloud_sdk
 
     cloud_sdk_adc_path = _cloud_sdk.get_application_default_credentials_path()
-    explicit_file = os.environ.get(environment_vars.CREDENTIALS)
+    explicit_file = os.environ.get(environment_vars.CREDENTIALS, "")
 
     _LOGGER.debug(
-        "Checking %s for explicit credentials as part of auth process...", explicit_file
+        "Checking '%s' for explicit credentials as part of auth process...",
+        explicit_file,
     )
 
-    if explicit_file is not None and explicit_file == cloud_sdk_adc_path:
+    if explicit_file != "" and explicit_file == cloud_sdk_adc_path:
         # Cloud sdk flow calls gcloud to fetch project id, so if the explicit
         # file path is cloud sdk credentials path, then we should fall back
         # to cloud sdk flow, otherwise project id cannot be obtained.
         _LOGGER.debug(
-            "Explicit credentials path %s is the same as Cloud SDK credentials path, fall back to Cloud SDK credentials flow...",
+            "Explicit credentials path '%s' is the same as Cloud SDK credentials path, fall back to Cloud SDK credentials flow...",
             explicit_file,
         )
         return _get_gcloud_sdk_credentials(quota_project_id=quota_project_id)
 
-    if explicit_file is not None:
+    if explicit_file != "":
         with warnings.catch_warnings():
             warnings.simplefilter("ignore", DeprecationWarning)
             credentials, project_id = load_credentials_from_file(
                 os.environ[environment_vars.CREDENTIALS],
                 quota_project_id=quota_project_id,
             )
             credentials._cred_file_path = f"{explicit_file} file via the GOOGLE_APPLICATION_CREDENTIALS environment variable"
-
             return credentials, project_id
 
     else:

google/auth/_helpers.py

@@ -124,6 +124,26 @@ def utcnow():
     return now
 
 
+def utcfromtimestamp(timestamp):
+    """Returns the UTC datetime from a timestamp.
+
+    Args:
+        timestamp (float): The timestamp to convert.
+
+    Returns:
+        datetime: The time in UTC.
+    """
+    # We used datetime.utcfromtimestamp() before, since it's deprecated from
+    # python 3.12, we are using datetime.fromtimestamp(timestamp, timezone.utc)
+    # now. "utcfromtimestamp()" is offset-native (no timezone info), but
+    # "fromtimestamp(timestamp, timezone.utc)" is offset-aware (with timezone
+    # info). This will cause datetime comparison problem. For backward
+    # compatibility, we need to remove the timezone info.
+    dt = datetime.datetime.fromtimestamp(timestamp, tz=datetime.timezone.utc)
+    dt = dt.replace(tzinfo=None)
+    return dt
+
+
 def datetime_to_secs(value):
     """Convert a datetime object to the number of seconds since the UNIX epoch.
 

google/auth/app_engine.py

@@ -22,7 +22,6 @@
     https://cloud.google.com/appengine/docs/python/appidentity/
 """
 
-import datetime
 
 from google.auth import _helpers
 from google.auth import credentials
@@ -128,7 +127,7 @@ def refresh(self, request):
         scopes = self._scopes if self._scopes is not None else self._default_scopes
         # pylint: disable=unused-argument
         token, ttl = app_identity.get_access_token(scopes, self._service_account_id)
-        expiry = datetime.datetime.utcfromtimestamp(ttl)
+        expiry = _helpers.utcfromtimestamp(ttl)
 
         self.token, self.expiry = token, expiry
 

google/auth/compute_engine/_metadata.py

@@ -101,6 +101,10 @@ def _get_metadata_ip_root(use_mtls: bool):
 except ValueError:  # pragma: NO COVER
     _METADATA_DEFAULT_TIMEOUT = 3
 
+# This is used to disable checking for the GCE metadata server and directly
+# assuming it's not available.
+_NO_GCE_CHECK = os.getenv(environment_vars.NO_GCE_CHECK) == "true"
+
 # Detect GCE Residency
 _GOOGLE = "Google"
 _GCE_PRODUCT_NAME_FILE = "/sys/class/dmi/id/product_name"
@@ -116,6 +120,9 @@ def is_on_gce(request):
     Returns:
         bool: True if the code runs on Google Compute Engine, False otherwise.
     """
+    if _NO_GCE_CHECK:
+        return False
+
     if ping(request):
         return True
 

google/auth/compute_engine/credentials.py

@@ -498,7 +498,7 @@ def _call_metadata_identity_endpoint(self, request):
             raise new_exc from caught_exc
 
         _, payload, _, _ = jwt._unverified_decode(id_token)
-        return id_token, datetime.datetime.utcfromtimestamp(payload["exp"])
+        return id_token, _helpers.utcfromtimestamp(payload["exp"])
 
     def refresh(self, request):
         """Refreshes the ID token.

google/auth/environment_vars.py

@@ -60,6 +60,12 @@
 """Environment variable providing an alternate ip:port to be used for ip-only
 GCE metadata requests."""
 
+NO_GCE_CHECK = "NO_GCE_CHECK"
+"""Environment variable controlling whether to check if running on GCE or not.
+
+The default value is false. Users have to explicitly set this value to true
+in order to disable the GCE check."""
+
 GCE_METADATA_MTLS_MODE = "GCE_METADATA_MTLS_MODE"
 """Environment variable controlling the mTLS behavior for GCE metadata requests.
 

google/auth/impersonated_credentials.py

@@ -649,7 +649,7 @@ def refresh(self, request):
             raise new_exc from caught_exc
 
         self.token = id_token
-        self.expiry = datetime.utcfromtimestamp(
+        self.expiry = _helpers.utcfromtimestamp(
             jwt.decode(id_token, verify=False)["exp"]
         )
 

google/oauth2/_client.py

@@ -368,7 +368,7 @@ def call_iam_generate_id_token_endpoint(
         raise new_exc from caught_exc
 
     payload = jwt.decode(id_token, verify=False)
-    expiry = datetime.datetime.utcfromtimestamp(payload["exp"])
+    expiry = _helpers.utcfromtimestamp(payload["exp"])
 
     return id_token, expiry
 
@@ -420,7 +420,7 @@ def id_token_jwt_grant(request, token_uri, assertion, can_retry=True):
         raise new_exc from caught_exc
 
     payload = jwt.decode(id_token, verify=False)
-    expiry = datetime.datetime.utcfromtimestamp(payload["exp"])
+    expiry = _helpers.utcfromtimestamp(payload["exp"])
 
     return id_token, expiry, response_data
 

google/oauth2/_client_async.py

@@ -23,12 +23,12 @@
 .. _Section 3.1 of rfc6749: https://tools.ietf.org/html/rfc6749#section-3.2
 """
 
-import datetime
 import http.client as http_client
 import json
 import urllib
 
 from google.auth import _exponential_backoff
+from google.auth import _helpers
 from google.auth import exceptions
 from google.auth import jwt
 from google.oauth2 import _client as client
@@ -227,7 +227,7 @@ async def id_token_jwt_grant(request, token_uri, assertion, can_retry=True):
         raise new_exc from caught_exc
 
     payload = jwt.decode(id_token, verify=False)
-    expiry = datetime.datetime.utcfromtimestamp(payload["exp"])
+    expiry = _helpers.utcfromtimestamp(payload["exp"])
 
     return id_token, expiry, response_data
 

tests/compute_engine/test__metadata.py

@@ -108,6 +108,20 @@ def test_is_on_gce_ping_success():
     assert _metadata.is_on_gce(request)
 
 
+def test_is_on_gce_no_gce_check():
+    request = make_request("", headers=_metadata._METADATA_HEADERS)
+
+    os.environ[environment_vars.NO_GCE_CHECK] = "true"
+    importlib.reload(_metadata)
+
+    try:
+        assert not _metadata.is_on_gce(request)
+        assert request.call_count == 0
+    finally:
+        del os.environ[environment_vars.NO_GCE_CHECK]
+        importlib.reload(_metadata)
+
+
 @mock.patch("os.name", new="nt")
 def test_is_on_gce_windows_success():
     request = make_request("", headers={_metadata._METADATA_FLAVOR_HEADER: "meep"})