diff --git a/backend/windmill-api/src/capture.rs b/backend/windmill-api/src/capture.rs index 1eed038c25..9f02727bf0 100644 --- a/backend/windmill-api/src/capture.rs +++ b/backend/windmill-api/src/capture.rs @@ -381,8 +381,7 @@ async fn set_postgres_trigger_config( postgres_config.publication.table_to_track.as_deref(), &postgres_config.publication.transaction_to_track, ) - .await - .map_err(to_anyhow)?; + .await?; tx.commit().await.map_err(to_anyhow)?; diff --git a/backend/windmill-trigger-postgres/src/handler.rs b/backend/windmill-trigger-postgres/src/handler.rs index 92dd2e1e19..230c64c53f 100644 --- a/backend/windmill-trigger-postgres/src/handler.rs +++ b/backend/windmill-trigger-postgres/src/handler.rs @@ -27,9 +27,9 @@ use super::{ check_if_valid_publication_for_postgres_version, create_logical_replication_slot, create_pg_publication, drop_publication, generate_random_string, get_default_pg_connection, mapper::{Mapper, MappingInfo}, - PostgresConfig, PostgresConfigRequest, PostgresPublicationReplication, PostgresTrigger, - PublicationData, Relations, Slot, SlotList, TableToTrack, TemplateScript, TestPostgresConfig, - ERROR_PUBLICATION_NAME_NOT_EXISTS, + validate_publication_row_filters, PostgresConfig, PostgresConfigRequest, + PostgresPublicationReplication, PostgresTrigger, PublicationData, Relations, Slot, SlotList, + TableToTrack, TemplateScript, TestPostgresConfig, ERROR_PUBLICATION_NAME_NOT_EXISTS, }; // Lazy static template cache @@ -698,6 +698,8 @@ pub async fn update_pg_publication( ) -> Result<()> { let quoted_publication_name = quote_identifier(publication_name); let transaction_to_track_as_str = transaction_to_track.iter().join(","); + // Before the branch below drops the publication, so a rejected filter is not a lost publication. + validate_publication_row_filters(pg_connection, table_to_track.as_deref()).await?; match table_to_track { Some(ref relations) if !relations.is_empty() => { // If all_table is None, the publication does not exist yet @@ -748,6 +750,7 @@ pub async fn update_pg_publication( query.push_str(&format!(" ({})", cols)); } + // Raw, and kept in bounds by `validate_publication_row_filters` above. if let Some(where_clause) = &table.where_clause { query.push_str(&format!(" WHERE ({})", where_clause)); } diff --git a/backend/windmill-trigger-postgres/src/lib.rs b/backend/windmill-trigger-postgres/src/lib.rs index 571aac0a66..32ddd13f6c 100644 --- a/backend/windmill-trigger-postgres/src/lib.rs +++ b/backend/windmill-trigger-postgres/src/lib.rs @@ -466,6 +466,54 @@ pub async fn check_if_valid_publication_for_postgres_version( Ok(pg_14) } +/// Wraps `where_clause` exactly the way publication DDL does, then continues past the closing +/// parenthesis on a new line. +/// +/// A filter can only parse here by being a self-contained expression: closing the wrapper early +/// strands the trailing `) IS NOT FALSE`, a line comment cannot reach across the newline to hide +/// it, and a block comment left open never terminates. +fn build_row_filter_probe(schema_name: &str, table_name: &str, where_clause: &str) -> String { + format!( + "SELECT 1 FROM {}.{} WHERE ({}\n) IS NOT FALSE", + quote_identifier(schema_name), + quote_identifier(table_name), + where_clause + ) +} + +/// A publication row filter has no bind parameter, so it is interpolated raw into +/// `... WHERE ()`. Postgres decides where a literal ends from server state and syntax we +/// cannot see from here (`standard_conforming_strings`, `E'…'`, dollar quoting, a `$tag$` read as +/// part of a preceding identifier), so the filter is handed to the server itself to parse rather +/// than lexed here. The probe is only prepared, never executed. +/// +/// Call this before emitting any publication DDL, so a rejected filter leaves the publication as it +/// was. Publication DDL must also keep running through `Client::execute`, whose `Parse` refuses +/// more than one command: under `simple_query` or `batch_execute` a filter could stack statements. +pub(crate) async fn validate_publication_row_filters( + pg_connection: &Client, + table_to_track: Option<&[Relations]>, +) -> Result<()> { + for relation in table_to_track.unwrap_or_default() { + for table in relation.table_to_track.iter() { + let Some(where_clause) = table.where_clause.as_deref() else { + continue; + }; + let probe = + build_row_filter_probe(&relation.schema_name, &table.table_name, where_clause); + pg_connection.prepare(&probe).await.map_err(|e| { + Error::BadRequest(format!( + "Invalid row filter for table {}.{}: {}", + relation.schema_name, + table.table_name, + windmill_common::error::pg_error_message(&e) + )) + })?; + } + } + Ok(()) +} + pub async fn create_pg_publication( pg_connection: &Client, publication_name: &str, @@ -474,6 +522,7 @@ pub async fn create_pg_publication( ) -> Result<()> { let pg_14 = check_if_valid_publication_for_postgres_version(pg_connection, table_to_track).await?; + validate_publication_row_filters(pg_connection, table_to_track).await?; let mut query = String::from("CREATE PUBLICATION "); query.push_str("e_identifier(publication_name)); @@ -508,10 +557,7 @@ pub async fn create_pg_publication( query.push_str(")"); } - // A row filter has no bind parameter, so it goes in raw. Publication - // DDL must therefore keep running through `Client::execute`, whose - // `Parse` refuses more than one command: under `simple_query` or - // `batch_execute` a filter could stack statements. + // Raw, and kept in bounds by `validate_publication_row_filters` above. if let Some(where_clause) = &table.where_clause { query.push_str(" WHERE ("); query.push_str(where_clause); @@ -726,6 +772,17 @@ mod tests { assert_eq!(tt.where_clause, Some("status = 'active'".to_string())); } + #[test] + fn test_row_filter_probe_shields_its_tail_from_a_line_comment() { + let probe = build_row_filter_probe("public", "orders", "status = 'active') --"); + let tail = probe.lines().last().unwrap(); + assert_eq!(tail, ") IS NOT FALSE"); + assert_eq!( + probe, + "SELECT 1 FROM public.orders WHERE (status = 'active') --\n) IS NOT FALSE" + ); + } + #[test] fn test_table_to_track_with_columns() { let tt = TableToTrack::new(