From 3921336b23da9d247e276e05b80298e58d4deab1 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Fri, 31 Jul 2026 06:29:49 +0800 Subject: [PATCH] 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/api_types.rs | 53 +- crates/kms/src/backends/vault.rs | 75 +- crates/kms/src/backends/vault_credentials.rs | 1201 ++++++++++++++++-- crates/kms/src/backends/vault_transit.rs | 68 +- crates/kms/src/config.rs | 409 +++++- crates/kms/src/error.rs | 10 + crates/kms/src/service_manager.rs | 20 +- docs/operations/kms-backend-security.md | 2 + docs/operations/vault-kms-authentication.md | 122 ++ 9 files changed, 1813 insertions(+), 147 deletions(-) create mode 100644 docs/operations/vault-kms-authentication.md diff --git a/crates/kms/src/api_types.rs b/crates/kms/src/api_types.rs index 221529f5c..7d96951c9 100644 --- a/crates/kms/src/api_types.rs +++ b/crates/kms/src/api_types.rs @@ -235,15 +235,55 @@ pub struct StartKmsRequest { #[derive(Deserialize)] #[serde(deny_unknown_fields)] enum StrictVaultAuthMethod { - Token { token: String }, - AppRole { role_id: String, secret_id: String }, + Token { + token: String, + }, + AppRole { + role_id: String, + #[serde(default)] + secret_id: String, + #[serde(default)] + secret_id_file: Option, + #[serde(default)] + mount: Option, + #[serde(default)] + refresh_safety_window_secs: Option, + }, + TokenFile { + path: std::path::PathBuf, + #[serde(default)] + poll_interval_secs: Option, + #[serde(default)] + refresh_safety_window_secs: Option, + }, } impl From for VaultAuthMethod { fn from(value: StrictVaultAuthMethod) -> Self { match value { StrictVaultAuthMethod::Token { token } => Self::Token { token }, - StrictVaultAuthMethod::AppRole { role_id, secret_id } => Self::AppRole { role_id, secret_id }, + StrictVaultAuthMethod::AppRole { + role_id, + secret_id, + secret_id_file, + mount, + refresh_safety_window_secs, + } => Self::AppRole { + role_id, + secret_id, + secret_id_file, + mount: mount.unwrap_or_else(|| crate::config::DEFAULT_VAULT_APPROLE_MOUNT.to_string()), + refresh_safety_window_secs, + }, + StrictVaultAuthMethod::TokenFile { + path, + poll_interval_secs, + refresh_safety_window_secs, + } => Self::TokenFile { + path, + poll_interval_secs, + refresh_safety_window_secs, + }, } } } @@ -403,6 +443,7 @@ impl From<&KmsConfig> for KmsConfigSummary { auth_method_type: match &vault_config.auth_method { VaultAuthMethod::Token { .. } => "token".to_string(), VaultAuthMethod::AppRole { .. } => "approle".to_string(), + VaultAuthMethod::TokenFile { .. } => "token_file".to_string(), }, has_stored_credentials: true, namespace: vault_config.namespace.clone(), @@ -416,6 +457,7 @@ impl From<&KmsConfig> for KmsConfigSummary { auth_method_type: match &vault_config.auth_method { VaultAuthMethod::Token { .. } => "token".to_string(), VaultAuthMethod::AppRole { .. } => "approle".to_string(), + VaultAuthMethod::TokenFile { .. } => "token_file".to_string(), }, has_stored_credentials: true, namespace: vault_config.namespace.clone(), @@ -872,10 +914,7 @@ mod tests { }); let approle = ConfigureKmsRequest::VaultKv2(ConfigureVaultKmsRequest { address: "https://vault.example.com:8200".to_string(), - auth_method: VaultAuthMethod::AppRole { - role_id: "configure-role-id".to_string(), - secret_id: "configure-approle-secret-id".to_string(), - }, + auth_method: VaultAuthMethod::approle("configure-role-id".to_string(), "configure-approle-secret-id".to_string()), namespace: None, mount_path: Some("transit".to_string()), kv_mount: Some("secret".to_string()), diff --git a/crates/kms/src/backends/vault.rs b/crates/kms/src/backends/vault.rs index 97e1bc613..0b77d6e39 100644 --- a/crates/kms/src/backends/vault.rs +++ b/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, 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 { - 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 { 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 { + /// applies to subsequent calls without interrupting in-flight ones. Fails + /// closed when the credentials could not be refreshed in time. + fn vault(&self) -> Result> { 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 { 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 { 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> { // 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 { + 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"); diff --git a/crates/kms/src/backends/vault_credentials.rs b/crates/kms/src/backends/vault_credentials.rs index 486100dcd..066e0bb3a 100644 --- a/crates/kms/src/backends/vault_credentials.rs +++ b/crates/kms/src/backends/vault_credentials.rs @@ -17,21 +17,74 @@ //! [`VaultCredentialProvider`] owns the authenticated [`VaultClient`] and hands //! out per-request snapshots. Backends take a fresh snapshot via //! [`VaultCredentialProvider::current`] for every Vault call instead of holding -//! a client for their own lifetime: a future credential rotation then applies -//! to the next call, while calls already in flight finish on the generation -//! they captured (their `Arc` keeps it alive). +//! a client for their own lifetime: a credential rotation applies to the next +//! call, while calls already in flight finish on the generation they captured +//! (their `Arc` keeps it alive). +//! +//! Lease-bound tokens (AppRole) are kept fresh by a background renewal task +//! (see [`VaultCredentialProvider::spawn_renewal_task`]): it renews at half the +//! lease TTL, falls back to a fresh login when renewal is not possible, and +//! keeps retrying after failures. If the token still reaches the configured +//! safety window before expiry, [`VaultCredentialProvider::current`] fails +//! closed rather than handing out a token that may lapse mid-request. use std::fmt; +use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; use arc_swap::ArcSwap; use async_trait::async_trait; +use sha2::Digest; +use tokio::time::Instant; +use tokio_util::sync::CancellationToken; +use tracing::{info, warn}; use vaultrs::client::{VaultClient, VaultClientSettingsBuilder}; use zeroize::{Zeroize, ZeroizeOnDrop}; -use crate::config::{VaultAuthMethod, redacted_secret}; +use crate::config::{KmsConfig, VaultAuthMethod, redacted_secret}; use crate::error::{KmsError, Result}; +use crate::policy::{self, AttemptError, ErrorClass, OpClass, RetryPolicy}; + +/// Result of a single authentication attempt, classified for the Auth retry +/// policy. +type AttemptResult = std::result::Result; + +/// Cadence for refresh retries after a failed cycle (on top of the bounded +/// retries inside one [`policy::execute`] call). +const DEFAULT_REFRESH_RETRY_INTERVAL: Duration = Duration::from_secs(5); + +/// Default seconds between token file re-reads for [`TokenFileSource`]. +const DEFAULT_TOKEN_FILE_POLL_INTERVAL_SECS: u64 = 30; + +/// A crate-owned secret value, zeroized on drop and redacted in Debug output. +#[derive(Clone, Zeroize, ZeroizeOnDrop)] +pub(crate) struct SecretString(String); + +impl SecretString { + pub(crate) fn new(value: String) -> Self { + Self(value) + } + + pub(crate) fn expose(&self) -> &str { + &self.0 + } +} + +impl fmt::Debug for SecretString { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(redacted_secret(&self.0)) + } +} + +/// Expiry attributes of a lease-bound token. +#[derive(Debug, Clone, Copy)] +pub(crate) struct LeaseInfo { + /// Time-to-live granted at issue or renewal. + pub(crate) ttl: Duration, + /// Whether `renew-self` can extend this token. + pub(crate) renewable: bool, +} /// A Vault token handed out by a [`TokenSource`]. /// @@ -39,52 +92,87 @@ use crate::error::{KmsError, Result}; /// copy `vaultrs` keeps inside its client settings (or the HTTP headers built /// from it); it bounds how long the token lingers in memory owned by this /// module. -/// -/// AppRole login (PR-2) will extend this with the lease metadata returned by -/// the login endpoint (`lease_duration`, `renewable`, accessor). #[derive(Clone, Zeroize, ZeroizeOnDrop)] pub(crate) struct TokenLease { token: String, + /// `None` for tokens without an expiry (static configuration tokens and + /// non-expiring root-like tokens). + #[zeroize(skip)] + lease: Option, } impl TokenLease { - pub(crate) fn new(token: String) -> Self { - Self { token } + pub(crate) fn new(token: String, lease: Option) -> Self { + Self { token, lease } + } + + /// Map a Vault auth response onto a lease. A `lease_duration` of zero + /// means the token never expires, so no lease is tracked and no renewal is + /// scheduled. + fn from_auth(auth: vaultrs::api::AuthInfo) -> Self { + let lease = (auth.lease_duration > 0).then_some(LeaseInfo { + ttl: Duration::from_secs(auth.lease_duration), + renewable: auth.renewable, + }); + Self { + token: auth.client_token, + lease, + } } /// Expose the raw token for handing to the Vault client builder. pub(crate) fn expose(&self) -> &str { &self.token } + + pub(crate) fn lease_info(&self) -> Option { + self.lease + } } impl fmt::Debug for TokenLease { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("TokenLease") .field("token", &redacted_secret(&self.token)) + .field("lease", &self.lease) .finish() } } +/// Wrap a `vaultrs` failure as a classified attempt failure. +fn attempt_error(operation: &str, error: vaultrs::error::ClientError) -> AttemptError { + AttemptError { + class: policy::classify_vaultrs(&error), + error: KmsError::backend_error(format!("Vault {operation} failed: {error}")), + } +} + /// Source of Vault authentication tokens. /// -/// Only [`StaticToken`] exists today. The trait is async and fallible so -/// future sources can perform I/O when acquiring a token without changing the -/// provider: -/// - `AppRoleLogin` (PR-2): performs an `auth/approle/login` round trip and -/// returns the issued token with its lease metadata; -/// - `TokenFile` (PR-3): re-reads an agent-managed token file. +/// Implementations perform one attempt per call; bounded retries and backoff +/// are owned by the caller through [`policy::execute`] with [`OpClass::Auth`]. +/// Current sources are [`StaticToken`], [`AppRoleLogin`], and the agent-managed +/// [`TokenFileSource`]. #[async_trait] pub(crate) trait TokenSource: fmt::Debug + Send + Sync { - /// Acquire a token for a new client generation. + /// One login attempt yielding a token for a new client generation. + async fn acquire(&self) -> AttemptResult; + + /// One `renew-self` attempt for the token held by `client`. /// - /// Called once at provider construction today; rotation (PR-2) will call - /// it again for every re-authentication. - async fn acquire(&self) -> Result; + /// Sources whose tokens cannot be renewed fail fatally; callers fall back + /// to [`TokenSource::acquire`]. + async fn renew(&self, _client: &VaultClient) -> AttemptResult { + Err(AttemptError { + class: ErrorClass::Fatal, + error: KmsError::invalid_operation("this token source does not support renewal"), + }) + } } /// Token source for [`VaultAuthMethod::Token`]: always yields the token fixed -/// at configuration time. +/// at configuration time. The token carries no lease, so it is never renewed +/// and never expires from the provider's point of view. pub(crate) struct StaticToken { token: TokenLease, } @@ -92,14 +180,14 @@ pub(crate) struct StaticToken { impl StaticToken { pub(crate) fn new(token: String) -> Self { Self { - token: TokenLease::new(token), + token: TokenLease::new(token, None), } } } #[async_trait] impl TokenSource for StaticToken { - async fn acquire(&self) -> Result { + async fn acquire(&self) -> AttemptResult { Ok(self.token.clone()) } } @@ -111,16 +199,247 @@ impl fmt::Debug for StaticToken { } } -/// Map the configured auth method onto a token source. +/// Token source for [`VaultAuthMethod::AppRole`]: exchanges `role_id` + +/// `secret_id` for a lease-bound token via the AppRole auth engine. +pub(crate) struct AppRoleLogin { + /// Unauthenticated client used only for the login exchange. + login_client: VaultClient, + mount: String, + role_id: String, + /// Inline secret_id fallback, used when no file is configured. + secret_id: SecretString, + /// Secret-id file, re-read on every login so external rotation of the + /// secret_id is picked up without a restart. Takes precedence over the + /// inline value. + secret_id_file: Option, +} + +impl AppRoleLogin { + pub(crate) fn new( + settings: &VaultConnectionSettings, + mount: String, + role_id: String, + secret_id: String, + secret_id_file: Option, + ) -> Result { + Ok(Self { + login_client: settings.build_login_client()?, + mount, + role_id, + secret_id: SecretString::new(secret_id), + secret_id_file, + }) + } + + /// Resolve the secret_id for one login attempt. + /// + /// File problems are fatal for the attempt (replaying the same read within + /// one retry cycle cannot help), but the renewal loop keeps retrying on + /// its cadence, so repairing the file heals the source without a restart. + async fn resolve_secret_id(&self) -> AttemptResult { + let Some(path) = &self.secret_id_file else { + return Ok(self.secret_id.clone()); + }; + + let mut raw = tokio::fs::read_to_string(path).await.map_err(|error| AttemptError { + class: ErrorClass::Fatal, + error: KmsError::configuration_error(format!("Failed to read AppRole secret_id file {}: {error}", path.display())), + })?; + let trimmed = raw.trim(); + if trimmed.is_empty() { + raw.zeroize(); + return Err(AttemptError { + class: ErrorClass::Fatal, + error: KmsError::configuration_error(format!("AppRole secret_id file {} is empty", path.display())), + }); + } + let secret_id = SecretString::new(trimmed.to_string()); + raw.zeroize(); + Ok(secret_id) + } +} + +#[async_trait] +impl TokenSource for AppRoleLogin { + async fn acquire(&self) -> AttemptResult { + let secret_id = self.resolve_secret_id().await?; + let auth = vaultrs::auth::approle::login(&self.login_client, &self.mount, &self.role_id, secret_id.expose()) + .await + .map_err(|error| attempt_error("AppRole login", error))?; + Ok(TokenLease::from_auth(auth)) + } + + async fn renew(&self, client: &VaultClient) -> AttemptResult { + let auth = vaultrs::token::renew_self(client, None) + .await + .map_err(|error| attempt_error("token renewal", error))?; + Ok(TokenLease::from_auth(auth)) + } +} + +impl fmt::Debug for AppRoleLogin { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + // The login client embeds Vault client settings and must stay out of + // Debug output; role_id is not a secret in Vault's AppRole model. + f.debug_struct("AppRoleLogin") + .field("mount", &self.mount) + .field("role_id", &self.role_id) + .field("secret_id", &self.secret_id) + .field("secret_id_file", &self.secret_id_file) + .finish_non_exhaustive() + } +} + +/// Token source for [`VaultAuthMethod::TokenFile`]: reads an agent-managed +/// token file (for example a Vault Agent auto-auth sink). /// -/// AppRole is still rejected at construction time; PR-2 replaces this arm with -/// an `AppRoleLogin` source. -pub(crate) fn token_source_for(auth_method: &VaultAuthMethod) -> Result> { +/// The agent owns the token's lifecycle; this source only tracks the file. +/// Every read grants the token an observed validity of `validity`, so the +/// renewal loop re-reads the file at half that interval and installs a new +/// client generation from whatever the file holds. 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. +pub(crate) struct TokenFileSource { + path: PathBuf, + /// Observed validity granted per successful read; the renewal loop + /// re-reads at half this value. + validity: Duration, + /// Digest of the last token read, to tell agent-driven rotations apart + /// from plain refreshes in the logs. Never holds the token itself. + last_digest: std::sync::Mutex>, +} + +impl TokenFileSource { + pub(crate) fn new(path: PathBuf, validity: Duration) -> Self { + Self { + path, + validity, + last_digest: std::sync::Mutex::new(None), + } + } + + fn fatal(error: KmsError) -> AttemptError { + AttemptError { + class: ErrorClass::Fatal, + error, + } + } + + /// Reject a token file readable by group or other, mirroring the SFTP + /// host-key rule: a Vault token grants use of the KMS keys, so a mode + /// wider than owner-only is a deployment error, not a warning. + #[cfg(unix)] + fn check_permissions(&self, metadata: &std::fs::Metadata) -> AttemptResult<()> { + use std::os::unix::fs::PermissionsExt; + let mode = metadata.permissions().mode() & 0o777; + if mode & 0o077 != 0 { + return Err(Self::fatal(KmsError::configuration_error(format!( + "Vault token file {} has insecure permissions {mode:#o} (group and other permission bits must be unset)", + self.path.display() + )))); + } + Ok(()) + } +} + +#[async_trait] +impl TokenSource for TokenFileSource { + async fn acquire(&self) -> AttemptResult { + // Synchronous reads on purpose: the token file is tiny, this runs at + // the renewal cadence, and staying off the blocking pool keeps the + // paused-clock tests of the renewal timing deterministic. + let metadata = std::fs::metadata(&self.path).map_err(|error| { + Self::fatal(KmsError::configuration_error(format!( + "Failed to read Vault token file {}: {error}", + self.path.display() + ))) + })?; + #[cfg(unix)] + self.check_permissions(&metadata)?; + + let mut raw = std::fs::read_to_string(&self.path).map_err(|error| { + Self::fatal(KmsError::configuration_error(format!( + "Failed to read Vault token file {}: {error}", + self.path.display() + ))) + })?; + let trimmed = raw.trim(); + if trimmed.is_empty() { + raw.zeroize(); + return Err(Self::fatal(KmsError::configuration_error(format!( + "Vault token file {} is empty", + self.path.display() + )))); + } + + let digest: [u8; 32] = sha2::Sha256::digest(trimmed.as_bytes()).into(); + let previous = self + .last_digest + .lock() + .expect("token file digest mutex poisoned") + .replace(digest); + if previous.is_some_and(|previous| previous != digest) { + info!( + path = %self.path.display(), + modified = ?metadata.modified().ok(), + "Vault token file rotated; installing a new client generation" + ); + } + + let token = trimmed.to_string(); + raw.zeroize(); + Ok(TokenLease::new( + token, + Some(LeaseInfo { + ttl: self.validity, + renewable: false, + }), + )) + } +} + +impl fmt::Debug for TokenFileSource { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + // The token itself is never stored on the source; only its digest is. + f.debug_struct("TokenFileSource") + .field("path", &self.path) + .field("validity", &self.validity) + .finish_non_exhaustive() + } +} + +/// Map the configured auth method onto a token source. +pub(crate) fn token_source_for( + auth_method: &VaultAuthMethod, + settings: &VaultConnectionSettings, +) -> Result> { match auth_method { VaultAuthMethod::Token { token } => Ok(Box::new(StaticToken::new(token.clone()))), - VaultAuthMethod::AppRole { .. } => Err(KmsError::backend_error( - "AppRole authentication not yet implemented. Please use token authentication.", - )), + VaultAuthMethod::AppRole { + role_id, + secret_id, + secret_id_file, + mount, + .. + } => Ok(Box::new(AppRoleLogin::new( + settings, + mount.clone(), + role_id.clone(), + secret_id.clone(), + secret_id_file.clone(), + )?)), + VaultAuthMethod::TokenFile { + path, + poll_interval_secs, + .. + } => { + let poll_interval = Duration::from_secs(poll_interval_secs.unwrap_or(DEFAULT_TOKEN_FILE_POLL_INTERVAL_SECS).max(1)); + // Validity is twice the poll interval because the renewal loop + // fires at half the lease TTL: the file is then re-read once per + // poll interval, and a file that stops being readable expires the + // token after roughly two missed polls minus the safety window. + Ok(Box::new(TokenFileSource::new(path.clone(), poll_interval * 2))) + } } } @@ -135,7 +454,7 @@ pub(crate) struct VaultConnectionSettings { impl VaultConnectionSettings { /// Build an authenticated client for one generation. - fn build_client(&self, token: &TokenLease) -> Result { + fn build_client(&self, token: &str) -> Result { let mut settings_builder = VaultClientSettingsBuilder::default(); settings_builder.address(&self.address); // Defense in depth against stalled connections: vaultrs leaves the @@ -143,7 +462,7 @@ impl VaultConnectionSettings { // request would otherwise wait forever regardless of the // operation-level retry policy. settings_builder.timeout(Some(self.attempt_timeout)); - settings_builder.token(token.expose()); + settings_builder.token(token); if let Some(namespace) = &self.namespace { settings_builder.namespace(Some(namespace.clone())); @@ -155,6 +474,51 @@ impl VaultConnectionSettings { VaultClient::new(settings).map_err(|e| KmsError::backend_error(format!("Failed to create Vault client: {e}"))) } + + /// Build the tokenless client used for login exchanges. + fn build_login_client(&self) -> Result { + self.build_client("") + } +} + +/// Refresh and fail-closed tuning for a [`VaultCredentialProvider`]. +#[derive(Debug, Clone)] +pub(crate) struct VaultCredentialPolicy { + /// Retry budget for one login/renewal cycle. + pub(crate) retry: RetryPolicy, + /// Fail-closed margin: once the current token is within this window of + /// expiry without a successful refresh, [`VaultCredentialProvider::current`] + /// refuses to hand it out. + pub(crate) safety_window: Duration, + /// Pause between refresh cycles after a failed one. + pub(crate) retry_interval: Duration, +} + +impl VaultCredentialPolicy { + /// Derive the policy from the KMS configuration. + /// + /// The default safety window equals the per-attempt timeout: a request + /// issued now can stay in flight for up to one attempt timeout, so the + /// token must outlive at least that. + pub(crate) fn from_kms_config(config: &KmsConfig, auth_method: &VaultAuthMethod) -> Self { + let retry = RetryPolicy::from_config(config); + let safety_window = match auth_method { + VaultAuthMethod::AppRole { + refresh_safety_window_secs: Some(secs), + .. + } + | VaultAuthMethod::TokenFile { + refresh_safety_window_secs: Some(secs), + .. + } => Duration::from_secs(*secs), + _ => retry.attempt_timeout, + }; + Self { + retry, + safety_window, + retry_interval: DEFAULT_REFRESH_RETRY_INTERVAL, + } + } } /// One authenticated client generation. @@ -164,10 +528,27 @@ impl VaultConnectionSettings { /// client out from under an in-flight request. pub(crate) struct VaultClientHandle { /// Monotonic counter identifying the credential generation this client was - /// built from. Static tokens never rotate, so only generation 0 exists - /// today; rotation (PR-2) bumps it on every re-authentication. + /// built from; bumped on every successful refresh. pub(crate) generation: u64, pub(crate) client: VaultClient, + /// When this generation's token was issued (or last renewed). + issued_at: Instant, + /// Lease of this generation's token; `None` when it never expires. + lease: Option, +} + +impl VaultClientHandle { + /// Absolute expiry of this generation's token. + fn expires_at(&self) -> Option { + self.lease.map(|lease| self.issued_at + lease.ttl) + } + + /// When the renewal task should refresh this generation: half the TTL, + /// leaving the second half as budget for retries before the fail-closed + /// window is reached. + fn renew_at(&self) -> Option { + self.lease.map(|lease| self.issued_at + lease.ttl / 2) + } } impl fmt::Debug for VaultClientHandle { @@ -176,46 +557,215 @@ impl fmt::Debug for VaultClientHandle { // never appear in Debug output. f.debug_struct("VaultClientHandle") .field("generation", &self.generation) + .field("lease", &self.lease) .finish_non_exhaustive() } } /// Owns the authenticated Vault client for a backend and hands out /// per-request snapshots. -/// -/// The provider keeps neither the settings nor the source after construction -/// because a static token can never be refreshed. Rotation (PR-2) will retain -/// both and add a refresh path that acquires a fresh lease, rebuilds the -/// client, and stores it under a bumped generation. pub(crate) struct VaultCredentialProvider { + settings: VaultConnectionSettings, + source: Box, + policy: VaultCredentialPolicy, current: ArcSwap, + /// Serializes refreshes so concurrent triggers coalesce into one login. + refresh_lock: tokio::sync::Mutex<()>, } impl VaultCredentialProvider { /// Authenticate with `source` and build the initial client generation. - pub(crate) async fn new(settings: VaultConnectionSettings, source: Box) -> Result { - let lease = source.acquire().await?; - let client = settings.build_client(&lease)?; + pub(crate) async fn new( + settings: VaultConnectionSettings, + source: Box, + policy: VaultCredentialPolicy, + ) -> Result { + let startup_cancel = CancellationToken::new(); + let lease = policy::execute("vault_login", OpClass::Auth, &policy.retry, &startup_cancel, || source.acquire()).await?; + let client = settings.build_client(lease.expose())?; Ok(Self { - current: ArcSwap::from_pointee(VaultClientHandle { generation: 0, client }), + current: ArcSwap::from_pointee(VaultClientHandle { + generation: 0, + client, + issued_at: Instant::now(), + lease: lease.lease_info(), + }), + settings, + source, + policy, + refresh_lock: tokio::sync::Mutex::new(()), }) } - /// Snapshot the current client generation. + /// Snapshot the current generation without the expiry gate. Internal use + /// (renewal scheduling) and tests only; request paths go through + /// [`VaultCredentialProvider::current`]. + pub(crate) fn snapshot(&self) -> Arc { + self.current.load_full() + } + + /// Snapshot the current client generation for a single request. /// /// Take one snapshot per Vault call: the returned `Arc` pins the /// generation for exactly that call, so a concurrent rotation applies to /// the next call without interrupting this one. - pub(crate) fn current(&self) -> Arc { - self.current.load_full() + /// + /// Fails closed when the token is inside the safety window of its expiry: + /// a request signed with such a token could lapse mid-flight, so refusing + /// it locally is strictly safer than an unpredictable remote failure. + pub(crate) fn current(&self) -> Result> { + let handle = self.current.load_full(); + if let Some(expires_at) = handle.expires_at() { + let now = Instant::now(); + if now + self.policy.safety_window >= expires_at { + return Err(KmsError::credentials_unavailable(format!( + "Vault token (generation {}) is within {:?} of expiry and has not been refreshed; refusing to use it", + handle.generation, self.policy.safety_window + ))); + } + } + Ok(handle) + } + + /// Refresh the credentials if generation `observed` is still current. + /// + /// Single-flight: concurrent callers serialize on the refresh lock, and a + /// caller that finds a newer generation already installed returns without + /// touching Vault. Renewable tokens are renewed in place; anything else + /// (or a failed renewal) falls back to a fresh login. + pub(crate) async fn refresh(&self, observed: u64, cancel: &CancellationToken) -> Result<()> { + let _guard = self.refresh_lock.lock().await; + let current = self.snapshot(); + if current.generation != observed { + return Ok(()); + } + + let renewable = current.lease.map(|lease| lease.renewable).unwrap_or(false); + let renewed = if renewable { + match policy::execute("vault_token_renew", OpClass::Auth, &self.policy.retry, cancel, || { + self.source.renew(¤t.client) + }) + .await + { + Ok(lease) => Some(lease), + Err(error @ KmsError::OperationCancelled { .. }) => return Err(error), + Err(error) => { + warn!( + generation = current.generation, + error = %error, + "Vault token renewal failed; falling back to a fresh login" + ); + None + } + } + } else { + None + }; + + let lease = match renewed { + Some(lease) => lease, + None => policy::execute("vault_login", OpClass::Auth, &self.policy.retry, cancel, || self.source.acquire()).await?, + }; + + let client = self.settings.build_client(lease.expose())?; + self.current.store(Arc::new(VaultClientHandle { + generation: current.generation + 1, + client, + issued_at: Instant::now(), + lease: lease.lease_info(), + })); + Ok(()) + } + + /// Spawn the background renewal task for lease-bound credentials. + /// + /// Returns `None` when the current token never expires (static tokens): + /// there is nothing to renew. The returned handle cancels the task when + /// dropped, tying the task's lifetime to whoever owns the handle (the + /// service version that owns this backend). + pub(crate) fn spawn_renewal_task(self: &Arc) -> Option { + self.snapshot().lease?; + let cancel = CancellationToken::new(); + let join = tokio::spawn(renewal_loop(Arc::clone(self), cancel.clone())); + Some(CredentialTaskHandle { + cancel, + join: std::sync::Mutex::new(Some(join)), + }) } } impl fmt::Debug for VaultCredentialProvider { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("VaultCredentialProvider") + .field("source", &self.source) .field("current", &self.current.load()) - .finish() + .field("policy", &self.policy) + .finish_non_exhaustive() + } +} + +/// Drive credential refreshes until cancelled. +/// +/// Each cycle sleeps until the current generation's renewal point (half TTL), +/// then refreshes. A failed cycle logs, waits `retry_interval`, and tries +/// again immediately (the renewal point is already in the past), so the +/// provider keeps trying to recover even after the fail-closed window has +/// been reached. +async fn renewal_loop(provider: Arc, cancel: CancellationToken) { + loop { + let handle = provider.snapshot(); + let Some(renew_at) = handle.renew_at() else { + // The current generation never expires; nothing left to schedule. + return; + }; + tokio::select! { + biased; + _ = cancel.cancelled() => return, + _ = tokio::time::sleep_until(renew_at) => {} + } + + match provider.refresh(handle.generation, &cancel).await { + Ok(()) => {} + Err(KmsError::OperationCancelled { .. }) => return, + Err(error) => { + warn!( + generation = handle.generation, + error = %error, + "Vault credential refresh failed; retrying until the credentials recover" + ); + tokio::select! { + biased; + _ = cancel.cancelled() => return, + _ = tokio::time::sleep(provider.policy.retry_interval) => {} + } + } + } + } +} + +/// Owner handle for a spawned renewal task. +/// +/// Dropping the handle cancels the task, so hanging it off the service version +/// recycles the task on stop and reconfigure without explicit lifecycle calls. +pub(crate) struct CredentialTaskHandle { + cancel: CancellationToken, + join: std::sync::Mutex>>, +} + +impl CredentialTaskHandle { + /// Cancel the renewal task and wait for it to exit. + pub(crate) async fn shutdown(&self) { + self.cancel.cancel(); + let join = self.join.lock().expect("credential task join mutex poisoned").take(); + if let Some(join) = join { + let _ = join.await; + } + } +} + +impl Drop for CredentialTaskHandle { + fn drop(&mut self) { + self.cancel.cancel(); } } @@ -223,8 +773,10 @@ impl fmt::Debug for VaultCredentialProvider { mod tests { use super::*; use crate::config::REDACTED_SECRET; + use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; const TEST_TOKEN: &str = "vault-token-debug-leak-canary"; + const TEST_SECRET_ID: &str = "approle-secret-id-leak-canary"; fn test_settings() -> VaultConnectionSettings { VaultConnectionSettings { @@ -234,19 +786,113 @@ mod tests { } } - /// Building a provider never contacts Vault, so these tests run offline. - async fn test_provider() -> VaultCredentialProvider { - VaultCredentialProvider::new(test_settings(), Box::new(StaticToken::new(TEST_TOKEN.to_string()))) + /// Tight retry budget so paused-clock tests stay deterministic: one + /// attempt per cycle, failed cycles spaced by `retry_interval`. + fn test_policy(safety_window: Duration, retry_interval: Duration) -> VaultCredentialPolicy { + VaultCredentialPolicy { + retry: RetryPolicy { + attempt_timeout: Duration::from_secs(1), + op_deadline: Duration::from_secs(1), + max_attempts: 1, + base_backoff: Duration::from_millis(10), + max_backoff: Duration::from_millis(10), + }, + safety_window, + retry_interval, + } + } + + /// Shared observable state of a [`ScriptedSource`]. + #[derive(Debug, Default)] + struct ScriptedState { + login_calls: AtomicU32, + renew_calls: AtomicU32, + fail_login: AtomicBool, + fail_renew: AtomicBool, + } + + /// Token source with scriptable outcomes for driving the provider without + /// a Vault server. + #[derive(Debug)] + struct ScriptedSource { + state: Arc, + ttl: Duration, + renewable: bool, + login_delay: Duration, + } + + impl ScriptedSource { + fn lease(&self) -> Option { + (!self.ttl.is_zero()).then_some(LeaseInfo { + ttl: self.ttl, + renewable: self.renewable, + }) + } + + fn failure() -> AttemptError { + AttemptError { + class: ErrorClass::RetryableStatus, + error: KmsError::backend_error("scripted auth failure (503)"), + } + } + } + + #[async_trait] + impl TokenSource for ScriptedSource { + async fn acquire(&self) -> AttemptResult { + if !self.login_delay.is_zero() { + tokio::time::sleep(self.login_delay).await; + } + let call = self.state.login_calls.fetch_add(1, Ordering::SeqCst); + if self.state.fail_login.load(Ordering::SeqCst) { + return Err(Self::failure()); + } + Ok(TokenLease::new(format!("scripted-login-{call}"), self.lease())) + } + + async fn renew(&self, _client: &VaultClient) -> AttemptResult { + let call = self.state.renew_calls.fetch_add(1, Ordering::SeqCst); + if self.state.fail_renew.load(Ordering::SeqCst) { + return Err(Self::failure()); + } + Ok(TokenLease::new(format!("scripted-renew-{call}"), self.lease())) + } + } + + async fn scripted_provider( + ttl: Duration, + renewable: bool, + policy: VaultCredentialPolicy, + ) -> (Arc, Arc) { + let state = Arc::new(ScriptedState::default()); + let source = ScriptedSource { + state: state.clone(), + ttl, + renewable, + login_delay: Duration::ZERO, + }; + let provider = VaultCredentialProvider::new(test_settings(), Box::new(source), policy) .await - .expect("provider construction must not require a live Vault") + .expect("scripted provider must build without a live Vault"); + (Arc::new(provider), state) + } + + async fn static_provider() -> VaultCredentialProvider { + VaultCredentialProvider::new( + test_settings(), + Box::new(StaticToken::new(TEST_TOKEN.to_string())), + test_policy(Duration::from_secs(10), Duration::from_secs(5)), + ) + .await + .expect("static provider must build without a live Vault") } #[tokio::test] async fn test_static_token_snapshots_pin_one_generation() { - let provider = test_provider().await; + let provider = static_provider().await; - let first = provider.current(); - let second = provider.current(); + let first = provider.current().expect("static tokens never expire"); + let second = provider.current().expect("static tokens never expire"); assert_eq!(first.generation, 0); assert!( @@ -255,50 +901,457 @@ mod tests { ); } + #[tokio::test] + async fn test_static_token_spawns_no_renewal_task() { + let provider = Arc::new(static_provider().await); + assert!(provider.spawn_renewal_task().is_none(), "a token without a lease has nothing to renew"); + } + #[tokio::test] async fn test_static_token_source_yields_configured_token() { - let source = token_source_for(&VaultAuthMethod::Token { - token: TEST_TOKEN.to_string(), - }) + let settings = test_settings(); + let source = token_source_for( + &VaultAuthMethod::Token { + token: TEST_TOKEN.to_string(), + }, + &settings, + ) .expect("token auth must map to a source"); let lease = source.acquire().await.expect("static acquire cannot fail"); assert_eq!(lease.expose(), TEST_TOKEN); + assert!(lease.lease_info().is_none(), "static tokens must not carry a lease"); } - /// Behavior pin: AppRole keeps failing at construction with the same - /// user-visible message until the login source lands (PR-2). - #[test] - fn test_approle_auth_method_still_rejected() { - let error = token_source_for(&VaultAuthMethod::AppRole { - role_id: "role".to_string(), - secret_id: "approle-secret-canary".to_string(), - }) - .expect_err("approle must stay rejected until the login source lands"); + #[tokio::test] + async fn test_approle_auth_method_maps_to_login_source() { + let settings = test_settings(); + let source = token_source_for(&VaultAuthMethod::approle("role".to_string(), TEST_SECRET_ID.to_string()), &settings) + .expect("approle auth must map to a login source"); - let rendered = error.to_string(); - assert!(rendered.contains("AppRole authentication not yet implemented"), "got: {rendered}"); - assert!(!rendered.contains("approle-secret-canary"), "error must not echo the secret id"); + assert!(format!("{source:?}").contains("AppRoleLogin")); + } + + #[tokio::test(start_paused = true)] + async fn test_renewal_task_renews_at_half_ttl() { + let (provider, state) = scripted_provider( + Duration::from_secs(60), + true, + test_policy(Duration::from_secs(10), Duration::from_secs(5)), + ) + .await; + let task = provider.spawn_renewal_task().expect("lease-bound tokens need renewal"); + + tokio::time::sleep(Duration::from_secs(29)).await; + assert_eq!(state.renew_calls.load(Ordering::SeqCst), 0, "renewal must not run before half TTL"); + assert_eq!(provider.snapshot().generation, 0); + + tokio::time::sleep(Duration::from_secs(2)).await; + assert_eq!(state.renew_calls.load(Ordering::SeqCst), 1, "renewal must run at half TTL"); + assert_eq!(state.login_calls.load(Ordering::SeqCst), 1, "renewable tokens must not re-login"); + assert_eq!(provider.snapshot().generation, 1, "a successful renewal must install a new generation"); + + task.shutdown().await; + } + + #[tokio::test(start_paused = true)] + async fn test_failed_renewal_falls_back_to_login() { + let (provider, state) = scripted_provider( + Duration::from_secs(60), + true, + test_policy(Duration::from_secs(10), Duration::from_secs(5)), + ) + .await; + state.fail_renew.store(true, Ordering::SeqCst); + let task = provider.spawn_renewal_task().expect("renewal task"); + + tokio::time::sleep(Duration::from_secs(31)).await; + assert_eq!(state.renew_calls.load(Ordering::SeqCst), 1, "renewal must be attempted first"); + assert_eq!( + state.login_calls.load(Ordering::SeqCst), + 2, + "failed renewal must fall back to a fresh login" + ); + assert_eq!(provider.snapshot().generation, 1); + + task.shutdown().await; + } + + #[tokio::test(start_paused = true)] + async fn test_current_fails_closed_inside_safety_window_and_recovers() { + let (provider, state) = scripted_provider( + Duration::from_secs(60), + true, + test_policy(Duration::from_secs(10), Duration::from_secs(5)), + ) + .await; + state.fail_renew.store(true, Ordering::SeqCst); + state.fail_login.store(true, Ordering::SeqCst); + let task = provider.spawn_renewal_task().expect("renewal task"); + + // Refresh cycles at 30s, 35s, ... keep failing; the token stays usable + // until 50s (60s TTL minus the 10s safety window). + tokio::time::sleep(Duration::from_secs(49)).await; + provider + .current() + .expect("token outside the safety window must still be served"); + + tokio::time::sleep(Duration::from_secs(2)).await; + let error = provider + .current() + .expect_err("token inside the safety window must be refused"); + assert!( + matches!(error, KmsError::CredentialsUnavailable { .. }), + "expected CredentialsUnavailable, got {error:?}" + ); + + // Recovery: the next retry cycle succeeds, installs a fresh + // generation, and the provider serves requests again. + state.fail_renew.store(false, Ordering::SeqCst); + state.fail_login.store(false, Ordering::SeqCst); + tokio::time::sleep(Duration::from_secs(6)).await; + let handle = provider.current().expect("provider must recover after a successful refresh"); + assert!(handle.generation >= 1); + + task.shutdown().await; + } + + #[tokio::test(start_paused = true)] + async fn test_shutdown_recycles_renewal_task_promptly() { + let (provider, _state) = scripted_provider( + Duration::from_secs(60), + true, + test_policy(Duration::from_secs(10), Duration::from_secs(5)), + ) + .await; + let task = provider.spawn_renewal_task().expect("renewal task"); + + tokio::time::timeout(Duration::from_secs(1), task.shutdown()) + .await + .expect("cancelled renewal task must exit promptly"); + } + + #[tokio::test(start_paused = true)] + async fn test_dropping_task_handle_cancels_renewal_task() { + let (provider, _state) = scripted_provider( + Duration::from_secs(60), + true, + test_policy(Duration::from_secs(10), Duration::from_secs(5)), + ) + .await; + let task = provider.spawn_renewal_task().expect("renewal task"); + let cancel_probe = task.cancel.clone(); + + drop(task); + + tokio::time::timeout(Duration::from_secs(1), cancel_probe.cancelled()) + .await + .expect("dropping the handle must cancel the renewal task"); + } + + #[tokio::test(start_paused = true)] + async fn test_concurrent_refreshes_coalesce_into_one_login() { + let state = Arc::new(ScriptedState::default()); + let source = ScriptedSource { + state: state.clone(), + ttl: Duration::from_secs(60), + renewable: false, + login_delay: Duration::from_millis(100), + }; + let provider = Arc::new( + VaultCredentialProvider::new( + test_settings(), + Box::new(source), + test_policy(Duration::from_secs(10), Duration::from_secs(5)), + ) + .await + .expect("provider"), + ); + assert_eq!(state.login_calls.load(Ordering::SeqCst), 1, "initial login"); + + let cancel = CancellationToken::new(); + let first = { + let provider = provider.clone(); + let cancel = cancel.clone(); + tokio::spawn(async move { provider.refresh(0, &cancel).await }) + }; + let second = { + let provider = provider.clone(); + let cancel = cancel.clone(); + tokio::spawn(async move { provider.refresh(0, &cancel).await }) + }; + first.await.expect("join").expect("refresh"); + second.await.expect("join").expect("refresh"); + + assert_eq!( + state.login_calls.load(Ordering::SeqCst), + 2, + "concurrent refreshes of the same generation must coalesce into one login" + ); + assert_eq!(provider.snapshot().generation, 1); + } + + #[tokio::test] + async fn test_approle_secret_id_file_missing_fails_fatally() { + let dir = tempfile::tempdir().expect("tempdir"); + let source = AppRoleLogin::new( + &test_settings(), + "approle".to_string(), + "role".to_string(), + String::new(), + Some(dir.path().join("absent-secret-id")), + ) + .expect("source"); + + let failure = source + .resolve_secret_id() + .await + .expect_err("missing secret_id file must fail the attempt"); + assert_eq!(failure.class, ErrorClass::Fatal); + assert!(matches!(failure.error, KmsError::ConfigurationError { .. })); + } + + #[tokio::test] + async fn test_approle_secret_id_file_empty_fails_fatally() { + let file = tempfile::NamedTempFile::new().expect("tempfile"); + std::fs::write(file.path(), " \n\t\n").expect("write whitespace"); + let source = AppRoleLogin::new( + &test_settings(), + "approle".to_string(), + "role".to_string(), + String::new(), + Some(file.path().to_path_buf()), + ) + .expect("source"); + + let failure = source + .resolve_secret_id() + .await + .expect_err("effectively empty secret_id file must fail the attempt"); + assert_eq!(failure.class, ErrorClass::Fatal); + assert!(failure.error.to_string().contains("is empty")); + } + + #[tokio::test] + async fn test_approle_secret_id_file_takes_precedence_and_is_trimmed() { + let file = tempfile::NamedTempFile::new().expect("tempfile"); + std::fs::write(file.path(), format!(" {TEST_SECRET_ID}\n")).expect("write secret"); + let source = AppRoleLogin::new( + &test_settings(), + "approle".to_string(), + "role".to_string(), + "inline-secret-id-must-lose".to_string(), + Some(file.path().to_path_buf()), + ) + .expect("source"); + + let secret_id = source.resolve_secret_id().await.expect("readable file must resolve"); + assert_eq!(secret_id.expose(), TEST_SECRET_ID); } /// Leak regression: the Debug output of every credential-carrying type - /// must stay free of the token literal. + /// must stay free of token and secret_id literals. #[tokio::test] async fn test_credential_types_debug_redacts_token() { - let provider = test_provider().await; - let handle = provider.current(); - let lease = TokenLease::new(TEST_TOKEN.to_string()); - let source = StaticToken::new(TEST_TOKEN.to_string()); + let provider = static_provider().await; + let handle = provider.current().expect("static token"); + let lease = TokenLease::new( + TEST_TOKEN.to_string(), + Some(LeaseInfo { + ttl: Duration::from_secs(60), + renewable: true, + }), + ); + let static_source = StaticToken::new(TEST_TOKEN.to_string()); + let approle_source = AppRoleLogin::new( + &test_settings(), + "approle".to_string(), + "leak-test-role-id".to_string(), + TEST_SECRET_ID.to_string(), + None, + ) + .expect("approle source"); for rendered in [ format!("{provider:?}"), format!("{handle:?}"), format!("{lease:?}"), - format!("{source:?}"), + format!("{static_source:?}"), + format!("{approle_source:?}"), ] { assert!(!rendered.contains(TEST_TOKEN), "debug output must not leak the vault token: {rendered}"); + assert!( + !rendered.contains(TEST_SECRET_ID), + "debug output must not leak the approle secret_id: {rendered}" + ); } assert!(format!("{lease:?}").contains(REDACTED_SECRET)); + let approle_rendered = format!("{approle_source:?}"); + assert!(approle_rendered.contains("leak-test-role-id"), "role_id is not a secret"); + assert!(approle_rendered.contains(REDACTED_SECRET)); + } + + /// Write a token file with owner-only permissions, as a Vault Agent sink + /// would. + fn write_owner_only(path: &std::path::Path, contents: &str) { + std::fs::write(path, contents).expect("write token file"); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600)).expect("chmod token file"); + } + } + + async fn token_file_provider(path: std::path::PathBuf, policy: VaultCredentialPolicy) -> Arc { + Arc::new( + VaultCredentialProvider::new(test_settings(), Box::new(TokenFileSource::new(path, Duration::from_secs(60))), policy) + .await + .expect("token file provider must build without a live Vault"), + ) + } + + #[tokio::test] + async fn test_token_file_source_reads_and_trims_token() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("vault-token"); + write_owner_only(&path, &format!(" {TEST_TOKEN}\n")); + let source = TokenFileSource::new(path, Duration::from_secs(60)); + + let lease = source.acquire().await.expect("readable token file must resolve"); + assert_eq!(lease.expose(), TEST_TOKEN); + let lease_info = lease.lease_info().expect("token file leases carry an observed validity"); + assert_eq!(lease_info.ttl, Duration::from_secs(60)); + assert!(!lease_info.renewable, "agent-managed tokens are refreshed by re-reading, not renew-self"); + } + + #[tokio::test] + async fn test_token_file_missing_fails_fatally() { + let dir = tempfile::tempdir().expect("tempdir"); + let source = TokenFileSource::new(dir.path().join("absent-token"), Duration::from_secs(60)); + + let failure = source.acquire().await.expect_err("missing token file must fail the attempt"); + assert_eq!(failure.class, ErrorClass::Fatal); + assert!(matches!(failure.error, KmsError::ConfigurationError { .. })); + } + + #[tokio::test] + async fn test_token_file_empty_fails_fatally() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("vault-token"); + write_owner_only(&path, " \n\t\n"); + let source = TokenFileSource::new(path, Duration::from_secs(60)); + + let failure = source + .acquire() + .await + .expect_err("effectively empty token file must fail the attempt"); + assert_eq!(failure.class, ErrorClass::Fatal); + assert!(failure.error.to_string().contains("is empty")); + } + + #[cfg(unix)] + #[tokio::test] + async fn test_token_file_rejects_group_or_other_permission_bits() { + use std::os::unix::fs::PermissionsExt; + + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("vault-token"); + write_owner_only(&path, TEST_TOKEN); + let source = TokenFileSource::new(path.clone(), Duration::from_secs(60)); + + for mode in [0o640u32, 0o604, 0o622, 0o660] { + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(mode)).expect("chmod"); + let failure = source + .acquire() + .await + .expect_err("token file readable or writable beyond the owner must be rejected"); + assert_eq!(failure.class, ErrorClass::Fatal, "mode {mode:#o}"); + let rendered = failure.error.to_string(); + assert!(rendered.contains("insecure permissions"), "mode {mode:#o}: {rendered}"); + assert!(!rendered.contains(TEST_TOKEN), "permission errors must not echo the token"); + } + + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600)).expect("chmod"); + source.acquire().await.expect("owner-only token file must resolve"); + } + + #[tokio::test(start_paused = true)] + async fn test_token_file_replacement_installs_new_generation_next_cycle() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("vault-token"); + write_owner_only(&path, "agent-token-v1"); + let provider = token_file_provider(path.clone(), test_policy(Duration::from_secs(10), Duration::from_secs(5))).await; + let task = provider.spawn_renewal_task().expect("token file credentials are lease-bound"); + + tokio::time::sleep(Duration::from_secs(29)).await; + assert_eq!(provider.snapshot().generation, 0, "no re-read before half the observed validity"); + + // Atomic replacement, as a Vault Agent sink writes: temp file + rename. + let staged = dir.path().join("vault-token.tmp"); + write_owner_only(&staged, "agent-token-v2"); + std::fs::rename(&staged, &path).expect("atomic replace"); + + tokio::time::sleep(Duration::from_secs(2)).await; + assert_eq!( + provider.snapshot().generation, + 1, + "the poll after a rotation must install a new generation" + ); + + // A poll without a rotation still installs a fresh generation: each + // successful read re-extends the token's observed validity. + tokio::time::sleep(Duration::from_secs(30)).await; + assert_eq!(provider.snapshot().generation, 2); + + task.shutdown().await; + } + + #[tokio::test(start_paused = true)] + async fn test_token_file_deletion_fails_closed_and_recovers() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("vault-token"); + write_owner_only(&path, "agent-token-v1"); + let provider = token_file_provider(path.clone(), test_policy(Duration::from_secs(10), Duration::from_secs(5))).await; + let task = provider.spawn_renewal_task().expect("renewal task"); + + std::fs::remove_file(&path).expect("remove token file"); + + // Polls at 30s, 35s, ... keep failing; the last read keeps the token + // usable until 50s (60s observed validity minus the 10s safety window). + tokio::time::sleep(Duration::from_secs(49)).await; + provider + .current() + .expect("token outside the safety window must still be served"); + + tokio::time::sleep(Duration::from_secs(2)).await; + let error = provider + .current() + .expect_err("token inside the safety window must be refused"); + assert!( + matches!(error, KmsError::CredentialsUnavailable { .. }), + "expected CredentialsUnavailable, got {error:?}" + ); + + // Restoring the file heals the provider on the next retry cycle. + write_owner_only(&path, "agent-token-v2"); + tokio::time::sleep(Duration::from_secs(6)).await; + let handle = provider.current().expect("provider must recover once the token file is back"); + assert!(handle.generation >= 1); + + task.shutdown().await; + } + + #[tokio::test] + async fn test_token_file_debug_redacts_token() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("vault-token"); + write_owner_only(&path, TEST_TOKEN); + let source = TokenFileSource::new(path.clone(), Duration::from_secs(60)); + source.acquire().await.expect("token file must resolve"); + + let rendered = format!("{source:?}"); + assert!(!rendered.contains(TEST_TOKEN), "debug output must not leak the vault token: {rendered}"); + assert!(rendered.contains("vault-token"), "the file path is not a secret"); } } diff --git a/crates/kms/src/backends/vault_transit.rs b/crates/kms/src/backends/vault_transit.rs index a4153cdfc..a1355af75 100644 --- a/crates/kms/src/backends/vault_transit.rs +++ b/crates/kms/src/backends/vault_transit.rs @@ -14,7 +14,10 @@ //! Vault Transit-based KMS backend. -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, VaultTransitConfig}; use crate::encryption::{DataKeyEnvelope, generate_key_material}; @@ -129,7 +132,7 @@ impl From for TransitKeyMetadata { } pub struct VaultTransitKmsClient { - credentials: VaultCredentialProvider, + credentials: Arc, config: VaultTransitConfig, /// KV v2 mount path for persisting transit key metadata metadata_kv_mount: String, @@ -141,15 +144,18 @@ pub struct VaultTransitKmsClient { impl VaultTransitKmsClient { /// Create a new Vault Transit KMS client /// - /// `attempt_timeout` caps every HTTP request issued through this client. - pub async fn new(config: VaultTransitConfig, attempt_timeout: Duration) -> Result { - 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: VaultTransitConfig, kms_config: &KmsConfig) -> Result { 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?); Ok(Self { credentials, @@ -163,8 +169,9 @@ impl VaultTransitKmsClient { /// 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 { + /// applies to subsequent calls without interrupting in-flight ones. Fails + /// closed when the credentials could not be refreshed in time. + fn vault(&self) -> Result> { self.credentials.current() } @@ -192,7 +199,7 @@ impl VaultTransitKmsClient { } async fn read_transit_key(&self, key_id: &str) -> Result { - key::read(&self.vault().client, &self.config.mount_path, key_id) + key::read(&self.vault()?.client, &self.config.mount_path, key_id) .await .or_else(|e| Self::map_vault_error(key_id, e, "read")) } @@ -200,7 +207,7 @@ impl VaultTransitKmsClient { async fn create_transit_key(&self, key_id: &str) -> Result<()> { let mut builder = CreateKeyRequestBuilder::default(); builder.key_type(KeyType::Aes256Gcm96); - key::create(&self.vault().client, &self.config.mount_path, key_id, Some(&mut builder)) + key::create(&self.vault()?.client, &self.config.mount_path, key_id, Some(&mut builder)) .await .map_err(|e| KmsError::backend_error(format!("Failed to create Vault Transit key {key_id}: {e}"))) } @@ -217,7 +224,7 @@ impl VaultTransitKmsClient { builder.associated_data(aad); } - let response = data::encrypt(&self.vault().client, &self.config.mount_path, key_id, &plaintext_b64, Some(&mut builder)) + let response = data::encrypt(&self.vault()?.client, &self.config.mount_path, key_id, &plaintext_b64, Some(&mut builder)) .await .map_err(|e| KmsError::backend_error(format!("Failed to encrypt data with Vault Transit key {key_id}: {e}")))?; @@ -235,7 +242,7 @@ impl VaultTransitKmsClient { builder.associated_data(aad); } - let response = data::decrypt(&self.vault().client, &self.config.mount_path, key_id, ciphertext, Some(&mut builder)) + let response = data::decrypt(&self.vault()?.client, &self.config.mount_path, key_id, ciphertext, Some(&mut builder)) .await .map_err(|e| KmsError::backend_error(format!("Failed to decrypt data with Vault Transit key {key_id}: {e}")))?; @@ -250,7 +257,7 @@ impl VaultTransitKmsClient { async fn read_metadata_from_kv(&self, key_id: &str) -> Result> { let path = self.metadata_key_path(key_id); - match kv2::read::(&self.vault().client, &self.metadata_kv_mount, &path).await { + match kv2::read::(&self.vault()?.client, &self.metadata_kv_mount, &path).await { Ok(persisted) => Ok(Some(persisted.into())), Err(vaultrs::error::ClientError::ResponseWrapError) | Err(vaultrs::error::ClientError::APIError { code: 404, .. }) => Ok(None), @@ -261,7 +268,7 @@ impl VaultTransitKmsClient { async fn write_metadata_to_kv(&self, key_id: &str, metadata: &TransitKeyMetadata) -> Result<()> { let path = self.metadata_key_path(key_id); let persisted: TransitKeyMetadataPersisted = metadata.clone().into(); - kv2::set(&self.vault().client, &self.metadata_kv_mount, &path, &persisted) + kv2::set(&self.vault()?.client, &self.metadata_kv_mount, &path, &persisted) .await .map(|_| ()) .map_err(|e| KmsError::backend_error(format!("Failed to write transit key metadata to Vault KV: {e}"))) @@ -269,7 +276,7 @@ impl VaultTransitKmsClient { async fn delete_metadata_from_kv(&self, key_id: &str) -> Result<()> { let path = self.metadata_key_path(key_id); - match kv2::delete_metadata(&self.vault().client, &self.metadata_kv_mount, &path).await { + match kv2::delete_metadata(&self.vault()?.client, &self.metadata_kv_mount, &path).await { Ok(_) => Ok(()), Err(vaultrs::error::ClientError::ResponseWrapError) | Err(vaultrs::error::ClientError::APIError { code: 404, .. }) => Ok(()), @@ -483,7 +490,7 @@ impl KmsClient for VaultTransitKmsClient { } async fn list_keys(&self, request: &ListKeysRequest, _context: Option<&OperationContext>) -> Result { - let all_keys = key::list(&self.vault().client, &self.config.mount_path) + let all_keys = key::list(&self.vault()?.client, &self.config.mount_path) .await .map_err(|e| KmsError::backend_error(format!("Failed to list Vault Transit keys: {e}")))? .keys; @@ -553,7 +560,7 @@ impl KmsClient for VaultTransitKmsClient { } async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result { - key::rotate(&self.vault().client, &self.config.mount_path, key_id) + key::rotate(&self.vault()?.client, &self.config.mount_path, key_id) .await .map_err(|e| KmsError::backend_error(format!("Failed to rotate Vault Transit key {key_id}: {e}")))?; @@ -576,7 +583,7 @@ impl KmsClient for VaultTransitKmsClient { } async fn health_check(&self) -> Result<()> { - key::list(&self.vault().client, &self.config.mount_path) + key::list(&self.vault()?.client, &self.config.mount_path) .await .map(|_| ()) .map_err(|e| KmsError::backend_error(format!("Vault Transit health check failed: {e}"))) @@ -612,9 +619,16 @@ impl VaultTransitKmsBackend { } }; - let client = VaultTransitKmsClient::new(vault_config, config.effective_timeout()).await?; + let client = VaultTransitKmsClient::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 { + self.client.credentials.spawn_renewal_task() + } } #[async_trait] @@ -698,7 +712,7 @@ impl KmsBackend for VaultTransitKmsBackend { let mut update_builder = UpdateKeyConfigurationRequestBuilder::default(); update_builder.deletion_allowed(true); key::update( - &self.client.vault().client, + &self.client.vault()?.client, &self.client.config.mount_path, &key_id, Some(&mut update_builder), @@ -708,7 +722,7 @@ impl KmsBackend for VaultTransitKmsBackend { KmsError::backend_error(format!("Failed to allow deletion of Vault Transit key {key_id}: {e}")) })?; } - key::delete(&self.client.vault().client, &self.client.config.mount_path, &key_id) + key::delete(&self.client.vault()?.client, &self.client.config.mount_path, &key_id) .await .map_err(|e| KmsError::backend_error(format!("Failed to delete Vault Transit key {key_id}: {e}")))?; self.client.delete_key_metadata(&key_id).await?; @@ -810,7 +824,7 @@ mod tests { let config = test_vault_transit_config(); // --- First "process": create a key and disable it --- - let client1 = VaultTransitKmsClient::new(config.clone(), Duration::from_secs(30)) + let client1 = VaultTransitKmsClient::new(config.clone(), &KmsConfig::default()) .await .expect("Failed to create VaultTransit client"); @@ -833,7 +847,7 @@ mod tests { assert_eq!(info_after_disable.status, KeyStatus::Disabled, "key must be Disabled after disable_key"); // --- Simulate restart: create a brand new client with empty cache --- - let client2 = VaultTransitKmsClient::new(config, Duration::from_secs(30)) + let client2 = VaultTransitKmsClient::new(config, &KmsConfig::default()) .await .expect("Failed to create second VaultTransit client (restart simulation)"); @@ -863,7 +877,7 @@ mod tests { async fn test_transit_pending_deletion_survives_restart_simulation() { let config = test_vault_transit_config(); - let client1 = VaultTransitKmsClient::new(config.clone(), Duration::from_secs(30)) + let client1 = VaultTransitKmsClient::new(config.clone(), &KmsConfig::default()) .await .expect("Failed to create VaultTransit client"); @@ -887,7 +901,7 @@ mod tests { "key must be PendingDeletion after schedule_key_deletion" ); - let client2 = VaultTransitKmsClient::new(config, Duration::from_secs(30)) + let client2 = VaultTransitKmsClient::new(config, &KmsConfig::default()) .await .expect("Failed to create second VaultTransit client (restart simulation)"); @@ -929,7 +943,7 @@ mod tests { #[tokio::test] #[ignore] // Requires a running Vault instance with transit engine enabled async fn test_transit_old_ciphertext_decrypts_after_rotate() { - let client = VaultTransitKmsClient::new(test_vault_transit_config(), Duration::from_secs(30)) + let client = VaultTransitKmsClient::new(test_vault_transit_config(), &KmsConfig::default()) .await .expect("Failed to create VaultTransit client"); diff --git a/crates/kms/src/config.rs b/crates/kms/src/config.rs index 2e2f5e80e..be4376ca5 100644 --- a/crates/kms/src/config.rs +++ b/crates/kms/src/config.rs @@ -29,8 +29,14 @@ pub const ENV_KMS_VAULT_TRANSIT_METADATA_KV_MOUNT: &str = "RUSTFS_KMS_VAULT_TRAN pub const ENV_KMS_VAULT_TRANSIT_METADATA_PREFIX: &str = "RUSTFS_KMS_VAULT_TRANSIT_METADATA_PREFIX"; pub const ENV_KMS_STATIC_SECRET_KEY: &str = "RUSTFS_KMS_STATIC_SECRET_KEY"; pub const ENV_KMS_STATIC_SECRET_KEY_FILE: &str = "RUSTFS_KMS_STATIC_SECRET_KEY_FILE"; +pub const ENV_KMS_VAULT_APPROLE_ROLE_ID: &str = "RUSTFS_KMS_VAULT_APPROLE_ROLE_ID"; +pub const ENV_KMS_VAULT_APPROLE_SECRET_ID: &str = "RUSTFS_KMS_VAULT_APPROLE_SECRET_ID"; +pub const ENV_KMS_VAULT_APPROLE_SECRET_ID_FILE: &str = "RUSTFS_KMS_VAULT_APPROLE_SECRET_ID_FILE"; +pub const ENV_KMS_VAULT_APPROLE_MOUNT: &str = "RUSTFS_KMS_VAULT_APPROLE_MOUNT"; +pub const ENV_KMS_VAULT_TOKEN_FILE: &str = "RUSTFS_KMS_VAULT_TOKEN_FILE"; pub const DEFAULT_VAULT_TRANSIT_METADATA_KV_MOUNT: &str = "secret"; pub const DEFAULT_VAULT_TRANSIT_METADATA_KEY_PREFIX: &str = "rustfs/kms/transit-metadata"; +pub const DEFAULT_VAULT_APPROLE_MOUNT: &str = "approle"; /// Upper bound applied to `KmsConfig::timeout` when deriving backend behavior. /// @@ -53,6 +59,10 @@ fn default_vault_kv2_mount_path() -> String { "transit".to_string() } +fn default_vault_approle_mount() -> String { + DEFAULT_VAULT_APPROLE_MOUNT.to_string() +} + pub const KMS_CONFIG_REDACTION_RULES: &[RedactionRule] = &[ RedactionRule::new("kms.local.master_key", RedactionLevel::Secret, "local backend key encryption material"), RedactionRule::new("kms.vault.token", RedactionLevel::Secret, "vault authentication token"), @@ -388,18 +398,94 @@ impl Default for VaultTransitConfig { pub enum VaultAuthMethod { /// Token authentication Token { token: String }, - /// AppRole authentication - AppRole { role_id: String, secret_id: String }, + /// AppRole authentication: login with `role_id` + `secret_id` for a + /// lease-bound token that is renewed in the background. + AppRole { + role_id: String, + /// Inline secret_id; used only when `secret_id_file` is unset. + secret_id: String, + /// Path to a file holding the secret_id. Re-read on every login so an + /// externally rotated secret_id is picked up; takes precedence over the + /// inline value. + #[serde(default)] + secret_id_file: Option, + /// AppRole auth engine mount path. + #[serde(default = "default_vault_approle_mount")] + mount: String, + /// Fail-closed margin in seconds: once the current token is within this + /// window of expiry without a successful refresh, requests are refused + /// instead of sent with a token that may lapse mid-flight. Defaults to + /// the per-attempt timeout. + #[serde(default)] + refresh_safety_window_secs: Option, + }, + /// Agent-managed token file (for example a Vault Agent auto-auth sink): + /// the token is read from `path` and re-read periodically so a token + /// rotated by the agent is picked up without a restart. + TokenFile { + path: PathBuf, + /// Seconds between token file re-reads. Each successful read also + /// extends the token's observed validity to twice this value, so a + /// file that stops being readable eventually trips the fail-closed + /// window. Defaults to 30 seconds. + #[serde(default)] + poll_interval_secs: Option, + /// Fail-closed margin in seconds, as on `AppRole`. Defaults to the + /// per-attempt timeout. + #[serde(default)] + refresh_safety_window_secs: Option, + }, +} + +impl VaultAuthMethod { + /// AppRole authentication with the default mount and no secret-id file. + pub fn approle(role_id: String, secret_id: String) -> Self { + Self::AppRole { + role_id, + secret_id, + secret_id_file: None, + mount: default_vault_approle_mount(), + refresh_safety_window_secs: None, + } + } + + /// Agent-managed token file with the default poll interval. + pub fn token_file(path: PathBuf) -> Self { + Self::TokenFile { + path, + poll_interval_secs: None, + refresh_safety_window_secs: None, + } + } } impl fmt::Debug for VaultAuthMethod { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { Self::Token { token } => f.debug_struct("Token").field("token", &redacted_secret(token)).finish(), - Self::AppRole { role_id, secret_id } => f + Self::AppRole { + role_id, + secret_id, + secret_id_file, + mount, + refresh_safety_window_secs, + } => f .debug_struct("AppRole") .field("role_id", role_id) .field("secret_id", &redacted_secret(secret_id)) + .field("secret_id_file", secret_id_file) + .field("mount", mount) + .field("refresh_safety_window_secs", refresh_safety_window_secs) + .finish(), + Self::TokenFile { + path, + poll_interval_secs, + refresh_safety_window_secs, + } => f + .debug_struct("TokenFile") + .field("path", path) + .field("poll_interval_secs", poll_interval_secs) + .field("refresh_safety_window_secs", refresh_safety_window_secs) .finish(), } } @@ -479,7 +565,7 @@ impl KmsConfig { backend: KmsBackend::VaultKv2, backend_config: BackendConfig::VaultKv2(Box::new(VaultConfig { address: address.to_string(), - auth_method: VaultAuthMethod::AppRole { role_id, secret_id }, + auth_method: VaultAuthMethod::approle(role_id, secret_id), ..Default::default() })), ..Default::default() @@ -637,6 +723,8 @@ impl KmsConfig { return Err(KmsError::configuration_error("Vault KV2 address must use http or https scheme")); } + validate_vault_auth_method("Vault KV2", &config.auth_method)?; + if !self.allow_insecure_dev_defaults { validate_vault_development_defaults("Vault KV2", &config.address, &config.auth_method, config.tls.as_ref())?; } @@ -660,6 +748,8 @@ impl KmsConfig { return Err(KmsError::configuration_error("Vault Transit address must use http or https scheme")); } + validate_vault_auth_method("Vault Transit", &config.auth_method)?; + if !self.allow_insecure_dev_defaults { validate_vault_development_defaults( "Vault Transit", @@ -764,7 +854,7 @@ impl KmsConfig { } KmsBackend::VaultKv2 => { let address = get_env_str("RUSTFS_KMS_VAULT_ADDRESS", "http://localhost:8200"); - let token = get_env_str("RUSTFS_KMS_VAULT_TOKEN", "dev-token"); + let auth_method = vault_auth_method_from_env()?; let skip_tls_verify = get_env_bool(ENV_KMS_VAULT_SKIP_TLS_VERIFY, false); let mount_path = match get_env_opt_str("RUSTFS_KMS_VAULT_MOUNT_PATH") { @@ -779,7 +869,7 @@ impl KmsConfig { config.backend_config = BackendConfig::VaultKv2(Box::new(VaultConfig { address, - auth_method: VaultAuthMethod::Token { token }, + auth_method, namespace: get_env_opt_str("RUSTFS_KMS_VAULT_NAMESPACE"), mount_path, kv_mount: get_env_str("RUSTFS_KMS_VAULT_KV_MOUNT", "secret"), @@ -789,12 +879,12 @@ impl KmsConfig { } KmsBackend::VaultTransit => { let address = get_env_str("RUSTFS_KMS_VAULT_ADDRESS", "http://localhost:8200"); - let token = get_env_str("RUSTFS_KMS_VAULT_TOKEN", "dev-token"); + let auth_method = vault_auth_method_from_env()?; let skip_tls_verify = get_env_bool(ENV_KMS_VAULT_SKIP_TLS_VERIFY, false); config.backend_config = BackendConfig::VaultTransit(Box::new(VaultTransitConfig { address, - auth_method: VaultAuthMethod::Token { token }, + auth_method, namespace: get_env_opt_str("RUSTFS_KMS_VAULT_NAMESPACE"), mount_path: get_env_str("RUSTFS_KMS_VAULT_MOUNT_PATH", "transit"), metadata_kv_mount: get_env_str( @@ -873,6 +963,95 @@ fn is_under_temp_dir(path: &Path) -> bool { path.starts_with(std::env::temp_dir()) } +/// Resolve the Vault auth method from environment variables. +/// +/// Setting `RUSTFS_KMS_VAULT_APPROLE_ROLE_ID` selects AppRole authentication; +/// the secret_id then comes from `RUSTFS_KMS_VAULT_APPROLE_SECRET_ID_FILE` +/// (re-read on every login, mirroring the `RUSTFS_KMS_STATIC_SECRET_KEY_FILE` +/// precedent) or inline from `RUSTFS_KMS_VAULT_APPROLE_SECRET_ID`, with the +/// file taking precedence. Without a role id the legacy token flow applies. +fn vault_auth_method_from_env() -> Result { + if let Some(token_file) = get_env_opt_str(ENV_KMS_VAULT_TOKEN_FILE) { + // A token file names one authoritative credential source; combining it + // with another one would leave the effective identity ambiguous, so + // that is a configuration error rather than a precedence rule. + if get_env_opt_str(ENV_KMS_VAULT_APPROLE_ROLE_ID).is_some() { + return Err(KmsError::configuration_error(format!( + "{ENV_KMS_VAULT_TOKEN_FILE} cannot be combined with {ENV_KMS_VAULT_APPROLE_ROLE_ID}; configure exactly one Vault auth method" + ))); + } + if get_env_opt_str("RUSTFS_KMS_VAULT_TOKEN").is_some() { + return Err(KmsError::configuration_error(format!( + "{ENV_KMS_VAULT_TOKEN_FILE} cannot be combined with RUSTFS_KMS_VAULT_TOKEN; configure exactly one Vault auth method" + ))); + } + return Ok(VaultAuthMethod::token_file(PathBuf::from(token_file))); + } + + let Some(role_id) = get_env_opt_str(ENV_KMS_VAULT_APPROLE_ROLE_ID) else { + return Ok(VaultAuthMethod::Token { + token: get_env_str("RUSTFS_KMS_VAULT_TOKEN", "dev-token"), + }); + }; + + let secret_id_file = get_env_opt_str(ENV_KMS_VAULT_APPROLE_SECRET_ID_FILE).map(PathBuf::from); + let secret_id = get_env_opt_str(ENV_KMS_VAULT_APPROLE_SECRET_ID).unwrap_or_default(); + if secret_id.is_empty() && secret_id_file.is_none() { + return Err(KmsError::configuration_error(format!( + "Vault AppRole requires {ENV_KMS_VAULT_APPROLE_SECRET_ID} or {ENV_KMS_VAULT_APPROLE_SECRET_ID_FILE} to be set" + ))); + } + + Ok(VaultAuthMethod::AppRole { + role_id, + secret_id, + secret_id_file, + mount: get_env_str(ENV_KMS_VAULT_APPROLE_MOUNT, DEFAULT_VAULT_APPROLE_MOUNT), + refresh_safety_window_secs: None, + }) +} + +fn validate_vault_auth_method(backend_name: &str, auth_method: &VaultAuthMethod) -> Result<()> { + match auth_method { + VaultAuthMethod::Token { .. } => Ok(()), + VaultAuthMethod::AppRole { + role_id, + secret_id, + secret_id_file, + mount, + .. + } => { + if role_id.is_empty() { + return Err(KmsError::configuration_error(format!("{backend_name} AppRole role_id cannot be empty"))); + } + if secret_id.is_empty() && secret_id_file.is_none() { + return Err(KmsError::configuration_error(format!( + "{backend_name} AppRole requires a secret_id or a secret_id_file" + ))); + } + if mount.is_empty() { + return Err(KmsError::configuration_error(format!("{backend_name} AppRole mount cannot be empty"))); + } + Ok(()) + } + VaultAuthMethod::TokenFile { + path, + poll_interval_secs, + .. + } => { + if path.as_os_str().is_empty() { + return Err(KmsError::configuration_error(format!("{backend_name} token file path cannot be empty"))); + } + if poll_interval_secs == &Some(0) { + return Err(KmsError::configuration_error(format!( + "{backend_name} token file poll interval must be greater than 0" + ))); + } + Ok(()) + } + } +} + fn validate_vault_development_defaults( backend_name: &str, address: &str, @@ -1297,6 +1476,220 @@ mod tests { ); } + #[test] + fn test_from_env_selects_approle_when_role_id_is_set() { + with_vars( + vec![ + ("RUSTFS_KMS_BACKEND", Some("vault")), + ("RUSTFS_KMS_VAULT_ADDRESS", Some("https://vault.example.com")), + (ENV_KMS_VAULT_APPROLE_ROLE_ID, Some("env-role-id")), + (ENV_KMS_VAULT_APPROLE_SECRET_ID, Some("env-approle-secret-id")), + (ENV_KMS_VAULT_APPROLE_MOUNT, Some("approle-alt")), + // A stale token env var must not override the AppRole selection. + ("RUSTFS_KMS_VAULT_TOKEN", Some("vault-token")), + ], + || { + let config = KmsConfig::from_env().expect("kms config should load from env"); + let vault = config.vault_config().expect("vault backend config"); + let VaultAuthMethod::AppRole { + role_id, + secret_id, + secret_id_file, + mount, + refresh_safety_window_secs, + } = &vault.auth_method + else { + panic!("role id in the environment must select AppRole auth, got {:?}", vault.auth_method); + }; + assert_eq!(role_id, "env-role-id"); + assert_eq!(secret_id, "env-approle-secret-id"); + assert_eq!(secret_id_file, &None); + assert_eq!(mount, "approle-alt"); + assert_eq!(refresh_safety_window_secs, &None); + }, + ); + } + + #[test] + fn test_from_env_approle_secret_id_file_is_stored_as_path() { + with_vars( + vec![ + ("RUSTFS_KMS_BACKEND", Some("vault-transit")), + ("RUSTFS_KMS_VAULT_ADDRESS", Some("https://vault.example.com")), + (ENV_KMS_VAULT_APPROLE_ROLE_ID, Some("env-role-id")), + (ENV_KMS_VAULT_APPROLE_SECRET_ID_FILE, Some("/etc/rustfs/approle-secret-id")), + ], + || { + let config = KmsConfig::from_env().expect("kms config should load from env"); + let vault = config.vault_transit_config().expect("vault transit backend config"); + let VaultAuthMethod::AppRole { + secret_id, + secret_id_file, + mount, + .. + } = &vault.auth_method + else { + panic!("role id in the environment must select AppRole auth"); + }; + // The path is stored, not read: the secret_id file is re-read on + // every login so external rotation is picked up. + assert_eq!(secret_id_file.as_deref(), Some(std::path::Path::new("/etc/rustfs/approle-secret-id"))); + assert!(secret_id.is_empty()); + assert_eq!(mount, DEFAULT_VAULT_APPROLE_MOUNT); + }, + ); + } + + #[test] + fn test_from_env_approle_requires_secret_id_or_file() { + with_vars( + vec![ + ("RUSTFS_KMS_BACKEND", Some("vault")), + (ENV_KMS_VAULT_APPROLE_ROLE_ID, Some("env-role-id")), + (ENV_KMS_VAULT_APPROLE_SECRET_ID, None::<&str>), + (ENV_KMS_VAULT_APPROLE_SECRET_ID_FILE, None::<&str>), + ], + || { + let error = KmsConfig::from_env().expect_err("approle without a secret_id source must be rejected"); + assert!(error.to_string().contains(ENV_KMS_VAULT_APPROLE_SECRET_ID)); + }, + ); + } + + #[test] + fn test_from_env_selects_token_file() { + with_vars( + vec![ + ("RUSTFS_KMS_BACKEND", Some("vault")), + ("RUSTFS_KMS_VAULT_ADDRESS", Some("https://vault.example.com")), + (ENV_KMS_VAULT_TOKEN_FILE, Some("/run/vault-agent/token")), + ], + || { + let config = KmsConfig::from_env().expect("kms config should load from env"); + let vault = config.vault_config().expect("vault backend config"); + let VaultAuthMethod::TokenFile { + path, + poll_interval_secs, + refresh_safety_window_secs, + } = &vault.auth_method + else { + panic!("token file in the environment must select TokenFile auth, got {:?}", vault.auth_method); + }; + assert_eq!(path, std::path::Path::new("/run/vault-agent/token")); + assert_eq!(poll_interval_secs, &None); + assert_eq!(refresh_safety_window_secs, &None); + }, + ); + } + + #[test] + fn test_from_env_token_file_is_mutually_exclusive_with_other_auth() { + with_vars( + vec![ + ("RUSTFS_KMS_BACKEND", Some("vault")), + (ENV_KMS_VAULT_TOKEN_FILE, Some("/run/vault-agent/token")), + (ENV_KMS_VAULT_APPROLE_ROLE_ID, Some("env-role-id")), + ], + || { + let error = KmsConfig::from_env().expect_err("token file combined with approle must be rejected"); + assert!(error.to_string().contains(ENV_KMS_VAULT_TOKEN_FILE)); + assert!(error.to_string().contains(ENV_KMS_VAULT_APPROLE_ROLE_ID)); + }, + ); + + with_vars( + vec![ + ("RUSTFS_KMS_BACKEND", Some("vault-transit")), + (ENV_KMS_VAULT_TOKEN_FILE, Some("/run/vault-agent/token")), + ("RUSTFS_KMS_VAULT_TOKEN", Some("vault-token")), + ], + || { + let error = KmsConfig::from_env().expect_err("token file combined with a static token must be rejected"); + assert!(error.to_string().contains(ENV_KMS_VAULT_TOKEN_FILE)); + assert!(error.to_string().contains("RUSTFS_KMS_VAULT_TOKEN")); + }, + ); + } + + #[test] + fn test_validate_rejects_bad_token_file_settings() { + let vault_config = |auth_method: VaultAuthMethod| KmsConfig { + backend: KmsBackend::VaultKv2, + backend_config: BackendConfig::VaultKv2(Box::new(VaultConfig { + address: "https://vault.example.com:8200".to_string(), + auth_method, + ..Default::default() + })), + ..Default::default() + }; + + let error = vault_config(VaultAuthMethod::token_file(PathBuf::new())) + .validate() + .expect_err("empty token file path must be rejected"); + assert!(error.to_string().contains("path")); + + let error = vault_config(VaultAuthMethod::TokenFile { + path: PathBuf::from("/run/vault-agent/token"), + poll_interval_secs: Some(0), + refresh_safety_window_secs: None, + }) + .validate() + .expect_err("zero poll interval must be rejected"); + assert!(error.to_string().contains("poll interval")); + + vault_config(VaultAuthMethod::token_file(PathBuf::from("/run/vault-agent/token"))) + .validate() + .expect("well-formed token file auth must validate"); + } + + #[test] + fn test_approle_config_deserializes_legacy_shape_with_defaults() { + // Persisted configurations from before the AppRole implementation only + // carry role_id and secret_id; the new fields must fill with defaults. + let legacy = serde_json::json!({ + "AppRole": { + "role_id": "legacy-role", + "secret_id": "legacy-secret-id", + } + }); + let auth: VaultAuthMethod = serde_json::from_value(legacy).expect("legacy AppRole config must keep deserializing"); + let VaultAuthMethod::AppRole { + role_id, + secret_id_file, + mount, + refresh_safety_window_secs, + .. + } = auth + else { + panic!("expected AppRole"); + }; + assert_eq!(role_id, "legacy-role"); + assert_eq!(secret_id_file, None); + assert_eq!(mount, DEFAULT_VAULT_APPROLE_MOUNT); + assert_eq!(refresh_safety_window_secs, None); + } + + #[test] + fn test_validate_rejects_incomplete_approle() { + let mut config = KmsConfig::vault_approle( + Url::parse("https://vault.example.com:8200").expect("vault URL"), + String::new(), + "secret-id".to_string(), + ); + let error = config.validate().expect_err("empty role_id must be rejected"); + assert!(error.to_string().contains("role_id")); + + config = KmsConfig::vault_approle( + Url::parse("https://vault.example.com:8200").expect("vault URL"), + "role-id".to_string(), + String::new(), + ); + let error = config + .validate() + .expect_err("approle without secret_id or secret_id_file must be rejected"); + assert!(error.to_string().contains("secret_id")); + } + #[test] fn test_from_env_requires_vault_development_opt_in() { with_vars( diff --git a/crates/kms/src/error.rs b/crates/kms/src/error.rs index 409494b08..0eeff00e9 100644 --- a/crates/kms/src/error.rs +++ b/crates/kms/src/error.rs @@ -132,6 +132,11 @@ pub enum KmsError { /// Operation is not supported by the active KMS backend #[error("Operation '{operation}' is not supported by KMS backend '{backend}'")] UnsupportedCapability { backend: String, operation: String }, + + /// Backend credentials expired or could not be refreshed in time; requests + /// fail closed instead of being sent with credentials that may lapse mid-flight + #[error("KMS credentials unavailable: {message}")] + CredentialsUnavailable { message: String }, } impl KmsError { @@ -281,6 +286,11 @@ impl KmsError { operation: operation.into(), } } + + /// Create a credentials unavailable error + pub fn credentials_unavailable>(message: S) -> Self { + Self::CredentialsUnavailable { message: message.into() } + } } /// Convert from standard library errors diff --git a/crates/kms/src/service_manager.rs b/crates/kms/src/service_manager.rs index be463c42d..7ec4453e8 100644 --- a/crates/kms/src/service_manager.rs +++ b/crates/kms/src/service_manager.rs @@ -14,6 +14,7 @@ //! KMS service manager for dynamic configuration and runtime management +use crate::backends::vault_credentials::CredentialTaskHandle; use crate::backends::{KmsBackend, local::LocalKmsBackend}; use crate::config::{BackendConfig, KmsConfig}; use crate::error::{KmsError, Result}; @@ -110,6 +111,10 @@ struct ServiceVersion { service: Arc, /// The KMS manager instance manager: Arc, + /// Owner of the backend's credential renewal task, if the backend needs + /// one. Stop shuts it down explicitly; reconfigure recycles it through + /// the handle's cancel-on-drop behavior when the old version is discarded. + credential_task: Option>, } #[derive(Clone)] @@ -331,6 +336,13 @@ impl KmsServiceManager { current_service: None, })); + // Shut down the stopped version's credential renewal task before + // reporting stopped, so stop deterministically recycles the background + // task even while in-flight operations still hold the old service Arc. + if let Some(task) = state.current_service.as_ref().and_then(|sv| sv.credential_task.clone()) { + task.shutdown().await; + } + debug!( event = EVENT_KMS_SERVICE_STATE, component = LOG_COMPONENT_KMS, @@ -488,7 +500,10 @@ impl KmsServiceManager { info!("Creating KMS service version {} with backend: {:?}", version, config.backend); - // Create backend + // Create backend. Vault backends may also spawn a background + // credential renewal task whose owner handle lives on the service + // version, so replacing the version recycles the task. + let mut credential_task = None; let backend = match &config.backend_config { BackendConfig::Local(_) => { info!("Creating Local KMS backend for version {}", version); @@ -498,11 +513,13 @@ impl KmsServiceManager { BackendConfig::VaultKv2(_) => { info!("Creating Vault KV2 KMS backend for version {}", version); let backend = crate::backends::vault::VaultKmsBackend::new(config.clone()).await?; + credential_task = backend.spawn_credential_renewal().map(Arc::new); Arc::new(backend) as Arc } BackendConfig::VaultTransit(_) => { info!("Creating Vault Transit KMS backend for version {}", version); let backend = crate::backends::vault_transit::VaultTransitKmsBackend::new(config.clone()).await?; + credential_task = backend.spawn_credential_renewal().map(Arc::new); Arc::new(backend) as Arc } BackendConfig::Static(_) => { @@ -522,6 +539,7 @@ impl KmsServiceManager { version, service: encryption_service, manager: kms_manager, + credential_task, }) } diff --git a/docs/operations/kms-backend-security.md b/docs/operations/kms-backend-security.md index e8063f5b3..badc78bae 100644 --- a/docs/operations/kms-backend-security.md +++ b/docs/operations/kms-backend-security.md @@ -2,6 +2,8 @@ RustFS ships several KMS backends. They differ not only in deployment effort but in **where master key material lives and who can read it**. Pick a backend based on the confidentiality boundary you need, not on the name alone. +For how the Vault backends authenticate (static token, AppRole, Vault Agent token file) and how credential refresh and the fail-closed window behave, see the [Vault KMS authentication runbook](vault-kms-authentication.md). + ## Backend comparison | Backend | Config tag | Master key material location | At-rest protection of key material | Durability | Rotation | Intended use | diff --git a/docs/operations/vault-kms-authentication.md b/docs/operations/vault-kms-authentication.md new file mode 100644 index 000000000..6ee0a0c2c --- /dev/null +++ b/docs/operations/vault-kms-authentication.md @@ -0,0 +1,122 @@ +# Vault KMS authentication runbook + +This runbook covers how the RustFS Vault KMS backends (KV2 and Transit) authenticate to Vault, how to deploy AppRole and Vault Agent token-file authentication, and how the fail-closed credential window behaves in production. For what each backend stores in Vault and the KV2/Transit policy scopes, see [KMS backend security properties](kms-backend-security.md). + +## Choosing an authentication method + +| Method | Config tag | Credential lifetime | Background renewal | Recommended for | +| --- | --- | --- | --- | --- | +| Static token | `Token` | Whatever the operator provisioned; RustFS never renews it | None | Development; short-lived experiments | +| AppRole | `AppRole` | Lease-bound token obtained by login; renewed by RustFS | Renew at half TTL, re-login on failure | Production without a Vault Agent sidecar | +| Agent token file | `TokenFile` | Owned by Vault Agent; RustFS only re-reads the sink file | File re-read once per poll interval | Production with a Vault Agent (or equivalent) managing auth | + +Exactly one method must be configured. Setting `RUSTFS_KMS_VAULT_TOKEN_FILE` together with `RUSTFS_KMS_VAULT_APPROLE_ROLE_ID` or an explicit `RUSTFS_KMS_VAULT_TOKEN` is rejected at startup with a configuration error, because the effective identity would be ambiguous. + +The default `dev-token` fallback for `RUSTFS_KMS_VAULT_TOKEN` is rejected outside explicit development mode (`RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS=true`), as are plain-HTTP Vault addresses and disabled TLS verification. + +## AppRole authentication + +### Vault-side setup + +Create a policy scoped to exactly what the backend needs (see the KV2 and Transit policy examples in [KMS backend security properties](kms-backend-security.md)), then an AppRole that issues tokens carrying it: + +```shell +vault policy write rustfs-kms rustfs-kms-policy.hcl + +vault auth enable approle + +vault write auth/approle/role/rustfs-kms \ + token_policies="rustfs-kms" \ + token_ttl=1h \ + token_max_ttl=24h \ + secret_id_ttl=90d \ + secret_id_num_uses=0 +``` + +Keep `token_ttl` comfortably above the RustFS per-attempt timeout (default 30s): the fail-closed window defaults to one attempt timeout, so a token TTL close to it leaves almost no usable lifetime. + +### RustFS configuration + +```shell +RUSTFS_KMS_BACKEND=vault-transit # or "vault" for the KV2 backend +RUSTFS_KMS_VAULT_ADDRESS=https://vault.example.com:8200 +RUSTFS_KMS_VAULT_APPROLE_ROLE_ID= +RUSTFS_KMS_VAULT_APPROLE_SECRET_ID_FILE=/etc/rustfs/approle-secret-id +# Alternatively, inline (the file takes precedence when both are set): +# RUSTFS_KMS_VAULT_APPROLE_SECRET_ID= +# Optional, defaults to "approle": +# RUSTFS_KMS_VAULT_APPROLE_MOUNT=approle +``` + +RustFS logs in at startup, then renews the token at half its TTL in the background. If renewal fails (network, Vault sealed, token revoked), it falls back to a full re-login; if that also fails, it keeps retrying every few seconds until Vault recovers. + +### SecretID delivery and rotation + +Deliver the SecretID out of band — a secrets-manager-mounted file, an init-container writing `RUSTFS_KMS_VAULT_APPROLE_SECRET_ID_FILE`, or Vault response wrapping unwrapped by your deployment tooling. Treat it like a password: owner-readable file permissions, never in logs or shell history. + +The secret_id file is re-read on every login attempt, so rotating the SecretID is a two-step operation with no restart: generate a new SecretID (`vault write -f auth/approle/role/rustfs-kms/secret-id`), atomically replace the file, then revoke the old SecretID accessor. The already-issued token keeps renewing; the new SecretID is only needed at the next full re-login. + +An empty or missing secret_id file fails the login attempt immediately (no Vault round trip) and is retried on the normal refresh cadence, so repairing the file heals the backend without a restart. + +## Vault Agent token file + +In this mode a Vault Agent (or any equivalent process) owns authentication and token renewal, and RustFS only reads the token sink file. + +### Vault Agent example + +```hcl +auto_auth { + method "approle" { + config = { + role_id_file_path = "/etc/vault-agent/role-id" + secret_id_file_path = "/etc/vault-agent/secret-id" + } + } + + sink "file" { + config = { + path = "/run/vault-agent/token" + mode = 0600 + } + } +} +``` + +### RustFS configuration + +```shell +RUSTFS_KMS_BACKEND=vault-transit +RUSTFS_KMS_VAULT_ADDRESS=https://vault.example.com:8200 +RUSTFS_KMS_VAULT_TOKEN_FILE=/run/vault-agent/token +``` + +The poll interval (`poll_interval_secs` in the `TokenFile` auth configuration, default 30 seconds) controls how often the file is re-read. Each successful read grants the token an observed validity of twice the poll interval and installs a fresh client generation, so an agent-rotated token is picked up within one poll interval of the atomic replace. + +Requirements enforced at every read, each failing the refresh without contacting Vault: + +- The file must exist and be non-empty after trimming whitespace. +- On Unix, the file must not be readable or writable by group or other (mode `0600` or stricter). Wider permissions are a hard error naming the offending mode, mirroring the SFTP host-key rule. RustFS must run as the file's owner. + +If the agent stops refreshing the file that is fine — RustFS re-reads the same token and keeps going as long as the token itself is valid on the Vault side. If the file disappears or turns empty, RustFS keeps serving requests on the last-read token until the fail-closed window trips, and heals automatically once the file is restored. + +## Fail-closed window + +For lease-bound credentials (AppRole tokens, token files), `current()` refuses to hand out a token that is within the safety window of its expiry and has not been refreshed. Requests then fail with `KMS credentials unavailable: ...` instead of being sent with a token that could lapse mid-flight and fail unpredictably on the Vault side. + +- Default window: one per-attempt timeout (`RUSTFS_KMS_TIMEOUT_SECS`, default 30s) — a request issued now can legitimately stay in flight that long, so the token must outlive it. +- Override: `refresh_safety_window_secs` on the `AppRole` or `TokenFile` auth configuration. +- Static tokens never trip the window: they carry no lease and are assumed valid until Vault says otherwise. + +The window is a symptom threshold, not the fault itself: by the time it trips, refresh has been failing for roughly half the token TTL (AppRole) or two poll intervals (token file). + +### Troubleshooting + +| Symptom | Log line to look for | Likely cause and fix | +| --- | --- | --- | +| Requests fail with `KMS credentials unavailable` | `Vault credential refresh failed; retrying until the credentials recover` (warn, repeated) | Vault unreachable/sealed, or the credential source is broken; the provider recovers on its own once refresh succeeds — fix the cause, no restart needed | +| Renewal succeeded but re-login later fails | `Vault token renewal failed; falling back to a fresh login` followed by login errors | SecretID expired/revoked or AppRole role changed; rotate the secret_id file | +| Token file mode error at startup or during polls | `has insecure permissions` in the error | Fix the sink `mode` (0600) and the file owner; the next poll heals the provider | +| Token file missing/empty errors | `Failed to read Vault token file` / `token file ... is empty` | Vault Agent down or sink misconfigured; restart the agent, the next poll heals the provider | +| Startup fails immediately with a configuration error naming two env vars | — | Two auth methods configured at once; keep exactly one of token, AppRole, token file | + +When diagnosing, confirm three clocks/lifetimes in order: the Vault token TTL (`vault token lookup` with the token's accessor), the RustFS refresh cadence (half TTL or the poll interval), and the fail-closed window. The renewal task logs every failed cycle, so a silent gap in warnings combined with `CredentialsUnavailable` errors points at the process clock or a paused runtime rather than Vault.