feat: Add smbios check to detect GCE residency (#1276)

* feat: smbios check for GCE residency

* Refresh test token

Author: wangyutongg

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.

system_tests/secrets.tar.enc

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

tests/compute_engine/data/smbios_product_name

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

tests/compute_engine/data/smbios_product_name_non_google

@@ -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/compute_engine/test__metadata.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/test__default.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)