From a2fc6e15df4208d2e410426923bd0711205b62b1 Mon Sep 17 00:00:00 2001 From: cxymds Date: Fri, 24 Jul 2026 10:30:43 +0800 Subject: [PATCH] fix(s3): honor CopyObject tagging directives (#5165) Closes rustfs/backlog#1462. --- .config/nextest.toml | 2 +- Cargo.lock | 1 + .../e2e_test/src/copy_object_tagging_test.rs | 468 ++++++++++++++++++ crates/e2e_test/src/lib.rs | 3 + docs/testing/e2e-suite-inventory.md | 3 +- rustfs/Cargo.toml | 1 + rustfs/src/app/object_usecase.rs | 17 +- rustfs/src/app/storage_api.rs | 4 + rustfs/src/storage/access.rs | 107 +++- rustfs/src/storage/s3_api/tagging.rs | 224 ++++++++- rustfs/src/storage/storage_api.rs | 4 + 11 files changed, 797 insertions(+), 37 deletions(-) create mode 100644 crates/e2e_test/src/copy_object_tagging_test.rs diff --git a/.config/nextest.toml b/.config/nextest.toml index 03529d03e..84e6107ca 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -192,7 +192,7 @@ test-group = 'ecstore-serial-flaky' [profile.e2e-smoke] default-filter = """ package(e2e_test) & ( - test(/^(delete_marker_migration_semantics|version_id_regression|list_objects_v2_pagination|list_object_versions_regression|list_objects_duplicates|list_buckets_double_slash|leading_slash_key|special_chars|create_bucket_region|delete_objects_versioning|head_object_consistency|head_object_range|copy_object_metadata|copy_source_invalid_date|content_encoding|anonymous_access|bucket_policy_check|presigned_negative|negative_sigv4|admin_auth|notification_webhook|tls_hot_reload|console_smoke|admin_iam_crud|admin_pools)_test::|^fake_s3_target::/) + test(/^(delete_marker_migration_semantics|version_id_regression|list_objects_v2_pagination|list_object_versions_regression|list_objects_duplicates|list_buckets_double_slash|leading_slash_key|special_chars|create_bucket_region|delete_objects_versioning|head_object_consistency|head_object_range|copy_object_metadata|copy_object_tagging|copy_source_invalid_date|content_encoding|anonymous_access|bucket_policy_check|presigned_negative|negative_sigv4|admin_auth|notification_webhook|tls_hot_reload|console_smoke|admin_iam_crud|admin_pools)_test::|^fake_s3_target::/) | test(/^replication_extension_test::(test_replication_check_succeeds_with_remote_target|test_replication_check_rejects_target_without_object_lock|test_set_remote_target_rejects_unversioned_source_bucket|test_replication_check_rejects_unversioned_source_bucket|test_replication_check_rejects_missing_replication_config|test_replication_check_rejects_invalid_bucket|test_set_remote_target_rejects_same_bucket_on_same_deployment|test_set_remote_target_rejects_unversioned_target_bucket|test_set_remote_target_update_requires_arn|test_set_remote_target_update_rejects_missing_target|test_set_remote_target_rejects_invalid_target_url|test_set_remote_target_rejects_self_signed_https_target_without_skip_tls_verify|test_set_remote_target_rejects_private_ca_https_target_without_ca_cert_pem|test_list_remote_targets_rejects_empty_bucket|test_list_remote_targets_rejects_invalid_bucket|test_remove_remote_target_rejects_missing_target|test_remove_remote_target_rejects_missing_arn|test_remove_remote_target_rejects_invalid_bucket|test_remove_remote_target_rejects_target_used_by_replication|test_delete_bucket_replication_removes_remote_target)$/) | test(/^reliant::lifecycle::/) | test(/^reliant::tiering::/) diff --git a/Cargo.lock b/Cargo.lock index 6a7966d7a..9dbd02202 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8885,6 +8885,7 @@ dependencies = [ "quick-xml", "rand 0.10.2", "rcgen", + "regex", "reqwest", "rmp-serde", "rsa 0.10.0-rc.18", diff --git a/crates/e2e_test/src/copy_object_tagging_test.rs b/crates/e2e_test/src/copy_object_tagging_test.rs new file mode 100644 index 000000000..23861a705 --- /dev/null +++ b/crates/e2e_test/src/copy_object_tagging_test.rs @@ -0,0 +1,468 @@ +// 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. + +//! CopyObject tagging directive regression tests. + +#[cfg(test)] +mod tests { + use crate::common::{RustFSTestEnvironment, init_logging, local_http_client}; + use aws_sdk_s3::Client; + use aws_sdk_s3::error::ProvideErrorMetadata; + use aws_sdk_s3::primitives::ByteStream; + use aws_sdk_s3::types::{BucketVersioningStatus, MetadataDirective, TaggingDirective, VersioningConfiguration}; + use serial_test::serial; + use std::collections::BTreeMap; + + async fn object_tags(client: &Client, bucket: &str, key: &str) -> BTreeMap { + client + .get_object_tagging() + .bucket(bucket) + .key(key) + .send() + .await + .expect("GetObjectTagging should succeed") + .tag_set() + .iter() + .map(|tag| (tag.key().to_string(), tag.value().to_string())) + .collect() + } + + #[tokio::test] + #[serial] + async fn copy_object_applies_copy_replace_and_empty_tagging_directives() { + init_logging(); + let mut env = RustFSTestEnvironment::new() + .await + .expect("test environment should initialize"); + env.start_rustfs_server(vec![]).await.expect("RustFS should start"); + + let client = env.create_s3_client(); + let bucket = "copy-object-tagging-directive"; + let source = "source.txt"; + + client + .create_bucket() + .bucket(bucket) + .send() + .await + .expect("bucket creation should succeed"); + client + .put_bucket_versioning() + .bucket(bucket) + .versioning_configuration( + VersioningConfiguration::builder() + .status(BucketVersioningStatus::Enabled) + .build(), + ) + .send() + .await + .expect("versioning should be enabled"); + + let first_version = client + .put_object() + .bucket(bucket) + .key(source) + .tagging("project=rustfs&stage=first") + .body(ByteStream::from_static(b"first")) + .send() + .await + .expect("first source version should be written") + .version_id() + .expect("versioned PUT should return a version ID") + .to_string(); + + client + .put_object() + .bucket(bucket) + .key(source) + .tagging("project=rustfs&stage=current") + .body(ByteStream::from_static(b"current")) + .send() + .await + .expect("current source version should be written"); + + client + .copy_object() + .bucket(bucket) + .key("default-copy.txt") + .copy_source(format!("{bucket}/{source}")) + .send() + .await + .expect("default CopyObject should preserve current source tags"); + assert_eq!( + object_tags(&client, bucket, "default-copy.txt").await, + BTreeMap::from([ + ("project".to_string(), "rustfs".to_string()), + ("stage".to_string(), "current".to_string()), + ]) + ); + + client + .copy_object() + .bucket(bucket) + .key("explicit-copy.txt") + .copy_source(format!("{bucket}/{source}?versionId={first_version}")) + .tagging_directive(TaggingDirective::Copy) + .send() + .await + .expect("COPY should preserve the selected historical version's tags"); + assert_eq!( + object_tags(&client, bucket, "explicit-copy.txt").await, + BTreeMap::from([ + ("project".to_string(), "rustfs".to_string()), + ("stage".to_string(), "first".to_string()), + ]) + ); + + client + .copy_object() + .bucket(bucket) + .key("replace.txt") + .copy_source(format!("{bucket}/{source}")) + .tagging_directive(TaggingDirective::Replace) + .tagging("project=cli&label=copy%20test") + .send() + .await + .expect("REPLACE should atomically apply requested tags"); + assert_eq!( + object_tags(&client, bucket, "replace.txt").await, + BTreeMap::from([ + ("label".to_string(), "copy test".to_string()), + ("project".to_string(), "cli".to_string()), + ]) + ); + let replace_head = client + .head_object() + .bucket(bucket) + .key("replace.txt") + .send() + .await + .expect("HEAD should succeed after tag replacement"); + assert_eq!(replace_head.tag_count(), Some(2)); + + client + .copy_object() + .bucket(bucket) + .key("empty-replace.txt") + .copy_source(format!("{bucket}/{source}")) + .tagging_directive(TaggingDirective::Replace) + .send() + .await + .expect("REPLACE without Tagging should clear the destination tag set"); + assert!(object_tags(&client, bucket, "empty-replace.txt").await.is_empty()); + let empty_head = client + .head_object() + .bucket(bucket) + .key("empty-replace.txt") + .send() + .await + .expect("HEAD should succeed after empty tag replacement"); + assert_eq!(empty_head.tag_count(), None); + + client + .copy_object() + .bucket(bucket) + .key("metadata-replace-tag-copy.txt") + .copy_source(format!("{bucket}/{source}")) + .metadata_directive(MetadataDirective::Replace) + .metadata("updated", "true") + .send() + .await + .expect("metadata REPLACE must preserve tags under the default COPY directive"); + assert_eq!( + object_tags(&client, bucket, "metadata-replace-tag-copy.txt").await, + BTreeMap::from([ + ("project".to_string(), "rustfs".to_string()), + ("stage".to_string(), "current".to_string()), + ]) + ); + + client + .copy_object() + .bucket(bucket) + .key("combined-replace.txt") + .copy_source(format!("{bucket}/{source}")) + .metadata_directive(MetadataDirective::Replace) + .metadata("updated", "true") + .tagging_directive(TaggingDirective::Replace) + .tagging("project=combined") + .send() + .await + .expect("metadata and tagging REPLACE directives must be independent"); + assert_eq!( + object_tags(&client, bucket, "combined-replace.txt").await, + BTreeMap::from([("project".to_string(), "combined".to_string())]) + ); + + client + .copy_object() + .bucket(bucket) + .key(source) + .copy_source(format!("{bucket}/{source}")) + .tagging_directive(TaggingDirective::Replace) + .tagging("project=self-copy") + .send() + .await + .expect("self-copy with tag replacement should update tags atomically"); + assert_eq!( + object_tags(&client, bucket, source).await, + BTreeMap::from([("project".to_string(), "self-copy".to_string())]) + ); + let self_copy_body = client + .get_object() + .bucket(bucket) + .key(source) + .send() + .await + .expect("self-copy destination should remain readable") + .body + .collect() + .await + .expect("self-copy body should be complete") + .into_bytes(); + assert_eq!(self_copy_body.as_ref(), b"current", "tag-only self-copy must preserve the object body"); + + client + .put_object() + .bucket(bucket) + .key("malformed.txt") + .tagging("state=original") + .body(ByteStream::from_static(b"original destination")) + .send() + .await + .expect("preexisting malformed-test destination should be written"); + + let malformed = client + .copy_object() + .bucket(bucket) + .key("malformed.txt") + .copy_source(format!("{bucket}/{source}")) + .tagging_directive(TaggingDirective::Replace) + .tagging("project=rustfs%ZZ") + .send() + .await + .expect_err("malformed tags must fail CopyObject"); + assert_eq!(malformed.as_service_error().and_then(ProvideErrorMetadata::code), Some("InvalidTag")); + assert_eq!( + object_tags(&client, bucket, "malformed.txt").await, + BTreeMap::from([("state".to_string(), "original".to_string())]) + ); + let preserved_body = client + .get_object() + .bucket(bucket) + .key("malformed.txt") + .send() + .await + .expect("malformed tags must not replace an existing destination") + .body + .collect() + .await + .expect("preserved destination body should be readable") + .into_bytes(); + assert_eq!( + preserved_body.as_ref(), + b"original destination", + "malformed tags must leave destination data unchanged" + ); + + let discarded = client + .copy_object() + .bucket(bucket) + .key("discarded.txt") + .copy_source(format!("{bucket}/{source}")) + .tagging("project=must-not-be-discarded") + .send() + .await + .expect_err("Tagging without REPLACE must fail instead of discarding requested tags"); + assert_eq!(discarded.as_service_error().and_then(ProvideErrorMetadata::code), Some("InvalidRequest")); + + let invalid_directive = client + .copy_object() + .bucket(bucket) + .key("invalid-directive.txt") + .copy_source(format!("{bucket}/{source}")) + .tagging_directive(TaggingDirective::from("UNKNOWN")) + .send() + .await + .expect_err("an unknown TaggingDirective must fail"); + assert_eq!( + invalid_directive.as_service_error().and_then(ProvideErrorMetadata::code), + Some("InvalidArgument") + ); + + env.stop_server(); + } + + #[tokio::test] + #[serial] + async fn copy_object_tag_replacement_honors_request_tag_policy_denial() -> Result<(), Box> + { + init_logging(); + let source_bucket = "copy-tags-policy-source"; + let destination_bucket = "copy-tags-policy-destination"; + + let mut env = RustFSTestEnvironment::new().await?; + env.start_rustfs_server(vec![]).await?; + let admin = env.create_s3_client(); + admin.create_bucket().bucket(source_bucket).send().await?; + admin.create_bucket().bucket(destination_bucket).send().await?; + admin + .put_object() + .bucket(source_bucket) + .key("source.txt") + .tagging("source=allowed") + .body(ByteStream::from_static(b"source")) + .send() + .await?; + admin + .put_object() + .bucket(source_bucket) + .key("conditioned.txt") + .body(ByteStream::from_static(b"conditioned source")) + .send() + .await?; + + let source_policy = serde_json::json!({ + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Principal": "*", + "Action": ["s3:GetObject"], + "Resource": [format!("arn:aws:s3:::{source_bucket}/source.txt")] + }, + { + "Effect": "Allow", + "Principal": "*", + "Action": ["s3:GetObject"], + "Resource": [format!("arn:aws:s3:::{source_bucket}/conditioned.txt")], + "Condition": { + "StringEquals": { + "s3:RequestObjectTag/classification": "public" + } + } + } + ] + }) + .to_string(); + admin + .put_bucket_policy() + .bucket(source_bucket) + .policy(source_policy) + .send() + .await?; + + let destination_policy = serde_json::json!({ + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Principal": "*", + "Action": ["s3:PutObject"], + "Resource": [format!("arn:aws:s3:::{destination_bucket}/*")] + }, + { + "Effect": "Deny", + "Principal": "*", + "Action": ["s3:PutObject"], + "Resource": [format!("arn:aws:s3:::{destination_bucket}/*")], + "Condition": { + "StringEquals": { + "s3:RequestObjectTag/classification": "restricted" + } + } + } + ] + }) + .to_string(); + admin + .put_bucket_policy() + .bucket(destination_bucket) + .policy(destination_policy) + .send() + .await?; + + let copy_source = format!("/{source_bucket}/source.txt"); + let allowed = local_http_client() + .put(format!("{}/{destination_bucket}/allowed.txt", env.url)) + .header("x-amz-copy-source", ©_source) + .header("x-amz-tagging-directive", "REPLACE") + .header("x-amz-tagging", "classification=public") + .send() + .await?; + assert_eq!( + allowed.status(), + reqwest::StatusCode::OK, + "a tag set allowed by the request-tag policy should copy successfully" + ); + assert_eq!( + object_tags(&admin, destination_bucket, "allowed.txt").await, + BTreeMap::from([("classification".to_string(), "public".to_string())]) + ); + + let denied = local_http_client() + .put(format!("{}/{destination_bucket}/denied.txt", env.url)) + .header("x-amz-copy-source", copy_source) + .header("x-amz-tagging-directive", "REPLACE") + .header("x-amz-tagging", "classification=restricted") + .send() + .await?; + assert_eq!( + denied.status(), + reqwest::StatusCode::FORBIDDEN, + "CopyObject must honor a request-tag policy Deny" + ); + + let source_condition_bypass = local_http_client() + .put(format!("{}/{destination_bucket}/source-condition.txt", env.url)) + .header("x-amz-copy-source", format!("/{source_bucket}/conditioned.txt")) + .header("x-amz-tagging-directive", "REPLACE") + .header("x-amz-tagging", "classification=public") + .send() + .await?; + assert_eq!( + source_condition_bypass.status(), + reqwest::StatusCode::FORBIDDEN, + "destination request tags must not satisfy source GetObject policy conditions" + ); + + let missing_destination = admin + .head_object() + .bucket(destination_bucket) + .key("denied.txt") + .send() + .await + .expect_err("an access-denied copy must not create a destination object"); + assert_eq!( + missing_destination.as_service_error().and_then(ProvideErrorMetadata::code), + Some("NotFound") + ); + let missing_bypass_destination = admin + .head_object() + .bucket(destination_bucket) + .key("source-condition.txt") + .send() + .await + .expect_err("a source authorization denial must not create a destination object"); + assert_eq!( + missing_bypass_destination + .as_service_error() + .and_then(ProvideErrorMetadata::code), + Some("NotFound") + ); + + env.stop_server(); + Ok(()) + } +} diff --git a/crates/e2e_test/src/lib.rs b/crates/e2e_test/src/lib.rs index 8363305ba..5e8c9043d 100644 --- a/crates/e2e_test/src/lib.rs +++ b/crates/e2e_test/src/lib.rs @@ -187,6 +187,9 @@ mod heal_erasure_disk_rebuild_test; #[cfg(test)] mod copy_object_metadata_test; +#[cfg(test)] +mod copy_object_tagging_test; + #[cfg(test)] mod copy_object_version_restore_test; diff --git a/docs/testing/e2e-suite-inventory.md b/docs/testing/e2e-suite-inventory.md index d72e4b329..e3d06db72 100644 --- a/docs/testing/e2e-suite-inventory.md +++ b/docs/testing/e2e-suite-inventory.md @@ -36,6 +36,7 @@ | content_encoding_test | 3 | ✅ | | copy_object_checksum_test | 3 | | | copy_object_metadata_test | 1 | ✅ | +| copy_object_tagging_test | 2 | ✅ | | copy_object_version_restore_test | 2 | | | copy_source_invalid_date_test | 1 | ✅ | | create_bucket_region_test | 2 | ✅ | @@ -83,4 +84,4 @@ `notification_webhook_test` also has 1 ignored store-and-forward regression tracked by rustfs#4852; ignored tests are excluded from the active counts above. -**Total listed: 467 tests across 63 modules · PR smoke subset: 114 tests / 28 modules** (26 full modules + 4 `reliant` tests + 20 of `replication_extension_test`) **· nightly `e2e-repl-nightly`: 27 tests** · generated 2026-07-21. +**Total listed: 469 tests across 64 modules · PR smoke subset: 116 tests / 29 modules** (27 full modules + 4 `reliant` tests + 20 of `replication_extension_test`) **· nightly `e2e-repl-nightly`: 27 tests** · generated 2026-07-23. diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index cd1570eb0..8f2a50a99 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -179,6 +179,7 @@ uuid = { workspace = true, features = ["v4", "fast-rng", "macro-diagnostics"] } zip = { workspace = true } libc = { workspace = true } rand = { workspace = true, features = ["serde"] } +regex = { workspace = true } aes-gcm = { workspace = true, features = ["rand_core"] } chacha20poly1305 = { workspace = true } diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 6d3fc37d2..dd7aad37c 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -160,7 +160,7 @@ use s3s::dto::{ ObjectLockLegalHoldStatus, ObjectLockMode, ObjectLockRetention, ObjectLockRetentionMode, ObjectPart, PutObjectInput, PutObjectOutput, Range, RequestCharged, RestoreObjectInput, RestoreObjectOutput, RestoreStatus, SSECustomerAlgorithm, SSECustomerKeyMD5, SSEKMSKeyId, SelectObjectContentInput, SelectObjectContentOutput, ServerSideEncryption, StorageClass, - StreamingBlob, TaggingHeader, Timestamp, TimestampFormat, WebsiteRedirectLocation, + StreamingBlob, TaggingDirective, TaggingHeader, Timestamp, TimestampFormat, WebsiteRedirectLocation, }; use s3s::header::{X_AMZ_RESTORE, X_AMZ_RESTORE_OUTPUT_PATH}; use s3s::stream::{ByteStream, DynByteStream, RemainingLength}; @@ -5736,6 +5736,8 @@ impl DefaultObjectUsecase { copy_source_sse_customer_key_md5, metadata_directive, metadata, + tagging, + tagging_directive, copy_source_if_match, copy_source_if_none_match, content_type, @@ -5787,7 +5789,13 @@ impl DefaultObjectUsecase { // when an explicit storage class change is requested, or when restoring a specific historical // version onto the current key (source carries a versionId). Reject only a true no-op self-copy // where none of these apply (issue #4238). + let replacement_tags = super::storage_api::object_usecase::s3_api::tagging::resolve_copy_object_tags( + tagging.as_deref(), + tagging_directive.as_ref(), + )?; + if metadata_directive.as_ref().map(|d| d.as_str()) != Some(MetadataDirective::REPLACE) + && tagging_directive.as_ref().map(TaggingDirective::as_str) != Some(TaggingDirective::REPLACE) && storage_class.is_none() && version_id.is_none() && src_bucket == bucket @@ -5940,6 +5948,7 @@ impl DefaultObjectUsecase { // Extract user_defined from Arc for mutation; it will be re-wrapped after all edits. let mut user_defined = (*src_info.user_defined).clone(); + let effective_tags = replacement_tags.unwrap_or_else(|| (*src_info.user_tags).clone()); strip_managed_encryption_metadata(&mut user_defined); @@ -5983,6 +5992,12 @@ impl DefaultObjectUsecase { } } + user_defined.retain(|key, _| !key.eq_ignore_ascii_case(AMZ_OBJECT_TAGGING)); + if !effective_tags.is_empty() { + user_defined.insert(AMZ_OBJECT_TAGGING.to_string(), effective_tags.clone()); + } + src_info.user_tags = Arc::new(effective_tags); + let has_explicit_object_lock_retention = object_lock_mode.is_some() || object_lock_retain_until_date.is_some(); remove_object_lock_metadata_for_copy(&mut user_defined); if let Some(object_lock_metadata) = build_put_like_object_lock_metadata( diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index d3b0b426d..73fadc23c 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -928,6 +928,10 @@ pub(crate) mod s3_api { parse_list_multipart_uploads_params, parse_list_parts_params, parse_upload_part_number, }; } + + pub(crate) mod tagging { + pub(crate) use crate::storage::storage_api::s3_api_consumer::tagging::resolve_copy_object_tags; + } } pub(crate) mod admin_usecase { diff --git a/rustfs/src/storage/access.rs b/rustfs/src/storage/access.rs index 7f7149fff..8689bb105 100644 --- a/rustfs/src/storage/access.rs +++ b/rustfs/src/storage/access.rs @@ -24,6 +24,7 @@ use crate::server::RemoteAddr; use crate::storage::request_context::RequestContext; use crate::storage::storage_api::access_consumer::contract::bucket::BucketOperations; use crate::storage::storage_api::runtime_sources_consumer::runtime_sources; +use http::HeaderMap; use metrics::counter; use rustfs_iam::error::Error as IamError; use rustfs_policy::policy::action::{Action, AdminAction, S3Action}; @@ -246,6 +247,57 @@ fn merge_list_bucket_query_conditions(action: Action, query: Option<&str>, condi } } +fn merge_request_object_tag_conditions( + action: Action, + headers: &HeaderMap, + conditions: &mut HashMap>, +) -> S3Result<()> { + if action != Action::S3Action(S3Action::PutObjectAction) { + return Ok(()); + } + + let Some(tagging) = headers.get("x-amz-tagging").and_then(|value| value.to_str().ok()) else { + return Ok(()); + }; + + let mut tag_keys = Vec::new(); + for tag in crate::storage::s3_api::tagging::parse_object_tag_header(tagging)? { + let Some(key) = tag.key else { + continue; + }; + let Some(value) = tag.value else { + continue; + }; + tag_keys.push(key.clone()); + conditions.entry(format!("RequestObjectTag/{key}")).or_default().push(value); + } + conditions.insert("RequestObjectTagKeys".to_string(), tag_keys); + Ok(()) +} + +fn authorization_conditions( + req: &S3Request, + cred: &rustfs_credentials::Credentials, + version_id: Option<&str>, + region: Option, + remote_addr: Option, + client_info: Option<&ClientInfo>, + action: Action, +) -> S3Result>> { + let mut conditions = get_condition_values_with_query_and_client_info( + &req.headers, + cred, + version_id, + region, + remote_addr, + req.uri.query(), + client_info, + ); + merge_list_bucket_query_conditions(action, req.uri.query(), &mut conditions); + merge_request_object_tag_conditions(action, &req.headers, &mut conditions)?; + Ok(conditions) +} + fn auth_fs() -> &'static FS { static AUTH_FS: OnceLock = OnceLock::new(); AUTH_FS.get_or_init(FS::new) @@ -358,16 +410,7 @@ pub async fn authorize_request(req: &mut S3Request, action: Action) -> S3R let default_claims = HashMap::new(); let claims = cred.claims.as_ref().unwrap_or(&default_claims); let client_info = req.extensions.get::(); - let mut conditions = get_condition_values_with_query_and_client_info( - &req.headers, - cred, - version_id.as_deref(), - None, - remote_addr, - req.uri.query(), - client_info, - ); - merge_list_bucket_query_conditions(action, req.uri.query(), &mut conditions); + let mut conditions = authorization_conditions(req, cred, version_id.as_deref(), None, remote_addr, client_info, action)?; let action_args = Args { account: &cred.access_key, @@ -579,16 +622,15 @@ pub async fn authorize_request(req: &mut S3Request, action: Action) -> S3R } else { let default_cred = rustfs_credentials::Credentials::default(); let client_info = req.extensions.get::(); - let mut conditions = get_condition_values_with_query_and_client_info( - &req.headers, + let mut conditions = authorization_conditions( + req, &default_cred, version_id.as_deref(), req.region.clone(), remote_addr, - req.uri.query(), client_info, - ); - merge_list_bucket_query_conditions(action, req.uri.query(), &mut conditions); + action, + )?; let no_groups: Option> = None; let bucket_tag_hint = if !bucket.is_empty() && !object.is_empty() { @@ -2105,12 +2147,12 @@ mod tests { bucket_policy_needs_existing_object_tag_from_hint, classify_bucket_policy_raw_load_error, complete_multipart_upload_authorize_action, get_bucket_policy_authorize_action, has_write_offset_bytes_header, legal_hold_write_requested, list_parts_authorize_action, load_bucket_policy_existing_object_tag_hint, - merge_list_bucket_query_conditions, owner_can_bypass_policy_deny, post_object_authorize_action, - put_bucket_policy_authorize_action, request_context_from_req, retention_write_requested, secondary_tag_hint_action, - table_data_plane_admin_action, validate_post_object_success_controls, versioned_read_action, + merge_list_bucket_query_conditions, merge_request_object_tag_conditions, owner_can_bypass_policy_deny, + post_object_authorize_action, put_bucket_policy_authorize_action, request_context_from_req, retention_write_requested, + secondary_tag_hint_action, table_data_plane_admin_action, validate_post_object_success_controls, versioned_read_action, }; use crate::error::ApiError; - use http::{Extensions, HeaderMap, Method, Uri}; + use http::{Extensions, HeaderMap, HeaderValue, Method, Uri}; use rustfs_policy::policy::action::{Action, S3Action}; use rustfs_policy::policy::{BucketPolicy, bucket_policy_uses_existing_object_tag_conditions}; use s3s::{S3ErrorCode, S3Request, dto::*}; @@ -2323,6 +2365,33 @@ mod tests { assert!(conditions.is_empty()); } + #[test] + fn test_merge_request_object_tag_conditions_applies_only_to_put_object() { + let mut headers = HeaderMap::new(); + headers.insert("x-amz-tagging", HeaderValue::from_static("classification=restricted&label=copy%20test")); + + let mut source_conditions = HashMap::new(); + merge_request_object_tag_conditions(Action::S3Action(S3Action::GetObjectAction), &headers, &mut source_conditions) + .expect("GetObject condition construction should succeed"); + assert!( + source_conditions.is_empty(), + "destination request tags must not affect CopyObject source authorization" + ); + + let mut destination_conditions = HashMap::new(); + merge_request_object_tag_conditions(Action::S3Action(S3Action::PutObjectAction), &headers, &mut destination_conditions) + .expect("PutObject condition construction should succeed"); + assert_eq!( + destination_conditions.get("RequestObjectTag/classification"), + Some(&vec!["restricted".to_string()]) + ); + assert_eq!(destination_conditions.get("RequestObjectTag/label"), Some(&vec!["copy test".to_string()])); + assert_eq!( + destination_conditions.get("RequestObjectTagKeys"), + Some(&vec!["classification".to_string(), "label".to_string()]) + ); + } + #[test] fn test_object_tag_conditions_key_format() { let mut tags = HashMap::new(); diff --git a/rustfs/src/storage/s3_api/tagging.rs b/rustfs/src/storage/s3_api/tagging.rs index d8ac35504..1b22a3f88 100644 --- a/rustfs/src/storage/s3_api/tagging.rs +++ b/rustfs/src/storage/s3_api/tagging.rs @@ -12,14 +12,27 @@ // See the License for the specific language governing permissions and // limitations under the License. -use s3s::dto::Tag; +use percent_encoding::percent_decode_str; +use s3s::dto::{Tag, TaggingDirective}; use s3s::{S3Error, S3ErrorCode, S3Result}; use std::collections::HashSet; +use std::sync::OnceLock; +use url::form_urlencoded; + +use crate::storage::storage_api::encode_tags; fn invalid_tag_error(message: &str) -> S3Error { S3Error::with_message(S3ErrorCode::InvalidTag, message.to_string()) } +fn is_valid_tag_text(value: &str) -> bool { + static VALID_OBJECT_TAG_TEXT: OnceLock> = OnceLock::new(); + VALID_OBJECT_TAG_TEXT + .get_or_init(|| regex::Regex::new(r"^[\p{L}\p{N}\p{Z}+\-=._:/@]*$")) + .as_ref() + .is_ok_and(|pattern| pattern.is_match(value)) +} + pub(crate) fn validate_object_tag_set(tag_set: &[Tag]) -> S3Result<()> { if tag_set.len() > 10 { return Err(invalid_tag_error("Cannot have more than 10 tags per object")); @@ -33,8 +46,8 @@ pub(crate) fn validate_object_tag_set(tag_set: &[Tag]) -> S3Result<()> { .filter(|key| !key.is_empty()) .ok_or_else(|| invalid_tag_error("Tag key cannot be empty"))?; - if key.len() > 128 { - return Err(invalid_tag_error("Tag key is too long, maximum allowed length is 128 characters")); + if key.encode_utf16().count() > 128 || !is_valid_tag_text(key) { + return Err(invalid_tag_error("The TagKey you have provided is invalid")); } let value = tag @@ -42,8 +55,8 @@ pub(crate) fn validate_object_tag_set(tag_set: &[Tag]) -> S3Result<()> { .as_deref() .ok_or_else(|| invalid_tag_error("Tag value cannot be null"))?; - if value.len() > 256 { - return Err(invalid_tag_error("Tag value is too long, maximum allowed length is 256 characters")); + if value.encode_utf16().count() > 256 || !is_valid_tag_text(value) { + return Err(invalid_tag_error("The TagValue you have provided is invalid")); } if !tag_keys.insert(key) { @@ -54,11 +67,80 @@ pub(crate) fn validate_object_tag_set(tag_set: &[Tag]) -> S3Result<()> { Ok(()) } +fn validate_tag_component_encoding(component: &str) -> S3Result<()> { + let bytes = component.as_bytes(); + let mut index = 0; + while index < bytes.len() { + if bytes[index] == b'%' { + if index + 2 >= bytes.len() || !bytes[index + 1].is_ascii_hexdigit() || !bytes[index + 2].is_ascii_hexdigit() { + return Err(invalid_tag_error("The Tagging header contains invalid percent encoding")); + } + index += 3; + } else { + index += 1; + } + } + + percent_decode_str(component) + .decode_utf8() + .map_err(|_| invalid_tag_error("The Tagging header contains invalid UTF-8"))?; + Ok(()) +} + +pub(crate) fn parse_object_tag_header(tagging: &str) -> S3Result> { + let mut tag_set = Vec::with_capacity(10); + for pair in tagging.split('&').filter(|pair| !pair.is_empty()) { + if tag_set.len() == 10 { + return Err(invalid_tag_error("Cannot have more than 10 tags per object")); + } + + let (key, value) = pair.split_once('=').unwrap_or((pair, "")); + validate_tag_component_encoding(key)?; + validate_tag_component_encoding(value)?; + let mut parsed = form_urlencoded::parse(pair.as_bytes()); + if let Some((key, value)) = parsed.next() { + tag_set.push(Tag { + key: Some(key.into_owned()), + value: Some(value.into_owned()), + }); + } + } + + validate_object_tag_set(&tag_set)?; + Ok(tag_set) +} + +pub(crate) fn parse_copy_object_tags(tagging: &str) -> S3Result { + parse_object_tag_header(tagging).map(encode_tags) +} + +pub(crate) fn resolve_copy_object_tags( + tagging: Option<&str>, + tagging_directive: Option<&TaggingDirective>, +) -> S3Result> { + match tagging_directive.map(TaggingDirective::as_str) { + None | Some(TaggingDirective::COPY) => { + if tagging.is_some() { + return Err(S3Error::with_message( + S3ErrorCode::InvalidRequest, + "The Tagging header requires the REPLACE tagging directive".to_string(), + )); + } + Ok(None) + } + Some(TaggingDirective::REPLACE) => parse_copy_object_tags(tagging.unwrap_or_default()).map(Some), + Some(_) => Err(S3Error::with_message( + S3ErrorCode::InvalidArgument, + "The TaggingDirective header is invalid".to_string(), + )), + } +} + #[cfg(test)] mod tests { - use super::validate_object_tag_set; + use super::{parse_copy_object_tags, resolve_copy_object_tags, validate_object_tag_set}; use s3s::S3ErrorCode; - use s3s::dto::Tag; + use s3s::dto::{Tag, TaggingDirective}; fn tag(key: Option<&str>, value: Option<&str>) -> Tag { Tag { @@ -95,10 +177,7 @@ mod tests { let long_key = "k".repeat(129); let err = validate_object_tag_set(&[tag(Some(&long_key), Some("v1"))]).expect_err("too long tag key must be rejected"); assert_eq!(*err.code(), S3ErrorCode::InvalidTag); - assert!( - err.to_string() - .contains("Tag key is too long, maximum allowed length is 128 characters") - ); + assert!(err.to_string().contains("The TagKey you have provided is invalid")); } #[test] @@ -114,10 +193,7 @@ mod tests { let err = validate_object_tag_set(&[tag(Some("k1"), Some(&long_value))]).expect_err("too long tag value must be rejected"); assert_eq!(*err.code(), S3ErrorCode::InvalidTag); - assert!( - err.to_string() - .contains("Tag value is too long, maximum allowed length is 256 characters") - ); + assert!(err.to_string().contains("The TagValue you have provided is invalid")); } #[test] @@ -127,4 +203,122 @@ mod tests { assert_eq!(*err.code(), S3ErrorCode::InvalidTag); assert!(err.to_string().contains("Cannot provide multiple Tags with the same key")); } + + #[test] + fn test_validate_object_tag_set_accepts_maximum_lengths_and_allowed_characters() { + let key = "𐐀".repeat(64); + let value = format!("{}=", "v".repeat(255)); + + assert!(validate_object_tag_set(&[tag(Some(&key), Some(&value))]).is_ok()); + } + + #[test] + fn test_validate_object_tag_set_counts_utf16_code_units() { + let too_long_key = "𐐀".repeat(65); + let err = validate_object_tag_set(&[tag(Some(&too_long_key), Some("value"))]) + .expect_err("a key longer than 128 UTF-16 code units must be rejected"); + + assert_eq!(*err.code(), S3ErrorCode::InvalidTag); + } + + #[test] + fn test_validate_object_tag_set_rejects_unsupported_characters() { + let key_err = validate_object_tag_set(&[tag(Some("invalid?key"), Some("value"))]).expect_err("unsupported key character"); + assert_eq!(*key_err.code(), S3ErrorCode::InvalidTag); + + let value_err = + validate_object_tag_set(&[tag(Some("key"), Some("invalid&value"))]).expect_err("unsupported value character"); + assert_eq!(*value_err.code(), S3ErrorCode::InvalidTag); + + let combining_mark_err = validate_object_tag_set(&[tag(Some("invalid\u{0345}key"), Some("value"))]) + .expect_err("a combining mark outside the S3 letter/number/separator categories must be rejected"); + assert_eq!(*combining_mark_err.code(), S3ErrorCode::InvalidTag); + } + + #[test] + fn test_parse_copy_object_tags_accepts_encoded_tag_set() { + let tags = + parse_copy_object_tags("project=rustfs&label=copy%20test").expect("valid URL-encoded CopyObject tags should parse"); + + assert_eq!(tags, "project=rustfs&label=copy+test"); + } + + #[test] + fn test_parse_copy_object_tags_accepts_empty_replacement() { + let tags = parse_copy_object_tags("").expect("an empty replacement tag set should parse"); + + assert!(tags.is_empty()); + } + + #[test] + fn test_parse_copy_object_tags_rejects_invalid_percent_encoding() { + let err = parse_copy_object_tags("project=rustfs%ZZ") + .expect_err("invalid percent encoding must be rejected before CopyObject writes"); + + assert_eq!(*err.code(), S3ErrorCode::InvalidTag); + } + + #[test] + fn test_parse_copy_object_tags_rejects_duplicate_keys() { + let err = parse_copy_object_tags("project=rustfs&project=cli").expect_err("duplicate tag keys must be rejected"); + + assert_eq!(*err.code(), S3ErrorCode::InvalidTag); + } + + #[test] + fn test_parse_copy_object_tags_stops_at_object_tag_limit() { + let tagging = (0..11) + .map(|index| format!("k{index}=v{index}")) + .collect::>() + .join("&"); + let err = parse_copy_object_tags(&tagging).expect_err("the eleventh object tag must be rejected"); + + assert_eq!(*err.code(), S3ErrorCode::InvalidTag); + assert!(err.to_string().contains("Cannot have more than 10 tags per object")); + } + + #[test] + fn test_parse_copy_object_tags_rejects_control_whitespace() { + let err = parse_copy_object_tags("project=line%0Abreak").expect_err("control whitespace is not valid in S3 object tags"); + + assert_eq!(*err.code(), S3ErrorCode::InvalidTag); + } + + #[test] + fn test_resolve_copy_object_tags_preserves_source_for_default_and_copy() { + assert_eq!( + resolve_copy_object_tags(None, None).expect("default directive should preserve source tags"), + None + ); + let copy = TaggingDirective::from_static(TaggingDirective::COPY); + assert_eq!( + resolve_copy_object_tags(None, Some(©)).expect("COPY directive should preserve source tags"), + None + ); + } + + #[test] + fn test_resolve_copy_object_tags_replaces_with_exact_requested_set() { + let replace = TaggingDirective::from_static(TaggingDirective::REPLACE); + assert_eq!( + resolve_copy_object_tags(Some("project=rustfs"), Some(&replace)) + .expect("REPLACE directive should accept a valid tag set"), + Some("project=rustfs".to_string()) + ); + assert_eq!( + resolve_copy_object_tags(None, Some(&replace)).expect("REPLACE without Tagging should clear tags"), + Some(String::new()) + ); + } + + #[test] + fn test_resolve_copy_object_tags_rejects_discarded_or_unknown_requests() { + let err = resolve_copy_object_tags(Some("project=rustfs"), None) + .expect_err("a Tagging header without REPLACE must not be silently discarded"); + assert_eq!(*err.code(), S3ErrorCode::InvalidRequest); + + let unknown = TaggingDirective::from_static("UNKNOWN"); + let err = resolve_copy_object_tags(None, Some(&unknown)).expect_err("an unknown tagging directive must fail"); + assert_eq!(*err.code(), S3ErrorCode::InvalidArgument); + } } diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 6df459857..d25c9ec43 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -311,6 +311,10 @@ pub(crate) mod s3_api_consumer { super::super::to_s3s_etag(etag) } } + + pub(crate) mod tagging { + pub(crate) use super::super::super::s3_api::tagging::resolve_copy_object_tags; + } } pub(crate) mod sse_consumer {