gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 16:28:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

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

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

* fix(e2e): use expect_err to satisfy clippy err_expect lint
This commit is contained in:
Zhengchao An
2026-07-08 22:01:46 +08:00
committed by GitHub
parent 322b585f71
commit dee8e4e639
2 changed files with 172 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/e2e_test/src/lib.rs
+4
View File
@@ -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
+168 -97
View File
@@ -14,59 +14,94 @@
//! Security boundary tests for RustFS
//!
//! These tests verify that RustFS properly handles security-sensitive scenarios:
//! - DoS protection (large payloads, excessive multipart parts)
//! - SSRF prevention (internal URL validation)
//! - Race condition handling (concurrent operations)
//! These tests verify that RustFS properly enforces security-sensitive
//! controls by issuing real requests against a running server and asserting
//! the concrete outcome of each control:
//! - 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?;
let client = ctx.client();
init_logging();
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)
// This should be rejected by the server's body size limit
let large_body = format!("<Tagging><TagSet>{}</TagSet></Tagging>", "<Tag><Key>test</Key><Value>test</Value></Tag>".repeat(10000));
let object_key = "oversized-tagging-target";
client
.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(
aws_sdk_s3::types::Tag::builder().key("test").value("test").build()?,
).build()?)
.key(object_key)
.tagging(oversized_tagging)
.send()
.await;
// Cleanup
client.delete_bucket().bucket(&bucket_name).send().await?;
// Cleanup (best effort) before assertions so a failed assertion still
// 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
// We're testing that it doesn't crash or hang
assert!(result.is_ok() || result.is_err(), "Server should handle large bodies gracefully");
assert!(result.is_err(), "Server must reject an oversized tagging payload, but it was accepted");
let err = result.expect_err("checked is_err above");
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?;
let client = ctx.client();
init_logging();
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(
CompletedPart::builder()
.part_number(i)
.e_tag(format!("etag-{i}"))
.build(),
);
parts.push(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)
.multipart_upload(
CompletedMultipartUpload::builder()
.set_parts(Some(parts))
.build(),
)
.upload_id(&upload_id)
.multipart_upload(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?;
let client = ctx.client();
init_logging();
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
client
.put_object()
.bucket(&bucket_name)
.key("concurrent-test")
.body(ByteStream::from_static(b"initial content"))
.send()
.await?;
let key = "concurrent-test";
let writer_count = 10;
let expected_contents: Vec<String> = (0..writer_count).map(|i| format!("content-{i}")).collect();
// 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
let content = format!("content-{i}");
let _ = client_clone
client_clone
.put_object()
.bucket(&bucket_clone)
.key("concurrent-test")
.key(key)
.body(ByteStream::from(content.into_bytes()))
.send()
.await;
// 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;
.await
.is_ok()
}));
}
// 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
let _ = client.delete_object().bucket(&bucket_name).key("concurrent-test").send().await;
client.delete_bucket().bucket(&bucket_name).send().await?;
// The final object must be exactly one of the written contents (no torn or
// corrupted state from interleaved writes).
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
// We don't check for specific results since operations are concurrent
// After deletion, the object must be absent.
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?;
let client = ctx.client();
init_logging();
if !awscurl_available() {
info!("Skipping tiering URL validation test because awscurl is not available");
return Ok(());
}
// Try to configure tiering with internal URLs
// This should be rejected by the server's SSRF protection
let internal_urls = vec![
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
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 {
// The server should reject internal URLs
// We're testing that it doesn't allow SSRF attacks
// Note: This test may need to be adjusted based on the actual API
println!("Testing internal URL rejection: {url}");
for endpoint in internal_endpoints {
// AddTier expects an uppercase tier name and a backend configuration.
let body = serde_json::json!({
"type": "s3",
"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(())
}