From 94a6da6e830418c1cde5f5dc678842e17a7ed7e4 Mon Sep 17 00:00:00 2001 From: cxymds Date: Thu, 27 Aug 2026 18:33:56 +0800 Subject: [PATCH] feat(s3): enforce multipart presigned size limits (#6732) --- crates/ecstore/src/error/mod.rs | 5 + crates/ecstore/src/set_disk/ops/multipart.rs | 319 ++++++++++++++++-- crates/storage-api/src/error.rs | 4 + crates/utils/src/http/metadata_compat.rs | 2 + .../presigned-multipart-size-limit.md | 55 +++ rustfs/src/app/multipart_usecase.rs | 138 +++++++- rustfs/src/app/object/mod.rs | 1 + rustfs/src/app/object/put.rs | 4 +- rustfs/src/auth.rs | 136 ++++++++ rustfs/src/error.rs | 1 + rustfs/src/storage/access.rs | 22 +- 11 files changed, 629 insertions(+), 58 deletions(-) create mode 100644 docs/operations/presigned-multipart-size-limit.md diff --git a/crates/ecstore/src/error/mod.rs b/crates/ecstore/src/error/mod.rs index e7e50deb1..706f6e926 100644 --- a/crates/ecstore/src/error/mod.rs +++ b/crates/ecstore/src/error/mod.rs @@ -157,6 +157,8 @@ pub enum StorageError { InvalidPartNumber(usize), #[error("Your proposed upload is smaller than the minimum allowed size. Part {0} size {1} is less than minimum {2}")] EntityTooSmall(usize, i64, i64), + #[error("multipart upload size {0} exceeds the configured limit {1}")] + EntityTooLarge(u64, u64), // ── Erasure / Quorum ───────────────────────────────────────────── #[error("erasure read quorum")] @@ -554,6 +556,7 @@ impl Clone for StorageError { StorageError::DecommissionNotStarted => StorageError::DecommissionNotStarted, StorageError::InvalidPart(a, b, c) => StorageError::InvalidPart(*a, b.clone(), c.clone()), StorageError::EntityTooSmall(a, b, c) => StorageError::EntityTooSmall(*a, *b, *c), + StorageError::EntityTooLarge(a, b) => StorageError::EntityTooLarge(*a, *b), StorageError::DoneForNow => StorageError::DoneForNow, StorageError::DecommissionAlreadyRunning => StorageError::DecommissionAlreadyRunning, StorageError::RebalanceAlreadyRunning => StorageError::RebalanceAlreadyRunning, @@ -673,6 +676,7 @@ impl StorageError { StorageError::InsufficientWriteQuorum(_, _) => StorageErrorCode::InsufficientWriteQuorum, StorageError::PreconditionFailed => StorageErrorCode::PreconditionFailed, StorageError::EntityTooSmall(_, _, _) => StorageErrorCode::EntityTooSmall, + StorageError::EntityTooLarge(_, _) => StorageErrorCode::EntityTooLarge, StorageError::InvalidRangeSpec(_) => StorageErrorCode::InvalidRangeSpec, StorageError::NotModified => StorageErrorCode::NotModified, StorageError::InvalidPartNumber(_) => StorageErrorCode::InvalidPartNumber, @@ -795,6 +799,7 @@ impl StorageError { StorageErrorCode::EntityTooSmall => { Some(StorageError::EntityTooSmall(Default::default(), Default::default(), Default::default())) } + StorageErrorCode::EntityTooLarge => Some(StorageError::EntityTooLarge(Default::default(), Default::default())), StorageErrorCode::InvalidRangeSpec => Some(StorageError::InvalidRangeSpec(Default::default())), StorageErrorCode::NotModified => Some(StorageError::NotModified), StorageErrorCode::InvalidPartNumber => Some(StorageError::InvalidPartNumber(Default::default())), diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 13b120afc..52e905877 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -84,11 +84,13 @@ use rustfs_rio::TryGetIndex; use rustfs_utils::http::SSEC_ALGORITHM_HEADER; #[cfg(test)] use rustfs_utils::http::SUFFIX_COMPRESSION; +use rustfs_utils::http::{SUFFIX_MAX_TOTAL_OBJECT_SIZE, get_consistent_str}; use std::future::Future; #[cfg(test)] use std::sync::atomic::AtomicBool; #[cfg(any(test, feature = "test-util"))] use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Mutex, OnceLock}; #[cfg(any(test, feature = "test-util"))] use std::time::Duration; #[cfg(test)] @@ -97,6 +99,83 @@ use tokio::task::JoinSet; const MULTIPART_LIST_IO_CONCURRENCY: usize = 16; +static CAPPED_MULTIPART_STAGING: OnceLock>>> = OnceLock::new(); + +struct CappedMultipartStagingGuard { + upload_id_path: String, + permit: Option, +} + +impl Drop for CappedMultipartStagingGuard { + fn drop(&mut self) { + // Release the permit before checking the Arc count so a concurrent + // Abort/Complete cleanup can remove the now-unused map entry. + self.permit.take(); + remove_capped_multipart_staging_semaphore(&self.upload_id_path); + } +} + +fn capped_multipart_staging_semaphore(upload_id_path: &str) -> Arc { + CAPPED_MULTIPART_STAGING + .get_or_init(|| Mutex::new(HashMap::new())) + .lock() + .expect("capped multipart staging semaphore map should not be poisoned") + .entry(upload_id_path.to_owned()) + .or_insert_with(|| Arc::new(tokio::sync::Semaphore::new(1))) + .clone() +} + +fn remove_capped_multipart_staging_semaphore(upload_id_path: &str) { + if let Some(map) = CAPPED_MULTIPART_STAGING.get() { + let mut map = map + .lock() + .expect("capped multipart staging semaphore map should not be poisoned"); + let removable = map + .get(upload_id_path) + .is_some_and(|semaphore| Arc::strong_count(semaphore) == 1); + if removable { + map.remove(upload_id_path); + } + } +} + +fn multipart_size_limit_from_metadata(metadata: &HashMap) -> Result> { + if !contains_key_str(metadata, SUFFIX_MAX_TOTAL_OBJECT_SIZE) { + return Ok(None); + } + + let Some(value) = get_consistent_str(metadata, SUFFIX_MAX_TOTAL_OBJECT_SIZE) else { + return Err(Error::InvalidArgument( + "multipart upload".to_string(), + SUFFIX_MAX_TOTAL_OBJECT_SIZE.to_string(), + "missing or conflicting internal size limit".to_string(), + )); + }; + + let limit = value.parse::().map_err(|_| { + Error::InvalidArgument( + "multipart upload".to_string(), + SUFFIX_MAX_TOTAL_OBJECT_SIZE.to_string(), + "invalid internal size limit".to_string(), + ) + })?; + Ok(Some(limit)) +} + +fn admitted_multipart_size(current: u64, candidate: u64, limit: u64) -> Result { + let total = current.checked_add(candidate).ok_or_else(|| { + Error::InvalidArgument( + "multipart upload".to_string(), + SUFFIX_MAX_TOTAL_OBJECT_SIZE.to_string(), + "logical size overflow".to_string(), + ) + })?; + if total > limit { + return Err(Error::EntityTooLarge(total, limit)); + } + Ok(total) +} + pub(crate) struct StaleMultipartCleanupGuard { file_info: FileInfo, upload_path: String, @@ -115,8 +194,13 @@ impl StaleMultipartCleanupGuard { pub(crate) async fn delete(self, set: &SetDisks) -> Result<()> { fence_commit_on_lock_loss(Some(&self.lock_guard), "stale_multipart_cleanup", &self.upload_path)?; - set.delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &self.upload_path, self.write_quorum) - .await + let result = set + .delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &self.upload_path, self.write_quorum) + .await; + if result.is_ok() { + remove_capped_multipart_staging_semaphore(&self.upload_path); + } + result } } @@ -635,6 +719,61 @@ async fn multipart_upload_paths_on_disk(disk: DiskStore, bucket: &str) -> disk:: } impl SetDisks { + async fn current_multipart_logical_size( + &self, + bucket: &str, + object: &str, + upload_id: &str, + upload_id_path: &str, + fi: &FileInfo, + replacing_part: usize, + ) -> Result { + let online_disks = self.get_disks_internal().await; + let read_quorum = fi.read_quorum(self.default_read_quorum()); + let part_path = format!( + "{}{}", + path_join_buf(&[ + upload_id_path, + fi.data_dir.map(|v| v.to_string()).unwrap_or_default().as_str(), + ]), + SLASH_SEPARATOR + ); + let part_numbers = match Self::list_parts(&online_disks, &part_path, read_quorum).await { + Ok(parts) => parts, + Err(DiskError::FileNotFound) => return Ok(0), + Err(err) => return Err(to_object_err(err.into(), vec![bucket, object, upload_id])), + }; + if part_numbers.is_empty() { + return Ok(0); + } + let part_meta_paths = part_numbers + .iter() + .map(|number| format!("{part_path}part.{number}.meta")) + .collect::>(); + let existing_parts = + Self::read_parts(&online_disks, RUSTFS_META_MULTIPART_BUCKET, &part_meta_paths, &part_numbers, read_quorum) + .await + .map_err(|err| to_object_err(err.into(), vec![bucket, object, upload_id]))?; + + existing_parts.into_iter().try_fold(0_u64, |total, part| { + if part.error.is_some() || part.number == replacing_part { + return if part.error.is_some() { + Err(Error::PartMissingOrCorrupt) + } else { + Ok(total) + }; + } + let part_size = u64::try_from(part.actual_size).map_err(|_| Error::PartMissingOrCorrupt)?; + total.checked_add(part_size).ok_or_else(|| { + Error::InvalidArgument( + "multipart upload".to_string(), + SUFFIX_MAX_TOTAL_OBJECT_SIZE.to_string(), + "logical size overflow".to_string(), + ) + }) + }) + } + async fn discover_multipart_upload_paths( &self, orig_bucket: &str, @@ -1130,9 +1269,18 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { crate::hp_guard!("SetDisks::put_object_part"); let upload_id_path = Self::get_multipart_upload_dir(bucket, object, upload_id, opts.data_movement); - let (fi, _) = self + let (fi, _) = match self .check_upload_id_exists_with_opts(bucket, object, upload_id, true, opts) - .await?; + .await + { + Ok(value) => value, + Err(err @ Error::InvalidUploadID(..)) => { + remove_capped_multipart_staging_semaphore(&upload_id_path); + return Err(err); + } + Err(err) => return Err(err), + }; + let multipart_size_limit = multipart_size_limit_from_metadata(&fi.metadata)?; ensure_data_movement_upload_access(&fi, bucket, object, upload_id, opts)?; ensure_multipart_bucket_incarnation(&self.ctx, &fi, bucket, object, upload_id, opts.expected_bucket_incarnation_id) .await?; @@ -1165,6 +1313,44 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { let part_suffix = format!("part.{part_id}"); let tmp_part = format!("{}x{}", Uuid::new_v4(), OffsetDateTime::now_utc().unix_timestamp()); let tmp_part_path = Arc::new(format!("{tmp_part}/{part_suffix}")); + let part_path = format!("{}/{}/{}", upload_id_path, fi.data_dir.unwrap_or_default(), part_suffix); + let part_lock_path = format!("{upload_id_path}/{part_suffix}"); + + // Keep at most one capped part staging locally per upload. The + // distributed lock below is held only for the durable admission check; + // it is reacquired for the short final rename, so Complete/Abort are + // not blocked behind a slow body upload. + let _capped_staging_guard = if multipart_size_limit.is_some() { + Some(CappedMultipartStagingGuard { + upload_id_path: upload_id_path.clone(), + permit: Some( + capped_multipart_staging_semaphore(&upload_id_path) + .acquire_owned() + .await + .map_err(|_| Error::other("capped multipart staging semaphore closed"))?, + ), + }) + } else { + None + }; + + if let Some(limit) = multipart_size_limit { + let admission_guard = self + .acquire_write_lock_diag("put_object_part_admission", RUSTFS_META_MULTIPART_BUCKET, &upload_id_path) + .await?; + let declared_size = if data.size() >= 0 { + u64::try_from(data.size()).map_err(|_| Error::PartMissingOrCorrupt)? + } else if data.actual_size() >= 0 { + u64::try_from(data.actual_size()).map_err(|_| Error::PartMissingOrCorrupt)? + } else { + return Err(Error::PartMissingOrCorrupt); + }; + let current_size = self + .current_multipart_logical_size(bucket, object, upload_id, &upload_id_path, &fi, part_id) + .await?; + admitted_multipart_size(current_size, declared_size, limit)?; + drop(admission_guard); + } let result: Result = async { let erasure = @@ -1365,30 +1551,21 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { .await?; } - let part_path = format!("{}/{}/{}", upload_id_path, fi.data_dir.unwrap_or_default(), part_suffix); - let part_lock_path = format!("{upload_id_path}/{part_suffix}"); - #[cfg(test)] pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartBeforeLockAcquire).await; - // Serialize only same-part commits (rename_part), not the whole upload. - // Each concurrent stream writes to its own unique temp dir (see - // `tmp_part` above), so the encode/stream phase never conflicts and must - // stay lock-free — holding a lock across it would serialize slow - // re-transmits of the same part and defeat the S3 "last finisher wins" - // semantics. The mixed-generation hazard is confined to rename_part, - // where two temp parts are moved cross-disk onto the SAME final - // part_path: interleaving there can leave shards from two generations, - // each individually bitrot-valid, that only surface as silent corruption - // at read time (backlog#853). A write lock scoped to this part number - // makes each same-part commit atomic across disks, so the last committer - // wins consistently, while different part numbers commit onto disjoint - // part paths and stay concurrent (issue#5961 — an uploadId-wide write - // lock serialized them into 503 lock-acquire timeouts). The shared - // uploadId read lock keeps completion/abort (which take the uploadId - // write lock) from racing any in-flight part commit; a guarded - // completion takes the object lock before the upload lock to preserve - // global ordering. - let (_upload_commit_guard, _part_commit_guard) = if opts.no_lock { + // Capped uploads reacquire the upload-wide write lock for the + // final durable check and rename. Uncapped uploads retain the + // concurrent encode path and only serialize the final same-part + // rename; completion/abort use the upload-wide write lock. + let (_upload_commit_guard, _part_commit_guard) = if multipart_size_limit.is_some() { + let upload_guard = self + .acquire_write_lock_diag("put_object_part_commit", RUSTFS_META_MULTIPART_BUCKET, &upload_id_path) + .await?; + let part_guard = self + .acquire_write_lock_diag("put_object_part_commit", RUSTFS_META_MULTIPART_BUCKET, &part_lock_path) + .await?; + (Some(upload_guard), Some(part_guard)) + } else if opts.no_lock { (None, None) } else { let upload_guard = self @@ -1400,8 +1577,16 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { (Some(upload_guard), Some(part_guard)) }; let (commit_fi, _) = self - .check_upload_id_exists_with_opts(bucket, object, upload_id, false, opts) + .check_upload_id_exists_with_opts(bucket, object, upload_id, multipart_size_limit.is_some(), opts) .await?; + let commit_size_limit = multipart_size_limit_from_metadata(&commit_fi.metadata)?; + if commit_size_limit != multipart_size_limit { + return Err(Error::InvalidArgument( + "multipart upload".to_string(), + SUFFIX_MAX_TOTAL_OBJECT_SIZE.to_string(), + "size limit metadata changed or is missing".to_string(), + )); + } ensure_data_movement_upload_access(&commit_fi, bucket, object, upload_id, opts)?; ensure_multipart_bucket_incarnation( &self.ctx, @@ -1431,6 +1616,14 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { } ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?; + if let Some(limit) = commit_size_limit { + let current_size = self + .current_multipart_logical_size(bucket, object, upload_id, &upload_id_path, &commit_fi, part_id) + .await?; + let candidate_size = u64::try_from(actual_size).map_err(|_| Error::PartMissingOrCorrupt)?; + admitted_multipart_size(current_size, candidate_size, limit)?; + } + let _ = self .rename_part( &shuffle_disks, @@ -1891,12 +2084,17 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?; let upload_id_path = Self::get_multipart_upload_dir(bucket, object, upload_id, opts.data_movement); - self.delete_all_with_quorum( - RUSTFS_META_MULTIPART_BUCKET, - &upload_id_path, - fi.write_quorum(self.default_write_quorum()), - ) - .await + let result = self + .delete_all_with_quorum( + RUSTFS_META_MULTIPART_BUCKET, + &upload_id_path, + fi.write_quorum(self.default_write_quorum()), + ) + .await; + if result.is_ok() { + remove_capped_multipart_staging_semaphore(&upload_id_path); + } + result } // complete_multipart_upload finished #[tracing::instrument(skip(self))] @@ -2016,6 +2214,27 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { return Err(Error::other("part result number err")); } + if let Some(limit) = multipart_size_limit_from_metadata(&fi.metadata)? { + let mut total = 0_u64; + for part in &object_parts { + if part.error.is_some() { + return Err(Error::PartMissingOrCorrupt); + } + let part_size = u64::try_from(part.actual_size).map_err(|_| Error::PartMissingOrCorrupt)?; + total = total.checked_add(part_size).ok_or_else(|| { + Error::InvalidArgument( + "multipart upload".to_string(), + SUFFIX_MAX_TOTAL_OBJECT_SIZE.to_string(), + "logical size overflow".to_string(), + ) + })?; + } + if total > limit { + return Err(Error::EntityTooLarge(total, limit)); + } + rustfs_utils::http::metadata_compat::remove_str(&mut fi.metadata, SUFFIX_MAX_TOTAL_OBJECT_SIZE); + } + let mut checksum_type = rustfs_rio::ChecksumType::NONE; if let Some(cs) = fi.metadata.get(rustfs_rio::RUSTFS_MULTIPART_CHECKSUM) { @@ -3024,13 +3243,17 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { Ok(ObjectInfo::from_file_info(&fi, &commit_bucket, &commit_object, commit_is_versioned)) }; - if detach_commit_owner { + let result = if detach_commit_owner { tokio::spawn(commit) .await .map_err(|err| Error::other(format!("complete_multipart_upload commit task failed: {err}")))? } else { commit.await + }; + if result.is_ok() { + remove_capped_multipart_staging_semaphore(&upload_id_path); } + result } } @@ -3096,6 +3319,34 @@ mod tests { assert!(multipart_bucket_incarnation_id(&nil_metadata).is_err()); } + #[test] + fn multipart_size_limit_metadata_is_dual_key_and_fail_closed() { + let mut metadata = HashMap::new(); + insert_str(&mut metadata, SUFFIX_MAX_TOTAL_OBJECT_SIZE, "100".to_string()); + assert_eq!(multipart_size_limit_from_metadata(&metadata).unwrap(), Some(100)); + + metadata.insert("x-minio-internal-max-total-object-size".to_string(), "101".to_string()); + assert!(multipart_size_limit_from_metadata(&metadata).is_err()); + + let mut invalid = HashMap::new(); + insert_str(&mut invalid, SUFFIX_MAX_TOTAL_OBJECT_SIZE, "-1".to_string()); + assert!(multipart_size_limit_from_metadata(&invalid).is_err()); + } + + #[test] + fn multipart_size_admission_handles_boundaries_and_overflow() { + assert_eq!(admitted_multipart_size(90, 10, 100).unwrap(), 100); + assert!(matches!( + admitted_multipart_size(90, 11, 100), + Err(StorageError::EntityTooLarge(101, 100)) + )); + assert!(admitted_multipart_size(u64::MAX - 1, 1, u64::MAX).is_ok()); + assert!(matches!( + admitted_multipart_size(u64::MAX, 1, u64::MAX), + Err(StorageError::InvalidArgument(_, _, _)) + )); + } + #[test] fn multipart_bucket_incarnation_gate_accepts_only_current_or_same_lifetime_legacy_uploads() { let expected = Uuid::new_v4(); diff --git a/crates/storage-api/src/error.rs b/crates/storage-api/src/error.rs index 85873ae8b..b24fedfe4 100644 --- a/crates/storage-api/src/error.rs +++ b/crates/storage-api/src/error.rs @@ -81,6 +81,7 @@ pub enum StorageErrorCode { InsufficientWriteQuorum, PreconditionFailed, EntityTooSmall, + EntityTooLarge, InvalidRangeSpec, NotModified, InvalidPartNumber, @@ -169,6 +170,7 @@ impl StorageErrorCode { Self::InsufficientWriteQuorum => 0x3A, Self::PreconditionFailed => 0x3B, Self::EntityTooSmall => 0x3C, + Self::EntityTooLarge => 0x56, Self::InvalidRangeSpec => 0x3D, Self::NotModified => 0x3E, Self::InvalidPartNumber => 0x3F, @@ -257,6 +259,7 @@ impl StorageErrorCode { 0x3A => Some(Self::InsufficientWriteQuorum), 0x3B => Some(Self::PreconditionFailed), 0x3C => Some(Self::EntityTooSmall), + 0x56 => Some(Self::EntityTooLarge), 0x3D => Some(Self::InvalidRangeSpec), 0x3E => Some(Self::NotModified), 0x3F => Some(Self::InvalidPartNumber), @@ -350,6 +353,7 @@ mod tests { (StorageErrorCode::InsufficientWriteQuorum, 0x3A), (StorageErrorCode::PreconditionFailed, 0x3B), (StorageErrorCode::EntityTooSmall, 0x3C), + (StorageErrorCode::EntityTooLarge, 0x56), (StorageErrorCode::InvalidRangeSpec, 0x3D), (StorageErrorCode::NotModified, 0x3E), (StorageErrorCode::InvalidPartNumber, 0x3F), diff --git a/crates/utils/src/http/metadata_compat.rs b/crates/utils/src/http/metadata_compat.rs index aa403d108..e1857d354 100644 --- a/crates/utils/src/http/metadata_compat.rs +++ b/crates/utils/src/http/metadata_compat.rs @@ -44,6 +44,8 @@ pub const SUFFIX_COMPRESSION: &str = "compression"; pub const SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT: &str = "replication-preserve-ciphertext"; pub const SUFFIX_COMPRESSION_SIZE: &str = "compression-size"; pub const SUFFIX_ACTUAL_SIZE: &str = "actual-size"; +/// Maximum logical object size for a capability-bound multipart upload. +pub const SUFFIX_MAX_TOTAL_OBJECT_SIZE: &str = "max-total-object-size"; pub const SUFFIX_ACTUAL_OBJECT_SIZE: &str = "actual-object-size"; /// Used by replication; key stored with capital A pub const SUFFIX_ACTUAL_OBJECT_SIZE_CAP: &str = "Actual-Object-Size"; diff --git a/docs/operations/presigned-multipart-size-limit.md b/docs/operations/presigned-multipart-size-limit.md new file mode 100644 index 000000000..4769b6247 --- /dev/null +++ b/docs/operations/presigned-multipart-size-limit.md @@ -0,0 +1,55 @@ +# Presigned multipart total-size limit + +RustFS V2 supports an optional capability on a signed or SigV4-presigned +`CreateMultipartUpload` request: + +```text +x-rustfs-max-total-object-size= +``` + +The backend must include the parameter before calculating the SigV4 +signature. It is part of the canonical query and cannot be added, removed, or +changed by the browser. RustFS stores the verified limit in the multipart +upload session and applies it to every `UploadPart` and to +`CompleteMultipartUpload`. + +Backend pseudocode (the custom query must be present before signing): + +```text +uri = "/photos/archive.zip?uploads" +uri += "&x-rustfs-max-total-object-size=104857600" +presigned_url = sigv4_presign("POST", uri, credentials) +# Return presigned_url to the browser. Never append the parameter afterwards. +``` + +The resulting flow is: + +1. The backend signs `CreateMultipartUpload?...&x-rustfs-max-total-object-size=104857600`. +2. RustFS verifies the SigV4 request and persists the limit with the upload ID. +3. The browser uploads parts using the returned upload ID. +4. RustFS rejects a part whose declared logical size would exceed the remaining + budget and rejects completion if the server-side part metadata exceeds the + limit. + +The limit is measured in logical object bytes (`actual_size`), not erasure, +encryption, or compression bytes. Replacing an existing part uses replacement +semantics: the old part size is removed before the new part size is admitted. +Unknown-length parts are rejected for capped sessions rather than buffered +without a bound. Capped parts are admitted under an upload-wide write lock +before temporary shards are created and use a per-upload staging permit to +bound local in-flight data. The distributed lock is released while the body is +read and reacquired for the final check/rename, so Complete and Abort are not +blocked behind a slow upload. The normal request-body stall timeout releases +the staging permit when a client stops sending. + +The parameter is accepted only on `CreateMultipartUpload`. Supplying it on +`UploadPart`, `CompleteMultipartUpload`, `AbortMultipartUpload`, listing, or +copy operations returns `InvalidRequest`; those requests use the persisted +session state. A multipart upload created without this parameter remains +unlimited for backward compatibility. The V1 single-request capability +(`x-rustfs-max-content-length`) is independent and is not a multipart limit. + +Because enforcement happens in the multipart data plane, every node that may +receive requests for a capped upload must run the V2 implementation. During a +rolling upgrade, route capped uploads only to upgraded nodes; older nodes treat +the internal metadata as unknown and cannot enforce the limit. diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 3accb0fe9..557f2d834 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -67,6 +67,7 @@ use super::storage_api::multipart_usecase::sse::{ use super::storage_api::multipart_usecase::{ StorageObjectInfo as ObjectInfo, StorageObjectOptions as ObjectOptions, StoragePutObjReader as PutObjReader, }; +use crate::app::object::{guard_put_object_body_read_timeout, put_object_body_read_timeout}; use crate::app::object_data_cache::{ ObjectDataCacheAdapter, invalidate_object_data_cache_after_complete_multipart_success, invalidate_object_data_cache_before_mutation, @@ -78,7 +79,11 @@ use crate::app::object_usecase::{ use crate::app::runtime_sources::{ AppContext, current_app_context, current_object_data_cache_for_context, current_object_store_handle_for_context, }; -use crate::auth::{VerifiedPresignedRequest, reject_presigned_put_max_content_length_for_other_operation}; +use crate::auth::{ + VerifiedPresignedRequest, VerifiedSigV4Request, parse_presigned_multipart_max_total_object_size, + reject_presigned_multipart_max_total_object_size_for_other_operation, + reject_presigned_put_max_content_length_for_other_operation, +}; use crate::capacity::record_capacity_write; use crate::error::ApiError; use crate::table_catalog; @@ -92,8 +97,9 @@ use rustfs_utils::CompressionAlgorithm; #[cfg(test)] use rustfs_utils::http::insert_header; use rustfs_utils::http::{ - SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT, SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, - SUFFIX_SOURCE_REPLICATION_REQUEST, contains_key_str, get_header, get_source_scheme, + SUFFIX_MAX_TOTAL_OBJECT_SIZE, SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT, SUFFIX_REPLICATION_STATUS, + SUFFIX_REPLICATION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_REQUEST, contains_key_str, get_consistent_str, get_header, + get_source_scheme, headers::{AMZ_DECODED_CONTENT_LENGTH, AMZ_OBJECT_TAGGING, AMZ_STORAGE_CLASS}, insert_str, }; @@ -108,6 +114,7 @@ use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; use std::collections::{HashMap, HashSet}; use std::str::FromStr; use std::sync::Arc; +use std::time::Duration; use tokio::sync::RwLock; use tokio_util::io::StreamReader; use tracing::{instrument, warn}; @@ -226,6 +233,22 @@ fn create_multipart_upload_metadata( metadata } +fn multipart_max_total_object_size(metadata: &HashMap) -> S3Result> { + if !contains_key_str(metadata, SUFFIX_MAX_TOTAL_OBJECT_SIZE) { + return Ok(None); + } + + let value = get_consistent_str(metadata, SUFFIX_MAX_TOTAL_OBJECT_SIZE).ok_or_else(|| { + S3Error::with_message( + S3ErrorCode::InvalidRequest, + "multipart size capability metadata is missing or inconsistent".to_string(), + ) + })?; + value.parse::().map(Some).map_err(|_| { + S3Error::with_message(S3ErrorCode::InvalidRequest, "multipart size capability metadata is invalid".to_string()) + }) +} + /// A multipart session advertises disk compression only when the staged-rollout /// switch (`RUSTFS_COMPRESSION_MULTIPART_ENABLED`) is on, the object key/headers /// qualify, AND the session is not an SSE-C ciphertext-passthrough replication @@ -398,6 +421,11 @@ impl DefaultMultipartUsecase { &self, req: S3Request, ) -> S3Result> { + reject_presigned_multipart_max_total_object_size_for_other_operation( + &req.headers, + req.uri.query(), + req.extensions.get::().is_some(), + )?; reject_presigned_put_max_content_length_for_other_operation( &req.headers, req.uri.query(), @@ -444,6 +472,11 @@ impl DefaultMultipartUsecase { &self, req: S3Request, ) -> S3Result> { + reject_presigned_multipart_max_total_object_size_for_other_operation( + &req.headers, + req.uri.query(), + req.extensions.get::().is_some(), + )?; reject_presigned_put_max_content_length_for_other_operation( &req.headers, req.uri.query(), @@ -752,6 +785,11 @@ impl DefaultMultipartUsecase { &self, req: S3Request, ) -> S3Result> { + let multipart_max_total_object_size = parse_presigned_multipart_max_total_object_size( + &req.headers, + req.uri.query(), + req.extensions.get::().is_some(), + )?; reject_presigned_put_max_content_length_for_other_operation( &req.headers, req.uri.query(), @@ -807,6 +845,9 @@ impl DefaultMultipartUsecase { )?; let mut metadata = create_multipart_upload_metadata(input_metadata, &req.headers, tagging, storage_class.as_ref()); + if let Some(limit) = multipart_max_total_object_size { + insert_str(&mut metadata, SUFFIX_MAX_TOTAL_OBJECT_SIZE, limit.to_string()); + } let has_explicit_object_lock_retention = object_lock_mode.is_some() || object_lock_retain_until_date.is_some() @@ -978,6 +1019,11 @@ impl DefaultMultipartUsecase { #[instrument(level = "debug", skip(self, req))] #[hotpath::measure(impl_type = "MultipartUsecase")] pub async fn execute_upload_part(&self, req: S3Request) -> S3Result> { + reject_presigned_multipart_max_total_object_size_for_other_operation( + &req.headers, + req.uri.query(), + req.extensions.get::().is_some(), + )?; reject_presigned_put_max_content_length_for_other_operation( &req.headers, req.uri.query(), @@ -1006,6 +1052,40 @@ impl DefaultMultipartUsecase { let mut size = resolve_upload_part_size(&req.headers, content_length)?; let mut body_stream = body.ok_or_else(|| s3_error!(IncompleteBody))?; + let Some(store) = self.object_store() else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + let fi = store + .get_multipart_info(&bucket, &key, &upload_id, &opts) + .await + .map_err(ApiError::from)?; + let max_total_object_size = multipart_max_total_object_size(&fi.user_defined)?; + if max_total_object_size.is_some() && size.is_some_and(|size| size < 0) { + return Err(S3Error::new(S3ErrorCode::UnexpectedContent)); + } + if max_total_object_size.is_some() && size.is_none() { + return Err(S3Error::new(S3ErrorCode::UnexpectedContent)); + } + if let (Some(limit), Some(size)) = (max_total_object_size, size) + && u64::try_from(size).is_ok_and(|size| size > limit) + { + return Err(S3Error::new(S3ErrorCode::EntityTooLarge)); + } + if max_total_object_size.is_some() { + let request_id = req + .extensions + .get::() + .map(|ctx| ctx.request_id.clone()) + .unwrap_or_default(); + body_stream = guard_put_object_body_read_timeout( + body_stream, + &bucket, + &key, + &request_id, + content_length, + put_object_body_read_timeout().max(Duration::from_secs(rustfs_config::DEFAULT_HTTP_REQUEST_BODY_READ_TIMEOUT)), + ); + } if size.is_none() { let mut total = 0i64; @@ -1026,16 +1106,6 @@ impl DefaultMultipartUsecase { body_stream = StreamingBlob::wrap(stream); } - // Get multipart info early to check if managed encryption will be applied - let Some(store) = self.object_store() else { - return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); - }; - - let fi = store - .get_multipart_info(&bucket, &key, &upload_id, &opts) - .await - .map_err(ApiError::from)?; - let mut size = size.ok_or_else(|| s3_error!(UnexpectedContent))?; let ingress_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(std::time::Instant::now); @@ -1250,6 +1320,11 @@ impl DefaultMultipartUsecase { &self, req: S3Request, ) -> S3Result> { + reject_presigned_multipart_max_total_object_size_for_other_operation( + &req.headers, + req.uri.query(), + req.extensions.get::().is_some(), + )?; reject_presigned_put_max_content_length_for_other_operation( &req.headers, req.uri.query(), @@ -1302,6 +1377,11 @@ impl DefaultMultipartUsecase { } pub async fn execute_list_parts(&self, req: S3Request) -> S3Result> { + reject_presigned_multipart_max_total_object_size_for_other_operation( + &req.headers, + req.uri.query(), + req.extensions.get::().is_some(), + )?; reject_presigned_put_max_content_length_for_other_operation( &req.headers, req.uri.query(), @@ -1338,6 +1418,11 @@ impl DefaultMultipartUsecase { &self, req: S3Request, ) -> S3Result> { + reject_presigned_multipart_max_total_object_size_for_other_operation( + &req.headers, + req.uri.query(), + req.extensions.get::().is_some(), + )?; reject_presigned_put_max_content_length_for_other_operation( &req.headers, req.uri.query(), @@ -1441,6 +1526,7 @@ impl DefaultMultipartUsecase { .get_multipart_info(&bucket, &key, &upload_id, &dst_opts) .await .map_err(ApiError::from)?; + let destination_size_limit = multipart_max_total_object_size(&mp_info.user_defined)?; EncryptionRequest { bucket: &bucket, key: &key, @@ -1523,19 +1609,25 @@ impl DefaultMultipartUsecase { return Err(s3_error!(PreconditionFailed)); } + let source_logical_size = match src_info.get_actual_size() { + Ok(size) if size >= 0 => size, + Ok(_) | Err(_) if destination_size_limit.is_some() => { + return Err(S3Error::new(S3ErrorCode::UnexpectedContent)); + } + Ok(_) | Err(_) => src_info.size, + }; + let (_start_offset, length) = if let Some(ref range_spec) = rs { // Copy-source ranges are expressed over the logical plaintext object. // Encrypted (and compressed) objects have a larger or smaller physical // representation, so validating against `size` rejects valid later parts. - let validation_size = src_info.get_actual_size().unwrap_or(src_info.size); - - validate_copy_source_range_not_exceeds(range_spec, validation_size)?; + validate_copy_source_range_not_exceeds(range_spec, source_logical_size)?; range_spec - .get_offset_length(validation_size) + .get_offset_length(source_logical_size) .map_err(|e| S3Error::with_message(S3ErrorCode::InvalidRange, e.to_string()))? } else { - (0, src_info.size) + (0, source_logical_size) }; let is_disk_compressed = @@ -2137,6 +2229,16 @@ mod tests { assert_eq!(metadata.get(AMZ_OBJECT_TAGGING), Some(&"project=rustfs".to_string())); } + #[test] + fn multipart_max_total_object_size_reads_compatible_internal_metadata() { + let mut metadata = HashMap::new(); + insert_str(&mut metadata, SUFFIX_MAX_TOTAL_OBJECT_SIZE, "104857600".to_string()); + assert_eq!(multipart_max_total_object_size(&metadata).unwrap(), Some(104_857_600)); + + metadata.insert("x-minio-internal-max-total-object-size".to_string(), "1".to_string()); + assert!(multipart_max_total_object_size(&metadata).is_err()); + } + #[tokio::test] async fn execute_complete_multipart_upload_rejects_missing_parts_payload() { let input = CompleteMultipartUploadInput::builder() diff --git a/rustfs/src/app/object/mod.rs b/rustfs/src/app/object/mod.rs index cb30c01b9..1e42d7036 100644 --- a/rustfs/src/app/object/mod.rs +++ b/rustfs/src/app/object/mod.rs @@ -195,6 +195,7 @@ pub(crate) use self::delete::*; pub(crate) use self::extract::*; pub(crate) use self::get::*; use self::put::*; +pub(crate) use self::put::{guard_put_object_body_read_timeout, put_object_body_read_timeout}; pub(crate) use self::shared::*; #[cfg(test)] use self::test_support::*; diff --git a/rustfs/src/app/object/put.rs b/rustfs/src/app/object/put.rs index 2638cea36..e44720ef7 100644 --- a/rustfs/src/app/object/put.rs +++ b/rustfs/src/app/object/put.rs @@ -90,7 +90,7 @@ fn resolve_put_object_authoritative_size(headers: &HeaderMap, content_length: Op /// Returns `Duration::ZERO` when disabled (`RUSTFS_HTTP_REQUEST_BODY_READ_TIMEOUT=0`), /// in which case [`guard_put_object_body_read_timeout`] passes the body through /// untouched. -fn put_object_body_read_timeout() -> Duration { +pub(crate) fn put_object_body_read_timeout() -> Duration { Duration::from_secs(rustfs_utils::get_env_u64( rustfs_config::ENV_HTTP_REQUEST_BODY_READ_TIMEOUT, rustfs_config::DEFAULT_HTTP_REQUEST_BODY_READ_TIMEOUT, @@ -260,7 +260,7 @@ impl ByteStream for RequestBodyReadTimeout { /// Wrap an incoming request body with [`RequestBodyReadTimeout`] unless the /// feature is disabled (`timeout == 0`), in which case the body is returned /// untouched. `remaining_length` is preserved via [`StreamingBlob::new`]. -fn guard_put_object_body_read_timeout( +pub(crate) fn guard_put_object_body_read_timeout( body: StreamingBlob, bucket: &str, key: &str, diff --git a/rustfs/src/auth.rs b/rustfs/src/auth.rs index ed20d6e1f..7b7c3ff54 100644 --- a/rustfs/src/auth.rs +++ b/rustfs/src/auth.rs @@ -53,6 +53,7 @@ const EVENT_SESSION_TOKEN_EXTRACTION: &str = "session_token_extraction"; /// RustFS-specific query capability for a single presigned PutObject request. pub(crate) const RUSTFS_MAX_CONTENT_LENGTH_QUERY: &str = "x-rustfs-max-content-length"; +pub(crate) const RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY: &str = "x-rustfs-max-total-object-size"; /// Inserted by the S3 access boundary after the upstream verifier accepts a /// request as SigV4 presigned. Downstream capability parsing must require this @@ -60,6 +61,9 @@ pub(crate) const RUSTFS_MAX_CONTENT_LENGTH_QUERY: &str = "x-rustfs-max-content-l #[derive(Debug, Clone, Copy)] pub(crate) struct VerifiedPresignedRequest; +#[derive(Debug, Clone, Copy)] +pub(crate) struct VerifiedSigV4Request; + /// Performs constant-time string comparison to prevent timing attacks. /// /// This function should be used when comparing sensitive values like passwords, @@ -1111,6 +1115,92 @@ pub(crate) fn reject_presigned_put_max_content_length_for_other_operation( Ok(()) } +/// Parse the V2 multipart total-size capability after SigV4 authentication. +/// Header-authenticated CreateMultipartUpload requests are accepted because the +/// custom query is covered by the SigV4 canonical request; later multipart +/// operations read the immutable value from the upload session metadata. +pub(crate) fn parse_presigned_multipart_max_total_object_size( + header: &HeaderMap, + query: Option<&str>, + verified_sigv4: bool, +) -> S3Result> { + let Some(query) = query else { + return Ok(None); + }; + + let mut value = None; + let mut decoded_query = Vec::new(); + for (name, candidate) in form_urlencoded::parse(query.as_bytes()) { + decoded_query.push((name.to_string(), candidate.to_string())); + if name == RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY { + if value.is_some() { + return Err(S3Error::with_message( + S3ErrorCode::InvalidRequest, + format!("{RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY} must appear exactly once"), + )); + } + value = Some(candidate.into_owned()); + } else if name.eq_ignore_ascii_case(RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY) { + return Err(S3Error::with_message( + S3ErrorCode::InvalidRequest, + format!("query parameter name must be exactly {RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY}"), + )); + } + } + + let Some(value) = value else { + return Ok(None); + }; + + let auth_type = get_request_auth_type_with_query(header, Some(query)); + let is_presigned = matches!(auth_type, AuthType::Presigned); + let is_header_signed = matches!(auth_type, AuthType::Signed); + let complete_presigned_query = [ + ("x-amz-algorithm", "AWS4-HMAC-SHA256"), + ("x-amz-date", ""), + ("x-amz-expires", ""), + ("x-amz-signedheaders", ""), + ("x-amz-credential", ""), + ("x-amz-signature", ""), + ] + .into_iter() + .all(|(name, expected)| { + decoded_query + .iter() + .find(|(candidate, _)| candidate.eq_ignore_ascii_case(name)) + .is_some_and(|(_, candidate)| !candidate.is_empty() && (expected.is_empty() || candidate == expected)) + }); + + let authenticated = verified_sigv4 && (is_header_signed || (is_presigned && complete_presigned_query)); + if !authenticated { + return Err(S3Error::with_message( + S3ErrorCode::InvalidRequest, + format!("{RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY} requires a verified SigV4 request"), + )); + } + + value.parse::().map(Some).map_err(|_| { + S3Error::with_message( + S3ErrorCode::InvalidRequest, + format!("{RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY} must be a non-negative 64-bit integer"), + ) + }) +} + +pub(crate) fn reject_presigned_multipart_max_total_object_size_for_other_operation( + header: &HeaderMap, + query: Option<&str>, + verified_sigv4: bool, +) -> S3Result<()> { + if parse_presigned_multipart_max_total_object_size(header, query, verified_sigv4)?.is_some() { + return Err(S3Error::with_message( + S3ErrorCode::InvalidRequest, + format!("{RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY} is only supported for CreateMultipartUpload"), + )); + } + Ok(()) +} + #[cfg(test)] mod tests { use super::*; @@ -1816,6 +1906,52 @@ mod tests { } } + #[test] + fn multipart_max_total_object_size_requires_signed_create_request() { + let headers = HeaderMap::new(); + let signed_prefix = "X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Date=20260827T000000Z&X-Amz-Expires=900&X-Amz-SignedHeaders=host&X-Amz-Credential=test/20260827/us-east-1/s3/aws4_request&X-Amz-Signature=signature"; + let query = format!("{signed_prefix}&x-rustfs-max-total-object-size=104857600"); + + assert_eq!( + parse_presigned_multipart_max_total_object_size(&headers, Some(&query), true).unwrap(), + Some(104_857_600) + ); + assert_eq!( + reject_presigned_multipart_max_total_object_size_for_other_operation(&headers, Some(&query), true) + .unwrap_err() + .code(), + &S3ErrorCode::InvalidRequest + ); + } + + #[test] + fn multipart_max_total_object_size_rejects_tampering_and_invalid_values() { + let headers = HeaderMap::new(); + let signed_prefix = "X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Date=20260827T000000Z&X-Amz-Expires=900&X-Amz-SignedHeaders=host&X-Amz-Credential=test/20260827/us-east-1/s3/aws4_request&X-Amz-Signature=signature"; + for query in [ + "x-rustfs-max-total-object-size=1", + "X-RustFS-Max-Total-Object-Size=1", + "X-Amz-Algorithm=AWS4-HMAC-SHA256&x-rustfs-max-total-object-size=1&x-rustfs-max-total-object-size=2", + "X-Amz-Algorithm=AWS4-HMAC-SHA256&x-rustfs-max-total-object-size=-1", + "X-Amz-Algorithm=AWS4-HMAC-SHA256&x-rustfs-max-total-object-size=18446744073709551616", + ] { + assert_eq!( + parse_presigned_multipart_max_total_object_size(&headers, Some(query), true) + .unwrap_err() + .code(), + &S3ErrorCode::InvalidRequest + ); + } + + let forged = format!("{signed_prefix}&x-rustfs-max-total-object-size=1"); + assert_eq!( + parse_presigned_multipart_max_total_object_size(&headers, Some(&forged), false) + .unwrap_err() + .code(), + &S3ErrorCode::InvalidRequest + ); + } + #[test] fn test_credentials_is_expired() { let mut cred = create_test_credentials(); diff --git a/rustfs/src/error.rs b/rustfs/src/error.rs index 6d8f179cf..d72026ea8 100644 --- a/rustfs/src/error.rs +++ b/rustfs/src/error.rs @@ -372,6 +372,7 @@ impl From for ApiError { StorageError::ObjectExistsAsDirectory(_, _) => S3ErrorCode::InvalidArgument, StorageError::InvalidPart(_, _, _) => S3ErrorCode::InvalidPart, StorageError::EntityTooSmall(_, _, _) => S3ErrorCode::EntityTooSmall, + StorageError::EntityTooLarge(_, _) => S3ErrorCode::EntityTooLarge, StorageError::PreconditionFailed => S3ErrorCode::PreconditionFailed, StorageError::NotModified => S3ErrorCode::NotModified, StorageError::InvalidRangeSpec(_) => S3ErrorCode::InvalidRange, diff --git a/rustfs/src/storage/access.rs b/rustfs/src/storage/access.rs index bbee97eca..42775eabc 100644 --- a/rustfs/src/storage/access.rs +++ b/rustfs/src/storage/access.rs @@ -16,9 +16,10 @@ use super::ObjectOptions; use super::ecfs::FS; use super::{ECStore, PolicySys, ReplicationStatusType, StorageError, get_lock_acquire_timeout, is_err_bucket_not_found}; use crate::auth::{ - AuthType, RUSTFS_MAX_CONTENT_LENGTH_QUERY, VerifiedPresignedRequest, check_key_valid_with_context, - get_condition_values_with_client_info, get_condition_values_with_query_and_client_info, get_request_auth_type_with_query, - get_session_token, parse_presigned_put_max_content_length, + AuthType, RUSTFS_MAX_CONTENT_LENGTH_QUERY, RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY, VerifiedPresignedRequest, + VerifiedSigV4Request, check_key_valid_with_context, get_condition_values_with_client_info, + get_condition_values_with_query_and_client_info, get_request_auth_type_with_query, get_session_token, + parse_presigned_multipart_max_total_object_size, parse_presigned_put_max_content_length, }; use crate::error::ApiError; use crate::license::license_check; @@ -1771,7 +1772,9 @@ impl S3Access for FS { // Publish this server's context slot so downstream data-plane handlers // resolve the same store (backlog#1052 S6). - let verified_presigned = matches!(get_request_auth_type_with_query(cx.headers(), cx.uri().query()), AuthType::Presigned); + let auth_type = get_request_auth_type_with_query(cx.headers(), cx.uri().query()); + let verified_presigned = matches!(auth_type, AuthType::Presigned); + let verified_sigv4 = matches!(auth_type, AuthType::Presigned | AuthType::Signed); { let ext = cx.extensions_mut(); ext.insert(self.server_ctx().clone()); @@ -1779,6 +1782,9 @@ impl S3Access for FS { if verified_presigned { ext.insert(VerifiedPresignedRequest); } + if verified_sigv4 { + ext.insert(VerifiedSigV4Request); + } } // The size capability is intentionally scoped to the single-object @@ -1793,6 +1799,14 @@ impl S3Access for FS { format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} is only supported for presigned PutObject"), )); } + if parse_presigned_multipart_max_total_object_size(cx.headers(), cx.uri().query(), verified_sigv4)?.is_some() + && cx.s3_op().name() != "CreateMultipartUpload" + { + return Err(S3Error::with_message( + S3ErrorCode::InvalidRequest, + format!("{RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY} is only supported for CreateMultipartUpload"), + )); + } license_check().map_err(|er| match er.kind() { std::io::ErrorKind::PermissionDenied => s3_error!(AccessDenied, "{er}"), _ => {