fix(auth): return ExpiredToken for retained STS credentials (#8344)

* fix(auth): return ExpiredToken for retained STS credentials

Validate session identity independently of credential expiry in both IAM
rejection and claims validation, and preserve typed S3 errors at the caller.
Keep token binding, disabled-account precedence, and service-account behavior.
Add exact expiry, IAM, and HTTP/XML regression coverage.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test(ecstore): synchronize the metadata slowtail hedge regression

Park the slow metadata task at a test-only fault checkpoint and release
healthy reads after the fault is active. Assert the spare response and
cancelled-task drain without imposing a 300 ms disk/scheduler deadline.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test(ilm): synchronize copy compensation transition stages

Observe automatic compensation reaching the mock tier PUT checkpoint before
waiting for the final metadata commit. Assert forced enqueue failure,
automatic backfill, one completed tier upload, and payload integrity.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>

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::<Vec<_>>();
                 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;

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)

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]

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<Hash
     check_claims_from_token_with_context(token, cred, None)
 }
 
+fn check_session_token(token: &str, cred: &Credentials) -> 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<HashMap<String, Value>> {
-    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("<Code>ExpiredToken</Code>"));
+        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("<Code>ExpiredToken</Code>"));
+            }
+        }
     }
 
     #[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]