diff --git a/backend/windmill-api-settings/src/lib.rs b/backend/windmill-api-settings/src/lib.rs
index 7a278afc49..3f7fed4489 100644
--- a/backend/windmill-api-settings/src/lib.rs
+++ b/backend/windmill-api-settings/src/lib.rs
@@ -256,52 +256,300 @@ pub async fn test_s3_bucket(
     use bytes::Bytes;
     use futures::StreamExt;
 
-    require_super_admin(&db, &authed.email).await?;
+    // The probe executes on the API server itself. On multi-tenant Cloud that is a shared control
+    // plane, so we constrain untrusted callers to remove the SSRF / credential-exfiltration /
+    // local-filesystem surface (see validate_object_storage_test). On self-hosted instances the
+    // object store usually lives on the local/private network and all authenticated users are
+    // trusted, so testing there stays unrestricted. Super admins keep the unrestricted path too.
+    let is_super_admin = is_super_admin_email(&db, &authed.email).await?;
+    let restrict = !is_super_admin && *CLOUD_HOSTED;
+    if restrict {
+        validate_object_storage_test(&test_s3_bucket).await?;
+    }
 
     let client = build_object_store_from_settings(test_s3_bucket, Some(&db))
         .await?
         .store;
 
-    let mut list = client.list(Some(
-        &windmill_object_store::object_store_reexports::Path::from("".to_string()),
-    ));
-    let first_file = list.next().await;
-    if first_file.is_some() {
-        if let Err(e) = first_file.as_ref().unwrap() {
-            tracing::error!("error listing bucket: {e:#}");
-            error::Error::internal_err(format!("Failed to list files in blob storage: {e:#}"));
+    let run = async {
+        let mut list = client.list(Some(
+            &windmill_object_store::object_store_reexports::Path::from("".to_string()),
+        ));
+        let first_file = list.next().await;
+        if first_file.is_some() {
+            if let Err(e) = first_file.as_ref().unwrap() {
+                tracing::error!("error listing bucket: {e:#}");
+                error::Error::internal_err(format!("Failed to list files in blob storage: {e:#}"));
+            }
+            tracing::info!("Listed files: {:?}", first_file.unwrap());
+        } else {
+            tracing::info!("No files in blob storage");
         }
-        tracing::info!("Listed files: {:?}", first_file.unwrap());
+
+        let path = windmill_object_store::object_store_reexports::Path::from(format!(
+            "/test-s3-bucket-{uuid}",
+            uuid = uuid::Uuid::new_v4()
+        ));
+        tracing::info!("Testing blob storage at path: {path}");
+        client
+            .put(
+                &path,
+                windmill_object_store::object_store_reexports::PutPayload::from_static(b"hello"),
+            )
+            .await
+            .map_err(|e| anyhow::anyhow!("error writing file to {path}: {e:#}"))?;
+        let content = client
+            .get(&path)
+            .await
+            .map_err(to_anyhow)?
+            .bytes()
+            .await
+            .map_err(to_anyhow)?;
+        if content != Bytes::from_static(b"hello") {
+            return Err(error::Error::internal_err(
+                "Failed to read back from blob storage".to_string(),
+            ));
+        }
+        client.delete(&path).await.map_err(to_anyhow)?;
+        Ok::<String, error::Error>("Tested blob storage successfully".to_string())
+    };
+
+    if restrict {
+        // The object-store client is built with timeouts disabled, so a malicious endpoint could
+        // otherwise hold the API server connection open indefinitely.
+        tokio::time::timeout(Duration::from_secs(15), run)
+            .await
+            .map_err(|_| {
+                error::Error::internal_err("Object storage connectivity test timed out".to_string())
+            })?
     } else {
-        tracing::info!("No files in blob storage");
+        run.await
+    }
+}
+
+// Hardening for the object-storage connectivity test by an untrusted (non-super-admin) caller on
+// Cloud. The probe runs on the shared API server, so without these constraints an authenticated
+// user could coerce the server into connecting to arbitrary internal endpoints (SSRF), signing
+// requests with the instance role (credential exfiltration), or reading/writing the server's local
+// disk (filesystem object store).
+#[cfg(feature = "parquet")]
+async fn validate_object_storage_test(settings: &ObjectSettings) -> error::Result<()> {
+    fn non_empty(opt: &Option<String>) -> bool {
+        opt.as_ref().is_some_and(|s| !s.is_empty())
     }
 
-    let path = windmill_object_store::object_store_reexports::Path::from(format!(
-        "/test-s3-bucket-{uuid}",
-        uuid = uuid::Uuid::new_v4()
-    ));
-    tracing::info!("Testing blob storage at path: {path}");
-    client
-        .put(
-            &path,
-            windmill_object_store::object_store_reexports::PutPayload::from_static(b"hello"),
-        )
+    // Reject backends that rely on the server's identity or local filesystem, require explicit
+    // credentials for the rest (so the server never falls back to its own ambient credentials), and
+    // resolve the host the client will actually connect to. We derive the *effective* endpoint here
+    // — mirroring build_*_from_settings: the region/account-derived default and the virtual-hosted
+    // bucket prefix — rather than only validating a caller-supplied `endpoint`, so caller-controlled
+    // `region`/`account_name`/`bucket` cannot smuggle an internal host past the check (e.g. an empty
+    // endpoint with region = "@169.254.169.254/" otherwise resolves to the cloud metadata service).
+    let effective_endpoint: Option<String> = match settings {
+        ObjectSettings::Filesystem(_) => {
+            return Err(error::Error::NotAuthorized(
+                "Testing a local filesystem object store requires a super admin".to_string(),
+            ));
+        }
+        ObjectSettings::AwsOidc(_) => {
+            return Err(error::Error::NotAuthorized(
+                "Testing OIDC-based object storage requires a super admin".to_string(),
+            ));
+        }
+        ObjectSettings::S3(s3) => {
+            if !(non_empty(&s3.access_key) && non_empty(&s3.secret_key)) {
+                return Err(error::Error::NotAuthorized(
+                    "Testing S3 storage without explicit credentials requires a super admin"
+                        .to_string(),
+                ));
+            }
+            let region = s3
+                .region
+                .clone()
+                .filter(|r| !r.is_empty())
+                .or_else(|| std::env::var("AWS_REGION").ok().filter(|r| !r.is_empty()))
+                .unwrap_or_else(|| "us-east-1".to_string());
+            let raw_endpoint = s3
+                .endpoint
+                .clone()
+                .filter(|e| !e.is_empty())
+                .or_else(|| std::env::var("S3_ENDPOINT").ok().filter(|e| !e.is_empty()))
+                .unwrap_or_else(|| format!("s3.{region}.amazonaws.com"));
+            Some(windmill_object_store::render_endpoint(
+                raw_endpoint,
+                !s3.allow_http.unwrap_or(true),
+                s3.port,
+                s3.path_style,
+                s3.bucket.clone().unwrap_or_default(),
+            ))
+        }
+        ObjectSettings::Azure(azure) => {
+            if !non_empty(&azure.access_key) {
+                return Err(error::Error::NotAuthorized(
+                    "Testing Azure storage without an explicit access key requires a super admin"
+                        .to_string(),
+                ));
+            }
+            Some(
+                azure
+                    .endpoint
+                    .clone()
+                    .filter(|e| !e.is_empty())
+                    .unwrap_or_else(|| format!("{}.blob.core.windows.net", azure.account_name)),
+            )
+        }
+        ObjectSettings::Gcs(gcs) => {
+            if gcs.service_account_key.is_empty() {
+                return Err(error::Error::NotAuthorized(
+                    "Testing GCS storage without a service account key requires a super admin"
+                        .to_string(),
+                ));
+            }
+            // The service-account-key JSON can override the data-plane URL (`gcs_base_url`) and the
+            // OAuth token endpoint (`token_uri`); the GCS client connects to whatever they point at.
+            // Validate every http(s) URL embedded in the key. When none override it, the host stays
+            // the public storage.googleapis.com, so no further check is needed.
+            if let Ok(serde_json::Value::Object(map)) =
+                serde_json::from_str::<serde_json::Value>(&gcs.service_account_key)
+            {
+                for value in map.values() {
+                    if let Some(url) = value.as_str() {
+                        // Match how the URL parser reads the value: leading whitespace/control is
+                        // ignored and the scheme is case-insensitive.
+                        let url =
+                            url.trim_start_matches(|c: char| c.is_whitespace() || c.is_control());
+                        if strip_http_scheme(url).is_some() {
+                            validate_public_endpoint(url).await?;
+                        }
+                    }
+                }
+            }
+            None
+        }
+    };
+
+    // Block non-public network targets (internal services, cloud metadata, loopback, ...).
+    if let Some(endpoint) = effective_endpoint {
+        validate_public_endpoint(&endpoint).await?;
+    }
+    Ok(())
+}
+
+#[cfg(feature = "parquet")]
+async fn validate_public_endpoint(endpoint: &str) -> error::Result<()> {
+    let host = extract_host(endpoint).ok_or_else(|| {
+        error::Error::BadRequest(format!("Invalid object storage endpoint: {endpoint}"))
+    })?;
+
+    let addrs: Vec<std::net::SocketAddr> = tokio::net::lookup_host((host.as_str(), 443u16))
         .await
-        .map_err(|e| anyhow::anyhow!("error writing file to {path}: {e:#}"))?;
-    let content = client
-        .get(&path)
-        .await
-        .map_err(to_anyhow)?
-        .bytes()
-        .await
-        .map_err(to_anyhow)?;
-    if content != Bytes::from_static(b"hello") {
-        return Err(error::Error::internal_err(
-            "Failed to read back from blob storage".to_string(),
-        ));
+        .map_err(|e| {
+            error::Error::BadRequest(format!(
+                "Could not resolve object storage endpoint '{host}': {e}"
+            ))
+        })?
+        .collect();
+
+    if addrs.is_empty() {
+        return Err(error::Error::BadRequest(format!(
+            "Could not resolve object storage endpoint '{host}'"
+        )));
+    }
+
+    // Reject if any resolved address is non-public, which also defeats the simplest DNS-rebinding
+    // attempts (a name resolving to both a public and a private address).
+    for addr in addrs {
+        if is_forbidden_ip(addr.ip()) {
+            return Err(error::Error::NotAuthorized(
+                "Testing object storage at a private, loopback, or link-local endpoint requires a super admin"
+                    .to_string(),
+            ));
+        }
+    }
+    Ok(())
+}
+
+// Strip a leading `http://`/`https://` scheme case-insensitively (URL schemes are
+// case-insensitive), returning the remainder when one was present.
+#[cfg(feature = "parquet")]
+fn strip_http_scheme(s: &str) -> Option<&str> {
+    for scheme in ["https://", "http://"] {
+        let b = scheme.as_bytes();
+        if s.len() >= b.len() && s.as_bytes()[..b.len()].eq_ignore_ascii_case(b) {
+            return Some(&s[b.len()..]);
+        }
+    }
+    None
+}
+
+#[cfg(feature = "parquet")]
+fn extract_host(endpoint: &str) -> Option<String> {
+    let mut s = endpoint.trim();
+    if let Some(rest) = strip_http_scheme(s) {
+        s = rest;
+    }
+    s = s.split(['/', '?', '#', '\\']).next().unwrap_or(s);
+    if let Some((_, rest)) = s.rsplit_once('@') {
+        s = rest;
+    }
+    let host = if let Some(rest) = s.strip_prefix('[') {
+        // IPv6 literal, e.g. [::1]:9000
+        rest.split(']').next().unwrap_or(rest)
+    } else {
+        // host or host:port
+        s.split(':').next().unwrap_or(s)
+    }
+    .trim();
+    if host.is_empty() {
+        None
+    } else {
+        Some(host.to_string())
+    }
+}
+
+#[cfg(feature = "parquet")]
+fn is_forbidden_ip(ip: std::net::IpAddr) -> bool {
+    use std::net::{IpAddr, Ipv4Addr};
+    match ip {
+        IpAddr::V4(v4) => {
+            v4.is_loopback()
+                || v4.is_private()
+                || v4.is_link_local() // 169.254.0.0/16, incl. the cloud metadata endpoint
+                || v4.is_unspecified()
+                || v4.is_broadcast()
+                || v4.is_documentation()
+                || v4.is_multicast()
+                || v4.octets()[0] == 0 // 0.0.0.0/8
+                || (v4.octets()[0] == 100 && (v4.octets()[1] & 0xc0) == 64) // 100.64.0.0/10 CGNAT
+        }
+        IpAddr::V6(v6) => {
+            // Any IPv4 embedded in an IPv6 address (IPv4-mapped ::ffff:0:0/96, IPv4-compatible
+            // ::/96, or NAT64 64:ff9b::/96) is re-checked against the IPv4 rules, so e.g.
+            // 64:ff9b::169.254.169.254 cannot route to the metadata endpoint in a NAT64 network.
+            let seg = v6.segments();
+            let is_v4_compatible = seg[0..6] == [0, 0, 0, 0, 0, 0];
+            let is_nat64 = seg[0] == 0x0064 && seg[1] == 0xff9b && seg[2..6] == [0, 0, 0, 0];
+            if let Some(v4) = v6.to_ipv4_mapped() {
+                return is_forbidden_ip(IpAddr::V4(v4));
+            }
+            if is_v4_compatible || is_nat64 {
+                let embedded = Ipv4Addr::new(
+                    (seg[6] >> 8) as u8,
+                    (seg[6] & 0xff) as u8,
+                    (seg[7] >> 8) as u8,
+                    (seg[7] & 0xff) as u8,
+                );
+                if is_forbidden_ip(IpAddr::V4(embedded)) {
+                    return true;
+                }
+            }
+            v6.is_loopback()
+                || v6.is_unspecified()
+                || v6.is_multicast()
+                || (seg[0] & 0xfe00) == 0xfc00 // fc00::/7 unique local
+                || (seg[0] & 0xffc0) == 0xfe80 // fe80::/10 link-local
+        }
     }
-    client.delete(&path).await.map_err(to_anyhow)?;
-    Ok("Tested blob storage successfully".to_string())
 }
 
 #[cfg(feature = "parquet")]
@@ -1861,3 +2109,111 @@ mod tests {
         );
     }
 }
+
+#[cfg(all(test, feature = "parquet"))]
+mod object_storage_test_hardening {
+    use super::{extract_host, is_forbidden_ip, validate_object_storage_test};
+    use std::net::IpAddr;
+    use windmill_object_store::ObjectSettings;
+
+    // IP literals (not hostnames) keep validate_public_endpoint deterministic — `lookup_host`
+    // parses them without any network round-trip.
+    fn gcs_settings(gcs_base_url: &str) -> ObjectSettings {
+        serde_json::from_value(serde_json::json!({
+            "type": "Gcs",
+            "bucket": "b",
+            "serviceAccountKey": { "gcs_base_url": gcs_base_url, "client_email": "x@y.z" }
+        }))
+        .unwrap()
+    }
+
+    #[tokio::test]
+    async fn rejects_gcs_internal_base_url() {
+        // gcs_base_url in the service-account key must not smuggle an internal host past the check,
+        // including via a mixed-case scheme (URL schemes are case-insensitive).
+        for url in [
+            "http://169.254.169.254",
+            "HTTP://169.254.169.254",
+            "Https://10.0.0.5",
+        ] {
+            assert!(
+                validate_object_storage_test(&gcs_settings(url))
+                    .await
+                    .is_err(),
+                "{url} should be rejected"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn allows_gcs_public_base_url() {
+        assert!(
+            validate_object_storage_test(&gcs_settings("https://8.8.8.8"))
+                .await
+                .is_ok()
+        );
+    }
+
+    fn ip(s: &str) -> IpAddr {
+        s.parse().unwrap()
+    }
+
+    #[test]
+    fn forbids_internal_ips() {
+        for s in [
+            "127.0.0.1",                // loopback
+            "169.254.169.254",          // cloud metadata (link-local)
+            "10.0.0.5",                 // private
+            "172.16.3.4",               // private
+            "192.168.1.10",             // private
+            "0.0.0.0",                  // unspecified
+            "100.64.0.1",               // CGNAT
+            "::1",                      // IPv6 loopback
+            "fe80::1",                  // IPv6 link-local
+            "fc00::1",                  // IPv6 unique local
+            "::ffff:127.0.0.1",         // IPv4-mapped loopback
+            "::ffff:169.254.169.254",   // IPv4-mapped metadata
+            "::169.254.169.254",        // IPv4-compatible metadata
+            "64:ff9b::169.254.169.254", // NAT64-embedded metadata
+            "64:ff9b::a9fe:a9fe",       // NAT64-embedded metadata (hex form)
+        ] {
+            assert!(is_forbidden_ip(ip(s)), "{s} should be forbidden");
+        }
+    }
+
+    #[test]
+    fn allows_public_ips() {
+        for s in ["8.8.8.8", "1.1.1.1", "52.95.110.1", "2606:4700:4700::1111"] {
+            assert!(!is_forbidden_ip(ip(s)), "{s} should be allowed");
+        }
+    }
+
+    #[test]
+    fn extracts_host_from_endpoint() {
+        let cases = [
+            ("s3.amazonaws.com", Some("s3.amazonaws.com")),
+            ("https://minio.internal:9000", Some("minio.internal")),
+            ("http://10.0.0.5:9000/bucket", Some("10.0.0.5")),
+            ("user:pass@host.example:443", Some("host.example")),
+            ("[::1]:9000", Some("::1")),
+            ("https://[fe80::1]/x", Some("fe80::1")),
+            ("", None),
+            // Injection via region/bucket interpolation into the default endpoint string: the
+            // userinfo `@` and the path `/` must not hide the real authority from the host check.
+            (
+                "https://s3.@169.254.169.254/.amazonaws.com",
+                Some("169.254.169.254"),
+            ),
+            (
+                "https://@169.254.169.254/mybucket.s3.amazonaws.com",
+                Some("169.254.169.254"),
+            ),
+            ("s3.#@169.254.169.254/x.amazonaws.com", Some("s3.")),
+            // Scheme is case-insensitive.
+            ("HTTP://169.254.169.254", Some("169.254.169.254")),
+        ];
+        for (input, expected) in cases {
+            assert_eq!(extract_host(input).as_deref(), expected, "input: {input}");
+        }
+    }
+}