diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index c25c64e58..e21c85484 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -6175,6 +6175,7 @@ pub(crate) mod rename_fanout_barrier { Cleanup, Rollback, ReadVersion, + MetadataSlowtailFault, NonInlineHedgeTimer, } @@ -6185,6 +6186,7 @@ pub(crate) mod rename_fanout_barrier { super::rename_fanout_barrier_phase::CLEANUP => Self::Cleanup, super::rename_fanout_barrier_phase::ROLLBACK => Self::Rollback, super::rename_fanout_barrier_phase::READ_VERSION => Self::ReadVersion, + "metadata_slowtail_fault" => Self::MetadataSlowtailFault, "non_inline_hedge_timer" => Self::NonInlineHedgeTimer, _ => panic!("unknown commit fault phase: {label}"), } @@ -6195,6 +6197,7 @@ pub(crate) mod rename_fanout_barrier { pub const PHASE_CLEANUP: Phase = Phase::Cleanup; pub const PHASE_ROLLBACK: Phase = Phase::Rollback; pub const PHASE_READ_VERSION: Phase = Phase::ReadVersion; + pub const PHASE_METADATA_SLOWTAIL_FAULT: Phase = Phase::MetadataSlowtailFault; /// Object-scoped hedge timer checkpoint; slot zero identifies the timer, not a disk. pub const PHASE_NON_INLINE_HEDGE_TIMER: Phase = Phase::NonInlineHedgeTimer; @@ -7059,18 +7062,46 @@ mod tests { (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")), ], async { + let slow_fault = + rename_fanout_barrier::arm(object, slow_disk, rename_fanout_barrier::PHASE_METADATA_SLOWTAIL_FAULT); + // Park the initial healthy reads until the slowtail hook is active, + // so cancellation cannot race ahead of fault injection on a busy runner. + let initial_reads = order + .iter() + .take(DISKS - 1) + .copied() + .filter(|&index| index != slow_disk) + .map(|index| rename_fanout_barrier::arm(object, index, rename_fanout_barrier::PHASE_READ_VERSION)) + .collect::>(); let calls = disk_call_counters::observe(object); - let read_with_data = - SetDisks::read_all_fileinfo_observed(&disks, bucket, bucket, object, "", true, false, false, true, 2); - let (parts_metadata, errs, diagnostics) = tokio::time::timeout(Duration::from_millis(300), read_with_data) + let tracker = rename_fanout_barrier::observe_tasks(object); + let disks_for_read = disks.clone(); + let mut read = tokio::spawn(async move { + SetDisks::read_all_fileinfo_observed(&disks_for_read, bucket, bucket, object, "", true, false, false, true, 2) + .await + }); + tokio::time::timeout(BARRIER_PAUSE_GUARD, slow_fault.wait_until_paused()) .await - .expect("gated metadata read should hedge the initial slow shard") + .expect("initial slow shard should reach the slowtail fault hook"); + for initial_read in &initial_reads { + initial_read.release(); + } + + // Keep the slow shard parked: only a successful hedge can finish + // this read. The timeout bounds a hang, not disk or scheduler latency. + let (parts_metadata, errs, diagnostics) = tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read) + .await + .expect("gated metadata read should hedge the paused initial slow shard") + .expect("metadata read task should join") .expect("gated metadata fanout should resolve"); - assert!(parts_metadata.iter().filter(|fi| fi.name == object).count() >= 3); + assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS - 1); + assert!(parts_metadata[slow_disk].name.is_empty(), "the slow shard must not supply a response"); + assert_eq!(parts_metadata[spare_disk].name, object); assert!(errs.iter().all(Option::is_none)); - assert!(diagnostics.total_responses() < DISKS); + assert_eq!(diagnostics.total_responses(), DISKS - 1); assert_eq!(calls.for_disk(disk_call_counters::KIND_METADATA_SLOWTAIL_FAULT, slow_disk), 1); assert_eq!(calls.for_disk(disk_call_counters::KIND_READ_VERSION, spare_disk), 1); + assert_eq!(tracker.running(), 0, "early-stop must cancel and drain the paused slow shard"); }, ) .await; diff --git a/crates/ecstore/src/set_disk/core/metadata_read.rs b/crates/ecstore/src/set_disk/core/metadata_read.rs index 82dd3b8a4..ab1700b51 100644 --- a/crates/ecstore/src/set_disk/core/metadata_read.rs +++ b/crates/ecstore/src/set_disk/core/metadata_read.rs @@ -1145,6 +1145,9 @@ impl SetDisks { Self::read_version_fanout_barrier(&object, index).await; if let Some(delay) = slowtail_fault.as_ref().and_then(|fault| fault.delay_for_disk(index)) { Self::record_metadata_slowtail_fault(&object, index); + #[cfg(test)] + rename_fanout_barrier::checkpoint(&object, index, rename_fanout_barrier::PHASE_METADATA_SLOWTAIL_FAULT) + .await; tokio::time::sleep(delay).await; } read_version_via_coalescer(disk, &org_bucket, &bucket, &object, &version_id, &task_opts, allow_coalescing) diff --git a/rustfs/src/app/lifecycle_transition_api_test.rs b/rustfs/src/app/lifecycle_transition_api_test.rs index cc4b3a9c7..4af5540b1 100644 --- a/rustfs/src/app/lifecycle_transition_api_test.rs +++ b/rustfs/src/app/lifecycle_transition_api_test.rs @@ -26,7 +26,9 @@ use super::storage_api::test::contract::{ }; use super::storage_api::test::ecfs::FS; use super::storage_api::test::object_utils::to_s3s_etag; -use super::storage_api::test::runtime::{MockWarmBackend, MockWarmOp, register_mock_tier as register_mock_tier_util}; +use super::storage_api::test::runtime::{ + MockWarmBackend, MockWarmOp, get_global_transition_state, register_mock_tier as register_mock_tier_util, +}; use super::storage_api::test::{ ECStore, Endpoint, EndpointServerPools, Endpoints, PoolEndpoints, StorageObjectInfo as ObjectInfo, StorageObjectOptions as ObjectOptions, StoragePutObjReader as PutObjReader, install_all_v6_fleet_capability_proof, @@ -1687,6 +1689,11 @@ async fn compensation_driven_copy_still_completes_transition() { .build() .unwrap(); + let transition_state = get_global_transition_state(); + let missed_before = transition_state.missed_immediate_tasks(); + let compensation_before = transition_state.compensation_scheduled_tasks(); + let put_barrier = backend.arm_put_barrier().await; + with_forced_immediate_enqueue_timeout(|| async { Box::pin(usecase.execute_copy_object(build_request(copy_input, Method::PUT))) .await @@ -1694,13 +1701,34 @@ async fn compensation_driven_copy_still_completes_transition() { }) .await; + assert!( + transition_state.missed_immediate_tasks() > missed_before, + "copy should exercise the forced immediate enqueue failure" + ); + assert!( + transition_state.compensation_scheduled_tasks() > compensation_before, + "copy should schedule automatic compensation backfill" + ); + + // Observe automatic compensation reaching the tier before timing its final + // metadata commit; discovery and queue latency are separate from that commit. + put_barrier.wait_until_paused().await; + assert_eq!( + backend.object_count().await, + 1, + "compensation should store the copied body before committing" + ); + put_barrier.release(); + let info = wait_for_transition(&ecstore, dst_bucket.as_str(), dst_object, TRANSITION_WAIT_TIMEOUT) .await .expect("copied object should eventually transition after compensation backfill"); + assert_eq!(backend.put_count().await, 1, "compensation should upload the copied version once"); assert_eq!(info.transitioned_object.status, "complete"); assert_eq!(info.transitioned_object.tier, tier_name); assert!(backend.contains(&info.transitioned_object.name).await); + assert_eq!(read_object_bytes(&ecstore, dst_bucket.as_str(), dst_object).await, payload); } #[serial] diff --git a/rustfs/src/auth.rs b/rustfs/src/auth.rs index 4f1fbc36a..aa25b9432 100644 --- a/rustfs/src/auth.rs +++ b/rustfs/src/auth.rs @@ -450,6 +450,8 @@ pub async fn check_key_valid_with_context( return Err(s3_error!(InvalidAccessKeyId, "check key failed")); } + check_session_token(session_token, &u.credentials)?; + warn!( event = EVENT_SECRET_KEY_LOOKUP_FAILED, component = LOG_COMPONENT_AUTH, @@ -468,8 +470,7 @@ pub async fn check_key_valid_with_context( cred = u.credentials; } - let claims = check_claims_from_token_with_context(session_token, &cred, ctx) - .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("check claims failed {e}")))?; + let claims = check_claims_from_token_with_context(session_token, &cred, ctx)?; cred.claims = if !claims.is_empty() { Some(claims) } else { None }; @@ -481,30 +482,39 @@ pub fn check_claims_from_token(token: &str, cred: &Credentials) -> S3Result S3Result<()> { + // Session identity survives expiration; is_temp() only describes active credentials. + let has_session_token = !cred.session_token.is_empty(); + + if !token.is_empty() && cred.access_key.is_empty() { + return Err(s3_error!(InvalidRequest, "no access key")); + } + + if token.is_empty() && has_session_token && !cred.is_service_account() { + return Err(s3_error!(InvalidRequest, "invalid token1")); + } + + if !token.is_empty() && !has_session_token { + return Err(s3_error!(InvalidRequest, "invalid token2")); + } + + if !cred.is_service_account() && has_session_token && !constant_time_eq(token, &cred.session_token) { + return Err(s3_error!(InvalidRequest, "invalid token3")); + } + + if has_session_token && !cred.is_service_account() && cred.is_expired() { + return Err(s3_error!(ExpiredToken, "The provided token has expired.")); + } + + Ok(()) +} + fn check_claims_from_token_with_context( token: &str, cred: &Credentials, ctx: Option<&AppContext>, ) -> S3Result> { - if !token.is_empty() && cred.access_key.is_empty() { - return Err(s3_error!(InvalidRequest, "no access key")); - } - - if token.is_empty() && cred.is_temp() && !cred.is_service_account() { - return Err(s3_error!(InvalidRequest, "invalid token1")); - } - - if !token.is_empty() && !cred.is_temp() { - return Err(s3_error!(InvalidRequest, "invalid token2")); - } - - if !cred.is_service_account() && cred.is_temp() && !constant_time_eq(token, &cred.session_token) { - return Err(s3_error!(InvalidRequest, "invalid token3")); - } - - if cred.is_temp() && cred.is_expired() { - return Err(s3_error!(InvalidRequest, "invalid access key is temp and expired")); - } + check_session_token(token, cred)?; let sys_cred = match ctx { Some(context) => context.action_credentials().get(), @@ -1611,22 +1621,118 @@ mod tests { #[test] fn test_check_claims_from_token_expired_credentials() { let mut cred = create_temp_credentials(); - cred.expiration = Some(OffsetDateTime::now_utc() - time::Duration::hours(1)); // Expired - cred.claims = None; // Make sure it's not a service account + cred.expiration = Some(OffsetDateTime::now_utc() - time::Duration::hours(1)); - let result = check_claims_from_token(&cred.session_token, &cred); + let error = check_claims_from_token(&cred.session_token, &cred).unwrap_err(); + assert_eq!(error.code(), &S3ErrorCode::ExpiredToken); + let response = error.to_http_response().expect("serialize the S3 error"); + assert_eq!(response.status(), http::StatusCode::BAD_REQUEST); + let body = response.body().bytes().expect("error XML body"); + let xml = std::str::from_utf8(&body).expect("UTF-8 error XML"); + assert!(xml.contains("ExpiredToken")); + assert!(!xml.contains(&cred.session_token)); + assert!(!xml.contains(&cred.secret_key)); + } - assert!(result.is_err()); - let error = result.unwrap_err(); - assert_eq!(error.code(), &S3ErrorCode::InvalidRequest); + #[test] + fn expired_sts_credentials_require_the_matching_token() { + let mut cred = create_temp_credentials(); + cred.expiration = Some(OffsetDateTime::now_utc() - time::Duration::hours(1)); + for token in ["", "wrong-session-token"] { + let error = check_claims_from_token(token, &cred).unwrap_err(); + assert_eq!(error.code(), &S3ErrorCode::InvalidRequest); + } + } - // The function checks various conditions in order. An expired temp credential - // might trigger other validation errors first (like token mismatch) - let msg = error.message().unwrap_or(""); - let is_valid_error = msg.contains("invalid access key is temp and expired") - || msg.contains("invalid token") - || msg.contains("action cred not init"); - assert!(is_valid_error, "Unexpected error message: '{msg}'"); + #[tokio::test] + async fn check_key_valid_preserves_sts_error_codes() { + use rustfs_iam::{cache::Cache, manager::IamCache}; + use rustfs_policy::auth::UserIdentity; + + let (_temp_dir, _disk_paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await; + let (send_chan, _receiver) = tokio::sync::mpsc::channel(1); + let iam_cache = Arc::new(IamCache { + cache: Cache::default(), + api: ObjectStore::new(store.clone()), + state: Default::default(), + loading: Default::default(), + roles: HashMap::new(), + send_chan, + last_timestamp: Default::default(), + sync_failures: Default::default(), + sync_successes: Default::default(), + last_sync_duration_millis: Default::default(), + }); + let iam = Arc::new(IamSys::new(iam_cache.clone())); + let context = AppContext::new(store, Arc::new(ContextIam { handle: iam }), Arc::new(TestKms)); + let root = Credentials { + access_key: "sts-test-root".to_string(), + secret_key: "sts-test-signing-secret".to_string(), + status: "on".to_string(), + ..Default::default() + }; + assert!(context.publish_action_credentials(root.clone())); + let now = OffsetDateTime::now_utc(); + let claims = HashMap::from([("exp".to_string(), json!((now + time::Duration::hours(1)).unix_timestamp()))]); + let mut valid = get_new_credentials_with_metadata(&claims, &root.secret_key).expect("signed STS credentials"); + valid.parent_user = root.access_key.clone(); + iam_cache + .cache + .add_or_update_sts_account(&valid.access_key, &UserIdentity::new(valid.clone()), now); + assert!( + check_key_valid_with_context(&valid.session_token, &valid.access_key, Some(&context)) + .await + .is_ok() + ); + assert!( + check_key_valid_with_context("", &root.access_key, Some(&context)) + .await + .is_ok() + ); + + // Seed the state of an issued identity after expiration without sleeping + // or changing the clock used by signature verification. + let expired_claims = HashMap::from([("exp".to_string(), json!((now - time::Duration::hours(1)).unix_timestamp()))]); + let mut expired = + get_new_credentials_with_metadata(&expired_claims, &root.secret_key).expect("expired signed STS credentials"); + expired.parent_user = root.access_key.clone(); + let mut disabled = expired.clone(); + disabled.status = "off".to_string(); + let mut service_account = create_service_account_credentials(); + service_account.expiration = expired.expiration; + let mut invalid_key = create_test_credentials(); + invalid_key.secret_key = "short".to_string(); + for (name, credential, token, expected) in [ + ("expired", &expired, expired.session_token.as_str(), S3ErrorCode::ExpiredToken), + ("expired without token", &expired, "", S3ErrorCode::InvalidRequest), + ("expired with wrong token", &expired, "wrong-token", S3ErrorCode::InvalidRequest), + ( + "disabled expired", + &disabled, + disabled.session_token.as_str(), + S3ErrorCode::InvalidAccessKeyId, + ), + ("expired service account", &service_account, "", S3ErrorCode::InvalidRequest), + ("active without token", &valid, "", S3ErrorCode::InvalidRequest), + ("active with wrong token", &valid, "wrong-token", S3ErrorCode::InvalidRequest), + ("invalid key", &invalid_key, "", S3ErrorCode::InvalidRequest), + ] { + iam_cache.cache.add_or_update_sts_account( + &credential.access_key, + &UserIdentity::new(credential.clone()), + OffsetDateTime::now_utc(), + ); + let error = check_key_valid_with_context(token, &credential.access_key, Some(&context)) + .await + .unwrap_err(); + assert_eq!(error.code(), &expected, "{name}"); + if expected == S3ErrorCode::ExpiredToken { + let response = error.to_http_response().expect("serialize the S3 error"); + assert_eq!(response.status(), http::StatusCode::BAD_REQUEST); + let body = response.body().bytes().expect("error XML body"); + assert!(std::str::from_utf8(&body).unwrap().contains("ExpiredToken")); + } + } } #[test] @@ -1690,12 +1796,10 @@ mod tests { verified.claims.as_ref().and_then(|claims| claims.get("context")), Some(&json!("matching")) ); - assert!( - check_key_valid_with_context(&credential.session_token, &credential.access_key, Some(&mismatching)) - .await - .is_err(), - "a different server context must not validate the token" - ); + let error = check_key_valid_with_context(&credential.session_token, &credential.access_key, Some(&mismatching)) + .await + .expect_err("a different server context must not validate the token"); + assert_eq!(error.code(), &S3ErrorCode::InvalidRequest); } #[test]