From 2cdba03deef5793ffc638bd6be19a1a397213050 Mon Sep 17 00:00:00 2001 From: cxymds Date: Sun, 2 Aug 2026 19:28:54 +0800 Subject: [PATCH] fix(authz): gate replication-only PUT headers (#5625) --- .../src/admin/handlers/object_zip_download.rs | 2 + rustfs/src/app/multipart_usecase.rs | 19 ++- rustfs/src/app/object_usecase.rs | 51 +++++-- rustfs/src/app/storage_api.rs | 11 +- rustfs/src/protocols/client.rs | 1 + rustfs/src/storage/access.rs | 43 +++++- rustfs/src/storage/options.rs | 125 +++++++++++++++--- rustfs/src/storage/storage_api.rs | 11 +- 8 files changed, 213 insertions(+), 50 deletions(-) diff --git a/rustfs/src/admin/handlers/object_zip_download.rs b/rustfs/src/admin/handlers/object_zip_download.rs index 5cec480bb..b3597790e 100644 --- a/rustfs/src/admin/handlers/object_zip_download.rs +++ b/rustfs/src/admin/handlers/object_zip_download.rs @@ -206,6 +206,7 @@ impl From for ReqInfo { bucket: snapshot.bucket, object: snapshot.object, version_id: snapshot.version_id, + replication_request_authorized: false, region: current_region(), request_context: None, } @@ -1424,6 +1425,7 @@ mod tests { bucket: Some("photos".to_string()), object: None, version_id: None, + replication_request_authorized: false, region: current_region(), request_context: None, } diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index cc2108ff8..7af8643c5 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -15,7 +15,7 @@ //! Multipart application use-case contracts. use super::storage_api::multipart_usecase::ECStore; -use super::storage_api::multipart_usecase::access::has_bypass_governance_header; +use super::storage_api::multipart_usecase::access::{has_bypass_governance_header, replication_request_authorized}; use super::storage_api::multipart_usecase::bucket::quota::checker::QuotaChecker; use super::storage_api::multipart_usecase::bucket::{ lifecycle::{bucket_lifecycle_audit::LcEventSrc, bucket_lifecycle_ops::enqueue_transition_immediate}, @@ -39,8 +39,9 @@ use super::storage_api::multipart_usecase::io::{DecryptReader, EncryptReader, Ha use super::storage_api::multipart_usecase::io::{HashReader, WriteEncryption, WritePlan}; use super::storage_api::multipart_usecase::object_utils::to_s3s_etag; use super::storage_api::multipart_usecase::options::{ - copy_src_opts, extract_metadata_from_mime, get_complete_multipart_upload_opts, get_content_sha256_with_query, get_opts, - namespace_reserved_user_metadata, parse_copy_source_range, put_opts, validate_archive_content_encoding, + copy_src_opts, extract_metadata_from_mime, get_complete_multipart_upload_opts_with_replication_authorization, + get_content_sha256_with_query, get_opts, namespace_reserved_user_metadata, parse_copy_source_range, + put_opts_with_replication_authorization, validate_archive_content_encoding, }; use super::storage_api::multipart_usecase::s3_api::multipart::{ ListMultipartUploadsParams, build_list_multipart_uploads_output, build_list_parts_output, @@ -384,6 +385,7 @@ impl DefaultMultipartUsecase { EventName::ObjectCreatedCompleteMultipartUpload, S3Operation::CompleteMultipartUpload, ); + let replication_authorized = replication_request_authorized(&req); let input = req.input; let CompleteMultipartUploadInput { multipart_upload, @@ -448,7 +450,8 @@ impl DefaultMultipartUsecase { )); }; - let mut opts = get_complete_multipart_upload_opts(&req.headers).map_err(ApiError::from)?; + let mut opts = get_complete_multipart_upload_opts_with_replication_authorization(&req.headers, replication_authorized) + .map_err(ApiError::from)?; let versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await; let version_suspended = BucketVersioningSys::prefix_suspended(&bucket, &key).await; opts.versioned = versioned; @@ -661,6 +664,7 @@ impl DefaultMultipartUsecase { let helper = OperationHelper::new(&req, EventName::ObjectCreatedCreateMultipartUpload, S3Operation::CreateMultipartUpload) .suppress_event(); + let replication_authorized = replication_request_authorized(&req); let CreateMultipartUploadInput { bucket, key, @@ -761,9 +765,10 @@ impl DefaultMultipartUsecase { // compression here would make GET decode the completed object as one stream. let mt2 = metadata.clone(); - let mut opts: ObjectOptions = put_opts(&bucket, &key, version_id, &req.headers, metadata) - .await - .map_err(ApiError::from)?; + let mut opts: ObjectOptions = + put_opts_with_replication_authorization(&bucket, &key, version_id, &req.headers, metadata, replication_authorized) + .await + .map_err(ApiError::from)?; let dsc = must_replicate_object(&bucket, &key, &mt2, "".to_string(), opts.delete_marker_replication_status(), opts.clone()) diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 265fe3afc..b01f1d784 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -20,8 +20,8 @@ use rustfs_io_metrics::buffered_write; use crate::storage_api::table::get_bucket_metadata; use super::storage_api::object_usecase::access::{ - PostObjectRequestMarker, authorize_request, has_bypass_governance_header, recursive_force_delete_is_authorized, req_info_mut, - req_info_ref, + PostObjectRequestMarker, authorize_request, has_bypass_governance_header, recursive_force_delete_is_authorized, + replication_request_authorized, req_info_mut, req_info_ref, }; use super::storage_api::object_usecase::bucket::quota::checker::QuotaChecker; #[cfg(test)] @@ -81,9 +81,10 @@ use super::storage_api::object_usecase::object_cache::lookup_get_object_body_cac use super::storage_api::object_usecase::object_cache::{GetObjectBodyCacheHookLookup, get_object_body_cache_plaintext_len}; use super::storage_api::object_usecase::object_utils::to_s3s_etag; use super::storage_api::object_usecase::options::{ - copy_dst_opts, copy_src_opts, del_opts_with_versioning, extract_metadata, extract_metadata_from_mime_with_object_name, - filter_object_metadata, get_content_sha256_with_query, get_opts, namespace_reserved_user_metadata, - normalize_content_encoding_for_storage, put_opts, validate_archive_content_encoding, + copy_dst_opts_with_replication_authorization, copy_src_opts, del_opts_with_versioning, extract_metadata, + extract_metadata_from_mime_with_object_name, filter_object_metadata, get_content_sha256_with_query, get_opts, + namespace_reserved_user_metadata, normalize_content_encoding_for_storage, put_opts_with_replication_authorization, + validate_archive_content_encoding, }; use super::storage_api::object_usecase::request_context::{self, spawn_traced}; use super::storage_api::object_usecase::s3_api::multipart::parse_list_parts_params; @@ -4923,9 +4924,16 @@ impl DefaultObjectUsecase { )?; apply_bucket_default_lock_retention(&bucket, &mut metadata, has_explicit_object_lock_retention).await?; - let mut opts: ObjectOptions = put_opts(&bucket, &key, version_id.clone(), &req.headers, metadata.clone()) - .await - .map_err(ApiError::from)?; + let mut opts: ObjectOptions = put_opts_with_replication_authorization( + &bucket, + &key, + version_id.clone(), + &req.headers, + metadata.clone(), + replication_request_authorized(&req), + ) + .await + .map_err(ApiError::from)?; apply_put_request_object_lock_opts( &bucket, object_lock_legal_hold_status, @@ -6194,9 +6202,16 @@ impl DefaultObjectUsecase { ..Default::default() }; - let mut dst_opts = copy_dst_opts(&bucket, &key, dest_version_id.clone(), &req.headers, HashMap::new()) - .await - .map_err(ApiError::from)?; + let mut dst_opts = copy_dst_opts_with_replication_authorization( + &bucket, + &key, + dest_version_id.clone(), + &req.headers, + HashMap::new(), + replication_request_authorized(&req), + ) + .await + .map_err(ApiError::from)?; let cp_src_dst_same = path_join_buf(&[&src_bucket, &src_key]) == path_join_buf(&[&bucket, &key]); let expected_current_version_id = expected_current_version_id(&req.headers)?; @@ -7978,6 +7993,7 @@ impl DefaultObjectUsecase { if is_sse_kms_requested(&req.input, &req.headers) { return Err(s3_error!(NotImplemented, "SSE-KMS is not supported for extract uploads")); } + let replication_authorized = replication_request_authorized(&req); let input = req.input; let PutObjectInput { @@ -8249,9 +8265,16 @@ impl DefaultObjectUsecase { storage_class.clone(), )?; apply_bucket_default_lock_retention(&bucket, &mut metadata, has_explicit_object_lock_retention).await?; - let mut opts = put_opts(&bucket, &fpath, None, &req.headers, metadata.clone()) - .await - .map_err(ApiError::from)?; + let mut opts = put_opts_with_replication_authorization( + &bucket, + &fpath, + None, + &req.headers, + metadata.clone(), + replication_authorized, + ) + .await + .map_err(ApiError::from)?; apply_extract_entry_pax_extensions(&mut f, &mut metadata, &mut opts).await?; if archive_entry_mod_time.is_some() { opts.mod_time = archive_entry_mod_time; diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index c6df2bd75..838a02c5b 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -218,7 +218,7 @@ pub(crate) mod runtime_sources { pub(crate) mod access { pub(crate) use crate::storage::storage_api::access_consumer::{ PostObjectRequestMarker, ReqInfo, authorize_request, has_bypass_governance_header, recursive_force_delete_is_authorized, - req_info_mut, req_info_ref, + replication_request_authorized, req_info_mut, req_info_ref, }; } @@ -885,10 +885,11 @@ pub(crate) mod options { #[cfg(test)] pub(crate) use crate::storage::storage_api::options_consumer::VERSIONING_CONFIG_LOOKUPS; pub(crate) use crate::storage::storage_api::options_consumer::{ - copy_dst_opts, copy_src_opts, del_opts_with_versioning, extract_metadata, extract_metadata_from_mime, - extract_metadata_from_mime_with_object_name, filter_object_metadata, get_complete_multipart_upload_opts, - get_content_sha256_with_query, get_opts, namespace_reserved_user_metadata, normalize_content_encoding_for_storage, - parse_copy_source_range, put_opts, validate_archive_content_encoding, + copy_dst_opts_with_replication_authorization, copy_src_opts, del_opts_with_versioning, extract_metadata, + extract_metadata_from_mime, extract_metadata_from_mime_with_object_name, filter_object_metadata, + get_complete_multipart_upload_opts_with_replication_authorization, get_content_sha256_with_query, get_opts, + namespace_reserved_user_metadata, normalize_content_encoding_for_storage, parse_copy_source_range, + put_opts_with_replication_authorization, validate_archive_content_encoding, }; } diff --git a/rustfs/src/protocols/client.rs b/rustfs/src/protocols/client.rs index f7854f78e..6961172db 100644 --- a/rustfs/src/protocols/client.rs +++ b/rustfs/src/protocols/client.rs @@ -168,6 +168,7 @@ impl ProtocolStorageClient { bucket: params.bucket, object: params.object, version_id: None, + replication_request_authorized: false, region: None, request_context: Some(RequestContext::fallback()), }); diff --git a/rustfs/src/storage/access.rs b/rustfs/src/storage/access.rs index 09a51f6f0..570846091 100644 --- a/rustfs/src/storage/access.rs +++ b/rustfs/src/storage/access.rs @@ -37,7 +37,11 @@ use rustfs_policy::policy::{ bucket_policy_uses_existing_object_tag_conditions, }; use rustfs_trusted_proxies::ClientInfo; -use rustfs_utils::http::{AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_LOCK_BYPASS_GOVERNANCE, SUFFIX_FORCE_DELETE, get_header}; +use rustfs_utils::http::{ + AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_LOCK_BYPASS_GOVERNANCE, SUFFIX_FORCE_DELETE, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, + SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_CHECK, + SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_VERSION_ID, get_header, +}; use s3s::access::{S3Access, S3AccessContext}; use s3s::{S3Error, S3ErrorCode, S3Request, S3Result, dto::*, s3_error}; use std::collections::HashMap; @@ -51,11 +55,39 @@ pub(crate) struct ReqInfo { pub bucket: Option, pub object: Option, pub version_id: Option, + pub replication_request_authorized: bool, #[allow(dead_code)] pub region: Option, pub request_context: Option, } +pub(crate) fn replication_request_authorized(req: &S3Request) -> bool { + req.extensions + .get::() + .is_some_and(|req_info| req_info.replication_request_authorized) +} + +fn has_replication_only_put_headers(headers: &HeaderMap) -> bool { + headers.contains_key(AMZ_BUCKET_REPLICATION_STATUS) + || get_header(headers, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE).is_some() + || get_header(headers, SUFFIX_REPLICATION_SSEC_CRC).is_some() + || get_header(headers, SUFFIX_SOURCE_ETAG).is_some() + || get_header(headers, SUFFIX_SOURCE_MTIME).is_some() + || get_header(headers, SUFFIX_SOURCE_REPLICATION_CHECK).is_some() + || get_header(headers, SUFFIX_SOURCE_REPLICATION_REQUEST).is_some() + || get_header(headers, SUFFIX_SOURCE_VERSION_ID).is_some() +} + +async fn authorize_replication_only_put_headers(req: &mut S3Request) -> S3Result<()> { + if !has_replication_only_put_headers(&req.headers) { + return Ok(()); + } + + authorize_request(req, Action::S3Action(S3Action::ReplicateObjectAction)).await?; + req_info_mut(req)?.replication_request_authorized = true; + Ok(()) +} + pub(crate) fn recursive_force_delete_is_authorized(headers: &HeaderMap, is_owner: bool, replica_request: bool) -> bool { !get_header(headers, SUFFIX_FORCE_DELETE).is_some_and(|value| value.eq_ignore_ascii_case("true")) || is_owner @@ -1106,7 +1138,8 @@ impl S3Access for FS { req_info.bucket = Some(req.input.bucket.clone()); req_info.object = Some(req.input.key.clone()); - authorize_request(req, complete_multipart_upload_authorize_action()).await + authorize_request(req, complete_multipart_upload_authorize_action()).await?; + authorize_replication_only_put_headers(req).await } /// Checks whether the CopyObject request has accesses to the resources. @@ -1140,6 +1173,8 @@ impl S3Access for FS { authorize_request(req, Action::S3Action(S3Action::PutObjectAction)).await?; + authorize_replication_only_put_headers(req).await?; + if legal_hold_write_requested(req.input.object_lock_legal_hold_status.as_ref()) { authorize_request(req, Action::S3Action(S3Action::PutObjectLegalHoldAction)).await?; } @@ -1159,6 +1194,8 @@ impl S3Access for FS { authorize_request(req, Action::S3Action(S3Action::PutObjectAction)).await?; + authorize_replication_only_put_headers(req).await?; + if legal_hold_write_requested(req.input.object_lock_legal_hold_status.as_ref()) { authorize_request(req, Action::S3Action(S3Action::PutObjectLegalHoldAction)).await?; } @@ -2033,6 +2070,8 @@ impl S3Access for FS { return Ok(()); } + authorize_replication_only_put_headers(req).await?; + if legal_hold_write_requested(req.input.object_lock_legal_hold_status.as_ref()) { authorize_request(req, Action::S3Action(S3Action::PutObjectLegalHoldAction)).await?; } diff --git a/rustfs/src/storage/options.rs b/rustfs/src/storage/options.rs index f452a83dd..12362fa0c 100644 --- a/rustfs/src/storage/options.rs +++ b/rustfs/src/storage/options.rs @@ -154,7 +154,7 @@ pub fn del_opts_with_versioning( }; let mut opts = if replication_request_authorized { - put_opts_from_headers(headers, metadata) + put_opts_from_headers_with_replication_authorization(headers, metadata, replication_request_authorized) } else { get_default_opts(headers, metadata, false) } @@ -280,12 +280,23 @@ pub async fn put_opts( vid: Option, headers: &HeaderMap, metadata: HashMap, +) -> Result { + put_opts_with_replication_authorization(bucket, object, vid, headers, metadata, false).await +} + +pub async fn put_opts_with_replication_authorization( + bucket: &str, + object: &str, + vid: Option, + headers: &HeaderMap, + metadata: HashMap, + replication_request_authorized: bool, ) -> Result { let versioning_cfg = bucket_versioning_config(bucket).await; let versioned = versioning_cfg.prefix_enabled(object); let version_suspended = versioning_cfg.prefix_suspended(object); - let vid = if vid.is_none() { + let vid = if vid.is_none() && replication_request_authorized { get_header(headers, SUFFIX_SOURCE_VERSION_ID).map(|s| s.into_owned()) } else { vid @@ -300,7 +311,7 @@ pub async fn put_opts( return Err(StorageError::InvalidVersionID(bucket.to_owned(), object.to_owned(), id.clone())); } - let mut opts = put_opts_from_headers(headers, metadata) + let mut opts = put_opts_from_headers_with_replication_authorization(headers, metadata, replication_request_authorized) .map_err(|err| StorageError::InvalidArgument(bucket.to_owned(), object.to_owned(), err.to_string()))?; opts.version_id = { @@ -319,10 +330,17 @@ pub async fn put_opts( } pub fn get_complete_multipart_upload_opts(headers: &HeaderMap) -> std::io::Result { + get_complete_multipart_upload_opts_with_replication_authorization(headers, false) +} + +pub fn get_complete_multipart_upload_opts_with_replication_authorization( + headers: &HeaderMap, + replication_request_authorized: bool, +) -> std::io::Result { let mut user_defined = HashMap::new(); let mut replication_request = false; - if get_header(headers, SUFFIX_SOURCE_REPLICATION_REQUEST).as_deref() == Some("true") { + if replication_request_authorized && get_header(headers, SUFFIX_SOURCE_REPLICATION_REQUEST).as_deref() == Some("true") { replication_request = true; if let Some(actual_size_str) = get_header(headers, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE) { rustfs_utils::http::insert_str( @@ -335,7 +353,7 @@ pub fn get_complete_multipart_upload_opts(headers: &HeaderMap) -> s } } - if let Some(v) = get_header(headers, SUFFIX_REPLICATION_SSEC_CRC) { + if replication_request_authorized && let Some(v) = get_header(headers, SUFFIX_REPLICATION_SSEC_CRC) { insert_header_map(&mut user_defined, SUFFIX_REPLICATION_SSEC_CRC, v.into_owned()); } @@ -345,7 +363,7 @@ pub fn get_complete_multipart_upload_opts(headers: &HeaderMap) -> s replication_request, ..Default::default() }; - apply_replica_status_from_headers(headers, &mut opts); + apply_replica_status_from_headers(headers, &mut opts, replication_request_authorized); fill_conditional_writes_opts_from_header(headers, &mut opts)?; Ok(opts) @@ -359,7 +377,18 @@ pub async fn copy_dst_opts( headers: &HeaderMap, metadata: HashMap, ) -> Result { - put_opts(bucket, object, vid, headers, metadata).await + copy_dst_opts_with_replication_authorization(bucket, object, vid, headers, metadata, false).await +} + +pub async fn copy_dst_opts_with_replication_authorization( + bucket: &str, + object: &str, + vid: Option, + headers: &HeaderMap, + metadata: HashMap, + replication_request_authorized: bool, +) -> Result { + put_opts_with_replication_authorization(bucket, object, vid, headers, metadata, replication_request_authorized).await } pub fn copy_src_opts(_bucket: &str, _object: &str, headers: &HeaderMap) -> Result { @@ -367,9 +396,17 @@ pub fn copy_src_opts(_bucket: &str, _object: &str, headers: &HeaderMap, metadata: HashMap) -> Result { + put_opts_from_headers_with_replication_authorization(headers, metadata, false) +} + +pub fn put_opts_from_headers_with_replication_authorization( + headers: &HeaderMap, + metadata: HashMap, + replication_request_authorized: bool, +) -> Result { let mut opts = get_default_opts(headers, metadata, false)?; - apply_replica_status_from_headers(headers, &mut opts); - if get_header(headers, SUFFIX_SOURCE_REPLICATION_REQUEST).as_deref() == Some("true") { + apply_replica_status_from_headers(headers, &mut opts, replication_request_authorized); + if replication_request_authorized && get_header(headers, SUFFIX_SOURCE_REPLICATION_REQUEST).as_deref() == Some("true") { opts.replication_request = true; if let Some(v) = get_header(headers, SUFFIX_SOURCE_MTIME) { let trimmed_s = v.trim(); @@ -385,7 +422,11 @@ pub fn put_opts_from_headers(headers: &HeaderMap, metadata: HashMap Ok(opts) } -fn apply_replica_status_from_headers(headers: &HeaderMap, opts: &mut ObjectOptions) { +fn apply_replica_status_from_headers(headers: &HeaderMap, opts: &mut ObjectOptions, authorized: bool) { + if !authorized { + return; + } + if headers .get(AMZ_BUCKET_REPLICATION_STATUS) .and_then(|value| value.to_str().ok()) @@ -928,7 +969,9 @@ mod tests { ENV_REJECT_ARCHIVE_CONTENT_ENCODING, ReplicationStatusType, SUPPORTED_HEADERS, copy_dst_opts, copy_src_opts, del_opts, del_opts_with_versioning, detect_content_type_from_object_name, extract_metadata, extract_metadata_from_mime, extract_metadata_from_mime_with_object_name, filter_object_metadata, get_complete_multipart_upload_opts, - get_default_opts, get_opts, namespace_reserved_user_metadata, parse_copy_source_range, put_opts, put_opts_from_headers, + get_complete_multipart_upload_opts_with_replication_authorization, get_default_opts, get_opts, + namespace_reserved_user_metadata, parse_copy_source_range, put_opts, put_opts_from_headers, + put_opts_from_headers_with_replication_authorization, put_opts_with_replication_authorization, validate_archive_content_encoding, }; use http::{HeaderMap, HeaderValue}; @@ -1330,7 +1373,7 @@ mod tests { } #[test] - fn test_put_opts_from_headers_with_replication_request() { + fn test_put_opts_from_headers_ignores_replication_request_without_authorization() { let mut headers = HeaderMap::new(); insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true"); let valid_mtime = "2024-05-20T10:30:00+08:00"; @@ -1343,6 +1386,20 @@ mod tests { assert!(result.is_ok()); let opts = result.unwrap(); + assert!(!opts.replication_request); + assert!(opts.mod_time.is_none()); + } + + #[test] + fn test_put_opts_from_headers_accepts_replication_request_after_authorization() { + let mut headers = HeaderMap::new(); + insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true"); + let valid_mtime = "2024-05-20T10:30:00+08:00"; + insert_header(&mut headers, SUFFIX_SOURCE_MTIME, valid_mtime); + + let opts = put_opts_from_headers_with_replication_authorization(&headers, HashMap::new(), true) + .expect("authorized replication request should parse"); + assert!(opts.replication_request); let expected_mtime = time::OffsetDateTime::parse(valid_mtime, &time::format_description::well_known::Rfc3339).unwrap(); @@ -1351,7 +1408,7 @@ mod tests { let mut headers_invalid_mtime = HeaderMap::new(); insert_header(&mut headers_invalid_mtime, SUFFIX_SOURCE_REPLICATION_REQUEST, "true"); insert_header(&mut headers_invalid_mtime, SUFFIX_SOURCE_MTIME, "invalid-time"); - let result_invalid = put_opts_from_headers(&headers_invalid_mtime, HashMap::new()); + let result_invalid = put_opts_from_headers_with_replication_authorization(&headers_invalid_mtime, HashMap::new(), true); assert!(result_invalid.is_ok()); let opts_invalid = result_invalid.unwrap(); assert!(opts_invalid.replication_request); @@ -1366,9 +1423,39 @@ mod tests { HeaderValue::from_static(ReplicationStatusType::Replica.as_str()), ); - let opts = put_opts_from_headers(&headers, HashMap::new()).expect("replica status header should parse"); + let opts = put_opts_from_headers(&headers, HashMap::new()).expect("replica status header should be ignored"); - assert_eq!(opts.delete_marker_replication_status(), ReplicationStatusType::Replica); + assert_eq!(opts.delete_marker_replication_status(), ReplicationStatusType::Empty); + + let authorized = put_opts_from_headers_with_replication_authorization(&headers, HashMap::new(), true) + .expect("authorized replica status header should parse"); + assert_eq!(authorized.delete_marker_replication_status(), ReplicationStatusType::Replica); + } + + #[tokio::test] + async fn test_put_opts_only_accepts_replication_source_headers_after_authorization() { + let source_version_id = Uuid::new_v4().to_string(); + let mut headers = HeaderMap::new(); + insert_header(&mut headers, SUFFIX_SOURCE_VERSION_ID, source_version_id.clone()); + insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true"); + headers.insert( + AMZ_BUCKET_REPLICATION_STATUS, + HeaderValue::from_static(ReplicationStatusType::Replica.as_str()), + ); + + let untrusted = put_opts("test-bucket", "test-object", None, &headers, HashMap::new()) + .await + .expect("ordinary PUT options should be created"); + assert_eq!(untrusted.version_id, None); + assert!(!untrusted.replication_request); + assert_eq!(untrusted.delete_marker_replication_status(), ReplicationStatusType::Empty); + + let trusted = put_opts_with_replication_authorization("test-bucket", "test-object", None, &headers, HashMap::new(), true) + .await + .expect("authorized replication PUT options should be created"); + assert_eq!(trusted.version_id.as_deref(), Some(source_version_id.as_str())); + assert!(trusted.replication_request); + assert_eq!(trusted.delete_marker_replication_status(), ReplicationStatusType::Replica); } #[test] @@ -1379,9 +1466,13 @@ mod tests { HeaderValue::from_static(ReplicationStatusType::Replica.as_str()), ); - let opts = get_complete_multipart_upload_opts(&headers).expect("replica status header should parse"); + let opts = get_complete_multipart_upload_opts(&headers).expect("replica status header should be ignored"); - assert_eq!(opts.delete_marker_replication_status(), ReplicationStatusType::Replica); + assert_eq!(opts.delete_marker_replication_status(), ReplicationStatusType::Empty); + + let authorized = get_complete_multipart_upload_opts_with_replication_authorization(&headers, true) + .expect("authorized replica status header should parse"); + assert_eq!(authorized.delete_marker_replication_status(), ReplicationStatusType::Replica); } #[test] diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index b5cd55ace..fa7a4e9b5 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -103,7 +103,7 @@ pub(crate) use super::sse::{ pub(crate) mod access_consumer { pub(crate) use super::super::access::{ PostObjectRequestMarker, ReqInfo, authorize_request, has_bypass_governance_header, recursive_force_delete_is_authorized, - req_info_mut, req_info_ref, + replication_request_authorized, req_info_mut, req_info_ref, }; pub(crate) mod contract { @@ -180,10 +180,11 @@ pub(crate) mod options_consumer { #[cfg(test)] pub(crate) use super::super::options::VERSIONING_CONFIG_LOOKUPS; pub(crate) use super::super::options::{ - copy_dst_opts, copy_src_opts, del_opts_with_versioning, extract_metadata, extract_metadata_from_mime, - extract_metadata_from_mime_with_object_name, filter_object_metadata, get_complete_multipart_upload_opts, - get_content_sha256_with_query, get_opts, namespace_reserved_user_metadata, normalize_content_encoding_for_storage, - parse_copy_source_range, put_opts, validate_archive_content_encoding, + copy_dst_opts_with_replication_authorization, copy_src_opts, del_opts_with_versioning, extract_metadata, + extract_metadata_from_mime, extract_metadata_from_mime_with_object_name, filter_object_metadata, + get_complete_multipart_upload_opts_with_replication_authorization, get_content_sha256_with_query, get_opts, + namespace_reserved_user_metadata, normalize_content_encoding_for_storage, parse_copy_source_range, + put_opts_with_replication_authorization, validate_archive_content_encoding, }; pub(crate) mod contract {