From f9b02c1675e3547a9e5e08c33c694dc50387d9f8 Mon Sep 17 00:00:00 2001 From: James Sadler Date: Fri, 6 Nov 2026 22:00:31 +1100 Subject: [PATCH] refactor(stack-kms-vendor): share one credential cache across vendors The AWS, Google Cloud, Azure and Vault crates each had a copy of the same credential cache. Google Cloud and Azure also had the same access token and error types. A fix to one copy did not go to the others. stack-kms-vendor now has one generic CachedToken. Each vendor crate declares a marker type that names its credential, its cache key and its error label. The crates keep their public names as type aliases (CachedToken, CachedCredentials, AuthError, CredentialsError) and re-export the shared AccessToken. The cache rules do not change: refresh at min(5 minutes, life / 3) before expiry, one fetch at a time, and a credential that has not expired is kept when a fetch fails. AWS credentials now also use the life / 3 rule. The cache tests are now in stack-kms-vendor, once. --- .../kms-vendor-shared-credential-cache.md | 12 + Cargo.lock | 5 - languages/golang/encrypt/guest/Cargo.lock | 5 - .../typescript/packages/auth/src/keyset.rs | 6 +- packages/stack-kms-aws/Cargo.toml | 1 - packages/stack-kms-aws/src/credentials/mod.rs | 226 +----- packages/stack-kms-aws/src/lib.rs | 5 +- packages/stack-kms-aws/src/time.rs | 7 +- packages/stack-kms-azure/Cargo.toml | 1 - packages/stack-kms-azure/src/auth/mod.rs | 364 +--------- packages/stack-kms-azure/src/lib.rs | 4 +- packages/stack-kms-azure/src/time.rs | 11 +- packages/stack-kms-gcp/Cargo.toml | 1 - packages/stack-kms-gcp/src/auth/mod.rs | 336 +-------- packages/stack-kms-gcp/src/lib.rs | 4 +- packages/stack-kms-gcp/src/time.rs | 8 +- packages/stack-kms-vault/Cargo.toml | 2 - packages/stack-kms-vault/src/auth.rs | 265 +------ packages/stack-kms-vault/src/lib.rs | 4 +- packages/stack-kms-vault/src/time.rs | 9 +- packages/stack-kms-vendor/src/credential.rs | 680 ++++++++++++++++++ packages/stack-kms-vendor/src/lib.rs | 8 + 22 files changed, 873 insertions(+), 1091 deletions(-) create mode 100644 .changeset/kms-vendor-shared-credential-cache.md create mode 100644 packages/stack-kms-vendor/src/credential.rs diff --git a/.changeset/kms-vendor-shared-credential-cache.md b/.changeset/kms-vendor-shared-credential-cache.md new file mode 100644 index 000000000..bf831dc93 --- /dev/null +++ b/.changeset/kms-vendor-shared-credential-cache.md @@ -0,0 +1,12 @@ +--- +"@cipherstash/auth": patch +--- + +The key vendor backends that `stash keyset` uses now share one credential +cache. Before, the AWS, Google Cloud, Azure and Vault backends each had their +own copy. The cache behavior is the same: it gets a new credential 5 minutes +before expiry, or a third of its life before expiry for a short-lived +credential. It does one fetch at a time and keeps a credential that has not +expired if a fetch fails. AWS temporary credentials that live less than 15 +minutes now also refresh a third of their life before expiry. The package's +API is unchanged. diff --git a/Cargo.lock b/Cargo.lock index 364fd8de6..1df6f93c6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3640,7 +3640,6 @@ version = "0.1.0" dependencies = [ "aes-gcm", "base64ct", - "futures", "hmac", "serde", "serde_json", @@ -3663,7 +3662,6 @@ version = "0.1.0" dependencies = [ "aes", "base64ct", - "futures", "rand_core 0.6.4", "rsa", "serde", @@ -3686,7 +3684,6 @@ version = "0.1.0" dependencies = [ "aes-gcm", "base64ct", - "futures", "rand_core 0.6.4", "rsa", "serde", @@ -3709,7 +3706,6 @@ version = "0.1.0" dependencies = [ "aes-gcm", "base64ct", - "futures", "serde", "serde_json", "sha2 0.11.0", @@ -3721,7 +3717,6 @@ dependencies = [ "url", "uuid", "vitaminc 0.9.0", - "web-time", "zeroize", ] diff --git a/languages/golang/encrypt/guest/Cargo.lock b/languages/golang/encrypt/guest/Cargo.lock index c9aca8ef9..746d84cdc 100644 --- a/languages/golang/encrypt/guest/Cargo.lock +++ b/languages/golang/encrypt/guest/Cargo.lock @@ -2239,7 +2239,6 @@ name = "stack-kms-aws" version = "0.1.0" dependencies = [ "base64ct", - "futures", "hmac", "serde", "serde_json", @@ -2258,7 +2257,6 @@ name = "stack-kms-azure" version = "0.1.0" dependencies = [ "base64ct", - "futures", "serde", "serde_json", "stack-kms", @@ -2275,7 +2273,6 @@ name = "stack-kms-gcp" version = "0.1.0" dependencies = [ "base64ct", - "futures", "serde", "serde_json", "stack-kms", @@ -2292,7 +2289,6 @@ name = "stack-kms-vault" version = "0.1.0" dependencies = [ "base64ct", - "futures", "serde", "serde_json", "stack-kms", @@ -2300,7 +2296,6 @@ dependencies = [ "thiserror 1.0.69", "url", "vitaminc 0.9.0", - "web-time", "zeroize", ] diff --git a/languages/typescript/packages/auth/src/keyset.rs b/languages/typescript/packages/auth/src/keyset.rs index 017f942f9..156f41db1 100644 --- a/languages/typescript/packages/auth/src/keyset.rs +++ b/languages/typescript/packages/auth/src/keyset.rs @@ -504,12 +504,14 @@ async fn run_vendor(vendor: KeyVendor, operation: Operation) -> Result Result { - use stack_kms_gcp::{AccessToken, CachedToken, GcpKms, GcpKmsConfig, MetadataServerLogin}; + use stack_kms_gcp::{ + AccessToken, CachedToken, GcpKms, GcpKmsConfig, MetadataServerLogin, ACCESS_TOKEN_VARIABLE, + }; let vendor = KeyVendor::Gcp; let env = stack_kms_gcp::process_env(); let transport = HttpVendorTransport::new().map_err(|error| vendor_config(vendor, error))?; - if let Some(token) = AccessToken::from_env_lookup(&env) { + if let Some(token) = env(ACCESS_TOKEN_VARIABLE).map(AccessToken::new) { let kms = GcpKms::new(GcpKmsConfig::new(), transport, token); return operation.run(&kms).await; } diff --git a/packages/stack-kms-aws/Cargo.toml b/packages/stack-kms-aws/Cargo.toml index 540cd4d0e..310519ef1 100644 --- a/packages/stack-kms-aws/Cargo.toml +++ b/packages/stack-kms-aws/Cargo.toml @@ -25,7 +25,6 @@ stack-kms-vendor = { path = "../stack-kms-vendor", version = "0.1.0" } vitaminc = { workspace = true, features = ["protected"] } base64ct = { version = "1.7", features = ["alloc"] } -futures = "0.3.25" hmac = "0.13" sha2 = "0.11" serde = { workspace = true } diff --git a/packages/stack-kms-aws/src/credentials/mod.rs b/packages/stack-kms-aws/src/credentials/mod.rs index 62e77ec2f..50820e1c9 100644 --- a/packages/stack-kms-aws/src/credentials/mod.rs +++ b/packages/stack-kms-aws/src/credentials/mod.rs @@ -19,15 +19,12 @@ use std::fmt; use std::future::Future; -use std::time::Duration; -use futures::lock::Mutex; -use stack_kms::{BackendErrorKind, MaybeSend}; +use stack_kms::MaybeSend; +use stack_kms_vendor::{CredentialError, CredentialKind, Expiring, Expiry, FetchCredential}; use web_time::SystemTime; use zeroize::Zeroizing; -use crate::time::{system_clock, Clock}; - #[cfg(not(target_arch = "wasm32"))] mod chain; #[cfg(not(target_arch = "wasm32"))] @@ -144,40 +141,27 @@ impl Credentials { } } -/// The credentials could not be got. -#[derive(Debug, thiserror::Error)] -#[error("AWS credentials: {message}")] -pub struct CredentialsError { - kind: BackendErrorKind, - message: String, -} - -impl CredentialsError { - /// A failure of `kind`, described by `message`. Never put a secret in - /// the message. - pub fn new(kind: BackendErrorKind, message: impl Into) -> Self { - Self { - kind, - message: message.into(), - } - } - - /// No credentials are configured, or they were refused. - pub fn access_denied(message: impl Into) -> Self { - Self::new(BackendErrorKind::AccessDenied, message) +impl Expiring for Credentials { + fn expiry(&self) -> Expiry { + self.expires_at.map_or(Expiry::Never, Expiry::At) } +} - /// The credentials source did not answer. - pub fn unavailable(message: impl Into) -> Self { - Self::new(BackendErrorKind::Unavailable, message) - } +/// The credential kind of this crate: [`Credentials`], one set for the +/// client. It names [`CredentialsError`] and [`CachedCredentials`]. +#[derive(Debug)] +pub enum AwsCredentials {} - /// The failure kind the client reports. - pub fn kind(&self) -> BackendErrorKind { - self.kind - } +impl CredentialKind for AwsCredentials { + const LABEL: &'static str = "AWS credentials"; + type Key = (); + type Credential = Credentials; } +/// The credentials could not be got. The message starts with "AWS +/// credentials". +pub type CredentialsError = CredentialError; + /// Where a client gets its credentials. The client asks before it signs /// each request, so a provider that fetches should cache /// ([`CachedCredentials`]). @@ -199,98 +183,40 @@ impl CredentialsProvider for Credentials { } } -/// The default time before expiry at which [`CachedCredentials`] fetches -/// new credentials. -pub const DEFAULT_REFRESH_BEFORE_EXPIRY: Duration = Duration::from_secs(5 * 60); +pub use stack_kms_vendor::DEFAULT_REFRESH_BEFORE_EXPIRY; /// Keeps the credentials `inner` returns, and fetches new ones when they -/// are within the refresh window of their expiry -/// ([`DEFAULT_REFRESH_BEFORE_EXPIRY`]). +/// are near their expiry: [`DEFAULT_REFRESH_BEFORE_EXPIRY`] before, or a +/// third of their life before when they live less than three times that. /// /// One fetch runs at a time: callers that arrive during a fetch wait for /// it and share its result. If a fetch fails while the cached credentials /// have not yet expired, they are still returned. Credentials with no /// expiry are kept for good. -pub struct CachedCredentials

{ - inner: P, - cached: Mutex>, - refresh_before: Duration, - clock: Clock, -} - -impl fmt::Debug for CachedCredentials

{ - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("CachedCredentials") - .field("inner", &self.inner) - .field("refresh_before", &self.refresh_before) - .finish_non_exhaustive() - } -} - -impl CachedCredentials

