fix(object): bound snowball auto extract (#3715)

rustfs/src/app/object_usecase.rs

@@ -117,7 +117,7 @@ use rustfs_utils::http::{
     insert_str, remove_str,
 };
 use rustfs_utils::path::{encode_dir_object, is_dir_object, path_join_buf};
-use rustfs_zip::CompressionFormat;
+use rustfs_zip::{ArchiveLimits, CompressionFormat};
 use s3s::dto::*;
 use s3s::header::{X_AMZ_RESTORE, X_AMZ_RESTORE_OUTPUT_PATH};
 use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
@@ -1256,6 +1256,67 @@ fn resolve_put_object_extract_options(headers: &HeaderMap) -> S3Result<PutObject
     })
 }
 
+fn put_object_extract_limits() -> ArchiveLimits {
+    ArchiveLimits::default()
+}
+
+fn put_object_extract_quota_exceeded(current_usage: u64, quota_limit: u64) -> S3Error {
+    S3Error::with_message(
+        S3ErrorCode::InvalidRequest,
+        format!("Bucket quota exceeded. Current usage: {current_usage} bytes, limit: {quota_limit} bytes"),
+    )
+}
+
+fn validate_put_object_extract_entry_count(count: usize, limits: ArchiveLimits) -> S3Result<()> {
+    if count > limits.max_entries {
+        return Err(s3_error!(
+            InvalidArgument,
+            "Archive entry count exceeds limit: count={}, limit={}",
+            count,
+            limits.max_entries
+        ));
+    }
+    Ok(())
+}
+
+fn validate_put_object_extract_entry_size(path: &str, size: u64, limits: ArchiveLimits) -> S3Result<()> {
+    if size > limits.max_entry_size {
+        return Err(s3_error!(
+            InvalidArgument,
+            "Archive entry size exceeds limit for {}: size={}, limit={}",
+            path,
+            size,
+            limits.max_entry_size
+        ));
+    }
+    Ok(())
+}
+
+fn validate_put_object_extract_total_size(total_size: u64, limits: ArchiveLimits) -> S3Result<()> {
+    if total_size > limits.max_total_unpacked_size {
+        return Err(s3_error!(
+            InvalidArgument,
+            "Archive total unpacked size exceeds limit: size={}, limit={}",
+            total_size,
+            limits.max_total_unpacked_size
+        ));
+    }
+    Ok(())
+}
+
+fn validate_put_object_extract_entry_path(path: &str, limits: ArchiveLimits) -> S3Result<()> {
+    if path.len() > limits.max_path_length {
+        return Err(s3_error!(
+            InvalidArgument,
+            "Archive entry path exceeds limit for {}: length={}, limit={}",
+            path,
+            path.len(),
+            limits.max_path_length
+        ));
+    }
+    Ok(())
+}
+
 fn is_sse_kms_requested(input: &PutObjectInput, headers: &HeaderMap) -> bool {
     input
         .server_side_encryption
@@ -4634,6 +4695,31 @@ impl DefaultObjectUsecase {
         };
 
         let extract_options = resolve_put_object_extract_options(&req.headers)?;
+        let extract_limits = put_object_extract_limits();
+        let extract_quota_snapshot = if let Some(metadata_sys) = self.bucket_metadata_sys() {
+            let quota_checker = QuotaChecker::new(metadata_sys);
+            match quota_checker.get_quota_config(&bucket).await {
+                Ok(quota) => {
+                    if let Some(limit) = quota.quota {
+                        match quota_checker.get_real_time_usage(&bucket).await {
+                            Ok(current_usage) => Some((current_usage, limit)),
+                            Err(err) => {
+                                warn!(bucket, error = %err, state = "extract_usage_snapshot_failed", "Bucket quota snapshot degraded to allow");
+                                None
+                            }
+                        }
+                    } else {
+                        None
+                    }
+                }
+                Err(err) => {
+                    warn!(bucket, error = %err, state = "extract_quota_snapshot_failed", "Bucket quota snapshot degraded to allow");
+                    None
+                }
+            }
+        } else {
+            None
+        };
         let version_id = match event_version_id {
             Some(v) => v.to_string(),
             None => String::new(),
@@ -4649,6 +4735,8 @@ impl DefaultObjectUsecase {
         let port = get_request_port(&req.headers);
         let user_agent = get_request_user_agent(&req.headers);
         let mut wrote_any_entry = false;
+        let mut extracted_entry_count = 0usize;
+        let mut total_unpacked_size = 0u64;
 
         while let Some(entry) = entries.next().await {
             let mut f = match entry {
@@ -4662,6 +4750,8 @@ impl DefaultObjectUsecase {
                     return Err(s3_error!(InvalidArgument, "Failed to read archive entry: {:?}", e));
                 }
             };
+            extracted_entry_count = extracted_entry_count.saturating_add(1);
+            validate_put_object_extract_entry_count(extracted_entry_count, extract_limits)?;
 
             let fpath = match f.path() {
                 Ok(path) => path,
@@ -4685,6 +4775,7 @@ impl DefaultObjectUsecase {
                     return Err(err);
                 }
             };
+            validate_put_object_extract_entry_path(&fpath, extract_limits)?;
             validate_table_catalog_object_mutation(&bucket, &fpath).await?;
 
             let mut auth_req = S3Request {
@@ -4706,7 +4797,19 @@ impl DefaultObjectUsecase {
             }
             authorize_request(&mut auth_req, Action::S3Action(S3Action::PutObjectAction)).await?;
 
-            let mut size = f.header().size().unwrap_or_default() as i64;
+            let entry_size = f.header().size().unwrap_or_default();
+            validate_put_object_extract_entry_size(&fpath, entry_size, extract_limits)?;
+            total_unpacked_size = total_unpacked_size
+                .checked_add(entry_size)
+                .ok_or_else(|| s3_error!(InvalidArgument, "Archive total unpacked size overflowed while processing entries"))?;
+            validate_put_object_extract_total_size(total_unpacked_size, extract_limits)?;
+            if let Some((current_usage, quota_limit)) = extract_quota_snapshot
+                && current_usage.saturating_add(total_unpacked_size) > quota_limit
+            {
+                return Err(put_object_extract_quota_exceeded(current_usage, quota_limit));
+            }
+            let mut size =
+                i64::try_from(entry_size).map_err(|_| s3_error!(InvalidArgument, "Archive entry size does not fit into i64"))?;
             let archive_entry_mod_time = f
                 .header()
                 .mtime()
@@ -5534,6 +5637,57 @@ mod tests {
         assert!(resolve_put_object_extract_options(&headers).is_err());
     }
 
+    #[test]
+    fn validate_put_object_extract_entry_count_rejects_limit_overflow() {
+        let limits = ArchiveLimits {
+            max_entries: 1,
+            ..ArchiveLimits::default()
+        };
+
+        let err = validate_put_object_extract_entry_count(2, limits).unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+    }
+
+    #[test]
+    fn validate_put_object_extract_entry_size_rejects_oversized_entry() {
+        let limits = ArchiveLimits {
+            max_entry_size: 8,
+            ..ArchiveLimits::default()
+        };
+
+        let err = validate_put_object_extract_entry_size("payload.bin", 9, limits).unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+    }
+
+    #[test]
+    fn validate_put_object_extract_total_size_rejects_cumulative_overflow() {
+        let limits = ArchiveLimits {
+            max_total_unpacked_size: 16,
+            ..ArchiveLimits::default()
+        };
+
+        let err = validate_put_object_extract_total_size(17, limits).unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+    }
+
+    #[test]
+    fn validate_put_object_extract_entry_path_rejects_overlong_path() {
+        let limits = ArchiveLimits {
+            max_path_length: 8,
+            ..ArchiveLimits::default()
+        };
+
+        let err = validate_put_object_extract_entry_path("toolong-path", limits).unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+    }
+
+    #[test]
+    fn put_object_extract_quota_exceeded_matches_existing_error_shape() {
+        let err = put_object_extract_quota_exceeded(10, 8);
+        assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
+        assert_eq!(err.message(), Some("Bucket quota exceeded. Current usage: 10 bytes, limit: 8 bytes"));
+    }
+
     #[tokio::test]
     async fn execute_put_object_rejects_post_object_sse_kms_from_input() {
         let input = PutObjectInput::builder()