diff --git a/backend/.sqlx/query-bc87664b87a84e1033589f729148961cfbd0171f835faa756762919c021bfd2b.json b/backend/.sqlx/query-bc87664b87a84e1033589f729148961cfbd0171f835faa756762919c021bfd2b.json deleted file mode 100644 index b452b6fc16..0000000000 --- a/backend/.sqlx/query-bc87664b87a84e1033589f729148961cfbd0171f835faa756762919c021bfd2b.json +++ /dev/null @@ -1,24 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT workspace_id || '/' || path AS \"stream!\" FROM postgres_trigger\n WHERE workspace_id = $1 AND mode <> 'disabled'::TRIGGER_MODE\n AND (postgres_resource_path = $2 OR starts_with(postgres_resource_path, $3))\n UNION ALL\n SELECT workspace_id || '/' || path || ' (capture)' FROM capture_config\n WHERE workspace_id = $1 AND trigger_kind = 'postgres'\n AND last_client_ping > now() - interval '10 seconds'\n AND (trigger_config->>'postgres_resource_path' = $2\n OR starts_with(trigger_config->>'postgres_resource_path', $3))", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "stream!", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Text", - "Text", - "Text" - ] - }, - "nullable": [ - null - ] - }, - "hash": "bc87664b87a84e1033589f729148961cfbd0171f835faa756762919c021bfd2b" -} diff --git a/backend/windmill-api-integration-tests/tests/datatable_roles.rs b/backend/windmill-api-integration-tests/tests/datatable_roles.rs index 9179095c6a..719b8e42db 100644 --- a/backend/windmill-api-integration-tests/tests/datatable_roles.rs +++ b/backend/windmill-api-integration-tests/tests/datatable_roles.rs @@ -786,8 +786,27 @@ async fn roles_cannot_be_turned_on_while_a_trigger_streams_the_data_table( "the refusal does not name the trigger to disable" ); + // Disabled, but its listener pinged just now and stops only at its next heartbeat. sqlx::query( - "UPDATE postgres_trigger SET mode = 'disabled' WHERE path = 'u/test-user-2/fork_stream'", + "UPDATE postgres_trigger SET mode = 'disabled', server_id = NULL, last_server_ping = now() + WHERE path = 'u/test-user-2/fork_stream'", + ) + .execute(&db) + .await?; + let resp = authed(client().post(&url), "SECRET_TOKEN") + .json(&turn_on) + .send() + .await?; + assert_eq!( + resp.status(), + 400, + "roles went on while a disabled trigger's listener was still attached: {}", + resp.text().await? + ); + + sqlx::query( + "UPDATE postgres_trigger SET last_server_ping = now() - interval '20 seconds' + WHERE path = 'u/test-user-2/fork_stream'", ) .execute(&db) .await?; @@ -852,3 +871,36 @@ async fn roles_going_on_wait_for_a_trigger_being_enabled(db: Pool) -> ); Ok(()) } + +#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))] +async fn a_stored_name_containing_a_question_mark_resolves_as_itself( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + // Names could contain `?` before they were restricted, and such an entry is still stored. + sqlx::query( + "UPDATE workspace_settings + SET datatable = jsonb_set(datatable, '{datatables,legacy?dt}', datatable->'datatables'->'main') + WHERE workspace_id = 'test-workspace'", + ) + .execute(&db) + .await?; + + let resolve = |reference: &'static str| { + let db = db.clone(); + async move { + windmill_common::workspaces::parse_datatable_ref_for(&db, "test-workspace", reference) + .await + } + }; + assert_eq!(resolve("legacy?dt").await?, ("legacy?dt".to_string(), None)); + assert_eq!( + resolve("main?role=analytics").await?, + ("main".to_string(), Some("analytics".to_string())) + ); + assert!( + resolve("main?dt").await.is_err(), + "an unknown parameter was ignored" + ); + Ok(()) +} diff --git a/backend/windmill-api-workspaces/src/datatable_permissions.rs b/backend/windmill-api-workspaces/src/datatable_permissions.rs index 4014e81f1e..f774448d64 100644 --- a/backend/windmill-api-workspaces/src/datatable_permissions.rs +++ b/backend/windmill-api-workspaces/src/datatable_permissions.rs @@ -481,23 +481,28 @@ async fn ensure_no_streams_reaching(db: &DB, governing: &GoverningDatatable) -> for (w_id, name) in reached { let reference = format!("datatable://{name}"); let with_query = format!("{reference}?"); - // A suspended trigger keeps its listener, so only a disabled one is not streaming; a - // capture streams for as long as its client keeps pinging. + // A suspended trigger keeps its listener, and a capture streams while its client pings. A + // listener also outlives its trigger being disabled, or its capture's client going quiet, + // until its next heartbeat notices; one that pinged within the 15 seconds a server holds a + // listener for may still be dispatching. streams.extend( - sqlx::query_scalar!( - r#"SELECT workspace_id || '/' || path AS "stream!" FROM postgres_trigger - WHERE workspace_id = $1 AND mode <> 'disabled'::TRIGGER_MODE + sqlx::query_scalar::<_, String>( + r#"SELECT workspace_id || '/' || path FROM postgres_trigger + WHERE workspace_id = $1 + AND (mode <> 'disabled'::TRIGGER_MODE + OR last_server_ping > now() - interval '15 seconds') AND (postgres_resource_path = $2 OR starts_with(postgres_resource_path, $3)) UNION ALL SELECT workspace_id || '/' || path || ' (capture)' FROM capture_config WHERE workspace_id = $1 AND trigger_kind = 'postgres' - AND last_client_ping > now() - interval '10 seconds' + AND (last_client_ping > now() - interval '10 seconds' + OR last_server_ping > now() - interval '15 seconds') AND (trigger_config->>'postgres_resource_path' = $2 OR starts_with(trigger_config->>'postgres_resource_path', $3))"#, - &w_id, - &reference, - &with_query, ) + .bind(&w_id) + .bind(&reference) + .bind(&with_query) .fetch_all(db) .await?, ); @@ -505,8 +510,8 @@ async fn ensure_no_streams_reaching(db: &DB, governing: &GoverningDatatable) -> if !streams.is_empty() { return Err(Error::BadRequest(format!( "Data table '{}' cannot be put under roles while a Postgres trigger or capture streams \ - it: a replication stream reads every row whatever the roles grant. Disable them \ - first: {}", + it: a replication stream reads every row whatever the roles grant. Disable them, then \ + allow their listeners up to 15 seconds to stop: {}", governing.name, streams.join(", ") ))); diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index ab81e50b9d..e5a1690c7b 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -45,12 +45,12 @@ use windmill_common::workspaces::GitRepositorySettings; #[cfg(feature = "enterprise")] use windmill_common::workspaces::WorkspaceDeploymentUISettings; use windmill_common::workspaces::{ - check_deploy_rules, check_user_against_rule, datatable_ref_name, - get_datatable_resource_from_db, get_datatable_resource_from_db_unchecked, - resolve_governing_datatable, validate_dev_workspace_id, validate_fork_workspace_id, - validate_workspace_name, DataTable, DataTableCatalogResourceType, DataTableForkBehavior, - DatatableAccess, GoverningDatatable, ProtectionRuleKind, ProtectionRules, ProtectionRuleset, - RuleCheckResult, WorkspaceGitSyncSettings, DEV_WORKSPACE_LOCK_RULE_NAME, + check_deploy_rules, check_user_against_rule, get_datatable_resource_from_db, + get_datatable_resource_from_db_unchecked, parse_datatable_ref_for, resolve_governing_datatable, + validate_dev_workspace_id, validate_fork_workspace_id, validate_workspace_name, DataTable, + DataTableCatalogResourceType, DataTableForkBehavior, DatatableAccess, GoverningDatatable, + ProtectionRuleKind, ProtectionRules, ProtectionRuleset, RuleCheckResult, + WorkspaceGitSyncSettings, DEV_WORKSPACE_LOCK_RULE_NAME, }; use windmill_common::workspaces::{Ducklake, DucklakeCatalogResourceType}; use windmill_common::PgDatabase; @@ -3313,8 +3313,8 @@ async fn create_pg_database( // database that no data table entry names. Refuse here too, so the clone stops before one // exists rather than leaving an empty registered `wm_fork_…` behind. if let Some(reference) = req.source.strip_prefix("datatable://") { - let name = datatable_ref_name(reference); - ensure_datatable_is_clonable(&db, &w_id, name).await?; + let (name, _) = parse_datatable_ref_for(&db, &w_id, reference).await?; + ensure_datatable_is_clonable(&db, &w_id, &name).await?; } // Non-superadmin: restrict dbname to wm_fork_ prefix @@ -3457,8 +3457,8 @@ async fn import_pg_database( } if let Some(reference) = req.source.strip_prefix("datatable://") { - let name = datatable_ref_name(reference); - ensure_datatable_is_clonable(&db, &w_id, name).await?; + let (name, _) = parse_datatable_ref_for(&db, &w_id, reference).await?; + ensure_datatable_is_clonable(&db, &w_id, &name).await?; } if req.fork_behavior == DataTableForkBehavior::SchemaAndData { diff --git a/backend/windmill-common/src/workspaces.rs b/backend/windmill-common/src/workspaces.rs index bddf85f0c1..0e9a7ed3c9 100644 --- a/backend/windmill-common/src/workspaces.rs +++ b/backend/windmill-common/src/workspaces.rs @@ -2106,13 +2106,30 @@ pub fn strip_datatable_permissions( Some(datatable) } -/// The data table a `datatable://` reference names, ignoring its query string. For callers that -/// only need to find the entry; use [`parse_datatable_ref`] wherever the role is acted on. -pub fn datatable_ref_name(reference: &str) -> &str { - reference - .split_once('?') - .map(|(name, _)| name) - .unwrap_or(reference) +/// As [`parse_datatable_ref`], except that an entry whose stored name itself contains `?` — which +/// names could before they were restricted — resolves by that exact name, without a role. It is +/// looked up first, so `sales?role=x` never reaches a different entry than the one stored so. +pub async fn parse_datatable_ref_for( + db: &DB, + w_id: &str, + reference: &str, +) -> Result<(String, Option)> { + if reference.contains('?') { + let exists = sqlx::query_scalar::<_, Option>( + "SELECT (datatable->'datatables') ? $2 FROM workspace_settings WHERE workspace_id = $1", + ) + .bind(w_id) + .bind(reference) + .fetch_optional(db) + .await? + .flatten() + .unwrap_or(false); + if exists { + return Ok((reference.to_string(), None)); + } + } + let (name, role) = parse_datatable_ref(reference)?; + Ok((name.to_string(), role.map(str::to_string))) } /// Split a `datatable://` reference into its name and the role its query string names. @@ -3431,9 +3448,6 @@ mod tests { "silently ignored: {malformed}" ); } - - // The name-only helper stays lenient — it is used where the role is never acted on. - assert_eq!(datatable_ref_name("sales?role="), "sales"); } #[test] diff --git a/backend/windmill-worker/src/duckdb_executor.rs b/backend/windmill-worker/src/duckdb_executor.rs index 764859d29b..bc5660b5fb 100644 --- a/backend/windmill-worker/src/duckdb_executor.rs +++ b/backend/windmill-worker/src/duckdb_executor.rs @@ -2602,31 +2602,28 @@ fn fork_defer_statements( } struct AttachedDatatable<'a> { - name: &'a str, - role: Option<&'a str>, + /// The data table reference, query string included; a bare `datatable` is `main`. + reference: String, alias: &'a str, } /// `ATTACH 'datatable[://][?role=]' AS `. A bare `datatable` names the default -/// data table, so the role query string has to be accepted with and without an explicit name. -fn parse_attach_datatable(query: &str) -> Result>> { +/// data table, so the role query string has to be accepted with and without an explicit name. The +/// reference is split only once the workspace can be read, because a stored name may contain `?`. +fn parse_attach_datatable(query: &str) -> Option> { lazy_static::lazy_static! { static ref RE: regex::Regex = regex::Regex::new( - r"(?i)ATTACH\s*'datatable(://[^'?:]+)?(\?[^':]*)?'\s*AS\s+([^ ;]+)" + r"(?i)ATTACH\s*'datatable(://[^':]+|\?[^':]*)?'\s*AS\s+([^ ;]+)" ).unwrap(); } - let Some(cap) = RE.captures(query) else { - return Ok(None); + let cap = RE.captures(query)?; + let reference = match cap.get(1).map(|m| m.as_str()) { + Some(named) if named.starts_with("://") => named[3..].to_string(), + Some(query) => format!("main{query}"), + None => "main".to_string(), }; - let name = cap.get(1).map(|m| &m.as_str()[3..]).unwrap_or("main"); - // A query string that does not parse is refused rather than dropped: attaching under the - // default role when the statement asked for another one is the failure this guards. - let role = match cap.get(2) { - Some(m) => windmill_common::workspaces::parse_datatable_ref(m.as_str())?.1, - None => None, - }; - let alias = cap.get(3).map(|m| m.as_str()).unwrap_or(""); - Ok(Some(AttachedDatatable { name, role, alias })) + let alias = cap.get(2).map(|m| m.as_str()).unwrap_or(""); + Some(AttachedDatatable { reference, alias }) } async fn transform_attach_datatable( @@ -2635,27 +2632,31 @@ async fn transform_attach_datatable( hidden_passwords: &mut Arc>>, job: &MiniPulledJob, ) -> Result>> { - let Some(attached) = parse_attach_datatable(query)? else { + let Some(attached) = parse_attach_datatable(query) else { return Ok(None); }; + // A query string that does not parse is refused rather than dropped: attaching under the + // default role when the statement asked for another one is the failure this guards. let db_resource = match conn { Connection::Http(client) => { - get_datatable_resource_from_agent_http( - client, - attached.name, - &job.workspace_id, - attached.role, - &job.id, - ) - .await? + let (name, role) = + windmill_common::workspaces::parse_datatable_ref(&attached.reference)?; + get_datatable_resource_from_agent_http(client, name, &job.workspace_id, role, &job.id) + .await? } Connection::Sql(db) => { + let (name, role) = windmill_common::workspaces::parse_datatable_ref_for( + db, + &job.workspace_id, + &attached.reference, + ) + .await?; get_datatable_resource_from_db( db, &job.workspace_id, - attached.name, - attached.role, + &name, + role.as_deref(), DatatableAccess::PermissionedAs { permissioned_as: &job.permissioned_as, email: &job.permissioned_as_email, @@ -2792,45 +2793,40 @@ mod tests { #[test] fn attach_datatable_parses_name_and_role() { - let named = parse_attach_datatable("ATTACH 'datatable://sales?role=analytics' AS dt") - .unwrap() - .unwrap(); + let reference_of = |q: &str| parse_attach_datatable(q).unwrap().reference; + let named = + parse_attach_datatable("ATTACH 'datatable://sales?role=analytics' AS dt").unwrap(); assert_eq!( - (named.name, named.role, named.alias), - ("sales", Some("analytics"), "dt") + (named.reference.as_str(), named.alias), + ("sales?role=analytics", "dt") ); // A bare `datatable` is the default one, and still takes a role. - let default = parse_attach_datatable("ATTACH 'datatable?role=analytics' AS dt") - .unwrap() - .unwrap(); - assert_eq!((default.name, default.role), ("main", Some("analytics"))); - let no_role = parse_attach_datatable("ATTACH 'datatable://sales' AS dt") - .unwrap() - .unwrap(); - assert_eq!((no_role.name, no_role.role), ("sales", None)); - let bare = parse_attach_datatable("ATTACH 'datatable' AS dt") - .unwrap() - .unwrap(); - assert_eq!((bare.name, bare.role), ("main", None)); - assert!(parse_attach_datatable("SELECT 1").unwrap().is_none()); + assert_eq!( + reference_of("ATTACH 'datatable?role=analytics' AS dt"), + "main?role=analytics" + ); + assert_eq!(reference_of("ATTACH 'datatable://sales' AS dt"), "sales"); + assert_eq!(reference_of("ATTACH 'datatable' AS dt"), "main"); + assert!(parse_attach_datatable("SELECT 1").is_none()); + // A stored name can contain `?`, so that is left to the workspace lookup to split. + assert_eq!(reference_of("ATTACH 'datatable://a?b' AS dt"), "a?b"); - // The key matches case-insensitively, as the `-- role` annotation does. - let cased = parse_attach_datatable("ATTACH 'datatable://sales?Role=analytics' AS dt") - .unwrap() - .unwrap(); - assert_eq!(cased.role, Some("analytics")); - - // A query string that does not parse is refused rather than attached under the default - // role: the statement asked for a specific one. + // The key matches case-insensitively, as the `-- role` annotation does, and a query string + // that does not parse is refused rather than attached under the default role. + let parse = |q: &str| { + windmill_common::workspaces::parse_datatable_ref(&reference_of(q)) + .map(|(name, role)| (name.to_string(), role.map(str::to_string))) + }; + assert_eq!( + parse("ATTACH 'datatable://sales?Role=analytics' AS dt").unwrap(), + ("sales".to_string(), Some("analytics".to_string())) + ); for malformed in [ "ATTACH 'datatable://sales?role=' AS dt", "ATTACH 'datatable://sales?role=an;alytics' AS dt", "ATTACH 'datatable://sales?x=1&role=analytics' AS dt", ] { - assert!( - parse_attach_datatable(malformed).is_err(), - "silently ignored: {malformed}" - ); + assert!(parse(malformed).is_err(), "silently ignored: {malformed}"); } } diff --git a/backend/windmill-worker/src/pg_executor.rs b/backend/windmill-worker/src/pg_executor.rs index 03560690f3..69aa3ee0fb 100644 --- a/backend/windmill-worker/src/pg_executor.rs +++ b/backend/windmill-worker/src/pg_executor.rs @@ -29,7 +29,7 @@ use windmill_common::worker::{ to_raw_value, Connection, SqlAnnotations, SqlResultCollectionStrategy, CLOUD_HOSTED, }; use windmill_common::workspaces::{ - get_datatable_resource_from_db, parse_datatable_ref, DatatableAccess, + get_datatable_resource_from_db, parse_datatable_ref, parse_datatable_ref_for, DatatableAccess, }; use windmill_common::{PgDatabase, PrepareQueryColumnInfo, PrepareQueryResult, DB}; use windmill_parser::{Arg, Typ}; @@ -683,28 +683,29 @@ pub async fn do_postgresql( match pg_args.get("database").cloned() { Some(Value::String(db_str)) if db_str.starts_with("datatable://") => { let reference = db_str.trim_start_matches("datatable://"); - let (db_str, uri_role) = parse_datatable_ref(reference)?; // The annotation wins: a generated query can carry a `?role=` in the reference it // was handed, but only the script's author writes the leading comment block. let annotated = SqlAnnotations::datatable_role(&query)?; - let role = annotated.as_deref().or(uri_role); Some(match conn { Connection::Http(client) => { + let (name, uri_role) = parse_datatable_ref(reference)?; get_datatable_resource_from_agent_http( client, - db_str, + name, &job.workspace_id, - role, + annotated.as_deref().or(uri_role), &job.id, ) .await? } Connection::Sql(db) => { + let (name, uri_role) = + parse_datatable_ref_for(db, &job.workspace_id, reference).await?; get_datatable_resource_from_db( db, &job.workspace_id, - db_str, - role, + &name, + annotated.as_deref().or(uri_role.as_deref()), DatatableAccess::PermissionedAs { permissioned_as: &job.permissioned_as, email: &job.permissioned_as_email,