From 5cedab09ab3c9dde61ae23c1a920816d0e15888e Mon Sep 17 00:00:00 2001 From: GatewayJ <835269233@qq.com> Date: Tue, 28 Jul 2026 10:39:17 +0800 Subject: [PATCH] fix(sts): return Query API-compatible responses (#5282) * fix(sts): return Query API-compatible responses * fix(sts): resolve review and CI failures * fix(sts): harden query response conversion * test(e2e): stabilize transition conflict coverage * fix(ilm): retry vanished transition admission CAS * test(e2e): narrow transition overlap coverage --------- Co-authored-by: houseme --- .config/nextest.toml | 2 +- Cargo.lock | 1 + Cargo.toml | 1 + crates/e2e_test/Cargo.toml | 1 + crates/e2e_test/src/lib.rs | 3 + crates/e2e_test/src/reliant/tiering.rs | 177 +----- crates/e2e_test/src/sts_query_compat_test.rs | 229 +++++++ .../bucket/lifecycle/bucket_lifecycle_ops.rs | 42 ++ .../bucket/lifecycle/manual_transition_job.rs | 8 +- rustfs/src/admin/handlers/sts.rs | 4 +- rustfs/src/admin/router.rs | 65 +- rustfs/src/server/compress.rs | 12 +- rustfs/src/server/http.rs | 46 +- rustfs/src/server/layer.rs | 562 +++++++++++++++++- rustfs/src/server/mod.rs | 2 +- rustfs/src/storage/request_context.rs | 36 +- 16 files changed, 982 insertions(+), 209 deletions(-) create mode 100644 crates/e2e_test/src/sts_query_compat_test.rs diff --git a/.config/nextest.toml b/.config/nextest.toml index f264f7559..16418d2e6 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -209,7 +209,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_object_tagging|copy_source_invalid_date|content_encoding|multipart_storage_class|storage_class_capability|ssec_copy|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|multipart_storage_class|storage_class_capability|ssec_copy|anonymous_access|bucket_policy_check|presigned_negative|negative_sigv4|admin_auth|notification_webhook|tls_hot_reload|console_smoke|admin_iam_crud|admin_pools|sts_query_compat)_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 253b51d35..6086ee54c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3662,6 +3662,7 @@ dependencies = [ "async-trait", "aws-config", "aws-sdk-s3", + "aws-sdk-sts", "aws-smithy-http-client", "base64 0.23.0", "bytes", diff --git a/Cargo.toml b/Cargo.toml index 93306bca1..e6e66d836 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -228,6 +228,7 @@ atomic_enum = "0.3.0" aws-config = { version = "1.10.1" } aws-credential-types = { version = "1.3.0" } aws-sdk-s3 = { default-features = false, version = "1.140.0" } +aws-sdk-sts = { default-features = false, version = "1.110.0" } aws-smithy-http-client = { default-features = false, version = "1.2.0" } aws-smithy-runtime-api = { version = "1.14.0" } aws-smithy-types = { version = "1.6.1" } diff --git a/crates/e2e_test/Cargo.toml b/crates/e2e_test/Cargo.toml index 135a976c3..028441008 100644 --- a/crates/e2e_test/Cargo.toml +++ b/crates/e2e_test/Cargo.toml @@ -50,6 +50,7 @@ rustfs-filemeta.workspace = true bytes = { workspace = true, features = ["serde"] } serial_test = { workspace = true } aws-sdk-s3 = { workspace = true, default-features = false, features = ["sigv4a", "default-https-client", "rt-tokio"] } +aws-sdk-sts = { workspace = true, default-features = false, features = ["default-https-client", "rt-tokio"] } aws-config = { workspace = true } aws-smithy-http-client = { workspace = true, default-features = false, features = ["rustls-aws-lc"] } async-compression = { workspace = true, features = ["tokio", "bzip2", "xz"] } diff --git a/crates/e2e_test/src/lib.rs b/crates/e2e_test/src/lib.rs index 3f5abcb24..6730e2b8d 100644 --- a/crates/e2e_test/src/lib.rs +++ b/crates/e2e_test/src/lib.rs @@ -101,6 +101,9 @@ mod admin_auth_test; #[cfg(test)] mod existing_object_tag_policy_test; +#[cfg(test)] +mod sts_query_compat_test; + // Regression tests for Issue #2036: anonymous access with PublicAccessBlock #[cfg(test)] mod anonymous_access_test; diff --git a/crates/e2e_test/src/reliant/tiering.rs b/crates/e2e_test/src/reliant/tiering.rs index 1ee8b7ecc..ee4d89e81 100644 --- a/crates/e2e_test/src/reliant/tiering.rs +++ b/crates/e2e_test/src/reliant/tiering.rs @@ -74,7 +74,6 @@ const MANUAL_ASYNC_STATUS_BUCKET: &str = "ilm7-manual-async-status"; const MANUAL_CONTINUATION_BUCKET: &str = "ilm7-manual-continuation"; const MANUAL_ASYNC_LIMIT_BUCKET: &str = "ilm7-manual-async-limit"; const MANUAL_ASYNC_CONFLICT_BUCKET: &str = "ilm7-manual-async-conflict"; -const MANUAL_ASYNC_SAME_SCOPE_CONFLICT_BUCKET: &str = "ilm7-manual-async-same-scope-conflict"; const MANUAL_ASYNC_PARALLEL_BUCKET_A: &str = "ilm7-manual-async-parallel-a"; const MANUAL_ASYNC_PARALLEL_BUCKET_B: &str = "ilm7-manual-async-parallel-b"; const MANUAL_TIER_FAILURE_BUCKET: &str = "ilm7-manual-tier-failure"; @@ -86,7 +85,6 @@ const MANUAL_CONTINUATION_PREFIX: &str = "manual-continuation/"; const MANUAL_ASYNC_LIMIT_PREFIX: &str = "manual-async-limit/"; const MANUAL_ASYNC_CONFLICT_PREFIX: &str = "manual-async-conflict/"; const MANUAL_ASYNC_CONFLICT_NESTED_PREFIX: &str = "manual-async-conflict/nested/"; -const MANUAL_ASYNC_SAME_SCOPE_CONFLICT_PREFIX: &str = "manual-async-same-scope-conflict/"; const MANUAL_ASYNC_PARALLEL_PREFIX: &str = "manual-async-parallel/"; const MANUAL_TIER_FAILURE_PREFIX: &str = "manual-tier-failure/"; const MANUAL_WORKER_FAILURE_PREFIX: &str = "manual-worker-failure/"; @@ -97,6 +95,7 @@ const MANUAL_ASYNC_PARALLEL_OBJECTS: usize = 64; const MANUAL_ACTIVE_CANCEL_OBJECTS: usize = 512; const MANUAL_RESTART_CANCEL_OBJECTS: usize = 512; const MANUAL_ACTIVE_CANCEL_RUNNING_TIMEOUT: StdDuration = StdDuration::from_secs(15); +const MANUAL_ASYNC_CONFLICT_TERMINAL_TIMEOUT: StdDuration = StdDuration::from_secs(90); const MANUAL_RESTART_RECOVERY_TIMEOUT: StdDuration = StdDuration::from_secs(80); const OBJECT_KEY: &str = "tier/鲁A12345/report.bin"; const MANUAL_DUE_KEY: &str = "manual-due/report.bin"; @@ -1267,7 +1266,7 @@ async fn test_manual_transition_async_limit_reports_terminal_partial() -> TestRe } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn test_manual_transition_async_overlapping_scope_conflict_reports_active_job() -> TestResult { +async fn test_manual_transition_async_scope_conflicts_report_active_job() -> TestResult { let mut cold = RustFSTestEnvironment::new().await?; cold.access_key = "manualasyncconflictcoldtieradmin".to_string(); cold.secret_key = "manualasyncconflictcoldtiersecret".to_string(); @@ -1296,34 +1295,14 @@ async fn test_manual_transition_async_overlapping_scope_conflict_reports_active_ .await?; let before_remote_count = cold_tier_object_count(&cold_client).await?; - let (parent, nested) = tokio::join!( - manual_transition_async_run_raw( - &hot, - MANUAL_ASYNC_CONFLICT_BUCKET, - MANUAL_ASYNC_CONFLICT_PREFIX, - false, - MANUAL_ASYNC_CONFLICT_OBJECTS as u64, - ), - manual_transition_async_run_raw( - &hot, - MANUAL_ASYNC_CONFLICT_BUCKET, - MANUAL_ASYNC_CONFLICT_NESTED_PREFIX, - false, - MANUAL_ASYNC_CONFLICT_OBJECTS as u64, - ) - ); - let responses = [parent?, nested?]; - let accepted = responses - .iter() - .find(|(status, _)| *status == reqwest::StatusCode::ACCEPTED) - .ok_or("one concurrent overlapping-scope async run must be accepted")?; - let conflict = responses - .iter() - .find(|(status, _)| *status == reqwest::StatusCode::CONFLICT) - .ok_or("one concurrent overlapping-scope async run must report conflict")?; - - let accepted: ManualTransitionRunResponse = serde_json::from_str(&accepted.1)?; - let conflict: ManualTransitionJobConflictResponse = serde_json::from_str(&conflict.1)?; + let accepted = manual_transition_async_run( + &hot, + MANUAL_ASYNC_CONFLICT_BUCKET, + MANUAL_ASYNC_CONFLICT_PREFIX, + false, + MANUAL_ASYNC_CONFLICT_OBJECTS as u64, + ) + .await?; let job_id = accepted .job_id .as_deref() @@ -1341,13 +1320,25 @@ async fn test_manual_transition_async_overlapping_scope_conflict_reports_active_ assert_eq!(accepted.state, "accepted"); assert_eq!(accepted.mode, "durable_job"); assert_eq!(accepted.report.bucket, MANUAL_ASYNC_CONFLICT_BUCKET); - assert!( - matches!( - accepted.report.prefix.as_str(), - MANUAL_ASYNC_CONFLICT_PREFIX | MANUAL_ASYNC_CONFLICT_NESTED_PREFIX - ), - "overlapping async winner prefix: {accepted:#?}" + assert_eq!(accepted.report.prefix, MANUAL_ASYNC_CONFLICT_PREFIX); + let active = wait_for_manual_transition_job_running(&hot, status_endpoint, MANUAL_ACTIVE_CANCEL_RUNNING_TIMEOUT).await?; + assert_eq!(active.job_id, job_id); + + let (conflict_status, conflict_body) = manual_transition_async_run_raw( + &hot, + MANUAL_ASYNC_CONFLICT_BUCKET, + MANUAL_ASYNC_CONFLICT_NESTED_PREFIX, + false, + MANUAL_ASYNC_CONFLICT_OBJECTS as u64, + ) + .await?; + assert_eq!( + conflict_status, + reqwest::StatusCode::CONFLICT, + "active async run must reject nested prefix {}: {conflict_body}", + MANUAL_ASYNC_CONFLICT_NESTED_PREFIX ); + let conflict: ManualTransitionJobConflictResponse = serde_json::from_str(&conflict_body)?; assert_eq!(conflict.state, "conflict"); assert_eq!(conflict.mode, "durable_job"); assert_eq!(conflict.active_job_id, job_id); @@ -1355,7 +1346,7 @@ async fn test_manual_transition_async_overlapping_scope_conflict_reports_active_ assert_eq!(conflict.cancel_endpoint, status_endpoint); assert!(!conflict.scope_key.is_empty()); - let terminal = wait_for_manual_transition_job_terminal(&hot, status_endpoint, StdDuration::from_secs(30)).await?; + let terminal = wait_for_manual_transition_job_terminal(&hot, status_endpoint, MANUAL_ASYNC_CONFLICT_TERMINAL_TIMEOUT).await?; assert_eq!(terminal.job_id, job_id); assert!(!terminal.report.dry_run); assert_eq!(terminal.report.bucket, MANUAL_ASYNC_CONFLICT_BUCKET); @@ -1376,116 +1367,6 @@ async fn test_manual_transition_async_overlapping_scope_conflict_reports_active_ Ok(()) } -#[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn test_manual_transition_async_same_scope_conflict_reports_active_job() -> TestResult { - let mut cold = RustFSTestEnvironment::new().await?; - cold.access_key = "manualasyncsameconflictcoldadmin".to_string(); - cold.secret_key = "manualasyncsameconflictcoldsecret".to_string(); - cold.start_rustfs_server_without_cleanup(vec![]).await?; - let cold_client = cold.create_s3_client(); - cold_client.create_bucket().bucket(TIER_BUCKET).send().await?; - - let mut hot = RustFSTestEnvironment::new().await?; - hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]) - .await?; - let hot_client = hot.create_s3_client(); - add_rustfs_tier(&hot, &cold).await?; - - hot_client - .create_bucket() - .bucket(MANUAL_ASYNC_SAME_SCOPE_CONFLICT_BUCKET) - .send() - .await?; - for idx in 0..50 { - let key = format!("{MANUAL_ASYNC_SAME_SCOPE_CONFLICT_PREFIX}obj-{idx:02}"); - put_single_part_object( - &hot_client, - MANUAL_ASYNC_SAME_SCOPE_CONFLICT_BUCKET, - &key, - b"async same-scope conflict payload", - ) - .await?; - } - put_lifecycle_transition_rule( - &hot_client, - MANUAL_ASYNC_SAME_SCOPE_CONFLICT_BUCKET, - "manual-async-same-scope-conflict", - MANUAL_ASYNC_SAME_SCOPE_CONFLICT_PREFIX, - 0, - ) - .await?; - - let (first, second) = tokio::join!( - manual_transition_async_run_raw( - &hot, - MANUAL_ASYNC_SAME_SCOPE_CONFLICT_BUCKET, - MANUAL_ASYNC_SAME_SCOPE_CONFLICT_PREFIX, - false, - 50 - ), - manual_transition_async_run_raw( - &hot, - MANUAL_ASYNC_SAME_SCOPE_CONFLICT_BUCKET, - MANUAL_ASYNC_SAME_SCOPE_CONFLICT_PREFIX, - false, - 50 - ) - ); - let responses = [first?, second?]; - let accepted = responses - .iter() - .find(|(status, _)| *status == reqwest::StatusCode::ACCEPTED) - .ok_or("one concurrent same-scope async run must be accepted")?; - let conflict = responses - .iter() - .find(|(status, _)| *status == reqwest::StatusCode::CONFLICT) - .ok_or("one concurrent same-scope async run must report conflict")?; - - let accepted: ManualTransitionRunResponse = serde_json::from_str(&accepted.1)?; - let conflict: ManualTransitionJobConflictResponse = serde_json::from_str(&conflict.1)?; - let job_id = accepted - .job_id - .as_deref() - .ok_or("accepted same-scope async response must include job_id")?; - let status_endpoint = accepted - .status_endpoint - .as_deref() - .ok_or("accepted same-scope async response must include status_endpoint")?; - let cancel_endpoint = accepted - .cancel_endpoint - .as_deref() - .ok_or("accepted same-scope async response must include cancel_endpoint")?; - assert_eq!(cancel_endpoint, status_endpoint); - - assert_eq!(accepted.state, "accepted"); - assert_eq!(accepted.mode, "durable_job"); - assert_eq!(accepted.report.bucket, MANUAL_ASYNC_SAME_SCOPE_CONFLICT_BUCKET); - assert_eq!(accepted.report.prefix, MANUAL_ASYNC_SAME_SCOPE_CONFLICT_PREFIX); - assert_eq!(conflict.state, "conflict"); - assert_eq!(conflict.mode, "durable_job"); - assert_eq!(conflict.active_job_id, job_id); - assert_eq!(conflict.status_endpoint, status_endpoint); - assert_eq!(conflict.cancel_endpoint, status_endpoint); - assert!(!conflict.scope_key.is_empty()); - - let terminal = wait_for_manual_transition_job_terminal(&hot, status_endpoint, StdDuration::from_secs(30)).await?; - assert_eq!(terminal.job_id, job_id); - assert_eq!(terminal.report.bucket, MANUAL_ASYNC_SAME_SCOPE_CONFLICT_BUCKET); - assert_eq!(terminal.report.prefix, MANUAL_ASYNC_SAME_SCOPE_CONFLICT_PREFIX); - assert_conflict_winner_report(&terminal.status, &terminal.report, 50, "terminal same-scope conflict winner response"); - assert_eq!( - terminal.report.transition_failed, 0, - "terminal same-scope conflict winner response: {terminal:#?}" - ); - assert_eq!( - terminal.report.tier_failure, 0, - "terminal same-scope conflict winner response: {terminal:#?}" - ); - assert!(cold_tier_object_count(&cold_client).await? <= 50); - - Ok(()) -} - #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn test_manual_transition_async_different_buckets_admit_concurrently() -> TestResult { let mut cold = RustFSTestEnvironment::new().await?; diff --git a/crates/e2e_test/src/sts_query_compat_test.rs b/crates/e2e_test/src/sts_query_compat_test.rs new file mode 100644 index 000000000..9637bc0e1 --- /dev/null +++ b/crates/e2e_test/src/sts_query_compat_test.rs @@ -0,0 +1,229 @@ +// 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. + +use crate::common::{RustFSTestEnvironment, init_logging, local_http_client}; +use aws_sdk_sts::config::retry::RetryConfig; +use aws_sdk_sts::config::{Credentials, Region}; +use aws_sdk_sts::error::ProvideErrorMetadata; +use aws_sdk_sts::operation::RequestId; +use aws_sdk_sts::{Client, Config}; +use aws_smithy_http_client::Builder as SmithyHttpClientBuilder; +use http::header::{CONTENT_TYPE, HOST}; +use rustfs_signer::constants::UNSIGNED_PAYLOAD; +use rustfs_signer::sign_v4; +use s3s::Body; +use serial_test::serial; +use std::error::Error; + +type BoxError = Box; +type TestResult = Result<(), BoxError>; + +fn sts_client(url: &str, access_key: &str, secret_key: &str, session_token: Option<&str>) -> Client { + let mut config = Config::builder() + .credentials_provider(Credentials::new( + access_key, + secret_key, + session_token.map(str::to_owned), + None, + "e2e-sts-query-compat", + )) + .region(Region::new("us-east-1")) + .endpoint_url(url) + .retry_config(RetryConfig::standard().with_max_attempts(1)) + .behavior_version_latest(); + if url.starts_with("http://") { + config = config.http_client(SmithyHttpClientBuilder::new().build_http()); + } + Client::from_conf(config.build()) +} + +async fn create_root_service_account(env: &RustFSTestEnvironment) -> Result<(String, String), BoxError> { + let path = "/rustfs/admin/v3/add-service-accounts"; + let url = format!("{}{path}", env.url); + let uri = url.parse::()?; + let authority = uri.authority().ok_or("admin URL missing authority")?.to_string(); + let body = serde_json::json!({ "targetUser": env.access_key.clone() }).to_string(); + let request = http::Request::builder() + .method(http::Method::PUT) + .uri(uri) + .header(HOST, authority) + .header(CONTENT_TYPE, "application/json") + .header("x-amz-content-sha256", UNSIGNED_PAYLOAD) + .body(Body::empty())?; + let content_length = i64::try_from(body.len()).map_err(|_| "service account request body is too large")?; + let signed = sign_v4(request, content_length, &env.access_key, &env.secret_key, "", "us-east-1"); + let mut request = local_http_client().put(&url); + for (name, value) in signed.headers() { + request = request.header(name, value); + } + let response = request.body(body).send().await?; + let status = response.status(); + let body = response.text().await?; + if !status.is_success() { + return Err(format!("create service account failed: {status} {body}").into()); + } + + let response: serde_json::Value = serde_json::from_str(&body)?; + let access_key = response["credentials"]["accessKey"] + .as_str() + .ok_or("service account response should contain credentials.accessKey")? + .to_owned(); + let secret_key = response["credentials"]["secretKey"] + .as_str() + .ok_or("service account response should contain credentials.secretKey")? + .to_owned(); + Ok((access_key, secret_key)) +} + +async fn assert_chaining_denied(client: &Client, credential_kind: &str) -> TestResult { + let error = client + .assume_role() + .role_arn("arn:aws:iam::123456789012:role/test") + .role_session_name("sts-query-compat-e2e") + .send() + .await + .expect_err("credential chaining must be denied"); + let service_error = error + .as_service_error() + .ok_or_else(|| format!("{credential_kind} denial should deserialize as an STS service error: {error:?}"))?; + + assert_eq!(error.raw_response().map(|response| response.status().as_u16()), Some(403)); + assert_eq!(service_error.code(), Some("AccessDenied")); + assert_eq!(service_error.message(), Some("Access Denied")); + assert!( + error.request_id().is_some_and(|request_id| !request_id.is_empty()), + "{credential_kind} denial should include a request ID" + ); + Ok(()) +} + +#[tokio::test] +#[serial] +async fn test_sts_query_responses_are_aws_sdk_compatible() -> TestResult { + init_logging(); + + let mut env = RustFSTestEnvironment::new().await?; + env.start_rustfs_server(vec![]).await?; + + let assumed = sts_client(&env.url, &env.access_key, &env.secret_key, None) + .assume_role() + .role_arn("arn:aws:iam::123456789012:role/test") + .role_session_name("sts-query-compat-e2e") + .send() + .await?; + assert!( + assumed.request_id().is_some_and(|request_id| !request_id.is_empty()), + "successful AssumeRole should include a request ID" + ); + let temporary = assumed + .credentials() + .ok_or("successful AssumeRole response should contain credentials")?; + + let invalid_signature = sts_client(&env.url, &env.access_key, "incorrect-secret-key", None) + .assume_role() + .role_arn("arn:aws:iam::123456789012:role/test") + .role_session_name("sts-query-invalid-signature") + .send() + .await + .expect_err("an invalid signature must be rejected"); + assert_eq!(invalid_signature.raw_response().map(|response| response.status().as_u16()), Some(403)); + let invalid_signature_service_error = invalid_signature + .as_service_error() + .ok_or_else(|| format!("invalid signature should deserialize as an STS service error: {invalid_signature:?}"))?; + assert_eq!(invalid_signature_service_error.code(), Some("SignatureDoesNotMatch")); + assert!( + invalid_signature_service_error + .message() + .is_some_and(|message| message.starts_with("The request signature we calculated does not match")), + "signature rejection should preserve the canonical error message" + ); + assert!( + invalid_signature + .request_id() + .is_some_and(|request_id| !request_id.is_empty()), + "signature rejection should include a request ID" + ); + + assert_chaining_denied( + &sts_client( + &env.url, + temporary.access_key_id(), + temporary.secret_access_key(), + Some(temporary.session_token()), + ), + "temporary credential", + ) + .await?; + + let (service_access_key, service_secret_key) = create_root_service_account(&env).await?; + assert_chaining_denied(&sts_client(&env.url, &service_access_key, &service_secret_key, None), "service account").await?; + + env.stop_server(); + Ok(()) +} + +#[tokio::test] +#[serial] +async fn test_sts_query_rate_limit_error_is_aws_sdk_compatible() -> TestResult { + init_logging(); + + let mut env = RustFSTestEnvironment::new().await?; + env.start_rustfs_server_with_env( + vec![], + &[ + ("RUSTFS_API_RATE_LIMIT_ENABLE", "true"), + ("RUSTFS_API_RATE_LIMIT_RPM", "60"), + ("RUSTFS_API_RATE_LIMIT_BURST", "1"), + ], + ) + .await?; + + let client = sts_client(&env.url, &env.access_key, &env.secret_key, None); + let mut throttled = None; + let request = || { + client + .assume_role() + .role_arn("arn:aws:iam::123456789012:role/test") + .role_session_name("sts-query-rate-limit") + .send() + }; + let (first, second, third, fourth) = tokio::join!(request(), request(), request(), request()); + for result in [first, second, third, fourth] { + if let Err(error) = result + && error.raw_response().map(|response| response.status().as_u16()) == Some(429) + { + throttled = Some(error); + break; + } + } + + let error = throttled.ok_or("at least one concurrent STS request should be throttled at burst one")?; + let service_error = error + .as_service_error() + .ok_or_else(|| format!("rate limit response should deserialize as an STS service error: {error:?}"))?; + assert_eq!(service_error.code(), Some("TooManyRequests")); + assert!( + service_error + .message() + .is_some_and(|message| message.starts_with("Request rate limit exceeded")), + "rate limit response should preserve the server message" + ); + assert!( + error.request_id().is_some_and(|request_id| !request_id.is_empty()), + "rate limit response should include a request ID" + ); + + env.stop_server(); + Ok(()) +} diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index e06a25dd2..f0d26f72a 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -9483,6 +9483,48 @@ mod tests { assert_eq!(current.lease_id, second.lease_id); } + #[tokio::test] + #[serial] + async fn manual_transition_scope_missing_current_is_retryable_cas_miss() { + let (_paths, ecstore) = setup_test_env().await; + let options = ManualTransitionRunOptions { + prefix: "logs/".to_string(), + ..Default::default() + }; + let first = ManualTransitionJobRecord::new(Uuid::new_v4(), "manual-replace-race-bucket", &options, "first-owner"); + save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&first)) + .await + .expect("first scope admission should save"); + let (_loaded, etag) = load_manual_transition_scope_admission_with_etag(ecstore.clone(), &first.scope_key) + .await + .expect("first scope admission should load with an ETag"); + let object = manual_transition_scope_record_object_name(&first.scope_key).expect("scope admission path should encode"); + config_boundary::delete_config(ecstore.clone(), &object) + .await + .expect("current scope admission should be deleted"); + + let replacement = + ManualTransitionJobRecord::new(Uuid::new_v4(), "manual-replace-race-bucket", &options, "replacement-owner"); + let stale_replace = save_manual_transition_scope_admission_if_current( + ecstore.clone(), + &ManualTransitionScopeAdmission::from_job(&replacement), + &etag, + ) + .await + .expect_err("replacing a disappeared scope admission must report a CAS miss"); + assert_eq!(stale_replace, Error::PreconditionFailed); + + let claim = + claim_manual_transition_scope_admission(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&replacement)) + .await + .expect("replacement claim should recover through the create path"); + assert_eq!(claim, ManualTransitionScopeAdmissionClaim::Claimed); + let current = load_manual_transition_scope_admission(ecstore, &replacement.scope_key) + .await + .expect("replacement scope admission should be saved"); + assert_eq!(current.job_id, replacement.job_id); + } + #[tokio::test] #[serial] async fn manual_transition_admission_reclaims_stale_scope_records() { diff --git a/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs b/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs index a8db7d722..2555e2359 100644 --- a/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs +++ b/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs @@ -1466,7 +1466,7 @@ pub async fn save_manual_transition_scope_admission_if_current( admission.validate().map_err(manual_transition_job_store_error)?; let object = manual_transition_scope_record_object_name(&admission.scope_key).map_err(manual_transition_job_store_error)?; let data = serde_json::to_vec(admission).map_err(Error::other)?; - config_boundary::save_config_with_opts( + match config_boundary::save_config_with_opts( api, &object, data, @@ -1480,6 +1480,12 @@ pub async fn save_manual_transition_scope_admission_if_current( }, ) .await + { + Err(Error::ObjectNotFound(bucket, current_object)) if bucket == RUSTFS_META_BUCKET && current_object == object => { + Err(Error::PreconditionFailed) + } + result => result, + } } pub async fn claim_manual_transition_scope_admission( diff --git a/rustfs/src/admin/handlers/sts.rs b/rustfs/src/admin/handlers/sts.rs index e9e1cb24f..c3d83df26 100644 --- a/rustfs/src/admin/handlers/sts.rs +++ b/rustfs/src/admin/handlers/sts.rs @@ -162,13 +162,13 @@ async fn handle_assume_role( let session_token = get_session_token(&uri, &headers); if session_token.is_some() { - return Err(s3_error!(InvalidRequest, "AccessDenied1")); + return Err(s3_error!(AccessDenied, "Access Denied")); } let (cred, owner) = check_key_valid(get_session_token(&uri, &headers).unwrap_or_default(), &user.access_key).await?; if cred.is_temp() || cred.is_service_account() { - return Err(s3_error!(InvalidRequest, "AccessDenied")); + return Err(s3_error!(AccessDenied, "Access Denied")); } let Ok(iam_store) = crate::admin::runtime_sources::current_ready_iam_handle() else { diff --git a/rustfs/src/admin/router.rs b/rustfs/src/admin/router.rs index dd4e537eb..869f636ee 100644 --- a/rustfs/src/admin/router.rs +++ b/rustfs/src/admin/router.rs @@ -38,6 +38,7 @@ use crate::error::ApiError; use crate::license::license_check; use crate::server::{ ADMIN_PREFIX, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, is_admin_path, + is_sts_query_request, }; use crate::storage::storage_api::lock_bucket_targets_metadata; use aws_sdk_s3::primitives::ByteStream as AwsByteStream; @@ -2769,15 +2770,7 @@ where } // AssumeRole - if method == Method::POST - && path == "/" - && headers - .get(header::CONTENT_TYPE) - .and_then(|v| v.to_str().ok()) - .map(|ct| ct.split(';').next().unwrap_or("").trim().to_lowercase()) - .map(|ct| ct == "application/x-www-form-urlencoded") - .unwrap_or(false) - { + if is_sts_query_request(method, uri, headers) { return true; } @@ -2827,22 +2820,7 @@ where // The handler dispatches on the Action parameter: AssumeRole will reject if // credentials are missing, AssumeRoleWithWebIdentity will validate the JWT. // Require application/x-www-form-urlencoded Content-Type to narrow the bypass. - if req.method == Method::POST - && path == "/" - && req.credentials.is_none() - && req - .headers - .get(header::CONTENT_TYPE) - .and_then(|v| v.to_str().ok()) - .map(|ct| { - ct.split(';') - .next() - .unwrap_or("") - .trim() - .eq_ignore_ascii_case("application/x-www-form-urlencoded") - }) - .unwrap_or(false) - { + if req.credentials.is_none() && is_sts_query_request(&req.method, &req.uri, &req.headers) { return Ok(()); } @@ -4794,6 +4772,43 @@ mod tests { assert_eq!(err.code(), &S3ErrorCode::AccessDenied); } + #[tokio::test] + async fn check_access_limits_anonymous_sts_exemption_to_form_content_type() { + let router: S3Router = S3Router::new(false); + let request = |content_type: Option<&'static str>| { + let mut headers = HeaderMap::new(); + if let Some(content_type) = content_type { + headers.insert(http::header::CONTENT_TYPE, HeaderValue::from_static(content_type)); + } + S3Request { + input: Body::from(String::from("Action=AssumeRoleWithWebIdentity")), + method: Method::POST, + uri: Uri::from_static("/"), + headers, + extensions: http::Extensions::new(), + credentials: None, + region: None, + service: None, + trailing_headers: None, + } + }; + + for content_type in [None, Some("application/json")] { + let mut req = request(content_type); + let err = router + .check_access(&mut req) + .await + .expect_err("anonymous STS request without form content type must be denied"); + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + } + + let mut req = request(Some("application/x-www-form-urlencoded")); + router + .check_access(&mut req) + .await + .expect("anonymous form-encoded STS request should reach identity validation"); + } + #[tokio::test] async fn check_access_allows_object_zip_download_token_navigation() { let router: S3Router = S3Router::new(false); diff --git a/rustfs/src/server/compress.rs b/rustfs/src/server/compress.rs index bccd80c8a..b678c1277 100644 --- a/rustfs/src/server/compress.rs +++ b/rustfs/src/server/compress.rs @@ -407,6 +407,8 @@ pub(crate) enum PathCategory { InternodeRpc, /// Health/probe paths — skip compression (tiny responses) Probe, + /// STS Query API — response compatibility is applied outside compression. + StsQueryApi, } impl PathCategory { @@ -544,7 +546,14 @@ where } fn call(&mut self, req: Request) -> Self::Future { - let category = PathCategory::classify(req.uri().path()); + let category = if req.method() == http::Method::POST + && req.uri().path() == "/" + && req.extensions().get::().is_some() + { + PathCategory::StsQueryApi + } else { + PathCategory::classify(req.uri().path()) + }; InjectCategoryFut { inner: self.inner.call(req), category, @@ -785,5 +794,6 @@ mod tests { assert!(!PathCategory::Console.should_evaluate_compression()); assert!(!PathCategory::InternodeRpc.should_evaluate_compression()); assert!(!PathCategory::Probe.should_evaluate_compression()); + assert!(!PathCategory::StsQueryApi.should_evaluate_compression()); } } diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index b88a4e53a..c7c49bbdf 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -24,8 +24,8 @@ use crate::server::{ layer::{ BodylessStatusFixLayer, ConditionalCorsLayer, DoubleSlashListBucketsCompatLayer, EmptyBodyContentLengthCompatLayer, HeadRequestBodyFixLayer, IcebergRestErrorCompatLayer, ObjectAttributesEtagFixLayer, PublicHealthEndpointLayer, - RedirectLayer, RequestContextLayer, RequestLoggingLayer, S3ErrorMessageCompatLayer, VirtualHostStyleHintLayer, - redact_sensitive_uri_query, + RedirectLayer, RequestContextLayer, RequestLoggingLayer, S3ErrorMessageCompatLayer, StsQueryApiCompatLayer, + VirtualHostStyleHintLayer, redact_sensitive_uri_query, }, rate_limit::{RateLimitLayer, api_rate_limit_layer_from_env}, tls_material::{ @@ -1390,26 +1390,27 @@ fn process_connection( // 3. TrustedProxyLayer — conditional, parses X-Forwarded-For // 4. SetRequestIdLayer — generates X-Request-ID // 5. RequestContextLayer — creates RequestContext in extensions - // 6. EmptyBodyContentLengthCompatLayer — adds Content-Length: 0 for known empty-body API routes - // 7. CatchPanicLayer — panic → 500 - // 8. RateLimitLayer — conditional (external stack only), per-client 429 throttling - // 9. ReadinessGateLayer — blocks until ready - // 10. KeystoneAuthLayer — X-Auth-Token validation - // 11. TraceLayer — request span creation + metrics - // 12. RequestLoggingLayer — single completion event per request - // 13. PropagateRequestIdLayer — X-Request-ID → response - // 14. CompressionLayer — response compression (whitelist, path-aware) - // 15. PathCategoryInjectionLayer — injects path category for compression predicate - // 16. S3ErrorMessageCompatLayer — missing S3 error message compatibility - // 17. IcebergRestErrorCompatLayer — Iceberg REST JSON error compatibility - // 18. ObjectAttributesEtagFixLayer — ETag fix for GetObjectAttributes - // 19. ConditionalCorsLayer — S3 API CORS - // 20. RedirectLayer — console redirect (conditional) - // 21. BodylessStatusFixLayer — clears body for 1xx/204/205/304 responses - // 22. HeadRequestBodyFixLayer — strips actual body bytes from HEAD responses - // 23. PublicHealthEndpointLayer — handles public health before s3s host parsing - // 24. VirtualHostStyleHintLayer — actionable error for unroutable virtual-hosted-style (conditional) - // 25. DoubleSlashListBucketsCompatLayer — rewrites `GET //` to `GET /` for ListBuckets (MinIO browser compat) + // 6. StsQueryApiCompatLayer — route-scoped STS envelopes, including outer short-circuit errors + // 7. EmptyBodyContentLengthCompatLayer — adds Content-Length: 0 for known empty-body API routes + // 8. CatchPanicLayer — panic → 500 + // 9. RateLimitLayer — conditional (external stack only), per-client 429 throttling + // 10. ReadinessGateLayer — blocks until ready + // 11. KeystoneAuthLayer — X-Auth-Token validation + // 12. TraceLayer — request span creation + metrics + // 13. RequestLoggingLayer — single completion event per request + // 14. PropagateRequestIdLayer — X-Request-ID → response + // 15. CompressionLayer — response compression (whitelist, path-aware) + // 16. PathCategoryInjectionLayer — injects path category for compression predicate + // 17. S3ErrorMessageCompatLayer — missing S3 error message compatibility + // 18. IcebergRestErrorCompatLayer — Iceberg REST JSON error compatibility + // 19. ObjectAttributesEtagFixLayer — ETag fix for GetObjectAttributes + // 20. ConditionalCorsLayer — S3 API CORS + // 21. RedirectLayer — console redirect (conditional) + // 22. BodylessStatusFixLayer — clears body for 1xx/204/205/304 responses + // 23. HeadRequestBodyFixLayer — strips actual body bytes from HEAD responses + // 24. PublicHealthEndpointLayer — handles public health before s3s host parsing + // 25. VirtualHostStyleHintLayer — actionable error for unroutable virtual-hosted-style (conditional) + // 26. DoubleSlashListBucketsCompatLayer — rewrites `GET //` to `GET /` for ListBuckets (MinIO browser compat) // ───────────────────────────────────────────────────────────── // Batch 1 intentionally keeps the external and internode stacks behaviorally // identical while giving each path family a named construction boundary. @@ -1431,6 +1432,7 @@ fn process_connection( .option_layer(trusted_proxy_layer.clone()) .layer(SetRequestIdLayer::x_request_id(MakeRequestUuid)) .layer(InternodeRequestContextLiteLayer) + .layer(StsQueryApiCompatLayer) .layer(EmptyBodyContentLengthCompatLayer) .layer(CatchPanicLayer::new()) // Per-client API rate limit (backlog#1191): rejects over-limit diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index 0c1d0aaec..aab384932 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -28,16 +28,19 @@ use crate::storage_api::server::layer::apply_cors_headers; use crate::storage_api::server::layer::request_context::{ RequestContext, extract_request_id_from_headers, extract_trace_context_ids_from_headers, spawn_traced, }; -use bytes::Bytes; +use bytes::{Bytes, BytesMut}; +use futures::future::Either; use http::{HeaderMap, HeaderValue, Method, Request as HttpRequest, Response, StatusCode, Uri}; use http_body::Body; -use http_body_util::BodyExt; +use http_body_util::{BodyExt, Full}; use hyper::body::Incoming; +use quick_xml::events::Event; use rustfs_trusted_proxies::ClientInfo; use rustfs_utils::get_env_opt_str; use rustfs_utils::http::headers::AMZ_REQUEST_ID; use s3s::S3ErrorCode; use serde::{Deserialize, Serialize}; +use std::borrow::Cow; use std::future::Future; use std::pin::Pin; use std::sync::Arc; @@ -56,6 +59,9 @@ const LOG_SUBSYSTEM_HTTP: &str = "http"; const REDACTED_QUERY_VALUE: &str = "redacted"; const OBJECT_ZIP_DOWNLOADS_PATH: &str = "/v3/object-zip-downloads/"; const HTTP_REQUEST_INFLIGHT_WARN_THRESHOLD: Duration = Duration::from_secs(5); +const STS_RESPONSE_METADATA_TAG: &str = "ResponseMetadata"; +const STS_REQUEST_ID_TAG: &str = "RequestId"; +const STS_SUCCESS_RESPONSE_TAGS: [&str; 2] = ["AssumeRoleResponse", "AssumeRoleWithWebIdentityResponse"]; pub(crate) fn redact_sensitive_uri_query(uri: &http::Uri) -> String { let path = uri.path(); @@ -601,12 +607,14 @@ where } fn call(&mut self, req: HttpRequest) -> Self::Future { + let is_sts_query = + req.method() == Method::POST && req.uri().path() == "/" && req.extensions().get::().is_some(); let mut inner = self.inner.clone(); Box::pin(async move { let response = inner.call(req).await?; let (parts, body) = response.into_parts(); - let should_fix = parts.status == StatusCode::FORBIDDEN && is_xml_response(&parts.headers); + let should_fix = !is_sts_query && parts.status == StatusCode::FORBIDDEN && is_xml_response(&parts.headers); let response = match body { HybridBody::Rest { rest_body } => { @@ -629,6 +637,81 @@ where } } +#[derive(Clone)] +pub struct StsQueryApiCompatLayer; + +#[derive(Clone, Copy, Debug)] +pub(crate) struct StsQueryRequest; + +pub(crate) fn is_sts_query_request(method: &Method, uri: &Uri, headers: &HeaderMap) -> bool { + method == Method::POST + && uri.path() == "/" + && headers + .get(http::header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.split(';').next()) + .is_some_and(|value| value.trim().eq_ignore_ascii_case("application/x-www-form-urlencoded")) +} + +impl Layer for StsQueryApiCompatLayer { + type Service = StsQueryApiCompatService; + + fn layer(&self, inner: S) -> Self::Service { + StsQueryApiCompatService { inner } + } +} + +#[derive(Clone)] +pub struct StsQueryApiCompatService { + inner: S, +} + +type StsBoxError = Box; +type StsBoxBody = tower_http::body::UnsyncBoxBody; + +impl Service> for StsQueryApiCompatService +where + S: Service, Response = Response> + Send + 'static, + S::Future: Send + 'static, + S::Error: From + Send + 'static, +{ + type Response = Response; + type Error = S::Error; + type Future = Either> + Send>>>; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.inner.poll_ready(cx) + } + + fn call(&mut self, mut req: HttpRequest) -> Self::Future { + if !is_sts_query_request(req.method(), req.uri(), req.headers()) { + return Either::Left(self.inner.call(req)); + } + + req.extensions_mut().insert(StsQueryRequest); + let (request_id, request_id_header) = sts_response_request_id(&req); + let future = self.inner.call(req); + + Either::Right(Box::pin(async move { + let response = future.await?; + let (mut parts, body) = response.into_parts(); + parts.headers.insert(AMZ_REQUEST_ID, request_id_header); + + let (body, changed) = convert_sts_query_response(body, parts.status, &request_id).await?; + if changed { + parts.headers.remove(http::header::CONTENT_LENGTH); + parts + .headers + .insert(http::header::CONTENT_TYPE, HeaderValue::from_static("application/xml")); + } + let body = tower_http::body::UnsyncBoxBody::new(Full::from(body).map_err(|error| -> StsBoxError { match error {} })); + let response = Response::from_parts(parts, body); + + Ok(response) + })) + } +} + #[derive(Clone)] pub struct IcebergRestErrorCompatLayer; @@ -1338,6 +1421,153 @@ fn is_xml_response(headers: &HeaderMap) -> bool { } } +fn sts_response_request_id(req: &HttpRequest) -> (String, HeaderValue) { + let request_id = req + .extensions() + .get::() + .map(|context| context.request_id.as_str()) + .filter(|request_id| !request_id.trim().is_empty()) + .unwrap_or("sts-request-id") + .to_owned(); + + match HeaderValue::from_str(&request_id) { + Ok(header) => (request_id, header), + Err(_) => ("sts-request-id".to_owned(), HeaderValue::from_static("sts-request-id")), + } +} + +async fn convert_sts_query_response( + body: RestBody, + status: StatusCode, + request_id: &str, +) -> Result<(Bytes, bool), RestBody::Error> +where + RestBody: Body, +{ + let bytes = BodyExt::collect(body).await?.to_bytes(); + let Ok(xml) = std::str::from_utf8(&bytes) else { + return Ok((bytes, false)); + }; + + let converted = if status.is_success() { + add_sts_response_metadata(xml, request_id) + } else { + Some(Bytes::from(wrap_sts_error_response(xml, status, request_id))) + }; + + match converted { + Some(xml) => Ok((xml, true)), + None => Ok((bytes, false)), + } +} + +fn add_sts_response_metadata(xml: &str, request_id: &str) -> Option { + let mut reader = quick_xml::Reader::from_str(xml); + let mut depth = 0_usize; + let mut root_prefix = None; + + loop { + let event_start = usize::try_from(reader.buffer_position()).ok()?; + match reader.read_event() { + Ok(Event::Start(element)) => { + if depth == 0 { + if !is_sts_success_response_tag(element.local_name().as_ref()) { + return None; + } + root_prefix = element + .name() + .prefix() + .map(|prefix| std::str::from_utf8(prefix.as_ref()).map(str::to_owned)) + .transpose() + .ok()?; + } else if depth == 1 && element.local_name().as_ref() == STS_RESPONSE_METADATA_TAG.as_bytes() { + return None; + } + depth += 1; + } + Ok(Event::Empty(element)) => { + if depth == 0 { + return None; + } + if depth == 1 && element.local_name().as_ref() == STS_RESPONSE_METADATA_TAG.as_bytes() { + return None; + } + } + Ok(Event::End(element)) => { + if depth == 0 { + return None; + } + if depth == 1 { + if !is_sts_success_response_tag(element.local_name().as_ref()) { + return None; + } + let prefix = root_prefix.as_deref().map(|prefix| format!("{prefix}:")).unwrap_or_default(); + let metadata = format!( + "<{prefix}{STS_RESPONSE_METADATA_TAG}>\ + <{prefix}{STS_REQUEST_ID_TAG}>{request_id}\ + ", + request_id = xml_escape(request_id), + ); + let capacity = xml.len().checked_add(metadata.len())?; + let mut converted = BytesMut::with_capacity(capacity); + converted.extend_from_slice(&xml.as_bytes()[..event_start]); + converted.extend_from_slice(metadata.as_bytes()); + converted.extend_from_slice(&xml.as_bytes()[event_start..]); + return Some(converted.freeze()); + } + depth -= 1; + } + Ok(Event::Eof) | Err(_) => return None, + _ => {} + } + } +} + +fn is_sts_success_response_tag(tag: &[u8]) -> bool { + STS_SUCCESS_RESPONSE_TAGS.iter().any(|candidate| candidate.as_bytes() == tag) +} + +fn wrap_sts_error_response(xml: &str, status: StatusCode, request_id: &str) -> String { + let parsed = quick_xml::de::from_str::(xml).ok(); + let (code, message) = parsed + .as_ref() + .map(|error| { + let message = match error.message.as_deref() { + Some(message) => Cow::Borrowed(message), + None if error.code == S3ErrorCode::SignatureDoesNotMatch.as_str() => { + Cow::Owned(ApiError::error_code_to_message(&S3ErrorCode::SignatureDoesNotMatch)) + } + None => Cow::Borrowed(error.code.as_str()), + }; + (error.code.as_str(), message) + }) + .unwrap_or_else(|| { + let (code, message) = sts_error_for_status(status); + (code, Cow::Borrowed(message)) + }); + let error_type = if status.is_server_error() { "Receiver" } else { "Sender" }; + + format!( + "\ + \ + {error_type}{code}{message}\ + {request_id}", + code = xml_escape(code), + message = xml_escape(&message), + request_id = xml_escape(request_id), + ) +} + +fn sts_error_for_status(status: StatusCode) -> (&'static str, &'static str) { + match status { + StatusCode::BAD_REQUEST => ("InvalidRequest", "Invalid Request"), + StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN => ("AccessDenied", "Access Denied"), + StatusCode::TOO_MANY_REQUESTS => ("TooManyRequests", "Request rate limit exceeded"), + StatusCode::SERVICE_UNAVAILABLE => ("ServiceUnavailable", "Service Unavailable"), + _ => ("InternalError", "Internal Server Error"), + } +} + async fn fix_object_attributes_etag_in_xml(body: RestBody) -> Result where RestBody: Body + From, @@ -1780,6 +2010,7 @@ fn rewrite_double_slash_root(uri: &Uri) -> Option { #[cfg(test)] mod tests { use super::*; + use crate::server::compress::{HttpCompressionConfig, PathAwareHttpCompressionPredicate, PathCategoryInjectionLayer}; use crate::server::{FAVICON_PATH, LICENSE, RemoteAddr, VERSION}; use futures::future::{Ready, ready}; use http::Request; @@ -2884,6 +3115,331 @@ mod tests { assert_eq!(bytes, input); } + #[derive(Clone)] + struct FixedStsResponse { + status: StatusCode, + body: Bytes, + upstream_request_id: Option<&'static str>, + } + + impl Service> for FixedStsResponse { + type Response = Response; + type Error = StsBoxError; + type Future = Ready>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, _req: Request) -> Self::Future { + let body = tower_http::body::UnsyncBoxBody::new( + Full::from(self.body.clone()).map_err(|error| -> StsBoxError { match error {} }), + ); + let mut response = Response::builder() + .status(self.status) + .header(http::header::CONTENT_TYPE, "application/xml") + .header(http::header::CONTENT_LENGTH, self.body.len()); + if let Some(request_id) = self.upstream_request_id { + response = response.header(AMZ_REQUEST_ID, request_id); + } + ready(Ok(response.body(body).expect("fixed STS response"))) + } + } + + async fn call_sts_compat_layer( + status: StatusCode, + body: &'static [u8], + upstream_request_id: Option<&'static str>, + ) -> (HeaderMap, String) { + let mut service = StsQueryApiCompatLayer.layer(FixedStsResponse { + status, + body: Bytes::from_static(body), + upstream_request_id, + }); + let mut request = Request::builder() + .method(Method::POST) + .uri("/") + .header(http::header::CONTENT_TYPE, "application/x-www-form-urlencoded") + .body(()) + .expect("STS request"); + request.extensions_mut().insert(RequestContext { + request_id: "server-request-id".to_string(), + x_amz_request_id: "untrusted-upstream-id".to_string(), + trace_id: None, + span_id: None, + start_time: Instant::now(), + }); + + let response = service.call(request).await.expect("STS compatibility response"); + let headers = response.headers().clone(); + let body = BodyExt::collect(response.into_body()) + .await + .expect("collect STS response") + .to_bytes(); + (headers, String::from_utf8(body.to_vec()).expect("STS response should be UTF-8 XML")) + } + + #[tokio::test] + async fn sts_compat_layer_uses_canonical_request_id_for_success_header_and_xml() { + let (headers, body) = call_sts_compat_layer( + StatusCode::OK, + b"", + Some("upstream-request-id"), + ) + .await; + + assert_eq!( + headers.get(AMZ_REQUEST_ID).and_then(|value| value.to_str().ok()), + Some("server-request-id") + ); + assert!(!headers.contains_key(http::header::CONTENT_LENGTH)); + assert!(body.contains("server-request-id")); + assert!(!body.contains("untrusted-upstream-id")); + } + + #[tokio::test] + async fn sts_compat_layer_uses_same_request_id_for_error_header_and_xml() { + let (headers, body) = call_sts_compat_layer( + StatusCode::FORBIDDEN, + b"AccessDeniedAccess Denied", + None, + ) + .await; + + assert_eq!( + headers.get(AMZ_REQUEST_ID).and_then(|value| value.to_str().ok()), + Some("server-request-id") + ); + assert!(body.contains("")); + assert!(body.contains("server-request-id")); + } + + #[test] + fn sts_response_request_id_rejects_empty_context_id() { + let mut request = Request::new(()); + request.extensions_mut().insert(RequestContext { + request_id: String::new(), + x_amz_request_id: String::new(), + trace_id: None, + span_id: None, + start_time: Instant::now(), + }); + + let (request_id, header) = sts_response_request_id(&request); + assert_eq!(request_id, "sts-request-id"); + assert_eq!(header, HeaderValue::from_static("sts-request-id")); + } + + #[test] + fn sts_query_route_match_requires_post_root_and_form_content_type() { + let uri = Uri::from_static("/"); + let mut headers = HeaderMap::new(); + + assert!(!is_sts_query_request(&Method::POST, &uri, &headers)); + + headers.insert(http::header::CONTENT_TYPE, HeaderValue::from_static("application/json")); + assert!(!is_sts_query_request(&Method::POST, &uri, &headers)); + + headers.insert( + http::header::CONTENT_TYPE, + HeaderValue::from_static("application/x-www-form-urlencoded; charset=utf-8"), + ); + assert!(is_sts_query_request(&Method::POST, &uri, &headers)); + assert!(!is_sts_query_request(&Method::GET, &uri, &headers)); + assert!(!is_sts_query_request(&Method::POST, &Uri::from_static("/bucket"), &headers)); + } + + #[test] + fn sts_success_response_gets_response_metadata() { + let xml = "\ + "; + let converted = add_sts_response_metadata(xml, "request-123").expect("AssumeRole response should be converted"); + let converted = std::str::from_utf8(&converted).expect("converted response should be UTF-8 XML"); + + assert!(converted.contains("")); + assert!(converted.contains("request-123")); + assert!(converted.ends_with("")); + } + + #[test] + fn sts_web_identity_success_response_gets_response_metadata() { + let xml = "\ + \ + "; + let converted = + add_sts_response_metadata(xml, "request-123").expect("AssumeRoleWithWebIdentity response should be converted"); + let converted = std::str::from_utf8(&converted).expect("converted response should be UTF-8 XML"); + + assert!(converted.contains("")); + assert!(converted.contains("request-123")); + assert!(converted.ends_with("")); + } + + #[test] + fn sts_namespaced_success_response_gets_namespaced_metadata() { + let xml = "\ + "; + let converted = add_sts_response_metadata(xml, "request-123").expect("namespaced response should be converted"); + let converted = std::str::from_utf8(&converted).expect("converted response should be UTF-8 XML"); + + assert!(converted.contains("request-123")); + assert!(converted.ends_with("")); + } + + #[test] + fn sts_success_response_with_existing_metadata_is_unchanged() { + let xml = "\ + \ + existing\ + "; + + assert!(add_sts_response_metadata(xml, "request-123").is_none()); + } + + #[test] + fn sts_error_response_uses_query_api_envelope() { + let converted = wrap_sts_error_response( + "AccessDeniedAccess & Denied", + StatusCode::FORBIDDEN, + "request-456", + ); + + assert!(converted.contains("")); + assert!(converted.contains("Sender")); + assert!(converted.contains("AccessDenied")); + assert!(converted.contains("Access & Denied")); + assert!(converted.contains("request-456")); + } + + #[test] + fn sts_signature_error_response_fills_missing_message() { + let converted = + wrap_sts_error_response("SignatureDoesNotMatch", StatusCode::FORBIDDEN, "request-456"); + let expected = ApiError::error_code_to_message(&S3ErrorCode::SignatureDoesNotMatch); + + assert!(converted.contains(&format!("{expected}"))); + } + + #[tokio::test] + async fn sts_compat_layer_wraps_non_xml_short_circuit_errors() { + let (headers, body) = + call_sts_compat_layer(StatusCode::SERVICE_UNAVAILABLE, b"Service not ready: waiting for iam", None).await; + + assert_eq!( + headers.get(AMZ_REQUEST_ID).and_then(|value| value.to_str().ok()), + Some("server-request-id") + ); + assert!(body.contains("Receiver")); + assert!(body.contains("ServiceUnavailable")); + assert!(body.contains("Service Unavailable")); + assert!(body.contains("server-request-id")); + assert!(!body.contains("waiting for iam")); + } + + #[tokio::test] + async fn sts_compat_layer_replaces_upstream_and_duplicate_error_request_ids() { + let (headers, body) = call_sts_compat_layer( + StatusCode::FORBIDDEN, + b"AccessDeniedAccess Denied\ + body-request-id-1body-request-id-2", + Some("header-request-id"), + ) + .await; + + assert_eq!( + headers.get(AMZ_REQUEST_ID).and_then(|value| value.to_str().ok()), + Some("server-request-id") + ); + assert_eq!(body.matches("").count(), 1); + assert!(body.contains("server-request-id")); + for stale_request_id in ["header-request-id", "body-request-id-1", "body-request-id-2"] { + assert!(!body.contains(stale_request_id)); + } + } + + #[tokio::test] + async fn sts_compat_layer_falls_back_for_empty_and_malformed_error_xml() { + for (body, status, expected_code, expected_message) in [ + ( + b"".as_slice(), + StatusCode::SERVICE_UNAVAILABLE, + "ServiceUnavailable", + "Service Unavailable", + ), + ( + b"AccessDenied".as_slice(), + StatusCode::BAD_GATEWAY, + "InternalError", + "Internal Server Error", + ), + ] { + let (headers, body) = call_sts_compat_layer(status, body, None).await; + + assert_eq!( + headers.get(AMZ_REQUEST_ID).and_then(|value| value.to_str().ok()), + Some("server-request-id") + ); + assert!(body.contains("Receiver")); + assert!(body.contains(&format!("{expected_code}"))); + assert!(body.contains(&format!("{expected_message}"))); + assert_eq!(body.matches("").count(), 1); + assert!(body.contains("server-request-id")); + } + } + + #[tokio::test] + async fn sts_compat_layer_prevents_pre_conversion_compression() { + let xml = format!( + "{}", + "x".repeat(2048) + ); + let mut service = tower::ServiceBuilder::new() + .layer(StsQueryApiCompatLayer) + .layer(tower_http::catch_panic::CatchPanicLayer::new()) + .layer( + tower_http::compression::CompressionLayer::new().compress_when(PathAwareHttpCompressionPredicate::new( + HttpCompressionConfig { + enabled: true, + extensions: Vec::new(), + mime_patterns: vec!["application/xml".to_owned()], + min_size: 0, + }, + )), + ) + .layer(PathCategoryInjectionLayer) + .service(FixedStsResponse { + status: StatusCode::OK, + body: Bytes::from(xml), + upstream_request_id: None, + }); + let mut request = Request::builder() + .method(Method::POST) + .uri("/") + .header(http::header::CONTENT_TYPE, "application/x-www-form-urlencoded") + .header(http::header::ACCEPT_ENCODING, "gzip") + .body(()) + .expect("STS request"); + request.extensions_mut().insert(RequestContext { + request_id: "server-request-id".to_string(), + x_amz_request_id: String::new(), + trace_id: None, + span_id: None, + start_time: Instant::now(), + }); + + let response = service.call(request).await.expect("STS compatibility response"); + assert!( + !response.headers().contains_key(http::header::CONTENT_ENCODING), + "STS XML must stay uncompressed until the compatibility envelope is complete" + ); + let body = BodyExt::collect(response.into_body()) + .await + .expect("collect STS response") + .to_bytes(); + let body = String::from_utf8(body.to_vec()).expect("STS response should be UTF-8 XML"); + assert!(body.contains("server-request-id")); + } + #[tokio::test] async fn iceberg_rest_error_conversion_returns_standard_json_envelope() { let body = Full::from(Bytes::from_static( diff --git a/rustfs/src/server/mod.rs b/rustfs/src/server/mod.rs index ed54ceb89..8c2af988a 100644 --- a/rustfs/src/server/mod.rs +++ b/rustfs/src/server/mod.rs @@ -58,7 +58,7 @@ pub(crate) use health::{ pub(crate) use health::{HealthProbe, build_health_response_parts, collect_probe_readiness, probe_from_path}; pub(crate) use http::HeaderMapCarrier; pub(crate) use http::active_http_requests; -pub(crate) use layer::RequestContextLayer; +pub(crate) use layer::{RequestContextLayer, is_sts_query_request}; pub(crate) use module_switch::{ MODULE_SWITCHES_SIGNAL_SUBSYSTEM, ModuleSwitchSnapshot, ModuleSwitchSource, PersistedModuleSwitches, current_module_switch_snapshot, refresh_persisted_module_switches_from, refresh_persisted_module_switches_from_store, diff --git a/rustfs/src/storage/request_context.rs b/rustfs/src/storage/request_context.rs index 6f3965140..e096a332a 100644 --- a/rustfs/src/storage/request_context.rs +++ b/rustfs/src/storage/request_context.rs @@ -164,15 +164,26 @@ pub fn extract_request_id_from_headers(headers: &HeaderMap) -> String { let request_id = headers .get(REQUEST_ID_HEADER) .and_then(|v| v.to_str().ok()) + .filter(|value| !value.trim().is_empty()) .map(String::from) - .or_else(|| headers.get(AMZ_REQUEST_ID).and_then(|v| v.to_str().ok()).map(String::from)) - .unwrap_or_else(generate_fallback_request_id); + .or_else(|| { + headers + .get(AMZ_REQUEST_ID) + .and_then(|v| v.to_str().ok()) + .filter(|value| !value.trim().is_empty()) + .map(String::from) + }); - if !headers.contains_key(REQUEST_ID_HEADER) && !headers.contains_key(AMZ_REQUEST_ID) { - counter!("rustfs_log_chain_fallback_request_id_total", "source" => "headers_missing").increment(1); + if request_id.is_none() { + let source = if headers.contains_key(REQUEST_ID_HEADER) || headers.contains_key(AMZ_REQUEST_ID) { + "headers_empty_or_invalid" + } else { + "headers_missing" + }; + counter!("rustfs_log_chain_fallback_request_id_total", "source" => source).increment(1); } - request_id + request_id.unwrap_or_else(generate_fallback_request_id) } /// Spawn a request-internal task that inherits the current tracing span. @@ -283,6 +294,21 @@ mod tests { assert_eq!(id, "x-req-789"); } + #[test] + fn test_extract_request_id_ignores_empty_header_values() { + let mut headers = HeaderMap::new(); + headers.insert("x-request-id", http::HeaderValue::from_static("")); + headers.insert("x-amz-request-id", http::HeaderValue::from_static("amz-req-000")); + assert_eq!(extract_request_id_from_headers(&headers), "amz-req-000"); + + headers.insert( + "x-amz-request-id", + http::HeaderValue::from_bytes(b" \t").expect("optional whitespace is a valid header value"), + ); + let id = extract_request_id_from_headers(&headers); + assert!(id.starts_with("req-"), "empty request ID headers must generate a fallback ID"); + } + #[test] fn test_extract_trace_context_ids_from_traceparent_header() { global::set_text_map_propagator(TraceContextPropagator::new());