feat: Add smbios check to detect GCE residency

google/auth/_default.py

@@ -281,7 +281,7 @@ def _get_gce_credentials(request=None, quota_project_id=None):
  if request is None:
  request = google.auth.transport._http_client.Request()
 
- if _metadata.ping(request=request):
+ if _metadata.is_on_gce(request=request):
  # Get the project ID.
  try:
  project_id = _metadata.get_project_id(request=request)

google/auth/compute_engine/_metadata.py

@@ -59,6 +59,47 @@
 except ValueError: # pragma: NO COVER
  _METADATA_DEFAULT_TIMEOUT = 3
 
+# Detect GCE Residency
+_GOOGLE = "Google"
+_GCE_PRODUCT_NAME_FILE = "/sys/class/dmi/id/product_name"
+
+
+def is_on_gce(request):
+ """Checks to see if the code runs on Google Compute Engine
+
+ Args:
+ request (google.auth.transport.Request): A callable used to make
+ HTTP requests.
+
+ Returns:
+ bool: True if the code runs on Google Compute Engine, False otherwise.
+ """
+ if ping(request):
+ return True
+
+ if os.name == "nt":
+ # TODO: implement GCE residency detection on Windows
+ return False
+
+ # Detect GCE residency on Linux
+ return detect_gce_residency_linux()
+
+
+def detect_gce_residency_linux():
+ """Detect Google Compute Engine residency by smbios check on Linux
+
+ Returns:
+ bool: True if the GCE product name file is detected, False otherwise.
+ """
+ try:
+ with open(_GCE_PRODUCT_NAME_FILE, "r") as file_obj:
+ content = file_obj.read().strip()
+
+ except Exception:
+ return False
+
+ return content.startswith(_GOOGLE)
+
 
 def ping(request, timeout=_METADATA_DEFAULT_TIMEOUT, retry_count=3):
  """Checks to see if the metadata server is available.

tests/compute_engine/data/smbios_product_name

@@ -0,0 +1 @@
+Google Compute Engine

tests/compute_engine/data/smbios_product_name_non_google

@@ -0,0 +1 @@
+ABC Compute Engine

tests/compute_engine/test__metadata.py

@@ -29,6 +29,15 @@
 
 PATH = "instance/service-accounts/default"
 
+DATA_DIR = os.path.join(os.path.dirname(__file__), "data")
+SMBIOS_PRODUCT_NAME_FILE = os.path.join(DATA_DIR, "smbios_product_name")
+SMBIOS_PRODUCT_NAME_NONEXISTENT_FILE = os.path.join(
+ DATA_DIR, "smbios_product_name_nonexistent"
+)
+SMBIOS_PRODUCT_NAME_NON_GOOGLE = os.path.join(
+ DATA_DIR, "smbios_product_name_non_google"
+)
+
 
 def make_request(data, status=http_client.OK, headers=None, retry=False):
  response = mock.create_autospec(transport.Response, instance=True)
@@ -45,6 +54,39 @@ def make_request(data, status=http_client.OK, headers=None, retry=False):
  return request
 
 
+def test_detect_gce_residency_linux_success():
+ _metadata._GCE_PRODUCT_NAME_FILE = SMBIOS_PRODUCT_NAME_FILE
+ assert _metadata.detect_gce_residency_linux()
+
+
+def test_detect_gce_residency_linux_non_google():
+ _metadata._GCE_PRODUCT_NAME_FILE = SMBIOS_PRODUCT_NAME_NON_GOOGLE
+ assert not _metadata.detect_gce_residency_linux()
+
+
+def test_detect_gce_residency_linux_nonexistent():
+ _metadata._GCE_PRODUCT_NAME_FILE = SMBIOS_PRODUCT_NAME_NONEXISTENT_FILE
+ assert not _metadata.detect_gce_residency_linux()
+
+
+def test_is_on_gce_ping_success():
+ request = make_request("", headers=_metadata._METADATA_HEADERS)
+ assert _metadata.is_on_gce(request)
+
+
+@mock.patch("os.name", new="nt")
+def test_is_on_gce_windows_success():
+ request = make_request("", headers={_metadata._METADATA_FLAVOR_HEADER: "meep"})
+ assert not _metadata.is_on_gce(request)
+
+
+@mock.patch("os.name", new="posix")
+def test_is_on_gce_linux_success():
+ request = make_request("", headers={_metadata._METADATA_FLAVOR_HEADER: "meep"})
+ _metadata._GCE_PRODUCT_NAME_FILE = SMBIOS_PRODUCT_NAME_FILE
+ assert _metadata.is_on_gce(request)
+
+
 def test_ping_success():
  request = make_request("", headers=_metadata._METADATA_HEADERS)
 

tests/test__default.py

@@ -781,7 +781,7 @@ def test__get_gae_credentials_no_apis():
 
 
 @mock.patch(
- "google.auth.compute_engine._metadata.ping", return_value=True, autospec=True
+ "google.auth.compute_engine._metadata.is_on_gce", return_value=True, autospec=True
 )
 @mock.patch(
  "google.auth.compute_engine._metadata.get_project_id",
@@ -796,7 +796,7 @@ def test__get_gce_credentials(unused_get, unused_ping):
 
 
 @mock.patch(
- "google.auth.compute_engine._metadata.ping", return_value=False, autospec=True
+ "google.auth.compute_engine._metadata.is_on_gce", return_value=False, autospec=True
 )
 def test__get_gce_credentials_no_ping(unused_ping):
  credentials, project_id = _default._get_gce_credentials()
@@ -806,7 +806,7 @@ def test__get_gce_credentials_no_ping(unused_ping):
 
 
 @mock.patch(
- "google.auth.compute_engine._metadata.ping", return_value=True, autospec=True
+ "google.auth.compute_engine._metadata.is_on_gce", return_value=True, autospec=True
 )
 @mock.patch(
  "google.auth.compute_engine._metadata.get_project_id",
@@ -831,7 +831,7 @@ def test__get_gce_credentials_no_compute_engine():
 
 
 @mock.patch(
- "google.auth.compute_engine._metadata.ping", return_value=False, autospec=True
+ "google.auth.compute_engine._metadata.is_on_gce", return_value=False, autospec=True
 )
 def test__get_gce_credentials_explicit_request(ping):
  _default._get_gce_credentials(mock.sentinel.request)
@@ -1246,7 +1246,7 @@ def test_quota_project_from_environment(get_adc_path):
 
 
 @mock.patch(
- "google.auth.compute_engine._metadata.ping", return_value=True, autospec=True
+ "google.auth.compute_engine._metadata.is_on_gce", return_value=True, autospec=True
 )
 @mock.patch(
  "google.auth.compute_engine._metadata.get_project_id",

tests_async/test__default_async.py

@@ -366,7 +366,7 @@ def test__get_gae_credentials_no_apis():
 
 
 @mock.patch(
- "google.auth.compute_engine._metadata.ping", return_value=True, autospec=True
+ "google.auth.compute_engine._metadata.is_on_gce", return_value=True, autospec=True
 )
 @mock.patch(
  "google.auth.compute_engine._metadata.get_project_id",
@@ -381,7 +381,7 @@ def test__get_gce_credentials(unused_get, unused_ping):
 
 
 @mock.patch(
- "google.auth.compute_engine._metadata.ping", return_value=False, autospec=True
+ "google.auth.compute_engine._metadata.is_on_gce", return_value=False, autospec=True
 )
 def test__get_gce_credentials_no_ping(unused_ping):
  credentials, project_id = _default._get_gce_credentials()
@@ -391,7 +391,7 @@ def test__get_gce_credentials_no_ping(unused_ping):
 
 
 @mock.patch(
- "google.auth.compute_engine._metadata.ping", return_value=True, autospec=True
+ "google.auth.compute_engine._metadata.is_on_gce", return_value=True, autospec=True
 )
 @mock.patch(
  "google.auth.compute_engine._metadata.get_project_id",
@@ -416,7 +416,7 @@ def test__get_gce_credentials_no_compute_engine():
 
 
 @mock.patch(
- "google.auth.compute_engine._metadata.ping", return_value=False, autospec=True
+ "google.auth.compute_engine._metadata.is_on_gce", return_value=False, autospec=True
 )
 def test__get_gce_credentials_explicit_request(ping):
  _default._get_gce_credentials(mock.sentinel.request)