From 4e2f1359d13b194f83051d1e0b8383181035a87d Mon Sep 17 00:00:00 2001 From: szedan Date: Tue, 1 Sep 2026 18:28:17 +0300 Subject: [PATCH 1/6] feat(auth): add cache-through token caching for azure_ad and gcp_adc Introduce apis::token_cache::TokenCache, a reusable cache-through credential cache: a request that finds the cache stale acquires a fresh value inline, double-checked under an exclusive lock so concurrent callers racing an empty cache trigger at most one fetch. There is no background refresh loop and no server-side retry/backoff; a failed fetch is simply not cached, so the next request tries again. Retrofit azure_ad onto it, removing its per-instance background refresher thread and private tokio runtime entirely. Implement gcp_adc's metadata-server token fetch (GCE/GKE Workload Identity) on the same cache, unblocking what was previously a permanent 503 stub; service-account key file fetch remains a documented follow-up, now failing closed with a clear "not implemented" reason instead of silently 503ing forever. refresh_ratio is removed from both filters' config (meaningless under cache-through, since there is no scheduled ahead-of-time refresh). gcp_adc gains a metadata_host config field, mirroring azure_ad's authority_host, so a non-default or test metadata endpoint can be configured directly. Signed-off-by: szedan --- apis/Cargo.toml | 2 +- apis/src/lib.rs | 1 + apis/src/token_cache.rs | 257 +++++++ examples/configs/azure-ad.yaml | 7 +- examples/configs/gcp-adc.yaml | 21 +- filters/src/azure/azure_ad.rs | 699 ++++-------------- filters/src/gcp/config.rs | 43 +- filters/src/gcp/filter.rs | 160 ++-- filters/src/gcp/mod.rs | 5 +- filters/src/gcp/tests.rs | 179 +++-- filters/src/gcp/token.rs | 99 ++- .../tests/suite/examples/azure_ad.rs | 12 +- .../tests/suite/examples/gcp_adc.rs | 26 +- 13 files changed, 757 insertions(+), 754 deletions(-) create mode 100644 apis/src/token_cache.rs diff --git a/apis/Cargo.toml b/apis/Cargo.toml index 4261f14114..cd7aea603f 100644 --- a/apis/Cargo.toml +++ b/apis/Cargo.toml @@ -40,7 +40,7 @@ serde_yaml = { workspace = true } sqlx = { workspace = true, optional = true } thiserror = { workspace = true } tiktoken-rs = { workspace = true } -tokio = { workspace = true, features = ["rt", "time", "net"] } +tokio = { workspace = true, features = ["rt", "sync", "time", "net"] } tracing = { workspace = true } url = { workspace = true } utoipa = { workspace = true } diff --git a/apis/src/lib.rs b/apis/src/lib.rs index 02788223bf..8f4bfd2472 100644 --- a/apis/src/lib.rs +++ b/apis/src/lib.rs @@ -18,6 +18,7 @@ pub mod promotion; #[cfg(feature = "store")] pub mod store; pub(crate) mod subrequest; +pub mod token_cache; pub(crate) mod web_search; /// Whether a `Content-Type` header value indicates `text/event-stream`, diff --git a/apis/src/token_cache.rs b/apis/src/token_cache.rs new file mode 100644 index 0000000000..72805bdc61 --- /dev/null +++ b/apis/src/token_cache.rs @@ -0,0 +1,257 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (c) 2026 Praxis Contributors + +//! [`TokenCache`] — a cache-through, on-demand-refreshed cache for a +//! single short-lived credential (an upstream bearer token, an access +//! token, ...). +//! +//! # Semantics +//! +//! There is no background refresh. Every call to +//! [`TokenCache::get_or_refresh`] checks the cache; if the cached value +//! is still valid it is returned immediately (a shared read lock, no +//! network call). If it is missing or past its safety margin, the +//! caller takes an exclusive lock, checks again in case a concurrent +//! caller already refreshed it while this one was waiting for the lock, +//! and only then calls `fetch` — at most once per group of callers that +//! all observe a stale cache at the same time. A failed fetch is +//! propagated to the caller and nothing is cached, so the next call +//! tries again; there is no server-side retry/backoff loop. + +use std::time::{Duration, Instant}; + +use tokio::sync::RwLock; + +// ----------------------------------------------------------------------------- +// Margin +// ----------------------------------------------------------------------------- + +/// Safety margin to subtract from a TTL before treating a cached value +/// as expired, capped at half the TTL so a short-lived credential stays +/// usable for part of its life instead of being cached already-expired. +pub(crate) fn effective_margin(ttl: Duration, margin: Duration) -> Duration { + margin.min(ttl / 2) +} + +// ----------------------------------------------------------------------------- +// TokenCache +// ----------------------------------------------------------------------------- + +/// A cached value and the instant after which it must not be used. +struct Entry { + /// The cached value itself. + value: T, + + /// Instant, already adjusted by the cache's margin, after which the + /// value must not be used. + expires_at: Instant, +} + +/// Return a clone of `entry`'s value if it is still valid, `None` +/// otherwise (missing or past its safety-margin-adjusted expiry). +fn valid_cached_value(entry: Option<&Entry>) -> Option { + entry + .filter(|entry| entry.expires_at > Instant::now()) + .map(|entry| entry.value.clone()) +} + +/// Cache-through cache for one credential. See the module docs. +pub struct TokenCache { + /// Safety window subtracted from a fetched value's TTL (capped at + /// half the TTL) before it is treated as expired. + margin: Duration, + + /// The cached entry, if any. + cache: RwLock>>, +} + +impl TokenCache { + /// Build an empty cache. `margin` is the safety window subtracted + /// from a fetched value's TTL (capped at half the TTL) before it is + /// treated as expired. + pub fn new(margin: Duration) -> Self { + Self { + margin, + cache: RwLock::new(None), + } + } + + /// Return a cached, valid value — fetching a new one first if the + /// cache is empty or the cached value is within its safety margin + /// of expiry. + /// + /// At most one concurrent caller ever calls `fetch`: everyone else + /// who finds the cache stale queues on the exclusive lock behind + /// the first caller and, once it releases the lock, re-checks and + /// finds the value that caller just published. + /// + /// # Errors + /// + /// Returns whatever `fetch` returns on failure. Nothing is cached + /// in that case, so the next call tries again. + pub async fn get_or_refresh(&self, fetch: F) -> Result + where + F: FnOnce() -> Fut + Send, + Fut: Future> + Send, + { + // Fast path: shared read lock, no fetch. + let fresh = valid_cached_value(self.cache.read().await.as_ref()); + if let Some(value) = fresh { + return Ok(value); + } + + // Slow path: exclusive lock, re-check (another caller may have + // already refreshed it while this one waited for the lock). + let mut guard = self.cache.write().await; + let fresh = valid_cached_value(guard.as_ref()); + if let Some(value) = fresh { + drop(guard); + return Ok(value); + } + + let (value, ttl) = fetch().await?; + let expires_at = Instant::now() + ttl.saturating_sub(effective_margin(ttl, self.margin)); + *guard = Some(Entry { + value: value.clone(), + expires_at, + }); + drop(guard); + Ok(value) + } +} + +// ----------------------------------------------------------------------------- +// Tests +// ----------------------------------------------------------------------------- + +#[cfg(test)] +#[expect(clippy::allow_attributes, reason = "blanket test suppressions")] +#[allow(clippy::unwrap_used, clippy::expect_used, reason = "tests")] +mod tests { + use std::sync::atomic::{AtomicU32, Ordering}; + + use super::{TokenCache, effective_margin}; + + #[test] + fn effective_margin_caps_at_half_ttl() { + use std::time::Duration; + assert_eq!(effective_margin(Duration::from_secs(3600), Duration::from_secs(30)), Duration::from_secs(30)); + assert_eq!(effective_margin(Duration::from_secs(40), Duration::from_secs(30)), Duration::from_secs(20)); + assert_eq!(effective_margin(Duration::from_secs(10), Duration::from_secs(30)), Duration::from_secs(5)); + } + + #[tokio::test] + async fn first_call_fetches_and_caches() { + let cache: TokenCache = TokenCache::new(std::time::Duration::from_millis(5)); + let calls = AtomicU32::new(0); + + let value = cache + .get_or_refresh(|| async { + calls.fetch_add(1, Ordering::SeqCst); + Ok::<_, &str>((42_u32, std::time::Duration::from_secs(60))) + }) + .await + .expect("fetch must succeed"); + + assert_eq!(value, 42); + assert_eq!(calls.load(Ordering::SeqCst), 1, "fetch must be called exactly once"); + } + + #[tokio::test] + async fn second_call_within_ttl_does_not_refetch() { + let cache: TokenCache = TokenCache::new(std::time::Duration::from_millis(5)); + let calls = AtomicU32::new(0); + let fetch = || async { + calls.fetch_add(1, Ordering::SeqCst); + Ok::<_, &str>((7_u32, std::time::Duration::from_secs(60))) + }; + + let first = cache.get_or_refresh(fetch).await.expect("first fetch must succeed"); + let second = cache.get_or_refresh(fetch).await.expect("second call must succeed"); + + assert_eq!(first, 7); + assert_eq!(second, 7); + assert_eq!(calls.load(Ordering::SeqCst), 1, "a still-valid cache must not trigger a second fetch"); + } + + #[tokio::test] + async fn refetch_after_expiry() { + let cache: TokenCache = TokenCache::new(std::time::Duration::from_millis(1)); + let calls = AtomicU32::new(0); + let fetch = || async { + let n = calls.fetch_add(1, Ordering::SeqCst) + 1; + Ok::<_, &str>((n, std::time::Duration::from_millis(5))) + }; + + let first = cache.get_or_refresh(fetch).await.expect("first fetch must succeed"); + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + let second = cache.get_or_refresh(fetch).await.expect("second fetch must succeed"); + + assert_eq!(first, 1); + assert_eq!(second, 2, "an expired cache must trigger a fresh fetch"); + assert_eq!(calls.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn concurrent_calls_with_empty_cache_fetch_exactly_once() { + let cache: TokenCache = TokenCache::new(std::time::Duration::from_millis(5)); + let calls = std::sync::Arc::new(AtomicU32::new(0)); + let cache = std::sync::Arc::new(cache); + + let mut handles = Vec::new(); + for _ in 0..10 { + let cache = std::sync::Arc::clone(&cache); + let calls = std::sync::Arc::clone(&calls); + handles.push(tokio::spawn(async move { + cache + .get_or_refresh(|| async move { + calls.fetch_add(1, Ordering::SeqCst); + // Widen the race window so all 10 tasks are + // guaranteed to observe the empty cache before + // the first one publishes a value. + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + Ok::<_, &str>((99_u32, std::time::Duration::from_secs(60))) + }) + .await + .expect("fetch must succeed") + })); + } + + for handle in handles { + assert_eq!(handle.await.expect("task must not panic"), 99); + } + assert_eq!( + calls.load(Ordering::SeqCst), + 1, + "10 concurrent callers against an empty cache must fetch exactly once" + ); + } + + #[tokio::test] + async fn fetch_error_is_propagated_and_not_cached() { + let cache: TokenCache = TokenCache::new(std::time::Duration::from_millis(5)); + let calls = AtomicU32::new(0); + + let err = cache + .get_or_refresh(|| async { + calls.fetch_add(1, Ordering::SeqCst); + Err::<(u32, std::time::Duration), _>("boom") + }) + .await + .expect_err("failed fetch must propagate the error"); + + assert_eq!(err, "boom"); + assert_eq!(calls.load(Ordering::SeqCst), 1); + + // Nothing was cached, so the next call must try again. + let value = cache + .get_or_refresh(|| async { + calls.fetch_add(1, Ordering::SeqCst); + Ok::<_, &str>((5_u32, std::time::Duration::from_secs(60))) + }) + .await + .expect("retry must succeed"); + assert_eq!(value, 5); + assert_eq!(calls.load(Ordering::SeqCst), 2, "a failed fetch must not be cached"); + } +} diff --git a/examples/configs/azure-ad.yaml b/examples/configs/azure-ad.yaml index b194a2d1c2..75fc8066db 100644 --- a/examples/configs/azure-ad.yaml +++ b/examples/configs/azure-ad.yaml @@ -2,9 +2,9 @@ # # Acquires an Entra ID bearer token via the client-credentials grant # and injects "Authorization: Bearer " on every proxied -# request to Azure OpenAI. The token is cached in memory and refreshed -# in the background before it expires; the downstream client never -# sees Entra ID. +# request to Azure OpenAI. The token is cached in memory (cache-through: +# a request that finds the cache stale fetches a fresh token inline +# before proceeding); the downstream client never sees Entra ID. # # This filter only injects the Authorization header — it does NOT # manage the upstream Host. Point the backend cluster at your Azure @@ -40,7 +40,6 @@ filter_chains: scope: https://cognitiveservices.azure.com/.default client_secret_env_var: AZURE_CLIENT_SECRET # authority_host: login.microsoftonline.com # optional, for sovereign clouds - # refresh_ratio: 0.75 # optional, refresh at 75% of TTL insecure_options: allow_private_endpoints: true # example proxies to a local backend diff --git a/examples/configs/gcp-adc.yaml b/examples/configs/gcp-adc.yaml index 1d06df70d3..72f21bcde3 100644 --- a/examples/configs/gcp-adc.yaml +++ b/examples/configs/gcp-adc.yaml @@ -1,13 +1,15 @@ -# GCP ADC upstream authentication (experimental skeleton) +# GCP ADC upstream authentication (experimental) # -# Establishes the gcp_adc filter's configuration surface and -# fail-closed behavior. Token acquisition is NOT implemented -# yet: the token cache is never populated, so every proxied -# request is rejected with 503. Fetch (GKE metadata or a -# service-account key file) lands with the shared background -# refresh primitive tracked in praxis#555/#1042/#1043; once it -# does, the filter will inject "Authorization: Bearer " -# on every proxied request to Vertex AI. +# Acquires an OAuth2 access token from the GCE/GKE metadata server +# (source: adc or metadata) and injects "Authorization: Bearer +# " on every proxied request to Vertex AI. The token is +# cached in memory (cache-through: a request that finds the cache +# stale fetches a fresh token inline before proceeding); the +# downstream client never sees GCP credentials. +# +# source: key_file (a service-account key JSON file) is not +# implemented yet -- it needs JWT signing -- and fails closed with +# a clear "not implemented" reason. # # This filter only injects Authorization — it does NOT set Host # or path. Point the cluster at your Vertex regional endpoint @@ -41,7 +43,6 @@ filter_chains: - filter: gcp_adc # source: adc # default: GOOGLE_APPLICATION_CREDENTIALS, else GKE metadata # scope: https://www.googleapis.com/auth/cloud-platform - # refresh_ratio: 0.75 insecure_options: allow_private_endpoints: true # example proxies to a local backend diff --git a/filters/src/azure/azure_ad.rs b/filters/src/azure/azure_ad.rs index 070882f124..695259fed5 100644 --- a/filters/src/azure/azure_ad.rs +++ b/filters/src/azure/azure_ad.rs @@ -11,19 +11,30 @@ //! # Overview //! //! Enterprise Azure deployments prohibit static API keys and require an -//! `OAuth2` bearer token from Entra ID. This filter acquires such a token -//! via the **client-credentials** grant, caches it, refreshes it in the -//! background before it expires, and injects `Authorization: Bearer -//! ` on every proxied request. The downstream client is unaware -//! of Entra ID. +//! `OAuth2` bearer token from Entra ID. This filter acquires such a +//! token via the **client-credentials** grant and injects +//! `Authorization: Bearer ` on every proxied request. The +//! downstream client is unaware of Entra ID. +//! +//! # Caching: cache-through, not refresh-ahead +//! +//! There is no background refresh thread. Every request checks the +//! cached token; if it is still valid (outside the [`EXPIRY_SKEW`] +//! safety margin) it is used immediately with no network call. If it is +//! missing or stale, that request's handling acquires a fresh token +//! inline before proceeding — see +//! [`praxis_ai_apis::token_cache::TokenCache`] for the exact +//! cache-through/double-checked-locking contract, including how +//! concurrent requests that all observe a stale cache still trigger at +//! most one token-endpoint call. //! //! # Scope //! //! This filter currently supports the **client-secret** credential only. //! Managed identity (AKS/IMDS), client certificates (`private_key_jwt` //! assertions), and OIDC/workload-identity federation are planned -//! follow-ups; they slot into the same token-cache/refresh machinery -//! this filter already uses, with no change to request handling. +//! follow-ups; they slot into the same cache-through machinery this +//! filter already uses, with no change to request handling. //! //! # Routing vs. authentication //! @@ -38,14 +49,11 @@ //! //! # Failure behavior //! -//! The filter **fails closed**: until the first token is acquired, and -//! whenever the cached token is missing or expired, requests are -//! rejected with `503` rather than forwarded unauthenticated. Token -//! acquisition happens asynchronously in the background so that building -//! or hot-reloading a pipeline never blocks on a network round-trip to -//! Entra ID. Repeated acquisition failures back off exponentially (up to -//! 15 minutes) so an unreachable endpoint does not become a tight retry -//! loop. +//! The filter **fails closed**: whenever no valid token can be produced +//! — none cached and the inline fetch fails — the request is rejected +//! with `503` rather than forwarded unauthenticated. There is no +//! server-side retry loop; a failed fetch is not cached, so the next +//! request simply tries again. //! //! # YAML config //! @@ -56,23 +64,17 @@ //! scope: https://cognitiveservices.azure.com/.default //! client_secret_env_var: AZURE_CLIENT_SECRET //! authority_host: login.microsoftonline.com # optional, for sovereign clouds -//! refresh_ratio: 0.75 # optional, refresh at 75% of the usable lifetime //! ``` use std::{ - sync::{ - Arc, - atomic::{AtomicBool, Ordering}, - }, - thread::JoinHandle, - time::{Duration, Instant}, + sync::atomic::{AtomicBool, Ordering}, + time::Duration, }; -use arc_swap::ArcSwap; use http::{HeaderValue, header}; +use praxis_ai_apis::token_cache::TokenCache; use praxis_filter::FilterError; use serde::Deserialize; -use tokio_util::sync::CancellationToken; use tracing::warn; // ----------------------------------------------------------------------------- @@ -81,56 +83,12 @@ use tracing::warn; /// Treat a cached token as expired this long before its real expiry, so /// a token is never injected onto a request that could outlive it in -/// flight. +/// flight. Passed to [`TokenCache::new`] as its safety margin. const EXPIRY_SKEW: Duration = Duration::from_secs(30); -/// Base delay before retrying after a failed token acquisition. Grows -/// exponentially with consecutive failures, capped at -/// [`MAX_RETRY_BACKOFF`]. -const RETRY_BACKOFF: Duration = Duration::from_secs(30); - -/// Upper bound on the exponential retry backoff, so a persistently -/// unreachable token endpoint settles into an infrequent poll rather than -/// a tight loop. -const MAX_RETRY_BACKOFF: Duration = Duration::from_secs(900); - -/// Lower bound on the scheduled refresh delay, so a pathologically small -/// `expires_in` from the token endpoint cannot spin the refresher. -const MIN_REFRESH_DELAY: Duration = Duration::from_secs(1); - /// Timeout for a single token-endpoint round-trip. const TOKEN_REQUEST_TIMEOUT: Duration = Duration::from_secs(30); -/// How long [`RefreshHandle::drop`] waits for the refresher thread to -/// exit before giving up and logging. -const JOIN_TIMEOUT: Duration = Duration::from_secs(2); - -// ----------------------------------------------------------------------------- -// Cached token -// ----------------------------------------------------------------------------- - -/// A bearer token cached in memory, pre-formatted for injection. -/// -/// The token is stored as a ready-to-use `Authorization` header value so -/// the request hot path does no string work — it clones a `HeaderValue` -/// and nothing more. -struct CachedToken { - /// The complete `Authorization` header value (`"Bearer "`), - /// marked sensitive so it is redacted from header debug output. - authorization: HeaderValue, - - /// Instant, already adjusted by [`EXPIRY_SKEW`], after which the - /// token must not be used. - expires_at: Instant, -} - -impl CachedToken { - /// Whether the token is still safe to inject at `now`. - fn is_valid(&self, now: Instant) -> bool { - now < self.expires_at - } -} - // ----------------------------------------------------------------------------- // Token endpoint // ----------------------------------------------------------------------------- @@ -150,8 +108,9 @@ struct TokenResponse { /// client-credentials grant. /// /// Returns the ready-to-inject `Authorization` header value and the -/// token's lifetime. Kept free of [`CachedToken`]/[`Instant`] so it can -/// be unit-tested against a local mock endpoint. +/// token's lifetime. Kept free of caching concerns so it can be +/// unit-tested against a local mock endpoint, and so it fits +/// [`TokenCache::get_or_refresh`]'s `fetch` closure shape directly. /// /// # Errors /// @@ -201,208 +160,6 @@ async fn fetch_token( Ok((authorization, Duration::from_secs(token.expires_in))) } -/// Compute how long to wait before the next refresh: `lifetime * ratio`, -/// floored at [`MIN_REFRESH_DELAY`]. Callers pass the skew-adjusted -/// usable lifetime, not the raw TTL, so the refresh always fires while -/// the cached token is still valid. -fn refresh_delay(lifetime: Duration, ratio: f64) -> Duration { - lifetime.mul_f64(ratio).max(MIN_REFRESH_DELAY) -} - -/// Safety margin to subtract from a token's TTL before caching its -/// expiry. Normally [`EXPIRY_SKEW`], but never more than half the TTL, so -/// a short-lived token stays usable for part of its life instead of being -/// cached already-expired (which would fail every request closed). -fn effective_skew(ttl: Duration) -> Duration { - EXPIRY_SKEW.min(ttl / 2) -} - -/// Exponential backoff after `failures` consecutive fetch failures: -/// `RETRY_BACKOFF * 2^(failures - 1)`, capped at [`MAX_RETRY_BACKOFF`]. -fn retry_backoff(failures: u32) -> Duration { - // Bound the shift so it can never exceed u32 width; the result is - // capped anyway, so a large shift just saturates to the ceiling. - let shift = failures.saturating_sub(1).min(20); - RETRY_BACKOFF.saturating_mul(1_u32 << shift).min(MAX_RETRY_BACKOFF) -} - -// ----------------------------------------------------------------------------- -// Background refresher -// ----------------------------------------------------------------------------- - -/// Inputs the background refresher needs to acquire and cache tokens. -struct RefresherParams { - /// Fully-formed token endpoint URL. - token_url: String, - - /// Application (client) ID. - client_id: String, - - /// Client secret, resolved from the configured environment variable. - client_secret: String, - - /// `OAuth2` scope (e.g. `https://cognitiveservices.azure.com/.default`). - scope: String, - - /// Fraction of a token's usable lifetime (TTL minus the expiry - /// safety margin) at which to refresh it. - refresh_ratio: f64, - - /// Shared cache the filter's hot path reads from. - shared: Arc>>, -} - -/// Owns the background refresher thread and stops it on drop. -/// -/// Dropping the handle cancels the refresher (the filter pipeline was -/// swapped or shut down) and joins the thread, bounded by -/// [`JOIN_TIMEOUT`]. This is the shutdown signal required for background -/// tasks. -struct RefreshHandle { - /// Cancellation signal for the refresher loop. - shutdown: CancellationToken, - - /// Refresher thread join handle. - thread: Option>, -} - -impl Drop for RefreshHandle { - #[expect( - clippy::disallowed_methods, - reason = "Drop is sync; tokio::time::sleep cannot be used here (mirrors routing::overlay)" - )] - fn drop(&mut self) { - self.shutdown.cancel(); - if let Some(handle) = self.thread.take() { - let start = Instant::now(); - while !handle.is_finished() { - if start.elapsed() >= JOIN_TIMEOUT { - warn!( - timeout_secs = JOIN_TIMEOUT.as_secs(), - "azure_ad: token refresher thread did not exit within timeout" - ); - return; - } - std::thread::sleep(Duration::from_millis(5)); - } - drop(handle.join()); - } - } -} - -/// Spawn the background token refresher on a dedicated thread with its -/// own current-thread runtime (mirrors `routing::overlay`), so token -/// acquisition never blocks the pipeline build or the request hot path. -fn spawn_refresher(params: RefresherParams) -> RefreshHandle { - let shutdown = CancellationToken::new(); - let token = shutdown.clone(); - - let thread = std::thread::Builder::new() - .name("azure-ad-token-refresher".to_owned()) - .spawn( - move || match tokio::runtime::Builder::new_current_thread().enable_all().build() { - Ok(runtime) => runtime.block_on(refresh_loop(params, token)), - Err(error) => { - warn!(%error, "azure_ad: failed to create refresher runtime; tokens will not be acquired"); - }, - }, - ); - - let thread = match thread { - Ok(handle) => Some(handle), - Err(error) => { - // Without the refresher the cache stays empty and every - // request fails closed (503) — safe, but log loudly. - warn!(%error, "azure_ad: failed to spawn token refresher thread; requests will fail closed"); - None - }, - }; - - RefreshHandle { shutdown, thread } -} - -/// Acquire a token once and publish it to the shared cache. -/// -/// Returns `Some(delay)` — the delay until the next scheduled refresh — -/// on success, or `None` on failure. On failure the cache is left -/// untouched, so a still-valid token keeps serving. -async fn refresh_once(client: &reqwest::Client, params: &RefresherParams) -> Option { - match fetch_token( - client, - ¶ms.token_url, - ¶ms.client_id, - ¶ms.client_secret, - ¶ms.scope, - ) - .await - { - Ok((authorization, ttl)) => { - if ttl <= EXPIRY_SKEW { - warn!( - ttl_secs = ttl.as_secs(), - "azure_ad: token TTL is unusually short; validity margin reduced" - ); - } - Some(publish_token(params, authorization, ttl)) - }, - Err(error) => { - warn!(%error, "azure_ad: token refresh failed; will retry"); - None - }, - } -} - -/// Cache a freshly fetched token and return the delay until the next -/// scheduled refresh. -/// -/// Both the cached expiry and the refresh schedule are computed from the -/// skew-adjusted usable lifetime. Scheduling from the raw TTL instead -/// (`ratio * ttl`) can land past `ttl - skew`, leaving a window every -/// cycle where the cached token is already invalid but the refresh has -/// not fired yet — deterministic 503s on a healthy token endpoint. -fn publish_token(params: &RefresherParams, authorization: HeaderValue, ttl: Duration) -> Duration { - let usable = ttl.saturating_sub(effective_skew(ttl)); - params.shared.store(Arc::new(Some(CachedToken { - authorization, - expires_at: Instant::now() + usable, - }))); - refresh_delay(usable, params.refresh_ratio) -} - -/// Repeatedly acquire a token, publish it to the shared cache, and sleep -/// until the next refresh — until cancelled. -async fn refresh_loop(params: RefresherParams, shutdown: CancellationToken) { - let client = match reqwest::Client::builder().timeout(TOKEN_REQUEST_TIMEOUT).build() { - Ok(client) => client, - Err(error) => { - warn!(%error, "azure_ad: failed to build HTTP client; requests will fail closed"); - return; - }, - }; - - let mut failures: u32 = 0; - loop { - // Race the fetch against cancellation so a pipeline drop is not - // blocked behind an in-flight token request (up to its 30s - // timeout) — `RefreshHandle::drop` only waits `JOIN_TIMEOUT`. - let refreshed = tokio::select! { - refreshed = refresh_once(&client, ¶ms) => refreshed, - () = shutdown.cancelled() => break, - }; - let delay = if let Some(delay) = refreshed { - failures = 0; - delay - } else { - failures = failures.saturating_add(1); - retry_backoff(failures) - }; - tokio::select! { - () = tokio::time::sleep(delay) => {}, - () = shutdown.cancelled() => break, - } - } -} - // ----------------------------------------------------------------------------- // Filter // ----------------------------------------------------------------------------- @@ -417,29 +174,41 @@ async fn refresh_loop(params: RefresherParams, shutdown: CancellationToken) { /// See the module docs for scope (client-secret only), the /// routing-vs-authentication separation, and the fail-closed behavior. pub struct AzureAdFilter { - /// Lock-free cache of the current token, populated by the background - /// refresher and read on every request. - token: Arc>>, + /// Cache-through token cache; see the module docs and + /// [`praxis_ai_apis::token_cache`]. + cache: TokenCache, + + /// HTTP client used for token-endpoint requests, built once with + /// [`TOKEN_REQUEST_TIMEOUT`]. + client: reqwest::Client, + + /// Fully-formed token endpoint URL. + token_url: String, + + /// Application (client) ID. + client_id: String, + + /// Client secret, resolved from the configured environment variable. + client_secret: String, + + /// `OAuth2` scope (e.g. `https://cognitiveservices.azure.com/.default`). + scope: String, /// Whether the filter is currently failing closed, so the missing- /// token condition is logged on state transitions instead of once /// per rejected request. failing: AtomicBool, - - /// Background refresher; stops when the filter is dropped. - _refresh: RefreshHandle, } impl AzureAdFilter { /// Build a filter from parsed config, resolving the client secret - /// from its environment variable and spawning the refresher. + /// from its environment variable. /// /// # Errors /// - /// Returns [`FilterError`] if `refresh_ratio` is out of range, - /// `authority_host` or `tenant_id` contain URL-structural characters, - /// or the configured secret environment variable is unset or not - /// UTF-8. + /// Returns [`FilterError`] if `authority_host` or `tenant_id` contain + /// URL-structural characters, the configured secret environment + /// variable is unset or not UTF-8, or the HTTP client fails to build. fn new(config: AzureAdConfig) -> Result { validate_config(&config)?; @@ -454,21 +223,19 @@ impl AzureAdFilter { "https://{}/{}/oauth2/v2.0/token", config.authority_host, config.tenant_id ); - let shared = Arc::new(ArcSwap::from_pointee(None)); + let client = reqwest::Client::builder() + .timeout(TOKEN_REQUEST_TIMEOUT) + .build() + .map_err(|e| FilterError::from(format!("azure_ad: failed to build HTTP client: {e}")))?; - let refresh = spawn_refresher(RefresherParams { + Ok(Self { + cache: TokenCache::new(EXPIRY_SKEW), + client, token_url, client_id: config.client_id, client_secret, scope: config.scope, - refresh_ratio: config.refresh_ratio, - shared: Arc::clone(&shared), - }); - - Ok(Self { - token: shared, failing: AtomicBool::new(false), - _refresh: refresh, }) } @@ -502,23 +269,25 @@ impl praxis_filter::HttpFilter for AzureAdFilter { &self, ctx: &mut praxis_filter::HttpFilterContext<'_>, ) -> Result { - let cached = self.token.load(); - match cached.as_ref() { - Some(token) if token.is_valid(Instant::now()) => { + let fetched = self + .cache + .get_or_refresh(|| fetch_token(&self.client, &self.token_url, &self.client_id, &self.client_secret, &self.scope)) + .await; + match fetched { + Ok(authorization) => { if self.failing.swap(false, Ordering::Relaxed) { warn!("azure_ad: valid token available again; resuming request forwarding"); } // Cheap: clone a pre-formatted, sensitive HeaderValue. - ctx.request_headers_to_set - .push((header::AUTHORIZATION, token.authorization.clone())); + ctx.request_headers_to_set.push((header::AUTHORIZATION, authorization)); Ok(praxis_filter::FilterAction::Continue) }, - _ => { + Err(error) => { // Fail closed: never forward an unauthenticated request. // Log on the state transition, not per rejected request, // so a token outage under load cannot flood the logs. if !self.failing.swap(true, Ordering::Relaxed) { - warn!("azure_ad: no valid cached token; rejecting requests with 503 until one is available"); + warn!(%error, "azure_ad: no valid token available; rejecting requests with 503 until one is acquired"); } Ok(praxis_filter::FilterAction::Reject(praxis_filter::Rejection::status( 503, @@ -553,23 +322,11 @@ pub(crate) struct AzureAdConfig { /// `login.microsoftonline.us`). #[serde(default = "default_authority_host")] pub(crate) authority_host: String, - - /// Fraction of a token's usable lifetime (TTL minus the expiry - /// safety margin) at which to refresh it. Must be in the open - /// interval `(0, 1)`. - #[serde(default = "default_refresh_ratio")] - pub(crate) refresh_ratio: f64, } /// Validate the config fields [`AzureAdFilter::new`] relies on before it -/// reads the secret and spawns the refresher. +/// reads the secret. fn validate_config(config: &AzureAdConfig) -> Result<(), FilterError> { - if !(config.refresh_ratio > 0.0 && config.refresh_ratio < 1.0) { - return Err(FilterError::from(format!( - "azure_ad: refresh_ratio must be between 0 and 1 (exclusive), got {}", - config.refresh_ratio - ))); - } validate_url_component("authority_host", &config.authority_host)?; validate_url_component("tenant_id", &config.tenant_id)?; Ok(()) @@ -603,11 +360,6 @@ fn default_authority_host() -> String { "login.microsoftonline.com".to_owned() } -/// Default value for [`AzureAdConfig::refresh_ratio`]. -fn default_refresh_ratio() -> f64 { - 0.75 -} - /// Parse and validate the `azure_ad` filter's YAML config. /// /// # Errors @@ -637,11 +389,7 @@ mod tests { use http::Method; use praxis_filter::{FilterAction, HttpFilter as _}; - use super::{ - AzureAdFilter, CachedToken, EXPIRY_SKEW, MAX_RETRY_BACKOFF, MIN_REFRESH_DELAY, RETRY_BACKOFF, RefreshHandle, - RefresherParams, effective_skew, fetch_token, parse_azure_ad_config, refresh_delay, refresh_once, - retry_backoff, validate_url_component, - }; + use super::{AzureAdConfig, AzureAdFilter, fetch_token, parse_azure_ad_config, validate_url_component}; use crate::test_utils::{make_filter_context, make_request}; fn yaml(body: &str) -> serde_yaml::Value { @@ -667,26 +415,20 @@ mod tests { config.authority_host, "login.microsoftonline.com", "authority_host should default" ); - assert!( - (config.refresh_ratio - 0.75).abs() < f64::EPSILON, - "refresh_ratio should default to 0.75" - ); } #[test] - fn parses_optional_authority_host_and_refresh_ratio() { + fn parses_optional_authority_host() { let config = parse_azure_ad_config(&yaml( "tenant_id: tid\n\ client_id: cid\n\ scope: s\n\ client_secret_env_var: AZURE_CLIENT_SECRET\n\ - authority_host: login.microsoftonline.us\n\ - refresh_ratio: 0.5\n", + authority_host: login.microsoftonline.us\n", )) .expect("full config should parse"); assert_eq!(config.authority_host, "login.microsoftonline.us"); - assert!((config.refresh_ratio - 0.5).abs() < f64::EPSILON); } #[test] @@ -711,31 +453,8 @@ mod tests { assert!(err.is_err(), "unknown field should be rejected by deny_unknown_fields"); } - #[test] - fn new_rejects_out_of_range_refresh_ratio() { - for bad in [0.0, 1.0, 1.5, -0.1] { - let cfg = super::AzureAdConfig { - tenant_id: "tid".to_owned(), - client_id: "cid".to_owned(), - scope: "s".to_owned(), - client_secret_env_var: "AZURE_TEST_UNSET_SECRET".to_owned(), - authority_host: super::default_authority_host(), - refresh_ratio: bad, - }; - match AzureAdFilter::new(cfg) { - Ok(_) => panic!("out-of-range refresh_ratio ({bad}) must be rejected"), - Err(err) => assert!( - format!("{err}").contains("refresh_ratio"), - "error must name the offending field, got: {err}" - ), - } - } - } - #[test] fn from_config_propagates_missing_secret() { - // refresh_ratio is valid, so construction proceeds to the - // credential resolution step and fails there. let err = AzureAdFilter::from_config(&yaml( "tenant_id: tid\n\ client_id: cid\n\ @@ -777,13 +496,12 @@ mod tests { #[test] fn new_rejects_authority_host_with_userinfo_override() { - let cfg = super::AzureAdConfig { + let cfg = AzureAdConfig { tenant_id: "tid".to_owned(), client_id: "cid".to_owned(), scope: "s".to_owned(), client_secret_env_var: "AZURE_TEST_UNSET_SECRET".to_owned(), authority_host: "login.microsoftonline.com@evil.com".to_owned(), - refresh_ratio: 0.75, }; match AzureAdFilter::new(cfg) { Ok(_) => panic!("malicious authority_host must be rejected"), @@ -794,125 +512,6 @@ mod tests { } } - // -- Pure helpers --------------------------------------------------------- - - #[test] - fn refresh_delay_is_ttl_times_ratio() { - let delay = refresh_delay(std::time::Duration::from_secs(3600), 0.75); - assert_eq!(delay, std::time::Duration::from_secs(2700)); - } - - #[test] - fn refresh_delay_is_floored() { - // 1s * 0.75 = 750ms, below the floor. - let delay = refresh_delay(std::time::Duration::from_secs(1), 0.75); - assert_eq!(delay, MIN_REFRESH_DELAY); - } - - #[test] - fn effective_skew_is_capped_at_half_ttl() { - use std::time::Duration; - // Long tokens get the full skew. - assert_eq!(effective_skew(Duration::from_secs(3600)), EXPIRY_SKEW); - // Short tokens get at most half their TTL, so they are never - // cached already-expired. - assert_eq!(effective_skew(Duration::from_secs(40)), Duration::from_secs(20)); - assert_eq!(effective_skew(Duration::from_secs(10)), Duration::from_secs(5)); - } - - #[test] - fn retry_backoff_grows_then_caps() { - assert_eq!(retry_backoff(0), RETRY_BACKOFF, "no failures yet -> base delay"); - assert_eq!(retry_backoff(1), RETRY_BACKOFF, "first failure -> base delay"); - assert_eq!(retry_backoff(2), RETRY_BACKOFF * 2); - assert_eq!(retry_backoff(3), RETRY_BACKOFF * 4); - assert_eq!(retry_backoff(1000), MAX_RETRY_BACKOFF, "large failure count -> capped"); - } - - #[test] - fn cached_token_expiry() { - let now = std::time::Instant::now(); - let token = CachedToken { - authorization: http::HeaderValue::from_static("Bearer x"), - expires_at: now + std::time::Duration::from_secs(60), - }; - assert!(token.is_valid(now), "token in the future must be valid"); - assert!( - !token.is_valid(now + std::time::Duration::from_secs(61)), - "token past expiry must be invalid" - ); - } - - // -- on_request ----------------------------------------------------------- - - /// Build a filter directly around a given cache, without spawning a - /// refresher thread (the `RefreshHandle` has no thread, so its - /// `Drop` only cancels a token). - fn test_filter(token: Option) -> AzureAdFilter { - AzureAdFilter { - token: std::sync::Arc::new(arc_swap::ArcSwap::from_pointee(token)), - failing: std::sync::atomic::AtomicBool::new(false), - _refresh: RefreshHandle { - shutdown: tokio_util::sync::CancellationToken::new(), - thread: None, - }, - } - } - - #[tokio::test] - async fn on_request_injects_bearer_when_token_valid() { - let filter = test_filter(Some(CachedToken { - authorization: http::HeaderValue::from_static("Bearer test-token"), - expires_at: std::time::Instant::now() + std::time::Duration::from_secs(300), - })); - let request = make_request(Method::POST, "/v1/chat/completions"); - let mut ctx = make_filter_context(&request); - - let action = filter.on_request(&mut ctx).await.expect("must not error"); - assert!(matches!(action, FilterAction::Continue)); - - let auth = ctx - .request_headers_to_set - .iter() - .find(|(name, _)| *name == http::header::AUTHORIZATION) - .map(|(_, value)| value.to_str().expect("ascii")); - assert_eq!(auth, Some("Bearer test-token"), "must inject the cached bearer token"); - } - - #[tokio::test] - async fn on_request_fails_closed_when_no_token() { - let filter = test_filter(None); - let request = make_request(Method::POST, "/v1/chat/completions"); - let mut ctx = make_filter_context(&request); - - let action = filter.on_request(&mut ctx).await.expect("must reject, not error"); - assert!( - matches!(action, FilterAction::Reject(r) if r.status == 503), - "no cached token must fail closed with 503" - ); - assert!( - ctx.request_headers_to_set.is_empty(), - "no headers must be set when failing closed" - ); - } - - #[tokio::test] - async fn on_request_fails_closed_when_token_expired() { - let filter = test_filter(Some(CachedToken { - authorization: http::HeaderValue::from_static("Bearer stale"), - // Expired well beyond the skew. - expires_at: std::time::Instant::now() - (EXPIRY_SKEW + std::time::Duration::from_secs(1)), - })); - let request = make_request(Method::POST, "/v1/chat/completions"); - let mut ctx = make_filter_context(&request); - - let action = filter.on_request(&mut ctx).await.expect("must reject, not error"); - assert!( - matches!(action, FilterAction::Reject(r) if r.status == 503), - "expired token must fail closed with 503" - ); - } - // -- fetch_token against a mock endpoint ---------------------------------- /// Spawn a one-shot HTTP/1.1 server on loopback that replies with @@ -950,69 +549,6 @@ mod tests { server.join().unwrap(); } - #[tokio::test] - async fn refresh_once_caches_usable_token_for_short_ttl() { - // A TTL below the full EXPIRY_SKEW must still yield a token that - // is valid right now — regression for the skew-saturates-to-zero - // bug that would cache an already-expired token and 503 forever. - let (url, server) = mock_token_endpoint(r#"{"access_token":"short","expires_in":40}"#); - let shared = std::sync::Arc::new(arc_swap::ArcSwap::from_pointee(None)); - let params = RefresherParams { - token_url: url, - client_id: "cid".to_owned(), - client_secret: "secret".to_owned(), - scope: "scope".to_owned(), - refresh_ratio: 0.75, - shared: std::sync::Arc::clone(&shared), - }; - let client = reqwest::Client::new(); - - let delay = refresh_once(&client, ¶ms) - .await - .expect("short-ttl fetch must succeed"); - assert!(delay >= MIN_REFRESH_DELAY, "refresh delay must respect the floor"); - - let cached = shared.load_full(); - let token = cached.as_ref().as_ref().expect("a successful fetch must cache a token"); - let now = std::time::Instant::now(); - assert!( - token.is_valid(now), - "a 40s token must be usable now, not cached already-expired" - ); - // Regression: the refresh must be scheduled from the - // skew-adjusted usable lifetime. Scheduling from the raw - // TTL (0.75 * 40s = 30s) lands past the cached expiry - // (40s - 20s skew = 20s), leaving a guaranteed window of - // 503s every cycle even with a healthy token endpoint. - assert!( - token.is_valid(now + delay), - "the cached token must still be valid when the scheduled refresh fires" - ); - server.join().unwrap(); - } - - #[tokio::test] - async fn refresh_once_returns_none_on_failure() { - // Closed port -> fetch fails -> no token cached, None returned so - // the loop applies backoff. - let shared = std::sync::Arc::new(arc_swap::ArcSwap::from_pointee(None)); - let params = RefresherParams { - token_url: "http://127.0.0.1:1/token".to_owned(), - client_id: "cid".to_owned(), - client_secret: "secret".to_owned(), - scope: "scope".to_owned(), - refresh_ratio: 0.75, - shared: std::sync::Arc::clone(&shared), - }; - let client = reqwest::Client::new(); - - assert!( - refresh_once(&client, ¶ms).await.is_none(), - "a failed fetch must return None" - ); - assert!(shared.load_full().is_none(), "a failed fetch must not cache a token"); - } - #[tokio::test] async fn fetch_token_errors_on_non_success_status() { let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); @@ -1034,4 +570,91 @@ mod tests { assert!(format!("{err}").contains("401"), "error must carry the status: {err}"); server.join().unwrap(); } + + // -- on_request: cache-through end to end ---------------------------------- + + /// Build a filter with a real, always-set secret env var (so + /// construction never fails on credential resolution) and then point + /// its token endpoint at `token_url` — bypassing `authority_host`/ + /// `tenant_id` URL construction entirely so tests can target a local + /// mock server directly. + fn filter_at(token_url: &str) -> AzureAdFilter { + let config = parse_azure_ad_config(&yaml( + "tenant_id: tid\n\ + client_id: cid\n\ + scope: scope\n\ + client_secret_env_var: CARGO_PKG_NAME\n", + )) + .expect("test config must parse"); + let mut filter = + AzureAdFilter::new(config).expect("construction must succeed with an always-set secret env var"); + filter.token_url = token_url.to_owned(); + filter + } + + #[tokio::test] + async fn on_request_injects_bearer_on_first_fetch() { + let (url, server) = mock_token_endpoint(r#"{"access_token":"fresh","expires_in":3600}"#); + let filter = filter_at(&url); + let request = make_request(Method::POST, "/openai/deployments/gpt-4o/chat/completions"); + let mut ctx = make_filter_context(&request); + + let action = filter.on_request(&mut ctx).await.expect("must not error"); + assert!(matches!(action, FilterAction::Continue)); + + let auth = ctx + .request_headers_to_set + .iter() + .find(|(name, _)| *name == http::header::AUTHORIZATION) + .map(|(_, value)| value.to_str().expect("ascii")); + assert_eq!(auth, Some("Bearer fresh"), "must inject the freshly fetched bearer token"); + server.join().unwrap(); + } + + #[tokio::test] + async fn on_request_reuses_cached_token_without_a_second_fetch() { + // The mock endpoint accepts exactly one connection; a second + // on_request call must not attempt a second fetch, or it would + // fail to connect and 503 instead of continuing. + let (url, server) = mock_token_endpoint(r#"{"access_token":"once","expires_in":3600}"#); + let filter = filter_at(&url); + let request = make_request(Method::POST, "/openai/deployments/gpt-4o/chat/completions"); + + let mut first_ctx = make_filter_context(&request); + let first = filter.on_request(&mut first_ctx).await.expect("first call must not error"); + assert!(matches!(first, FilterAction::Continue)); + + let mut second_ctx = make_filter_context(&request); + let second = filter.on_request(&mut second_ctx).await.expect("second call must not error"); + assert!( + matches!(second, FilterAction::Continue), + "a still-valid cache must serve the second request without a new connection" + ); + let auth = second_ctx + .request_headers_to_set + .iter() + .find(|(name, _)| *name == http::header::AUTHORIZATION) + .map(|(_, value)| value.to_str().expect("ascii")); + assert_eq!(auth, Some("Bearer once")); + server.join().unwrap(); + } + + #[tokio::test] + async fn on_request_fails_closed_when_fetch_fails() { + // A closed local port: the connection is refused immediately, no + // real network dependency. + let filter = filter_at("http://127.0.0.1:1/token"); + let request = make_request(Method::POST, "/openai/deployments/gpt-4o/chat/completions"); + let mut ctx = make_filter_context(&request); + + let action = filter.on_request(&mut ctx).await.expect("must reject, not error"); + assert!( + matches!(action, FilterAction::Reject(r) if r.status == 503), + "a failed fetch must fail closed with 503" + ); + assert!( + ctx.request_headers_to_set.is_empty(), + "no headers must be set when failing closed" + ); + } } diff --git a/filters/src/gcp/config.rs b/filters/src/gcp/config.rs index d1dfea36e3..8897a013ee 100644 --- a/filters/src/gcp/config.rs +++ b/filters/src/gcp/config.rs @@ -39,7 +39,6 @@ pub(super) enum GcpAdcSource { /// filter: gcp_adc /// source: adc /// scope: https://www.googleapis.com/auth/cloud-platform -/// refresh_ratio: 0.75 /// ``` #[derive(Debug, Deserialize)] #[serde(deny_unknown_fields)] @@ -66,10 +65,11 @@ pub(super) struct GcpAdcConfig { #[serde(default)] pub credentials_file: Option, - /// Fraction of a token's TTL at which to refresh it. Must be in the - /// open interval `(0, 1)`. - #[serde(default = "default_refresh_ratio")] - pub refresh_ratio: f64, + /// GCE/GKE metadata server host. Override for testing (point at a + /// local mock) or a non-default metadata server; production + /// deployments should not need to set this. + #[serde(default = "default_metadata_host")] + pub metadata_host: String, } /// Default `OAuth2` scope for Vertex AI and other GCP APIs. @@ -77,9 +77,9 @@ fn default_scope() -> String { "https://www.googleapis.com/auth/cloud-platform".to_owned() } -/// Default value for [`GcpAdcConfig::refresh_ratio`]. -fn default_refresh_ratio() -> f64 { - 0.75 +/// Default value for [`GcpAdcConfig::metadata_host`]. +fn default_metadata_host() -> String { + "metadata.google.internal".to_owned() } /// Parse and validate the `gcp_adc` filter's YAML config. @@ -98,22 +98,35 @@ pub(super) fn parse_gcp_adc_config(config: &serde_yaml::Value) -> Result Result<(), FilterError> { - if !(config.refresh_ratio > 0.0 && config.refresh_ratio < 1.0) { - return Err(format!( - "gcp_adc: refresh_ratio must be between 0 and 1 (exclusive), got {}", - config.refresh_ratio - ) - .into()); - } if config.scope.is_empty() { return Err("gcp_adc: scope must not be empty".into()); } + validate_url_component("metadata_host", &config.metadata_host)?; match config.source { GcpAdcSource::KeyFile => validate_key_file_fields(config), GcpAdcSource::Metadata | GcpAdcSource::Adc => validate_metadata_fields(config), } } +/// Reject a config value that could break out of its URL component — +/// `metadata_host` is interpolated as the URL authority of the metadata +/// request. This does not attempt full hostname validation — it only +/// forbids the characters that change URL structure. +pub(super) fn validate_url_component(field: &str, value: &str) -> Result<(), FilterError> { + if value.is_empty() { + return Err(format!("gcp_adc: {field} must not be empty").into()); + } + let forbidden = |c: char| matches!(c, '/' | '\\' | '?' | '#' | '@') || c.is_whitespace() || c.is_control(); + if value.contains(forbidden) { + return Err(format!( + "gcp_adc: {field} '{value}' is invalid: it must be a bare value with no scheme, \ + path, query, '@', or whitespace" + ) + .into()); + } + Ok(()) +} + /// `key_file` requires `credentials_file` and does not use /// `service_account`. fn validate_key_file_fields(config: &GcpAdcConfig) -> Result<(), FilterError> { diff --git a/filters/src/gcp/filter.rs b/filters/src/gcp/filter.rs index 871961c78b..bcec5bfa15 100644 --- a/filters/src/gcp/filter.rs +++ b/filters/src/gcp/filter.rs @@ -4,71 +4,31 @@ //! [`GcpAdcFilter`] implementation and `HttpFilter` trait impl. use std::{ - sync::{ - Arc, - atomic::{AtomicBool, Ordering}, - }, - time::Instant, + sync::atomic::{AtomicBool, Ordering}, + time::Duration, }; -use arc_swap::ArcSwap; -use http::header; +use http::{HeaderValue, header}; +use praxis_ai_apis::token_cache::TokenCache; use praxis_filter::{FilterError, parse_filter_config}; use tracing::warn; use super::{ config::{GcpAdcConfig, validate_config}, - token::resolve_token_source, + token::{self, TokenSource, resolve_token_source}, }; // ----------------------------------------------------------------------------- -// Cached token +// Constants // ----------------------------------------------------------------------------- -/// A bearer token cached in memory, pre-formatted for injection. -/// -/// The only way to construct one is [`CachedToken::new`], which formats -/// the `Authorization` value and marks it sensitive, so a token can -/// never reach the cache unredacted (`HeaderValue`'s `Debug` output -/// redacts sensitive values). -#[derive(Debug)] -pub(super) struct CachedToken { - /// The complete `Authorization` header value (`"Bearer "`), - /// marked sensitive so it is redacted from header debug output. - authorization: http::HeaderValue, - - /// Instant after which the token must not be used. The producer is - /// responsible for subtracting a safety skew from the real expiry. - expires_at: Instant, -} - -impl CachedToken { - /// Build a cached token from a raw access token, formatting the - /// `Authorization` value and marking it sensitive. - /// - /// Currently only exercised by tests; token acquisition will call - /// this once fetch is implemented on top of the core background-task - /// primitive (praxis#1043). - /// - /// # Errors - /// - /// Returns [`FilterError`] if the token is not a valid header value. - #[cfg(test)] - pub(super) fn new(access_token: &str, expires_at: Instant) -> Result { - let mut authorization = http::HeaderValue::from_str(&format!("Bearer {access_token}")) - .map_err(|e| FilterError::from(format!("gcp_adc: token is not a valid header value: {e}")))?; - authorization.set_sensitive(true); - Ok(Self { - authorization, - expires_at, - }) - } +/// Treat a cached token as expired this long before its real expiry, so +/// a token is never injected onto a request that could outlive it in +/// flight. Passed to [`TokenCache::new`] as its safety margin. +const EXPIRY_SKEW: Duration = Duration::from_secs(30); - /// Whether the token is still safe to inject at `now`. - pub(super) fn is_valid(&self, now: Instant) -> bool { - now < self.expires_at - } -} +/// Timeout for a single metadata-server round-trip. +const TOKEN_REQUEST_TIMEOUT: Duration = Duration::from_secs(30); // ----------------------------------------------------------------------------- // Filter @@ -81,18 +41,18 @@ impl CachedToken { /// is a work in progress and its configuration surface may change /// between releases. /// -/// **Token acquisition is not implemented yet.** This filter currently -/// establishes the configuration surface, credential-source resolution, -/// and the fail-closed request path: the token cache is never populated, -/// so every request is rejected with `503`. Fetch (metadata server, key -/// file) arrives with the shared background-refresh primitive tracked in -/// praxis#555, praxis#1042, and praxis#1043 — filters must not spawn -/// their own refresher threads. +/// Acquires a token via Application Default Credentials (GKE metadata +/// server) and injects `Authorization: Bearer ` on every proxied +/// request, keeping GCP credentials invisible to the downstream client. +/// There is no background refresh thread: caching is cache-through, the +/// same as [`crate::azure::azure_ad`] — see +/// [`praxis_ai_apis::token_cache::TokenCache`] for the exact contract. /// -/// Once implemented, the filter will acquire a token via Application -/// Default Credentials (GKE metadata or a service-account key file) and -/// inject `Authorization: Bearer ` on every proxied request, -/// keeping GCP credentials invisible to the downstream client. +/// **Service-account key file (`source: key_file`) token fetch is not +/// implemented yet** — it needs `JWT` signing, which this workspace does +/// not currently depend on. Config parsing, file resolution, and +/// validation for `key_file` all work; `on_request` fails closed with a +/// clear "not implemented" reason instead of silently 503ing forever. /// /// Credential-source resolution happens at construct time: /// `GOOGLE_APPLICATION_CREDENTIALS` is read once when the pipeline is @@ -104,9 +64,9 @@ impl CachedToken { /// the correct Vertex endpoint (cluster `endpoints` + `tls.sni`) is /// the operator's responsibility. /// -/// Until a token is cached, and whenever the cached token is missing -/// or expired, requests are rejected with `503` rather than forwarded -/// unauthenticated. +/// Whenever no valid token can be produced — none cached and the inline +/// fetch fails — the request is rejected with `503` rather than +/// forwarded unauthenticated. /// /// # YAML configuration /// @@ -114,13 +74,24 @@ impl CachedToken { /// filter: gcp_adc /// source: adc /// scope: https://www.googleapis.com/auth/cloud-platform -/// refresh_ratio: 0.75 /// ``` pub struct GcpAdcFilter { - /// Lock-free cache of the current token, read on every request. - /// Nothing populates it until token fetch is implemented on top of - /// the core background-task primitive (praxis#1043). - token: Arc>>, + /// Cache-through token cache; see the struct docs and + /// [`praxis_ai_apis::token_cache`]. + cache: TokenCache, + + /// HTTP client used for metadata-server requests, built once with + /// [`TOKEN_REQUEST_TIMEOUT`]. + client: reqwest::Client, + + /// Resolved credential source. + source: TokenSource, + + /// `OAuth2` scope requested with the access token. + scope: String, + + /// GCE/GKE metadata server host. + metadata_host: String, /// Whether the filter is currently failing closed, so the missing- /// token condition is logged on state transitions instead of once @@ -133,17 +104,23 @@ impl GcpAdcFilter { /// /// # Errors /// - /// Returns [`FilterError`] if `refresh_ratio` is out of range, - /// `service_account` is structurally unsafe, a field is set that its - /// `source` does not use, or ADC file resolution fails. + /// Returns [`FilterError`] if `service_account` or `metadata_host` + /// are structurally unsafe, a field is set that its `source` does + /// not use, ADC file resolution fails, or the HTTP client fails to + /// build. fn new(config: &GcpAdcConfig, application_credentials: Option<&std::path::Path>) -> Result { validate_config(config)?; - // Resolution validates the credential source (file readable, - // supported `type`) at the config boundary; the resolved source - // is not stored until token fetch is implemented. - resolve_token_source(config, application_credentials)?; + let source = resolve_token_source(config, application_credentials)?; + let client = reqwest::Client::builder() + .timeout(TOKEN_REQUEST_TIMEOUT) + .build() + .map_err(|e| FilterError::from(format!("gcp_adc: failed to build HTTP client: {e}")))?; Ok(Self { - token: Arc::new(ArcSwap::from_pointee(None)), + cache: TokenCache::new(EXPIRY_SKEW), + client, + source, + scope: config.scope.clone(), + metadata_host: config.metadata_host.clone(), failing: AtomicBool::new(false), }) } @@ -163,15 +140,6 @@ impl GcpAdcFilter { .map(std::path::Path::new), )?)) } - - /// Build a filter around a given cache. - #[cfg(test)] - pub(super) fn for_test(token: Option) -> Self { - Self { - token: Arc::new(ArcSwap::from_pointee(token)), - failing: AtomicBool::new(false), - } - } } #[async_trait::async_trait] @@ -192,19 +160,21 @@ impl praxis_filter::HttpFilter for GcpAdcFilter { &self, ctx: &mut praxis_filter::HttpFilterContext<'_>, ) -> Result { - let cached = self.token.load(); - match cached.as_ref() { - Some(token) if token.is_valid(Instant::now()) => { + let fetched = self + .cache + .get_or_refresh(|| token::fetch(&self.client, &self.source, &self.metadata_host, &self.scope)) + .await; + match fetched { + Ok(authorization) => { if self.failing.swap(false, Ordering::Relaxed) { warn!("gcp_adc: valid token available again; resuming request forwarding"); } - ctx.request_headers_to_set - .push((header::AUTHORIZATION, token.authorization.clone())); + ctx.request_headers_to_set.push((header::AUTHORIZATION, authorization)); Ok(praxis_filter::FilterAction::Continue) }, - _ => { + Err(error) => { if !self.failing.swap(true, Ordering::Relaxed) { - warn!("gcp_adc: no valid cached token; rejecting requests with 503 until one is available"); + warn!(%error, "gcp_adc: no valid token available; rejecting requests with 503 until one is acquired"); } Ok(praxis_filter::FilterAction::Reject(praxis_filter::Rejection::status( 503, diff --git a/filters/src/gcp/mod.rs b/filters/src/gcp/mod.rs index c59118615d..0db3afc94d 100644 --- a/filters/src/gcp/mod.rs +++ b/filters/src/gcp/mod.rs @@ -7,8 +7,9 @@ //! is off by default and activates the `experimental` marker. The //! configuration surface may change between releases. //! -//! Token acquisition is not implemented yet — see [`GcpAdcFilter`] for -//! the current fail-closed behavior and the upstream tracking issues. +//! Token acquisition for the metadata-server sources (`adc`, `metadata`) +//! is implemented; `key_file` is not yet (needs `JWT` signing) — see +//! [`GcpAdcFilter`] for the current behavior. //! //! Classic GKE Workload Identity is the metadata server (ADC), not //! STS/WIF. Vertex AI needs an `OAuth2` access token with a **scope**, diff --git a/filters/src/gcp/tests.rs b/filters/src/gcp/tests.rs index 77a5ed631b..2a9ea0addc 100644 --- a/filters/src/gcp/tests.rs +++ b/filters/src/gcp/tests.rs @@ -3,20 +3,16 @@ //! Unit tests for the GCP ADC upstream-auth filter. -use std::{ - io::Write as _, - time::{Duration, Instant}, -}; +use std::io::{Read as _, Write as _}; use http::{HeaderValue, Method, header}; -use praxis_filter::{FilterAction, HttpFilter as _}; +use praxis_filter::FilterAction; use tempfile::NamedTempFile; use super::{ GcpAdcFilter, config::{parse_gcp_adc_config, validate_service_account}, - filter::CachedToken, - token::{TokenSource, resolve_token_source}, + token::{self, TokenSource, resolve_token_source}, }; use crate::test_utils::{make_filter_context, make_request}; @@ -35,8 +31,23 @@ fn write_json(body: &str) -> NamedTempFile { file } -fn valid_token(access_token: &str) -> CachedToken { - CachedToken::new(access_token, Instant::now() + Duration::from_secs(300)).expect("valid test token") +/// Spawn a one-shot HTTP/1.1 server on loopback that replies with `body` +/// to any request, and returns its bound `host:port`. +fn mock_metadata_endpoint(body: &'static str) -> (String, std::thread::JoinHandle<()>) { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let handle = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + let mut buf = [0_u8; 4096]; + let _ = stream.read(&mut buf).unwrap(); + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + body.len(), + body + ); + stream.write_all(response.as_bytes()).unwrap(); + }); + (addr.to_string(), handle) } // ----------------------------------------------------------------------------- @@ -51,9 +62,9 @@ fn parses_minimal_valid_config() { assert_eq!(config.scope, "https://www.googleapis.com/auth/cloud-platform"); assert!(config.service_account.is_none(), "service_account should be unset"); assert!(config.credentials_file.is_none(), "credentials_file should be unset"); - assert!( - (config.refresh_ratio - 0.75).abs() < f64::EPSILON, - "default refresh_ratio should be 0.75" + assert_eq!( + config.metadata_host, "metadata.google.internal", + "metadata_host should default" ); } @@ -64,7 +75,6 @@ fn parses_explicit_metadata_and_key_file() { source: metadata service_account: foo@project.iam.gserviceaccount.com scope: https://www.googleapis.com/auth/cloud-platform -refresh_ratio: 0.5 ", )) .expect("metadata config should parse"); @@ -136,16 +146,14 @@ fn rejects_unknown_field() { } #[test] -fn from_config_rejects_out_of_range_refresh_ratio() { - for ratio in ["0", "1", "-0.1", "1.5"] { - let err = GcpAdcFilter::from_config(&yaml(&format!("refresh_ratio: {ratio}"))) - .err() - .expect("refresh_ratio must be exclusive (0, 1)"); - assert!( - err.to_string().contains("refresh_ratio"), - "error should name refresh_ratio for {ratio}: {err}" - ); - } +fn rejects_structural_characters_in_metadata_host() { + let err = GcpAdcFilter::from_config(&yaml("metadata_host: evil.com/../x")) + .err() + .expect("path-injecting metadata_host must be rejected"); + assert!( + err.to_string().contains("metadata_host"), + "error should name metadata_host: {err}" + ); } #[test] @@ -248,76 +256,103 @@ fn explicit_metadata_ignores_credentials_path() { } // ----------------------------------------------------------------------------- -// CachedToken +// token::fetch against a mock metadata endpoint // ----------------------------------------------------------------------------- #[tokio::test] -async fn cached_token_formats_bearer_and_marks_sensitive() { - let request = make_request(Method::GET, "/"); - let filter = GcpAdcFilter::for_test(Some(valid_token("secret-token"))); - let mut ctx = make_filter_context(&request); - - // The constructor — not the test — must produce a sensitive, - // Bearer-formatted header value. - let action = filter.on_request(&mut ctx).await.expect("must not error"); - assert!(matches!(action, FilterAction::Continue)); - let injected = ctx - .request_headers_to_set - .iter() - .find(|(name, _)| *name == header::AUTHORIZATION) - .map(|(_, value)| value); - let injected = injected.expect("Authorization must be injected"); - assert_eq!(injected.to_str().expect("ascii"), "Bearer secret-token"); - assert!(injected.is_sensitive(), "constructor must mark the value sensitive"); -} - -#[test] -fn cached_token_rejects_invalid_header_value() { - CachedToken::new("bad\ntoken", Instant::now()).expect_err("control characters must be rejected"); +async fn fetch_parses_bearer_and_ttl_for_metadata_source() { + let (host, server) = mock_metadata_endpoint(r#"{"access_token":"abc123","expires_in":3600}"#); + let client = reqwest::Client::new(); + let source = TokenSource::Metadata { + service_account: "default".to_owned(), + }; + + let (authorization, ttl) = token::fetch(&client, &source, &host, "scope") + .await + .expect("mock metadata fetch must succeed"); + + assert_eq!(authorization.to_str().unwrap(), "Bearer abc123"); + assert!(authorization.is_sensitive(), "bearer header must be marked sensitive"); + assert_eq!(ttl, std::time::Duration::from_secs(3600)); + server.join().unwrap(); } -#[test] -fn cached_token_expiry() { - let now = Instant::now(); - let token = CachedToken::new("t", now + Duration::from_secs(60)).expect("valid token"); - assert!(token.is_valid(now), "token in the future must be valid"); +#[tokio::test] +async fn fetch_errors_for_service_account_key_source() { + let client = reqwest::Client::new(); + let err = token::fetch(&client, &TokenSource::ServiceAccountKey, "unused", "scope") + .await + .expect_err("key_file fetch is not implemented and must error, not hang or silently fail closed forever"); assert!( - !token.is_valid(now + Duration::from_secs(61)), - "token past expiry must be invalid" + err.to_string().contains("not implemented"), + "error must explain why, got: {err}" ); } // ----------------------------------------------------------------------------- -// on_request +// on_request: cache-through end to end // ----------------------------------------------------------------------------- #[tokio::test] -async fn on_request_injects_bearer_when_token_valid() { - let filter = GcpAdcFilter::for_test(Some(valid_token("test-token"))); +async fn on_request_injects_bearer_on_first_fetch() { + let (host, server) = mock_metadata_endpoint(r#"{"access_token":"fresh","expires_in":3600}"#); + let filter = + GcpAdcFilter::from_config(&yaml(&format!("source: metadata\nmetadata_host: {host}"))).expect("must construct"); let request = make_request(Method::POST, "/v1/models"); let mut ctx = make_filter_context(&request); let action = filter.on_request(&mut ctx).await.expect("must not error"); - assert!(matches!(action, FilterAction::Continue), "valid token must continue"); + assert!(matches!(action, FilterAction::Continue)); let auth = ctx .request_headers_to_set .iter() .find(|(name, _)| *name == header::AUTHORIZATION) .map(|(_, value)| value.to_str().expect("ascii")); - assert_eq!(auth, Some("Bearer test-token"), "must inject the cached bearer token"); + assert_eq!(auth, Some("Bearer fresh"), "must inject the freshly fetched bearer token"); + server.join().unwrap(); +} + +#[tokio::test] +async fn on_request_reuses_cached_token_without_a_second_fetch() { + // The mock endpoint accepts exactly one connection; a second + // on_request call must not attempt a second fetch, or it would fail + // to connect and 503 instead of continuing. + let (host, server) = mock_metadata_endpoint(r#"{"access_token":"once","expires_in":3600}"#); + let filter = + GcpAdcFilter::from_config(&yaml(&format!("source: metadata\nmetadata_host: {host}"))).expect("must construct"); + let request = make_request(Method::POST, "/v1/models"); + + let mut first_ctx = make_filter_context(&request); + let first = filter.on_request(&mut first_ctx).await.expect("first call must not error"); + assert!(matches!(first, FilterAction::Continue)); + + let mut second_ctx = make_filter_context(&request); + let second = filter.on_request(&mut second_ctx).await.expect("second call must not error"); + assert!( + matches!(second, FilterAction::Continue), + "a still-valid cache must serve the second request without a new connection" + ); + let auth = second_ctx + .request_headers_to_set + .iter() + .find(|(name, _)| *name == header::AUTHORIZATION) + .map(|(_, value)| value.to_str().expect("ascii")); + assert_eq!(auth, Some("Bearer once")); + server.join().unwrap(); } #[tokio::test] -async fn on_request_fails_closed_when_no_token() { - let filter = GcpAdcFilter::for_test(None); +async fn on_request_fails_closed_when_metadata_unreachable() { + let filter = GcpAdcFilter::from_config(&yaml("source: metadata\nmetadata_host: 127.0.0.1:1")) + .expect("must construct"); let request = make_request(Method::POST, "/v1/models"); let mut ctx = make_filter_context(&request); let action = filter.on_request(&mut ctx).await.expect("must reject, not error"); assert!( - matches!(action, FilterAction::Reject(rejection) if rejection.status == 503), - "no cached token must fail closed with 503" + matches!(action, FilterAction::Reject(r) if r.status == 503), + "a failed fetch must fail closed with 503" ); assert!( ctx.request_headers_to_set.is_empty(), @@ -326,23 +361,28 @@ async fn on_request_fails_closed_when_no_token() { } #[tokio::test] -async fn on_request_fails_closed_when_token_expired() { - let filter = GcpAdcFilter::for_test(Some( - CachedToken::new("stale", Instant::now() - Duration::from_secs(1)).expect("valid header value"), - )); +async fn on_request_fails_closed_for_key_file_source() { + let file = write_json(r#"{"type":"service_account","client_email":"sa@example.com"}"#); + let filter = GcpAdcFilter::from_config(&yaml(&format!( + "source: key_file\ncredentials_file: {}", + file.path().display() + ))) + .expect("must construct"); let request = make_request(Method::POST, "/v1/models"); let mut ctx = make_filter_context(&request); let action = filter.on_request(&mut ctx).await.expect("must reject, not error"); assert!( - matches!(action, FilterAction::Reject(rejection) if rejection.status == 503), - "expired token must fail closed with 503" + matches!(action, FilterAction::Reject(r) if r.status == 503), + "key_file token fetch is not implemented yet and must fail closed with 503" ); } #[tokio::test] async fn on_request_overwrites_client_authorization() { - let filter = GcpAdcFilter::for_test(Some(valid_token("gcp-token"))); + let (host, server) = mock_metadata_endpoint(r#"{"access_token":"gcp-token","expires_in":3600}"#); + let filter = + GcpAdcFilter::from_config(&yaml(&format!("source: metadata\nmetadata_host: {host}"))).expect("must construct"); let mut request = make_request(Method::POST, "/v1/models"); request .headers @@ -358,4 +398,5 @@ async fn on_request_overwrites_client_authorization() { .find(|(name, _)| *name == header::AUTHORIZATION) .map(|(_, value)| value.to_str().expect("ascii")); assert_eq!(auth, Some("Bearer gcp-token")); + server.join().unwrap(); } diff --git a/filters/src/gcp/token.rs b/filters/src/gcp/token.rs index a06fc3f462..6f5ab65dcc 100644 --- a/filters/src/gcp/token.rs +++ b/filters/src/gcp/token.rs @@ -1,13 +1,20 @@ // SPDX-License-Identifier: Apache-2.0 // Copyright (c) 2026 Praxis Contributors -//! Credential-source resolution for [`GcpAdcFilter`]. +//! Credential-source resolution and token fetch for [`GcpAdcFilter`]. //! -//! Token HTTP fetch (metadata server, `JWT` bearer) is added in follow-up -//! PRs. This module only classifies the source at construct time. +//! [`fetch`] acquires a token for the [`Metadata`](TokenSource::Metadata) +//! source from the GCE/GKE metadata server. The +//! [`ServiceAccountKey`](TokenSource::ServiceAccountKey) source (a parsed +//! `type: service_account` key file) is resolved and validated at +//! construct time but its token fetch is not implemented yet — it needs +//! `JWT` signing, which this workspace does not currently depend on — so +//! [`fetch`] returns a clear error for it rather than the cache silently +//! staying empty forever. -use std::path::Path; +use std::{path::Path, time::Duration}; +use http::HeaderValue; use praxis_filter::FilterError; use serde::Deserialize; @@ -17,7 +24,7 @@ use super::config::{GcpAdcConfig, GcpAdcSource}; // TokenSource // ----------------------------------------------------------------------------- -/// Resolved credential source used by the background refresher. +/// Resolved credential source used to fetch a token. #[derive(Clone, Debug, Eq, PartialEq)] pub(super) enum TokenSource { /// GCE/GKE/Cloud Run metadata server. @@ -27,10 +34,90 @@ pub(super) enum TokenSource { }, /// Parsed `type: service_account` key file. Fetch is not implemented - /// in this skeleton; the cache stays empty and requests fail closed. + /// yet (requires `JWT` signing); [`fetch`] returns an error for this + /// source so requests fail closed with a clear reason. ServiceAccountKey, } +// ----------------------------------------------------------------------------- +// Fetch +// ----------------------------------------------------------------------------- + +/// Response body from the GCE/GKE metadata server's token endpoint. +/// Extra fields (`token_type`, …) are ignored. +#[derive(Debug, Deserialize)] +struct MetadataTokenResponse { + /// The `OAuth2` access token. + access_token: String, + + /// Token lifetime in seconds. + expires_in: u64, +} + +/// Acquire a token for `source`. +/// +/// Kept free of caching concerns so it fits +/// [`TokenCache::get_or_refresh`](praxis_ai_apis::token_cache::TokenCache::get_or_refresh)'s +/// `fetch` closure shape directly. +/// +/// # Errors +/// +/// Returns [`FilterError`] if the metadata request fails, returns a +/// non-success status, or its body cannot be parsed; or, for +/// [`TokenSource::ServiceAccountKey`], always (not implemented). +pub(super) async fn fetch( + client: &reqwest::Client, + source: &TokenSource, + metadata_host: &str, + scope: &str, +) -> Result<(HeaderValue, Duration), FilterError> { + match source { + TokenSource::Metadata { service_account } => fetch_metadata_token(client, metadata_host, service_account, scope).await, + TokenSource::ServiceAccountKey => Err(FilterError::from( + "gcp_adc: token fetch for source key_file is not implemented yet (requires JWT signing); \ + use source: adc or source: metadata on a GCE/GKE instance instead", + )), + } +} + +/// Acquire a token from the GCE/GKE metadata server. +async fn fetch_metadata_token( + client: &reqwest::Client, + metadata_host: &str, + service_account: &str, + scope: &str, +) -> Result<(HeaderValue, Duration), FilterError> { + let mut url = format!( + "http://{metadata_host}/computeMetadata/v1/instance/service-accounts/{service_account}/token?scopes=" + ); + url::form_urlencoded::byte_serialize(scope.as_bytes()).for_each(|piece| url.push_str(piece)); + + let response = client + .get(url) + .header("Metadata-Flavor", "Google") + .send() + .await + .map_err(|e| FilterError::from(format!("gcp_adc: metadata token request failed: {e}")))?; + + let status = response.status(); + if !status.is_success() { + return Err(FilterError::from(format!( + "gcp_adc: metadata server returned HTTP status {status}" + ))); + } + + let token: MetadataTokenResponse = response + .json() + .await + .map_err(|e| FilterError::from(format!("gcp_adc: failed to parse metadata token response: {e}")))?; + + let mut authorization = HeaderValue::from_str(&format!("Bearer {}", token.access_token)) + .map_err(|e| FilterError::from(format!("gcp_adc: token is not a valid header value: {e}")))?; + authorization.set_sensitive(true); + + Ok((authorization, Duration::from_secs(token.expires_in))) +} + // ----------------------------------------------------------------------------- // GoogleApplicationCredentials // ----------------------------------------------------------------------------- diff --git a/tests/integration/tests/suite/examples/azure_ad.rs b/tests/integration/tests/suite/examples/azure_ad.rs index c32df8442d..73642d4948 100644 --- a/tests/integration/tests/suite/examples/azure_ad.rs +++ b/tests/integration/tests/suite/examples/azure_ad.rs @@ -11,12 +11,12 @@ //! sets for a test binary — so construction succeeds with no env //! mutation. //! -//! The token itself is acquired in the background from the configured -//! authority. This test points `authority_host` at a closed local port -//! so no real network call is made and no token is ever cached: the -//! cache stays empty and every request must fail closed with 503. The -//! full "token injected, reaches upstream" path is covered by the unit -//! tests in `filters/src/azure/azure_ad.rs`. +//! The token itself is acquired inline, on the request that finds the +//! cache stale (cache-through, not refresh-ahead). This test points +//! `authority_host` at a closed local port so the inline fetch fails +//! deterministically and no token is ever cached: every request must +//! fail closed with 503. The full "token injected, reaches upstream" +//! path is covered by the unit tests in `filters/src/azure/azure_ad.rs`. use std::collections::HashMap; diff --git a/tests/integration/tests/suite/examples/gcp_adc.rs b/tests/integration/tests/suite/examples/gcp_adc.rs index 4ef209cf8f..b77829f046 100644 --- a/tests/integration/tests/suite/examples/gcp_adc.rs +++ b/tests/integration/tests/suite/examples/gcp_adc.rs @@ -3,10 +3,13 @@ //! Tests for the GCP ADC example configuration. //! -//! Token fetch is not implemented in this skeleton, so the cache stays -//! empty and every request must fail closed with 503. The full "token -//! injected, reaches upstream" path is covered by unit tests in -//! `filters/src/gcp/`. +//! `GcpAdcFilter` acquires a token inline, on the request that finds the +//! cache stale (cache-through, not refresh-ahead). This test points +//! `metadata_host` at a closed local port so the inline fetch fails +//! deterministically (no real network call to `metadata.google.internal`) +//! and no token is ever cached: every request must fail closed with 503. +//! The full "token injected, reaches upstream" path is covered by the +//! unit tests in `filters/src/gcp/`. use std::collections::HashMap; @@ -29,11 +32,18 @@ fn gcp_adc_fails_closed_without_token() { let backend_guard = start_header_echo_backend(); let backend_port = backend_guard.port(); let proxy_port = free_port(); - let config = super::load_example_config( - "gcp-adc.yaml", - proxy_port, - HashMap::from([("127.0.0.1:3000", backend_port)]), + + let path = praxis_test_utils::example_config_path("gcp-adc.yaml"); + let yaml = std::fs::read_to_string(&path).unwrap_or_else(|e| panic!("read {path}: {e}")); + let patched = praxis_test_utils::patch_yaml(&yaml, proxy_port, &HashMap::from([("127.0.0.1:3000", backend_port)])); + // Point the metadata server at a closed port so the inline fetch + // fails deterministically, with no real network dependency. + let patched = patched.replace( + " - filter: gcp_adc", + " - filter: gcp_adc\n metadata_host: 127.0.0.1:1", ); + let config = + praxis_core::config::Config::from_yaml(&patched).unwrap_or_else(|e| panic!("parse gcp-adc.yaml: {e}")); let proxy = praxis_test_utils::start_proxy(&config); let raw = http_send( From bedcc8175b060eda9ddedad640bf47a2e17e3383 Mon Sep 17 00:00:00 2001 From: szedan Date: Tue, 1 Sep 2026 19:10:39 +0300 Subject: [PATCH 2/6] style: apply nightly rustfmt Fixes the lint CI failure on the cache-through credential caching change -- cargo clippy passed locally but nightly rustfmt was not run. Signed-off-by: szedan --- apis/src/token_cache.rs | 21 ++++++++++++--- filters/src/azure/azure_ad.rs | 26 ++++++++++++++++--- filters/src/gcp/tests.rs | 20 ++++++++++---- filters/src/gcp/token.rs | 9 ++++--- .../tests/suite/examples/gcp_adc.rs | 3 +-- 5 files changed, 60 insertions(+), 19 deletions(-) diff --git a/apis/src/token_cache.rs b/apis/src/token_cache.rs index 72805bdc61..9880f4d790 100644 --- a/apis/src/token_cache.rs +++ b/apis/src/token_cache.rs @@ -135,9 +135,18 @@ mod tests { #[test] fn effective_margin_caps_at_half_ttl() { use std::time::Duration; - assert_eq!(effective_margin(Duration::from_secs(3600), Duration::from_secs(30)), Duration::from_secs(30)); - assert_eq!(effective_margin(Duration::from_secs(40), Duration::from_secs(30)), Duration::from_secs(20)); - assert_eq!(effective_margin(Duration::from_secs(10), Duration::from_secs(30)), Duration::from_secs(5)); + assert_eq!( + effective_margin(Duration::from_secs(3600), Duration::from_secs(30)), + Duration::from_secs(30) + ); + assert_eq!( + effective_margin(Duration::from_secs(40), Duration::from_secs(30)), + Duration::from_secs(20) + ); + assert_eq!( + effective_margin(Duration::from_secs(10), Duration::from_secs(30)), + Duration::from_secs(5) + ); } #[tokio::test] @@ -171,7 +180,11 @@ mod tests { assert_eq!(first, 7); assert_eq!(second, 7); - assert_eq!(calls.load(Ordering::SeqCst), 1, "a still-valid cache must not trigger a second fetch"); + assert_eq!( + calls.load(Ordering::SeqCst), + 1, + "a still-valid cache must not trigger a second fetch" + ); } #[tokio::test] diff --git a/filters/src/azure/azure_ad.rs b/filters/src/azure/azure_ad.rs index 695259fed5..347ec8bc38 100644 --- a/filters/src/azure/azure_ad.rs +++ b/filters/src/azure/azure_ad.rs @@ -271,7 +271,15 @@ impl praxis_filter::HttpFilter for AzureAdFilter { ) -> Result { let fetched = self .cache - .get_or_refresh(|| fetch_token(&self.client, &self.token_url, &self.client_id, &self.client_secret, &self.scope)) + .get_or_refresh(|| { + fetch_token( + &self.client, + &self.token_url, + &self.client_id, + &self.client_secret, + &self.scope, + ) + }) .await; match fetched { Ok(authorization) => { @@ -607,7 +615,11 @@ mod tests { .iter() .find(|(name, _)| *name == http::header::AUTHORIZATION) .map(|(_, value)| value.to_str().expect("ascii")); - assert_eq!(auth, Some("Bearer fresh"), "must inject the freshly fetched bearer token"); + assert_eq!( + auth, + Some("Bearer fresh"), + "must inject the freshly fetched bearer token" + ); server.join().unwrap(); } @@ -621,11 +633,17 @@ mod tests { let request = make_request(Method::POST, "/openai/deployments/gpt-4o/chat/completions"); let mut first_ctx = make_filter_context(&request); - let first = filter.on_request(&mut first_ctx).await.expect("first call must not error"); + let first = filter + .on_request(&mut first_ctx) + .await + .expect("first call must not error"); assert!(matches!(first, FilterAction::Continue)); let mut second_ctx = make_filter_context(&request); - let second = filter.on_request(&mut second_ctx).await.expect("second call must not error"); + let second = filter + .on_request(&mut second_ctx) + .await + .expect("second call must not error"); assert!( matches!(second, FilterAction::Continue), "a still-valid cache must serve the second request without a new connection" diff --git a/filters/src/gcp/tests.rs b/filters/src/gcp/tests.rs index 2a9ea0addc..bb095c70b6 100644 --- a/filters/src/gcp/tests.rs +++ b/filters/src/gcp/tests.rs @@ -309,7 +309,11 @@ async fn on_request_injects_bearer_on_first_fetch() { .iter() .find(|(name, _)| *name == header::AUTHORIZATION) .map(|(_, value)| value.to_str().expect("ascii")); - assert_eq!(auth, Some("Bearer fresh"), "must inject the freshly fetched bearer token"); + assert_eq!( + auth, + Some("Bearer fresh"), + "must inject the freshly fetched bearer token" + ); server.join().unwrap(); } @@ -324,11 +328,17 @@ async fn on_request_reuses_cached_token_without_a_second_fetch() { let request = make_request(Method::POST, "/v1/models"); let mut first_ctx = make_filter_context(&request); - let first = filter.on_request(&mut first_ctx).await.expect("first call must not error"); + let first = filter + .on_request(&mut first_ctx) + .await + .expect("first call must not error"); assert!(matches!(first, FilterAction::Continue)); let mut second_ctx = make_filter_context(&request); - let second = filter.on_request(&mut second_ctx).await.expect("second call must not error"); + let second = filter + .on_request(&mut second_ctx) + .await + .expect("second call must not error"); assert!( matches!(second, FilterAction::Continue), "a still-valid cache must serve the second request without a new connection" @@ -344,8 +354,8 @@ async fn on_request_reuses_cached_token_without_a_second_fetch() { #[tokio::test] async fn on_request_fails_closed_when_metadata_unreachable() { - let filter = GcpAdcFilter::from_config(&yaml("source: metadata\nmetadata_host: 127.0.0.1:1")) - .expect("must construct"); + let filter = + GcpAdcFilter::from_config(&yaml("source: metadata\nmetadata_host: 127.0.0.1:1")).expect("must construct"); let request = make_request(Method::POST, "/v1/models"); let mut ctx = make_filter_context(&request); diff --git a/filters/src/gcp/token.rs b/filters/src/gcp/token.rs index 6f5ab65dcc..80f8fb055b 100644 --- a/filters/src/gcp/token.rs +++ b/filters/src/gcp/token.rs @@ -72,7 +72,9 @@ pub(super) async fn fetch( scope: &str, ) -> Result<(HeaderValue, Duration), FilterError> { match source { - TokenSource::Metadata { service_account } => fetch_metadata_token(client, metadata_host, service_account, scope).await, + TokenSource::Metadata { service_account } => { + fetch_metadata_token(client, metadata_host, service_account, scope).await + }, TokenSource::ServiceAccountKey => Err(FilterError::from( "gcp_adc: token fetch for source key_file is not implemented yet (requires JWT signing); \ use source: adc or source: metadata on a GCE/GKE instance instead", @@ -87,9 +89,8 @@ async fn fetch_metadata_token( service_account: &str, scope: &str, ) -> Result<(HeaderValue, Duration), FilterError> { - let mut url = format!( - "http://{metadata_host}/computeMetadata/v1/instance/service-accounts/{service_account}/token?scopes=" - ); + let mut url = + format!("http://{metadata_host}/computeMetadata/v1/instance/service-accounts/{service_account}/token?scopes="); url::form_urlencoded::byte_serialize(scope.as_bytes()).for_each(|piece| url.push_str(piece)); let response = client diff --git a/tests/integration/tests/suite/examples/gcp_adc.rs b/tests/integration/tests/suite/examples/gcp_adc.rs index b77829f046..42d878cf7d 100644 --- a/tests/integration/tests/suite/examples/gcp_adc.rs +++ b/tests/integration/tests/suite/examples/gcp_adc.rs @@ -42,8 +42,7 @@ fn gcp_adc_fails_closed_without_token() { " - filter: gcp_adc", " - filter: gcp_adc\n metadata_host: 127.0.0.1:1", ); - let config = - praxis_core::config::Config::from_yaml(&patched).unwrap_or_else(|e| panic!("parse gcp-adc.yaml: {e}")); + let config = praxis_core::config::Config::from_yaml(&patched).unwrap_or_else(|e| panic!("parse gcp-adc.yaml: {e}")); let proxy = praxis_test_utils::start_proxy(&config); let raw = http_send( From 2fbcd2ad3ce32e3dfc5b32c84a2281a897e577a2 Mon Sep 17 00:00:00 2001 From: szedan Date: Tue, 1 Sep 2026 19:22:48 +0300 Subject: [PATCH 3/6] docs(filters): regenerate azure_ad/gcp_adc docs and examples README Fixes the second lint CI failure -- xtask lint-filter-docs and sync-example-readme were not run locally after the cache-through change updated both filters' module docs and config surface. Signed-off-by: szedan --- docs/filters/azure_ad.md | 1 - docs/filters/gcp_adc.md | 9 ++++----- examples/README.md | 2 +- 3 files changed, 5 insertions(+), 7 deletions(-) diff --git a/docs/filters/azure_ad.md b/docs/filters/azure_ad.md index 126fdad677..3bc951376d 100644 --- a/docs/filters/azure_ad.md +++ b/docs/filters/azure_ad.md @@ -20,5 +20,4 @@ client_id: 11111111-1111-1111-1111-111111111111 scope: https://cognitiveservices.azure.com/.default client_secret_env_var: AZURE_CLIENT_SECRET authority_host: login.microsoftonline.com # optional, for sovereign clouds -refresh_ratio: 0.75 # optional, refresh at 75% of the usable lifetime ``` diff --git a/docs/filters/gcp_adc.md b/docs/filters/gcp_adc.md index f965862cb3..13f2f5c82d 100644 --- a/docs/filters/gcp_adc.md +++ b/docs/filters/gcp_adc.md @@ -9,15 +9,15 @@ Injects a GCP `OAuth2` access token into outbound requests. Experimental: requires the `gcp-adc-filter` cargo feature, which is off by default and activates the `experimental` marker. This filter is a work in progress and its configuration surface may change between releases. -**Token acquisition is not implemented yet.** This filter currently establishes the configuration surface, credential-source resolution, and the fail-closed request path: the token cache is never populated, so every request is rejected with `503`. Fetch (metadata server, key file) arrives with the shared background-refresh primitive tracked in praxis#555, praxis#1042, and praxis#1043 — filters must not spawn their own refresher threads. +Acquires a token via Application Default Credentials (GKE metadata server) and injects `Authorization: Bearer ` on every proxied request, keeping GCP credentials invisible to the downstream client. There is no background refresh thread: caching is cache-through, the same as [`crate::azure::azure_ad`] — see [`praxis_ai_apis::token_cache::TokenCache`] for the exact contract. -Once implemented, the filter will acquire a token via Application Default Credentials (GKE metadata or a service-account key file) and inject `Authorization: Bearer ` on every proxied request, keeping GCP credentials invisible to the downstream client. +**Service-account key file (`source: key_file`) token fetch is not implemented yet** — it needs `JWT` signing, which this workspace does not currently depend on. Config parsing, file resolution, and validation for `key_file` all work; `on_request` fails closed with a clear "not implemented" reason instead of silently 503ing forever. Credential-source resolution happens at construct time: `GOOGLE_APPLICATION_CREDENTIALS` is read once when the pipeline is built (reload the config to pick up changes), and a `gcloud` user credential file (`authorized_user`) is rejected as a configuration error rather than silently falling through the ADC chain. This filter only injects `Authorization`. Pointing the request at the correct Vertex endpoint (cluster `endpoints` + `tls.sni`) is the operator's responsibility. -Until a token is cached, and whenever the cached token is missing or expired, requests are rejected with `503` rather than forwarded unauthenticated. +Whenever no valid token can be produced — none cached and the inline fetch fails — the request is rejected with `503` rather than forwarded unauthenticated. ## Configuration @@ -27,7 +27,7 @@ Until a token is cached, and whenever the cached token is missing or expired, re | `scope` | string | no | `OAuth2` scope requested with the access token. | | `service_account` | string | no | Metadata service account email, or `default` (the default). Only used with the `metadata` and `adc` sources; rejected for `key_file`. Interpolated into the metadata path, so it must be `default` or a service-account email (letters, digits, `@`, `.`, `-`, `_`). | | `credentials_file` | string | no | Path to a service-account key JSON file. Required when `source` is `key_file`; rejected for the other sources (`adc` reads `GOOGLE_APPLICATION_CREDENTIALS` instead). | -| `refresh_ratio` | number | no | Fraction of a token's TTL at which to refresh it. Must be in the open interval `(0, 1)`. | +| `metadata_host` | string | no | GCE/GKE metadata server host. Override for testing (point at a local mock) or a non-default metadata server; production deployments should not need to set this. | ## Example @@ -35,5 +35,4 @@ Until a token is cached, and whenever the cached token is missing or expired, re filter: gcp_adc source: adc scope: https://www.googleapis.com/auth/cloud-platform -refresh_ratio: 0.75 ``` diff --git a/examples/README.md b/examples/README.md index 6849626ef0..d414c76053 100644 --- a/examples/README.md +++ b/examples/README.md @@ -26,7 +26,7 @@ before sending requests. | [aws-sigv4.yaml](configs/aws-sigv4.yaml) | Signs outbound requests to an AWS service (Bedrock, in this example) using Signature Version 4. Credentials are static, sourced from environment variables — see the module docs on Sigv4SignFilter for the planned OIDC/default-credential-chain follow-up | | [azure-ad.yaml](configs/azure-ad.yaml) | Acquires an Entra ID bearer token via the client-credentials grant and injects "Authorization: Bearer " on every proxied request to Azure OpenAI | | [credential-injection.yaml](configs/credential-injection.yaml) | Injects per-cluster API credentials into upstream requests and strips client-provided credentials to prevent forwarding | -| [gcp-adc.yaml](configs/gcp-adc.yaml) | Establishes the gcp_adc filter's configuration surface and fail-closed behavior | +| [gcp-adc.yaml](configs/gcp-adc.yaml) | Acquires an OAuth2 access token from the GCE/GKE metadata server (source: adc or metadata) and injects "Authorization: Bearer " on every proxied request to Vertex AI | | [intelligent-route-all-capabilities.yaml](configs/intelligent-route-all-capabilities.yaml) | Demonstrates every candidate capability and selection input handled by intelligent_route today | | [intelligent-route-inference.yaml](configs/intelligent-route-inference.yaml) | Routes requests to different upstream clusters based on the inference model name extracted from a configured request header. The header value is set by an earlier filter such as `json_body_field` | | [intelligent-route-mcp.yaml](configs/intelligent-route-mcp.yaml) | Routes MCP `tools/call` requests to the cluster that owns the requested tool, using the `mcp.name` metadata set by the `mcp` filter | From 1bcd738157572bd8b5733d89b17ff00ab988a789 Mon Sep 17 00:00:00 2001 From: szedan Date: Tue, 1 Sep 2026 19:58:26 +0300 Subject: [PATCH 4/6] review: address feedback from Alex and Aslak on #861 - effective_margin: drop unneeded pub(crate) visibility, it's only used within this file - replace the two explicit drop(guard) calls with a single #[expect(clippy::significant_drop_tightening)] on get_or_refresh -- the guard's last use is immediately before each return either way, so the manual drops were restating that rather than shortening the actual hold time - log a warning when a fetched token's TTL is at or below the cache's margin, so an unusually short TTL from the token endpoint is visible instead of silently reducing the validity window - fix a second stale "background refresher" comment in the azure_ad integration test that the doc-regeneration commit missed The write-lock-held-during-fetch latency pile-up under a sustained IdP network partition, also raised in review, is filed as a follow-up: ai#864. Signed-off-by: szedan --- apis/src/token_cache.rs | 17 ++++++++++++++--- .../tests/suite/examples/azure_ad.rs | 4 ++-- 2 files changed, 16 insertions(+), 5 deletions(-) diff --git a/apis/src/token_cache.rs b/apis/src/token_cache.rs index 9880f4d790..0104256216 100644 --- a/apis/src/token_cache.rs +++ b/apis/src/token_cache.rs @@ -21,6 +21,7 @@ use std::time::{Duration, Instant}; use tokio::sync::RwLock; +use tracing::warn; // ----------------------------------------------------------------------------- // Margin @@ -29,7 +30,7 @@ use tokio::sync::RwLock; /// Safety margin to subtract from a TTL before treating a cached value /// as expired, capped at half the TTL so a short-lived credential stays /// usable for part of its life instead of being cached already-expired. -pub(crate) fn effective_margin(ttl: Duration, margin: Duration) -> Duration { +fn effective_margin(ttl: Duration, margin: Duration) -> Duration { margin.min(ttl / 2) } @@ -89,6 +90,11 @@ impl TokenCache { /// /// Returns whatever `fetch` returns on failure. Nothing is cached /// in that case, so the next call tries again. + #[expect( + clippy::significant_drop_tightening, + reason = "guard's last use is immediately before each return; an explicit drop() would just \ + restate that, not shorten the actual hold time" + )] pub async fn get_or_refresh(&self, fetch: F) -> Result where F: FnOnce() -> Fut + Send, @@ -105,17 +111,22 @@ impl TokenCache { let mut guard = self.cache.write().await; let fresh = valid_cached_value(guard.as_ref()); if let Some(value) = fresh { - drop(guard); return Ok(value); } let (value, ttl) = fetch().await?; + if ttl <= self.margin { + warn!( + ttl_secs = ttl.as_secs(), + margin_secs = self.margin.as_secs(), + "token TTL is unusually short; validity margin reduced" + ); + } let expires_at = Instant::now() + ttl.saturating_sub(effective_margin(ttl, self.margin)); *guard = Some(Entry { value: value.clone(), expires_at, }); - drop(guard); Ok(value) } } diff --git a/tests/integration/tests/suite/examples/azure_ad.rs b/tests/integration/tests/suite/examples/azure_ad.rs index 73642d4948..09f410d684 100644 --- a/tests/integration/tests/suite/examples/azure_ad.rs +++ b/tests/integration/tests/suite/examples/azure_ad.rs @@ -44,8 +44,8 @@ fn azure_ad_fails_closed_without_token() { let yaml = std::fs::read_to_string(&path).unwrap_or_else(|e| panic!("read {path}: {e}")); let patched = praxis_test_utils::patch_yaml(&yaml, proxy_port, &HashMap::from([("127.0.0.1:3000", backend_port)])); // Use an always-set env var so filter construction succeeds, and - // point the authority at a closed port so the background refresher - // can never acquire a token (deterministic fail-closed). + // point the authority at a closed port so the inline cache-through + // fetch can never acquire a token (deterministic fail-closed). let patched = patched.replace( " client_secret_env_var: AZURE_CLIENT_SECRET", " client_secret_env_var: CARGO_PKG_NAME\n authority_host: 127.0.0.1:1", From df3de85428a7fe8790286df9c258add479bfbe83 Mon Sep 17 00:00:00 2001 From: szedan Date: Tue, 1 Sep 2026 20:23:47 +0300 Subject: [PATCH 5/6] fix(gcp_adc): constrain metadata_host to the metadata server or loopback Addresses CodeQL alert #8 (rust/cleartext-transmission) on #861. The metadata endpoint is only safe to reach over plain HTTP because it is link-local and never routable off the VM/host -- that part of the finding doesn't apply. But metadata_host is operator-configurable, and nothing stopped it from being pointed anywhere: a misconfiguration would send the same plaintext request, and receive the access token in the response, over a real network path instead of staying host-local. validate_metadata_host now rejects anything other than metadata.google.internal or a loopback address (127.0.0.1/localhost, which tests already use to point at a local mock server). Signed-off-by: szedan --- filters/src/gcp/config.rs | 23 +++++++++++++++++++++++ filters/src/gcp/tests.rs | 25 +++++++++++++++++++++++++ 2 files changed, 48 insertions(+) diff --git a/filters/src/gcp/config.rs b/filters/src/gcp/config.rs index 8897a013ee..c5d03749b7 100644 --- a/filters/src/gcp/config.rs +++ b/filters/src/gcp/config.rs @@ -102,6 +102,7 @@ pub(super) fn validate_config(config: &GcpAdcConfig) -> Result<(), FilterError> return Err("gcp_adc: scope must not be empty".into()); } validate_url_component("metadata_host", &config.metadata_host)?; + validate_metadata_host(&config.metadata_host)?; match config.source { GcpAdcSource::KeyFile => validate_key_file_fields(config), GcpAdcSource::Metadata | GcpAdcSource::Adc => validate_metadata_fields(config), @@ -127,6 +128,28 @@ pub(super) fn validate_url_component(field: &str, value: &str) -> Result<(), Fil Ok(()) } +/// Reject a `metadata_host` that isn't the real GCE/GKE metadata server or +/// a loopback address (used only to point tests at a local mock). +/// +/// The metadata endpoint is safe to reach over plain HTTP specifically +/// because it is link-local and never routable off the VM/host. Any other +/// host configured here would send the same plaintext request -- and +/// receive the access token in the response -- over a real network path. +fn validate_metadata_host(value: &str) -> Result<(), FilterError> { + let host = value.split(':').next().unwrap_or(value); + let is_safe = value == "metadata.google.internal" || host == "127.0.0.1" || host == "localhost"; + if !is_safe { + return Err(format!( + "gcp_adc: metadata_host '{value}' must be 'metadata.google.internal' or a loopback \ + address (127.0.0.1/localhost, for tests) -- the metadata endpoint is only safe over \ + plain HTTP because it never leaves the VM; anything else would send the access \ + token over a real network in cleartext" + ) + .into()); + } + Ok(()) +} + /// `key_file` requires `credentials_file` and does not use /// `service_account`. fn validate_key_file_fields(config: &GcpAdcConfig) -> Result<(), FilterError> { diff --git a/filters/src/gcp/tests.rs b/filters/src/gcp/tests.rs index bb095c70b6..f57021b8b7 100644 --- a/filters/src/gcp/tests.rs +++ b/filters/src/gcp/tests.rs @@ -156,6 +156,31 @@ fn rejects_structural_characters_in_metadata_host() { ); } +#[test] +fn rejects_non_loopback_non_default_metadata_host() { + // Structurally valid hostname, but not the real metadata server or a + // loopback test address -- must still be rejected, or a misconfigured + // metadata_host would send the access token over a real network in + // cleartext. + let err = GcpAdcFilter::from_config(&yaml("metadata_host: evil.example.com")) + .err() + .expect("non-loopback, non-default metadata_host must be rejected"); + assert!( + err.to_string().contains("metadata_host"), + "error should name metadata_host: {err}" + ); +} + +#[test] +fn accepts_loopback_and_default_metadata_host() { + GcpAdcFilter::from_config(&yaml("source: metadata\nmetadata_host: metadata.google.internal")) + .expect("the real metadata server must be accepted"); + GcpAdcFilter::from_config(&yaml("source: metadata\nmetadata_host: 127.0.0.1:9000")) + .expect("a loopback address must be accepted for tests"); + GcpAdcFilter::from_config(&yaml("source: metadata\nmetadata_host: localhost:9000")) + .expect("localhost must be accepted for tests"); +} + #[test] fn validate_service_account_accepts_email_and_default() { validate_service_account("default").expect("default is valid"); From 544326c6dfb88122cb1a2d23538d4296dfcf998c Mon Sep 17 00:00:00 2001 From: szedan Date: Tue, 1 Sep 2026 22:33:24 +0300 Subject: [PATCH 6/6] review: remove localhost from metadata_host allowlist, regen docs Addresses Alex's two review comments on the metadata_host restriction in #861: - localhost is a hostname resolved via DNS/etc/hosts, not a fixed address like 127.0.0.1 -- it could be remapped to point anywhere, which would defeat the loopback restriction entirely. Only 127.0.0.1 (and the real metadata.google.internal) are accepted now. - metadata_host's doc comment still described the old, broader behavior ("a non-default metadata server"); updated it to match the actual restriction and regenerated docs/filters/gcp_adc.md. Signed-off-by: szedan --- docs/filters/gcp_adc.md | 2 +- filters/src/gcp/config.rs | 20 ++++++++++++++------ filters/src/gcp/tests.rs | 18 +++++++++++++++--- 3 files changed, 30 insertions(+), 10 deletions(-) diff --git a/docs/filters/gcp_adc.md b/docs/filters/gcp_adc.md index 13f2f5c82d..3745dee50c 100644 --- a/docs/filters/gcp_adc.md +++ b/docs/filters/gcp_adc.md @@ -27,7 +27,7 @@ Whenever no valid token can be produced — none cached and the inline fetch fai | `scope` | string | no | `OAuth2` scope requested with the access token. | | `service_account` | string | no | Metadata service account email, or `default` (the default). Only used with the `metadata` and `adc` sources; rejected for `key_file`. Interpolated into the metadata path, so it must be `default` or a service-account email (letters, digits, `@`, `.`, `-`, `_`). | | `credentials_file` | string | no | Path to a service-account key JSON file. Required when `source` is `key_file`; rejected for the other sources (`adc` reads `GOOGLE_APPLICATION_CREDENTIALS` instead). | -| `metadata_host` | string | no | GCE/GKE metadata server host. Override for testing (point at a local mock) or a non-default metadata server; production deployments should not need to set this. | +| `metadata_host` | string | no | GCE/GKE metadata server host. Defaults to the real metadata server; the only other accepted value is a `127.0.0.1` loopback address, to point tests at a local mock. The metadata endpoint is only safe to reach over plain HTTP because it never leaves the VM/host, so nothing else is accepted (not even `localhost`, which is a resolvable hostname rather than a fixed address). | ## Example diff --git a/filters/src/gcp/config.rs b/filters/src/gcp/config.rs index c5d03749b7..ba5fb7bb90 100644 --- a/filters/src/gcp/config.rs +++ b/filters/src/gcp/config.rs @@ -65,9 +65,12 @@ pub(super) struct GcpAdcConfig { #[serde(default)] pub credentials_file: Option, - /// GCE/GKE metadata server host. Override for testing (point at a - /// local mock) or a non-default metadata server; production - /// deployments should not need to set this. + /// GCE/GKE metadata server host. Defaults to the real metadata + /// server; the only other accepted value is a `127.0.0.1` loopback + /// address, to point tests at a local mock. The metadata endpoint is + /// only safe to reach over plain HTTP because it never leaves the + /// VM/host, so nothing else is accepted (not even `localhost`, which + /// is a resolvable hostname rather than a fixed address). #[serde(default = "default_metadata_host")] pub metadata_host: String, } @@ -129,7 +132,12 @@ pub(super) fn validate_url_component(field: &str, value: &str) -> Result<(), Fil } /// Reject a `metadata_host` that isn't the real GCE/GKE metadata server or -/// a loopback address (used only to point tests at a local mock). +/// the `127.0.0.1` loopback address (used only to point tests at a local +/// mock). +/// +/// `localhost` is deliberately not accepted: unlike a literal loopback +/// IP, it is a hostname resolved via DNS/`/etc/hosts` and could be +/// remapped to point anywhere, which would defeat this check entirely. /// /// The metadata endpoint is safe to reach over plain HTTP specifically /// because it is link-local and never routable off the VM/host. Any other @@ -137,11 +145,11 @@ pub(super) fn validate_url_component(field: &str, value: &str) -> Result<(), Fil /// receive the access token in the response -- over a real network path. fn validate_metadata_host(value: &str) -> Result<(), FilterError> { let host = value.split(':').next().unwrap_or(value); - let is_safe = value == "metadata.google.internal" || host == "127.0.0.1" || host == "localhost"; + let is_safe = value == "metadata.google.internal" || host == "127.0.0.1"; if !is_safe { return Err(format!( "gcp_adc: metadata_host '{value}' must be 'metadata.google.internal' or a loopback \ - address (127.0.0.1/localhost, for tests) -- the metadata endpoint is only safe over \ + IP address (127.0.0.1, for tests) -- the metadata endpoint is only safe over \ plain HTTP because it never leaves the VM; anything else would send the access \ token over a real network in cleartext" ) diff --git a/filters/src/gcp/tests.rs b/filters/src/gcp/tests.rs index f57021b8b7..c1c8c79424 100644 --- a/filters/src/gcp/tests.rs +++ b/filters/src/gcp/tests.rs @@ -176,9 +176,21 @@ fn accepts_loopback_and_default_metadata_host() { GcpAdcFilter::from_config(&yaml("source: metadata\nmetadata_host: metadata.google.internal")) .expect("the real metadata server must be accepted"); GcpAdcFilter::from_config(&yaml("source: metadata\nmetadata_host: 127.0.0.1:9000")) - .expect("a loopback address must be accepted for tests"); - GcpAdcFilter::from_config(&yaml("source: metadata\nmetadata_host: localhost:9000")) - .expect("localhost must be accepted for tests"); + .expect("a loopback IP address must be accepted for tests"); +} + +#[test] +fn rejects_localhost_metadata_host() { + // Unlike a literal loopback IP, `localhost` is a hostname resolved + // via DNS/`/etc/hosts` and could be remapped to point anywhere -- + // accepting it would defeat the loopback restriction entirely. + let err = GcpAdcFilter::from_config(&yaml("metadata_host: localhost:9000")) + .err() + .expect("localhost must be rejected, it is not a fixed address"); + assert!( + err.to_string().contains("metadata_host"), + "error should name metadata_host: {err}" + ); } #[test]