kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: refresh custom instance user password if auth failed (#8787)

Browse Source 
* Refresh custom instance user pwd if connection failed

* No longer need to check on startup

* nit: unneeded inner function

* fix
This commit is contained in:
Diego Imbert
2026-04-10 16:26:53 +02:00
committed by GitHub
parent 8957d8f19b
commit 3d43d31aba
12 changed files with 76 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
backend/src/main.rs
View File

@@ -533,51 +533,6 @@ fn print_help() {
println!("- At startup, Windmill logs currently set configuration keys for visibility.");
}
async fn resync_custom_instance_user_pwd_if_needed(db: &Pool<Postgres>) {
use windmill_common::utils::get_custom_pg_instance_password;
use windmill_common::{get_database_url, PgDatabase};
let user_pwd = match get_custom_pg_instance_password(db).await {
Ok(pwd) => pwd,
Err(_) => {
// Setting doesn't exist yet (fresh install or pre-migration), skip check
return;
}
};
let mut pg_creds = match get_database_url().await {
Ok(url) => match PgDatabase::parse_uri(&url.as_str().await) {
Ok(creds) => creds,
Err(e) => {
tracing::warn!("Failed to parse database URL for custom_instance_user check: {e}");
return;
}
},
Err(e) => {
tracing::warn!("Failed to get database URL for custom_instance_user check: {e}");
return;
}
};
pg_creds.user = Some("custom_instance_user".to_string());
pg_creds.password = Some(user_pwd);
match pg_creds.connect().await {
Ok(_) => {
tracing::info!("custom_instance_user password is in sync");
}
Err(e) => {
tracing::warn!("custom_instance_user password is out of sync ({e}), refreshing...");
if let Err(e) = windmill_api_settings::refresh_custom_instance_user_pwd_inner(db).await
{
tracing::error!("Failed to refresh custom_instance_user password: {e}");
} else {
tracing::info!("Successfully refreshed custom_instance_user password");
}
}
}
}
async fn windmill_main() -> anyhow::Result<()> {
let (killpill_tx, mut killpill_rx) = KillpillSender::new(2);
let mut monitor_killpill_rx = killpill_tx.subscribe();
@@ -973,11 +928,6 @@ async fn windmill_main() -> anyhow::Result<()> {
// NOTE: Variable/resource cache initialization moved to API server in windmill-api
// Check if custom_instance_user password is in sync
if server_mode {
resync_custom_instance_user_pwd_if_needed(&db).await;
}
Connection::Sql(db)
};

41
backend/windmill-api-settings/src/lib.rs
View File

@@ -997,49 +997,12 @@ async fn list_custom_instance_pg_databases(
return Ok(Json(result));
}
pub async fn refresh_custom_instance_user_pwd_inner(db: &DB) -> Result<()> {
// 20251208123907_safety_custom_instance_db_user_pwd.up
let query = r#"
DO $$
DECLARE
pwd text;
BEGIN
SELECT gen_random_uuid()::text INTO pwd;
IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'custom_instance_user') THEN
EXECUTE format('ALTER USER custom_instance_user WITH PASSWORD %L', pwd);
RAISE NOTICE 'Updated password for existing user custom_instance_user';
ELSE
EXECUTE format('CREATE USER custom_instance_user WITH PASSWORD %L', pwd);
RAISE NOTICE 'Created new user custom_instance_user';
END IF;
IF NOT EXISTS (SELECT 1 FROM global_settings WHERE name = 'custom_instance_pg_databases') THEN
INSERT INTO global_settings (name, value)
VALUES ('custom_instance_pg_databases', jsonb_build_object(
'user_pwd', pwd::text,
'databases', jsonb_build_object()
));
RAISE NOTICE 'Inserted new global setting for custom_instance_pg_databases';
ELSE
UPDATE global_settings
SET value = jsonb_set(COALESCE(value, '{}'::jsonb), '{user_pwd}', to_jsonb(pwd::text)::jsonb)
WHERE name = 'custom_instance_pg_databases';
RAISE NOTICE 'Updated user_pwd in existing global setting for custom_instance_pg_databases';
END IF;
END
$$;
"#;
sqlx::query(query).execute(db).await?;
Ok(())
}
async fn refresh_custom_instance_user_pwd(
authed: ApiAuthed,
Extension(db): Extension<DB>,
) -> JsonResult<()> {
require_super_admin(&db, &authed.email).await?;
refresh_custom_instance_user_pwd_inner(&db).await?;
windmill_common::utils::refresh_custom_instance_user_pwd(&db).await?;
Ok(Json(()))
}
@@ -1140,7 +1103,7 @@ async fn setup_custom_instance_pg_database_inner(
}
// We have to connect to the newly created database as admin to grant permissions
let (client, connection) = pg_creds.connect().await?;
let (client, connection) = pg_creds.connect(Some(db)).await?;
let join_handle = tokio::spawn(async move { connection.await });
logs.db_connect = "OK".to_string();

8
backend/windmill-api-workspaces/src/workspaces.rs
View File

@@ -1397,7 +1397,7 @@ async fn get_datatable_schema(db: &DB, w_id: &str, datatable_name: &str) -> Resu
.map_err(|e| Error::internal_err(format!("Failed to parse database credentials: {}", e)))?;
// Connect to the datatable database
let (client, connection) = pg_db.connect().await?;
let (client, connection) = pg_db.connect(Some(db)).await?;
// Spawn the connection handler
tokio::spawn(async move {
@@ -1733,7 +1733,7 @@ async fn create_pg_database(
} else {
let source_pg =
resolve_pg_source_checked(&db, &user_db, &authed, &w_id, &req.source).await?;
let (client, connection) = source_pg.connect().await?;
let (client, connection) = source_pg.connect(Some(&db)).await?;
let join_handle = tokio::spawn(async move { connection.await });
let row = client
@@ -1865,7 +1865,7 @@ async fn get_datatable_full_schema(
Json(req): Json<GetDatatableFullSchemaRequest>,
) -> JsonResult<windmill_common::query_builders::FullDatabaseSchema> {
let pg = resolve_pg_source_checked(&db, &user_db, &authed, &w_id, &req.source).await?;
let (client, connection) = pg.connect().await?;
let (client, connection) = pg.connect(Some(&db)).await?;
let join_handle = tokio::spawn(async move { connection.await });
let result = windmill_common::query_builders::pg_get_full_schema(&client)
@@ -4190,7 +4190,7 @@ async fn snapshot_datatable_schema(
let pg = get_datatable_resource_from_db_unchecked(db, parent_w_id, dt_name).await?;
let pg: PgDatabase = serde_json::from_value(pg)
.map_err(|e| Error::internal_err(format!("Failed to parse db credentials: {}", e)))?;
let (client, connection) = pg.connect().await?;
let (client, connection) = pg.connect(Some(db)).await?;
let join_handle = tokio::spawn(async move { connection.await });
let schema = windmill_common::query_builders::pg_get_full_schema(&client)

2
backend/windmill-api-workspaces/src/workspaces_extra.rs
View File

@@ -983,7 +983,7 @@ pub async fn drop_forked_datatable_databases(
continue;
}
match parent_pg.connect().await {
match parent_pg.connect(Some(&db)).await {
Ok((client, connection)) => {
let join_handle = tokio::spawn(async move { connection.await });
if let Err(e) = client

4
backend/windmill-common/src/ai_bedrock.rs
View File

@@ -410,9 +410,7 @@ fn mime_to_document_format(mime_type: &str) -> DocumentFormat {
"application/vnd.openxmlformats-officedocument.wordprocessingml.document" => {
DocumentFormat::Docx
}
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" => {
DocumentFormat::Xlsx
}
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" => DocumentFormat::Xlsx,
_ => DocumentFormat::Pdf,
}
}

4
backend/windmill-common/src/ai_google.rs
View File

@@ -356,9 +356,7 @@ pub fn convert_content_to_gemini_parts(content: &OpenAIContent) -> Vec<GeminiPar
}
ContentPart::File { file } => {
parse_data_url(&file.file_data).map(|(mime_type, data)| {
GeminiPart::InlineData {
inline_data: GeminiInlineData { mime_type, data },
}
GeminiPart::InlineData { inline_data: GeminiInlineData { mime_type, data } }
})
}
// S3Objects are handled by the worker

3
backend/windmill-common/src/flows.rs
View File

@@ -81,8 +81,7 @@ pub async fn get_full_hub_flow_by_path(
.map_err(to_anyhow)
{
Ok(response) => response,
Err(_) if hub_base_url != DEFAULT_HUB_BASE_URL && flow_id < PRIVATE_HUB_MIN_VERSION =>
{
Err(_) if hub_base_url != DEFAULT_HUB_BASE_URL && flow_id < PRIVATE_HUB_MIN_VERSION => {
tracing::info!("Not found on private hub, fallback to default hub for hub flow {path}");
let fallback_url = format!("{DEFAULT_HUB_BASE_URL}/flows/{flow_id}/json");
http_get_from_hub(http_client, &fallback_url, false, None, db)

28
backend/windmill-common/src/lib.rs
View File

@@ -464,6 +464,32 @@ impl PgDatabase {
pub async fn connect(
&self,
main_db: Option<&DB>,
) -> Result<(tokio_postgres::Client, TokioPgConnection), error::Error> {
match self.connect_inner().await {
Ok(result) => Ok(result),
Err(e) => {
let err_str = e.to_string();
if err_str.contains("password authentication failed for user") && err_str.contains("custom_instance_user")
{
if let Some(db) = main_db {
tracing::warn!(
"custom_instance_user password auth failed, refreshing and retrying..."
);
crate::utils::refresh_custom_instance_user_pwd(db).await?;
let new_pwd = crate::utils::get_custom_pg_instance_password(db).await?;
let mut retried = self.clone();
retried.password = Some(new_pwd);
return retried.connect_inner().await;
}
}
Err(e)
}
}
}
async fn connect_inner(
&self,
) -> Result<(tokio_postgres::Client, TokioPgConnection), error::Error> {
use native_tls::{Certificate, TlsConnector};
use postgres_native_tls::MakeTlsConnector;
@@ -757,7 +783,7 @@ pub async fn create_custom_instance_database(
// Grant permissions to custom_instance_user
let wmill_pg_creds = PgDatabase::parse_uri(&get_database_url().await?.as_str().await)?;
let new_pg_creds = PgDatabase { dbname: dbname.to_string(), ..wmill_pg_creds };
let (client, connection) = new_pg_creds.connect().await?;
let (client, connection) = new_pg_creds.connect(Some(db)).await?;
let join_handle = tokio::spawn(async move { connection.await });
if let Err(e) = client

2
backend/windmill-common/src/secret_backend/mod.rs
View File

@@ -17,10 +17,10 @@ pub mod database;
#[cfg(feature = "private")]
pub mod vault_ee;
pub mod vault_oss;
#[cfg(feature = "private")]
pub mod azure_kv_ee;
pub mod azure_kv_oss;
pub mod vault_oss;
#[cfg(feature = "private")]
pub mod aws_sm_ee;

1
backend/windmill-common/src/tracing_init.rs
View File

@@ -273,4 +273,3 @@ where
}
}
}

32
backend/windmill-common/src/utils.rs
View File

@@ -986,6 +986,38 @@ impl<T> ExpiringCacheEntry<T> {
}
}
pub async fn refresh_custom_instance_user_pwd(db: &DB) -> Result<()> {
let query = r#"
DO $$
DECLARE
pwd text;
BEGIN
SELECT gen_random_uuid()::text INTO pwd;
IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'custom_instance_user') THEN
EXECUTE format('ALTER USER custom_instance_user WITH PASSWORD %L', pwd);
ELSE
EXECUTE format('CREATE USER custom_instance_user WITH PASSWORD %L', pwd);
END IF;
IF NOT EXISTS (SELECT 1 FROM global_settings WHERE name = 'custom_instance_pg_databases') THEN
INSERT INTO global_settings (name, value)
VALUES ('custom_instance_pg_databases', jsonb_build_object(
'user_pwd', pwd::text,
'databases', jsonb_build_object()
));
ELSE
UPDATE global_settings
SET value = jsonb_set(COALESCE(value, '{}'::jsonb), '{user_pwd}', to_jsonb(pwd::text)::jsonb)
WHERE name = 'custom_instance_pg_databases';
END IF;
END
$$;
"#;
sqlx::query(query).execute(db).await?;
Ok(())
}
pub async fn get_custom_pg_instance_password(db: &DB) -> Result<String> {
sqlx::query_scalar!(
"SELECT value->>'user_pwd' FROM global_settings WHERE name = 'custom_instance_pg_databases';"

11
backend/windmill-worker/src/pg_executor.rs
View File

@@ -28,7 +28,7 @@ use windmill_common::worker::{
to_raw_value, Connection, SqlResultCollectionStrategy, CLOUD_HOSTED,
};
use windmill_common::workspaces::get_datatable_resource_from_db_unchecked;
use windmill_common::{PgDatabase, PrepareQueryColumnInfo, PrepareQueryResult};
use windmill_common::{PgDatabase, PrepareQueryColumnInfo, PrepareQueryResult, DB};
use windmill_object_store::convert_json_line_stream;
use windmill_parser::{Arg, Typ};
use windmill_parser_sql::{
@@ -67,6 +67,7 @@ pub async fn clear_pg_cache() {
async fn new_pg_connection(
database: &PgDatabase,
_use_iam_auth: bool,
main_db: Option<&DB>,
) -> error::Result<(tokio_postgres::Client, tokio::task::JoinHandle<()>)> {
let (client, connection) = if _use_iam_auth {
#[cfg(all(feature = "enterprise", feature = "private"))]
@@ -80,7 +81,7 @@ async fn new_pg_connection(
));
}
} else {
database.connect().await?
database.connect(main_db).await?
};
let handle = tokio::spawn(async move {
if let Err(e) = connection.await {
@@ -361,18 +362,18 @@ pub async fn do_postgresql(
}
drop(guard);
cached_client = None;
new_client = Some(new_pg_connection(&database, use_iam_auth).await?);
new_client = Some(new_pg_connection(&database, use_iam_auth, conn.as_sql()).await?);
}
} else {
// Release the lock before connecting so the post-query caching
// code can re-acquire it.
drop(guard);
cached_client = None;
new_client = Some(new_pg_connection(&database, use_iam_auth).await?);
new_client = Some(new_pg_connection(&database, use_iam_auth, conn.as_sql()).await?);
}
} else {
cached_client = None;
new_client = Some(new_pg_connection(&database, use_iam_auth).await?);
new_client = Some(new_pg_connection(&database, use_iam_auth, conn.as_sql()).await?);
}
let (sig, typed_schema) = parse_pgsql_sig_with_typed_schema(&query)