From 74759b6e99b528fd4d849601ecf5c63141cba8a1 Mon Sep 17 00:00:00 2001 From: heihutu <30542132+heihutu@users.noreply.github.com> Date: Tue, 27 Jan 2026 13:32:56 +0800 Subject: [PATCH] fix: missing object.key in S3 event notifications for multipart uploads #1609 (#1624) Co-authored-by: houseme --- crates/notify/src/event.rs | 19 +++- crates/utils/src/notify/mod.rs | 164 +++++++++++++++++++++++++++++---- rustfs/src/storage/ecfs.rs | 19 ++-- rustfs/src/storage/helper.rs | 5 +- 4 files changed, 176 insertions(+), 31 deletions(-) diff --git a/crates/notify/src/event.rs b/crates/notify/src/event.rs index 569798c3a..cb7c81132 100644 --- a/crates/notify/src/event.rs +++ b/crates/notify/src/event.rs @@ -183,7 +183,7 @@ impl Event { pub fn new(args: EventArgs) -> Self { let event_time = Utc::now().naive_local(); - let unique_id = match args.object.mod_time { + let sequencer = match args.object.mod_time { Some(t) => format!("{:X}", t.unix_timestamp_nanos()), None => format!("{:X}", event_time.and_utc().timestamp_nanos_opt().unwrap_or(0)), }; @@ -213,7 +213,7 @@ impl Event { object: Object { key: key_name, version_id, - sequencer: unique_id, + sequencer, ..Default::default() }, }; @@ -249,7 +249,11 @@ impl Event { s3: s3_metadata, source: Source { host: args.host, - port: "".to_string(), + port: if args.port == 0 { + "".to_string() + } else { + args.port.to_string() + }, user_agent: args.user_agent, }, } @@ -271,6 +275,7 @@ pub struct EventArgs { pub resp_elements: HashMap, pub version_id: String, pub host: String, + pub port: u16, pub user_agent: String, } @@ -307,6 +312,7 @@ pub struct EventArgsBuilder { resp_elements: HashMap, version_id: String, host: String, + port: u16, user_agent: String, } @@ -375,6 +381,12 @@ impl EventArgsBuilder { self } + /// Sets the port. + pub fn port(mut self, port: u16) -> Self { + self.port = port; + self + } + /// Sets the user agent. pub fn user_agent(mut self, user_agent: impl Into) -> Self { self.user_agent = user_agent.into(); @@ -393,6 +405,7 @@ impl EventArgsBuilder { resp_elements: self.resp_elements, version_id: self.version_id, host: self.host, + port: self.port, user_agent: self.user_agent, } } diff --git a/crates/utils/src/notify/mod.rs b/crates/utils/src/notify/mod.rs index 9b2029d50..1d0ddb4d0 100644 --- a/crates/utils/src/notify/mod.rs +++ b/crates/utils/src/notify/mod.rs @@ -21,20 +21,20 @@ use s3s::{S3Request, S3Response}; pub use net::*; /// Extract request parameters from S3Request, mainly header information. -#[allow(dead_code)] pub fn extract_req_params(req: &S3Request) -> HashMap { - let mut params = HashMap::new(); - for (key, value) in req.headers.iter() { - if let Ok(val_str) = value.to_str() { - params.insert(key.as_str().to_string(), val_str.to_string()); - } - } - params + extract_params_header(&req.headers) } /// Extract request parameters from hyper::HeaderMap, mainly header information. /// This function is useful when you have a raw HTTP request and need to extract parameters. +#[deprecated(since = "0.1.0", note = "Use extract_params_header instead")] pub fn extract_req_params_header(head: &HeaderMap) -> HashMap { + extract_params_header(head) +} + +/// Extract parameters from hyper::HeaderMap, mainly header information. +/// This function is useful when you have a raw HTTP request and need to extract parameters. +pub fn extract_params_header(head: &HeaderMap) -> HashMap { let mut params = HashMap::new(); for (key, value) in head.iter() { if let Ok(val_str) = value.to_str() { @@ -45,19 +45,11 @@ pub fn extract_req_params_header(head: &HeaderMap) -> HashMap { } /// Extract response elements from S3Response, mainly header information. -#[allow(dead_code)] pub fn extract_resp_elements(resp: &S3Response) -> HashMap { - let mut params = HashMap::new(); - for (key, value) in resp.headers.iter() { - if let Ok(val_str) = value.to_str() { - params.insert(key.as_str().to_string(), val_str.to_string()); - } - } - params + extract_params_header(&resp.headers) } /// Get host from header information. -#[allow(dead_code)] pub fn get_request_host(headers: &HeaderMap) -> String { headers .get("host") @@ -66,8 +58,81 @@ pub fn get_request_host(headers: &HeaderMap) -> String { .to_string() } +/// Get Port from header information. +/// Priority: +/// 1. x-forwarded-port +/// 2. host header (parse port) +/// If host has no port, try to deduce from x-forwarded-proto (http->80, https->443) +/// 3. port header +/// +/// If the port cannot be determined, returns 0. +pub fn get_request_port(headers: &HeaderMap) -> u16 { + // 1. Try x-forwarded-port + if let Some(port) = headers + .get("x-forwarded-port") + .and_then(|v| v.to_str().ok()) + .and_then(|s| s.parse::().ok()) + { + return port; + } + + // 2. Try host header + if let Some(host) = headers.get("host").and_then(|v| v.to_str().ok()) { + if let Some(idx) = host.rfind(':') { + // Check if it's an IPv6 address with port, e.g., [::1]:8080 + // If ']' is present, the colon must be after it. + let valid_colon = match host.rfind(']') { + Some(close_bracket_idx) => idx > close_bracket_idx, + None => true, + }; + + if valid_colon + && let Ok(port) = host[idx + 1..].parse::() + && port > 0 + { + return port; + } + } + + // If host is present but no port found (or parsing failed, or port is 0), + // try to deduce from x-forwarded-proto + if let Some(proto) = headers.get("x-forwarded-proto").and_then(|v| v.to_str().ok()) { + match proto { + "http" => return 80, + "https" => return 443, + _ => {} + } + } + } + + // 3. Fallback to "port" header + headers + .get("port") + .and_then(|v| v.to_str().ok()) + .and_then(|s| s.parse::().ok()) + .unwrap_or(0) +} + +/// Get content-length from header information. +pub fn get_request_content_length(headers: &HeaderMap) -> u64 { + headers + .get("content-length") + .and_then(|v| v.to_str().ok()) + .and_then(|s| s.parse::().ok()) + .unwrap_or(0) +} + +/// Get referer from header information. +/// If the referer header is not present, returns an empty string. +pub fn get_request_referer(headers: &HeaderMap) -> String { + headers + .get("referer") + .and_then(|v| v.to_str().ok()) + .unwrap_or_default() + .to_string() +} + /// Get user-agent from header information. -#[allow(dead_code)] pub fn get_request_user_agent(headers: &HeaderMap) -> String { headers .get("user-agent") @@ -75,3 +140,66 @@ pub fn get_request_user_agent(headers: &HeaderMap) -> String { .unwrap_or_default() .to_string() } + +#[cfg(test)] +mod tests { + use super::*; + use hyper::header::HeaderValue; + + #[test] + fn test_get_request_port() { + let mut headers = HeaderMap::new(); + + // Case 1: No port info + assert_eq!(get_request_port(&headers), 0); + + // Case 2: port header + headers.insert("port", HeaderValue::from_static("8080")); + assert_eq!(get_request_port(&headers), 8080); + + // Case 3: host header with port + headers.remove("port"); + headers.insert("host", HeaderValue::from_static("example.com:9000")); + assert_eq!(get_request_port(&headers), 9000); + + // Case 4: host header without port, no proto + headers.insert("host", HeaderValue::from_static("example.com")); + assert_eq!(get_request_port(&headers), 0); + + // Case 5: IPv6 host with port + headers.insert("host", HeaderValue::from_static("[::1]:9001")); + assert_eq!(get_request_port(&headers), 9001); + + // Case 6: IPv6 host without port + headers.insert("host", HeaderValue::from_static("[::1]")); + assert_eq!(get_request_port(&headers), 0); + + // Case 7: x-forwarded-port + headers.insert("x-forwarded-port", HeaderValue::from_static("7000")); + // Even if host is present, x-forwarded-port takes precedence + assert_eq!(get_request_port(&headers), 7000); + + // Case 8: host without port, but x-forwarded-proto is http + headers.remove("x-forwarded-port"); + headers.insert("host", HeaderValue::from_static("example.com")); + headers.insert("x-forwarded-proto", HeaderValue::from_static("http")); + assert_eq!(get_request_port(&headers), 80); + + // Case 9: host without port, but x-forwarded-proto is https + headers.insert("x-forwarded-proto", HeaderValue::from_static("https")); + assert_eq!(get_request_port(&headers), 443); + + // Case 10: host without port, unknown proto + headers.insert("x-forwarded-proto", HeaderValue::from_static("ftp")); + assert_eq!(get_request_port(&headers), 0); + + // Case 11: host with port 0, should fallback to proto + headers.insert("host", HeaderValue::from_static("example.com:0")); + headers.insert("x-forwarded-proto", HeaderValue::from_static("https")); + assert_eq!(get_request_port(&headers), 443); + + // Case 12: host with port 0, no proto + headers.remove("x-forwarded-proto"); + assert_eq!(get_request_port(&headers), 0); + } +} diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index cda98ad9c..390165cf8 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -111,7 +111,8 @@ use rustfs_targets::{ arn::{ARN, TargetIDError}, }; use rustfs_utils::{ - CompressionAlgorithm, extract_req_params_header, extract_resp_elements, get_request_host, get_request_user_agent, + CompressionAlgorithm, extract_params_header, extract_resp_elements, get_request_host, get_request_port, + get_request_user_agent, http::{ AMZ_BUCKET_REPLICATION_STATUS, AMZ_CHECKSUM_MODE, AMZ_CHECKSUM_TYPE, headers::{ @@ -387,10 +388,11 @@ impl FS { event_name: EventName::ObjectCreatedPut, bucket_name: bucket.clone(), object: _obj_info.clone(), - req_params: extract_req_params_header(&req.headers), + 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), }; @@ -739,11 +741,12 @@ impl S3 for FS { } } + let region = rustfs_ecstore::global::get_global_region().unwrap_or_else(|| "us-east-1".to_string()); let output = CompleteMultipartUploadOutput { bucket: Some(bucket.clone()), key: Some(key.clone()), e_tag: obj_info.etag.clone().map(|etag| to_s3s_etag(&etag)), - location: Some("us-east-1".to_string()), + location: Some(region.clone()), server_side_encryption: server_side_encryption.clone(), // TDD: Return encryption info ssekms_key_id: ssekms_key_id.clone(), // TDD: Return KMS key ID if present checksum_crc32: checksum_crc32.clone(), @@ -764,7 +767,7 @@ impl S3 for FS { bucket: Some(bucket.clone()), key: Some(key.clone()), e_tag: obj_info.etag.clone().map(|etag| to_s3s_etag(&etag)), - location: Some("us-east-1".to_string()), + location: Some(region), server_side_encryption, // TDD: Return encryption info ssekms_key_id, // TDD: Return KMS key ID if present checksum_crc32, @@ -777,10 +780,10 @@ impl S3 for FS { }; let mt2 = HashMap::new(); - let repoptions = + let replicate_options = get_must_replicate_options(&mt2, "".to_string(), ReplicationStatusType::Empty, ReplicationType::Object, opts.clone()); - let dsc = must_replicate(&bucket, &key, repoptions).await; + let dsc = must_replicate(&bucket, &key, replicate_options).await; if dsc.replicate_any() { warn!("need multipart replication"); @@ -792,7 +795,7 @@ impl S3 for FS { ); // Set object info for event notification - helper = helper.object(obj_info.clone()); + helper = helper.object(obj_info); if let Some(version_id) = &mpu_version_for_event { helper = helper.version_id(version_id.clone()); } @@ -2031,7 +2034,7 @@ impl S3 for FS { }, ) .version_id(dobj.version_id.map(|v| v.to_string()).unwrap_or_default()) - .req_params(extract_req_params_header(&req_headers)) + .req_params(extract_params_header(&req_headers)) .resp_elements(extract_resp_elements(&S3Response::new(DeleteObjectsOutput::default()))) .host(get_request_host(&req_headers)) .user_agent(get_request_user_agent(&req_headers)) diff --git a/rustfs/src/storage/helper.rs b/rustfs/src/storage/helper.rs index aa4f00940..bdd2d72ff 100644 --- a/rustfs/src/storage/helper.rs +++ b/rustfs/src/storage/helper.rs @@ -21,7 +21,7 @@ use rustfs_ecstore::store_api::ObjectInfo; use rustfs_notify::{EventArgsBuilder, notifier_global}; use rustfs_targets::EventName; use rustfs_utils::{ - extract_req_params, extract_req_params_header, extract_resp_elements, get_request_host, get_request_user_agent, + extract_params_header, extract_req_params, extract_resp_elements, get_request_host, get_request_port, get_request_user_agent, }; use s3s::{S3Request, S3Response, S3Result}; use std::future::Future; @@ -96,8 +96,9 @@ impl OperationHelper { // object is a placeholder that must be set later using the `object()` method. let event_builder = EventArgsBuilder::new(event, bucket, ObjectInfo::default()) .host(get_request_host(&req.headers)) + .port(get_request_port(&req.headers)) .user_agent(get_request_user_agent(&req.headers)) - .req_params(extract_req_params_header(&req.headers)); + .req_params(extract_params_header(&req.headers)); Self { audit_builder: Some(audit_builder),