gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-02 19:39:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(authz): gate replication-only PUT headers (#5625)

Browse Source
This commit is contained in:
cxymds
2026-08-02 19:28:54 +08:00
committed by GitHub
parent 885096d1de
commit 2cdba03dee
8 changed files with 213 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/handlers/object_zip_download.rs
+2
View File
@@ -206,6 +206,7 @@ impl From<ObjectZipDownloadReqInfoSnapshot> 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,
}
rustfs/src/app/multipart_usecase.rs
+12 -7
View File
@@ -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())
rustfs/src/app/object_usecase.rs
+37 -14
View File
@@ -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;
rustfs/src/app/storage_api.rs
+6 -5
View File
@@ -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,
};
}
rustfs/src/protocols/client.rs
+1
View File
@@ -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()),
});
rustfs/src/storage/access.rs
+41 -2
View File
@@ -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<String>,
pub object: Option<String>,
pub version_id: Option<String>,
pub replication_request_authorized: bool,
#[allow(dead_code)]
pub region: Option<s3s::region::Region>,
pub request_context: Option<RequestContext>,
}
pub(crate) fn replication_request_authorized<T>(req: &S3Request<T>) -> bool {
req.extensions
.get::<ReqInfo>()
.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<T>(req: &mut S3Request<T>) -> 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?;
}
rustfs/src/storage/options.rs
+108 -17
View File
@@ -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<String>,
headers: &HeaderMap<HeaderValue>,
metadata: HashMap<String, String>,
) -> Result<ObjectOptions> {
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<String>,
headers: &HeaderMap<HeaderValue>,
metadata: HashMap<String, String>,
replication_request_authorized: bool,
) -> Result<ObjectOptions> {
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<HeaderValue>) -> std::io::Result<ObjectOptions> {
get_complete_multipart_upload_opts_with_replication_authorization(headers, false)
}
pub fn get_complete_multipart_upload_opts_with_replication_authorization(
headers: &HeaderMap<HeaderValue>,
replication_request_authorized: bool,
) -> std::io::Result<ObjectOptions> {
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<HeaderValue>) -> 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<HeaderValue>) -> 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<HeaderValue>,
metadata: HashMap<String, String>,
) -> Result<ObjectOptions> {
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<String>,
headers: &HeaderMap<HeaderValue>,
metadata: HashMap<String, String>,
replication_request_authorized: bool,
) -> Result<ObjectOptions> {
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<HeaderValue>) -> Result<ObjectOptions> {
@@ -367,9 +396,17 @@ pub fn copy_src_opts(_bucket: &str, _object: &str, headers: &HeaderMap<HeaderVal
}
pub fn put_opts_from_headers(headers: &HeaderMap<HeaderValue>, metadata: HashMap<String, String>) -> Result<ObjectOptions> {
put_opts_from_headers_with_replication_authorization(headers, metadata, false)
}
pub fn put_opts_from_headers_with_replication_authorization(
headers: &HeaderMap<HeaderValue>,
metadata: HashMap<String, String>,
replication_request_authorized: bool,
) -> Result<ObjectOptions> {
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<HeaderValue>, metadata: HashMap
Ok(opts)
}
fn apply_replica_status_from_headers(headers: &HeaderMap<HeaderValue>, opts: &mut ObjectOptions) {
fn apply_replica_status_from_headers(headers: &HeaderMap<HeaderValue>, 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]
rustfs/src/storage/storage_api.rs
+6 -5
View File
@@ -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 {