test(e2e): make security boundary tests assert real outcomes (#4466)

* test(e2e): make security boundary tests assert real outcomes

* fix(e2e): use expect_err to satisfy clippy err_expect lint

crates/e2e_test/src/lib.rs

@@ -49,6 +49,10 @@ mod quota_test;
 #[cfg(test)]
 mod bucket_policy_check_test;
 
+// Security boundary tests: DoS limits, SSRF prevention, concurrent-write integrity
+#[cfg(test)]
+mod security_boundary_test;
+
 /// IAM / bucket / STS session policy with `s3:ExistingObjectTag` conditions (E2E).
 #[cfg(test)]
 mod existing_object_tag_policy_test;

crates/e2e_test/src/security_boundary_test.rs

@@ -14,59 +14,94 @@
 
 //! Security boundary tests for RustFS
 //!
-//! These tests verify that RustFS properly handles security-sensitive scenarios:
+//! These tests verify that RustFS properly enforces security-sensitive
-//! - DoS protection (large payloads, excessive multipart parts)
+//! controls by issuing real requests against a running server and asserting
-//! - SSRF prevention (internal URL validation)
+//! the concrete outcome of each control:
-//! - Race condition handling (concurrent operations)
+//! - DoS protection (oversized tagging payloads, excessive multipart parts)
+//! - SSRF prevention (internal/private endpoints rejected for tiering)
+//! - Race condition handling (concurrent writes converge without corruption)
 
-use crate::common;
+use crate::common::{RustFSTestEnvironment, awscurl_available, awscurl_put, init_logging};
+use aws_sdk_s3::error::ProvideErrorMetadata;
 use aws_sdk_s3::primitives::ByteStream;
-use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
+use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart, Tag, Tagging};
+use serial_test::serial;
 use std::error::Error;
+use tracing::info;
 
-/// Test that large XML bodies are properly rejected
+/// Oversized tagging payloads must be rejected by the per-object tag limit.
+///
+/// RustFS caps object tagging at 10 tags (`InvalidTag`). This protects the
+/// server against unbounded tagging XML bodies. We transmit a payload that is
+/// far beyond that limit and assert the server rejects it with the specific
+/// error, rather than accepting an arbitrarily large control-plane body.
 #[tokio::test]
+#[serial]
 async fn test_large_xml_body_rejection() -> Result<(), Box<dyn Error + Send + Sync>> {
-    let ctx = common::TestContext::new("security-large-xml").await?;
+    init_logging();
-    let client = ctx.client();
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+    let client = env.create_s3_client();
 
-    // Create a bucket first
     let bucket_name = format!("security-large-xml-{}", uuid::Uuid::new_v4());
     client.create_bucket().bucket(&bucket_name).send().await?;
 
-    // Try to send a very large XML body (simulating DoS attempt)
+    let object_key = "oversized-tagging-target";
-    // This should be rejected by the server's body size limit
+    client
-    let large_body = format!("<Tagging><TagSet>{}</TagSet></Tagging>", "<Tag><Key>test</Key><Value>test</Value></Tag>".repeat(10000));
+        .put_object()
+        .bucket(&bucket_name)
+        .key(object_key)
+        .body(ByteStream::from_static(b"payload"))
+        .send()
+        .await?;
+
+    // Build a tag set that is well past the enforced 10-tag limit. This is a
+    // genuinely oversized tagging XML body that must be rejected, not a single
+    // in-limit tag.
+    let mut tag_builder = Tagging::builder();
+    for i in 0..500 {
+        tag_builder = tag_builder.tag_set(Tag::builder().key(format!("key-{i}")).value(format!("value-{i}")).build()?);
+    }
+    let oversized_tagging = tag_builder.build()?;
 
     let result = client
-        .put_bucket_tagging()
+        .put_object_tagging()
         .bucket(&bucket_name)
-        .tagging(aws_sdk_s3::types::Tagging::builder().tag_set(
+        .key(object_key)
-            aws_sdk_s3::types::Tag::builder().key("test").value("test").build()?,
+        .tagging(oversized_tagging)
-        ).build()?)
         .send()
         .await;
 
-    // Cleanup
+    // Cleanup (best effort) before assertions so a failed assertion still
-    client.delete_bucket().bucket(&bucket_name).send().await?;
+    // leaves no residue.
+    let _ = client.delete_object().bucket(&bucket_name).key(object_key).send().await;
+    let _ = client.delete_bucket().bucket(&bucket_name).send().await;
 
-    // The server should either accept it (if within limits) or reject it gracefully
+    assert!(result.is_err(), "Server must reject an oversized tagging payload, but it was accepted");
-    // We're testing that it doesn't crash or hang
+    let err = result.expect_err("checked is_err above");
-    assert!(result.is_ok() || result.is_err(), "Server should handle large bodies gracefully");
+    let code = err.as_service_error().and_then(|e| e.code());
+    assert_eq!(
+        code,
+        Some("InvalidTag"),
+        "Oversized tagging should be rejected with InvalidTag, got code {code:?}, err: {err:?}"
+    );
 
+    env.stop_server();
     Ok(())
 }
 
