Merge remote-tracking branch 'origin/datatable-external-instance-4' into datatable-external-instance-ui

backend/ee-repo-ref.txt

@@ -1 +1 @@
-8efda07e42b4154e50c6340eaf7f1be7d2a0f827
+438e5efda9674b37d25e6c02e286adbf2566d25c

backend/windmill-api-settings/src/lib.rs

@@ -893,6 +893,14 @@ pub async fn set_global_setting_internal(
         )));
     }
 
+    if key == EXTERNAL_INSTANCE_PG_SETTING {
+        return windmill_common::external_instance_pg::write_external_instance_pg_setting(
+            db,
+            Some(&value),
+        )
+        .await;
+    }
+
     run_setting_pre_write_hook(db, &key, &value).await?;
 
     match value {
@@ -954,13 +962,6 @@ async fn run_setting_pre_write_hook(
     value: &serde_json::Value,
 ) -> error::Result<()> {
     match key {
-        EXTERNAL_INSTANCE_PG_SETTING => {
-            windmill_common::external_instance_pg::check_external_instance_pg_write(
-                db,
-                Some(value),
-            )
-            .await?;
-        }
         // The instance AI config is written as an untyped blob through this generic
         // endpoint, so it never passes the typed check the workspace handler applies.
         // Rates that reach a cost total unbounded would make it negative or infinite.
@@ -1281,7 +1282,7 @@ async fn set_instance_config(
     let desired_map = desired.global_settings.to_settings_map();
     if !desired_map.is_empty() {
         let current_map = current.global_settings.to_settings_map();
-        let settings_diff =
+        let mut settings_diff =
             instance_config::diff_global_settings(&current_map, &desired_map, ApplyMode::Merge);
         let ai_config_changed = settings_diff
             .upserts
@@ -1310,16 +1311,15 @@ async fn set_instance_config(
         }
 
         for (key, value) in &settings_diff.upserts {
-            run_setting_pre_write_hook(&db, key, value).await?;
-        }
-        if settings_diff
-            .deletes
-            .iter()
-            .any(|k| k == EXTERNAL_INSTANCE_PG_SETTING)
-        {
-            windmill_common::external_instance_pg::check_external_instance_pg_write(&db, None)
-                .await?;
+            if key != EXTERNAL_INSTANCE_PG_SETTING {
+                run_setting_pre_write_hook(&db, key, value).await?;
+            }
         }
+        windmill_common::external_instance_pg::write_external_instance_pg_from_diff(
+            &db,
+            &mut settings_diff,
+        )
+        .await?;
 
         instance_config::apply_settings_diff(&db, &settings_diff)
             .await

backend/windmill-api-workspaces/src/workspaces.rs

@@ -3373,7 +3373,7 @@ fn apply_pg_tls_env(
                 cmd.env("PGSSLROOTCERT", "system");
             }
             Some("verify-ca") => {
-                if let Some(bundle) = system_ca_bundle() {
+                if let Some(bundle) = windmill_common::system_ca_bundle() {
                     cmd.env("PGSSLROOTCERT", bundle);
                 }
             }
@@ -3383,21 +3383,6 @@ fn apply_pg_tls_env(
     Ok(None)
 }
 
-fn system_ca_bundle() -> Option<std::path::PathBuf> {
-    std::env::var_os("SSL_CERT_FILE")
-        .map(std::path::PathBuf::from)
-        .into_iter()
-        .chain(
-            [
-                "/etc/ssl/certs/ca-certificates.crt",
-                "/etc/pki/tls/certs/ca-bundle.crt",
-                "/etc/ssl/cert.pem",
-                "/etc/ssl/ca-bundle.pem",
-            ]
-            .map(std::path::PathBuf::from),
-        )
-        .find(|path| path.is_file())
-}
 
 #[cfg(test)]
 mod pg_tls_env_tests {
@@ -8352,6 +8337,13 @@ async fn apply_forked_datatable(
         })?,
     };
 
+    if database.resource_type == DataTableCatalogResourceType::ExternalInstance {
+        windmill_common::external_instance_pg::ensure_external_instance_database_registered(
+            tx,
+            &fdt.new_dbname,
+        )
+        .await?;
+    }
     if database.resource_type.is_windmill_managed() {
         // The whole `database` object, not just its `resource_path`: a pointer entry has none to
         // patch. `reference` goes with it — exactly one of the two may be set. The copy was created

backend/windmill-common/src/external_instance_pg.rs

@@ -302,23 +302,73 @@ pub async fn ensure_external_instance_database_registered(
     )))
 }
 
-/// Check a write to [`EXTERNAL_INSTANCE_PG_SETTING`] before it happens: `None`, null or an empty
-/// string unsets it. Every writer of global settings calls this, the per-key and bulk endpoints
-/// as well as the declarative sync.
-pub async fn check_external_instance_pg_write(
+/// Write [`EXTERNAL_INSTANCE_PG_SETTING`]: `None`, null or an empty string unsets it. Every writer
+/// of global settings goes through this for that key — the per-key and bulk endpoints as well as
+/// the declarative sync — instead of writing the row itself.
+///
+/// The checks and the write share one transaction holding [`lock_external_instance_pg_state`]. A
+/// check taken outside it could pass while a database create still reads the old cluster, which
+/// would then register a database there after the setting names another one.
+///
+/// Authorization: checks nothing. Callers MUST be superadmin.
+pub async fn write_external_instance_pg_setting(
     db: &DB,
     value: Option<&serde_json::Value>,
 ) -> Result<()> {
+    let value = match value {
+        None | Some(serde_json::Value::Null) => None,
+        Some(serde_json::Value::String(s)) if s.trim().is_empty() => None,
+        Some(value) => Some(value),
+    };
+    let mut tx = db.begin().await?;
+    lock_external_instance_pg_state(&mut tx).await?;
     match value {
-        None | Some(serde_json::Value::Null) => ensure_external_instance_pg_removable(db).await,
-        Some(serde_json::Value::String(s)) if s.trim().is_empty() => {
-            ensure_external_instance_pg_removable(db).await
+        None => {
+            ensure_external_instance_pg_removable(db).await?;
+            sqlx::query("DELETE FROM global_settings WHERE name = $1")
+                .bind(EXTERNAL_INSTANCE_PG_SETTING)
+                .execute(&mut *tx)
+                .await?;
         }
         Some(value) => {
             crate::external_instance_pg_oss::validate_external_instance_pg_setting(value)?;
-            ensure_external_instance_pg_not_repointed(db, value).await
+            ensure_external_instance_pg_not_repointed(db, value).await?;
+            sqlx::query(
+                "INSERT INTO global_settings (name, value) VALUES ($1, $2)
+                 ON CONFLICT (name) DO UPDATE SET value = EXCLUDED.value, updated_at = now()",
+            )
+            .bind(EXTERNAL_INSTANCE_PG_SETTING)
+            .bind(value)
+            .execute(&mut *tx)
+            .await?;
         }
     }
+    tx.commit().await?;
+    tracing::info!(
+        "{} global setting {EXTERNAL_INSTANCE_PG_SETTING}",
+        if value.is_some() { "Set" } else { "Unset" }
+    );
+    Ok(())
+}
+
+/// [`write_external_instance_pg_setting`] for a settings diff: writes the key if the diff touches
+/// it, and takes it out of the diff so the generic apply does not write it again.
+pub async fn write_external_instance_pg_from_diff(
+    db: &DB,
+    diff: &mut crate::instance_config::SettingsDiff,
+) -> Result<()> {
+    if let Some(value) = diff.upserts.remove(EXTERNAL_INSTANCE_PG_SETTING) {
+        write_external_instance_pg_setting(db, Some(&value)).await?;
+    }
+    if let Some(i) = diff
+        .deletes
+        .iter()
+        .position(|k| k == EXTERNAL_INSTANCE_PG_SETTING)
+    {
+        diff.deletes.remove(i);
+        write_external_instance_pg_setting(db, None).await?;
+    }
+    Ok(())
 }
 
 /// Refuse pointing the setting at another host or port while databases or data table roles live on

backend/windmill-common/src/instance_config.rs

@@ -1395,14 +1395,8 @@ pub async fn sync_global_settings_declarative(
     crate::global_settings::parse_allowed_origins_setting(desired.get(origins_key))
         .map_err(|e| anyhow::anyhow!("{origins_key}: {e}"))?;
 
-    let diff = diff_global_settings(current, desired, ApplyMode::Replace);
-    let external_pg_key = crate::global_settings::EXTERNAL_INSTANCE_PG_SETTING;
-    if diff.deletes.iter().any(|k| k == external_pg_key) {
-        crate::external_instance_pg::check_external_instance_pg_write(db, None).await?;
-    }
-    if let Some(value) = diff.upserts.get(external_pg_key) {
-        crate::external_instance_pg::check_external_instance_pg_write(db, Some(value)).await?;
-    }
+    let mut diff = diff_global_settings(current, desired, ApplyMode::Replace);
+    crate::external_instance_pg::write_external_instance_pg_from_diff(db, &mut diff).await?;
     apply_settings_diff(db, &diff).await?;
 
     Ok(())

backend/windmill-common/src/lib.rs

@@ -1664,6 +1664,24 @@ pub async fn create_custom_instance_database(
     Ok(())
 }
 
+/// The system's CA bundle file, for libpq clients that cannot take `sslrootcert=system`: that value
+/// needs libpq 16, and verify-full only.
+pub fn system_ca_bundle() -> Option<std::path::PathBuf> {
+    std::env::var_os("SSL_CERT_FILE")
+        .map(std::path::PathBuf::from)
+        .into_iter()
+        .chain(
+            [
+                "/etc/ssl/certs/ca-certificates.crt",
+                "/etc/pki/tls/certs/ca-bundle.crt",
+                "/etc/ssl/cert.pem",
+                "/etc/ssl/ca-bundle.pem",
+            ]
+            .map(std::path::PathBuf::from),
+        )
+        .find(|path| path.is_file())
+}
+
 /// Connection options parsed from a database URL.
 ///
 /// The only place a database URL becomes `PgConnectOptions`. Providers that mint the password

backend/windmill-common/src/workspaces.rs

@@ -3018,6 +3018,14 @@ async fn register_fork_ducklake_namespace(
     {
         return Ok(());
     }
+    let mut tx = db.begin().await?;
+    // A row naming an external database counts as a use of it. Written under the lock a drop takes,
+    // and only while the database is still registered, so a drop cannot slip in between the
+    // settings this attach resolved and the row that protects the database.
+    if let Some(dbname) = catalog.strip_prefix("external_instance:") {
+        crate::external_instance_pg::ensure_external_instance_database_registered(&mut tx, dbname)
+            .await?;
+    }
     sqlx::query!(
         "INSERT INTO fork_ducklake_namespace
            (workspace_id, ducklake_name, metadata_schema, catalog, storage, storage_ref, data_path)
@@ -3032,9 +3040,10 @@ async fn register_fork_ducklake_namespace(
         &storage_ref,
         data_path,
     )
-    .execute(db)
+    .execute(&mut *tx)
     .await
     .map_err(|e| Error::internal_err(format!("registering fork ducklake namespace: {e:#}")))?;
+    tx.commit().await?;
     let mut locations = FORK_DUCKLAKE_REGISTERED
         .get(w_id)
         .filter(|(_, exp)| *exp > now)

backend/windmill-worker/src/duckdb_executor.rs

@@ -2240,11 +2240,64 @@ fn parse_attach_db_resource<'a>(query: &'a str) -> Option<ParsedAttachDbResource
     None
 }
 
+/// The verification a DuckDB postgres attach keeps, as its libpq `sslmode` and `sslrootcert`.
+///
+/// Attaches have always turned verify-ca and verify-full into `require`, which resources rely on.
+/// A connection that explicitly refuses invalid certificates — the external instance cluster's —
+/// keeps its mode instead: under `require` its shared password would go to whichever server
+/// answers. DuckDB's libpq takes one root file, so it gets the system bundle plus the configured
+/// certificate.
+fn pg_attach_verification(res: &PgDatabase) -> Result<Option<(&str, std::path::PathBuf)>> {
+    let mode = match res.sslmode.as_deref() {
+        Some(mode @ ("verify-ca" | "verify-full")) if res.accept_invalid_certs == Some(false) => mode,
+        _ => return Ok(None),
+    };
+    let bundle = windmill_common::system_ca_bundle()
+        .map(std::fs::read_to_string)
+        .transpose()
+        .map_err(|e| Error::ExecutionErr(format!("Failed to read the system CA bundle: {e}")))?
+        .unwrap_or_default();
+    let pem = res.root_certificate_pem.as_deref().unwrap_or_default();
+    if bundle.is_empty() && pem.is_empty() {
+        return Err(Error::ExecutionErr(format!(
+            "sslmode {mode} needs a root certificate, and this worker has no system CA bundle"
+        )));
+    }
+    let roots = format!("{bundle}\n{pem}\n");
+    use sha2::Digest;
+    let path = std::env::temp_dir().join(format!(
+        "windmill-pg-roots-{}.pem",
+        hex::encode(&sha2::Sha256::digest(roots.as_bytes())[..8])
+    ));
+    if !path.is_file() {
+        // Renamed into place: a job attaching concurrently must never read a half-written file.
+        let partial = path.with_extension(format!("{}.partial", Uuid::new_v4()));
+        std::fs::write(&partial, &roots)
+            .and_then(|()| std::fs::rename(&partial, &path))
+            .map_err(|e| Error::ExecutionErr(format!("Failed to write root certificates: {e}")))?;
+    }
+    Ok(Some((mode, path)))
+}
+
+fn pg_attach_uri(res: &PgDatabase) -> Result<String> {
+    let uri = res.to_uri();
+    let Some((mode, roots)) = pg_attach_verification(res)? else {
+        return Ok(uri);
+    };
+    let base = uri.strip_suffix("?sslmode=require").ok_or_else(|| {
+        Error::internal_err("unexpected sslmode in a postgres connection URI".to_string())
+    })?;
+    Ok(format!(
+        "{base}?sslmode={mode}&sslrootcert={}",
+        urlencoding::encode(&roots.to_string_lossy())
+    ))
+}
+
 fn format_attach_db_conn_str(db_resource: Value, db_type: &str) -> Result<String> {
     let s = match db_type.to_lowercase().as_str() {
         "postgres" | "postgresql" => {
             let res: PgDatabase = serde_json::from_value(db_resource)?;
-            res.to_uri()
+            pg_attach_uri(&res)?
         }
         #[cfg(feature = "mysql")]
         "mysql" => {
@@ -2703,10 +2756,21 @@ fn pg_secret_attach_statements(db_resource: Value, alias_name: &str) -> Result<V
     let esc = |s: &str| s.replace('\'', "''");
     // The postgres secret type has no sslmode parameter, so it goes in the ATTACH
     // string; only the libpq values PgDatabase::to_uri collapses to are forwarded.
-    let sslmode = match res.sslmode.as_deref() {
-        Some("disable") => "disable",
-        Some("require") | Some("verify-ca") | Some("verify-full") => "require",
-        _ => "prefer",
+    let sslmode = match pg_attach_verification(&res)? {
+        // A libpq keyword/value string: the path is quoted for libpq, then for the DuckDB literal.
+        Some((mode, roots)) => format!(
+            "{mode} sslrootcert=''{}''",
+            roots
+                .to_string_lossy()
+                .replace('\\', "\\\\")
+                .replace('\'', "\\''")
+        ),
+        None => match res.sslmode.as_deref() {
+            Some("disable") => "disable",
+            Some("require") | Some("verify-ca") | Some("verify-full") => "require",
+            _ => "prefer",
+        }
+        .to_string(),
     };
     let secret_name = datatable_secret_name(alias_name);
     Ok(vec![
@@ -2794,6 +2858,36 @@ pub struct Arg {
 mod tests {
     use super::*;
 
+    #[test]
+    fn pg_attach_keeps_verification_only_when_required() {
+        let pg = |sslmode: &str, accept_invalid_certs: Option<bool>| PgDatabase {
+            host: "db.internal".to_string(),
+            user: Some("custom_instance_user".to_string()),
+            password: Some("pw".to_string()),
+            port: None,
+            sslmode: Some(sslmode.to_string()),
+            dbname: "dt".to_string(),
+            root_certificate_pem: Some("-----BEGIN CERTIFICATE-----test".to_string()),
+            accept_invalid_certs,
+            use_iam_auth: None,
+            region: None,
+        };
+        let uri = pg_attach_uri(&pg("verify-full", Some(false))).unwrap();
+        assert!(uri.contains("?sslmode=verify-full&sslrootcert="), "{uri}");
+        let root = urlencoding::decode(uri.split("sslrootcert=").nth(1).unwrap()).unwrap();
+        let roots = std::fs::read_to_string(root.as_ref()).unwrap();
+        assert!(roots.contains("-----BEGIN CERTIFICATE-----test"));
+        let external = serde_json::to_value(pg("verify-full", Some(false))).unwrap();
+        let attach = &pg_secret_attach_statements(external, "dt").unwrap()[3];
+        assert!(
+            attach.starts_with(&format!("ATTACH 'sslmode=verify-full sslrootcert=''{}''", root)),
+            "{attach}"
+        );
+        // A resource that never opted in keeps the historical downgrade.
+        assert!(pg_attach_uri(&pg("verify-full", None)).unwrap().ends_with("?sslmode=require"));
+        assert!(pg_attach_uri(&pg("require", Some(false))).unwrap().ends_with("?sslmode=require"));
+    }
+
     #[test]
     fn attach_datatable_parses_name_and_role() {
         let reference_of = |q: &str| parse_attach_datatable(q).unwrap().reference;