fix(object_store): validate S3 bucket name before use

An invalid bucket name (e.g. `rw_data`, which contains an
underscore) previously only surfaced as an opaque error like
"Unexpected (permanent) at stat" on first access, making the real
cause hard to diagnose.

Add validate_s3_bucket_name() implementing the AWS S3 bucket naming
rules, and call it eagerly in build_remote_object_store() for the
s3:// scheme (both the native S3ObjectStore and OpenDAL S3 engine
paths) so misconfiguration is caught at startup with a clear panic
message instead of a confusing runtime error.

Closes #20263

Author: utkarshmankad

src/object_store/src/object/mod.rs

@@ -884,8 +884,11 @@ pub async fn build_remote_object_store(
     tracing::debug!(config=?config, "object store {ident}");
     match url {
         s3 if s3.starts_with("s3://") => {
+            let bucket = s3.strip_prefix("s3://").unwrap();
+            if let Err(e) = crate::object::s3::validate_s3_bucket_name(bucket) {
+                panic!("Invalid object store configuration: {e}");
+            }
             if config.s3.developer.use_opendal {
-                let bucket = s3.strip_prefix("s3://").unwrap();
                 tracing::info!("Using OpenDAL to access s3, bucket is {}", bucket);
                 ObjectStoreImpl::Opendal(
                     OpendalObjectStore::new_s3_engine(
@@ -898,13 +901,9 @@ pub async fn build_remote_object_store(
                 )
             } else {
                 ObjectStoreImpl::S3(
-                    S3ObjectStore::new_with_config(
-                        s3.strip_prefix("s3://").unwrap().to_owned(),
-                        metrics.clone(),
-                        config.clone(),
-                    )
-                    .await
-                    .monitored(metrics, config),
+                    S3ObjectStore::new_with_config(bucket.to_owned(), metrics.clone(), config.clone())
+                        .await
+                        .monitored(metrics, config),
                 )
             }
         }
@@ -1180,6 +1179,27 @@ mod tests {
         let err = reader.read_to_end(&mut output).await.unwrap_err();
         assert!(err.to_string().contains("injected stream error"));
     }
+
+    /// Integration test for issue #20263: an invalid bucket name (e.g. containing an
+    /// underscore) must be rejected eagerly with a clear panic message, instead of
+    /// surfacing as an opaque request error only when the store is first accessed.
+    #[tokio::test]
+    #[should_panic(expected = "Invalid object store configuration")]
+    async fn test_build_remote_object_store_rejects_invalid_bucket_name() {
+        use std::sync::Arc;
+
+        use risingwave_common::config::ObjectStoreConfig;
+
+        use super::{ObjectStoreMetrics, build_remote_object_store};
+
+        build_remote_object_store(
+            "s3://rw_data",
+            Arc::new(ObjectStoreMetrics::unused()),
+            "test",
+            Arc::new(ObjectStoreConfig::default()),
+        )
+        .await;
+    }
 }
 
 #[derive(Debug, Clone, Copy)]

src/object_store/src/object/s3.rs

@@ -403,6 +403,113 @@ impl StreamingUploader for S3StreamingUploader {
     }
 }
 
+/// Validates an S3 bucket name against the AWS bucket naming rules
+/// (<https://docs.aws.amazon.com/AmazonS3/latest/userguide/bucketnamingrules.html>),
+/// so that a misconfigured bucket name is rejected with a clear error message
+/// instead of surfacing as an opaque request error at first access.
+pub fn validate_s3_bucket_name(bucket: &str) -> Result<(), String> {
+    let len = bucket.len();
+    if !(3..=63).contains(&len) {
+        return Err(format!(
+            "invalid S3 bucket name {bucket:?}: length must be between 3 and 63 characters, got {len}"
+        ));
+    }
+    if bucket
+        .parse::<std::net::Ipv4Addr>()
+        .is_ok_and(|_| bucket.split('.').count() == 4)
+    {
+        return Err(format!(
+            "invalid S3 bucket name {bucket:?}: must not be formatted as an IP address"
+        ));
+    }
+    let is_valid_char =
+        |c: char| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-' || c == '.';
+    if !bucket.chars().all(is_valid_char) {
+        return Err(format!(
+            "invalid S3 bucket name {bucket:?}: only lowercase letters, numbers, dots and hyphens are allowed"
+        ));
+    }
+    let starts_ends_alnum = bucket
+        .chars()
+        .next()
+        .is_some_and(|c| c.is_ascii_lowercase() || c.is_ascii_digit())
+        && bucket
+            .chars()
+            .next_back()
+            .is_some_and(|c| c.is_ascii_lowercase() || c.is_ascii_digit());
+    if !starts_ends_alnum {
+        return Err(format!(
+            "invalid S3 bucket name {bucket:?}: must start and end with a lowercase letter or number"
+        ));
+    }
+    if bucket.contains("..") {
+        return Err(format!(
+            "invalid S3 bucket name {bucket:?}: must not contain two adjacent periods"
+        ));
+    }
+    if bucket.starts_with("xn--") || bucket.ends_with("-s3alias") || bucket.ends_with("--ol-s3") {
+        return Err(format!(
+            "invalid S3 bucket name {bucket:?}: must not start with the prefix \"xn--\" or end with the suffix \"-s3alias\" or \"--ol-s3\""
+        ));
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod validate_s3_bucket_name_tests {
+    use super::validate_s3_bucket_name;
+
+    #[test]
+    fn test_valid_bucket_names() {
+        assert!(validate_s3_bucket_name("rw-data").is_ok());
+        assert!(validate_s3_bucket_name("my.bucket.123").is_ok());
+        assert!(validate_s3_bucket_name("abc").is_ok());
+        assert!(validate_s3_bucket_name(&"a".repeat(63)).is_ok());
+    }
+
+    #[test]
+    fn test_rejects_underscore() {
+        // e.g. issue #20263: `rw_data` is not a valid bucket name.
+        let err = validate_s3_bucket_name("rw_data").unwrap_err();
+        assert!(err.contains("lowercase letters, numbers, dots and hyphens"));
+    }
+
+    #[test]
+    fn test_rejects_bad_length() {
+        assert!(validate_s3_bucket_name("ab").is_err());
+        assert!(validate_s3_bucket_name(&"a".repeat(64)).is_err());
+    }
+
+    #[test]
+    fn test_rejects_uppercase() {
+        assert!(validate_s3_bucket_name("MyBucket").is_err());
+    }
+
+    #[test]
+    fn test_rejects_ip_address() {
+        assert!(validate_s3_bucket_name("192.168.1.1").is_err());
+    }
+
+    #[test]
+    fn test_rejects_bad_start_end() {
+        assert!(validate_s3_bucket_name("-mybucket").is_err());
+        assert!(validate_s3_bucket_name("mybucket-").is_err());
+        assert!(validate_s3_bucket_name(".mybucket").is_err());
+    }
+
+    #[test]
+    fn test_rejects_adjacent_periods() {
+        assert!(validate_s3_bucket_name("my..bucket").is_err());
+    }
+
+    #[test]
+    fn test_rejects_reserved_prefix_suffix() {
+        assert!(validate_s3_bucket_name("xn--bucket").is_err());
+        assert!(validate_s3_bucket_name("mybucket-s3alias").is_err());
+        assert!(validate_s3_bucket_name("mybucket--ol-s3").is_err());
+    }
+}
+
 fn get_upload_body(data: Vec<Bytes>) -> ByteStream {
     // `ByteStream` is retryable when created from in-memory data.
     // This code path is used for non-multipart uploads, so a copy is acceptable.