-/// Test that excessive multipart parts are properly handled
+/// Excessive multipart parts must be rejected.
 #[tokio::test]
+#[serial]
 async fn test_excessive_multipart_parts() -> Result<(), Box<dyn Error + Send + Sync>> {
-    let ctx = common::TestContext::new("security-multipart").await?;
+    init_logging();
-    let client = ctx.client();
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+    let client = env.create_s3_client();
 
     let bucket_name = format!("security-multipart-{}", uuid::Uuid::new_v4());
     client.create_bucket().bucket(&bucket_name).send().await?;
 
-    // Create a multipart upload
     let create_result = client
         .create_multipart_upload()
         .bucket(&bucket_name)
@@ -74,136 +109,172 @@ async fn test_excessive_multipart_parts() -> Result<(), Box<dyn Error + Send + S
         .send()
         .await?;
 
-    let upload_id = create_result.upload_id().expect("upload_id should be present");
+    let upload_id = create_result.upload_id().expect("upload_id should be present").to_string();
 
-    // Try to complete with too many parts (should be rejected)
+    // Try to complete with too many parts (should be rejected).
     let mut parts = Vec::new();
     for i in 1..=10001 {
-        parts.push(
+        parts.push(CompletedPart::builder().part_number(i).e_tag(format!("etag-{i}")).build());
-            CompletedPart::builder()
-                .part_number(i)
-                .e_tag(format!("etag-{i}"))
-                .build(),
-        );
     }
 
     let result = client
         .complete_multipart_upload()
         .bucket(&bucket_name)
         .key("test-large")
-        .upload_id(upload_id)
+        .upload_id(&upload_id)
-        .multipart_upload(
+        .multipart_upload(CompletedMultipartUpload::builder().set_parts(Some(parts)).build())
-            CompletedMultipartUpload::builder()
-                .set_parts(Some(parts))
-                .build(),
-        )
         .send()
         .await;
 
-    // Cleanup
+    // Cleanup.
     let _ = client
         .abort_multipart_upload()
         .bucket(&bucket_name)
         .key("test-large")
-        .upload_id(upload_id)
+        .upload_id(&upload_id)
         .send()
         .await;
-    client.delete_bucket().bucket(&bucket_name).send().await?;
+    let _ = client.delete_bucket().bucket(&bucket_name).send().await;
 
-    // The server should reject excessive parts gracefully
     assert!(result.is_err(), "Server should reject excessive multipart parts");
 
+    env.stop_server();
     Ok(())
 }
 
-/// Test concurrent operations on the same object
+/// Concurrent writes to the same object must converge without corruption.
+///
+/// We fan out concurrent PUTs of distinct contents and assert every write is
+/// accepted, then assert the final object is exactly one of the written values
+/// (last-writer-wins, no torn/garbage state) and that it is absent after a
+/// subsequent delete.
 #[tokio::test]
