diff --git a/backend/.sqlx/query-09e2a19435068f9e9bfd5bcb44b4e283c71729f81550f6f7156ce4970345cc07.json b/backend/.sqlx/query-09e2a19435068f9e9bfd5bcb44b4e283c71729f81550f6f7156ce4970345cc07.json deleted file mode 100644 index 8ece28ff8f..0000000000 --- a/backend/.sqlx/query-09e2a19435068f9e9bfd5bcb44b4e283c71729f81550f6f7156ce4970345cc07.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE account SET refresh_error = $1 WHERE workspace_id = $2 AND id = $3", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Text", - "Int4" - ] - }, - "nullable": [] - }, - "hash": "09e2a19435068f9e9bfd5bcb44b4e283c71729f81550f6f7156ce4970345cc07" -} diff --git a/backend/.sqlx/query-23b2a39e37e93ca9367e954eba94b1a6f432edb76320c50e0d07088937c17a01.json b/backend/.sqlx/query-23b2a39e37e93ca9367e954eba94b1a6f432edb76320c50e0d07088937c17a01.json new file mode 100644 index 0000000000..de6cf870b5 --- /dev/null +++ b/backend/.sqlx/query-23b2a39e37e93ca9367e954eba94b1a6f432edb76320c50e0d07088937c17a01.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT expires_at FROM account WHERE workspace_id = $1 AND id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "expires_at", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Text", + "Int4" + ] + }, + "nullable": [ + false + ] + }, + "hash": "23b2a39e37e93ca9367e954eba94b1a6f432edb76320c50e0d07088937c17a01" +} diff --git a/backend/.sqlx/query-2c0ab7571e1a7c4290315bc3efccb4db9e0c9aee05596a594f81975a0cdb74d1.json b/backend/.sqlx/query-2c0ab7571e1a7c4290315bc3efccb4db9e0c9aee05596a594f81975a0cdb74d1.json new file mode 100644 index 0000000000..9c4b3eb842 --- /dev/null +++ b/backend/.sqlx/query-2c0ab7571e1a7c4290315bc3efccb4db9e0c9aee05596a594f81975a0cdb74d1.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT value FROM variable WHERE workspace_id = $1 AND path = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "value", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "2c0ab7571e1a7c4290315bc3efccb4db9e0c9aee05596a594f81975a0cdb74d1" +} diff --git a/backend/.sqlx/query-63c48fde8c0c0fff9abffc3be27e9948556b636b70b818cc31c2d50921a27366.json b/backend/.sqlx/query-3829559733fa087201f935ee10a5cecf2cec4f1646015ed7cf476efaba67707f.json similarity index 53% rename from backend/.sqlx/query-63c48fde8c0c0fff9abffc3be27e9948556b636b70b818cc31c2d50921a27366.json rename to backend/.sqlx/query-3829559733fa087201f935ee10a5cecf2cec4f1646015ed7cf476efaba67707f.json index 0ba2d18257..af7a60664d 100644 --- a/backend/.sqlx/query-63c48fde8c0c0fff9abffc3be27e9948556b636b70b818cc31c2d50921a27366.json +++ b/backend/.sqlx/query-3829559733fa087201f935ee10a5cecf2cec4f1646015ed7cf476efaba67707f.json @@ -1,42 +1,62 @@ { "db_name": "PostgreSQL", - "query": "SELECT client, refresh_token, grant_type, cc_client_id, cc_client_secret, cc_token_url, scopes FROM account WHERE workspace_id = $1 AND id = $2", + "query": "SELECT mcp_server_url, is_workspace_integration, client, refresh_token, grant_type,\n cc_client_id, cc_client_secret, cc_token_url, scopes, expires_at,\n now() > expires_at AS \"expired!\"\n FROM account WHERE workspace_id = $1 AND id = $2", "describe": { "columns": [ { "ordinal": 0, + "name": "mcp_server_url", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "is_workspace_integration", + "type_info": "Bool" + }, + { + "ordinal": 2, "name": "client", "type_info": "Varchar" }, { - "ordinal": 1, + "ordinal": 3, "name": "refresh_token", "type_info": "Varchar" }, { - "ordinal": 2, + "ordinal": 4, "name": "grant_type", "type_info": "Varchar" }, { - "ordinal": 3, + "ordinal": 5, "name": "cc_client_id", "type_info": "Varchar" }, { - "ordinal": 4, + "ordinal": 6, "name": "cc_client_secret", "type_info": "Varchar" }, { - "ordinal": 5, + "ordinal": 7, "name": "cc_token_url", "type_info": "Varchar" }, { - "ordinal": 6, + "ordinal": 8, "name": "scopes", "type_info": "TextArray" + }, + { + "ordinal": 9, + "name": "expires_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 10, + "name": "expired!", + "type_info": "Bool" } ], "parameters": { @@ -46,14 +66,18 @@ ] }, "nullable": [ + true, + false, false, false, false, true, true, true, - true + true, + false, + null ] }, - "hash": "63c48fde8c0c0fff9abffc3be27e9948556b636b70b818cc31c2d50921a27366" + "hash": "3829559733fa087201f935ee10a5cecf2cec4f1646015ed7cf476efaba67707f" } diff --git a/backend/.sqlx/query-8a0287e5b8f37c86d989a39516188484d036e2106acab33f6c459d9b9c9ba3be.json b/backend/.sqlx/query-8a0287e5b8f37c86d989a39516188484d036e2106acab33f6c459d9b9c9ba3be.json new file mode 100644 index 0000000000..c90368a92f --- /dev/null +++ b/backend/.sqlx/query-8a0287e5b8f37c86d989a39516188484d036e2106acab33f6c459d9b9c9ba3be.json @@ -0,0 +1,19 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE account SET refresh_token = COALESCE($1, refresh_token),\n expires_at = CASE WHEN $2 THEN $3 ELSE expires_at END,\n refresh_error = CASE WHEN $2 THEN NULL ELSE refresh_error END\n WHERE workspace_id = $4 AND id = $5 AND refresh_token = $6", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Bool", + "Timestamptz", + "Text", + "Int4", + "Text" + ] + }, + "nullable": [] + }, + "hash": "8a0287e5b8f37c86d989a39516188484d036e2106acab33f6c459d9b9c9ba3be" +} diff --git a/backend/.sqlx/query-8c2fcd9385931cfba49bcbe67a14cbfbb042a5930d406ec60939791a1d1d723b.json b/backend/.sqlx/query-8c2fcd9385931cfba49bcbe67a14cbfbb042a5930d406ec60939791a1d1d723b.json new file mode 100644 index 0000000000..bbf54e23b8 --- /dev/null +++ b/backend/.sqlx/query-8c2fcd9385931cfba49bcbe67a14cbfbb042a5930d406ec60939791a1d1d723b.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT path FROM variable WHERE workspace_id = $1 AND account = $2 AND is_secret", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text", + "Int4" + ] + }, + "nullable": [ + false + ] + }, + "hash": "8c2fcd9385931cfba49bcbe67a14cbfbb042a5930d406ec60939791a1d1d723b" +} diff --git a/backend/.sqlx/query-db9831a3218b36c32b305b9f2ba21abceec74c9c277f27231dba12868e68126b.json b/backend/.sqlx/query-db9831a3218b36c32b305b9f2ba21abceec74c9c277f27231dba12868e68126b.json deleted file mode 100644 index 3562a6f345..0000000000 --- a/backend/.sqlx/query-db9831a3218b36c32b305b9f2ba21abceec74c9c277f27231dba12868e68126b.json +++ /dev/null @@ -1,17 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE account SET refresh_token = $1, expires_at = $2, refresh_error = NULL WHERE workspace_id = $3 AND id = $4", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Timestamptz", - "Text", - "Int4" - ] - }, - "nullable": [] - }, - "hash": "db9831a3218b36c32b305b9f2ba21abceec74c9c277f27231dba12868e68126b" -} diff --git a/backend/.sqlx/query-e43b178e40ef39908def1746f7f2a6a2ad960a7b5e9536d15efae31bd2431742.json b/backend/.sqlx/query-e43b178e40ef39908def1746f7f2a6a2ad960a7b5e9536d15efae31bd2431742.json new file mode 100644 index 0000000000..2319cf80f7 --- /dev/null +++ b/backend/.sqlx/query-e43b178e40ef39908def1746f7f2a6a2ad960a7b5e9536d15efae31bd2431742.json @@ -0,0 +1,17 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE account SET refresh_error = $1\n WHERE workspace_id = $2 AND id = $3 AND refresh_token = $4", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text", + "Int4", + "Text" + ] + }, + "nullable": [] + }, + "hash": "e43b178e40ef39908def1746f7f2a6a2ad960a7b5e9536d15efae31bd2431742" +} diff --git a/backend/.sqlx/query-e910ae81fb98642e76644e9691157c3d02adce3ba55324375a2035231c640ac0.json b/backend/.sqlx/query-e910ae81fb98642e76644e9691157c3d02adce3ba55324375a2035231c640ac0.json new file mode 100644 index 0000000000..8805e4ee22 --- /dev/null +++ b/backend/.sqlx/query-e910ae81fb98642e76644e9691157c3d02adce3ba55324375a2035231c640ac0.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT EXISTS (SELECT 1 FROM account WHERE workspace_id = $1 AND id = $2) AS \"exists!\"", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "exists!", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text", + "Int4" + ] + }, + "nullable": [ + null + ] + }, + "hash": "e910ae81fb98642e76644e9691157c3d02adce3ba55324375a2035231c640ac0" +} diff --git a/backend/.sqlx/query-fa9cba875690fa5568a7bb3f33464ccea31da990bb66d6c29b48d39f714c7c8b.json b/backend/.sqlx/query-fa9cba875690fa5568a7bb3f33464ccea31da990bb66d6c29b48d39f714c7c8b.json new file mode 100644 index 0000000000..8515006507 --- /dev/null +++ b/backend/.sqlx/query-fa9cba875690fa5568a7bb3f33464ccea31da990bb66d6c29b48d39f714c7c8b.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT now() > expires_at AS \"expired!\" FROM account\n WHERE workspace_id = $1 AND id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "expired!", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text", + "Int4" + ] + }, + "nullable": [ + null + ] + }, + "hash": "fa9cba875690fa5568a7bb3f33464ccea31da990bb66d6c29b48d39f714c7c8b" +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 20ac124f54..92f89c68fd 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -c465c042eeae3c5d9a0180f62a56e8f8d3e9e162 +3546b6b080e0234335ed2c8e2e9cc3d595bbadca diff --git a/backend/windmill-api/src/mcp_tools.rs b/backend/windmill-api/src/mcp_tools.rs index ee7cd5e6b7..a7d32f6e9e 100644 --- a/backend/windmill-api/src/mcp_tools.rs +++ b/backend/windmill-api/src/mcp_tools.rs @@ -103,13 +103,12 @@ async fn connect_mcp_client( if let Some(info) = token_info { if let (Some(account_id), Some(true)) = (info.account_id, info.is_expired) { - let refresh_tx = user_db.clone().begin(authed).await?; if let Err(e) = crate::oauth2_oss::_refresh_token( - refresh_tx, token_var_path, w_id, account_id, db, + false, ) .await { diff --git a/backend/windmill-api/src/oauth2_oss.rs b/backend/windmill-api/src/oauth2_oss.rs index 6a9caee28f..9b8bcd1294 100644 --- a/backend/windmill-api/src/oauth2_oss.rs +++ b/backend/windmill-api/src/oauth2_oss.rs @@ -22,8 +22,6 @@ use hmac::Mac; use itertools::Itertools; #[cfg(not(feature = "private"))] use serde::{Deserialize, Serialize}; -#[cfg(not(feature = "private"))] -use sqlx::{Postgres, Transaction}; #[cfg(all(feature = "oauth2", not(feature = "private")))] use windmill_common::more_serde::maybe_number_opt; #[cfg(all(feature = "oauth2", not(feature = "private")))] @@ -133,12 +131,12 @@ async fn list_connects() -> windmill_common::error::JsonResult> { } #[cfg(not(feature = "private"))] -pub async fn _refresh_token<'c>( - _tx: Transaction<'c, Postgres>, +pub async fn _refresh_token( _path: &str, _w_id: &str, _id: i32, _db: &DB, + _force: bool, ) -> error::Result { // Implementation is not open source Err(error::Error::BadRequest( diff --git a/backend/windmill-oauth/src/lib.rs b/backend/windmill-oauth/src/lib.rs index 9768911813..a30d6007f8 100644 --- a/backend/windmill-oauth/src/lib.rs +++ b/backend/windmill-oauth/src/lib.rs @@ -20,12 +20,10 @@ use std::fmt::Debug; use anyhow::anyhow; use hmac::Mac; use serde::{de::DeserializeOwned, Deserialize, Serialize}; -use sqlx::{Postgres, Transaction}; use tower_cookies::{Cookie, Cookies}; 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::BASE_URL; pub type DB = sqlx::Pool; @@ -890,61 +888,39 @@ pub struct OAuthAccountInfo { pub scopes: Option>, } -/// 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>, - w_id: &str, - id: i32, - db: &DB, - oauth_clients: &AllClients, - http_client: &reqwest::Client, - connect_configs_json: &str, -) -> error::Result { - let account = sqlx::query_as!( - OAuthAccountInfo, - "SELECT client, refresh_token, grant_type, cc_client_id, cc_client_secret, cc_token_url, scopes FROM account WHERE workspace_id = $1 AND id = $2", - w_id, - id, - ) - .fetch_optional(&mut *tx) - .await?; - let account = windmill_common::utils::not_found_if_none(account, "Account", &id.to_string())?; - - refresh_token_for_account( - tx, - w_id, - id, - db, - account, - oauth_clients, - http_client, - connect_configs_json, - ) - .await +/// A refresh the provider answered with a new access token. +pub struct RefreshedToken { + pub access_token: String, + /// The rotated refresh token, when the provider issued one. + pub refresh_token: Option, + pub expires_in: i64, } -/// Refresh an OAuth token given pre-fetched account info (no additional SELECT). +pub enum RefreshError { + /// The provider refused the exchange: recorded on the account as `refresh_error`. + Provider(String), + Other(Error), +} + +impl From for RefreshError { + fn from(e: Error) -> Self { + RefreshError::Other(e) + } +} + +/// Exchange an account's refresh token (or client credentials) with its provider. /// -/// 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>, +/// Holds no connection across the HTTP call and writes nothing: the caller records the +/// outcome on the `account` row and persists the access token through the secret backend. +pub async fn exchange_account_token( + db: &DB, w_id: &str, id: i32, - db: &DB, - account: OAuthAccountInfo, + account: &OAuthAccountInfo, oauth_clients: &AllClients, http_client: &reqwest::Client, connect_configs_json: &str, -) -> error::Result { +) -> Result { // Instance-configured client: required for authorization_code (the refresh // token exchange uses the instance app's credentials). For client_credentials // it is resolved inside `build_client_credentials_oauth_client` instead. @@ -1043,48 +1019,13 @@ pub async fn refresh_token_for_account<'c>( ) .await; - if let Err(token_err) = token { - sqlx::query!( - "UPDATE account SET refresh_error = $1 WHERE workspace_id = $2 AND id = $3", - token_err.alt(), - w_id, - id, - ) - .execute(&mut *tx) - .await?; - tx.commit().await?; - return Err(error::Error::BadRequest(format!( - "Error refreshing token: {}", - token_err.alt() - ))); - }; + let token = token.map_err(|e| RefreshError::Provider(e.alt()))?; - let token = token.unwrap(); - - let expires_at = now_from_db(&mut *tx).await? - + chrono::Duration::try_seconds( - token - .expires_in - .ok_or_else(|| Error::InternalErr("expires_in expected and not found".to_string()))? - .try_into() - .unwrap(), - ) - .unwrap_or_default(); - sqlx::query!( - "UPDATE account SET refresh_token = $1, expires_at = $2, refresh_error = NULL WHERE workspace_id = $3 AND id = $4", - token - .refresh_token - .map(|x| x.to_string()) - .unwrap_or(account.refresh_token), - expires_at, - w_id, - id, - ) - .execute(&mut *tx) - .await?; - tx.commit().await?; - - let token_str = token.access_token.to_string(); + let expires_in = token + .expires_in + .ok_or_else(|| Error::InternalErr("expires_in expected and not found".to_string()))? + .try_into() + .unwrap_or(i64::MAX); tracing::info!( grant_type = %account.grant_type, @@ -1094,7 +1035,11 @@ pub async fn refresh_token_for_account<'c>( "OAuth token refreshed successfully" ); - Ok(token_str) + Ok(RefreshedToken { + access_token: token.access_token.to_string(), + refresh_token: token.refresh_token.map(|t| t.to_string()), + expires_in, + }) } /// Generate OAuth redirect URL with CSRF protection diff --git a/backend/windmill-store/src/oauth_refresh_oss.rs b/backend/windmill-store/src/oauth_refresh_oss.rs index 13e24dfd8b..17d49aae4f 100644 --- a/backend/windmill-store/src/oauth_refresh_oss.rs +++ b/backend/windmill-store/src/oauth_refresh_oss.rs @@ -6,39 +6,238 @@ * LICENSE-AGPL for a copy of the license. */ +use std::future::Future; +use std::sync::Arc; + +use windmill_common::db::DB; +use windmill_common::error::{self, Error}; +use windmill_oauth::{OAuthAccountInfo, RefreshError, RefreshedToken}; + #[cfg(feature = "private")] pub use crate::oauth_refresh_ee::_refresh_token; -#[cfg(not(feature = "private"))] -use sqlx::{Postgres, Transaction}; -#[cfg(not(feature = "private"))] -use windmill_common::db::DB; -#[cfg(not(feature = "private"))] -use windmill_common::error; +/// The `account` row a refresh starts from. +pub struct RefreshAccount { + pub mcp_server_url: Option, + pub is_workspace_integration: bool, + pub info: OAuthAccountInfo, +} -#[cfg(not(feature = "private"))] -pub async fn _refresh_token<'c>( - tx: Transaction<'c, Postgres>, - path: &str, - w_id: &str, - id: i32, +lazy_static::lazy_static! { + static ref REFRESH_LOCKS: dashmap::DashMap<(String, i32), Arc>> = + dashmap::DashMap::new(); +} + +/// Refresh the token of account `id`, which backs the secret variable at `path`. +/// +/// No connection is held across `exchange`: it does its own config reads and the provider +/// call, and every write here is a single autocommit statement. Concurrent refreshers of one +/// account are serialized within a process, and across processes the account is only +/// committed if its refresh token is still the one exchanged, so a rotated refresh token is +/// never overwritten by a stale one. An account refreshed by someone else while we waited for +/// the lock is not refreshed again: without `force` any fresh account counts, with it only one +/// whose expiry moved since we asked. Callers must have authorized `w_id`/`path`/`id`. +pub(crate) async fn refresh_account_token( db: &DB, -) -> error::Result { - let token = windmill_oauth::refresh_token( - tx, + w_id: &str, + path: &str, + id: i32, + force: bool, + exchange: F, +) -> error::Result +where + F: FnOnce(RefreshAccount) -> Fut, + Fut: Future>, +{ + let seen_expiry = if force { + sqlx::query_scalar!( + "SELECT expires_at FROM account WHERE workspace_id = $1 AND id = $2", + w_id, + id + ) + .fetch_optional(db) + .await? + } else { + None + }; + + let lock = REFRESH_LOCKS + .entry((w_id.to_string(), id)) + .or_default() + .clone(); + let _guard = lock.lock().await; + + let row = sqlx::query!( + "SELECT mcp_server_url, is_workspace_integration, client, refresh_token, grant_type, + cc_client_id, cc_client_secret, cc_token_url, scopes, expires_at, + now() > expires_at AS \"expired!\" + FROM account WHERE workspace_id = $1 AND id = $2", w_id, id, + ) + .fetch_optional(db) + .await? + .ok_or_else(|| Error::NotFound(format!("Account {id} not found")))?; + + // A failed store moves `expires_at` too, into the past: only a fresh account counts. + let refreshed_meanwhile = !row.expired && (!force || Some(row.expires_at) != seen_expiry); + if refreshed_meanwhile { + return current_token(db, w_id, path).await; + } + + let used = row.refresh_token.clone(); + let account = RefreshAccount { + mcp_server_url: row.mcp_server_url, + is_workspace_integration: row.is_workspace_integration, + info: OAuthAccountInfo { + client: row.client, + refresh_token: row.refresh_token, + grant_type: row.grant_type, + cc_client_id: row.cc_client_id, + cc_client_secret: row.cc_client_secret, + cc_token_url: row.cc_token_url, + scopes: row.scopes, + }, + }; + + let token = match exchange(account).await { + Ok(token) => token, + Err(RefreshError::Other(e)) => return Err(e), + Err(RefreshError::Provider(msg)) => { + let recorded = sqlx::query!( + "UPDATE account SET refresh_error = $1 + WHERE workspace_id = $2 AND id = $3 AND refresh_token = $4", + msg, + w_id, + id, + used, + ) + .execute(db) + .await? + .rows_affected(); + // Another process rotated the refresh token first, so the one we sent was spent: + // its token is the one to serve, unless storing it failed too. + if recorded == 0 { + let expired = sqlx::query_scalar!( + "SELECT now() > expires_at AS \"expired!\" FROM account + WHERE workspace_id = $1 AND id = $2", + w_id, + id + ) + .fetch_optional(db) + .await?; + if expired == Some(false) { + return current_token(db, w_id, path).await; + } + } + return Err(Error::BadRequest(format!("Error refreshing token: {msg}"))); + } + }; + + // The lifetime runs from the exchange, not from the end of the stores below. + let expires_at = windmill_common::utils::now_from_db(db).await? + + chrono::Duration::try_seconds(token.expires_in).unwrap_or_default(); + + // Store at every secret variable of the account before marking it fresh, so no reader of + // any of them sees the account fresh with the old token. A failed store already marks the + // account for a retry; the rotated refresh token must still be saved, or the next refresh + // sends one the provider has revoked. + let mut paths = sqlx::query_scalar!( + "SELECT path FROM variable WHERE workspace_id = $1 AND account = $2 AND is_secret", + w_id, + id + ) + .fetch_all(db) + .await?; + if !paths.iter().any(|p| p == path) { + paths.push(path.to_string()); + } + let mut stored = Ok(()); + for p in &paths { + stored = + crate::secret_backend_ext::store_oauth_token_value(db, w_id, p, &token.access_token) + .await; + if stored.is_err() { + break; + } + } + sqlx::query!( + "UPDATE account SET refresh_token = COALESCE($1, refresh_token), + expires_at = CASE WHEN $2 THEN $3 ELSE expires_at END, + refresh_error = CASE WHEN $2 THEN NULL ELSE refresh_error END + WHERE workspace_id = $4 AND id = $5 AND refresh_token = $6", + token.refresh_token, + stored.is_ok(), + expires_at, + w_id, + id, + used, + ) + .execute(db) + .await?; + stored?; + + Ok(token.access_token) +} + +/// Whether account `id` exists in `w_id`, read through the caller's user transaction. `account` +/// carries no RLS, so this gates nothing beyond the workspace the caller is already authorized +/// for: a refresh asked for by account id is scoped by that membership, not by ownership. +pub async fn account_visible( + user_conn: &mut sqlx::PgConnection, + w_id: &str, + id: i32, +) -> error::Result { + Ok(sqlx::query_scalar!( + "SELECT EXISTS (SELECT 1 FROM account WHERE workspace_id = $1 AND id = $2) AS \"exists!\"", + w_id, + id + ) + .fetch_one(user_conn) + .await?) +} + +async fn current_token(db: &DB, w_id: &str, path: &str) -> error::Result { + let value = sqlx::query_scalar!( + "SELECT value FROM variable WHERE workspace_id = $1 AND path = $2", + w_id, + path + ) + .fetch_optional(db) + .await? + .ok_or_else(|| Error::NotFound(format!("Variable {path} not found")))?; + crate::secret_backend_ext::get_secret_value(db, w_id, path, &value).await +} + +/// The standard OAuth / client-credentials exchange. +pub(crate) async fn exchange_standard( + db: &DB, + w_id: &str, + id: i32, + account: &OAuthAccountInfo, +) -> Result { + windmill_oauth::exchange_account_token( db, + w_id, + id, + account, &*windmill_oauth::OAUTH_CLIENTS.load(), &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) +#[cfg(not(feature = "private"))] +pub async fn _refresh_token( + path: &str, + w_id: &str, + id: i32, + db: &DB, + force: bool, +) -> error::Result { + refresh_account_token(db, w_id, path, id, force, |account| async move { + exchange_standard(db, w_id, id, &account.info).await + }) + .await } diff --git a/backend/windmill-store/src/secret_backend_ext.rs b/backend/windmill-store/src/secret_backend_ext.rs index d5272384c0..814a3174a9 100644 --- a/backend/windmill-store/src/secret_backend_ext.rs +++ b/backend/windmill-store/src/secret_backend_ext.rs @@ -77,14 +77,13 @@ pub async fn store_secret_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. +/// If persisting the token fails — most likely a transient error talking to an +/// external backend — the account must not end up marked fresh while the served +/// secret is stale, or the on-fetch refresh gate (`now() > expires_at`) would keep +/// serving the stale token for the whole token lifetime. So on failure we reset +/// `expires_at` to the past and record `refresh_error` — looking the account up via +/// `variable.account` — and the very next fetch retries the refresh. This also +/// covers another refresher of the account having marked it fresh meanwhile. /// /// Authorization contract: this performs NO access control. It writes the /// caller-supplied token into the secret variable at `path` and may mutate the diff --git a/backend/windmill-store/src/variables.rs b/backend/windmill-store/src/variables.rs index fca29736e6..42f8b4cc01 100644 --- a/backend/windmill-store/src/variables.rs +++ b/backend/windmill-store/src/variables.rs @@ -373,42 +373,42 @@ async fn get_variable( explain_variable_perm_error(&path, &w_id, &db, Some(&authed)).await?; unreachable!() } else { + drop(tx); explain_variable_perm_error(&path, &w_id, &db, Some(&authed)).await?; unreachable!() }; let decrypt_secret = q.decrypt_secret.unwrap_or(true); - let r = if variable.is_secret { - if decrypt_secret { - audit_log( - &mut *tx, - &authed, - "variables.decrypt_secret", - ActionKind::Execute, - &w_id, - Some(&variable.path), - None, - ) - .await?; - } + if variable.is_secret && decrypt_secret { + audit_log( + &mut *tx, + &authed, + "variables.decrypt_secret", + ActionKind::Execute, + &w_id, + Some(&variable.path), + None, + ) + .await?; + } + // Everything below reads the pool: holding this connection meanwhile would make each + // concurrent request wait on a second one. + tx.commit().await?; + let r = if variable.is_secret { let value = variable.value.unwrap_or_else(|| "".to_string()); ListableVariable { value: if variable.is_expired.unwrap_or(false) && variable.account.is_some() { #[cfg(feature = "oauth2")] { - let refresh_tx = db - .begin() - .await - .map_err(|e| Error::InternalErr(e.to_string()))?; Some( crate::oauth_refresh_oss::_refresh_token( - refresh_tx, &variable.path, &w_id, variable.account.unwrap(), &db, + false, ) .await?, ) @@ -416,7 +416,6 @@ async fn get_variable( #[cfg(not(feature = "oauth2"))] return Err(Error::internal_err("Require oauth2 feature".to_string())); } else if !value.is_empty() && decrypt_secret { - let _ = tx.commit().await; // Use secret backend for decryption (supports both DB and Vault) Some(get_secret_value(&db, &w_id, &variable.path, &value).await?) } else if q.include_encrypted.unwrap_or(false) { @@ -1543,17 +1542,12 @@ pub async fn get_value_internal<'a>( if variable.is_expired.unwrap_or(false) && variable.account.is_some() { #[cfg(feature = "oauth2")] { - let db = db_with_opt_authed.db(); - let refresh_tx = db - .begin() - .await - .map_err(|e| Error::InternalErr(e.to_string()))?; crate::oauth_refresh_oss::_refresh_token( - refresh_tx, &variable.path, &w_id, variable.account.unwrap(), - db, + db_with_opt_authed.db(), + false, ) .await? }