feat(kms): AppRole login with background token renewal and fail-closed expiry (#5487)

* feat(kms): add AppRole configuration surface for Vault auth

Extend VaultAuthMethod::AppRole with secret_id_file (re-read on every
login so external rotation is picked up), a configurable auth mount
(default "approle"), and an optional fail-closed safety window. All new
fields are serde(default) so previously persisted configurations keep
deserializing, and the strict admin-configure deserializer accepts them
as optional.

Environment selection: setting RUSTFS_KMS_VAULT_APPROLE_ROLE_ID switches
both Vault backends to AppRole; the secret_id comes from
RUSTFS_KMS_VAULT_APPROLE_SECRET_ID_FILE (path stored, file wins) or
RUSTFS_KMS_VAULT_APPROLE_SECRET_ID, following the static secret-key file
precedent. validate() rejects AppRole configs without a role_id, without
any secret_id source, or with an empty mount.

Also append the CredentialsUnavailable error variant used by the
fail-closed credential gate.

* feat(kms): implement AppRole login with background renewal and fail-closed expiry

Implement the AppRoleLogin token source (vaultrs approle login +
renew-self) and wire lease-bound credentials through the provider:

- Each successful login/renewal installs a new client generation in the
  ArcSwap; in-flight requests finish on the generation they captured.
- A background renewal task refreshes at half the lease TTL: renewable
  tokens are renewed in place, everything else (or a failed renewal)
  falls back to a fresh login. Auth exchanges run under the typed retry
  policy (OpClass::Auth) and failed cycles retry on a fixed cadence, so
  the provider recovers once Vault does.
- Fail-closed: current() refuses to hand out a token inside the
  configured safety window of its expiry (default: one attempt timeout),
  returning CredentialsUnavailable instead of sending a request whose
  token may lapse mid-flight.
- Refreshes are single-flight: concurrent triggers for the same
  generation coalesce into one login.
- The renewal task's owner handle lives on the KMS service version:
  stop() shuts it down explicitly and reconfigure recycles it via
  cancel-on-drop when the old version is discarded.
- The secret_id file is re-read on every login attempt; missing or empty
  files fail the attempt without contacting Vault. Crate-owned copies of
  tokens and secret_ids are zeroized on drop, and Debug output of every
  credential-carrying type stays redacted (leak regression tests).

The renewal machinery is covered by paused-clock tests driving a
scripted token source: renew-at-half-TTL timing, login fallback,
fail-closed window entry and recovery, prompt task recycling, and
coalesced concurrent refreshes.

* feat(kms): add Vault Agent token file authentication

Add the TokenFile source: the token is read from an agent-managed sink
file (RUSTFS_KMS_VAULT_TOKEN_FILE or the TokenFile auth config) and
re-read once per poll interval (default 30s) through the existing
renewal loop, so a token rotated by the agent installs a new client
generation within one poll of the atomic replace. Each successful read
extends the token's observed validity to twice the poll interval; a
file that disappears or turns empty keeps failing the refresh until the
fail-closed window trips, and heals the provider as soon as it is
restored.

Reads are strict and never contact Vault on failure: the file must be
non-empty after trimming, and on Unix group/other permission bits are a
hard error (mirroring the SFTP host-key rule). Rotation detection uses
a content digest; the token itself is never stored on the source and
the crate-owned copy is zeroized.

Configuring the token file together with AppRole or an explicit static
token is rejected as a configuration error. All new config fields are
serde(default) and the strict admin-configure deserializer accepts the
new variant.

Covered by paused-clock tests (atomic replacement installs a new
generation next cycle, deletion fails closed and recovers, prompt task
recycling) plus negatives for missing/empty/over-permissive files and a
Debug leak regression.

* docs(kms): add Vault authentication and credential lifecycle runbook

Cover choosing between static token, AppRole, and Vault Agent token
file auth; AppRole role setup with SecretID delivery and rotation;
Agent sink deployment with the permission requirements; and the
fail-closed window semantics with a troubleshooting table keyed on the
renewal task's log lines.

crates/kms/src/backends/vault.rs

@@ -14,7 +14,10 @@
 
 //! Vault-based KMS backend implementation using vaultrs
 
-use crate::backends::vault_credentials::{VaultClientHandle, VaultConnectionSettings, VaultCredentialProvider, token_source_for};
+use crate::backends::vault_credentials::{
+    CredentialTaskHandle, VaultClientHandle, VaultConnectionSettings, VaultCredentialPolicy, VaultCredentialProvider,
+    token_source_for,
+};
 use crate::backends::{BackendCapabilities, BackendInfo, KmsBackend, KmsClient};
 use crate::config::{KmsConfig, VaultConfig};
 use crate::encryption::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material};
