From e822fc1552395d2c4626ea13ac7b5afc50319b8d Mon Sep 17 00:00:00 2001 From: cxymds Date: Tue, 28 Jul 2026 17:05:04 +0800 Subject: [PATCH] fix(swift): enforce cumulative container quotas (#5378) --- crates/protocols/src/swift/container.rs | 142 ++++++++++++++-------- crates/protocols/src/swift/errors.rs | 4 + crates/protocols/src/swift/handler.rs | 14 +-- crates/protocols/src/swift/mod.rs | 2 +- crates/protocols/src/swift/quota.rs | 63 ++++++---- crates/protocols/src/swift/storage_api.rs | 20 ++- 6 files changed, 162 insertions(+), 83 deletions(-) diff --git a/crates/protocols/src/swift/container.rs b/crates/protocols/src/swift/container.rs index 8c9cf0e7c..ffbc24498 100644 --- a/crates/protocols/src/swift/container.rs +++ b/crates/protocols/src/swift/container.rs @@ -22,12 +22,19 @@ use super::storage_api::container::{ }; use super::types::Container; use super::{SwiftError, SwiftResult}; -use super::{get_swift_bucket_metadata, resolve_swift_object_store_handle, set_swift_bucket_metadata}; +use super::{get_swift_bucket_metadata, get_swift_bucket_usage, resolve_swift_object_store_handle, set_swift_bucket_metadata}; use rustfs_credentials::Credentials; use s3s::dto::{Tag, Tagging}; use sha2::{Digest, Sha256}; use tracing::{debug, error}; +fn required_bucket_usage(usage: &std::collections::HashMap, bucket: &str) -> SwiftResult<(u64, u64)> { + usage + .get(bucket) + .copied() + .ok_or_else(|| SwiftError::ServiceUnavailable("Container usage is unavailable".to_string())) +} + const LOG_COMPONENT_PROTOCOLS: &str = "protocols"; const LOG_SUBSYSTEM_SWIFT_CONTAINER: &str = "swift_container"; const EVENT_SWIFT_CONTAINER_STORAGE_STATE: &str = "swift_container_storage_state"; @@ -252,13 +259,24 @@ pub async fn list_containers(account: &str, credentials: &Credentials) -> SwiftR .list_bucket(&BucketOptions::default()) .await .map_err(|e| sanitize_storage_error("Container listing", e))?; + let usage = get_swift_bucket_usage() + .await + .map_err(|e| sanitize_storage_error("Container usage retrieval", e))? + .ok_or_else(|| SwiftError::ServiceUnavailable("Container usage is unavailable".to_string()))?; // Filter and convert buckets to containers let containers: Vec = bucket_infos .iter() .filter(|info| mapper.bucket_belongs_to_project(&info.name, &project_id)) - .filter_map(|info| bucket_info_to_container(info, &mapper, &project_id)) - .collect(); + .filter_map(|info| { + bucket_info_to_container(info, &mapper, &project_id).map(|mut container| { + let (count, bytes) = required_bucket_usage(&usage, &info.name)?; + container.count = count; + container.bytes = bytes; + Ok(container) + }) + }) + .collect::>>()?; debug!( event = EVENT_SWIFT_CONTAINER_STORAGE_STATE, @@ -368,6 +386,44 @@ pub struct ContainerMetadata { pub custom_metadata: std::collections::HashMap, } +async fn get_container_metadata_base( + account: &str, + container: &str, + credentials: &Credentials, +) -> SwiftResult<(String, BucketInfo, std::collections::HashMap)> { + let project_id = validate_account_access(account, credentials)?; + validate_container_name(container)?; + let bucket_name = ContainerMapper::default().swift_to_s3_bucket(container, &project_id); + let Some(store) = resolve_swift_object_store_handle() else { + return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string())); + }; + let bucket_info = store + .get_bucket_info(&bucket_name, &BucketOptions::default()) + .await + .map_err(|e| { + if e.to_string().contains("not found") || e.to_string().contains("NoSuchBucket") { + SwiftError::NotFound(format!("Container '{}' not found", container)) + } else { + sanitize_storage_error("Container metadata retrieval", e) + } + })?; + let custom_metadata = get_swift_bucket_metadata(&bucket_name) + .await + .ok() + .and_then(|bucket_meta| bucket_meta.tagging_config.as_ref().map(s3_tags_to_swift_metadata)) + .unwrap_or_default(); + Ok((bucket_name, bucket_info, custom_metadata)) +} + +pub(crate) async fn get_container_custom_metadata( + account: &str, + container: &str, + credentials: &Credentials, +) -> SwiftResult> { + let (_, _, custom_metadata) = get_container_metadata_base(account, container, credentials).await?; + Ok(custom_metadata) +} + /// Get container metadata (for HEAD operation) /// /// This function: @@ -382,60 +438,22 @@ pub struct ContainerMetadata { /// - Returns 404 Not Found if container doesn't exist #[allow(dead_code)] // Used by handler pub async fn get_container_metadata(account: &str, container: &str, credentials: &Credentials) -> SwiftResult { - // Validate account access and extract project_id - let project_id = validate_account_access(account, credentials)?; + let (bucket_name, bucket_info, custom_metadata) = get_container_metadata_base(account, container, credentials).await?; - // Validate container name - validate_container_name(container)?; - - // Create mapper with default config (tenant prefixing enabled) - let mapper = ContainerMapper::default(); - - // Convert Swift container name to S3 bucket name - let bucket_name = mapper.swift_to_s3_bucket(container, &project_id); - - // Get storage layer - let Some(store) = resolve_swift_object_store_handle() else { - return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string())); - }; - - // Get bucket info - let bucket_info = store - .get_bucket_info(&bucket_name, &BucketOptions::default()) + // Ecstore serves the persisted scanner snapshot through a bounded in-process cache. + // Single-node deployments overlay successful in-process mutations; distributed + // deployments use only the cluster-wide persisted snapshot. + let usage = get_swift_bucket_usage() .await - .map_err(|e| { - // Check if bucket not found - if e.to_string().contains("not found") || e.to_string().contains("NoSuchBucket") { - SwiftError::NotFound(format!("Container '{}' not found", container)) - } else { - sanitize_storage_error("Container metadata retrieval", e) - } - })?; + .map_err(|e| sanitize_storage_error("Container usage retrieval", e))? + .ok_or_else(|| SwiftError::ServiceUnavailable("Container usage is unavailable".to_string()))?; + let (object_count, bytes_used) = required_bucket_usage(&usage, &bucket_name)?; - // Load bucket metadata to get custom metadata from tags - let custom_metadata = match get_swift_bucket_metadata(&bucket_name).await { - Ok(bucket_meta) => { - if let Some(tagging) = &bucket_meta.tagging_config { - s3_tags_to_swift_metadata(tagging) - } else { - std::collections::HashMap::new() - } - } - Err(_) => { - // If metadata not available, return empty (container may be newly created) - std::collections::HashMap::new() - } - }; - - // Currently returns basic metadata with limitations: - // 1. Object count requires iterating all objects (expensive) - // 2. Bytes used requires summing all object sizes (expensive) - // 3. Custom metadata is now loaded from bucket tags ✅ Ok(ContainerMetadata { - object_count: 0, // TODO: implement object counting in backend - bytes_used: 0, // TODO: implement size aggregation in backend + object_count, + bytes_used, created: bucket_info.created, - custom_metadata, // ✅ Now populated from bucket tags! + custom_metadata, }) } @@ -1843,4 +1861,24 @@ mod tests { assert_eq!(tagging.tag_set.len(), 1); assert_eq!(tagging.tag_set[0].value.as_deref(), Some("new-archive")); } + + #[test] + fn nonempty_container_usage_preserves_authoritative_totals() { + let usage = std::collections::HashMap::from([("tenant-container".to_string(), (7, 4097))]); + + assert_eq!( + required_bucket_usage(&usage, "tenant-container").expect("authoritative bucket usage"), + (7, 4097) + ); + } + + #[test] + fn missing_container_usage_fails_closed() { + let usage = std::collections::HashMap::new(); + + assert!(matches!( + required_bucket_usage(&usage, "tenant-container"), + Err(SwiftError::ServiceUnavailable(_)) + )); + } } diff --git a/crates/protocols/src/swift/errors.rs b/crates/protocols/src/swift/errors.rs index 4526243b1..25275dc41 100644 --- a/crates/protocols/src/swift/errors.rs +++ b/crates/protocols/src/swift/errors.rs @@ -32,6 +32,8 @@ pub enum SwiftError { NotFound(String), /// 409 Conflict Conflict(String), + /// 411 Length Required + LengthRequired(String), /// 413 Request Entity Too Large (Payload Too Large) RequestEntityTooLarge(String), /// 422 Unprocessable Entity @@ -54,6 +56,7 @@ impl fmt::Display for SwiftError { SwiftError::Forbidden(msg) => write!(f, "Forbidden: {}", msg), SwiftError::NotFound(msg) => write!(f, "Not Found: {}", msg), SwiftError::Conflict(msg) => write!(f, "Conflict: {}", msg), + SwiftError::LengthRequired(msg) => write!(f, "Length Required: {}", msg), SwiftError::RequestEntityTooLarge(msg) => write!(f, "Request Entity Too Large: {}", msg), SwiftError::UnprocessableEntity(msg) => write!(f, "Unprocessable Entity: {}", msg), SwiftError::TooManyRequests { retry_after, .. } => { @@ -76,6 +79,7 @@ impl SwiftError { SwiftError::Forbidden(_) => StatusCode::FORBIDDEN, SwiftError::NotFound(_) => StatusCode::NOT_FOUND, SwiftError::Conflict(_) => StatusCode::CONFLICT, + SwiftError::LengthRequired(_) => StatusCode::LENGTH_REQUIRED, SwiftError::RequestEntityTooLarge(_) => StatusCode::PAYLOAD_TOO_LARGE, SwiftError::UnprocessableEntity(_) => StatusCode::UNPROCESSABLE_ENTITY, SwiftError::TooManyRequests { .. } => StatusCode::TOO_MANY_REQUESTS, diff --git a/crates/protocols/src/swift/handler.rs b/crates/protocols/src/swift/handler.rs index 7e9f02983..9961cddf2 100644 --- a/crates/protocols/src/swift/handler.rs +++ b/crates/protocols/src/swift/handler.rs @@ -652,14 +652,11 @@ async fn handle_authenticated_request( .await; } - // Check quota before upload (if Content-Length provided) - if let Some(content_length) = headers.get("content-length") - && let Ok(size_str) = content_length.to_str() - && let Ok(object_size) = size_str.parse::() - { - // Check if upload would exceed quota - super::quota::check_upload_quota(&account, &container, object_size, &credentials).await?; - } + let object_size = headers + .get("content-length") + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse::().ok()); + super::quota::check_upload_quota(&account, &container, object_size, &credentials).await?; // Check if versioning is enabled for this container if let Some(archive_container) = container::get_versions_location(&account, &container, &credentials).await? { @@ -1486,6 +1483,7 @@ fn swift_error_to_response(error: SwiftError) -> Response { SwiftError::Forbidden(msg) => (StatusCode::FORBIDDEN, msg.as_str()), SwiftError::NotFound(msg) => (StatusCode::NOT_FOUND, msg.as_str()), SwiftError::Conflict(msg) => (StatusCode::CONFLICT, msg.as_str()), + SwiftError::LengthRequired(msg) => (StatusCode::LENGTH_REQUIRED, msg.as_str()), SwiftError::RequestEntityTooLarge(msg) => (StatusCode::PAYLOAD_TOO_LARGE, msg.as_str()), SwiftError::UnprocessableEntity(msg) => (StatusCode::UNPROCESSABLE_ENTITY, msg.as_str()), SwiftError::TooManyRequests { .. } => (StatusCode::TOO_MANY_REQUESTS, "Rate limit exceeded"), diff --git a/crates/protocols/src/swift/mod.rs b/crates/protocols/src/swift/mod.rs index 528cce52b..2dba11b8d 100644 --- a/crates/protocols/src/swift/mod.rs +++ b/crates/protocols/src/swift/mod.rs @@ -61,7 +61,7 @@ pub use router::{SwiftRoute, SwiftRouter}; // Note: Container, Object, and SwiftMetadata types used by Swift implementation pub use storage_api::public_api::{SwiftGetObjectReader, SwiftObjectInfo, SwiftObjectOptions, SwiftPutObjReader}; pub(crate) use storage_api::public_api::{ - get_swift_bucket_metadata, resolve_swift_object_store_handle, set_swift_bucket_metadata, + get_swift_bucket_metadata, get_swift_bucket_usage, resolve_swift_object_store_handle, set_swift_bucket_metadata, }; #[allow(unused_imports)] pub use types::{Container, Object, SwiftMetadata}; diff --git a/crates/protocols/src/swift/quota.rs b/crates/protocols/src/swift/quota.rs index 9c79c3a9d..308161592 100644 --- a/crates/protocols/src/swift/quota.rs +++ b/crates/protocols/src/swift/quota.rs @@ -87,18 +87,17 @@ pub struct QuotaConfig { impl QuotaConfig { /// Load quota configuration from container metadata pub async fn load(account: &str, container_name: &str, credentials: &Credentials) -> SwiftResult { - // Get container metadata - let container_info = container::get_container_metadata(account, container_name, credentials).await?; + let custom_metadata = container::get_container_custom_metadata(account, container_name, credentials).await?; let mut config = QuotaConfig::default(); // Parse Quota-Bytes - if let Some(quota_bytes_str) = container_info.custom_metadata.get("x-container-meta-quota-bytes") { + if let Some(quota_bytes_str) = custom_metadata.get("x-container-meta-quota-bytes") { config.quota_bytes = quota_bytes_str.parse().ok(); } // Parse Quota-Count - if let Some(quota_count_str) = container_info.custom_metadata.get("x-container-meta-quota-count") { + if let Some(quota_count_str) = custom_metadata.get("x-container-meta-quota-count") { config.quota_count = quota_count_str.parse().ok(); } @@ -113,9 +112,12 @@ impl QuotaConfig { /// Check if adding an object would exceed quotas /// /// Returns Ok(()) if within quota, Err with 413 if exceeded - pub fn check_quota(&self, current_bytes: u64, current_count: u64, additional_bytes: u64) -> SwiftResult<()> { + pub fn check_quota(&self, current_bytes: u64, current_count: u64, additional_bytes: Option) -> SwiftResult<()> { // Check byte quota if let Some(max_bytes) = self.quota_bytes { + let additional_bytes = additional_bytes.ok_or_else(|| { + SwiftError::LengthRequired("Content-Length is required when a byte quota is configured".to_string()) + })?; let new_bytes = current_bytes.saturating_add(additional_bytes); if new_bytes > max_bytes { return Err(SwiftError::RequestEntityTooLarge(format!( @@ -147,7 +149,7 @@ impl QuotaConfig { pub async fn check_upload_quota( account: &str, container_name: &str, - object_size: u64, + object_size: Option, credentials: &Credentials, ) -> SwiftResult<()> { // Load quota config @@ -157,7 +159,6 @@ pub async fn check_upload_quota( if !quota.is_enabled() { return Ok(()); } - // Get current container usage let metadata = container::get_container_metadata(account, container_name, credentials).await?; @@ -173,7 +174,7 @@ pub async fn check_upload_quota( quota_bytes = ?quota.quota_bytes, current_count = metadata.object_count, quota_count = ?quota.quota_count, - object_size, + object_size = ?object_size, "swift quota state changed" ); @@ -235,7 +236,7 @@ mod tests { }; // Current: 500 bytes, adding 400 bytes = 900 total (within 1000 limit) - let result = config.check_quota(500, 0, 400); + let result = config.check_quota(500, 0, Some(400)); assert!(result.is_ok()); } @@ -247,7 +248,7 @@ mod tests { }; // Current: 500 bytes, adding 600 bytes = 1100 total (exceeds 1000 limit) - let result = config.check_quota(500, 0, 600); + let result = config.check_quota(500, 0, Some(600)); assert!(result.is_err()); match result { Err(SwiftError::RequestEntityTooLarge(msg)) => { @@ -265,7 +266,7 @@ mod tests { }; // Current: 500 bytes, adding 500 bytes = 1000 total (exactly at limit) - let result = config.check_quota(500, 0, 500); + let result = config.check_quota(500, 0, Some(500)); assert!(result.is_ok()); } @@ -277,7 +278,7 @@ mod tests { }; // Current: 5 objects, adding 1 = 6 total (within 10 limit) - let result = config.check_quota(0, 5, 100); + let result = config.check_quota(0, 5, Some(100)); assert!(result.is_ok()); } @@ -289,7 +290,7 @@ mod tests { }; // Current: 10 objects, adding 1 = 11 total (exceeds 10 limit) - let result = config.check_quota(0, 10, 100); + let result = config.check_quota(0, 10, Some(100)); assert!(result.is_err()); match result { Err(SwiftError::RequestEntityTooLarge(msg)) => { @@ -307,7 +308,7 @@ mod tests { }; // Current: 9 objects, adding 1 = 10 total (exactly at limit) - let result = config.check_quota(0, 9, 100); + let result = config.check_quota(0, 9, Some(100)); assert!(result.is_ok()); } @@ -319,7 +320,7 @@ mod tests { }; // Both within limits - let result = config.check_quota(500, 5, 400); + let result = config.check_quota(500, 5, Some(400)); assert!(result.is_ok()); } @@ -331,7 +332,7 @@ mod tests { }; // Bytes exceeded, count within - let result = config.check_quota(500, 5, 600); + let result = config.check_quota(500, 5, Some(600)); assert!(result.is_err()); } @@ -343,7 +344,7 @@ mod tests { }; // Count exceeded, bytes within - let result = config.check_quota(500, 10, 100); + let result = config.check_quota(500, 10, Some(100)); assert!(result.is_err()); } @@ -355,7 +356,7 @@ mod tests { }; // No limits, should always pass - let result = config.check_quota(999999, 999999, 999999); + let result = config.check_quota(999999, 999999, Some(999999)); assert!(result.is_ok()); } @@ -367,7 +368,7 @@ mod tests { }; // Zero limit means no uploads allowed - let result = config.check_quota(0, 0, 1); + let result = config.check_quota(0, 0, Some(1)); assert!(result.is_err()); } @@ -379,7 +380,7 @@ mod tests { }; // Zero limit means no objects allowed - let result = config.check_quota(0, 0, 100); + let result = config.check_quota(0, 0, Some(100)); assert!(result.is_err()); } @@ -391,8 +392,28 @@ mod tests { }; // Test saturating_add protection - let result = config.check_quota(u64::MAX - 100, 0, 200); + let result = config.check_quota(u64::MAX - 100, 0, Some(200)); // Should saturate to u64::MAX and compare against quota assert!(result.is_ok()); } + + #[test] + fn byte_quota_rejects_upload_without_content_length() { + let config = QuotaConfig { + quota_bytes: Some(1000), + quota_count: None, + }; + + assert!(matches!(config.check_quota(500, 0, None), Err(SwiftError::LengthRequired(_)))); + } + + #[test] + fn count_only_quota_checks_upload_without_content_length() { + let config = QuotaConfig { + quota_bytes: None, + quota_count: Some(10), + }; + + assert!(matches!(config.check_quota(0, 10, None), Err(SwiftError::RequestEntityTooLarge(_)))); + } } diff --git a/crates/protocols/src/swift/storage_api.rs b/crates/protocols/src/swift/storage_api.rs index 8dc6d3430..b65c711e5 100644 --- a/crates/protocols/src/swift/storage_api.rs +++ b/crates/protocols/src/swift/storage_api.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::collections::HashMap; use std::sync::Arc; pub(crate) use rustfs_ecstore::api::bucket::metadata::BucketMetadata as SwiftBucketMetadata; @@ -43,7 +44,9 @@ pub(crate) mod object { pub(crate) mod public_api { pub use super::{SwiftGetObjectReader, SwiftObjectInfo, SwiftObjectOptions, SwiftPutObjReader}; - pub(crate) use super::{get_swift_bucket_metadata, resolve_swift_object_store_handle, set_swift_bucket_metadata}; + pub(crate) use super::{ + get_swift_bucket_metadata, get_swift_bucket_usage, resolve_swift_object_store_handle, set_swift_bucket_metadata, + }; } pub(crate) mod versioning { @@ -62,3 +65,18 @@ pub(crate) async fn get_swift_bucket_metadata(bucket: &str) -> SwiftStorageResul pub(crate) async fn set_swift_bucket_metadata(bucket: String, metadata: SwiftBucketMetadata) -> SwiftStorageResult<()> { set_swift_bucket_metadata_in_backend(bucket, metadata).await } + +pub(crate) async fn get_swift_bucket_usage() -> SwiftStorageResult>> { + let Some(store) = resolve_swift_object_store_handle() else { + return Ok(None); + }; + let mut data_usage = rustfs_ecstore::api::data_usage::load_data_usage_from_backend_cached(store).await?; + rustfs_ecstore::api::data_usage::apply_bucket_usage_memory_overlay(&mut data_usage).await; + Ok(Some( + data_usage + .buckets_usage + .into_iter() + .map(|(bucket, usage)| (bucket, (usage.objects_count, usage.size))) + .collect(), + )) +}