mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-22 00:03:35 +00:00
fix: refresh oauth tokens without holding a pool connection (#11905)
* fix: refresh oauth tokens without holding a connection across the exchange Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * chore: pin ee ref for the oauth refresh fix Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * chore: update sqlx cache for the oauth refresh queries Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: coalesce forced oauth refreshes and drop the redundant mcp visibility check Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: store refreshed oauth tokens at every linked variable and only reuse a fresh one Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: date oauth expiry from the exchange and fan out only to secret variables Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * chore: update ee-repo-ref to 3546b6b080e0234335ed2c8e2e9cc3d595bbadca This commit updates the EE repository reference after PR #906 was merged in windmill-ee-private. Previous ee-repo-ref: 12d1fb5978e66af462398c789b861706a2367ed0 New ee-repo-ref: 3546b6b080e0234335ed2c8e2e9cc3d595bbadca Automated by sync-ee-ref workflow. --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
This commit is contained in:
Ruben Fiszel
co-authored by
Claude Opus 5.5
windmill-internal-app[bot]
Claude Opus 5.5
windmill-internal-app[bot]
parent
8afe0607ab
commit
8ddb358198
-16
@@ -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"
|
||||
}
|
||||
+23
@@ -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"
|
||||
}
|
||||
+23
@@ -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"
|
||||
}
|
||||
+33
-9
@@ -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"
|
||||
}
|
||||
+19
@@ -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"
|
||||
}
|
||||
+23
@@ -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"
|
||||
}
|
||||
-17
@@ -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"
|
||||
}
|
||||
+17
@@ -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"
|
||||
}
|
||||
+23
@@ -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"
|
||||
}
|
||||
+23
@@ -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"
|
||||
}
|
||||
@@ -1 +1 @@
|
||||
c465c042eeae3c5d9a0180f62a56e8f8d3e9e162
|
||||
3546b6b080e0234335ed2c8e2e9cc3d595bbadca
|
||||
|
||||
@@ -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
|
||||
{
|
||||
|
||||
@@ -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<Vec<String>> {
|
||||
}
|
||||
|
||||
#[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<String> {
|
||||
// Implementation is not open source
|
||||
Err(error::Error::BadRequest(
|
||||
|
||||
@@ -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<sqlx::Postgres>;
|
||||
@@ -890,61 +888,39 @@ pub struct OAuthAccountInfo {
|
||||
pub scopes: Option<Vec<String>>,
|
||||
}
|
||||
|
||||
/// 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<String> {
|
||||
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<String>,
|
||||
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<Error> 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<String> {
|
||||
) -> Result<RefreshedToken, RefreshError> {
|
||||
// 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
|
||||
|
||||
@@ -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<String>,
|
||||
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<tokio::sync::Mutex<()>>> =
|
||||
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<F, Fut>(
|
||||
db: &DB,
|
||||
) -> error::Result<String> {
|
||||
let token = windmill_oauth::refresh_token(
|
||||
tx,
|
||||
w_id: &str,
|
||||
path: &str,
|
||||
id: i32,
|
||||
force: bool,
|
||||
exchange: F,
|
||||
) -> error::Result<String>
|
||||
where
|
||||
F: FnOnce(RefreshAccount) -> Fut,
|
||||
Fut: Future<Output = Result<RefreshedToken, RefreshError>>,
|
||||
{
|
||||
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<bool> {
|
||||
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<String> {
|
||||
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<RefreshedToken, RefreshError> {
|
||||
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<String> {
|
||||
refresh_account_token(db, w_id, path, id, force, |account| async move {
|
||||
exchange_standard(db, w_id, id, &account.info).await
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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?
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user