+#[serial]
 async fn test_concurrent_object_operations() -> Result<(), Box<dyn Error + Send + Sync>> {
-    let ctx = common::TestContext::new("security-concurrent").await?;
+    init_logging();
-    let client = ctx.client();
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+    let client = env.create_s3_client();
 
     let bucket_name = format!("security-concurrent-{}", uuid::Uuid::new_v4());
     client.create_bucket().bucket(&bucket_name).send().await?;
 
-    // Upload initial object
+    let key = "concurrent-test";
-    client
+    let writer_count = 10;
-        .put_object()
+    let expected_contents: Vec<String> = (0..writer_count).map(|i| format!("content-{i}")).collect();
-        .bucket(&bucket_name)
-        .key("concurrent-test")
-        .body(ByteStream::from_static(b"initial content"))
-        .send()
-        .await?;
 
-    // Spawn multiple concurrent operations
+    // Fan out concurrent PUTs of distinct contents.
     let mut handles = Vec::new();
-    for i in 0..10 {
+    for content in expected_contents.iter().cloned() {
         let client_clone = client.clone();
         let bucket_clone = bucket_name.clone();
         handles.push(tokio::spawn(async move {
-            // Concurrent PUT
+            client_clone
-            let content = format!("content-{i}");
-            let _ = client_clone
                 .put_object()
                 .bucket(&bucket_clone)
-                .key("concurrent-test")
+                .key(key)
                 .body(ByteStream::from(content.into_bytes()))
                 .send()
-                .await;
+                .await
-
+                .is_ok()
-            // Concurrent GET
-            let _ = client_clone
-                .get_object()
-                .bucket(&bucket_clone)
-                .key("concurrent-test")
-                .send()
-                .await;
-
-            // Concurrent DELETE (might fail if object doesn't exist)
-            let _ = client_clone
-                .delete_object()
-                .bucket(&bucket_clone)
-                .key("concurrent-test")
-                .send()
-                .await;
         }));
     }
 
-    // Wait for all operations to complete
+    // Every concurrent write must be accepted by the server.
     for handle in handles {
-        let _ = handle.await;
+        let put_ok = handle.await?;
+        assert!(
+            put_ok,
+            "A concurrent PUT was rejected; server must accept concurrent writes to the same key"
+        );
     }
 
-    // Cleanup
+    // The final object must be exactly one of the written contents (no torn or
-    let _ = client.delete_object().bucket(&bucket_name).key("concurrent-test").send().await;
+    // corrupted state from interleaved writes).
-    client.delete_bucket().bucket(&bucket_name).send().await?;
+    let get = client.get_object().bucket(&bucket_name).key(key).send().await?;
+    let body = get.body.collect().await?.into_bytes();
+    let final_content = String::from_utf8(body.to_vec())?;
+    assert!(
+        expected_contents.contains(&final_content),
+        "Final object content {final_content:?} is not one of the concurrently written values"
+    );
 
-    // The server should handle concurrent operations without crashing
+    // After deletion, the object must be absent.
-    // We don't check for specific results since operations are concurrent
+    client.delete_object().bucket(&bucket_name).key(key).send().await?;
+    let get_after_delete = client.get_object().bucket(&bucket_name).key(key).send().await;
+    assert!(get_after_delete.is_err(), "Object must be absent after delete");
+    let code = get_after_delete
+        .err()
+        .and_then(|e| e.as_service_error().and_then(|se| se.code()).map(str::to_string));
+    assert_eq!(
+        code.as_deref(),
+        Some("NoSuchKey"),
+        "GET after delete should return NoSuchKey, got {code:?}"
+    );
 
+    let _ = client.delete_bucket().bucket(&bucket_name).send().await;
+
+    env.stop_server();
     Ok(())
 }
 
-/// Test that internal/private URLs are rejected for tiering
+/// Internal/private endpoints must be rejected as remote tier backends (SSRF).
+///
+/// This issues a real admin AddTier call (`PUT /rustfs/admin/v3/tier`) for each
+/// internal/private endpoint and asserts the server rejects it (non-2xx, so the
+/// signed request helper returns an error). An internal endpoint must never be
+/// accepted as a tier backend. The rejection may originate from explicit
+/// SSRF/internal-address filtering or from the backend connectivity/credential
+/// validation performed during AddTier; either way the security-relevant
+/// outcome — the internal endpoint is not accepted — is asserted here.
+///
+/// The admin API is exercised via signed `awscurl` requests, matching the
+/// pattern used by the other admin-API E2E tests in this crate; the test is
+/// skipped when `awscurl` is not installed.
 #[tokio::test]
+#[serial]
 async fn test_tiering_url_validation() -> Result<(), Box<dyn Error + Send + Sync>> {
-    let ctx = common::TestContext::new("security-tiering-url").await?;
+    init_logging();
-    let client = ctx.client();
+    if !awscurl_available() {
+        info!("Skipping tiering URL validation test because awscurl is not available");
+        return Ok(());
+    }
 
-    // Try to configure tiering with internal URLs
+    let mut env = RustFSTestEnvironment::new().await?;
-    // This should be rejected by the server's SSRF protection
+    env.start_rustfs_server(vec![]).await?;
-    let internal_urls = vec![
+
+    let tier_url = format!("{}/rustfs/admin/v3/tier", env.url);
+    let internal_endpoints = [
         "http://127.0.0.1:8080",
         "http://localhost:8080",
-        "http://169.254.169.254", // AWS metadata endpoint
+        "http://169.254.169.254", // cloud instance metadata endpoint
         "http://[::1]:8080",
     ];
 
-    for url in internal_urls {
+    for endpoint in internal_endpoints {
-        // The server should reject internal URLs
+        // AddTier expects an uppercase tier name and a backend configuration.
-        // We're testing that it doesn't allow SSRF attacks
+        let body = serde_json::json!({
-        // Note: This test may need to be adjusted based on the actual API
+            "type": "s3",
-        println!("Testing internal URL rejection: {url}");
+            "s3": {
+                "name": "SSRFTEST",
+                "endpoint": endpoint,
+                "accessKey": "ssrf-probe",
+                "secretKey": "ssrf-probe",
+                "bucket": "ssrf-probe-bucket",
+                "prefix": "",
+                "region": "us-east-1",
+                "storageClass": ""
+            }
+        })
+        .to_string();
+
+        let result = awscurl_put(&tier_url, &body, &env.access_key, &env.secret_key).await;
+        assert!(
+            result.is_err(),
+            "AddTier must reject internal endpoint {endpoint}, but it was accepted: {result:?}"
+        );
     }
 
+    env.stop_server();
     Ok(())
 }