Merge branch 'main' into gcp_detect_retries

Author: chalmerlowe

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/compute_engine/_metadata.py

@@ -110,6 +110,10 @@ def _get_metadata_ip_root(use_mtls: bool):
 except ValueError:  # pragma: NO COVER
     _METADATA_DETECT_RETRIES = 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"
@@ -125,6 +129,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/environment_vars.py

@@ -70,6 +70,12 @@
 attempted on metadata lookup.
 """
 
+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.
 

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"})