starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-10-13 08:02:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: key token write-back by account and serialize trigger creation with disconnect

Browse Source 
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
hugocasa
2026-09-28 09:01:28 +02:00
co-authored by Claude Opus 5.5
parent ac8637ac20
commit 560af04743
9 changed files with 152 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-17a29b1db3b9aea8ce4158ed1e475a4e26c766a15b0ec209504a24b2460575be.json
Generated
-24
View File
@@ -1,24 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE variable SET value = $1 WHERE workspace_id = $2 AND path = $3 RETURNING account",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "account",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text"
]
},
"nullable": [
true
]
},
"hash": "17a29b1db3b9aea8ce4158ed1e475a4e26c766a15b0ec209504a24b2460575be"
}
backend/.sqlx/query-5cb209fe91eeccdc4c208db7655b6349376f09dad74380302cb815fc2626c69c.json
Generated
-30
View File
@@ -1,30 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT v.value, a.refresh_token\n FROM variable v\n JOIN account a ON a.workspace_id = v.workspace_id AND a.id = v.account\n WHERE v.workspace_id = $1 AND v.path = $2\n AND a.client = $3 AND a.is_workspace_integration = true",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "value",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "refresh_token",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text"
]
},
"nullable": [
false,
false
]
},
"hash": "5cb209fe91eeccdc4c208db7655b6349376f09dad74380302cb815fc2626c69c"
}
backend/.sqlx/query-6be4bf59c404d2f557d1106c48c320bb3eff65255a44bd66799ae14288312ba4.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT EXISTS(SELECT 1 FROM variable WHERE workspace_id = $1 AND path = $2)",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "exists",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "6be4bf59c404d2f557d1106c48c320bb3eff65255a44bd66799ae14288312ba4"
}
backend/.sqlx/query-71b410dbc6cdd0252de2f8646395903c9fa60a5435fa73c587da2b913203cf18.json
Generated
+36
View File
@@ -0,0 +1,36 @@
{
"db_name": "PostgreSQL",
"query": "SELECT v.value, a.id AS account_id, a.refresh_token\n FROM variable v\n JOIN account a ON a.workspace_id = v.workspace_id AND a.id = v.account\n WHERE v.workspace_id = $1 AND v.path = $2\n AND a.client = $3 AND a.is_workspace_integration = true",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "value",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "account_id",
"type_info": "Int4"
},
{
"ordinal": 2,
"name": "refresh_token",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text"
]
},
"nullable": [
false,
false,
false
]
},
"hash": "71b410dbc6cdd0252de2f8646395903c9fa60a5435fa73c587da2b913203cf18"
}
backend/.sqlx/query-c427c9dad3de3acc361c341b359453197d7683bc8fd5a96be462e3493e957cee.json
Generated
+16
View File
@@ -0,0 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE variable SET value = $1 WHERE workspace_id = $2 AND account = $3",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text",
"Int4"
]
},
"nullable": []
},
"hash": "c427c9dad3de3acc361c341b359453197d7683bc8fd5a96be462e3493e957cee"
}
backend/windmill-api-integration-tests/tests/native_triggers.rs
+6 -2
View File
@@ -158,8 +158,9 @@ fn now_ms() -> i64 {
async fn test_connections_are_isolated(db: Pool<Postgres>) -> anyhow::Result<()> {
let path_a = "u/test-user/native_nextcloud";
let path_b = "u/other-user/native_nextcloud";
let mut account_a = 0;
for (path, token) in [(path_a, "a"), (path_b, "b")] {
setup_oauth_integration(
let account = setup_oauth_integration(
&db,
ServiceName::Nextcloud,
path,
@@ -168,12 +169,15 @@ async fn test_connections_are_isolated(db: Pool<Postgres>) -> anyhow::Result<()>
None,
)
.await?;
if path == path_a {
account_a = account;
}
}
update_oauth_token_resource(
&db,
"test-workspace",
path_a,
account_a,
"access-a2",
Some("refresh-a2"),
)
backend/windmill-native-triggers/src/handler.rs
+22 -5
View File
@@ -1,11 +1,12 @@
use crate::{
classify_read_failure, decrypt_oauth_data, delete_native_trigger, delete_token_by_hash,
get_native_trigger, list_native_triggers, list_usable_connections, lock::TriggerLock,
map_external_error, map_external_error_with, resolve_usable_connection, rotate_webhook_token,
set_native_trigger_enabled, store_native_trigger, sync::EXTERNAL_TRIGGER_MISSING_ERROR,
update_native_trigger_error, update_native_trigger_if_runnable_unchanged, webhook_token_label,
webhook_token_scopes, External, ExternalReadFailure, NativeTrigger, NativeTriggerConfig,
NativeTriggerData, ServiceName,
lock_connection, map_external_error, map_external_error_with, resolve_usable_connection,
rotate_webhook_token, set_native_trigger_enabled, store_native_trigger,
sync::EXTERNAL_TRIGGER_MISSING_ERROR, update_native_trigger_error,
update_native_trigger_if_runnable_unchanged, webhook_token_label, webhook_token_scopes,
External, ExternalReadFailure, NativeTrigger, NativeTriggerConfig, NativeTriggerData,
ServiceName,
};
use axum::{
extract::{Path, Query},
@@ -243,6 +244,22 @@ async fn create_native_trigger<T: External>(
webhook_token,
};
lock_connection(&mut tx, &workspace_id, &connection_path).await?;
let still_connected = sqlx::query_scalar!(
"SELECT EXISTS(SELECT 1 FROM variable WHERE workspace_id = $1 AND path = $2)",
workspace_id,
connection_path,
)
.fetch_one(&mut *tx)
.await?
.unwrap_or(false);
if !still_connected {
return Err(Error::BadRequest(format!(
"{connection_path} was disconnected while this trigger was being created. Its \
registration on {service_name} may remain; connect an account and create it again."
)));
}
store_native_trigger(
&mut *tx,
&workspace_id,
backend/windmill-native-triggers/src/lib.rs
+38 -22
View File
@@ -460,14 +460,14 @@ pub trait External: Send + Sync + 'static {
task::spawn({
let db_clone = db.clone();
let workspace_id_clone = workspace_id.to_string();
let connection_path = connection_path.to_string();
let account_id = refreshed_oauth_config.account_id;
let new_access_token = refreshed_oauth_config.access_token.clone();
let new_refresh_token = refreshed_oauth_config.refresh_token.clone();
async move {
update_oauth_token_resource(
&db_clone,
&workspace_id_clone,
&connection_path,
account_id,
&new_access_token,
new_refresh_token.as_deref(),
)
@@ -609,6 +609,7 @@ pub struct OAuthConfig {
pub client_id: String,
pub client_secret: String,
pub connection_path: String,
pub account_id: i32,
}
pub async fn make_http_request<T: DeserializeOwned + Send, B: Serialize>(
@@ -814,7 +815,7 @@ pub async fn decrypt_oauth_data<T: DeserializeOwned>(
let mc = build_crypt(db, workspace_id).await?;
let row = sqlx::query!(
"SELECT v.value, a.refresh_token
"SELECT v.value, a.id AS account_id, a.refresh_token
FROM variable v
JOIN account a ON a.workspace_id = v.workspace_id AND a.id = v.account
WHERE v.workspace_id = $1 AND v.path = $2
@@ -868,6 +869,7 @@ pub async fn decrypt_oauth_data<T: DeserializeOwned>(
"client_id": client_id,
"client_secret": client_secret,
"connection_path": connection_path,
"account_id": row.account_id,
});
serde_json::from_value(assembled)
@@ -994,6 +996,7 @@ pub async fn refresh_oauth_tokens(
client_id: oauth_config.client_id.clone(),
client_secret: oauth_config.client_secret.clone(),
connection_path: oauth_config.connection_path.clone(),
account_id: oauth_config.account_id,
})
}
@@ -1010,17 +1013,18 @@ pub async fn refresh_oauth_tokens(
})
}
/// Write a refreshed token back to the one connection it was refreshed for. The account is found
/// through that connection's variable, never by service: several connections share a service, and
/// a provider that rotates refresh tokens invalidates the old one, so writing it onto another
/// account would break that account's next refresh.
/// Write a refreshed token back to the account it was refreshed from, by that account's id, never
/// by service or path: several connections share a service, and a connection's path can be
/// reconnected or renamed while the refresh is in flight. A provider that rotates refresh tokens
/// invalidates the old one, so writing it onto another account would break that account's next
/// refresh.
///
/// No authorization happens here: it overwrites the connection's secret as the server. Call it
/// only with tokens just refreshed from that connection's own refresh token.
/// No authorization happens here: it overwrites the account's secrets as the server. Call it only
/// with tokens just refreshed from that account's own refresh token.
pub async fn update_oauth_token_resource(
db: &DB,
workspace_id: &str,
connection_path: &str,
account_id: i32,
new_access_token: &str,
new_refresh_token: Option<&str>,
) {
@@ -1028,19 +1032,14 @@ pub async fn update_oauth_token_resource(
let mc = build_crypt(db, workspace_id).await?;
let encrypted_token = encrypt(&mc, new_access_token);
let account_id = sqlx::query_scalar!(
"UPDATE variable SET value = $1 WHERE workspace_id = $2 AND path = $3 RETURNING account",
sqlx::query!(
"UPDATE variable SET value = $1 WHERE workspace_id = $2 AND account = $3",
encrypted_token,
workspace_id,
connection_path,
account_id,
)
.fetch_optional(db)
.await?
.flatten();
let Some(account_id) = account_id else {
return Ok(());
};
.execute(db)
.await?;
// Even without a new refresh token, expires_at moves so the background refresh does not
// re-refresh immediately.
@@ -1063,8 +1062,8 @@ pub async fn update_oauth_token_resource(
if let Err(e) = result {
tracing::error!(
"Failed to update OAuth tokens of connection {} in workspace {}: {}",
connection_path,
"Failed to update OAuth tokens of account {} in workspace {}: {}",
account_id,
workspace_id,
e
);
@@ -1813,6 +1812,23 @@ pub async fn resolve_usable_connection(
}
}
/// Serialize, until `tx` ends, the operations that must agree on whether a connection exists:
/// storing a trigger that uses it and disconnecting it. Whichever comes second waits and then
/// sees the other's outcome.
pub async fn lock_connection(
tx: &mut PgConnection,
workspace_id: &str,
connection_path: &str,
) -> Result<()> {
sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
.bind(format!(
"native_connection:{workspace_id}:{connection_path}"
))
.execute(&mut *tx)
.await?;
Ok(())
}
#[derive(Debug, Deserialize)]
pub struct ConnectionQuery {
pub connection_path: Option<String>,
backend/windmill-native-triggers/src/workspace_integrations.rs
+11 -7
View File
@@ -35,8 +35,8 @@ use windmill_api_auth::{check_scopes, require_is_writer, ApiAuthed};
#[cfg(feature = "native_trigger")]
use crate::{
decrypt_oauth_data, delete_token_by_hash, delete_workspace_integration,
list_usable_connections, nextcloud::OcsResponse, require_native_integration_use,
resolve_endpoint, Connection, ServiceName,
list_usable_connections, lock_connection, nextcloud::OcsResponse,
require_native_integration_use, resolve_endpoint, Connection, ServiceName,
};
#[cfg(feature = "native_trigger")]
@@ -1131,8 +1131,14 @@ async fn delete_connection(
)
.await;
let mut tx = user_db.begin(&authed).await?;
lock_connection(&mut tx, &workspace_id, &path).await?;
let account = cleanup_connection(&mut *tx, &workspace_id, &path).await?;
// A trigger moved or created meanwhile was skipped above and still acts through this
// connection, so removing the connection would strand it.
// connection, so removing the connection would strand it. Checked after the delete: a create
// holds the connection's variable until it commits, so the delete waits for it and this
// check then sees its trigger.
let still_used = sqlx::query_scalar!(
"SELECT EXISTS(SELECT 1 FROM native_trigger
WHERE workspace_id = $1 AND service_name = $2 AND connection_path = $3)",
@@ -1140,19 +1146,17 @@ async fn delete_connection(
service_name as ServiceName,
path,
)
.fetch_one(&db)
.fetch_one(&mut *tx)
.await?
.unwrap_or(false);
if still_used {
tx.rollback().await?;
return Err(Error::BadRequest(format!(
"A trigger using {path} changed while disconnecting, so the connection was kept. \
Disconnect again to remove it."
)));
}
let mut tx = user_db.begin(&authed).await?;
let account = cleanup_connection(&mut *tx, &workspace_id, &path).await?;
audit_log(
&mut *tx,
&authed,