feat(object-store): make GCS service account key optional for Workload Identity (#9842)

build_gcs_client always called `.with_service_account_key(...)`, so an
absent key (the settings UI stores "no key" as the empty JSON object `{}`)
was handed to the builder and failed to parse instead of falling through
to the object_store crate's InstanceCredentialProvider. Skip the call when
the key is blank so GCS uses the instance's ambient credentials (GKE
Workload Identity / the GCP metadata server).

"Blank" (empty/whitespace/`{}`/`null`) is centralized in a shared
`gcs_service_account_key_is_blank` predicate so the build path and the
non-super-admin connectivity-test SSRF guard (`validate_object_storage_test`)
agree on what counts as "no key" — otherwise a blank key would bypass the
guard yet still trigger the ambient-credential fallback, letting an
untrusted caller probe arbitrary buckets with the server's instance role.

Also clarify the settings UI hint that the key may be left empty for
ambient credentials, and add regression tests for the blank-key build path
and the guard.

Fixes WIN-2110

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/windmill-api-settings/src/lib.rs

@@ -406,7 +406,11 @@ async fn validate_object_storage_test(settings: &ObjectSettings) -> error::Resul
             )
         }
         ObjectSettings::Gcs(gcs) => {
-            if gcs.service_account_key.is_empty() {
+            // Mirror `build_gcs_client`'s blank-key check (shared predicate): a blank/`{}` key falls
+            // back to the instance's ambient credentials there, so it must be rejected here too —
+            // otherwise an untrusted caller could probe with the server's identity (the very
+            // SSRF/credential-exfil this function guards against).
+            if windmill_object_store::gcs_service_account_key_is_blank(&gcs.service_account_key) {
                 return Err(error::Error::NotAuthorized(
                     "Testing GCS storage without a service account key requires a super admin"
                         .to_string(),
@@ -2187,6 +2191,26 @@ mod object_storage_test_hardening {
         );
     }
 
+    #[tokio::test]
+    async fn rejects_gcs_blank_service_account_key() {
+        // A blank key makes build_gcs_client fall back to the instance's ambient credentials, so an
+        // untrusted caller must not be allowed to test with it. The `serviceAccountKey` field is
+        // serialized via serde's `as_string` (`to_string` of the JSON value), so the settings UI's
+        // "no key" empty object arrives as `"{}"` and a null as `"null"` — both must be rejected.
+        for key in [serde_json::json!({}), serde_json::json!(null)] {
+            let settings: ObjectSettings = serde_json::from_value(serde_json::json!({
+                "type": "Gcs",
+                "bucket": "b",
+                "serviceAccountKey": key
+            }))
+            .unwrap();
+            assert!(
+                validate_object_storage_test(&settings).await.is_err(),
+                "blank key {key:?} should be rejected"
+            );
+        }
+    }
+
     fn ip(s: &str) -> IpAddr {
         s.parse().unwrap()
     }

backend/windmill-object-store/src/lib.rs

@@ -494,6 +494,23 @@ fn build_azure_blob_client(
     return Ok(Arc::new(store));
 }
 
+/// Whether a GCS `service_account_key` carries no static credentials, in which case the client
+/// should fall back to the instance's ambient credentials (GKE Workload Identity / metadata server)
+/// instead of being handed an unparseable key. Besides an empty/whitespace string, the settings UI
+/// stores "no key" as an empty JSON object `{}` (and `serde_json` may yield `null`), so treat those
+/// as absent too. Shared with the connectivity-test SSRF guard so both agree on what "no key" means.
+pub fn gcs_service_account_key_is_blank(service_account_key: &str) -> bool {
+    let trimmed = service_account_key.trim();
+    if trimmed.is_empty() {
+        return true;
+    }
+    match serde_json::from_str::<serde_json::Value>(trimmed) {
+        Ok(serde_json::Value::Null) => true,
+        Ok(serde_json::Value::Object(map)) => map.is_empty(),
+        _ => false,
+    }
+}
+
 #[cfg(feature = "parquet")]
 async fn build_gcs_client(gcs_resource_ref: &GcsResource) -> error::Result<Arc<dyn ObjectStore>> {
     let gcs_resource = gcs_resource_ref.clone();
@@ -509,7 +526,12 @@ async fn build_gcs_client(gcs_resource_ref: &GcsResource) -> error::Result<Arc<d
         )
         .with_bucket_name(gcs_resource.bucket);
 
-    store_builder = store_builder.with_service_account_key(gcs_resource.service_account_key);
+    // A blank key means no static credentials: let the builder fall back to the metadata server
+    // (InstanceCredentialProvider) so GKE Workload Identity / ambient credentials work. Passing a
+    // blank/`{}` key to `with_service_account_key` would instead fail to parse.
+    if !gcs_service_account_key_is_blank(&gcs_resource.service_account_key) {
+        store_builder = store_builder.with_service_account_key(gcs_resource.service_account_key);
+    }
 
     let store = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| store_builder.build()))
         .map_err(|panic_info| {
@@ -1703,6 +1725,40 @@ mod tests {
             .contains("GCS is not supported"));
     }
 
+    #[test]
+    fn test_gcs_service_account_key_is_blank() {
+        for blank in ["", "   ", "\n\t", "{}", "  {}  ", "null"] {
+            assert!(
+                gcs_service_account_key_is_blank(blank),
+                "{blank:?} should be treated as no key"
+            );
+        }
+        for present in ["{\"client_email\":\"x@y.z\"}", "not json"] {
+            assert!(
+                !gcs_service_account_key_is_blank(present),
+                "{present:?} should be treated as a key"
+            );
+        }
+    }
+
+    #[cfg(feature = "parquet")]
+    #[tokio::test]
+    async fn test_build_gcs_client_blank_key_uses_instance_credentials() {
+        // A blank service account key must not be passed to `with_service_account_key`
+        // (which would fail to parse): the builder should fall back to instance credentials
+        // (GKE Workload Identity / metadata server) and construct successfully. `{}` is the
+        // settings UI's representation of "no key".
+        for key in ["", "   ", "{}"] {
+            let resource =
+                GcsResource { bucket: "bucket".to_string(), service_account_key: key.to_string() };
+            assert!(
+                build_gcs_client(&resource).await.is_ok(),
+                "blank key {:?} should build via instance credentials",
+                key
+            );
+        }
+    }
+
     #[test]
     fn test_duckdb_connection_settings_filesystem_unsupported() {
         let resource = ObjectStoreResource::Filesystem(FilesystemSettings {

frontend/src/lib/components/ObjectStoreConfigSettings.svelte

@@ -690,7 +690,10 @@
 					/>
 				</Label>
 				<Label label="Service Account Key">
-					<span class="text-primary text-2xs">JSON content of the service account key file</span>
+					<span class="text-primary text-2xs">
+						JSON content of the service account key file. Leave empty to use the instance's ambient
+						credentials (e.g. GKE Workload Identity / the GCP metadata server).
+					</span>
 					{#if hasServiceAccountKey && !showServiceAccountKey}
 						<div class="flex items-center gap-3 mt-1">
 							<span class="text-tertiary text-xs">