gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 16:28:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: missing object.key in S3 event notifications for multipart uploads #1609 (#1624)

Browse Source 
Co-authored-by: houseme <housemecn@gmail.com>
This commit is contained in:
heihutu
2026-01-27 13:32:56 +08:00
committed by GitHub
parent db29c0cae2
commit 74759b6e99
4 changed files with 176 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/notify/src/event.rs
+16 -3
View File
@@ -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<String, String>,
pub version_id: String,
pub host: String,
pub port: u16,
pub user_agent: String,
}
@@ -307,6 +312,7 @@ pub struct EventArgsBuilder {
resp_elements: HashMap<String, String>,
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<String>) -> 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,
}
}
crates/utils/src/notify/mod.rs
+146 -18
View File
@@ -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<T>(req: &S3Request<T>) -> HashMap<String, String> {
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<String, String> {
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<String, String> {
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<String, String> {
}
/// Extract response elements from S3Response, mainly header information.
#[allow(dead_code)]
pub fn extract_resp_elements<T>(resp: &S3Response<T>) -> HashMap<String, String> {
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::<u16>().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::<u16>()
&& 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::<u16>().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::<u64>().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);
}
}
rustfs/src/storage/ecfs.rs
+11 -8
View File
@@ -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))
rustfs/src/storage/helper.rs
+3 -2
View File
@@ -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),