diff --git a/Cargo.lock b/Cargo.lock index 8ac895b6c..7c209eb87 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -645,9 +645,9 @@ checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" [[package]] name = "aws-config" -version = "1.8.12" +version = "1.8.13" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "96571e6996817bf3d58f6b569e4b9fd2e9d2fcf9f7424eed07b2ce9bb87535e5" +checksum = "c456581cb3c77fafcc8c67204a70680d40b61112d6da78c77bd31d945b65f1b5" dependencies = [ "aws-credential-types", "aws-runtime", @@ -655,7 +655,7 @@ dependencies = [ "aws-sdk-ssooidc", "aws-sdk-sts", "aws-smithy-async", - "aws-smithy-http 0.62.6", + "aws-smithy-http", "aws-smithy-json", "aws-smithy-runtime", "aws-smithy-runtime-api", @@ -710,15 +710,15 @@ dependencies = [ [[package]] name = "aws-runtime" -version = "1.5.18" +version = "1.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "959dab27ce613e6c9658eb3621064d0e2027e5f2acb65bc526a43577facea557" +checksum = "c635c2dc792cb4a11ce1a4f392a925340d1bdf499289b5ec1ec6810954eb43f5" dependencies = [ "aws-credential-types", "aws-sigv4", "aws-smithy-async", "aws-smithy-eventstream", - "aws-smithy-http 0.62.6", + "aws-smithy-http", "aws-smithy-runtime", "aws-smithy-runtime-api", "aws-smithy-types", @@ -726,7 +726,9 @@ dependencies = [ "bytes", "fastrand", "http 0.2.12", + "http 1.4.0", "http-body 0.4.6", + "http-body 1.0.1", "percent-encoding", "pin-project-lite", "tracing", @@ -735,9 +737,9 @@ dependencies = [ [[package]] name = "aws-sdk-s3" -version = "1.121.0" +version = "1.122.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "61948728b681f88a1e49b9500469cf9e36575a424e745e2c5a651a42386e7d9c" +checksum = "94c2ca0cba97e8e279eb6c0b2d0aa10db5959000e602ab2b7c02de6b85d4c19b" dependencies = [ "aws-credential-types", "aws-runtime", @@ -745,7 +747,7 @@ dependencies = [ "aws-smithy-async", "aws-smithy-checksums", "aws-smithy-eventstream", - "aws-smithy-http 0.62.6", + "aws-smithy-http", "aws-smithy-json", "aws-smithy-observability", "aws-smithy-runtime", @@ -759,7 +761,7 @@ dependencies = [ "hmac 0.12.1", "http 0.2.12", "http 1.4.0", - "http-body 0.4.6", + "http-body 1.0.1", "lru", "percent-encoding", "regex-lite", @@ -770,14 +772,14 @@ dependencies = [ [[package]] name = "aws-sdk-sso" -version = "1.92.0" +version = "1.93.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b7d63bd2bdeeb49aa3f9b00c15e18583503b778b2e792fc06284d54e7d5b6566" +checksum = "9dcb38bb33fc0a11f1ffc3e3e85669e0a11a37690b86f77e75306d8f369146a0" dependencies = [ "aws-credential-types", "aws-runtime", "aws-smithy-async", - "aws-smithy-http 0.62.6", + "aws-smithy-http", "aws-smithy-json", "aws-smithy-observability", "aws-smithy-runtime", @@ -787,20 +789,21 @@ dependencies = [ "bytes", "fastrand", "http 0.2.12", + "http 1.4.0", "regex-lite", "tracing", ] [[package]] name = "aws-sdk-ssooidc" -version = "1.94.0" +version = "1.95.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "532d93574bf731f311bafb761366f9ece345a0416dbcc273d81d6d1a1205239b" +checksum = "2ada8ffbea7bd1be1f53df1dadb0f8fdb04badb13185b3321b929d1ee3caad09" dependencies = [ "aws-credential-types", "aws-runtime", "aws-smithy-async", - "aws-smithy-http 0.62.6", + "aws-smithy-http", "aws-smithy-json", "aws-smithy-observability", "aws-smithy-runtime", @@ -810,20 +813,21 @@ dependencies = [ "bytes", "fastrand", "http 0.2.12", + "http 1.4.0", "regex-lite", "tracing", ] [[package]] name = "aws-sdk-sts" -version = "1.96.0" +version = "1.97.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "357e9a029c7524db6a0099cd77fbd5da165540339e7296cca603531bc783b56c" +checksum = "e6443ccadc777095d5ed13e21f5c364878c9f5bad4e35187a6cdbd863b0afcad" dependencies = [ "aws-credential-types", "aws-runtime", "aws-smithy-async", - "aws-smithy-http 0.62.6", + "aws-smithy-http", "aws-smithy-json", "aws-smithy-observability", "aws-smithy-query", @@ -834,19 +838,20 @@ dependencies = [ "aws-types", "fastrand", "http 0.2.12", + "http 1.4.0", "regex-lite", "tracing", ] [[package]] name = "aws-sigv4" -version = "1.3.7" +version = "1.3.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "69e523e1c4e8e7e8ff219d732988e22bfeae8a1cafdbe6d9eca1546fa080be7c" +checksum = "efa49f3c607b92daae0c078d48a4571f599f966dce3caee5f1ea55c4d9073f99" dependencies = [ "aws-credential-types", "aws-smithy-eventstream", - "aws-smithy-http 0.62.6", + "aws-smithy-http", "aws-smithy-runtime-api", "aws-smithy-types", "bytes", @@ -879,17 +884,18 @@ dependencies = [ [[package]] name = "aws-smithy-checksums" -version = "0.63.13" +version = "0.64.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "23374b9170cbbcc6f5df8dc5ebb9b6c5c28a3c8f599f0e8b8b10eb6f4a5c6e74" +checksum = "ddcf418858f9f3edd228acb8759d77394fed7531cce78d02bdda499025368439" dependencies = [ - "aws-smithy-http 0.62.6", + "aws-smithy-http", "aws-smithy-types", "bytes", "crc-fast", "hex", - "http 0.2.12", - "http-body 0.4.6", + "http 1.4.0", + "http-body 1.0.1", + "http-body-util", "md-5 0.10.6", "pin-project-lite", "sha1 0.10.6", @@ -908,34 +914,13 @@ dependencies = [ "crc32fast", ] -[[package]] -name = "aws-smithy-http" -version = "0.62.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "826141069295752372f8203c17f28e30c464d22899a43a0c9fd9c458d469c88b" -dependencies = [ - "aws-smithy-eventstream", - "aws-smithy-runtime-api", - "aws-smithy-types", - "bytes", - "bytes-utils", - "futures-core", - "futures-util", - "http 0.2.12", - "http 1.4.0", - "http-body 0.4.6", - "percent-encoding", - "pin-project-lite", - "pin-utils", - "tracing", -] - [[package]] name = "aws-smithy-http" version = "0.63.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "630e67f2a31094ffa51b210ae030855cb8f3b7ee1329bdd8d085aaf61e8b97fc" dependencies = [ + "aws-smithy-eventstream", "aws-smithy-runtime-api", "aws-smithy-types", "bytes", @@ -977,9 +962,9 @@ dependencies = [ [[package]] name = "aws-smithy-json" -version = "0.61.9" +version = "0.62.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "49fa1213db31ac95288d981476f78d05d9cbb0353d22cdf3472cc05bb02f6551" +checksum = "3cb96aa208d62ee94104645f7b2ecaf77bf27edf161590b6224bfbac2832f979" dependencies = [ "aws-smithy-types", ] @@ -1010,7 +995,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f3df87c14f0127a0d77eb261c3bc45d5b4833e2a1f63583ebfb728e4852134ee" dependencies = [ "aws-smithy-async", - "aws-smithy-http 0.63.3", + "aws-smithy-http", "aws-smithy-http-client", "aws-smithy-observability", "aws-smithy-runtime-api", @@ -1605,9 +1590,9 @@ dependencies = [ [[package]] name = "clap" -version = "4.5.56" +version = "4.5.57" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a75ca66430e33a14957acc24c5077b503e7d374151b2b4b3a10c83b4ceb4be0e" +checksum = "6899ea499e3fb9305a65d5ebf6e3d2248c5fab291f300ad0a704fbe142eae31a" dependencies = [ "clap_builder", "clap_derive", @@ -1615,9 +1600,9 @@ dependencies = [ [[package]] name = "clap_builder" -version = "4.5.56" +version = "4.5.57" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "793207c7fa6300a0608d1080b858e5fdbe713cdc1c8db9fb17777d8a13e63df0" +checksum = "7b12c8b680195a62a8364d16b8447b01b6c2c8f9aaf68bee653be34d4245e238" dependencies = [ "anstream", "anstyle", @@ -7629,10 +7614,13 @@ name = "rustfs" version = "0.0.5" dependencies = [ "aes-gcm 0.11.0-rc.2", + "anyhow", "astral-tokio-tar", "async-trait", "atoi", "atomic_enum", + "aws-config", + "aws-sdk-s3", "axum", "axum-server", "backtrace", @@ -8566,7 +8554,7 @@ checksum = "a50f4cf475b65d88e057964e0e9bb1f0aa9bbb2036dc65c64596b42932536984" [[package]] name = "s3s" version = "0.13.0-alpha.2" -source = "git+https://github.com/rustfs/s3s.git?branch=dependabot%2Fcargo%2Fdep-0202#13f968fc89e3bd286d308ca855456b23734d4f74" +source = "git+https://github.com/rustfs/s3s.git?rev=13f968fc89e3bd286d308ca855456b23734d4f74#13f968fc89e3bd286d308ca855456b23734d4f74" dependencies = [ "arc-swap", "arrayvec", diff --git a/Cargo.toml b/Cargo.toml index 95ea9c8f6..357c8b31e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -174,16 +174,16 @@ arc-swap = "1.8.0" astral-tokio-tar = "0.5.6" atoi = "2.0.0" atomic_enum = "0.3.0" -aws-config = { version = "1.8.12" } +aws-config = { version = "1.8.13" } aws-credential-types = { version = "1.2.11" } -aws-sdk-s3 = { version = "1.121.0", default-features = false, features = ["sigv4a", "default-https-client", "rt-tokio"] } +aws-sdk-s3 = { version = "1.122.0", default-features = false, features = ["sigv4a", "default-https-client", "rt-tokio"] } aws-smithy-types = { version = "1.4.3" } backtrace = "0.3.76" base64 = "0.22.1" base64-simd = "0.8.0" brotli = "8.0.2" cfg-if = "1.0.4" -clap = { version = "4.5.56", features = ["derive", "env"] } +clap = { version = "4.5.57", features = ["derive", "env"] } const-str = { version = "1.1.0", features = ["std", "proc"] } convert_case = "0.11.0" criterion = { version = "0.8", features = ["html_reports"] } @@ -229,7 +229,7 @@ rumqttc = { version = "0.25.1" } rustix = { version = "1.1.3", features = ["fs"] } rust-embed = { version = "8.11.0" } rustc-hash = { version = "2.1.1" } -s3s = { version = "0.13.0-alpha.2", features = ["minio"], git = "https://github.com/rustfs/s3s.git", branch = "dependabot/cargo/dep-0202" } +s3s = { version = "0.13.0-alpha.2", features = ["minio"], git = "https://github.com/rustfs/s3s.git", rev = "13f968fc89e3bd286d308ca855456b23734d4f74" } serial_test = "3.3.1" shadow-rs = { version = "1.7.0", default-features = false } siphasher = "1.0.2" diff --git a/crates/ecstore/src/bucket/quota/checker.rs b/crates/ecstore/src/bucket/quota/checker.rs index 2a7f28588..36ad8782c 100644 --- a/crates/ecstore/src/bucket/quota/checker.rs +++ b/crates/ecstore/src/bucket/quota/checker.rs @@ -58,12 +58,12 @@ impl QuotaChecker { let current_usage = self.get_real_time_usage(bucket).await?; let expected_usage = match operation { - QuotaOperation::PutObject | QuotaOperation::CopyObject => current_usage + operation_size, + QuotaOperation::PutObject | QuotaOperation::PostObject | QuotaOperation::CopyObject => current_usage + operation_size, QuotaOperation::DeleteObject => current_usage.saturating_sub(operation_size), }; let allowed = match operation { - QuotaOperation::PutObject | QuotaOperation::CopyObject => { + QuotaOperation::PutObject | QuotaOperation::PostObject | QuotaOperation::CopyObject => { quota_config.check_operation_allowed(current_usage, operation_size) } QuotaOperation::DeleteObject => true, diff --git a/crates/ecstore/src/bucket/quota/mod.rs b/crates/ecstore/src/bucket/quota/mod.rs index 3bf00a05f..7c04bd5d0 100644 --- a/crates/ecstore/src/bucket/quota/mod.rs +++ b/crates/ecstore/src/bucket/quota/mod.rs @@ -90,6 +90,7 @@ pub struct QuotaCheckResult { #[derive(Debug)] pub enum QuotaOperation { PutObject, + PostObject, CopyObject, DeleteObject, } diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index 18761d631..113dc2c57 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -161,6 +161,10 @@ pprof = { workspace = true } uuid = { workspace = true, features = ["v4"] } serial_test = { workspace = true } tempfile = { workspace = true } +aws-sdk-s3 = { workspace = true } +aws-config = { workspace = true } +anyhow = { workspace = true } +tokio = { workspace = true, features = ["test-util"] } [build-dependencies] http.workspace = true diff --git a/rustfs/src/admin/handlers/quota.rs b/rustfs/src/admin/handlers/quota.rs index 84944fa82..ede7e6563 100644 --- a/rustfs/src/admin/handlers/quota.rs +++ b/rustfs/src/admin/handlers/quota.rs @@ -415,6 +415,7 @@ impl Operation for CheckBucketQuotaHandler { let operation: QuotaOperation = match request.operation_type.to_uppercase().as_str() { "PUT" | "PUTOBJECT" => QuotaOperation::PutObject, + "POST" | "POSTOBJECT" => QuotaOperation::PostObject, "COPY" | "COPYOBJECT" => QuotaOperation::CopyObject, "DELETE" | "DELETEOBJECT" => QuotaOperation::DeleteObject, _ => QuotaOperation::PutObject, // Default to PUT operation @@ -455,6 +456,7 @@ mod tests { fn test_quota_operation_parsing() { let parse_operation = |operation: &str| match operation.to_uppercase().as_str() { "PUT" | "PUTOBJECT" => QuotaOperation::PutObject, + "POST" | "POSTOBJECT" => QuotaOperation::PostObject, "COPY" | "COPYOBJECT" => QuotaOperation::CopyObject, "DELETE" | "DELETEOBJECT" => QuotaOperation::DeleteObject, _ => QuotaOperation::PutObject, @@ -463,6 +465,9 @@ mod tests { assert!(matches!(parse_operation("put"), QuotaOperation::PutObject)); assert!(matches!(parse_operation("PUT"), QuotaOperation::PutObject)); assert!(matches!(parse_operation("PutObject"), QuotaOperation::PutObject)); + assert!(matches!(parse_operation("post"), QuotaOperation::PostObject)); + assert!(matches!(parse_operation("POST"), QuotaOperation::PostObject)); + assert!(matches!(parse_operation("PostObject"), QuotaOperation::PostObject)); assert!(matches!(parse_operation("copy"), QuotaOperation::CopyObject)); assert!(matches!(parse_operation("DELETE"), QuotaOperation::DeleteObject)); assert!(matches!(parse_operation("unknown"), QuotaOperation::PutObject)); diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index 1acb436f6..8596a0d94 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -30,15 +30,15 @@ use crate::storage::{ access::{ReqInfo, authorize_request, has_bypass_governance_header}, ecfs_extend::RFC1123, options::{ - copy_dst_opts, copy_src_opts, del_opts, extract_metadata, extract_metadata_from_mime_with_object_name, - get_complete_multipart_upload_opts, get_opts, parse_copy_source_range, put_opts, + copy_dst_opts, copy_src_opts, del_opts, extract_metadata, get_complete_multipart_upload_opts, get_opts, + parse_copy_source_range, put_opts, }, }; use crate::storage::{ - apply_lock_retention, check_preconditions, get_buffer_size_opt_in, get_validated_store, has_replication_rules, - parse_object_lock_legal_hold, parse_object_lock_retention, process_lambda_configurations, process_queue_configurations, - process_topic_configurations, validate_bucket_object_lock_enabled, validate_list_object_unordered_with_delimiter, - validate_object_key, wrap_response_with_cors, + check_preconditions, get_buffer_size_opt_in, get_validated_store, has_replication_rules, parse_object_lock_legal_hold, + parse_object_lock_retention, process_lambda_configurations, process_queue_configurations, process_topic_configurations, + validate_bucket_object_lock_enabled, validate_list_object_unordered_with_delimiter, validate_object_key, + wrap_response_with_cors, }; use crate::storage::{entity, parse_part_number_i32_to_usize}; // base64 imports moved to sse module @@ -226,6 +226,8 @@ impl FS { Self {} } + #[instrument(level = "debug", skip(self, req))] + #[allow(dead_code)] async fn put_object_extract(&self, req: S3Request) -> S3Result> { let helper = OperationHelper::new(&req, EventName::ObjectCreatedPut, "s3:PutObject").suppress_event(); let input = req.input; @@ -4312,396 +4314,7 @@ impl S3 for FS { #[instrument(level = "debug", skip(self, req))] async fn put_object(&self, req: S3Request) -> S3Result> { - let mut helper = OperationHelper::new(&req, EventName::ObjectCreatedPut, "s3:PutObject"); - if req - .headers - .get("X-Amz-Meta-Snowball-Auto-Extract") - .is_some_and(|v| v.to_str().unwrap_or_default() == "true") - { - return self.put_object_extract(req).await; - } - - let input = req.input; - - // Save SSE-C parameters before moving input - if let Some(ref storage_class) = input.storage_class - && !is_valid_storage_class(storage_class.as_str()) - { - return Err(s3_error!(InvalidStorageClass)); - } - let PutObjectInput { - body, - bucket, - key, - content_length, - content_type, - tagging, - metadata, - version_id, - server_side_encryption, - sse_customer_algorithm, - sse_customer_key, - sse_customer_key_md5, - ssekms_key_id, - content_md5, - if_match, - if_none_match, - .. - } = input; - - // Validate object key - validate_object_key(&key, "PUT")?; - - if if_match.is_some() || if_none_match.is_some() { - let Some(store) = new_object_layer_fn() else { - return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); - }; - - match store.get_object_info(&bucket, &key, &ObjectOptions::default()).await { - Ok(info) => { - if !info.delete_marker { - if let Some(ifmatch) = if_match - && let Some(strong_etag) = ifmatch.into_etag() - && info - .etag - .as_ref() - .is_some_and(|etag| ETag::Strong(etag.clone()) != strong_etag) - { - return Err(s3_error!(PreconditionFailed)); - } - if let Some(ifnonematch) = if_none_match - && let Some(strong_etag) = ifnonematch.into_etag() - && info - .etag - .as_ref() - .is_some_and(|etag| ETag::Strong(etag.clone()) == strong_etag) - { - return Err(s3_error!(PreconditionFailed)); - } - } - } - Err(err) => { - if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) { - return Err(ApiError::from(err).into()); - } - - if if_match.is_some() && (is_err_object_not_found(&err) || is_err_version_not_found(&err)) { - return Err(ApiError::from(err).into()); - } - } - } - } - - // check quota for put operation - if let Some(size) = content_length - && let Some(metadata_sys) = rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys.get() - { - let quota_checker = QuotaChecker::new(metadata_sys.clone()); - - match quota_checker - .check_quota(&bucket, QuotaOperation::PutObject, size as u64) - .await - { - Ok(check_result) => { - if !check_result.allowed { - return Err(S3Error::with_message( - S3ErrorCode::InvalidRequest, - format!( - "Bucket quota exceeded. Current usage: {} bytes, limit: {} bytes", - check_result.current_usage, - check_result.quota_limit.unwrap_or(0) - ), - )); - } - } - Err(e) => { - warn!("Quota check failed for bucket {}: {}, allowing operation", bucket, e); - } - } - } - - let Some(body) = body else { return Err(s3_error!(IncompleteBody)) }; - - let mut size = match content_length { - Some(c) => c, - None => { - if let Some(val) = req.headers.get(AMZ_DECODED_CONTENT_LENGTH) { - match atoi::atoi::(val.as_bytes()) { - Some(x) => x, - None => return Err(s3_error!(UnexpectedContent)), - } - } else { - return Err(s3_error!(UnexpectedContent)); - } - } - }; - - if size == -1 { - return Err(s3_error!(UnexpectedContent)); - } - - // Apply adaptive buffer sizing based on file size for optimal streaming performance. - // Uses workload profile configuration (enabled by default) to select appropriate buffer size. - // Buffer sizes range from 32KB to 4MB depending on file size and configured workload profile. - let buffer_size = get_buffer_size_opt_in(size); - let body = tokio::io::BufReader::with_capacity( - buffer_size, - StreamReader::new(body.map(|f| f.map_err(|e| std::io::Error::other(e.to_string())))), - ); - - // let body = Box::new(StreamReader::new(body.map(|f| f.map_err(|e| std::io::Error::other(e.to_string()))))); - - // let mut reader = PutObjReader::new(body, content_length as usize); - - let store = get_validated_store(&bucket).await?; - - let mut metadata = metadata.unwrap_or_default(); - - let object_lock_configuration = match metadata_sys::get_object_lock_config(&bucket).await { - Ok((cfg, _created)) => Some(cfg), - Err(err) => { - if err == StorageError::ConfigNotFound { - None - } else { - warn!("get_object_lock_config err {:?}", err); - return Err(S3Error::with_message( - S3ErrorCode::InternalError, - "Failed to load Object Lock configuration".to_string(), - )); - } - } - }; - - apply_lock_retention(object_lock_configuration, &mut metadata); - - if let Some(content_type) = content_type { - metadata.insert("content-type".to_string(), content_type.to_string()); - } - - extract_metadata_from_mime_with_object_name(&req.headers, &mut metadata, true, Some(&key)); - - if let Some(tags) = tagging { - metadata.insert(AMZ_OBJECT_TAGGING.to_owned(), tags.to_string()); - } - - let mut opts: ObjectOptions = put_opts(&bucket, &key, version_id.clone(), &req.headers, metadata.clone()) - .await - .map_err(ApiError::from)?; - - let mut reader: Box = Box::new(WarpReader::new(body)); - - let actual_size = size; - - let mut md5hex = if let Some(base64_md5) = content_md5 { - let md5 = base64_simd::STANDARD - .decode_to_vec(base64_md5.as_bytes()) - .map_err(|e| ApiError::from(StorageError::other(format!("Invalid content MD5: {e}"))))?; - Some(hex_simd::encode_to_string(&md5, hex_simd::AsciiCase::Lower)) - } else { - None - }; - - let mut sha256hex = get_content_sha256(&req.headers); - - if is_compressible(&req.headers, &key) && size > MIN_COMPRESSIBLE_SIZE as i64 { - let algorithm = CompressionAlgorithm::default(); - metadata.insert(format!("{RESERVED_METADATA_PREFIX_LOWER}compression"), algorithm.to_string()); - - metadata.insert(format!("{RESERVED_METADATA_PREFIX_LOWER}actual-size",), size.to_string()); - - let mut hrd = HashReader::new(reader, size as i64, size as i64, md5hex, sha256hex, false).map_err(ApiError::from)?; - - if let Err(err) = hrd.add_checksum_from_s3s(&req.headers, req.trailing_headers.clone(), false) { - return Err(ApiError::from(StorageError::other(format!("add_checksum error={err:?}"))).into()); - } - - opts.want_checksum = hrd.checksum(); - opts.user_defined - .insert(format!("{RESERVED_METADATA_PREFIX_LOWER}compression"), algorithm.to_string()); - opts.user_defined - .insert(format!("{RESERVED_METADATA_PREFIX_LOWER}actual-size",), size.to_string()); - - reader = Box::new(CompressReader::new(hrd, algorithm)); - size = HashReader::SIZE_PRESERVE_LAYER; - md5hex = None; - sha256hex = None; - } - - let mut reader = HashReader::new(reader, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)?; - - if size >= 0 { - if let Err(err) = reader.add_checksum_from_s3s(&req.headers, req.trailing_headers.clone(), false) { - return Err(ApiError::from(StorageError::other(format!("add_checksum error={err:?}"))).into()); - } - - opts.want_checksum = reader.checksum(); - } - - // Apply encryption using unified SSE API - let encryption_request = EncryptionRequest { - bucket: &bucket, - key: &key, - server_side_encryption, - ssekms_key_id, - sse_customer_algorithm: sse_customer_algorithm.clone(), - sse_customer_key, - sse_customer_key_md5: sse_customer_key_md5.clone(), - content_size: actual_size, - part_number: None, - part_key: None, - part_nonce: None, - }; - - let (effective_sse, effective_kms_key_id) = match sse_encryption(encryption_request).await? { - Some(material) => { - let server_side_encryption = Some(material.server_side_encryption.clone()); - let ssekms_key_id = material.kms_key_id.clone(); - - // Apply encryption wrapper - let encrypted_reader = material.wrap_reader(reader); - reader = HashReader::new(encrypted_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false) - .map_err(ApiError::from)?; - - // Merge encryption metadata - metadata.extend(material.metadata); - - (server_side_encryption, ssekms_key_id) - } - None => (None, None), - }; - - let mut reader = PutObjReader::new(reader); - - let mt2 = metadata.clone(); - opts.user_defined.extend(metadata); - - let repoptions = - get_must_replicate_options(&mt2, "".to_string(), ReplicationStatusType::Empty, ReplicationType::Object, opts.clone()); - - let dsc = must_replicate(&bucket, &key, repoptions).await; - - if dsc.replicate_any() { - let k = format!("{}{}", RESERVED_METADATA_PREFIX_LOWER, "replication-timestamp"); - opts.user_defined.insert(k, jiff::Zoned::now().to_string()); - let k = format!("{}{}", RESERVED_METADATA_PREFIX_LOWER, "replication-status"); - opts.user_defined.insert(k, dsc.pending_status().unwrap_or_default()); - } - - let obj_info = store - .put_object(&bucket, &key, &mut reader, &opts) - .await - .map_err(ApiError::from)?; - - // Fast in-memory update for immediate quota consistency - rustfs_ecstore::data_usage::increment_bucket_usage_memory(&bucket, obj_info.size as u64).await; - - // Invalidate cache for the written object to prevent stale data - let manager = get_concurrency_manager(); - let put_bucket = bucket.clone(); - let put_key = key.clone(); - let mut put_version = obj_info.version_id.map(|v| v.to_string()); - if opts.version_suspended && obj_info.version_id.is_none_or(|v| v.is_nil()) { - put_version = Some("null".to_string()); - } - - helper = helper.object(obj_info.clone()); - if let Some(version_id) = &put_version { - helper = helper.version_id(version_id.clone()); - } - - let put_version_clone = put_version.clone(); - tokio::spawn(async move { - manager - .invalidate_cache_versioned(&put_bucket, &put_key, put_version_clone.as_deref()) - .await; - }); - - let e_tag = obj_info.etag.clone().map(|etag| to_s3s_etag(&etag)); - - let repoptions = - get_must_replicate_options(&mt2, "".to_string(), ReplicationStatusType::Empty, ReplicationType::Object, opts); - - let dsc = must_replicate(&bucket, &key, repoptions).await; - - if dsc.replicate_any() { - schedule_replication(obj_info, store, dsc, ReplicationType::Object).await; - } - - let mut checksum_crc32 = input.checksum_crc32; - let mut checksum_crc32c = input.checksum_crc32c; - let mut checksum_sha1 = input.checksum_sha1; - let mut checksum_sha256 = input.checksum_sha256; - let mut checksum_crc64nvme = input.checksum_crc64nvme; - - if let Some(alg) = &input.checksum_algorithm - && let Some(Some(checksum_str)) = req.trailing_headers.as_ref().map(|trailer| { - let key = match alg.as_str() { - ChecksumAlgorithm::CRC32 => rustfs_rio::ChecksumType::CRC32.key(), - ChecksumAlgorithm::CRC32C => rustfs_rio::ChecksumType::CRC32C.key(), - ChecksumAlgorithm::SHA1 => rustfs_rio::ChecksumType::SHA1.key(), - ChecksumAlgorithm::SHA256 => rustfs_rio::ChecksumType::SHA256.key(), - ChecksumAlgorithm::CRC64NVME => rustfs_rio::ChecksumType::CRC64_NVME.key(), - _ => return None, - }; - trailer.read(|headers| { - headers - .get(key.unwrap_or_default()) - .and_then(|value| value.to_str().ok().map(|s| s.to_string())) - }) - }) - { - match alg.as_str() { - ChecksumAlgorithm::CRC32 => checksum_crc32 = checksum_str, - ChecksumAlgorithm::CRC32C => checksum_crc32c = checksum_str, - ChecksumAlgorithm::SHA1 => checksum_sha1 = checksum_str, - ChecksumAlgorithm::SHA256 => checksum_sha256 = checksum_str, - ChecksumAlgorithm::CRC64NVME => checksum_crc64nvme = checksum_str, - _ => (), - } - } - - let output = PutObjectOutput { - e_tag, - server_side_encryption: effective_sse, // TDD: Return effective encryption config - sse_customer_algorithm, - sse_customer_key_md5, - ssekms_key_id: effective_kms_key_id, // TDD: Return effective KMS key ID - checksum_crc32, - checksum_crc32c, - checksum_sha1, - checksum_sha256, - checksum_crc64nvme, - version_id: put_version, - ..Default::default() - }; - - // TODO fix response for POST Policy (multipart/form-data) ，wait s3s crate update,fix issue #1564 - // // If it is a POST Policy(multipart/form-data) path, the PutObjectInput carries the success_action_* field - // // Here, the response is uniformly rewritten, with the default being 204, redirect prioritizing 303, and status supporting 200/201/204 - // if input.success_action_status.is_some() || input.success_action_redirect.is_some() { - // let mut form_fields = HashMap::::new(); - // if let Some(v) = &input.success_action_status { - // form_fields.insert("success_action_status".to_string(), v.to_string()); - // } - // if let Some(v) = &input.success_action_redirect { - // form_fields.insert("success_action_redirect".to_string(), v.to_string()); - // } - // - // // obj_info.etag has been converted to e_tag (s3s etag) above, so try to pass the original string here - // let etag_str = e_tag.as_ref().map(|v| v.as_str()); - // - // // Returns using POST semantics: 204/303/201/200 - // let resp = build_post_object_success_response(&form_fields, &bucket, &key, etag_str, None)?; - // - // // Keep helper event complete (note: (StatusCode, Body) is returned here instead of PutObjectOutput) - // let result = Ok(resp); - // let _ = helper.complete(&result); - // return result; - // } - - let result = Ok(S3Response::new(output)); - let _ = helper.complete(&result); - result + crate::storage::objects::GLOBAL_OBJECTS.put_object(req).await } async fn put_object_acl(&self, req: S3Request) -> S3Result> { diff --git a/rustfs/src/storage/ecfs_extend.rs b/rustfs/src/storage/ecfs_extend.rs index da5bf0ad2..bdb0a7156 100644 --- a/rustfs/src/storage/ecfs_extend.rs +++ b/rustfs/src/storage/ecfs_extend.rs @@ -18,7 +18,6 @@ use crate::config::workload_profiles::{ use crate::error::ApiError; use crate::server::cors; use crate::storage::ecfs::ListObjectUnorderedQuery; -use axum::body::Body; use http::{HeaderMap, HeaderValue, StatusCode}; use metrics::counter; use rustfs_ecstore::bucket::metadata_sys; @@ -91,140 +90,6 @@ pub(crate) fn apply_lock_retention(object_lock_config: Option` - Parsed success action or error -/// -/// Notes: -/// - Follows AWS S3 behavior: `success_action_redirect` takes precedence over `success_action_status`. -/// - Validates `success_action_status` values; invalid values result in MalformedPOSTRequest error. -/// -#[allow(dead_code)] -pub(crate) fn parse_success_action_from_form_fields(fields: &HashMap) -> S3Result { - // 1) success_action_redirect wins over success_action_status (AWS compatible behavior). - if let Some(loc) = fields - .get("success_action_redirect") - .map(|s| s.trim()) - .filter(|s| !s.is_empty()) - { - return Ok(PostObjectSuccessAction::Redirect303 { - location: loc.to_string(), - }); - } - - // 2) success_action_status is optional; default is 204. - let Some(status_str) = fields - .get("success_action_status") - .map(|s| s.trim()) - .filter(|s| !s.is_empty()) - else { - return Ok(PostObjectSuccessAction::NoContent204); - }; - - // AWS allows only 200/201/204 for POST form success_action_status. - // Treat invalid values as MalformedPOSTRequest to match S3 strictness. - match status_str { - "200" => Ok(PostObjectSuccessAction::Ok200), - "201" => Ok(PostObjectSuccessAction::Created201), - "204" => Ok(PostObjectSuccessAction::NoContent204), - _ => Err(S3Error::with_message( - S3ErrorCode::MalformedPOSTRequest, - format!("Invalid success_action_status: {status_str}. Allowed values are 200, 201, 204."), - )), - } -} - -/// Build the final S3Response for a successful Presigned POST upload. -/// -/// Integration point (manual): -/// - After `put_object` succeeds in the PostPolicy handler, call this function -/// with parsed form fields + object info to produce the correct HTTP status. -/// -/// Notes: -/// - For 204: empty body -/// - For 303: empty body + Location header -/// - For 200: empty body (some clients accept this); you may optionally return XML/HTML body if you already implement it. -/// - For 201: prefer returning PostResponse XML; if not available, empty body still satisfies most clients, but strict tests may require XML. If you have a PostResponse serializer already, plug it in here. -#[allow(dead_code)] -pub(crate) fn build_post_object_success_response( - form_fields: &HashMap, - // These are optional; used if you want to return richer responses for 200/201. - bucket: &str, - key: &str, - etag: Option<&str>, - location: Option<&str>, -) -> S3Result> { - let action = parse_success_action_from_form_fields(form_fields)?; - - match action { - PostObjectSuccessAction::NoContent204 => Ok(S3Response::new((StatusCode::NO_CONTENT, Body::empty()))), - - PostObjectSuccessAction::Redirect303 { location } => { - let mut headers = HeaderMap::new(); - headers.insert( - http::header::LOCATION, - HeaderValue::from_str(&location) - .map_err(|_| S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid success_action_redirect URL"))?, - ); - Ok(S3Response::with_headers((StatusCode::SEE_OTHER, Body::empty()), headers)) - } - - PostObjectSuccessAction::Ok200 => { - // AWS may return 200 with an HTML/redirect response for browser workflows. - // For compatibility, returning empty body is acceptable unless strict clients require content. - // Keep Content-Length implicit; Body::empty() -> 0. - Ok(S3Response::new((StatusCode::OK, Body::empty()))) - } - - PostObjectSuccessAction::Created201 => { - // AWS 201 response is XML: - // - // ... - // ... - // ... - // ... - // - // - // If RustFS already has a DTO for this, switch to it here. - // To keep this patch minimal and safe, return empty body with 201 by default. - // - // IMPORTANT: If you run strict s3-tests for POST Object, you may need to implement XML body. - let _ = (bucket, key, etag, location); - Ok(S3Response::new((StatusCode::CREATED, Body::empty()))) - } - } -} - /// Calculate adaptive buffer size with workload profile support. /// /// This enhanced version supports different workload profiles for optimal performance diff --git a/rustfs/src/storage/mod.rs b/rustfs/src/storage/mod.rs index 8739be93b..a15cb31aa 100644 --- a/rustfs/src/storage/mod.rs +++ b/rustfs/src/storage/mod.rs @@ -14,20 +14,19 @@ pub mod access; pub mod concurrency; +#[cfg(test)] +mod concurrent_get_object_test; pub mod ecfs; +mod ecfs_extend; pub(crate) mod entity; pub(crate) mod helper; pub mod options; pub mod tonic_service; - -#[cfg(test)] -mod concurrent_get_object_test; -mod ecfs_extend; +pub(crate) use ecfs_extend::*; #[cfg(test)] mod ecfs_test; pub(crate) mod head_prefix; +mod objects; mod sse; #[cfg(test)] mod sse_test; - -pub(crate) use ecfs_extend::*; diff --git a/rustfs/src/storage/objects/mod.rs b/rustfs/src/storage/objects/mod.rs new file mode 100644 index 000000000..1798ea46a --- /dev/null +++ b/rustfs/src/storage/objects/mod.rs @@ -0,0 +1,27 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::sync::LazyLock; + +mod put_object; + +pub(crate) struct Objects; + +pub(crate) static GLOBAL_OBJECTS: LazyLock = LazyLock::new(Objects::new); + +impl Objects { + pub fn new() -> Self { + Objects + } +} diff --git a/rustfs/src/storage/objects/put_object.rs b/rustfs/src/storage/objects/put_object.rs new file mode 100644 index 000000000..ea910727a --- /dev/null +++ b/rustfs/src/storage/objects/put_object.rs @@ -0,0 +1,666 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use crate::error::ApiError; +use crate::storage::concurrency::get_concurrency_manager; +use crate::storage::helper::OperationHelper; +use crate::storage::objects::Objects; +use crate::storage::options::{extract_metadata_from_mime_with_object_name, get_content_sha256, put_opts}; +use crate::storage::sse::{EncryptionRequest, sse_encryption}; +use crate::storage::{apply_lock_retention, get_buffer_size_opt_in, get_validated_store, validate_object_key}; +use futures_util::StreamExt; +use http::HeaderMap; +use rustfs_ecstore::bucket::metadata_sys; +use rustfs_ecstore::bucket::quota::QuotaOperation; +use rustfs_ecstore::bucket::quota::checker::QuotaChecker; +use rustfs_ecstore::bucket::replication::{get_must_replicate_options, must_replicate, schedule_replication}; +use rustfs_ecstore::client::object_api_utils::to_s3s_etag; +use rustfs_ecstore::compress::{MIN_COMPRESSIBLE_SIZE, is_compressible}; +use rustfs_ecstore::error::{StorageError, is_err_object_not_found, is_err_version_not_found}; +use rustfs_ecstore::set_disk::is_valid_storage_class; +use rustfs_ecstore::store_api::{ObjectIO, ObjectOptions, PutObjReader}; +use rustfs_ecstore::{StorageAPI, new_object_layer_fn}; +use rustfs_filemeta::{ReplicationStatusType, ReplicationType}; +use rustfs_notify::notifier_global; +use rustfs_rio::{CompressReader, HashReader, Reader, WarpReader}; +use rustfs_targets::EventName; +use rustfs_utils::http::{AMZ_DECODED_CONTENT_LENGTH, AMZ_OBJECT_TAGGING, RESERVED_METADATA_PREFIX_LOWER}; +use rustfs_utils::{ + CompressionAlgorithm, extract_params_header, extract_resp_elements, get_request_host, get_request_port, + get_request_user_agent, +}; +use rustfs_zip::CompressionFormat; +use s3s::dto::{ChecksumAlgorithm, ETag, PutObjectInput, PutObjectOutput}; +use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; +use std::collections::HashMap; +use std::path::Path; +use tokio_tar::Archive; +use tokio_util::io::StreamReader; +use tracing::{debug, error, instrument, warn}; + +impl Objects { + #[instrument(level = "debug", skip(self, req))] + pub async fn put_object(&self, req: S3Request) -> S3Result> { + let mut helper = OperationHelper::new(&req, EventName::ObjectCreatedPut, "s3:PutObject"); + if req + .headers + .get("X-Amz-Meta-Snowball-Auto-Extract") + .is_some_and(|v| v.to_str().unwrap_or_default() == "true") + { + return self.put_object_extract(req).await; + } + + let input = req.input; + + // Save SSE-C parameters before moving input + if let Some(ref storage_class) = input.storage_class + && !is_valid_storage_class(storage_class.as_str()) + { + return Err(s3_error!(InvalidStorageClass)); + } + let PutObjectInput { + body, + bucket, + key, + content_length, + content_type, + tagging, + metadata, + version_id, + server_side_encryption, + sse_customer_algorithm, + sse_customer_key, + sse_customer_key_md5, + ssekms_key_id, + content_md5, + if_match, + if_none_match, + .. + } = input; + + // Validate object key + validate_object_key(&key, "PUT")?; + + if if_match.is_some() || if_none_match.is_some() { + let Some(store) = new_object_layer_fn() else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + + match store.get_object_info(&bucket, &key, &ObjectOptions::default()).await { + Ok(info) => { + if !info.delete_marker { + if let Some(ifmatch) = if_match + && let Some(strong_etag) = ifmatch.into_etag() + && info + .etag + .as_ref() + .is_some_and(|etag| ETag::Strong(etag.clone()) != strong_etag) + { + return Err(s3_error!(PreconditionFailed)); + } + if let Some(rematching) = if_none_match + && let Some(strong_etag) = rematching.into_etag() + && info + .etag + .as_ref() + .is_some_and(|etag| ETag::Strong(etag.clone()) == strong_etag) + { + return Err(s3_error!(PreconditionFailed)); + } + } + } + Err(err) => { + if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) { + return Err(ApiError::from(err).into()); + } + + if if_match.is_some() && (is_err_object_not_found(&err) || is_err_version_not_found(&err)) { + return Err(ApiError::from(err).into()); + } + } + } + } + + // check quota for put operation + if let Some(size) = content_length + && let Some(metadata_sys) = rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys.get() + { + let quota_checker = QuotaChecker::new(metadata_sys.clone()); + + match quota_checker + .check_quota(&bucket, QuotaOperation::PutObject, size as u64) + .await + { + Ok(check_result) => { + if !check_result.allowed { + return Err(S3Error::with_message( + S3ErrorCode::InvalidRequest, + format!( + "Bucket quota exceeded. Current usage: {} bytes, limit: {} bytes", + check_result.current_usage, + check_result.quota_limit.unwrap_or(0) + ), + )); + } + } + Err(e) => { + warn!("Quota check failed for bucket {}: {}, allowing operation", bucket, e); + } + } + } + + let Some(body) = body else { return Err(s3_error!(IncompleteBody)) }; + + let mut size = match content_length { + Some(c) => c, + None => { + if let Some(val) = req.headers.get(AMZ_DECODED_CONTENT_LENGTH) { + match atoi::atoi::(val.as_bytes()) { + Some(x) => x, + None => return Err(s3_error!(UnexpectedContent)), + } + } else { + return Err(s3_error!(UnexpectedContent)); + } + } + }; + + if size == -1 { + return Err(s3_error!(UnexpectedContent)); + } + + // Apply adaptive buffer sizing based on file size for optimal streaming performance. + // Uses workload profile configuration (enabled by default) to select appropriate buffer size. + // Buffer sizes range from 32KB to 4MB depending on file size and configured workload profile. + let buffer_size = get_buffer_size_opt_in(size); + let body = tokio::io::BufReader::with_capacity( + buffer_size, + StreamReader::new(body.map(|f| f.map_err(|e| std::io::Error::other(e.to_string())))), + ); + + // let body = Box::new(StreamReader::new(body.map(|f| f.map_err(|e| std::io::Error::other(e.to_string()))))); + + // let mut reader = PutObjReader::new(body, content_length as usize); + + let store = get_validated_store(&bucket).await?; + + let mut metadata = metadata.unwrap_or_default(); + + let object_lock_configuration = match metadata_sys::get_object_lock_config(&bucket).await { + Ok((cfg, _created)) => Some(cfg), + Err(err) => { + if err == StorageError::ConfigNotFound { + None + } else { + warn!("get_object_lock_config err {:?}", err); + return Err(S3Error::with_message( + S3ErrorCode::InternalError, + "Failed to load Object Lock configuration".to_string(), + )); + } + } + }; + + apply_lock_retention(object_lock_configuration, &mut metadata); + + if let Some(content_type) = content_type { + metadata.insert("content-type".to_string(), content_type.to_string()); + } + + extract_metadata_from_mime_with_object_name(&req.headers, &mut metadata, true, Some(&key)); + + if let Some(tags) = tagging { + metadata.insert(AMZ_OBJECT_TAGGING.to_owned(), tags.to_string()); + } + + let mut opts: ObjectOptions = put_opts(&bucket, &key, version_id.clone(), &req.headers, metadata.clone()) + .await + .map_err(ApiError::from)?; + + let mut reader: Box = Box::new(WarpReader::new(body)); + + let actual_size = size; + + let mut md5hex = if let Some(base64_md5) = content_md5 { + let md5 = base64_simd::STANDARD + .decode_to_vec(base64_md5.as_bytes()) + .map_err(|e| ApiError::from(StorageError::other(format!("Invalid content MD5: {e}"))))?; + Some(hex_simd::encode_to_string(&md5, hex_simd::AsciiCase::Lower)) + } else { + None + }; + + let mut sha256hex = get_content_sha256(&req.headers); + + if is_compressible(&req.headers, &key) && size > MIN_COMPRESSIBLE_SIZE as i64 { + let algorithm = CompressionAlgorithm::default(); + metadata.insert(format!("{RESERVED_METADATA_PREFIX_LOWER}compression"), algorithm.to_string()); + + metadata.insert(format!("{RESERVED_METADATA_PREFIX_LOWER}actual-size",), size.to_string()); + + let mut hrd = HashReader::new(reader, size as i64, size as i64, md5hex, sha256hex, false).map_err(ApiError::from)?; + + if let Err(err) = hrd.add_checksum_from_s3s(&req.headers, req.trailing_headers.clone(), false) { + return Err(ApiError::from(StorageError::other(format!("add_checksum error={err:?}"))).into()); + } + + opts.want_checksum = hrd.checksum(); + opts.user_defined + .insert(format!("{RESERVED_METADATA_PREFIX_LOWER}compression"), algorithm.to_string()); + opts.user_defined + .insert(format!("{RESERVED_METADATA_PREFIX_LOWER}actual-size",), size.to_string()); + + reader = Box::new(CompressReader::new(hrd, algorithm)); + size = HashReader::SIZE_PRESERVE_LAYER; + md5hex = None; + sha256hex = None; + } + + let mut reader = HashReader::new(reader, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)?; + + if size >= 0 { + if let Err(err) = reader.add_checksum_from_s3s(&req.headers, req.trailing_headers.clone(), false) { + return Err(ApiError::from(StorageError::other(format!("add_checksum error={err:?}"))).into()); + } + + opts.want_checksum = reader.checksum(); + } + + // Apply encryption using unified SSE API + let encryption_request = EncryptionRequest { + bucket: &bucket, + key: &key, + server_side_encryption, + ssekms_key_id, + sse_customer_algorithm: sse_customer_algorithm.clone(), + sse_customer_key, + sse_customer_key_md5: sse_customer_key_md5.clone(), + content_size: actual_size, + part_number: None, + part_key: None, + part_nonce: None, + }; + + let (effective_sse, effective_kms_key_id) = match sse_encryption(encryption_request).await? { + Some(material) => { + let server_side_encryption = Some(material.server_side_encryption.clone()); + let ssekms_key_id = material.kms_key_id.clone(); + + // Apply encryption wrapper + let encrypted_reader = material.wrap_reader(reader); + reader = HashReader::new(encrypted_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false) + .map_err(ApiError::from)?; + + // Merge encryption metadata + metadata.extend(material.metadata); + + (server_side_encryption, ssekms_key_id) + } + None => (None, None), + }; + + let mut reader = PutObjReader::new(reader); + + let mt2 = metadata.clone(); + opts.user_defined.extend(metadata); + + let repoptions = + get_must_replicate_options(&mt2, "".to_string(), ReplicationStatusType::Empty, ReplicationType::Object, opts.clone()); + + let dsc = must_replicate(&bucket, &key, repoptions).await; + + if dsc.replicate_any() { + let k = format!("{}{}", RESERVED_METADATA_PREFIX_LOWER, "replication-timestamp"); + opts.user_defined.insert(k, jiff::Zoned::now().to_string()); + let k = format!("{}{}", RESERVED_METADATA_PREFIX_LOWER, "replication-status"); + opts.user_defined.insert(k, dsc.pending_status().unwrap_or_default()); + } + + let obj_info = store + .put_object(&bucket, &key, &mut reader, &opts) + .await + .map_err(ApiError::from)?; + + // Fast in-memory update for immediate quota consistency + rustfs_ecstore::data_usage::increment_bucket_usage_memory(&bucket, obj_info.size as u64).await; + + // Invalidate cache for the written object to prevent stale data + let manager = get_concurrency_manager(); + let put_bucket = bucket.clone(); + let put_key = key.clone(); + let mut put_version = obj_info.version_id.map(|v| v.to_string()); + if opts.version_suspended && obj_info.version_id.is_none_or(|v| v.is_nil()) { + put_version = Some("null".to_string()); + } + + helper = helper.object(obj_info.clone()); + if let Some(version_id) = &put_version { + helper = helper.version_id(version_id.clone()); + } + + let put_version_clone = put_version.clone(); + tokio::spawn(async move { + manager + .invalidate_cache_versioned(&put_bucket, &put_key, put_version_clone.as_deref()) + .await; + }); + + let e_tag = obj_info.etag.clone().map(|etag| to_s3s_etag(&etag)); + + let repoptions = + get_must_replicate_options(&mt2, "".to_string(), ReplicationStatusType::Empty, ReplicationType::Object, opts); + + let dsc = must_replicate(&bucket, &key, repoptions).await; + + if dsc.replicate_any() { + schedule_replication(obj_info, store, dsc, ReplicationType::Object).await; + } + + let mut checksum_crc32 = input.checksum_crc32; + let mut checksum_crc32c = input.checksum_crc32c; + let mut checksum_sha1 = input.checksum_sha1; + let mut checksum_sha256 = input.checksum_sha256; + let mut checksum_crc64nvme = input.checksum_crc64nvme; + + if let Some(alg) = &input.checksum_algorithm + && let Some(Some(checksum_str)) = req.trailing_headers.as_ref().map(|trailer| { + let key = match alg.as_str() { + ChecksumAlgorithm::CRC32 => rustfs_rio::ChecksumType::CRC32.key(), + ChecksumAlgorithm::CRC32C => rustfs_rio::ChecksumType::CRC32C.key(), + ChecksumAlgorithm::SHA1 => rustfs_rio::ChecksumType::SHA1.key(), + ChecksumAlgorithm::SHA256 => rustfs_rio::ChecksumType::SHA256.key(), + ChecksumAlgorithm::CRC64NVME => rustfs_rio::ChecksumType::CRC64_NVME.key(), + _ => return None, + }; + trailer.read(|headers| { + headers + .get(key.unwrap_or_default()) + .and_then(|value| value.to_str().ok().map(|s| s.to_string())) + }) + }) + { + match alg.as_str() { + ChecksumAlgorithm::CRC32 => checksum_crc32 = checksum_str, + ChecksumAlgorithm::CRC32C => checksum_crc32c = checksum_str, + ChecksumAlgorithm::SHA1 => checksum_sha1 = checksum_str, + ChecksumAlgorithm::SHA256 => checksum_sha256 = checksum_str, + ChecksumAlgorithm::CRC64NVME => checksum_crc64nvme = checksum_str, + _ => (), + } + } + + let output = PutObjectOutput { + e_tag, + server_side_encryption: effective_sse, // TDD: Return effective encryption config + sse_customer_algorithm, + sse_customer_key_md5, + ssekms_key_id: effective_kms_key_id, // TDD: Return effective KMS key ID + checksum_crc32, + checksum_crc32c, + checksum_sha1, + checksum_sha256, + checksum_crc64nvme, + version_id: put_version, + ..Default::default() + }; + + let result = Ok(S3Response::new(output)); + let _ = helper.complete(&result); + result + } + + async fn put_object_extract(&self, req: S3Request) -> S3Result> { + let helper = OperationHelper::new(&req, EventName::ObjectCreatedPut, "s3:PutObject").suppress_event(); + let input = req.input; + + let PutObjectInput { + body, + bucket, + key, + version_id, + content_length, + content_md5, + .. + } = input; + + let event_version_id = version_id; + let Some(body) = body else { return Err(s3_error!(IncompleteBody)) }; + + let size = match content_length { + Some(c) => c, + None => { + if let Some(val) = req.headers.get(AMZ_DECODED_CONTENT_LENGTH) { + match atoi::atoi::(val.as_bytes()) { + Some(x) => x, + None => return Err(s3_error!(UnexpectedContent)), + } + } else { + return Err(s3_error!(UnexpectedContent)); + } + } + }; + + // Apply adaptive buffer sizing based on file size for optimal streaming performance. + // Uses workload profile configuration (enabled by default) to select appropriate buffer size. + // Buffer sizes range from 32KB to 4MB depending on file size and configured workload profile. + let buffer_size = get_buffer_size_opt_in(size); + let body = tokio::io::BufReader::with_capacity( + buffer_size, + StreamReader::new(body.map(|f| f.map_err(|e| std::io::Error::other(e.to_string())))), + ); + + let Some(ext) = Path::new(&key).extension().and_then(|s| s.to_str()) else { + return Err(s3_error!(InvalidArgument, "key extension not found")); + }; + + let ext = ext.to_owned(); + + let md5hex = if let Some(base64_md5) = content_md5 { + let md5 = base64_simd::STANDARD + .decode_to_vec(base64_md5.as_bytes()) + .map_err(|e| ApiError::from(StorageError::other(format!("Invalid content MD5: {e}"))))?; + Some(hex_simd::encode_to_string(&md5, hex_simd::AsciiCase::Lower)) + } else { + None + }; + + let sha256hex = get_content_sha256(&req.headers); + let actual_size = size; + + let reader: Box = Box::new(WarpReader::new(body)); + + let mut hreader = HashReader::new(reader, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)?; + + if let Err(err) = hreader.add_checksum_from_s3s(&req.headers, req.trailing_headers.clone(), false) { + return Err(ApiError::from(StorageError::other(format!("add_checksum error={err:?}"))).into()); + } + + // TODO: support zip + let decoder = CompressionFormat::from_extension(&ext).get_decoder(hreader).map_err(|e| { + error!("get_decoder err {:?}", e); + s3_error!(InvalidArgument, "get_decoder err") + })?; + + let mut ar = Archive::new(decoder); + let mut entries = ar.entries().map_err(|e| { + error!("get entries err {:?}", e); + s3_error!(InvalidArgument, "get entries err") + })?; + + let Some(store) = new_object_layer_fn() else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + + let prefix = req + .headers + .get("X-Amz-Meta-Rustfs-Snowball-Prefix") + .map(|v| v.to_str().unwrap_or_default()) + .unwrap_or_default(); + let version_id = match event_version_id { + Some(v) => v.to_string(), + None => String::new(), + }; + while let Some(entry) = entries.next().await { + let f = match entry { + Ok(f) => f, + Err(e) => { + error!("Failed to read archive entry: {}", e); + return Err(s3_error!(InvalidArgument, "Failed to read archive entry: {:?}", e)); + } + }; + + if f.header().entry_type().is_dir() { + continue; + } + + if let Ok(fpath) = f.path() { + let mut fpath = fpath.to_string_lossy().to_string(); + + if !prefix.is_empty() { + fpath = format!("{prefix}/{fpath}"); + } + + let mut size = f.header().size().unwrap_or_default() as i64; + + debug!("Extracting file: {}, size: {} bytes", fpath, size); + + let mut reader: Box = Box::new(WarpReader::new(f)); + + let mut metadata = HashMap::new(); + + let actual_size = size; + + if is_compressible(&HeaderMap::new(), &fpath) && size > MIN_COMPRESSIBLE_SIZE as i64 { + metadata.insert( + format!("{RESERVED_METADATA_PREFIX_LOWER}compression"), + CompressionAlgorithm::default().to_string(), + ); + metadata.insert(format!("{RESERVED_METADATA_PREFIX_LOWER}actual-size",), size.to_string()); + + let hrd = HashReader::new(reader, size, actual_size, None, None, false).map_err(ApiError::from)?; + + reader = Box::new(CompressReader::new(hrd, CompressionAlgorithm::default())); + size = HashReader::SIZE_PRESERVE_LAYER; + } + + let hrd = HashReader::new(reader, size, actual_size, None, None, false).map_err(ApiError::from)?; + let mut reader = PutObjReader::new(hrd); + + let _obj_info = store + .put_object(&bucket, &fpath, &mut reader, &ObjectOptions::default()) + .await + .map_err(ApiError::from)?; + + // Invalidate cache for the written object to prevent stale data + let manager = get_concurrency_manager(); + let fpath_clone = fpath.clone(); + let bucket_clone = bucket.clone(); + tokio::spawn(async move { + manager.invalidate_cache_versioned(&bucket_clone, &fpath_clone, None).await; + }); + + let e_tag = _obj_info.etag.clone().map(|etag| to_s3s_etag(&etag)); + + // // store.put_object(bucket, object, data, opts); + + let output = PutObjectOutput { + e_tag, + ..Default::default() + }; + + let event_args = rustfs_notify::EventArgs { + event_name: EventName::ObjectCreatedPut, + bucket_name: bucket.clone(), + object: _obj_info.clone(), + req_params: extract_params_header(&req.headers), + resp_elements: extract_resp_elements(&S3Response::new(output.clone())), + version_id: version_id.clone(), + host: get_request_host(&req.headers), + port: get_request_port(&req.headers), + user_agent: get_request_user_agent(&req.headers), + }; + + // Asynchronous call will not block the response of the current request + tokio::spawn(async move { + notifier_global::notify(event_args).await; + }); + } + } + + // match decompress( + // body, + // CompressionFormat::from_extension(&ext), + // |entry: tokio_tar::Entry>>| async move { + // let path = entry.path().unwrap(); + // debug!("Extracted: {}", path.display()); + // Ok(()) + // }, + // ) + // .await + // { + // Ok(_) => info!("Decompression completed successfully"), + // Err(e) => error!("Decompression failed: {}", e), + // } + + let mut checksum_crc32 = input.checksum_crc32; + let mut checksum_crc32c = input.checksum_crc32c; + let mut checksum_sha1 = input.checksum_sha1; + let mut checksum_sha256 = input.checksum_sha256; + let mut checksum_crc64nvme = input.checksum_crc64nvme; + + if let Some(alg) = &input.checksum_algorithm + && let Some(Some(checksum_str)) = req.trailing_headers.as_ref().map(|trailer| { + let key = match alg.as_str() { + ChecksumAlgorithm::CRC32 => rustfs_rio::ChecksumType::CRC32.key(), + ChecksumAlgorithm::CRC32C => rustfs_rio::ChecksumType::CRC32C.key(), + ChecksumAlgorithm::SHA1 => rustfs_rio::ChecksumType::SHA1.key(), + ChecksumAlgorithm::SHA256 => rustfs_rio::ChecksumType::SHA256.key(), + ChecksumAlgorithm::CRC64NVME => rustfs_rio::ChecksumType::CRC64_NVME.key(), + _ => return None, + }; + trailer.read(|headers| { + headers + .get(key.unwrap_or_default()) + .and_then(|value| value.to_str().ok().map(|s| s.to_string())) + }) + }) + { + match alg.as_str() { + ChecksumAlgorithm::CRC32 => checksum_crc32 = checksum_str, + ChecksumAlgorithm::CRC32C => checksum_crc32c = checksum_str, + ChecksumAlgorithm::SHA1 => checksum_sha1 = checksum_str, + ChecksumAlgorithm::SHA256 => checksum_sha256 = checksum_str, + ChecksumAlgorithm::CRC64NVME => checksum_crc64nvme = checksum_str, + _ => (), + } + } + + warn!( + "put object extract checksum_crc32={checksum_crc32:?}, checksum_crc32c={checksum_crc32c:?}, checksum_sha1={checksum_sha1:?}, checksum_sha256={checksum_sha256:?}, checksum_crc64nvme={checksum_crc64nvme:?}", + ); + + // TODO: etag + let output = PutObjectOutput { + checksum_crc32, + checksum_crc32c, + checksum_sha1, + checksum_sha256, + checksum_crc64nvme, + ..Default::default() + }; + let result = Ok(S3Response::new(output)); + let _ = helper.complete(&result); + result + } +} diff --git a/rustfs/tests/lifecycle_minio_sdk_test.rs b/rustfs/tests/lifecycle_minio_sdk_test.rs new file mode 100644 index 000000000..d5d7be8ad --- /dev/null +++ b/rustfs/tests/lifecycle_minio_sdk_test.rs @@ -0,0 +1,188 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use anyhow::{Context, Result}; +use aws_config::BehaviorVersion; +use aws_sdk_s3::Client; +use aws_sdk_s3::config::{Credentials, Region}; +use aws_sdk_s3::presigning::PresigningConfig; +use aws_sdk_s3::types::{ + BucketLifecycleConfiguration, ExpirationStatus, LifecycleExpiration, LifecycleRule, LifecycleRuleFilter, +}; +use serial_test::serial; +use std::env; +use std::time::Duration; +use uuid::Uuid; + +struct Settings { + oss_endpoint: String, + oss_access_key: String, + oss_secret_key: String, + oss_bucket_name: String, + oss_lifecycle_days: i32, + #[allow(dead_code)] + oss_secure: bool, + oss_region: String, +} + +impl Settings { + fn new() -> Self { + Self { + oss_endpoint: env::var("TEST_RUSTFS_SERVER").unwrap_or_else(|_| "http://localhost:9000".to_string()), + oss_access_key: "rustfsadmin".to_string(), + oss_secret_key: "rustfsadmin".to_string(), + oss_bucket_name: "mblock99".to_string(), + oss_lifecycle_days: 1, + oss_secure: false, + oss_region: "us-east-1".to_string(), + } + } +} + +struct Oss { + client: Client, + bucket_name: String, + lifecycle_days: i32, +} + +impl Oss { + async fn new(settings: &Settings) -> Result { + let credentials = Credentials::new(&settings.oss_access_key, &settings.oss_secret_key, None, None, "test"); + + let config = aws_config::defaults(BehaviorVersion::latest()) + .credentials_provider(credentials) + .region(Region::new(settings.oss_region.clone())) + .endpoint_url(&settings.oss_endpoint) + .load() + .await; + + let client = Client::new(&config); + + Ok(Self { + client, + bucket_name: settings.oss_bucket_name.clone(), + lifecycle_days: settings.oss_lifecycle_days, + }) + } + + fn new_uuid(&self) -> String { + Uuid::new_v4().to_string() + } + + async fn create_bucket(&self) -> Result<()> { + match self.client.head_bucket().bucket(&self.bucket_name).send().await { + Ok(_) => { + println!("Bucket {} already exists", self.bucket_name); + Ok(()) + } + Err(_) => match self.client.create_bucket().bucket(&self.bucket_name).send().await { + Ok(_) => { + self.set_lifecycle_expiration(self.lifecycle_days, None).await?; + println!("Bucket {} created successfully", self.bucket_name); + Ok(()) + } + Err(err) => { + let err_msg = err.to_string(); + if err_msg.contains("BucketAlreadyOwnedByYou") { + println!("Bucket {} already owned by you; continuing", self.bucket_name); + Ok(()) + } else { + Err(anyhow::anyhow!("Failed to create bucket: {}", err)) + } + } + }, + } + } + + async fn set_lifecycle_expiration(&self, days: i32, prefix: Option<&str>) -> Result<()> { + let prefix_str = prefix.unwrap_or(""); + let rule_id = format!("expire-{}-{}d", if prefix_str.is_empty() { "all" } else { prefix_str }, days); + + let filter = LifecycleRuleFilter::builder().prefix(prefix_str.to_string()).build(); + + let expiration = LifecycleExpiration::builder().days(days).build(); + + let rule = LifecycleRule::builder() + .id(rule_id) + .status(ExpirationStatus::Enabled) + .filter(filter) + .expiration(expiration) + .build() + .context("Failed to build lifecycle rule")?; + + let lifecycle_config = BucketLifecycleConfiguration::builder() + .rules(rule) + .build() + .context("Failed to build bucket lifecycle configuration")?; + + self.client + .put_bucket_lifecycle_configuration() + .bucket(&self.bucket_name) + .lifecycle_configuration(lifecycle_config) + .send() + .await + .context("Failed to set bucket lifecycle")?; + + Ok(()) + } + + #[allow(dead_code)] + async fn upload_file(&self, filename: &str, content: &[u8]) -> Result { + let ext = std::path::Path::new(filename) + .extension() + .and_then(|s| s.to_str()) + .unwrap_or(""); + + let uuid = self.new_uuid(); + let object_name = if ext.is_empty() { uuid } else { format!("{}.{}", uuid, ext) }; + + self.client + .put_object() + .bucket(&self.bucket_name) + .key(&object_name) + .body(content.to_vec().into()) + .send() + .await + .context("Failed to upload file")?; + + Ok(object_name) + } + + #[allow(dead_code)] + async fn get_presigned_url(&self, filename: &str) -> Result { + let expires_in = Duration::from_secs((self.lifecycle_days * 24 * 60 * 60) as u64); + let presigning_config = PresigningConfig::expires_in(expires_in)?; + + let presigned_req = self + .client + .get_object() + .bucket(&self.bucket_name) + .key(filename) + .presigned(presigning_config) + .await + .context("Failed to get presigned URL")?; + + Ok(presigned_req.uri().to_string()) + } +} + +#[tokio::test] +#[serial] +#[ignore] +async fn test_lifecycle_minio_sdk() -> Result<()> { + let settings = Settings::new(); + let oss = Oss::new(&settings).await?; + oss.create_bucket().await?; + Ok(()) +} diff --git a/rustfs/tests/lifecycle_miniosdk_test.py b/rustfs/tests/lifecycle_miniosdk_test.py deleted file mode 100644 index 2b30a2254..000000000 --- a/rustfs/tests/lifecycle_miniosdk_test.py +++ /dev/null @@ -1,120 +0,0 @@ -# Copyright 2024 RustFS Team -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -from datetime import timedelta -import uuid -from fastapi import UploadFile -from minio import Minio -from minio.commonconfig import Filter -from minio.error import S3Error -from minio.lifecycleconfig import Expiration, LifecycleConfig, Rule -import os -from collections import namedtuple - -Settings = namedtuple("Settings", ['oss_endpoint', - 'oss_access_key', - 'oss_secret_key', - 'oss_bucket_name', - 'oss_lifecycle_days', - 'oss_secure', - 'oss_region']) -settings = Settings( - oss_endpoint=os.getenv("TEST_RUSTFS_SERVER", "localhost:9000"), - oss_access_key = "rustfsadmin", - oss_secret_key = "rustfsadmin", - oss_bucket_name = "mblock99", - oss_lifecycle_days = 1, - oss_secure = False, - oss_region = "" -) - -class OSS: - def __init__(self): - self.bucket_name = settings.oss_bucket_name - self.lifecycle_days = settings.oss_lifecycle_days - self.client = Minio( - endpoint=settings.oss_endpoint, - access_key=settings.oss_access_key, - secret_key=settings.oss_secret_key, - secure=settings.oss_secure, - region=settings.oss_region, - ) - - def new_uuid(self): - return str(uuid.uuid4()) - - def create_bucket(self): - found = self.client.bucket_exists(self.bucket_name) - if not found: - try: - self.client.make_bucket(self.bucket_name) - self.set_lifecycle_expiration(days=self.lifecycle_days) - print(f"Bucket {self.bucket_name} created successfully") - except S3Error as exc: - if exc.code == "BucketAlreadyOwnedByYou": - print(f"Bucket {self.bucket_name} already owned by you; continuing") - else: - raise - else: - print(f"Bucket {self.bucket_name} already exists") - - def set_lifecycle_expiration(self,days: int = 1, prefix: str = "") -> None: - """ - 设置按天自动过期删除（MinIO 按天、每天巡检一次；<1天不生效） - """ - rule_filter = Filter(prefix=prefix or "") - rule = Rule( - rule_id=f"expire-{prefix or 'all'}-{days}d", - status="Enabled", - rule_filter=rule_filter, - expiration=Expiration(days=int(days)), - ) - cfg = LifecycleConfig([rule]) - self.client.set_bucket_lifecycle(self.bucket_name, cfg) - - def upload_file(self, file: UploadFile): - """ - 上传文件到OSS，返回文件的UUID - """ - ext = os.path.splitext(file.filename)[1] - uuid = self.new_uuid() - filename = f'{uuid}{ext}' - file.file.seek(0) - self.client.put_object( - self.bucket_name, - filename, - file.file, - length=-1, - part_size=10*1024*1024, - ) - return filename - - def get_presigned_url(self, filename: str): - """ - 获取文件的预签名URL，用于下载文件 - """ - return self.client.presigned_get_object( - self.bucket_name, filename, expires=timedelta(days=self.lifecycle_days) - ) - - -def get_oss(): - """ - 获取OSS实例 - """ - return OSS() - -if __name__ == "__main__": - oss = get_oss() - oss.create_bucket()