starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-10 08:07:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: allow object storage test for non-super-admins, harden on cloud (#9739)

Browse Source 
* fix: allow non-super-admin object storage test, harden SSRF surface on cloud

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

* fix: validate effective object storage host to close region/bucket SSRF bypass

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

* fix: validate gcs_base_url/token_uri in GCS service account key to close SSRF bypass

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

* fix: match url scheme case-insensitively in object storage host validation

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

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
hugocasa
2026-06-23 20:50:40 +00:00
committed by GitHub
co-authored by Claude Opus 4.8
parent e90b2be8fa
commit 24446e8009
1 changed files with 391 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api-settings/src/lib.rs
+391 -35
View File
@@ -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}");
}
}
}