From 124b32742c6960b2cbe286ecf8d6ae0761bcdf0c Mon Sep 17 00:00:00 2001 From: overtrue Date: Fri, 31 Jul 2026 00:44:08 +0800 Subject: [PATCH] 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. --- crates/kms/src/backends/vault.rs | 75 +- crates/kms/src/backends/vault_credentials.rs | 900 +++++++++++++++++-- crates/kms/src/backends/vault_transit.rs | 68 +- crates/kms/src/service_manager.rs | 20 +- 4 files changed, 931 insertions(+), 132 deletions(-) diff --git a/crates/kms/src/backends/vault.rs b/crates/kms/src/backends/vault.rs index 01cc4f042..77846edeb 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::{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 @@ -1157,7 +1172,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"); @@ -1210,7 +1225,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"); @@ -1295,7 +1310,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"); @@ -1327,7 +1342,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"); @@ -1364,7 +1379,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"); @@ -1403,7 +1418,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"); @@ -1446,7 +1461,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"), ); @@ -1505,7 +1520,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"); @@ -1543,7 +1558,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..4d67b3568 100644 --- a/crates/kms/src/backends/vault_credentials.rs +++ b/crates/kms/src/backends/vault_credentials.rs @@ -17,21 +17,70 @@ //! [`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 tokio::time::Instant; +use tokio_util::sync::CancellationToken; +use tracing::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); + +/// 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 +88,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`] and [`AppRoleLogin`]; an agent-managed +/// `TokenFile` source is planned as a follow-up. #[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 +176,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 +195,117 @@ impl fmt::Debug for StaticToken { } } +/// 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() + } +} + /// Map the configured auth method onto a token source. -/// -/// 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> { +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(), + )?)), } } @@ -135,7 +320,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 +328,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 +340,47 @@ 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), + .. + } => Duration::from_secs(*secs), + _ => retry.attempt_timeout, + }; + Self { + retry, + safety_window, + retry_interval: DEFAULT_REFRESH_RETRY_INTERVAL, + } + } } /// One authenticated client generation. @@ -164,10 +390,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 +419,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 +635,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 +648,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 +763,294 @@ 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)); } } diff --git a/crates/kms/src/backends/vault_transit.rs b/crates/kms/src/backends/vault_transit.rs index a291f85b1..dc9d9e395 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::{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?; @@ -798,7 +812,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"); @@ -821,7 +835,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)"); @@ -851,7 +865,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"); @@ -875,7 +889,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)"); @@ -917,7 +931,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/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, }) }