{ - /// Cache the credentials of `inner`. - pub fn new(inner: P) -> Self { - Self { - inner, - cached: Mutex::new(None), - refresh_before: DEFAULT_REFRESH_BEFORE_EXPIRY, - clock: system_clock(), - } - } - - /// Fetch new credentials `refresh_before` before expiry. - pub fn with_refresh_before(mut self, refresh_before: Duration) -> Self { - self.refresh_before = refresh_before; - self - } - - /// Read the time from `clock`. - pub fn with_clock(mut self, clock: Clock) -> Self { - self.clock = clock; - self - } - - /// The provider whose credentials are cached. - pub fn inner(&self) -> &P { - &self.inner +pub type CachedCredentials

= stack_kms_vendor::CachedToken; + +impl FetchCredential

for AwsCredentials { + fn fetch<'a>( + provider: &'a P, + _key: &'a (), + ) -> impl Future> + MaybeSend + 'a { + provider.credentials() } } impl CredentialsProvider for CachedCredentials

{ - async fn credentials(&self) -> Result { - let mut cached = self.cached.lock().await; - let now = (self.clock)(); - if let Some(credentials) = cached.as_ref() { - let fresh = credentials - .expires_at - .is_none_or(|expiry| now + self.refresh_before < expiry); - if fresh { - return Ok(credentials.clone()); - } - } - match self.inner.credentials().await { - Ok(credentials) => { - *cached = Some(credentials.clone()); - Ok(credentials) - } - Err(error) => match cached.as_ref() { - Some(credentials) if credentials.expires_at.is_some_and(|e| now < e) => { - Ok(credentials.clone()) - } - _ => Err(error), - }, - } + fn credentials( + &self, + ) -> impl Future> + MaybeSend { + self.get(&()) } } #[cfg(test)] mod tests { use std::collections::HashMap; - use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use std::sync::Arc; - use web_time::UNIX_EPOCH; - use super::*; fn env(pairs: &[(&str, &str)]) -> EnvLookup { @@ -334,85 +260,11 @@ mod tests { assert!(!text.contains("the-token")); } - /// A provider that hands out credentials expiring `lifetime` after the - /// test clock, or fails when told to. - struct Counting { - calls: AtomicUsize, - fail: std::sync::atomic::AtomicBool, - now: Arc, - lifetime: Option, - } - - impl CredentialsProvider for Counting { - async fn credentials(&self) -> Result { - let call = self.calls.fetch_add(1, Ordering::SeqCst); - if self.fail.load(Ordering::SeqCst) { - return Err(CredentialsError::unavailable("down")); - } - let credentials = Credentials::new(format!("AKID{call}"), "secret", None); - Ok(match self.lifetime { - Some(lifetime) => credentials.with_expiry( - UNIX_EPOCH + Duration::from_secs(self.now.load(Ordering::SeqCst)) + lifetime, - ), - None => credentials, - }) - } - } - - fn cached(lifetime: Option) -> (CachedCredentials, Arc) { - let now = Arc::new(AtomicU64::new(1_000_000)); - let clock_now = Arc::clone(&now); - let provider = Counting { - calls: AtomicUsize::new(0), - fail: false.into(), - now: Arc::clone(&now), - lifetime, - }; - let clock: Clock = - Arc::new(move || UNIX_EPOCH + Duration::from_secs(clock_now.load(Ordering::SeqCst))); - (CachedCredentials::new(provider).with_clock(clock), now) - } - - #[tokio::test] - async fn credentials_are_fetched_again_before_they_expire() { - let (cache, now) = cached(Some(Duration::from_secs(3600))); - assert_eq!(cache.credentials().await.unwrap().access_key_id(), "AKID0"); - now.fetch_add(3600 - 301, Ordering::SeqCst); - assert_eq!(cache.credentials().await.unwrap().access_key_id(), "AKID0"); - now.fetch_add(1, Ordering::SeqCst); - assert_eq!(cache.credentials().await.unwrap().access_key_id(), "AKID1"); - assert_eq!(cache.inner().calls.load(Ordering::SeqCst), 2); - } - - #[tokio::test] - async fn credentials_without_expiry_are_kept() { - let (cache, now) = cached(None); - cache.credentials().await.unwrap(); - now.fetch_add(10 * 365 * 86_400, Ordering::SeqCst); - assert_eq!(cache.credentials().await.unwrap().access_key_id(), "AKID0"); - assert_eq!(cache.inner().calls.load(Ordering::SeqCst), 1); - } - - #[tokio::test] - async fn a_failed_refresh_keeps_unexpired_credentials_only() { - let (cache, now) = cached(Some(Duration::from_secs(3600))); - cache.credentials().await.unwrap(); - cache.inner().fail.store(true, Ordering::SeqCst); - now.fetch_add(3500, Ordering::SeqCst); - assert_eq!(cache.credentials().await.unwrap().access_key_id(), "AKID0"); - now.fetch_add(100, Ordering::SeqCst); - let error = cache.credentials().await.unwrap_err(); - assert_eq!(error.kind(), BackendErrorKind::Unavailable); - } - - #[tokio::test] - async fn the_refresh_window_is_configurable() { - let (cache, now) = cached(Some(Duration::from_secs(3600))); - let cache = cache.with_refresh_before(Duration::ZERO); - cache.credentials().await.unwrap(); - now.fetch_add(3599, Ordering::SeqCst); - assert_eq!(cache.credentials().await.unwrap().access_key_id(), "AKID0"); - now.fetch_add(1, Ordering::SeqCst); - assert_eq!(cache.credentials().await.unwrap().access_key_id(), "AKID1"); + #[test] + fn the_expiry_is_the_expiry_time() { + let credentials = Credentials::new("AKID", "secret", None); + assert_eq!(credentials.expiry(), Expiry::Never); + let at = web_time::UNIX_EPOCH + std::time::Duration::from_secs(60); + assert_eq!(credentials.with_expiry(at).expiry(), Expiry::At(at)); } } diff --git a/packages/stack-kms-aws/src/lib.rs b/packages/stack-kms-aws/src/lib.rs index 2c5f0d2ec..19a596fe3 100644 --- a/packages/stack-kms-aws/src/lib.rs +++ b/packages/stack-kms-aws/src/lib.rs @@ -74,7 +74,10 @@ pub use client::{ CONTEXT_KEYSET_ID, CONTEXT_PURPOSE, }; pub use config::{AwsKmsConfig, ConfigError}; -pub use credentials::{CachedCredentials, Credentials, CredentialsError, CredentialsProvider}; +pub use credentials::{ + AwsCredentials, CachedCredentials, Credentials, CredentialsError, CredentialsProvider, + DEFAULT_REFRESH_BEFORE_EXPIRY, +}; pub use time::{system_clock, Clock}; #[cfg(all(feature = "http", not(target_arch = "wasm32")))] diff --git a/packages/stack-kms-aws/src/time.rs b/packages/stack-kms-aws/src/time.rs index 409fef14b..71ba29a5a 100644 --- a/packages/stack-kms-aws/src/time.rs +++ b/packages/stack-kms-aws/src/time.rs @@ -7,12 +7,7 @@ use web_time::{SystemTime, UNIX_EPOCH}; /// The time source of a client: `SystemTime::now` unless a test sets /// another. -pub type Clock = std::sync::Arc SystemTime + Send + Sync>; - -/// The system clock. -pub fn system_clock() -> Clock { - std::sync::Arc::new(SystemTime::now) -} +pub use stack_kms_vendor::{system_clock, Clock}; /// The civil date of a day count since 1970-01-01 (Howard Hinnant's /// `civil_from_days`). diff --git a/packages/stack-kms-azure/Cargo.toml b/packages/stack-kms-azure/Cargo.toml index ae7260eef..956e9ed4c 100644 --- a/packages/stack-kms-azure/Cargo.toml +++ b/packages/stack-kms-azure/Cargo.toml @@ -25,7 +25,6 @@ stack-kms-vendor = { path = "../stack-kms-vendor", version = "0.1.0" } vitaminc = { workspace = true, features = ["protected"] } base64ct = { version = "1.7", features = ["alloc"] } -futures = "0.3.25" serde = { workspace = true } serde_json = { workspace = true } thiserror = { workspace = true } diff --git a/packages/stack-kms-azure/src/auth/mod.rs b/packages/stack-kms-azure/src/auth/mod.rs index eb593a036..8f2c17322 100644 --- a/packages/stack-kms-azure/src/auth/mod.rs +++ b/packages/stack-kms-azure/src/auth/mod.rs @@ -32,21 +32,19 @@ //! managed identity. Also not supported: Azure CLI credentials, Azure Arc //! and Service Fabric managed identity. -use std::collections::HashMap; -use std::fmt; use std::future::Future; -use std::sync::{Mutex as SyncMutex, PoisonError}; use std::time::Duration; -use futures::lock::Mutex; use serde::Deserialize; use stack_kms::{BackendErrorKind, MaybeSend}; -use stack_kms_vendor::{classify_status, VendorResponse}; +use stack_kms_vendor::{ + classify_status, CredentialError, CredentialKind, FetchCredential, VendorResponse, +}; use web_time::SystemTime; -use zeroize::Zeroizing; +#[cfg(not(target_arch = "wasm32"))] use crate::config::EnvLookup; -use crate::time::{system_clock, unix_seconds, Clock}; +use crate::time::unix_seconds; mod client_secret; #[cfg(not(target_arch = "wasm32"))] @@ -68,91 +66,21 @@ pub use workload_identity::WorkloadIdentityLogin; /// standard name for one; this one is ours. pub const ACCESS_TOKEN_VARIABLE: &str = "AZURE_KEY_VAULT_ACCESS_TOKEN"; -/// An Entra ID access token, and how long it lasts from when it was -/// issued. -/// -/// The token is wiped on drop and never printed. -#[derive(Clone)] -pub struct AccessToken { - token: Zeroizing, - expires_in: Option, -} +pub use stack_kms_vendor::{AccessToken, DEFAULT_REFRESH_BEFORE_EXPIRY}; -impl fmt::Debug for AccessToken { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("AccessToken") - .field("token", &"") - .field("expires_in", &self.expires_in) - .finish() - } -} - -impl AccessToken { - /// A token with no known expiry: it is used until Azure refuses it. - pub fn new(token: impl Into) -> Self { - Self { - token: Zeroizing::new(token.into()), - expires_in: None, - } - } - - /// The same token, which expires `expires_in` from now. - pub fn with_expires_in(mut self, expires_in: Duration) -> Self { - self.expires_in = Some(expires_in); - self - } +/// The credential kind of this crate: an Entra ID [`AccessToken`], one for +/// each scope. It names [`AuthError`] and [`CachedToken`]. +#[derive(Debug)] +pub enum AzureCredentials {} - /// The token from [`ACCESS_TOKEN_VARIABLE`], or `None` when it is not - /// set. - pub fn from_env_lookup(env: &EnvLookup) -> Option { - env(ACCESS_TOKEN_VARIABLE).map(Self::new) - } - - /// The token. - pub fn token(&self) -> &str { - &self.token - } - - /// How long the token lasts from when it was issued. `None` for a token - /// with no known expiry. - pub fn expires_in(&self) -> Option { - self.expires_in - } +impl CredentialKind for AzureCredentials { + const LABEL: &'static str = "Azure credentials"; + type Key = String; + type Credential = AccessToken; } -/// A token could not be got. -#[derive(Debug, thiserror::Error)] -#[error("Azure credentials: {message}")] -pub struct AuthError { - kind: BackendErrorKind, - message: String, -} - -impl AuthError { - /// A failure of `kind`, described by `message`. Never put a secret in - /// the message. - pub fn new(kind: BackendErrorKind, message: impl Into) -> Self { - Self { - kind, - message: message.into(), - } - } - - /// No credentials are configured, or they were refused. - pub fn access_denied(message: impl Into) -> Self { - Self::new(BackendErrorKind::AccessDenied, message) - } - - /// The token endpoint or the managed identity endpoint did not answer. - pub fn unavailable(message: impl Into) -> Self { - Self::new(BackendErrorKind::Unavailable, message) - } - - /// The failure kind the client reports. - pub fn kind(&self) -> BackendErrorKind { - self.kind - } -} +/// A token could not be got. The message starts with "Azure credentials". +pub type AuthError = CredentialError; /// Where a client gets its access tokens. The client asks before each /// request, so a provider that logs in should cache ([`CachedToken`]). @@ -258,17 +186,12 @@ pub(crate) fn read_token_response( let parsed: TokenResponse = serde_json::from_slice(&response.body).map_err(|_| { AuthError::unavailable(format!("{source} returned a response that is not JSON")) })?; - let token = AccessToken { - token: Zeroizing::new( - parsed - .access_token - .filter(|token| !token.is_empty()) - .ok_or_else(|| { - AuthError::unavailable(format!("{source} returned no access token")) - })?, - ), - expires_in: None, - }; + let token = AccessToken::new( + parsed + .access_token + .filter(|token| !token.is_empty()) + .ok_or_else(|| AuthError::unavailable(format!("{source} returned no access token")))?, + ); let life = match read_u64(parsed.expires_in.as_ref()) { Some(seconds) => Some(seconds), None => { @@ -281,19 +204,6 @@ pub(crate) fn read_token_response( }) } -/// The default time before a token expires at which [`CachedToken`] gets a -/// new one. For a short-lived token the window is a third of its life. -pub const DEFAULT_REFRESH_BEFORE_EXPIRY: Duration = Duration::from_secs(5 * 60); - -/// A token in the cache, and when to get the next one. -struct Cached { - token: AccessToken, - /// When to get a new token. `None`: never. - refresh_at: Option, - /// When the token expires. `None`: never. - expires_at: Option, -} - /// Keeps the token `inner` returns for each scope, and gets a new one when /// it is near its expiry: [`DEFAULT_REFRESH_BEFORE_EXPIRY`] before, or a /// third of its life before when it lives less than three times that. @@ -302,114 +212,31 @@ struct Cached { /// If a fetch fails while the cached token has not expired, the cached /// token is still returned. A token with no expiry is kept until Azure /// refuses it ([`TokenProvider::invalidate`], which drops every scope's). -pub struct CachedToken

{ - inner: P, - cached: SyncMutex>, - fetch: Mutex<()>, - refresh_before: Duration, - clock: Clock, -} - -impl fmt::Debug for CachedToken

{ - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("CachedToken") - .field("inner", &self.inner) - .field("refresh_before", &self.refresh_before) - .finish_non_exhaustive() - } -} - -impl CachedToken

{ - /// Cache the tokens of `inner`. - pub fn new(inner: P) -> Self { - Self { - inner, - cached: SyncMutex::new(HashMap::new()), - fetch: Mutex::new(()), - refresh_before: DEFAULT_REFRESH_BEFORE_EXPIRY, - clock: system_clock(), - } - } - - /// Get a new token at most `refresh_before` before the token expires. - pub fn with_refresh_before(mut self, refresh_before: Duration) -> Self { - self.refresh_before = refresh_before; - self - } - - /// Read the time from `clock`. - pub fn with_clock(mut self, clock: Clock) -> Self { - self.clock = clock; - self - } +pub type CachedToken

= stack_kms_vendor::CachedToken; - /// The provider whose tokens are cached. - pub fn inner(&self) -> &P { - &self.inner +impl FetchCredential

for AzureCredentials { + fn fetch<'a>( + provider: &'a P, + scope: &'a String, + ) -> impl Future> + MaybeSend + 'a { + provider.token(scope) } - /// The cached token of `scope`, if it does not need a refresh yet. - fn fresh(&self, scope: &str, now: SystemTime) -> Option { - let cached = self.cached.lock().unwrap_or_else(PoisonError::into_inner); - cached - .get(scope) - .filter(|cached| cached.refresh_at.is_none_or(|at| now < at)) - .map(|cached| cached.token.clone()) + fn invalidate(provider: &P) { + provider.invalidate(); } } impl TokenProvider for CachedToken

{ - async fn token(&self, scope: &str) -> Result { - if let Some(token) = self.fresh(scope, (self.clock)()) { - return Ok(token); - } - let _fetch = self.fetch.lock().await; - // Another caller may have got a token while this one waited. - let now = (self.clock)(); - if let Some(token) = self.fresh(scope, now) { - return Ok(token); - } - match self.inner.token(scope).await { - Ok(token) => { - let (refresh_at, expires_at) = match token.expires_in { - Some(life) => { - let window = self.refresh_before.min(life / 3); - (now.checked_add(life - window), now.checked_add(life)) - } - None => (None, None), - }; - let _ = self - .cached - .lock() - .unwrap_or_else(PoisonError::into_inner) - .insert( - scope.to_owned(), - Cached { - token: token.clone(), - refresh_at, - expires_at, - }, - ); - Ok(token) - } - Err(error) => { - let cached = self.cached.lock().unwrap_or_else(PoisonError::into_inner); - match cached.get(scope) { - Some(cached) if cached.expires_at.is_some_and(|end| now < end) => { - Ok(cached.token.clone()) - } - _ => Err(error), - } - } - } + fn token( + &self, + scope: &str, + ) -> impl Future> + MaybeSend { + self.get(scope) } fn invalidate(&self) { - self.cached - .lock() - .unwrap_or_else(PoisonError::into_inner) - .clear(); - self.inner.invalidate(); + stack_kms_vendor::CachedToken::invalidate(self); } } @@ -457,7 +284,7 @@ mod default { /// As [`from_env`](Self::from_env), reading the variables with `env`. pub fn from_env_lookup(transport: T, env: &EnvLookup) -> Result { - if let Some(token) = AccessToken::from_env_lookup(env) { + if let Some(token) = env(ACCESS_TOKEN_VARIABLE).map(AccessToken::new) { return Ok(Self::Token(token)); } let tenant = env("AZURE_TENANT_ID"); @@ -519,19 +346,10 @@ pub use default::DefaultCredentials; #[cfg(test)] mod tests { - use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; - use std::sync::Arc; - use web_time::UNIX_EPOCH; use super::*; - #[test] - fn debug_hides_the_token() { - let text = format!("{:?}", AccessToken::new("the-access-token")); - assert!(!text.contains("the-access-token")); - } - #[test] fn the_resource_is_the_scope_without_default() { assert_eq!( @@ -541,112 +359,6 @@ mod tests { assert_eq!(resource_of("https://x"), "https://x"); } - struct Counting { - calls: AtomicUsize, - fail: AtomicBool, - life: Option, - } - - impl TokenProvider for Counting { - async fn token(&self, scope: &str) -> Result { - let call = self.calls.fetch_add(1, Ordering::SeqCst); - if self.fail.load(Ordering::SeqCst) { - return Err(AuthError::unavailable("down")); - } - let token = AccessToken::new(format!("{scope}-{call}")); - Ok(match self.life { - Some(life) => token.with_expires_in(life), - None => token, - }) - } - } - - fn cached(life: Option) -> (CachedToken, Arc) { - let now = Arc::new(AtomicU64::new(1_000_000)); - let clock_now = Arc::clone(&now); - let clock: Clock = - Arc::new(move || UNIX_EPOCH + Duration::from_secs(clock_now.load(Ordering::SeqCst))); - let provider = Counting { - calls: AtomicUsize::new(0), - fail: false.into(), - life, - }; - (CachedToken::new(provider).with_clock(clock), now) - } - - async fn token_of(cache: &CachedToken, scope: &str) -> String { - cache.token(scope).await.unwrap().token().to_owned() - } - - #[tokio::test] - async fn a_token_is_got_again_before_it_expires() { - let (cache, now) = cached(Some(Duration::from_secs(3600))); - assert_eq!(token_of(&cache, "v").await, "v-0"); - now.fetch_add(3600 - 301, Ordering::SeqCst); - assert_eq!(token_of(&cache, "v").await, "v-0"); - now.fetch_add(1, Ordering::SeqCst); - assert_eq!(token_of(&cache, "v").await, "v-1"); - assert_eq!(cache.inner().calls.load(Ordering::SeqCst), 2); - } - - #[tokio::test] - async fn each_scope_has_its_own_token() { - let (cache, _) = cached(Some(Duration::from_secs(3600))); - assert_eq!(token_of(&cache, "vault").await, "vault-0"); - assert_eq!(token_of(&cache, "hsm").await, "hsm-1"); - assert_eq!(token_of(&cache, "vault").await, "vault-0"); - assert_eq!(token_of(&cache, "hsm").await, "hsm-1"); - cache.invalidate(); - assert_eq!(token_of(&cache, "hsm").await, "hsm-2"); - assert_eq!(token_of(&cache, "vault").await, "vault-3"); - } - - #[tokio::test] - async fn a_short_life_refreshes_a_third_before_its_end() { - let (cache, now) = cached(Some(Duration::from_secs(60))); - assert_eq!(token_of(&cache, "v").await, "v-0"); - now.fetch_add(39, Ordering::SeqCst); - assert_eq!(token_of(&cache, "v").await, "v-0"); - now.fetch_add(1, Ordering::SeqCst); - assert_eq!(token_of(&cache, "v").await, "v-1"); - } - - #[tokio::test] - async fn a_token_with_no_expiry_is_kept_until_invalidated() { - let (cache, now) = cached(None); - assert_eq!(token_of(&cache, "v").await, "v-0"); - now.fetch_add(10 * 365 * 86_400, Ordering::SeqCst); - assert_eq!(token_of(&cache, "v").await, "v-0"); - cache.invalidate(); - assert_eq!(token_of(&cache, "v").await, "v-1"); - } - - #[tokio::test] - async fn a_failed_fetch_keeps_a_token_that_has_not_expired() { - let (cache, now) = cached(Some(Duration::from_secs(3600))); - assert_eq!(token_of(&cache, "v").await, "v-0"); - cache.inner().fail.store(true, Ordering::SeqCst); - now.fetch_add(3500, Ordering::SeqCst); - assert_eq!(token_of(&cache, "v").await, "v-0"); - now.fetch_add(100, Ordering::SeqCst); - let error = cache.token("v").await.unwrap_err(); - assert_eq!(error.kind(), BackendErrorKind::Unavailable); - // Another scope has nothing cached to fall back on. - let error = cache.token("w").await.unwrap_err(); - assert_eq!(error.kind(), BackendErrorKind::Unavailable); - } - - #[tokio::test] - async fn the_refresh_window_is_configurable() { - let (cache, now) = cached(Some(Duration::from_secs(3600))); - let cache = cache.with_refresh_before(Duration::ZERO); - assert_eq!(token_of(&cache, "v").await, "v-0"); - now.fetch_add(3599, Ordering::SeqCst); - assert_eq!(token_of(&cache, "v").await, "v-0"); - now.fetch_add(1, Ordering::SeqCst); - assert_eq!(token_of(&cache, "v").await, "v-1"); - } - #[test] fn token_responses_are_read_and_failures_classified() { let now = UNIX_EPOCH + Duration::from_secs(1_000_000); diff --git a/packages/stack-kms-azure/src/lib.rs b/packages/stack-kms-azure/src/lib.rs index a1fba313e..d8c68c36c 100644 --- a/packages/stack-kms-azure/src/lib.rs +++ b/packages/stack-kms-azure/src/lib.rs @@ -89,8 +89,8 @@ mod config; mod time; pub use auth::{ - AccessToken, AuthError, CachedToken, ClientSecretLogin, TokenProvider, ACCESS_TOKEN_VARIABLE, - DEFAULT_AUTHORITY_HOST, DEFAULT_REFRESH_BEFORE_EXPIRY, + AccessToken, AuthError, AzureCredentials, CachedToken, ClientSecretLogin, TokenProvider, + ACCESS_TOKEN_VARIABLE, DEFAULT_AUTHORITY_HOST, DEFAULT_REFRESH_BEFORE_EXPIRY, }; #[cfg(not(target_arch = "wasm32"))] pub use auth::{DefaultCredentials, ManagedIdentityLogin, WorkloadIdentityLogin}; diff --git a/packages/stack-kms-azure/src/time.rs b/packages/stack-kms-azure/src/time.rs index 6a027a9fa..e1e7b26c7 100644 --- a/packages/stack-kms-azure/src/time.rs +++ b/packages/stack-kms-azure/src/time.rs @@ -2,14 +2,9 @@ use web_time::{SystemTime, UNIX_EPOCH}; -/// The time source of a token cache: -/// `SystemTime::now` unless a test sets another. -pub type Clock = std::sync::Arc SystemTime + Send + Sync>; - -/// The system clock. -pub fn system_clock() -> Clock { - std::sync::Arc::new(SystemTime::now) -} +/// The time source of a token cache: `SystemTime::now` unless a test sets +/// another. +pub use stack_kms_vendor::{system_clock, Clock}; /// Seconds since the Unix epoch. A time before the epoch is 0. pub(crate) fn unix_seconds(time: SystemTime) -> u64 { diff --git a/packages/stack-kms-gcp/Cargo.toml b/packages/stack-kms-gcp/Cargo.toml index 0ec8dd7e1..fa192504c 100644 --- a/packages/stack-kms-gcp/Cargo.toml +++ b/packages/stack-kms-gcp/Cargo.toml @@ -31,7 +31,6 @@ stack-kms-vendor = { path = "../stack-kms-vendor", version = "0.1.0" } vitaminc = { workspace = true, features = ["protected"] } base64ct = { version = "1.7", features = ["alloc"] } -futures = "0.3.25" # RS256 for the service account JWT. Pure Rust, so it builds for # wasm32-wasip1 (ring and aws-lc-rs need a C toolchain there). See the # CIP-5308 section of docs/plans/2026-11-05-key-vendor-backends-decisions.md diff --git a/packages/stack-kms-gcp/src/auth/mod.rs b/packages/stack-kms-gcp/src/auth/mod.rs index 502995b9f..cbeb70bcd 100644 --- a/packages/stack-kms-gcp/src/auth/mod.rs +++ b/packages/stack-kms-gcp/src/auth/mod.rs @@ -27,22 +27,21 @@ //! impersonation. Use a service account key, the metadata server, or a //! static token. -use std::fmt; use std::future::Future; -use std::sync::{Mutex as SyncMutex, PoisonError}; +#[cfg(any(feature = "service-account", not(target_arch = "wasm32")))] use std::time::Duration; -use futures::lock::Mutex; #[cfg(any(feature = "service-account", not(target_arch = "wasm32")))] use serde::Deserialize; -use stack_kms::{BackendErrorKind, MaybeSend}; +#[cfg(any(feature = "service-account", not(target_arch = "wasm32")))] +use stack_kms::BackendErrorKind; +use stack_kms::MaybeSend; #[cfg(any(feature = "service-account", not(target_arch = "wasm32")))] use stack_kms_vendor::{classify_status, VendorResponse}; -use web_time::SystemTime; -use zeroize::Zeroizing; +use stack_kms_vendor::{CredentialError, CredentialKind, FetchCredential}; +#[cfg(all(feature = "service-account", not(target_arch = "wasm32")))] use crate::config::EnvLookup; -use crate::time::{system_clock, Clock}; #[cfg(not(target_arch = "wasm32"))] mod metadata; @@ -60,91 +59,26 @@ pub use service_account::{ /// The OAuth scope of Cloud KMS. pub const CLOUD_KMS_SCOPE: &str = "https://www.googleapis.com/auth/cloudkms"; -/// An OAuth 2.0 access token, and how long it lasts from when it was -/// issued. -/// -/// The token is wiped on drop and never printed. -#[derive(Clone)] -pub struct AccessToken { - token: Zeroizing, - expires_in: Option, -} - -impl fmt::Debug for AccessToken { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("AccessToken") - .field("token", &"") - .field("expires_in", &self.expires_in) - .finish() - } -} - -impl AccessToken { - /// A token with no known expiry: it is used until Google refuses it. - pub fn new(token: impl Into) -> Self { - Self { - token: Zeroizing::new(token.into()), - expires_in: None, - } - } - - /// The same token, which expires `expires_in` from now. - pub fn with_expires_in(mut self, expires_in: Duration) -> Self { - self.expires_in = Some(expires_in); - self - } +pub use stack_kms_vendor::{AccessToken, DEFAULT_REFRESH_BEFORE_EXPIRY}; - /// The token from `GOOGLE_OAUTH_ACCESS_TOKEN`, or `None` when it is not - /// set. - pub fn from_env_lookup(env: &EnvLookup) -> Option { - env("GOOGLE_OAUTH_ACCESS_TOKEN").map(Self::new) - } +/// The environment variable of a static access token. +pub const ACCESS_TOKEN_VARIABLE: &str = "GOOGLE_OAUTH_ACCESS_TOKEN"; - /// The token. - pub fn token(&self) -> &str { - &self.token - } +/// The credential kind of this crate: an [`AccessToken`], one for the +/// client (Cloud KMS has one scope). It names [`AuthError`] and +/// [`CachedToken`]. +#[derive(Debug)] +pub enum GoogleCredentials {} - /// How long the token lasts from when it was issued. `None` for a token - /// with no known expiry. - pub fn expires_in(&self) -> Option { - self.expires_in - } +impl CredentialKind for GoogleCredentials { + const LABEL: &'static str = "Google credentials"; + type Key = (); + type Credential = AccessToken; } -/// A token could not be got. -#[derive(Debug, thiserror::Error)] -#[error("Google credentials: {message}")] -pub struct AuthError { - kind: BackendErrorKind, - message: String, -} - -impl AuthError { - /// A failure of `kind`, described by `message`. Never put a secret in - /// the message. - pub fn new(kind: BackendErrorKind, message: impl Into) -> Self { - Self { - kind, - message: message.into(), - } - } - - /// No credentials are configured, or they were refused. - pub fn access_denied(message: impl Into) -> Self { - Self::new(BackendErrorKind::AccessDenied, message) - } - - /// The token endpoint or the metadata server did not answer. - pub fn unavailable(message: impl Into) -> Self { - Self::new(BackendErrorKind::Unavailable, message) - } - - /// The failure kind the client reports. - pub fn kind(&self) -> BackendErrorKind { - self.kind - } -} +/// A token could not be got. The message starts with "Google +/// credentials". +pub type AuthError = CredentialError; /// Where a client gets its access token. The client asks before each /// request, so a provider that logs in should cache ([`CachedToken`]). @@ -225,35 +159,18 @@ pub(crate) fn read_token_response( let parsed: TokenResponse = serde_json::from_slice(&response.body).map_err(|_| { AuthError::unavailable(format!("{source} returned a response that is not JSON")) })?; - let token = Zeroizing::new( + let token = AccessToken::new( parsed .access_token .filter(|token| !token.is_empty()) .ok_or_else(|| AuthError::unavailable(format!("{source} returned no access token")))?, ); - let token = AccessToken { - token, - expires_in: None, - }; Ok(match parsed.expires_in { 0 => token, seconds => token.with_expires_in(Duration::from_secs(seconds)), }) } -/// The default time before a token expires at which [`CachedToken`] gets a -/// new one. For a short-lived token the window is a third of its life. -pub const DEFAULT_REFRESH_BEFORE_EXPIRY: Duration = Duration::from_secs(5 * 60); - -/// A token in the cache, and when to get the next one. -struct Cached { - token: AccessToken, - /// When to get a new token. `None`: never. - refresh_at: Option, - /// When the token expires. `None`: never. - expires_at: Option, -} - /// Keeps the token `inner` returns, and gets a new one when it is near its /// expiry: [`DEFAULT_REFRESH_BEFORE_EXPIRY`] before, or a third of its life /// before when it lives less than three times that. @@ -262,104 +179,28 @@ struct Cached { /// and share its result. If a fetch fails while the cached token has not /// expired, the cached token is still returned. A token with no expiry is /// kept until Google refuses it ([`TokenProvider::invalidate`]). -pub struct CachedToken

{ - inner: P, - cached: SyncMutex>, - fetch: Mutex<()>, - refresh_before: Duration, - clock: Clock, -} - -impl fmt::Debug for CachedToken

{ - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("CachedToken") - .field("inner", &self.inner) - .field("refresh_before", &self.refresh_before) - .finish_non_exhaustive() - } -} - -impl CachedToken

{ - /// Cache the token of `inner`. - pub fn new(inner: P) -> Self { - Self { - inner, - cached: SyncMutex::new(None), - fetch: Mutex::new(()), - refresh_before: DEFAULT_REFRESH_BEFORE_EXPIRY, - clock: system_clock(), - } - } - - /// Get a new token at most `refresh_before` before the token expires. - pub fn with_refresh_before(mut self, refresh_before: Duration) -> Self { - self.refresh_before = refresh_before; - self - } +pub type CachedToken

= stack_kms_vendor::CachedToken; - /// Read the time from `clock`. - pub fn with_clock(mut self, clock: Clock) -> Self { - self.clock = clock; - self +impl FetchCredential

for GoogleCredentials { + fn fetch<'a>( + provider: &'a P, + _key: &'a (), + ) -> impl Future> + MaybeSend + 'a { + provider.token() } - /// The provider whose token is cached. - pub fn inner(&self) -> &P { - &self.inner - } - - /// The cached token, if it does not need a refresh yet. - fn fresh(&self, now: SystemTime) -> Option { - let cached = self.cached.lock().unwrap_or_else(PoisonError::into_inner); - cached - .as_ref() - .filter(|cached| cached.refresh_at.is_none_or(|at| now < at)) - .map(|cached| cached.token.clone()) + fn invalidate(provider: &P) { + provider.invalidate(); } } impl TokenProvider for CachedToken

{ - async fn token(&self) -> Result { - if let Some(token) = self.fresh((self.clock)()) { - return Ok(token); - } - let _fetch = self.fetch.lock().await; - // Another caller may have got a token while this one waited. - let now = (self.clock)(); - if let Some(token) = self.fresh(now) { - return Ok(token); - } - match self.inner.token().await { - Ok(token) => { - let (refresh_at, expires_at) = match token.expires_in { - Some(life) => { - let window = self.refresh_before.min(life / 3); - (now.checked_add(life - window), now.checked_add(life)) - } - None => (None, None), - }; - *self.cached.lock().unwrap_or_else(PoisonError::into_inner) = Some(Cached { - token: token.clone(), - refresh_at, - expires_at, - }); - Ok(token) - } - Err(error) => { - let cached = self.cached.lock().unwrap_or_else(PoisonError::into_inner); - match cached.as_ref() { - Some(cached) if cached.expires_at.is_some_and(|end| now < end) => { - Ok(cached.token.clone()) - } - _ => Err(error), - } - } - } + fn token(&self) -> impl Future> + MaybeSend { + self.get(&()) } fn invalidate(&self) { - *self.cached.lock().unwrap_or_else(PoisonError::into_inner) = None; - self.inner.invalidate(); + stack_kms_vendor::CachedToken::invalidate(self); } } @@ -373,7 +214,7 @@ mod default { /// The credentials [`DefaultCredentials::from_env`] picks. #[derive(Debug)] pub enum DefaultCredentials { - /// `GOOGLE_OAUTH_ACCESS_TOKEN`. + /// [`ACCESS_TOKEN_VARIABLE`]. Token(AccessToken), /// The service account key file at `GOOGLE_APPLICATION_CREDENTIALS`. ServiceAccount(Box>>), @@ -384,7 +225,8 @@ mod default { impl DefaultCredentials { /// Pick credentials from the environment, in this order: /// - /// 1. `GOOGLE_OAUTH_ACCESS_TOKEN`: a static access token. + /// 1. [`ACCESS_TOKEN_VARIABLE`] (`GOOGLE_OAUTH_ACCESS_TOKEN`): a + /// static access token. /// 2. `GOOGLE_APPLICATION_CREDENTIALS`: the path of a service /// account key file. A file of another type (an /// `authorized_user` or `external_account` file) is refused. @@ -397,7 +239,7 @@ mod default { /// As [`from_env`](Self::from_env), reading the variables with `env`. pub fn from_env_lookup(transport: T, env: &EnvLookup) -> Result { - if let Some(token) = AccessToken::from_env_lookup(env) { + if let Some(token) = env(ACCESS_TOKEN_VARIABLE).map(AccessToken::new) { return Ok(Self::Token(token)); } if let Some(path) = env("GOOGLE_APPLICATION_CREDENTIALS") { @@ -436,110 +278,8 @@ pub use default::DefaultCredentials; #[cfg(test)] mod tests { - use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; - use std::sync::Arc; - - use web_time::UNIX_EPOCH; - use super::*; - #[test] - fn debug_hides_the_token() { - let text = format!("{:?}", AccessToken::new("the-access-token")); - assert!(!text.contains("the-access-token")); - } - - struct Counting { - calls: AtomicUsize, - fail: AtomicBool, - life: Option, - } - - impl TokenProvider for Counting { - async fn token(&self) -> Result { - let call = self.calls.fetch_add(1, Ordering::SeqCst); - if self.fail.load(Ordering::SeqCst) { - return Err(AuthError::unavailable("down")); - } - let token = AccessToken::new(format!("token-{call}")); - Ok(match self.life { - Some(life) => token.with_expires_in(life), - None => token, - }) - } - } - - fn cached(life: Option) -> (CachedToken, Arc) { - let now = Arc::new(AtomicU64::new(1_000_000)); - let clock_now = Arc::clone(&now); - let clock: Clock = - Arc::new(move || UNIX_EPOCH + Duration::from_secs(clock_now.load(Ordering::SeqCst))); - let provider = Counting { - calls: AtomicUsize::new(0), - fail: false.into(), - life, - }; - (CachedToken::new(provider).with_clock(clock), now) - } - - async fn token_of(cache: &CachedToken) -> String { - cache.token().await.unwrap().token().to_owned() - } - - #[tokio::test] - async fn a_token_is_got_again_before_it_expires() { - let (cache, now) = cached(Some(Duration::from_secs(3600))); - assert_eq!(token_of(&cache).await, "token-0"); - now.fetch_add(3600 - 301, Ordering::SeqCst); - assert_eq!(token_of(&cache).await, "token-0"); - now.fetch_add(1, Ordering::SeqCst); - assert_eq!(token_of(&cache).await, "token-1"); - assert_eq!(cache.inner().calls.load(Ordering::SeqCst), 2); - } - - #[tokio::test] - async fn a_short_life_refreshes_a_third_before_its_end() { - let (cache, now) = cached(Some(Duration::from_secs(60))); - assert_eq!(token_of(&cache).await, "token-0"); - now.fetch_add(39, Ordering::SeqCst); - assert_eq!(token_of(&cache).await, "token-0"); - now.fetch_add(1, Ordering::SeqCst); - assert_eq!(token_of(&cache).await, "token-1"); - } - - #[tokio::test] - async fn a_token_with_no_expiry_is_kept_until_invalidated() { - let (cache, now) = cached(None); - assert_eq!(token_of(&cache).await, "token-0"); - now.fetch_add(10 * 365 * 86_400, Ordering::SeqCst); - assert_eq!(token_of(&cache).await, "token-0"); - cache.invalidate(); - assert_eq!(token_of(&cache).await, "token-1"); - } - - #[tokio::test] - async fn a_failed_fetch_keeps_a_token_that_has_not_expired() { - let (cache, now) = cached(Some(Duration::from_secs(3600))); - assert_eq!(token_of(&cache).await, "token-0"); - cache.inner().fail.store(true, Ordering::SeqCst); - now.fetch_add(3500, Ordering::SeqCst); - assert_eq!(token_of(&cache).await, "token-0"); - now.fetch_add(100, Ordering::SeqCst); - let error = cache.token().await.unwrap_err(); - assert_eq!(error.kind(), BackendErrorKind::Unavailable); - } - - #[tokio::test] - async fn the_refresh_window_is_configurable() { - let (cache, now) = cached(Some(Duration::from_secs(3600))); - let cache = cache.with_refresh_before(Duration::ZERO); - assert_eq!(token_of(&cache).await, "token-0"); - now.fetch_add(3599, Ordering::SeqCst); - assert_eq!(token_of(&cache).await, "token-0"); - now.fetch_add(1, Ordering::SeqCst); - assert_eq!(token_of(&cache).await, "token-1"); - } - #[test] fn token_responses_are_read_and_failures_classified() { let ok = VendorResponse::new( diff --git a/packages/stack-kms-gcp/src/lib.rs b/packages/stack-kms-gcp/src/lib.rs index 8d892d3fd..05675c9f8 100644 --- a/packages/stack-kms-gcp/src/lib.rs +++ b/packages/stack-kms-gcp/src/lib.rs @@ -89,8 +89,8 @@ pub use auth::DefaultCredentials; #[cfg(not(target_arch = "wasm32"))] pub use auth::MetadataServerLogin; pub use auth::{ - AccessToken, AuthError, CachedToken, TokenProvider, CLOUD_KMS_SCOPE, - DEFAULT_REFRESH_BEFORE_EXPIRY, + AccessToken, AuthError, CachedToken, GoogleCredentials, TokenProvider, ACCESS_TOKEN_VARIABLE, + CLOUD_KMS_SCOPE, DEFAULT_REFRESH_BEFORE_EXPIRY, }; #[cfg(feature = "service-account")] pub use auth::{ServiceAccountKey, ServiceAccountLogin, DEFAULT_TOKEN_URI}; diff --git a/packages/stack-kms-gcp/src/time.rs b/packages/stack-kms-gcp/src/time.rs index d7085fea0..4134c614c 100644 --- a/packages/stack-kms-gcp/src/time.rs +++ b/packages/stack-kms-gcp/src/time.rs @@ -1,15 +1,11 @@ //! The time source of a client. +#[cfg(feature = "service-account")] use web_time::SystemTime; /// The time source of a token cache and of the service account JWT: /// `SystemTime::now` unless a test sets another. -pub type Clock = std::sync::Arc SystemTime + Send + Sync>; - -/// The system clock. -pub fn system_clock() -> Clock { - std::sync::Arc::new(SystemTime::now) -} +pub use stack_kms_vendor::{system_clock, Clock}; /// Seconds since the Unix epoch. A time before the epoch is 0. #[cfg(feature = "service-account")] diff --git a/packages/stack-kms-vault/Cargo.toml b/packages/stack-kms-vault/Cargo.toml index 50f27467d..44fa12c4e 100644 --- a/packages/stack-kms-vault/Cargo.toml +++ b/packages/stack-kms-vault/Cargo.toml @@ -25,12 +25,10 @@ stack-kms-vendor = { path = "../stack-kms-vendor", version = "0.1.0" } vitaminc = { workspace = true, features = ["protected"] } base64ct = { version = "1.7", features = ["alloc"] } -futures = "0.3.25" serde = { workspace = true } serde_json = { workspace = true } thiserror = { workspace = true } url = { workspace = true } -web-time = { workspace = true } zeroize = { workspace = true } [dev-dependencies] diff --git a/packages/stack-kms-vault/src/auth.rs b/packages/stack-kms-vault/src/auth.rs index b7fb79bee..db21ca633 100644 --- a/packages/stack-kms-vault/src/auth.rs +++ b/packages/stack-kms-vault/src/auth.rs @@ -25,18 +25,17 @@ use std::fmt; use std::future::Future; -use std::sync::{Mutex as SyncMutex, PoisonError}; use std::time::Duration; -use futures::lock::Mutex; use serde::{Deserialize, Serialize}; use stack_kms::{BackendErrorKind, MaybeSend}; -use stack_kms_vendor::{classify_status, VendorRequest, VendorTransport}; -use web_time::SystemTime; +use stack_kms_vendor::{ + classify_status, CredentialError, CredentialKind, Expiring, Expiry, FetchCredential, + VendorRequest, VendorTransport, +}; use zeroize::Zeroizing; use crate::config::{is_path, process_env, EnvLookup, VaultConfig}; -use crate::time::{system_clock, Clock}; /// A Vault token, and how long its lease lasts from when it was issued. /// @@ -93,40 +92,28 @@ impl VaultToken { } } -/// A token could not be got. -#[derive(Debug, thiserror::Error)] -#[error("Vault auth: {message}")] -pub struct AuthError { - kind: BackendErrorKind, - message: String, -} - -impl AuthError { - /// A failure of `kind`, described by `message`. Never put a secret in - /// the message. - pub fn new(kind: BackendErrorKind, message: impl Into) -> Self { - Self { - kind, - message: message.into(), - } +impl Expiring for VaultToken { + fn expiry(&self) -> Expiry { + self.lease.map_or(Expiry::Never, Expiry::After) } +} - /// No auth method is configured, or the login was refused. - pub fn access_denied(message: impl Into) -> Self { - Self::new(BackendErrorKind::AccessDenied, message) - } +pub use stack_kms_vendor::DEFAULT_REFRESH_BEFORE_EXPIRY; - /// The auth method did not answer. - pub fn unavailable(message: impl Into) -> Self { - Self::new(BackendErrorKind::Unavailable, message) - } +/// The credential kind of this crate: a [`VaultToken`], one for the +/// client. It names [`AuthError`] and [`CachedToken`]. +#[derive(Debug)] +pub enum VaultAuth {} - /// The failure kind the client reports. - pub fn kind(&self) -> BackendErrorKind { - self.kind - } +impl CredentialKind for VaultAuth { + const LABEL: &'static str = "Vault auth"; + type Key = (); + type Credential = VaultToken; } +/// A token could not be got. The message starts with "Vault auth". +pub type AuthError = CredentialError; + /// Where a client gets its Vault token. The client asks before each /// request, so a provider that logs in should cache ([`CachedToken`]). /// @@ -148,19 +135,6 @@ impl TokenProvider for VaultToken { } } -/// The default time before the end of a lease at which [`CachedToken`] -/// logs in again. For a short lease the window is a third of the lease. -pub const DEFAULT_REFRESH_BEFORE_EXPIRY: Duration = Duration::from_secs(5 * 60); - -/// A token in the cache, and when to get the next one. -struct Cached { - token: VaultToken, - /// When to log in again. `None`: never. - refresh_at: Option, - /// When the lease ends. `None`: never. - expires_at: Option, -} - /// Keeps the token `inner` returns, and gets a new one when its lease is /// near its end: [`DEFAULT_REFRESH_BEFORE_EXPIRY`] before, or a third of /// the lease before when the lease is shorter than three times that. @@ -169,104 +143,28 @@ struct Cached { /// it and share its result. If a login fails while the cached token's lease /// has not ended, the cached token is still returned. A token with no /// lease is kept until Vault refuses it ([`TokenProvider::invalidate`]). -pub struct CachedToken

{ - inner: P, - cached: SyncMutex>, - fetch: Mutex<()>, - refresh_before: Duration, - clock: Clock, -} +pub type CachedToken

= stack_kms_vendor::CachedToken; -impl fmt::Debug for CachedToken

{ - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("CachedToken") - .field("inner", &self.inner) - .field("refresh_before", &self.refresh_before) - .finish_non_exhaustive() - } -} - -impl CachedToken

{ - /// Cache the token of `inner`. - pub fn new(inner: P) -> Self { - Self { - inner, - cached: SyncMutex::new(None), - fetch: Mutex::new(()), - refresh_before: DEFAULT_REFRESH_BEFORE_EXPIRY, - clock: system_clock(), - } +impl FetchCredential

for VaultAuth { + fn fetch<'a>( + provider: &'a P, + _key: &'a (), + ) -> impl Future> + MaybeSend + 'a { + provider.token() } - /// Log in again at most `refresh_before` before the lease ends. - pub fn with_refresh_before(mut self, refresh_before: Duration) -> Self { - self.refresh_before = refresh_before; - self - } - - /// Read the time from `clock`. - pub fn with_clock(mut self, clock: Clock) -> Self { - self.clock = clock; - self - } - - /// The provider whose token is cached. - pub fn inner(&self) -> &P { - &self.inner - } - - /// The cached token, if it does not need a login yet. - fn fresh(&self, now: SystemTime) -> Option { - let cached = self.cached.lock().unwrap_or_else(PoisonError::into_inner); - cached - .as_ref() - .filter(|cached| cached.refresh_at.is_none_or(|at| now < at)) - .map(|cached| cached.token.clone()) + fn invalidate(provider: &P) { + provider.invalidate(); } } impl TokenProvider for CachedToken

{ - async fn token(&self) -> Result { - if let Some(token) = self.fresh((self.clock)()) { - return Ok(token); - } - let _fetch = self.fetch.lock().await; - // Another caller may have logged in while this one waited. - let now = (self.clock)(); - if let Some(token) = self.fresh(now) { - return Ok(token); - } - match self.inner.token().await { - Ok(token) => { - let (refresh_at, expires_at) = match token.lease { - Some(lease) => { - let window = self.refresh_before.min(lease / 3); - (now.checked_add(lease - window), now.checked_add(lease)) - } - None => (None, None), - }; - *self.cached.lock().unwrap_or_else(PoisonError::into_inner) = Some(Cached { - token: token.clone(), - refresh_at, - expires_at, - }); - Ok(token) - } - Err(error) => { - let cached = self.cached.lock().unwrap_or_else(PoisonError::into_inner); - match cached.as_ref() { - Some(cached) if cached.expires_at.is_some_and(|end| now < end) => { - Ok(cached.token.clone()) - } - _ => Err(error), - } - } - } + fn token(&self) -> impl Future> + MaybeSend { + self.get(&()) } fn invalidate(&self) { - *self.cached.lock().unwrap_or_else(PoisonError::into_inner) = None; - self.inner.invalidate(); + stack_kms_vendor::CachedToken::invalidate(self); } } @@ -612,11 +510,6 @@ pub use kubernetes::{ #[cfg(test)] mod tests { - use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; - use std::sync::Arc; - - use web_time::UNIX_EPOCH; - use super::*; #[test] @@ -636,94 +529,10 @@ mod tests { assert!(!text.contains("the-secret-id")); } - struct Counting { - calls: AtomicUsize, - fail: AtomicBool, - lease: Option, - } - - impl TokenProvider for Counting { - async fn token(&self) -> Result { - let call = self.calls.fetch_add(1, Ordering::SeqCst); - if self.fail.load(Ordering::SeqCst) { - return Err(AuthError::unavailable("down")); - } - let token = VaultToken::new(format!("token-{call}")); - Ok(match self.lease { - Some(lease) => token.with_lease(lease), - None => token, - }) - } - } - - fn cached(lease: Option) -> (CachedToken, Arc) { - let now = Arc::new(AtomicU64::new(1_000_000)); - let clock_now = Arc::clone(&now); - let clock: Clock = - Arc::new(move || UNIX_EPOCH + Duration::from_secs(clock_now.load(Ordering::SeqCst))); - let provider = Counting { - calls: AtomicUsize::new(0), - fail: false.into(), - lease, - }; - (CachedToken::new(provider).with_clock(clock), now) - } - - async fn token_of(cache: &CachedToken) -> String { - cache.token().await.unwrap().token().to_owned() - } - - #[tokio::test] - async fn a_token_is_got_again_before_its_lease_ends() { - let (cache, now) = cached(Some(Duration::from_secs(3600))); - assert_eq!(token_of(&cache).await, "token-0"); - now.fetch_add(3600 - 301, Ordering::SeqCst); - assert_eq!(token_of(&cache).await, "token-0"); - now.fetch_add(1, Ordering::SeqCst); - assert_eq!(token_of(&cache).await, "token-1"); - assert_eq!(cache.inner().calls.load(Ordering::SeqCst), 2); - } - - #[tokio::test] - async fn a_short_lease_refreshes_a_third_before_its_end() { - let (cache, now) = cached(Some(Duration::from_secs(60))); - assert_eq!(token_of(&cache).await, "token-0"); - now.fetch_add(39, Ordering::SeqCst); - assert_eq!(token_of(&cache).await, "token-0"); - now.fetch_add(1, Ordering::SeqCst); - assert_eq!(token_of(&cache).await, "token-1"); - } - - #[tokio::test] - async fn a_token_with_no_lease_is_kept_until_invalidated() { - let (cache, now) = cached(None); - assert_eq!(token_of(&cache).await, "token-0"); - now.fetch_add(10 * 365 * 86_400, Ordering::SeqCst); - assert_eq!(token_of(&cache).await, "token-0"); - cache.invalidate(); - assert_eq!(token_of(&cache).await, "token-1"); - } - - #[tokio::test] - async fn a_failed_login_keeps_a_token_whose_lease_has_not_ended() { - let (cache, now) = cached(Some(Duration::from_secs(3600))); - assert_eq!(token_of(&cache).await, "token-0"); - cache.inner().fail.store(true, Ordering::SeqCst); - now.fetch_add(3500, Ordering::SeqCst); - assert_eq!(token_of(&cache).await, "token-0"); - now.fetch_add(100, Ordering::SeqCst); - let error = cache.token().await.unwrap_err(); - assert_eq!(error.kind(), BackendErrorKind::Unavailable); - } - - #[tokio::test] - async fn the_refresh_window_is_configurable() { - let (cache, now) = cached(Some(Duration::from_secs(3600))); - let cache = cache.with_refresh_before(Duration::ZERO); - assert_eq!(token_of(&cache).await, "token-0"); - now.fetch_add(3599, Ordering::SeqCst); - assert_eq!(token_of(&cache).await, "token-0"); - now.fetch_add(1, Ordering::SeqCst); - assert_eq!(token_of(&cache).await, "token-1"); + #[test] + fn the_lease_is_the_expiry() { + assert_eq!(VaultToken::new("t").expiry(), Expiry::Never); + let token = VaultToken::new("t").with_lease(Duration::from_secs(60)); + assert_eq!(token.expiry(), Expiry::After(Duration::from_secs(60))); } } diff --git a/packages/stack-kms-vault/src/lib.rs b/packages/stack-kms-vault/src/lib.rs index a92332767..f106417d5 100644 --- a/packages/stack-kms-vault/src/lib.rs +++ b/packages/stack-kms-vault/src/lib.rs @@ -82,8 +82,8 @@ mod config; mod time; pub use auth::{ - AppRoleLogin, AuthError, CachedToken, TokenProvider, VaultToken, DEFAULT_APPROLE_MOUNT, - DEFAULT_REFRESH_BEFORE_EXPIRY, + AppRoleLogin, AuthError, CachedToken, TokenProvider, VaultAuth, VaultToken, + DEFAULT_APPROLE_MOUNT, DEFAULT_REFRESH_BEFORE_EXPIRY, }; #[cfg(not(target_arch = "wasm32"))] pub use auth::{DefaultAuth, KubernetesLogin}; diff --git a/packages/stack-kms-vault/src/time.rs b/packages/stack-kms-vault/src/time.rs index 7487a4a51..ce4ab058e 100644 --- a/packages/stack-kms-vault/src/time.rs +++ b/packages/stack-kms-vault/src/time.rs @@ -1,12 +1,5 @@ //! The time source of a client. -use web_time::SystemTime; - /// The time source of a token cache: `SystemTime::now` unless a test sets /// another. -pub type Clock = std::sync::Arc SystemTime + Send + Sync>; - -/// The system clock. -pub fn system_clock() -> Clock { - std::sync::Arc::new(SystemTime::now) -} +pub use stack_kms_vendor::{system_clock, Clock}; diff --git a/packages/stack-kms-vendor/src/credential.rs b/packages/stack-kms-vendor/src/credential.rs new file mode 100644 index 000000000..c1b10e1c6 --- /dev/null +++ b/packages/stack-kms-vendor/src/credential.rs @@ -0,0 +1,680 @@ +//! What the vendor crates share about credentials: the access token, the +//! credential error, and the cache that keeps a credential until it is +//! near its expiry. +//! +//! Each vendor crate declares one marker type that implements +//! [`CredentialKind`]. The marker names the credential the crate caches, +//! what the cache is keyed by (a scope, or nothing) and the text that +//! starts each [`CredentialError`]. The crate then implements +//! [`FetchCredential`] on the marker for each of its providers, and its own +//! provider trait for [`CachedToken`](CachedToken). The crate +//! names these with type aliases, for example +//! `type CachedToken

= stack_kms_vendor::CachedToken`. + +use std::borrow::Borrow; +use std::collections::HashMap; +use std::fmt; +use std::future::Future; +use std::hash::Hash; +use std::marker::PhantomData; +use std::sync::{Mutex as SyncMutex, PoisonError}; +use std::time::Duration; + +use futures::lock::Mutex; +use stack_kms::{BackendErrorKind, MaybeSend}; +use web_time::SystemTime; +use zeroize::Zeroizing; + +/// The time source of a credential cache: `SystemTime::now` unless a test +/// sets another. +pub type Clock = std::sync::Arc SystemTime + Send + Sync>; + +/// The system clock. +pub fn system_clock() -> Clock { + std::sync::Arc::new(SystemTime::now) +} + +/// When a credential stops working. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Expiry { + /// No known expiry: the credential is used until the vendor refuses it. + Never, + /// It lasts this long from when it was got (an OAuth `expires_in`, a + /// Vault lease). + After(Duration), + /// It stops working at this time (AWS temporary credentials). + At(SystemTime), +} + +/// A credential that [`CachedToken`] can keep. +pub trait Expiring: Clone + Send + Sync + 'static { + /// When the credential stops working. + fn expiry(&self) -> Expiry; +} + +/// An OAuth 2.0 bearer token, and how long it lasts from when it was +/// issued. Google Cloud and Azure use it. +/// +/// The token is wiped on drop and never printed. +#[derive(Clone)] +pub struct AccessToken { + token: Zeroizing, + expires_in: Option, +} + +impl fmt::Debug for AccessToken { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("AccessToken") + .field("token", &"") + .field("expires_in", &self.expires_in) + .finish() + } +} + +impl AccessToken { + /// A token with no known expiry: it is used until the vendor refuses + /// it. + pub fn new(token: impl Into) -> Self { + Self { + token: Zeroizing::new(token.into()), + expires_in: None, + } + } + + /// The same token, which expires `expires_in` from now. + pub fn with_expires_in(mut self, expires_in: Duration) -> Self { + self.expires_in = Some(expires_in); + self + } + + /// The token. + pub fn token(&self) -> &str { + &self.token + } + + /// How long the token lasts from when it was issued. `None` for a token + /// with no known expiry. + pub fn expires_in(&self) -> Option { + self.expires_in + } +} + +impl Expiring for AccessToken { + fn expiry(&self) -> Expiry { + self.expires_in.map_or(Expiry::Never, Expiry::After) + } +} + +/// The credential of a vendor crate, as its cache and its errors see it. +/// A vendor crate implements this on one marker type (an empty enum). +pub trait CredentialKind: Sized + 'static { + /// The text that starts each [`CredentialError`] of this kind, for + /// example `"Google credentials"`. + const LABEL: &'static str; + + /// What the cache keeps one credential for: `()` for one credential, a + /// scope for one credential per scope. + type Key: Eq + Hash + Clone + Send + Sync + 'static; + + /// The credential. + type Credential: Expiring; +} + +/// A credential could not be got. +/// +/// `K` names the vendor: the message starts with [`CredentialKind::LABEL`]. +/// Each vendor crate names this type with an alias (`AuthError`, +/// `CredentialsError`). +pub struct CredentialError { + kind: BackendErrorKind, + message: String, + vendor: PhantomData K>, +} + +impl CredentialError { + /// A failure of `kind`, described by `message`. Never put a secret in + /// the message. + pub fn new(kind: BackendErrorKind, message: impl Into) -> Self { + Self { + kind, + message: message.into(), + vendor: PhantomData, + } + } + + /// No credentials are configured, or they were refused. + pub fn access_denied(message: impl Into) -> Self { + Self::new(BackendErrorKind::AccessDenied, message) + } + + /// The credential source did not answer. + pub fn unavailable(message: impl Into) -> Self { + Self::new(BackendErrorKind::Unavailable, message) + } + + /// The failure kind the client reports. + pub fn kind(&self) -> BackendErrorKind { + self.kind + } +} + +impl fmt::Debug for CredentialError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("CredentialError") + .field("vendor", &K::LABEL) + .field("kind", &self.kind) + .field("message", &self.message) + .finish() + } +} + +impl fmt::Display for CredentialError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}: {}", K::LABEL, self.message) + } +} + +impl std::error::Error for CredentialError {} + +/// How [`CachedToken`] gets a credential of kind `Self` from a provider +/// `P`. A vendor crate implements this on its marker type, for every type +/// that implements its own provider trait. +pub trait FetchCredential

: CredentialKind { + /// Get a new credential for `key` from `provider`. + /// + /// The returned future is bounded by [`MaybeSend`]: `Send` on native + /// targets, unbounded on wasm32. + fn fetch<'a>( + provider: &'a P, + key: &'a Self::Key, + ) -> impl Future>> + MaybeSend + 'a; + + /// The vendor refused a credential of `provider`. The default does + /// nothing. + fn invalidate(provider: &P) { + let _ = provider; + } +} + +/// The default time before a credential expires at which [`CachedToken`] +/// gets a new one. For a short-lived credential the window is a third of +/// its life. +pub const DEFAULT_REFRESH_BEFORE_EXPIRY: Duration = Duration::from_secs(5 * 60); + +/// A credential in the cache, and when to get the next one. +struct Cached { + credential: C, + /// When to get a new credential. `None`: never. + refresh_at: Option, + /// When the credential expires. `None`: never. + expires_at: Option, +} + +/// Keeps the credential `inner` returns for each key, and gets a new one +/// when it is near its expiry: [`DEFAULT_REFRESH_BEFORE_EXPIRY`] before, or +/// a third of its life before when it lives less than three times that. +/// +/// One fetch runs at a time: callers that arrive during a fetch wait for it +/// and share its result. If a fetch fails while the cached credential has +/// not expired, the cached credential is still returned. A credential with +/// no expiry ([`Expiry::Never`]) is kept until +/// [`invalidate`](Self::invalidate). +/// +/// `K` is the vendor crate's marker type ([`CredentialKind`]). +pub struct CachedToken { + inner: P, + cached: SyncMutex>>, + fetch: Mutex<()>, + refresh_before: Duration, + clock: Clock, +} + +impl fmt::Debug for CachedToken { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("CachedToken") + .field("inner", &self.inner) + .field("refresh_before", &self.refresh_before) + .finish_non_exhaustive() + } +} + +impl CachedToken { + /// Cache the credentials of `inner`. + pub fn new(inner: P) -> Self { + Self { + inner, + cached: SyncMutex::new(HashMap::new()), + fetch: Mutex::new(()), + refresh_before: DEFAULT_REFRESH_BEFORE_EXPIRY, + clock: system_clock(), + } + } + + /// Get a new credential at most `refresh_before` before the credential + /// expires. + pub fn with_refresh_before(mut self, refresh_before: Duration) -> Self { + self.refresh_before = refresh_before; + self + } + + /// Read the time from `clock`. + pub fn with_clock(mut self, clock: Clock) -> Self { + self.clock = clock; + self + } + + /// The provider whose credentials are cached. + pub fn inner(&self) -> &P { + &self.inner + } + + /// The cached credential of `key`, if it does not need a refresh yet. + fn fresh(&self, key: &Q, now: SystemTime) -> Option + where + K::Key: Borrow, + Q: Hash + Eq + ?Sized, + { + let cached = self.cached.lock().unwrap_or_else(PoisonError::into_inner); + cached + .get(key) + .filter(|cached| cached.refresh_at.is_none_or(|at| now < at)) + .map(|cached| cached.credential.clone()) + } + + /// When to refresh a credential got at `now`, and when it expires. + fn schedule( + &self, + expiry: Expiry, + now: SystemTime, + ) -> (Option, Option) { + let expires_at = match expiry { + Expiry::Never => return (None, None), + Expiry::After(life) => now.checked_add(life), + Expiry::At(at) => Some(at), + }; + let Some(expires_at) = expires_at else { + // Past the end of time: as good as no expiry. + return (None, None); + }; + let life = expires_at.duration_since(now).unwrap_or(Duration::ZERO); + let window = self.refresh_before.min(life / 3); + (expires_at.checked_sub(window), Some(expires_at)) + } +} + +impl> CachedToken { + /// The credential of `key`: the cached one, or a new one from the + /// provider when the cached one is near its expiry. + pub async fn get(&self, key: &Q) -> Result> + where + K::Key: Borrow, + Q: Hash + Eq + ToOwned + ?Sized, + { + if let Some(credential) = self.fresh(key, (self.clock)()) { + return Ok(credential); + } + let _fetch = self.fetch.lock().await; + // Another caller may have got a credential while this one waited. + let now = (self.clock)(); + if let Some(credential) = self.fresh(key, now) { + return Ok(credential); + } + let key = key.to_owned(); + match K::fetch(&self.inner, &key).await { + Ok(credential) => { + let (refresh_at, expires_at) = self.schedule(credential.expiry(), now); + let _ = self + .cached + .lock() + .unwrap_or_else(PoisonError::into_inner) + .insert( + key, + Cached { + credential: credential.clone(), + refresh_at, + expires_at, + }, + ); + Ok(credential) + } + Err(error) => { + let cached = self.cached.lock().unwrap_or_else(PoisonError::into_inner); + match cached.get::(&key) { + Some(cached) if cached.expires_at.is_some_and(|end| now < end) => { + Ok(cached.credential.clone()) + } + _ => Err(error), + } + } + } + } + + /// The vendor refused a credential: drop every cached credential, so + /// that the next [`get`](Self::get) fetches a new one, and tell the + /// provider ([`FetchCredential::invalidate`]). + pub fn invalidate(&self) { + self.cached + .lock() + .unwrap_or_else(PoisonError::into_inner) + .clear(); + K::invalidate(&self.inner); + } +} + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; + use std::sync::Arc; + + use web_time::UNIX_EPOCH; + + use super::*; + + /// The test vendor: one token per key (a `String`). + enum Test {} + + impl CredentialKind for Test { + const LABEL: &'static str = "Test credentials"; + type Key = String; + type Credential = AccessToken; + } + + /// The expiry the provider gives its tokens. + #[derive(Clone, Copy)] + enum Life { + Never, + After(u64), + /// Expires at this many seconds after the test clock's start. + At(u64), + } + + struct Counting { + calls: AtomicUsize, + invalidations: AtomicUsize, + fail: AtomicBool, + life: Life, + /// Held by a test to keep a fetch waiting. + gate: Mutex<()>, + } + + /// A token whose `expiry` is not the one `AccessToken` reports, to + /// test [`Expiry::At`]. + #[derive(Clone)] + struct Timed(AccessToken, Expiry); + + impl Expiring for Timed { + fn expiry(&self) -> Expiry { + self.1 + } + } + + enum TimedTest {} + + impl CredentialKind for TimedTest { + const LABEL: &'static str = "Timed credentials"; + type Key = (); + type Credential = Timed; + } + + const START: u64 = 1_000_000; + + fn at(seconds: u64) -> SystemTime { + UNIX_EPOCH + Duration::from_secs(seconds) + } + + impl Counting { + async fn next(&self, key: &str) -> Result<(AccessToken, Expiry), String> { + let _gate = self.gate.lock().await; + let call = self.calls.fetch_add(1, Ordering::SeqCst); + if self.fail.load(Ordering::SeqCst) { + return Err("down".to_owned()); + } + let token = AccessToken::new(format!("{key}-{call}")); + Ok(match self.life { + Life::Never => (token, Expiry::Never), + Life::After(seconds) => { + let token = token.with_expires_in(Duration::from_secs(seconds)); + let expiry = token.expiry(); + (token, expiry) + } + Life::At(seconds) => (token, Expiry::At(at(START + seconds))), + }) + } + } + + impl FetchCredential for Test { + async fn fetch( + provider: &Counting, + key: &String, + ) -> Result> { + provider + .next(key) + .await + .map(|(token, _)| token) + .map_err(CredentialError::unavailable) + } + + fn invalidate(provider: &Counting) { + provider.invalidations.fetch_add(1, Ordering::SeqCst); + } + } + + impl FetchCredential for TimedTest { + async fn fetch(provider: &Counting, _key: &()) -> Result> { + provider + .next("t") + .await + .map(|(token, expiry)| Timed(token, expiry)) + .map_err(CredentialError::unavailable) + } + } + + fn provider(life: Life) -> Counting { + Counting { + calls: AtomicUsize::new(0), + invalidations: AtomicUsize::new(0), + fail: false.into(), + life, + gate: Mutex::new(()), + } + } + + fn clock() -> (Clock, Arc) { + let now = Arc::new(AtomicU64::new(START)); + let clock_now = Arc::clone(&now); + let clock: Clock = Arc::new(move || at(clock_now.load(Ordering::SeqCst))); + (clock, now) + } + + fn cached(life: Life) -> (CachedToken, Arc) { + let (clock, now) = clock(); + (CachedToken::new(provider(life)).with_clock(clock), now) + } + + async fn token_of(cache: &CachedToken, key: &str) -> String { + cache.get(key).await.unwrap().token().to_owned() + } + + #[tokio::test] + async fn a_credential_is_got_again_before_it_expires() { + let (cache, now) = cached(Life::After(3600)); + assert_eq!(token_of(&cache, "v").await, "v-0"); + now.fetch_add(3600 - 301, Ordering::SeqCst); + assert_eq!(token_of(&cache, "v").await, "v-0"); + now.fetch_add(1, Ordering::SeqCst); + assert_eq!(token_of(&cache, "v").await, "v-1"); + assert_eq!(cache.inner().calls.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn a_short_life_refreshes_a_third_before_its_end() { + let (cache, now) = cached(Life::After(60)); + assert_eq!(token_of(&cache, "v").await, "v-0"); + now.fetch_add(39, Ordering::SeqCst); + assert_eq!(token_of(&cache, "v").await, "v-0"); + now.fetch_add(1, Ordering::SeqCst); + assert_eq!(token_of(&cache, "v").await, "v-1"); + } + + #[tokio::test] + async fn a_credential_with_no_expiry_is_kept_until_invalidated() { + let (cache, now) = cached(Life::Never); + assert_eq!(token_of(&cache, "v").await, "v-0"); + now.fetch_add(10 * 365 * 86_400, Ordering::SeqCst); + assert_eq!(token_of(&cache, "v").await, "v-0"); + assert_eq!(cache.inner().invalidations.load(Ordering::SeqCst), 0); + cache.invalidate(); + assert_eq!(cache.inner().invalidations.load(Ordering::SeqCst), 1); + assert_eq!(token_of(&cache, "v").await, "v-1"); + } + + #[tokio::test] + async fn a_failed_fetch_keeps_a_credential_that_has_not_expired() { + let (cache, now) = cached(Life::After(3600)); + assert_eq!(token_of(&cache, "v").await, "v-0"); + cache.inner().fail.store(true, Ordering::SeqCst); + now.fetch_add(3500, Ordering::SeqCst); + assert_eq!(token_of(&cache, "v").await, "v-0"); + // At the expiry the credential is not returned. + now.fetch_add(100, Ordering::SeqCst); + let error = cache.get("v").await.unwrap_err(); + assert_eq!(error.kind(), BackendErrorKind::Unavailable); + // Another key has nothing cached to fall back on. + let error = cache.get("w").await.unwrap_err(); + assert_eq!(error.kind(), BackendErrorKind::Unavailable); + } + + #[tokio::test] + async fn a_failed_fetch_returns_a_credential_until_its_last_second() { + let (cache, now) = cached(Life::After(3600)); + assert_eq!(token_of(&cache, "v").await, "v-0"); + cache.inner().fail.store(true, Ordering::SeqCst); + now.fetch_add(3599, Ordering::SeqCst); + assert_eq!(token_of(&cache, "v").await, "v-0"); + } + + #[tokio::test] + async fn a_failed_fetch_with_no_expiry_is_an_error() { + let (cache, _) = cached(Life::Never); + cache.inner().fail.store(true, Ordering::SeqCst); + assert!(cache.get("v").await.is_err()); + } + + #[tokio::test] + async fn the_refresh_window_is_configurable() { + let (cache, now) = cached(Life::After(3600)); + let cache = cache.with_refresh_before(Duration::ZERO); + assert_eq!(token_of(&cache, "v").await, "v-0"); + now.fetch_add(3599, Ordering::SeqCst); + assert_eq!(token_of(&cache, "v").await, "v-0"); + now.fetch_add(1, Ordering::SeqCst); + assert_eq!(token_of(&cache, "v").await, "v-1"); + } + + #[tokio::test] + async fn each_key_has_its_own_credential() { + let (cache, _) = cached(Life::After(3600)); + assert_eq!(token_of(&cache, "vault").await, "vault-0"); + assert_eq!(token_of(&cache, "hsm").await, "hsm-1"); + assert_eq!(token_of(&cache, "vault").await, "vault-0"); + assert_eq!(token_of(&cache, "hsm").await, "hsm-1"); + cache.invalidate(); + assert_eq!(token_of(&cache, "hsm").await, "hsm-2"); + assert_eq!(token_of(&cache, "vault").await, "vault-3"); + } + + #[tokio::test] + async fn callers_during_a_fetch_share_its_result() { + let (cache, _) = cached(Life::After(3600)); + let gate = cache.inner().gate.lock().await; + let open = async { + // Let both callers start before the fetch can end. + tokio::task::yield_now().await; + drop(gate); + }; + let (first, second, ()) = futures::join!(cache.get("v"), cache.get("v"), open); + assert_eq!(first.unwrap().token(), "v-0"); + assert_eq!(second.unwrap().token(), "v-0"); + assert_eq!(cache.inner().calls.load(Ordering::SeqCst), 1); + } + + fn timed(life: Life) -> (CachedToken, Arc) { + let (clock, now) = clock(); + (CachedToken::new(provider(life)).with_clock(clock), now) + } + + async fn timed_token(cache: &CachedToken) -> String { + cache.get(&()).await.unwrap().0.token().to_owned() + } + + #[tokio::test] + async fn a_credential_with_an_expiry_time_is_refreshed_before_it() { + let (cache, now) = timed(Life::At(3600)); + assert_eq!(timed_token(&cache).await, "t-0"); + now.fetch_add(3600 - 301, Ordering::SeqCst); + assert_eq!(timed_token(&cache).await, "t-0"); + now.fetch_add(1, Ordering::SeqCst); + assert_eq!(timed_token(&cache).await, "t-1"); + } + + #[tokio::test] + async fn an_expiry_time_measures_the_life_from_the_fetch() { + // Got at START, expires at START + 60: refreshed 20 seconds early. + let (cache, now) = timed(Life::At(60)); + assert_eq!(timed_token(&cache).await, "t-0"); + now.fetch_add(39, Ordering::SeqCst); + assert_eq!(timed_token(&cache).await, "t-0"); + now.fetch_add(1, Ordering::SeqCst); + assert_eq!(timed_token(&cache).await, "t-1"); + } + + #[tokio::test] + async fn a_credential_that_has_already_expired_is_not_kept() { + let (cache, now) = timed(Life::At(0)); + now.fetch_add(10, Ordering::SeqCst); + assert_eq!(timed_token(&cache).await, "t-0"); + assert_eq!(timed_token(&cache).await, "t-1"); + cache.inner().fail.store(true, Ordering::SeqCst); + assert!(cache.get(&()).await.is_err()); + } + + #[test] + fn an_access_token_reports_its_expiry() { + assert_eq!(AccessToken::new("t").expiry(), Expiry::Never); + let token = AccessToken::new("t").with_expires_in(Duration::from_secs(9)); + assert_eq!(token.token(), "t"); + assert_eq!(token.expires_in(), Some(Duration::from_secs(9))); + assert_eq!(token.expiry(), Expiry::After(Duration::from_secs(9))); + } + + #[tokio::test] + async fn debug_hides_the_secret() { + let text = format!("{:?}", AccessToken::new("the-access-token")); + assert!(text.contains("AccessToken")); + assert!(!text.contains("the-access-token")); + let (cache, _) = cached(Life::Never); + let _ = cache.get("v").await.unwrap(); + let text = format!("{cache:?}"); + assert!(text.contains("CachedToken")); + assert!(!text.contains("v-0")); + } + + impl fmt::Debug for Counting { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("Counting") + } + } + + #[test] + fn an_error_names_its_vendor_and_kind() { + let error = CredentialError::::access_denied("no key"); + assert_eq!(error.kind(), BackendErrorKind::AccessDenied); + assert_eq!(error.to_string(), "Test credentials: no key"); + assert!(format!("{error:?}").contains("Test credentials")); + let error = CredentialError::::unavailable("down"); + assert_eq!(error.kind(), BackendErrorKind::Unavailable); + let error = CredentialError::::new(BackendErrorKind::Throttled, "slow"); + assert_eq!(error.kind(), BackendErrorKind::Throttled); + assert_eq!(error.to_string(), "Test credentials: slow"); + } +} diff --git a/packages/stack-kms-vendor/src/lib.rs b/packages/stack-kms-vendor/src/lib.rs index 90e776208..f50255d60 100644 --- a/packages/stack-kms-vendor/src/lib.rs +++ b/packages/stack-kms-vendor/src/lib.rs @@ -25,6 +25,9 @@ //! * [`run_batch`]: N vendor calls with a concurrency limit, backoff on //! [`Throttled`](stack_kms::BackendErrorKind::Throttled), and //! fail-whole-batch. +//! * [`CachedToken`]: keeps the credential a login returns, and gets a new +//! one before it expires. Every vendor crate caches its credentials with +//! it, and shares [`AccessToken`] and [`CredentialError`] too. //! * [`VendorTransport`]: the HTTP seam a vendor client sends through, of //! the same shape as [`ZeroKMSConnection`](stack_kms::ZeroKMSConnection), //! so the same vendor code runs on native Rust and in the WASI guest. @@ -236,6 +239,7 @@ mod batch; pub mod bound; +mod credential; pub mod derive; mod error; #[cfg(all(feature = "http", not(target_arch = "wasm32")))] @@ -255,6 +259,10 @@ mod fake; pub mod replay; pub use batch::{run_batch, BatchOptions}; +pub use credential::{ + system_clock, AccessToken, CachedToken, Clock, CredentialError, CredentialKind, Expiring, + Expiry, FetchCredential, DEFAULT_REFRESH_BEFORE_EXPIRY, +}; pub use error::RecordError; pub use record::{KeysetRecord, RECORD_FORMAT_VERSION, RECORD_PREFIX}; pub use router::KeyRouter;