@@ -32,7 +35,7 @@ use vaultrs::{api::kv2::requests::SetSecretRequestOptions, error::ClientError, k
 
 /// Vault KMS client implementation
 pub struct VaultKmsClient {
-    credentials: VaultCredentialProvider,
+    credentials: Arc<VaultCredentialProvider>,
     config: VaultConfig,
     /// Mount path for the KV engine (typically "kv" or "secret")
     kv_mount: String,
@@ -161,15 +164,18 @@ fn decode_stored_key_material(key_id: &str, encrypted_material: &str) -> Result<
 impl VaultKmsClient {
     /// Create a new Vault KMS client
     ///
-    /// `attempt_timeout` caps every HTTP request issued through this client.
-    pub async fn new(config: VaultConfig, attempt_timeout: Duration) -> Result<Self> {
-        let source = token_source_for(&config.auth_method)?;
+    /// `kms_config` supplies the per-attempt timeout that caps every HTTP
+    /// request issued through this client, plus the retry and fail-closed
+    /// budgets for credential refresh.
+    pub async fn new(config: VaultConfig, kms_config: &KmsConfig) -> Result<Self> {
         let settings = VaultConnectionSettings {
             address: config.address.clone(),
             namespace: config.namespace.clone(),
-            attempt_timeout,
+            attempt_timeout: kms_config.effective_timeout(),
         };
-        let credentials = VaultCredentialProvider::new(settings, source).await?;
+        let source = token_source_for(&config.auth_method, &settings)?;
+        let policy = VaultCredentialPolicy::from_kms_config(kms_config, &config.auth_method);
+        let credentials = Arc::new(VaultCredentialProvider::new(settings, source, policy).await?);
 
         info!(address = %config.address, "Vault KMS backend connected");
 
@@ -185,8 +191,9 @@ impl VaultKmsClient {
     /// Snapshot the authenticated Vault client for a single request.
     ///
     /// Every Vault call takes its own snapshot so a credential rotation
-    /// applies to subsequent calls without interrupting in-flight ones.
-    fn vault(&self) -> Arc<VaultClientHandle> {
+    /// applies to subsequent calls without interrupting in-flight ones. Fails
+    /// closed when the credentials could not be refreshed in time.
+    fn vault(&self) -> Result<Arc<VaultClientHandle>> {
         self.credentials.current()
     }
 
@@ -231,7 +238,7 @@ impl VaultKmsClient {
         let path = self.key_version_path(key_id, version);
 
         let record: VaultKeyVersionRecord =
-            kv2::read(&self.vault().client, &self.kv_mount, &path)
+            kv2::read(&self.vault()?.client, &self.kv_mount, &path)
                 .await
                 .map_err(|e| match e {
                     ClientError::ResponseWrapError => KmsError::key_version_not_found(key_id, version),
@@ -271,7 +278,7 @@ impl VaultKmsClient {
     async fn get_key_data_versioned(&self, key_id: &str) -> Result<(u32, VaultKeyData)> {
         let path = self.key_path(key_id);
 
-        let metadata = kv2::read_metadata(&self.vault().client, &self.kv_mount, &path)
+        let metadata = kv2::read_metadata(&self.vault()?.client, &self.kv_mount, &path)
             .await
             .map_err(|e| match e {
                 ClientError::ResponseWrapError => KmsError::key_not_found(key_id),
@@ -283,7 +290,7 @@ impl VaultKmsClient {
 
         // Read the exact secret version from the metadata to keep the (cas, data)
         // pair consistent even if another writer lands in between.
-        let key_data: VaultKeyData = kv2::read_version(&self.vault().client, &self.kv_mount, &path, metadata.current_version)
+        let key_data: VaultKeyData = kv2::read_version(&self.vault()?.client, &self.kv_mount, &path, metadata.current_version)
             .await
             .map_err(|e| match e {
                 ClientError::ResponseWrapError => KmsError::key_not_found(key_id),
@@ -303,7 +310,7 @@ impl VaultKmsClient {
         let path = self.key_path(key_id);
 
         let written =
-            kv2::set_with_options(&self.vault().client, &self.kv_mount, &path, key_data, SetSecretRequestOptions { cas })
+            kv2::set_with_options(&self.vault()?.client, &self.kv_mount, &path, key_data, SetSecretRequestOptions { cas })
                 .await
                 .map_err(|e| {
                     if is_cas_conflict(&e) {
@@ -327,7 +334,8 @@ impl VaultKmsClient {
     async fn try_create_key_version_record(&self, key_id: &str, record: &VaultKeyVersionRecord) -> Result<bool> {
         let path = self.key_version_path(key_id, record.version);
 
-        match kv2::set_with_options(&self.vault().client, &self.kv_mount, &path, record, SetSecretRequestOptions { cas: 0 }).await
+        match kv2::set_with_options(&self.vault()?.client, &self.kv_mount, &path, record, SetSecretRequestOptions { cas: 0 })
+            .await
         {
             Ok(_) => Ok(true),
             Err(e) if is_cas_conflict(&e) => Ok(false),
@@ -339,7 +347,7 @@ impl VaultKmsClient {
     async fn store_key_data(&self, key_id: &str, key_data: &VaultKeyData) -> Result<()> {
         let path = self.key_path(key_id);
 
-        kv2::set(&self.vault().client, &self.kv_mount, &path, key_data)
+        kv2::set(&self.vault()?.client, &self.kv_mount, &path, key_data)
             .await
             .map_err(|e| KmsError::backend_error(format!("Failed to store key in Vault: {e}")))?;
 
@@ -389,7 +397,7 @@ impl VaultKmsClient {
     async fn get_key_data(&self, key_id: &str) -> Result<VaultKeyData> {
         let path = self.key_path(key_id);
 
-        let secret: VaultKeyData = kv2::read(&self.vault().client, &self.kv_mount, &path)
+        let secret: VaultKeyData = kv2::read(&self.vault()?.client, &self.kv_mount, &path)
             .await
             .map_err(|e| match e {
                 vaultrs::error::ClientError::ResponseWrapError => KmsError::key_not_found(key_id),
@@ -404,7 +412,7 @@ impl VaultKmsClient {
     /// List all keys stored in Vault
     async fn list_vault_keys(&self) -> Result<Vec<String>> {
         // List keys under the prefix
-        match kv2::list(&self.vault().client, &self.kv_mount, &self.key_path_prefix).await {
+        match kv2::list(&self.vault()?.client, &self.kv_mount, &self.key_path_prefix).await {
             Ok(keys) => {
                 let keys = filter_key_directory_entries(keys);
                 debug!("Found {} keys in Vault", keys.len());
@@ -431,11 +439,11 @@ impl VaultKmsClient {
         // record still exists and the deletion can be retried. The reverse order
         // would leave orphaned master key material in Vault after the key vanished.
         let versions_dir = self.key_versions_dir(key_id);
-        match kv2::list(&self.vault().client, &self.kv_mount, &versions_dir).await {
+        match kv2::list(&self.vault()?.client, &self.kv_mount, &versions_dir).await {
             Ok(versions) => {
                 for version in versions {
                     let version_path = format!("{versions_dir}/{version}");
-                    kv2::delete_metadata(&self.vault().client, &self.kv_mount, &version_path)
+                    kv2::delete_metadata(&self.vault()?.client, &self.kv_mount, &version_path)
                         .await
                         .map_err(|e| KmsError::backend_error(format!("Failed to delete key version record from Vault: {e}")))?;
                 }
@@ -447,7 +455,7 @@ impl VaultKmsClient {
 
         // For this specific key path, we can safely delete the metadata
         // since each key has its own unique path under the prefix
-        kv2::delete_metadata(&self.vault().client, &self.kv_mount, &path)
+        kv2::delete_metadata(&self.vault()?.client, &self.kv_mount, &path)
             .await
             .map_err(|e| match e {
                 vaultrs::error::ClientError::APIError { code: 404, .. } => KmsError::key_not_found(key_id),
@@ -865,10 +873,17 @@ impl VaultKmsBackend {
             }
         };
 
-        let client = VaultKmsClient::new(vault_config, config.effective_timeout()).await?;
+        let client = VaultKmsClient::new(vault_config, &config).await?;
         Ok(Self { client })
     }
 
+    /// Spawn the background credential renewal task for this backend, if its
+    /// auth method issues lease-bound tokens. The caller owns the returned
+    /// handle; dropping it cancels the task.
+    pub(crate) fn spawn_credential_renewal(&self) -> Option<CredentialTaskHandle> {
+        self.client.credentials.spawn_renewal_task()
+    }
+
     /// Update key metadata in Vault storage
     async fn update_key_metadata_in_storage(&self, key_id: &str, metadata: &KeyMetadata) -> Result<()> {
         // Get the current key data from Vault
@@ -1167,7 +1182,7 @@ mod tests {
             tls: None,
         };
 
-        let client = VaultKmsClient::new(config, Duration::from_secs(30))
+        let client = VaultKmsClient::new(config, &KmsConfig::default())
             .await
             .expect("Failed to create Vault client");
 
@@ -1220,7 +1235,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_key_version_paths_stay_under_the_key() {
-        let client = VaultKmsClient::new(integration_vault_config(), Duration::from_secs(30))
+        let client = VaultKmsClient::new(integration_vault_config(), &KmsConfig::default())
             .await
             .expect("client");
 
@@ -1305,7 +1320,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_vault_kv2_backend_info_reports_at_rest_protection() {
-        let client = VaultKmsClient::new(integration_vault_config(), Duration::from_secs(30))
+        let client = VaultKmsClient::new(integration_vault_config(), &KmsConfig::default())
             .await
             .expect("client");
 
@@ -1337,7 +1352,7 @@ mod tests {
     #[tokio::test]
     #[ignore] // Requires a running Vault instance (dev mode)
     async fn test_vault_kv2_decrypt_after_rotate() {
-        let client = VaultKmsClient::new(integration_vault_config(), Duration::from_secs(30))
+        let client = VaultKmsClient::new(integration_vault_config(), &KmsConfig::default())
             .await
             .expect("client");
 
@@ -1374,7 +1389,7 @@ mod tests {
     #[tokio::test]
     #[ignore] // Requires a running Vault instance (dev mode)
     async fn test_vault_kv2_rotate_does_not_orphan_legacy_envelopes() {
-        let client = VaultKmsClient::new(integration_vault_config(), Duration::from_secs(30))
+        let client = VaultKmsClient::new(integration_vault_config(), &KmsConfig::default())
             .await
             .expect("client");
 
@@ -1413,7 +1428,7 @@ mod tests {
     #[tokio::test]
     #[ignore] // Requires a running Vault instance (dev mode)
     async fn test_vault_kv2_envelope_version_tampering_fails_closed() {
-        let client = VaultKmsClient::new(integration_vault_config(), Duration::from_secs(30))
+        let client = VaultKmsClient::new(integration_vault_config(), &KmsConfig::default())
             .await
             .expect("client");
 
@@ -1456,7 +1471,7 @@ mod tests {
         use std::sync::Arc;
 
         let client = Arc::new(
-            VaultKmsClient::new(integration_vault_config(), Duration::from_secs(30))
+            VaultKmsClient::new(integration_vault_config(), &KmsConfig::default())
                 .await
                 .expect("client"),
         );
@@ -1515,7 +1530,7 @@ mod tests {
         // Regression: get_key_material previously "self-healed" a decrypt/length failure by
         // minting a fresh random master key and overwriting the stored value — destroying the
         // original key and making every DEK wrapped by it permanently undecryptable.
-        let client = VaultKmsClient::new(integration_vault_config(), Duration::from_secs(30))
+        let client = VaultKmsClient::new(integration_vault_config(), &KmsConfig::default())
             .await
             .expect("client");
 
@@ -1553,7 +1568,7 @@ mod tests {
         // bootstrap case and silently generated + persisted a fresh master key on the
         // read path. Empty material must instead fail closed as MaterialMissing and
         // leave the stored record untouched.
-        let client = VaultKmsClient::new(integration_vault_config(), Duration::from_secs(30))
+        let client = VaultKmsClient::new(integration_vault_config(), &KmsConfig::default())
             .await
             .expect("client");