From 560af04743ff7dc22ef77e75aaa5d53ed3f89e5a Mon Sep 17 00:00:00 2001 From: hugocasa Date: Mon, 28 Sep 2026 09:00:17 +0200 Subject: [PATCH] fix: key token write-back by account and serialize trigger creation with disconnect Co-Authored-By: Claude Opus 5.5 --- ...75a4e26c766a15b0ec209504a24b2460575be.json | 24 -------- ...b6349376f09dad74380302cb815fc2626c69c.json | 30 ---------- ...320bb3eff65255a44bd66799ae14288312ba4.json | 23 +++++++ ...5903c9fa60a5435fa73c587da2b913203cf18.json | 36 +++++++++++ ...453197d7683bc8fd5a96be462e3493e957cee.json | 16 +++++ .../tests/native_triggers.rs | 8 ++- .../windmill-native-triggers/src/handler.rs | 27 +++++++-- backend/windmill-native-triggers/src/lib.rs | 60 ++++++++++++------- .../src/workspace_integrations.rs | 18 +++--- 9 files changed, 152 insertions(+), 90 deletions(-) delete mode 100644 backend/.sqlx/query-17a29b1db3b9aea8ce4158ed1e475a4e26c766a15b0ec209504a24b2460575be.json delete mode 100644 backend/.sqlx/query-5cb209fe91eeccdc4c208db7655b6349376f09dad74380302cb815fc2626c69c.json create mode 100644 backend/.sqlx/query-6be4bf59c404d2f557d1106c48c320bb3eff65255a44bd66799ae14288312ba4.json create mode 100644 backend/.sqlx/query-71b410dbc6cdd0252de2f8646395903c9fa60a5435fa73c587da2b913203cf18.json create mode 100644 backend/.sqlx/query-c427c9dad3de3acc361c341b359453197d7683bc8fd5a96be462e3493e957cee.json diff --git a/backend/.sqlx/query-17a29b1db3b9aea8ce4158ed1e475a4e26c766a15b0ec209504a24b2460575be.json b/backend/.sqlx/query-17a29b1db3b9aea8ce4158ed1e475a4e26c766a15b0ec209504a24b2460575be.json deleted file mode 100644 index 44554fde78..0000000000 --- a/backend/.sqlx/query-17a29b1db3b9aea8ce4158ed1e475a4e26c766a15b0ec209504a24b2460575be.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-5cb209fe91eeccdc4c208db7655b6349376f09dad74380302cb815fc2626c69c.json b/backend/.sqlx/query-5cb209fe91eeccdc4c208db7655b6349376f09dad74380302cb815fc2626c69c.json deleted file mode 100644 index d35d072611..0000000000 --- a/backend/.sqlx/query-5cb209fe91eeccdc4c208db7655b6349376f09dad74380302cb815fc2626c69c.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-6be4bf59c404d2f557d1106c48c320bb3eff65255a44bd66799ae14288312ba4.json b/backend/.sqlx/query-6be4bf59c404d2f557d1106c48c320bb3eff65255a44bd66799ae14288312ba4.json new file mode 100644 index 0000000000..0a39db6822 --- /dev/null +++ b/backend/.sqlx/query-6be4bf59c404d2f557d1106c48c320bb3eff65255a44bd66799ae14288312ba4.json @@ -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" +} diff --git a/backend/.sqlx/query-71b410dbc6cdd0252de2f8646395903c9fa60a5435fa73c587da2b913203cf18.json b/backend/.sqlx/query-71b410dbc6cdd0252de2f8646395903c9fa60a5435fa73c587da2b913203cf18.json new file mode 100644 index 0000000000..d74d5e6ba3 --- /dev/null +++ b/backend/.sqlx/query-71b410dbc6cdd0252de2f8646395903c9fa60a5435fa73c587da2b913203cf18.json @@ -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" +} diff --git a/backend/.sqlx/query-c427c9dad3de3acc361c341b359453197d7683bc8fd5a96be462e3493e957cee.json b/backend/.sqlx/query-c427c9dad3de3acc361c341b359453197d7683bc8fd5a96be462e3493e957cee.json new file mode 100644 index 0000000000..f851507f4e --- /dev/null +++ b/backend/.sqlx/query-c427c9dad3de3acc361c341b359453197d7683bc8fd5a96be462e3493e957cee.json @@ -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" +} diff --git a/backend/windmill-api-integration-tests/tests/native_triggers.rs b/backend/windmill-api-integration-tests/tests/native_triggers.rs index faa7efbe74..7ef2818f96 100644 --- a/backend/windmill-api-integration-tests/tests/native_triggers.rs +++ b/backend/windmill-api-integration-tests/tests/native_triggers.rs @@ -158,8 +158,9 @@ fn now_ms() -> i64 { async fn test_connections_are_isolated(db: Pool) -> 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) -> 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"), ) diff --git a/backend/windmill-native-triggers/src/handler.rs b/backend/windmill-native-triggers/src/handler.rs index 49fae69bba..fb6d8016d8 100644 --- a/backend/windmill-native-triggers/src/handler.rs +++ b/backend/windmill-native-triggers/src/handler.rs @@ -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( 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, diff --git a/backend/windmill-native-triggers/src/lib.rs b/backend/windmill-native-triggers/src/lib.rs index 73c42d6a8c..37cd34dd87 100644 --- a/backend/windmill-native-triggers/src/lib.rs +++ b/backend/windmill-native-triggers/src/lib.rs @@ -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( @@ -814,7 +815,7 @@ pub async fn decrypt_oauth_data( 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( "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, diff --git a/backend/windmill-native-triggers/src/workspace_integrations.rs b/backend/windmill-native-triggers/src/workspace_integrations.rs index 240ac31355..3d52c5a15a 100644 --- a/backend/windmill-native-triggers/src/workspace_integrations.rs +++ b/backend/windmill-native-triggers/src/workspace_integrations.rs @@ -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,