Improve GCP creds validation (dstackai#2322)

Author: r4victor

src/dstack/_internal/core/backends/gcp/auth.py

@@ -1,10 +1,11 @@
 import json
 from typing import Optional, Tuple
 
+import google.api_core.exceptions
 import google.auth
+import google.cloud.compute_v1 as compute_v1
 from google.auth.credentials import Credentials
 from google.auth.exceptions import DefaultCredentialsError
-from google.cloud import storage
 from google.oauth2 import service_account
 
 from dstack._internal.core.errors import BackendAuthError
@@ -16,13 +17,16 @@
 from dstack._internal.core.models.common import is_core_model_instance
 
 
-def authenticate(creds: AnyGCPCreds) -> Tuple[Credentials, Optional[str]]:
-    """
-    :raises BackendAuthError:
-    :return: GCP credentials and project_id
-    """
-    credentials, project_id = get_credentials(creds)
-    validate_credentials(credentials)
+def authenticate(creds: AnyGCPCreds, project_id: Optional[str] = None) -> Tuple[Credentials, str]:
+    credentials, credentials_project_id = get_credentials(creds)
+    if project_id is None:
+        # If project_id is not specified explicitly, try using credentials' project_id.
+        # Explicit project_id takes precedence bacause credentials' project_id may be irrelevant.
+        # For example, with Workload Identity Federation for GKE, it's cluster project_id.
+        project_id = credentials_project_id
+    if project_id is None:
+        raise BackendAuthError("Credentials require project_id to be specified")
+    validate_credentials(credentials, project_id)
     return credentials, project_id
 
 
@@ -40,17 +44,19 @@ def get_credentials(creds: AnyGCPCreds) -> Tuple[Credentials, Optional[str]]:
     try:
         default_credentials, project_id = google.auth.default()
     except DefaultCredentialsError:
-        raise BackendAuthError()
+        raise BackendAuthError("Failed to find default credentials")
 
     return default_credentials, project_id
 
 
-def validate_credentials(credentials: Credentials):
+def validate_credentials(credentials: Credentials, project_id: str):
     try:
-        storage_client = storage.Client(credentials=credentials)
-        storage_client.list_buckets(max_results=1)
+        regions_client = compute_v1.RegionsClient(credentials=credentials)
+        regions_client.list(project=project_id)
+    except google.api_core.exceptions.NotFound:
+        raise BackendAuthError(f"project_id {project_id} not found")
     except Exception:
-        raise BackendAuthError()
+        raise BackendAuthError("Insufficient permissions")
 
 
 def default_creds_available() -> bool:

src/dstack/_internal/core/backends/gcp/compute.py

@@ -70,7 +70,7 @@ class GCPCompute(Compute):
     def __init__(self, config: GCPConfig):
         super().__init__()
         self.config = config
-        self.credentials, self.project_id = auth.authenticate(config.creds)
+        self.credentials, _ = auth.authenticate(config.creds, self.config.project_id)
         self.instances_client = compute_v1.InstancesClient(credentials=self.credentials)
         self.firewalls_client = compute_v1.FirewallsClient(credentials=self.credentials)
         self.regions_client = compute_v1.RegionsClient(credentials=self.credentials)

src/dstack/_internal/server/services/backends/configurators/gcp.py

@@ -127,10 +127,6 @@ def get_default_configs(self) -> List[GCPConfigInfoWithCreds]:
             _, project_id = auth.authenticate(GCPDefaultCreds())
         except BackendAuthError:
             return []
-
-        if project_id is None:
-            return []
-
         return [
             GCPConfigInfoWithCreds(
                 project_id=project_id,
@@ -152,16 +148,15 @@ def get_config_values(self, config: GCPConfigInfoWithCredsPartial) -> GCPConfigV
         ):
             raise_invalid_credentials_error(fields=[["creds"]])
         try:
-            credentials, _ = auth.authenticate(creds=config.creds)
-            # We ignore credentials' project_id since it may be irrelevant.
-            # For example, with Workload Identity Federation for GKE, it's cluster project_id.
-            # config.project_id is not validated directly since it would require org-level permissions.
-            # config.project_id is validated indirectly when checking VPC.
-        except BackendAuthError:
+            credentials, _ = auth.authenticate(creds=config.creds, project_id=config.project_id)
+        except BackendAuthError as e:
+            details = None
+            if len(e.args) > 0:
+                details = e.args[0]
             if is_core_model_instance(config.creds, GCPServiceAccountCreds):
-                raise_invalid_credentials_error(fields=[["creds", "data"]])
+                raise_invalid_credentials_error(fields=[["creds", "data"]], details=details)
             else:
-                raise_invalid_credentials_error(fields=[["creds"]])
+                raise_invalid_credentials_error(fields=[["creds"]], details=details)
         config_values.regions = self._get_regions_element(
             selected=config.regions or DEFAULT_REGIONS
         )