From 76c0d970a18bf28ddc48dd746570e73486542606 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Mon, 8 Jun 2026 17:52:23 +0200 Subject: [PATCH] fix(oauth): persist refreshed token through configured secret backend (#9471) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The lazy on-fetch OAuth token refresh persisted the new access token with a raw `UPDATE variable SET value = `, bypassing the secret-backend abstraction. With an external secret backend (AWS Secrets Manager / Azure Key Vault / Vault), secret reads resolve through the backend and ignore `variable.value` entirely, so refresh advanced `account.expires_at` and updated Postgres but never wrote the new token to the external store. Every read that did not itself trigger a mint kept serving the frozen connect-time token, which expired ~1h after connect (RefreshError on Google clients). `windmill-oauth` can't depend on `windmill-store` (circular), so variable persistence moves out of `refresh_token{,_for_account}` (which now only exchange the token + update the `account` row and return the new token) into the `windmill-store` callers, via a new `store_oauth_token_value` helper that writes through the configured backend and stores the returned value (encrypted blob for the DB backend, `$...:` marker for external backends) in `variable.value`. If persisting the refreshed token fails (more likely now that it can be a network write to an external backend) after the account was committed fresh, `store_oauth_token_value` resets `expires_at` to the past and records `refresh_error` — looking the account up via `variable.account` — so the next fetch retries instead of serving the stale token for the whole token lifetime. Also add `windmill-store/tests/oauth_refresh_secret_backend.rs`, an opt-in e2e regression suite (RUN_SECRET_BACKEND_E2E / RUN_AWS_SM_TESTS) covering database and external (AWS SM via LocalStack) backends plus the self-healing reset. Verified against Postgres + LocalStack: 3 passed. EE companion (oauth_refresh_ee.rs: 3 refresh paths) merged via #607; this OSS half completes the fix (ee-repo-ref already at EE main 481ea7f). Co-authored-by: Claude Opus 4.8 (1M context) --- backend/windmill-oauth/src/lib.rs | 27 +- backend/windmill-store/src/lib.rs | 3 + .../windmill-store/src/oauth_refresh_oss.rs | 12 +- .../src/oauth_refresh_secret_backend_tests.rs | 337 ++++++++++++++++++ .../windmill-store/src/secret_backend_ext.rs | 107 +++++- 5 files changed, 459 insertions(+), 27 deletions(-) create mode 100644 backend/windmill-store/src/oauth_refresh_secret_backend_tests.rs diff --git a/backend/windmill-oauth/src/lib.rs b/backend/windmill-oauth/src/lib.rs index 59b4184cea..252f7ace28 100644 --- a/backend/windmill-oauth/src/lib.rs +++ b/backend/windmill-oauth/src/lib.rs @@ -26,7 +26,6 @@ use windmill_common::error::{self, to_anyhow, Error}; use windmill_common::more_serde::maybe_number_opt; use windmill_common::oauth2::*; use windmill_common::utils::now_from_db; -use windmill_common::variables::{build_crypt, encrypt}; use windmill_common::BASE_URL; pub type DB = sqlx::Pool; @@ -525,11 +524,14 @@ pub struct OAuthAccountInfo { pub scopes: Option>, } -/// Refresh an OAuth token and update the database. +/// Refresh an OAuth token and update the `account` row. /// Fetches the account from DB, then delegates to `refresh_token_for_account`. +/// +/// Returns the freshly minted access token. Persisting it to the secret variable +/// backing the resource is the caller's responsibility (it must route through the +/// configured secret backend — see `store_oauth_token_value` in `windmill-store`). pub async fn refresh_token<'c>( mut tx: Transaction<'c, Postgres>, - path: &str, w_id: &str, id: i32, db: &DB, @@ -549,7 +551,6 @@ pub async fn refresh_token<'c>( refresh_token_for_account( tx, - path, w_id, id, db, @@ -562,9 +563,14 @@ pub async fn refresh_token<'c>( } /// Refresh an OAuth token given pre-fetched account info (no additional SELECT). +/// +/// Exchanges the refresh token, updates the `account` row (`refresh_token`, +/// `expires_at`, `refresh_error`) and returns the new access token. It does NOT +/// persist the token to the secret variable — the caller must do that through the +/// configured secret backend (`store_oauth_token_value`), otherwise an external +/// secret backend would keep serving the stale connect-time token. pub async fn refresh_token_for_account<'c>( mut tx: Transaction<'c, Postgres>, - path: &str, w_id: &str, id: i32, db: &DB, @@ -676,17 +682,6 @@ pub async fn refresh_token_for_account<'c>( tx.commit().await?; let token_str = token.access_token.to_string(); - let mc = build_crypt(db, w_id).await?; - let encrypted_token = encrypt(&mc, token_str.as_str()); - - sqlx::query!( - "UPDATE variable SET value = $1 WHERE workspace_id = $2 AND path = $3", - encrypted_token, - w_id, - path - ) - .execute(db) - .await?; tracing::info!( grant_type = %account.grant_type, diff --git a/backend/windmill-store/src/lib.rs b/backend/windmill-store/src/lib.rs index c7e606de43..b14466c79b 100644 --- a/backend/windmill-store/src/lib.rs +++ b/backend/windmill-store/src/lib.rs @@ -14,3 +14,6 @@ pub mod resources; pub mod secret_backend_ext; pub mod var_resource_cache; pub mod variables; + +#[cfg(all(test, feature = "oauth2", feature = "private", feature = "enterprise"))] +mod oauth_refresh_secret_backend_tests; diff --git a/backend/windmill-store/src/oauth_refresh_oss.rs b/backend/windmill-store/src/oauth_refresh_oss.rs index e8edb19ccd..13e24dfd8b 100644 --- a/backend/windmill-store/src/oauth_refresh_oss.rs +++ b/backend/windmill-store/src/oauth_refresh_oss.rs @@ -24,9 +24,8 @@ pub async fn _refresh_token<'c>( id: i32, db: &DB, ) -> error::Result { - windmill_oauth::refresh_token( + let token = windmill_oauth::refresh_token( tx, - path, w_id, id, db, @@ -34,5 +33,12 @@ pub async fn _refresh_token<'c>( &windmill_oauth::OAUTH_HTTP_CLIENT, include_str!("../../oauth_connect.json"), ) - .await + .await?; + + // Persist the refreshed token through the configured secret backend so an + // external backend (Vault / Azure KV / AWS Secrets Manager) is updated too, + // not just the in-DB variable mirror. + crate::secret_backend_ext::store_oauth_token_value(db, w_id, path, &token).await?; + + Ok(token) } diff --git a/backend/windmill-store/src/oauth_refresh_secret_backend_tests.rs b/backend/windmill-store/src/oauth_refresh_secret_backend_tests.rs new file mode 100644 index 0000000000..5a11dea03e --- /dev/null +++ b/backend/windmill-store/src/oauth_refresh_secret_backend_tests.rs @@ -0,0 +1,337 @@ +//! E2E regression tests for OAuth token refresh persistence through the +//! configured secret backend. +//! +//! Regression for windmill#9471 / windmill-ee-private#607: the lazy on-fetch +//! OAuth refresh used to persist the freshly minted token with a raw +//! `UPDATE variable SET value = `, bypassing the secret-backend +//! abstraction. With an external backend (AWS Secrets Manager / Azure Key +//! Vault / Vault) reads resolve through the backend and ignore `variable.value` +//! entirely, so the external store stayed frozen at its connect-time token and +//! every read that did not itself trigger a mint served a stale/expired token. +//! +//! These tests exercise the persistence step (`store_oauth_token_value`) — the +//! exact code path that was fixed — against both the database backend and an +//! external (AWS Secrets Manager via LocalStack) backend, plus the self-healing +//! reset on a failed persist. They are opt-in (they mutate the shared +//! `global_settings.secret_backend` row and workspace/variable/account rows on a +//! real DB) and skip unless `RUN_SECRET_BACKEND_E2E=1` is set. +//! +//! ## Run +//! +//! Database-backend case (needs a migrated DB). Note `RUN_SECRET_BACKEND_E2E=1` +//! is required or every test skips: +//! +//! ```bash +//! RUN_SECRET_BACKEND_E2E=1 \ +//! DATABASE_URL=postgres://postgres:changeme@127.0.0.1:5432/windmill \ +//! cargo test -p windmill-store --features private,enterprise,oauth2 \ +//! oauth_refresh_secret_backend_tests -- --nocapture --test-threads=1 +//! ``` +//! +//! External-backend cases additionally need LocalStack `secretsmanager` and +//! `RUN_AWS_SM_TESTS=1`: +//! +//! ```bash +//! docker run -d -e SERVICES=secretsmanager localstack/localstack:3.8 +//! RUN_SECRET_BACKEND_E2E=1 RUN_AWS_SM_TESTS=1 AWS_SM_ENDPOINT=http://:4566 \ +//! DATABASE_URL=postgres://postgres:changeme@127.0.0.1:5432/windmill \ +//! cargo test -p windmill-store --features private,enterprise,oauth2 \ +//! oauth_refresh_secret_backend_tests -- --nocapture --test-threads=1 +//! ``` + +use crate::secret_backend_ext::{get_secret_value, store_oauth_token_value, store_secret_value}; +use sqlx::postgres::PgPoolOptions; +use sqlx::{Pool, Postgres}; + +// global_settings holds a single `secret_backend` row shared across tests; +// serialize the test bodies so concurrent runs don't clobber each other's +// configured backend. (Also run with --test-threads=1 for good measure.) +static SERIAL: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + +fn env_flag(name: &str) -> bool { + std::env::var(name) + .map(|v| v == "1" || v.eq_ignore_ascii_case("true")) + .unwrap_or(false) +} + +// Opt-in gate so the suite never runs (and mutates shared DB state) as part of a +// normal `cargo test` invocation. +fn run_e2e() -> bool { + env_flag("RUN_SECRET_BACKEND_E2E") +} + +fn run_aws_sm() -> bool { + env_flag("RUN_AWS_SM_TESTS") +} + +/// Restore the default (database) backend so we don't leave the instance +/// pointed at a test backend for any concurrently-running suite. +async fn reset_backend(db: &Pool) { + set_backend(db, serde_json::json!({ "type": "Database" })).await; +} + +fn aws_sm_endpoint() -> String { + std::env::var("AWS_SM_ENDPOINT").unwrap_or_else(|_| "http://localhost:4566".to_string()) +} + +async fn db() -> Pool { + let url = std::env::var("DATABASE_URL") + .expect("DATABASE_URL must point at a migrated windmill database"); + PgPoolOptions::new() + .max_connections(5) + .connect(&url) + .await + .expect("connect to DATABASE_URL") +} + +/// Fresh workspace + key + clean variable/account rows for `w_id`. +async fn setup_workspace(db: &Pool, w_id: &str) { + sqlx::query("DELETE FROM variable WHERE workspace_id = $1") + .bind(w_id) + .execute(db) + .await + .unwrap(); + sqlx::query("DELETE FROM account WHERE workspace_id = $1") + .bind(w_id) + .execute(db) + .await + .unwrap(); + sqlx::query("DELETE FROM workspace_key WHERE workspace_id = $1") + .bind(w_id) + .execute(db) + .await + .unwrap(); + sqlx::query("DELETE FROM workspace WHERE id = $1") + .bind(w_id) + .execute(db) + .await + .unwrap(); + + sqlx::query("INSERT INTO workspace (id, name, owner) VALUES ($1, $1, 'admin@windmill.dev')") + .bind(w_id) + .execute(db) + .await + .unwrap(); + sqlx::query( + "INSERT INTO workspace_key (workspace_id, kind, key) VALUES ($1, 'cloud', 'e2ekey')", + ) + .bind(w_id) + .execute(db) + .await + .unwrap(); +} + +async fn set_backend(db: &Pool, config: serde_json::Value) { + sqlx::query( + "INSERT INTO global_settings (name, value) VALUES ('secret_backend', $1) \ + ON CONFLICT (name) DO UPDATE SET value = EXCLUDED.value", + ) + .bind(config) + .execute(db) + .await + .unwrap(); +} + +fn aws_sm_config(endpoint: &str) -> serde_json::Value { + serde_json::json!({ + "type": "AwsSecretsManager", + "region": "us-east-1", + "access_key_id": "test", + "secret_access_key": "test", + "endpoint_url": endpoint, + "prefix": "windmill-e2e/" + }) +} + +/// Simulate `Connect`: store the initial token through the backend and create +/// the linked secret variable + account (expired, with a refresh token). +async fn simulate_connect(db: &Pool, w_id: &str, path: &str, initial_token: &str) -> i32 { + let stored = store_secret_value(db, w_id, path, initial_token) + .await + .expect("store initial token"); + + let account_id: i32 = sqlx::query_scalar( + "INSERT INTO account (workspace_id, expires_at, refresh_token, client, grant_type) \ + VALUES ($1, now() - interval '1 hour', 'rt_dummy', 'gdrive', 'authorization_code') \ + RETURNING id", + ) + .bind(w_id) + .fetch_one(db) + .await + .unwrap(); + + sqlx::query( + "INSERT INTO variable (workspace_id, path, value, is_secret, is_oauth, account, expires_at) \ + VALUES ($1, $2, $3, true, true, $4, now() - interval '1 hour')", + ) + .bind(w_id) + .bind(path) + .bind(&stored) + .bind(account_id) + .execute(db) + .await + .unwrap(); + + account_id +} + +async fn variable_value(db: &Pool, w_id: &str, path: &str) -> String { + sqlx::query_scalar("SELECT value FROM variable WHERE workspace_id = $1 AND path = $2") + .bind(w_id) + .bind(path) + .fetch_one(db) + .await + .unwrap() +} + +async fn account_expires_in_past(db: &Pool, w_id: &str, account_id: i32) -> bool { + sqlx::query_scalar("SELECT expires_at < now() FROM account WHERE workspace_id = $1 AND id = $2") + .bind(w_id) + .bind(account_id) + .fetch_one(db) + .await + .unwrap() +} + +/// Database backend (the "without external storage" case): refresh must +/// re-encrypt the new token into `variable.value`; reads serve the new token. +#[tokio::test] +async fn database_backend_persists_refreshed_token() { + if !run_e2e() { + println!( + "Skipping database_backend_persists_refreshed_token: set RUN_SECRET_BACKEND_E2E=1" + ); + return; + } + let _guard = SERIAL.lock().await; + let db = db().await; + let w_id = "wm_e2e_db"; + let path = "f/google/gdrive"; + + set_backend(&db, serde_json::json!({ "type": "Database" })).await; + setup_workspace(&db, w_id).await; + let _ = simulate_connect(&db, w_id, path, "OLD_TOKEN").await; + + // Connect-time token is served. + let v = variable_value(&db, w_id, path).await; + assert_eq!( + get_secret_value(&db, w_id, path, &v).await.unwrap(), + "OLD_TOKEN" + ); + + // Refresh persists the new token. + store_oauth_token_value(&db, w_id, path, "NEW_TOKEN") + .await + .unwrap(); + + let v = variable_value(&db, w_id, path).await; + assert_eq!( + get_secret_value(&db, w_id, path, &v).await.unwrap(), + "NEW_TOKEN", + "database backend should serve the refreshed token" + ); + println!(" ✓ database backend serves refreshed token"); + reset_backend(&db).await; +} + +/// External backend (the "with external storage" case): refresh must write +/// the new token to AWS Secrets Manager. Before the fix the external store +/// stayed frozen and reads served the stale connect-time token. +#[tokio::test] +async fn external_backend_persists_refreshed_token() { + if !run_e2e() || !run_aws_sm() { + println!("Skipping external_backend_persists_refreshed_token: set RUN_SECRET_BACKEND_E2E=1 and RUN_AWS_SM_TESTS=1"); + return; + } + let _guard = SERIAL.lock().await; + let db = db().await; + let w_id = "wm_e2e_awssm"; + let path = "f/google/gsheets"; + + set_backend(&db, aws_sm_config(&aws_sm_endpoint())).await; + setup_workspace(&db, w_id).await; + let _ = simulate_connect(&db, w_id, path, "OLD_TOKEN").await; + + // Connect-time token is served from the external store. + let marker = variable_value(&db, w_id, path).await; + assert!( + marker.starts_with("$aws_sm:"), + "external backend should store a marker in variable.value, got {marker}" + ); + assert_eq!( + get_secret_value(&db, w_id, path, &marker).await.unwrap(), + "OLD_TOKEN" + ); + + // Demonstrate the original bug shape: a raw DB write to variable.value is + // futile because reads resolve through the backend and ignore it. + sqlx::query( + "UPDATE variable SET value = 'ignored_db_blob' WHERE workspace_id = $1 AND path = $2", + ) + .bind(w_id) + .bind(path) + .execute(&db) + .await + .unwrap(); + assert_eq!( + get_secret_value(&db, w_id, path, "ignored_db_blob") + .await + .unwrap(), + "OLD_TOKEN", + "reads ignore variable.value for external backends — a raw UPDATE can't refresh the served token" + ); + + // The fix: persist through the backend. + store_oauth_token_value(&db, w_id, path, "NEW_TOKEN") + .await + .unwrap(); + + let marker = variable_value(&db, w_id, path).await; + assert_eq!( + get_secret_value(&db, w_id, path, &marker).await.unwrap(), + "NEW_TOKEN", + "external backend should serve the refreshed token written back to AWS SM" + ); + println!(" ✓ external (AWS SM) backend serves refreshed token written back to the store"); + reset_backend(&db).await; +} + +/// If persisting the refreshed token fails (e.g. transient external-backend +/// error) after the account was committed fresh, the account expiry must be +/// reset to the past so the next fetch retries instead of serving a stale +/// token for the whole token lifetime. +#[tokio::test] +async fn failed_persist_resets_account_expiry() { + if !run_e2e() || !run_aws_sm() { + println!("Skipping failed_persist_resets_account_expiry: set RUN_SECRET_BACKEND_E2E=1 and RUN_AWS_SM_TESTS=1"); + return; + } + let _guard = SERIAL.lock().await; + let db = db().await; + let w_id = "wm_e2e_selfheal"; + let path = "f/google/gdrive"; + + // Working backend first to seed the variable + a *fresh* account. + set_backend(&db, aws_sm_config(&aws_sm_endpoint())).await; + setup_workspace(&db, w_id).await; + let account_id = simulate_connect(&db, w_id, path, "OLD_TOKEN").await; + sqlx::query("UPDATE account SET expires_at = now() + interval '1 hour' WHERE workspace_id = $1 AND id = $2") + .bind(w_id) + .bind(account_id) + .execute(&db) + .await + .unwrap(); + assert!(!account_expires_in_past(&db, w_id, account_id).await); + + // Point the backend at an unreachable endpoint so the persist fails. + set_backend(&db, aws_sm_config("http://127.0.0.1:1")).await; + + let res = store_oauth_token_value(&db, w_id, path, "NEW_TOKEN").await; + assert!(res.is_err(), "persist to unreachable backend should fail"); + assert!( + account_expires_in_past(&db, w_id, account_id).await, + "a failed persist must reset account.expires_at to the past so refresh retries" + ); + println!(" ✓ failed persist reset account expiry (self-healing)"); + reset_backend(&db).await; +} diff --git a/backend/windmill-store/src/secret_backend_ext.rs b/backend/windmill-store/src/secret_backend_ext.rs index a83cb56b9e..1ae32c80ce 100644 --- a/backend/windmill-store/src/secret_backend_ext.rs +++ b/backend/windmill-store/src/secret_backend_ext.rs @@ -26,7 +26,10 @@ use windmill_common::{ #[cfg(all(feature = "private", feature = "enterprise"))] use windmill_common::{ global_settings::{load_value_from_global_settings, SECRET_BACKEND_SETTING}, - secret_backend::{AwsSecretsManagerBackend, AwsSecretsManagerSettings, AzureKeyVaultBackend, AzureKeyVaultSettings, SecretBackendConfig, VaultBackend, VaultSettings}, + secret_backend::{ + AwsSecretsManagerBackend, AwsSecretsManagerSettings, AzureKeyVaultBackend, + AzureKeyVaultSettings, SecretBackendConfig, VaultBackend, VaultSettings, + }, }; #[cfg(all(feature = "private", feature = "enterprise"))] @@ -225,7 +228,12 @@ pub async fn is_vault_backend_configured(db: &DB) -> Result { None => SecretBackendConfig::default(), }; - Ok(matches!(config, SecretBackendConfig::HashiCorpVault(_) | SecretBackendConfig::AzureKeyVault(_) | SecretBackendConfig::AwsSecretsManager(_))) + Ok(matches!( + config, + SecretBackendConfig::HashiCorpVault(_) + | SecretBackendConfig::AzureKeyVault(_) + | SecretBackendConfig::AwsSecretsManager(_) + )) } /// Get a secret value using the configured backend @@ -252,12 +260,8 @@ pub async fn get_secret_value( // Fetch from Vault directly backend.get_secret(workspace_id, path).await } - "azure_key_vault" => { - backend.get_secret(workspace_id, path).await - } - "aws_secrets_manager" => { - backend.get_secret(workspace_id, path).await - } + "azure_key_vault" => backend.get_secret(workspace_id, path).await, + "aws_secrets_manager" => backend.get_secret(workspace_id, path).await, _ => Err(Error::internal_err(format!( "Unknown backend: {}", backend.backend_name() @@ -303,6 +307,93 @@ pub async fn store_secret_value( } } +/// Persist a freshly minted OAuth access token to the secret variable backing +/// a resource, routing through the configured secret backend. +/// +/// This is the write counterpart of the lazy on-fetch OAuth refresh: it stores +/// the token via [`store_secret_value`] (which writes to the external backend — +/// AWS Secrets Manager / Azure Key Vault / Vault — when one is configured, or +/// encrypts for the database backend) and updates `variable.value` with the +/// returned value (the encrypted blob for the DB backend, or a `$...:` marker +/// for an external backend). Using a raw `UPDATE variable SET value = ` +/// here instead would leave the external store frozen at its connect-time token +/// while reads (which resolve through the backend) keep serving the stale value. +/// +/// The caller has already committed the `account` row as fresh (advanced +/// `expires_at`) by the time we get here. If persisting the token fails — most +/// likely a transient error talking to an external backend — that would leave +/// the account marked fresh while the served secret is stale, so the on-fetch +/// refresh gate (`now() > expires_at`) would skip refresh and keep serving the +/// stale token for the whole token lifetime. To avoid that we reset `expires_at` +/// to the past (and record `refresh_error`) on failure — looking the account up +/// via `variable.account` — so the very next fetch retries the refresh instead. +/// +/// Authorization contract: this performs NO access control. It writes the +/// caller-supplied token into the secret variable at `path` and may mutate the +/// linked `account` row, so callers MUST have already authorized the operation +/// against `workspace_id`/`path` (the OAuth refresh adapters only run after the +/// read path has resolved and gated the variable). It is therefore kept +/// `pub(crate)` and intended solely for the in-crate refresh adapters. +#[cfg(feature = "oauth2")] +pub(crate) async fn store_oauth_token_value( + db: &DB, + workspace_id: &str, + path: &str, + token: &str, +) -> Result<()> { + let persist = async { + let value = store_secret_value(db, workspace_id, path, token).await?; + sqlx::query("UPDATE variable SET value = $1 WHERE workspace_id = $2 AND path = $3") + .bind(value) + .bind(workspace_id) + .bind(path) + .execute(db) + .await?; + Ok::<(), Error>(()) + } + .await; + + if let Err(e) = persist { + // Mark the account expired again so the next fetch re-runs the refresh + // instead of serving the now-stale token until it naturally expires. The + // account id is the one linked from the variable being refreshed. + let account_id: Option = sqlx::query_scalar::<_, Option>( + "SELECT account FROM variable WHERE workspace_id = $1 AND path = $2", + ) + .bind(workspace_id) + .bind(path) + .fetch_optional(db) + .await + .ok() + .flatten() + .flatten(); + + if let Some(account_id) = account_id { + if let Err(reset_err) = sqlx::query( + "UPDATE account SET expires_at = now() - interval '1 minute', refresh_error = $1 \ + WHERE workspace_id = $2 AND id = $3", + ) + .bind(format!( + "OAuth token was refreshed but persisting it to the secret backend failed: {e}" + )) + .bind(workspace_id) + .bind(account_id) + .execute(db) + .await + { + tracing::error!( + workspace_id = %workspace_id, + account_id = %account_id, + "failed to reset account expiry after token persistence error: {reset_err}" + ); + } + } + return Err(e); + } + + Ok(()) +} + /// Delete a secret from the configured backend (if using Vault) /// /// For database backend: no-op (DB delete is handled separately)