From 166c6e84d3fd9cf4a0b998d6c79e7c60e417424e Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Thu, 29 Oct 2026 17:58:12 +0000 Subject: [PATCH 1/3] Fix SQL quoting --- Cargo.lock | 1 + Cargo.toml | 1 + scripts/test_upgrade_lifecycle.py | 16 ++++ scripts/upgrade_lifecycle.py | 3 +- src/lib.rs | 112 +++++++++++++++++++++++++- src/types.rs | 61 +++++++++++---- tests/e2e/sql/97_literal_quoting.sql | 113 +++++++++++++++++++++++++++ 7 files changed, 288 insertions(+), 19 deletions(-) create mode 100644 tests/e2e/sql/97_literal_quoting.sql diff --git a/Cargo.lock b/Cargo.lock index e821bc7a..623826d4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1936,6 +1936,7 @@ dependencies = [ "percent-encoding", "pgrx", "pgrx-tests", + "postgres-protocol", "reqwest", "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index c6a7eee5..43000eae 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -34,6 +34,7 @@ serde = { version = "1.0", features = ["derive"] } serde_json = { version = "1.0", features = ["preserve_order", "raw_value"] } uuid = { version = "1.0", features = ["v4", "serde"] } percent-encoding = "2.3" +postgres-protocol = "0.6" # duroxide integration duroxide = "=0.1.30" diff --git a/scripts/test_upgrade_lifecycle.py b/scripts/test_upgrade_lifecycle.py index 439414b3..1effe8f9 100644 --- a/scripts/test_upgrade_lifecycle.py +++ b/scripts/test_upgrade_lifecycle.py @@ -44,6 +44,22 @@ def case(shape="seq", waiting=False): } +class LiteralQuotingTests(unittest.TestCase): + def test_quotes_apostrophes_and_backslashes(self): + for value, expected in [ + ("", "''"), + ("plain", "'plain'"), + ("O'Reilly", "'O''Reilly'"), + ("\\", "E'\\\\'"), + ("\\' OR TRUE --", "E'\\\\'' OR TRUE --'"), + (r"\n\x41\u0041", r"E'\\n\\x41\\u0041'"), + ("line one\nline two", "'line one\nline two'"), + ("\u00e9\u4e2d\\'", "E'\u00e9\u4e2d\\\\'''"), + ]: + with self.subTest(value=value): + self.assertEqual(lifecycle.literal(value), expected) + + class ValidationTests(unittest.TestCase): def test_every_shape_passes_before_and_after(self): for shape in lifecycle.FAMILIES: diff --git a/scripts/upgrade_lifecycle.py b/scripts/upgrade_lifecycle.py index 312cf3c5..dc00f7f4 100644 --- a/scripts/upgrade_lifecycle.py +++ b/scripts/upgrade_lifecycle.py @@ -67,7 +67,8 @@ def require_equal(actual, expected): def literal(value): - return "'" + value.replace("'", "''") + "'" + escaped = value.replace("\\", "\\\\").replace("'", "''") + return ("E" if "\\" in value else "") + "'" + escaped + "'" def build(source, output, pg_config, pg_major): diff --git a/src/lib.rs b/src/lib.rs index 617763b2..9c3747a7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -964,6 +964,7 @@ CREATE OPERATOR @> ( mod tests { use crate::Durofut; use pgrx::prelude::*; + use postgres_protocol::escape::escape_literal as sql_literal; // ======================================================================== // Test Helpers for Integration Tests @@ -1094,10 +1095,6 @@ mod tests { }) } - fn sql_literal(s: &str) -> String { - format!("'{}'", s.replace('\'', "''")) - } - fn test_database_connection_string() -> String { use crate::types::{get_host, get_port}; @@ -2645,6 +2642,113 @@ mod tests { assert_eq!(steps, vec![1, 2], "Expected steps [1, 2], got {steps:?}"); } + #[pg_test] + fn test_literal_quoting_prevents_expression_injection() { + use crate::types::{substitute_all, SystemVars}; + use std::collections::HashMap; + + Spi::run("SET LOCAL standard_conforming_strings = off; SET LOCAL backslash_quote = on") + .unwrap(); + let results = HashMap::from([( + "payload".to_string(), + serde_json::json!({"rows": [{"value": "\\' OR true --"}]}).to_string(), + )]); + let query = substitute_all( + "SELECT 'victim' OPERATOR(pg_catalog.=) $payload", + &results, + &HashMap::new(), + &SystemVars { + instance_id: "literal-quoting".to_string(), + label: None, + }, + ) + .unwrap(); + assert_eq!(Spi::get_one::(&query).unwrap(), Some(false)); + } + + #[pg_test] + fn test_literal_quoting_round_trips_all_result_forms() { + use crate::types::{substitute_all, substitute_all_raw, SystemVars}; + use std::collections::HashMap; + + Spi::run( + "CREATE SCHEMA literal_quoting_shadow; + CREATE FUNCTION literal_quoting_shadow.quote_literal(pg_catalog.text) + RETURNS pg_catalog.text LANGUAGE sql AS $$ SELECT 'wrong'::pg_catalog.text $$; + SET LOCAL search_path = literal_quoting_shadow, pg_catalog, pg_temp", + ) + .unwrap(); + let vars = HashMap::new(); + let sys_vars = SystemVars { + instance_id: "literal-quoting".to_string(), + label: None, + }; + let check = |query: &str, result: &str, expected: &str| { + let results = HashMap::from([("value".to_string(), result.to_string())]); + let sql = substitute_all(query, &results, &vars, &sys_vars).unwrap(); + assert_eq!( + Spi::get_one::(&sql).unwrap().as_deref(), + Some(expected), + "{sql}" + ); + }; + + for standard_strings in ["on", "off"] { + for backslash_quote in ["on", "off"] { + Spi::run(&format!( + "SET LOCAL standard_conforming_strings = {standard_strings}; + SET LOCAL backslash_quote = {backslash_quote}" + )) + .unwrap(); + for text in [ + "", + "plain", + "O'Reilly", + "\\", + "ends with \\", + "\\' OR true --", + "\\\\'", + "\\\\\\'", + r"\n\t\r\b\f\v\123\x41\u0041\U00000041", + "line one\nline two\t\r", + "$value {variable} $$ $tag$", + "\u{e9}\u{4e2d}\u{1f642}\\'", + ] { + check("SELECT $value", text, text); + let envelope = serde_json::json!({"body": text}).to_string(); + check("SELECT $value", &envelope, &envelope); + check("SELECT $value.body", &envelope, text); + + for value in [ + serde_json::json!(text), + serde_json::json!({"nested": text}), + serde_json::json!([text]), + ] { + let expected = match &value { + serde_json::Value::String(s) => s.clone(), + other => other.to_string(), + }; + let result = serde_json::json!({"rows": [{"field": value}]}).to_string(); + for query in [ + "SELECT $value", + "SELECT $value?", + "SELECT $value.field", + "SELECT $value.field?", + "SELECT field FROM $value.*", + ] { + check(query, &result, &expected); + } + let results = HashMap::from([("value".to_string(), result)]); + assert_eq!( + substitute_all_raw("$value.field", &results, &vars, &sys_vars).unwrap(), + expected + ); + } + } + } + } + } + #[pg_test] #[ignore = "pgrx doesn't support shared_preload_libraries"] fn test_e2e_variable_substitution() { diff --git a/src/types.rs b/src/types.rs index df5aa6db..8abecf20 100644 --- a/src/types.rs +++ b/src/types.rs @@ -8,6 +8,7 @@ use pgrx::{pg_extern, Spi}; use chrono::{DateTime, Utc}; use cron::Schedule as CronSchedule; use percent_encoding::{utf8_percent_encode, NON_ALPHANUMERIC}; +use postgres_protocol::escape::escape_literal as quote_literal; use serde::{Deserialize, Serialize}; use std::ffi::{CStr, CString}; use std::str::FromStr; @@ -775,8 +776,7 @@ fn format_value(val: &serde_json::Value, for_sql: bool) -> String { match val { serde_json::Value::String(s) => { if for_sql { - let escaped = s.replace('\'', "''"); - format!("'{escaped}'") + quote_literal(s) } else { s.clone() } @@ -785,9 +785,7 @@ fn format_value(val: &serde_json::Value, for_sql: bool) -> String { serde_json::Value::Bool(b) => b.to_string(), _ => { if for_sql { - let s = val.to_string(); - let escaped = s.replace('\'', "''"); - format!("'{escaped}'") + quote_literal(&val.to_string()) } else { val.to_string() } @@ -807,8 +805,7 @@ fn extract_first_column_value( Err(_) => { // Not JSON — return raw value (backward compat for HTTP responses etc.) return Ok(if for_sql { - let escaped = json_str.replace('\'', "''"); - format!("'{escaped}'") + quote_literal(json_str) } else { json_str.to_string() }); @@ -844,8 +841,7 @@ fn extract_first_column_value( Ok(format_value(val, for_sql)) } else if for_sql { - let escaped = json_str.replace('\'', "''"); - Ok(format!("'{escaped}'")) + Ok(quote_literal(json_str)) } else { Ok(json_str.to_string()) } @@ -923,7 +919,7 @@ fn extract_column_value( /// Expand `$name.*` into an inline `VALUES` subquery (SQL) or JSON array (raw). /// -/// Preserve the generated SQL spelling (including `::text`) for replay of recorded +/// Preserve the generated casts (including `::text`) for replay of unaffected /// activity inputs. This SQL executes as the submitting user, not the worker role. fn expand_row_set(name: &str, json_str: &str, for_sql: bool) -> Result { /// Maximum number of rows allowed in `$name.*` expansion to prevent @@ -971,15 +967,13 @@ fn expand_row_set(name: &str, json_str: &str, for_sql: bool) -> Result { - let escaped = s.replace('\'', "''"); - format!("'{escaped}'::text") + format!("{}::text", quote_literal(s)) } Some(serde_json::Value::Number(n)) => n.to_string(), Some(serde_json::Value::Bool(b)) => b.to_string(), Some(serde_json::Value::Null) | None => "NULL".to_string(), Some(other) => { - let escaped = other.to_string().replace('\'', "''"); - format!("'{escaped}'::text") + format!("{}::text", quote_literal(&other.to_string())) } }) .collect(); @@ -3156,6 +3150,45 @@ mod tests { assert_eq!(out, "SELECT 1, 'Bob'"); } + #[test] + fn literal_quoting_preserves_unaffected_sql() { + let results = make_results(&[( + "row", + r#"{"rows":[{"empty":"","name":"O'Reilly","doc":{"key":"value"}}]}"#, + )]); + assert_eq!( + substitute_all( + "SELECT $row, $row.name, $row.doc FROM $row.*", + &results, + &empty_vars(), + &sys_vars() + ) + .unwrap(), + r#"SELECT '', 'O''Reilly', '{"key":"value"}' FROM (VALUES (''::text,'O''Reilly'::text,'{"key":"value"}'::text)) AS "row"("empty", "name", "doc")"# + ); + } + + #[test] + fn literal_quoting_is_deterministic_and_does_not_rescan() { + let value = serde_json::json!({"rows": [{"field": "\\' $b {var} \u{1f642}"}]}).to_string(); + let other = r#"{"rows":[{"field":"replacement"}]}"#; + let forward = make_results(&[("a", &value), ("b", other)]); + let reverse = make_results(&[("b", other), ("a", &value)]); + let vars = make_results(&[("var", "replacement")]); + for template in [ + "SELECT $a, $b", + "SELECT $a.field, $b.field", + "SELECT * FROM $a.*", + ] { + let sql = substitute_all(template, &forward, &vars, &sys_vars()).unwrap(); + assert_eq!( + sql, + substitute_all(template, &reverse, &vars, &sys_vars()).unwrap() + ); + assert!(sql.contains("$b {var}")); + } + } + #[test] fn test_substitution_order() { // Ensure $doc.id doesn't partially match as $doc first diff --git a/tests/e2e/sql/97_literal_quoting.sql b/tests/e2e/sql/97_literal_quoting.sql new file mode 100644 index 00000000..260d5e06 --- /dev/null +++ b/tests/e2e/sql/97_literal_quoting.sql @@ -0,0 +1,113 @@ +-- Copyright (c) Microsoft Corporation. +-- Licensed under the PostgreSQL License. + +DO $setup$ +BEGIN + IF NOT EXISTS (SELECT FROM pg_roles WHERE rolname = 'df_literal_quoting_user') THEN + CREATE ROLE df_literal_quoting_user LOGIN; + END IF; +END $setup$; + +-- Activities use separate connections, so a caller-session SET cannot test this. +ALTER ROLE df_literal_quoting_user SET standard_conforming_strings = off; +ALTER ROLE df_literal_quoting_user SET backslash_quote = on; +ALTER ROLE df_literal_quoting_user SET search_path = test_literal_quoting, pg_catalog, pg_temp; +SELECT df.grant_usage('df_literal_quoting_user'); + +DROP SCHEMA IF EXISTS test_literal_quoting CASCADE; +CREATE SCHEMA test_literal_quoting AUTHORIZATION df_literal_quoting_user; +SET SESSION AUTHORIZATION df_literal_quoting_user; + +CREATE FUNCTION test_literal_quoting.quote_literal(text) RETURNS text +LANGUAGE sql AS $$ SELECT 'wrong'::text $$; + +CREATE TABLE test_literal_quoting.source AS +SELECT id, value, jsonb_build_object('value', value) AS object_value, + jsonb_build_array(value) AS array_value +FROM (VALUES + (1, $value$\' OR true --$value$), + (2, $value$C:\new\test\123\x41\$value$), + (3, $value$O'Reilly$value$), + (4, ''), + (5, U&'\00E9\4E2D\+01F642') +) AS input(id, value); + +CREATE TABLE test_literal_quoting.copied ( + variant text, id integer, value text, object_value jsonb, array_value jsonb, + standard_strings text +); + +CREATE TEMP TABLE _literal_quoting_state AS +SELECT df.start( + df.seq( + df.seq( + df.as( + 'SELECT value, id, object_value, array_value FROM test_literal_quoting.source ORDER BY id', + 'rows' + ), + df.as( + 'SELECT count(*) AS matched FROM test_literal_quoting.source WHERE value = $rows AND id = -1', + 'guard' + ) + ), + df.seq( + $query$ + INSERT INTO test_literal_quoting.copied + SELECT 'rowset', id, value, object_value::jsonb, array_value::jsonb, + current_setting('standard_conforming_strings') + FROM $rows.* + $query$, + df.seq( + $query$ + INSERT INTO test_literal_quoting.copied VALUES + ('scalar', $rows.id, $rows, $rows.object_value::jsonb, $rows.array_value::jsonb, + current_setting('standard_conforming_strings')), + ('column', $rows.id, $rows.value, $rows.object_value::jsonb, $rows.array_value::jsonb, + current_setting('standard_conforming_strings')), + ('nullable-scalar', $rows.id, $rows?, $rows.object_value?::jsonb, $rows.array_value?::jsonb, + current_setting('standard_conforming_strings')), + ('nullable-column', $rows.id, $rows.value?, $rows.object_value?::jsonb, $rows.array_value?::jsonb, + current_setting('standard_conforming_strings')) + $query$, + 'SELECT $guard.matched AS matched' + ) + ) + ), + 'literal-quoting' +) AS instance_id; + +DO $assert$ +DECLARE + inst_id text; + final_status text; +BEGIN + SELECT instance_id INTO STRICT inst_id FROM _literal_quoting_state; + SELECT df.await_instance(inst_id, 30) INTO final_status; + IF final_status IS DISTINCT FROM 'completed' THEN + RAISE EXCEPTION 'TEST FAILED: literal quoting status %, result %', + final_status, df.result(inst_id); + END IF; + IF (df.result(inst_id)::jsonb->'rows'->0->>'matched')::integer IS DISTINCT FROM 0 THEN + RAISE EXCEPTION 'TEST FAILED: literal escaped into SQL expression'; + END IF; + IF (SELECT count(*) FROM test_literal_quoting.copied) <> 9 THEN + RAISE EXCEPTION 'TEST FAILED: expected five row-set rows and four scalar rows'; + END IF; + IF EXISTS ( + SELECT FROM test_literal_quoting.copied c + LEFT JOIN test_literal_quoting.source s USING (id) + WHERE c.value IS DISTINCT FROM s.value + OR c.object_value IS DISTINCT FROM s.object_value + OR c.array_value IS DISTINCT FROM s.array_value + OR c.standard_strings IS DISTINCT FROM 'off' + ) THEN + RAISE EXCEPTION 'TEST FAILED: result changed or activity did not use standard_conforming_strings=off'; + END IF; +END $assert$; + +DROP TABLE _literal_quoting_state; +RESET SESSION AUTHORIZATION; +DROP SCHEMA test_literal_quoting CASCADE; +DROP OWNED BY df_literal_quoting_user; +DROP ROLE df_literal_quoting_user; +SELECT 'TEST PASSED' AS result; From 806f04c1c942507c649a499321529536013f2315 Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Thu, 29 Oct 2026 18:49:57 +0000 Subject: [PATCH 2/3] Fix review comments, handle versioned execution --- scripts/test_upgrade_lifecycle.py | 21 ++- scripts/upgrade_lifecycle.py | 9 +- src/activities/execute_sql.rs | 65 ++++++- src/activities/load_function_graph.rs | 2 + src/lib.rs | 61 +++++++ src/orchestrations/execute_function_graph.rs | 55 ++++-- src/types.rs | 170 +++++++++++++++---- tests/e2e/sql/97_literal_quoting.sql | 16 +- tests/upgrade/lifecycle.sql | 26 +++ 9 files changed, 372 insertions(+), 53 deletions(-) diff --git a/scripts/test_upgrade_lifecycle.py b/scripts/test_upgrade_lifecycle.py index 1effe8f9..d9b49d8b 100644 --- a/scripts/test_upgrade_lifecycle.py +++ b/scripts/test_upgrade_lifecycle.py @@ -131,6 +131,15 @@ def test_break_requires_first_iteration_before_resume(self): self.assertIn("incorrect marker values at r.0", "; ".join(lifecycle.case_problems(value, {}))) + def test_quoting_requires_exact_values_before_and_after_resume(self): + for waiting in (True, False): + for invalid in (None, 0): + value = case("quoting", waiting=waiting) + value["marks"][0]["value"] = invalid + with self.subTest(waiting=waiting, invalid=invalid): + self.assertIn("incorrect marker values at r.0", + "; ".join(lifecycle.case_problems(value, {}))) + def test_race_requires_exact_loser_and_durable_timer(self): for waiting in (True, False): for field, invalid in (("race_loser_id", None), ("race_loser_timer_created", False)): @@ -267,9 +276,16 @@ def validate(names, completed): self.assertLess(alter, validations[3]) self.assertLess(validations[3], binary_wait_release) self.assertLess(binary_wait_release, validations[4]) + quoting_start = calls.index(unittest.mock.call.cluster.start_case( + "quoting-binary-b2", "quoting", True)) + quoting_release = calls.index(unittest.mock.call.cluster.release("quoting-binary-b2")) + self.assertLess(new_install, quoting_start) + self.assertLess(quoting_start, validations[2]) + self.assertLess(validations[3], quoting_release) + self.assertLess(quoting_release, validations[4]) baseline = {f"{family}-{boundary}" for family in lifecycle.FAMILIES for boundary in ("b1", "b2")} | {"seq-done-baseline"} - binary = baseline | {"seq-done-b1", "seq-binary-b2"} + binary = baseline | {"seq-done-b1", "seq-binary-b2", "quoting-binary-b2"} self.assertEqual(cohorts, [ baseline, baseline, binary, binary, binary | {"seq-done-b2"}, ]) @@ -291,7 +307,8 @@ def validate(names, completed): self.assertEqual(cluster.release.call_args_list, [unittest.mock.call(f"{family}-b1") for family in lifecycle.FAMILIES] + [unittest.mock.call(f"{family}-b2") for family in lifecycle.FAMILIES] - + [unittest.mock.call("seq-binary-b2")]) + + [unittest.mock.call("seq-binary-b2"), + unittest.mock.call("quoting-binary-b2")]) self.assertEqual(cluster.stop.call_count, 2) self.assertEqual(len(list(output.glob("*.json"))), 5) diff --git a/scripts/upgrade_lifecycle.py b/scripts/upgrade_lifecycle.py index dc00f7f4..c9a4ad28 100644 --- a/scripts/upgrade_lifecycle.py +++ b/scripts/upgrade_lifecycle.py @@ -25,7 +25,7 @@ # structure and resumed under both the B1 (new binary / old schema) and B2 # (new binary / new schema) boundaries. Every #410 combinator plus the explicit # else and break seeds is represented; df.sql is exercised by every shape. -FAMILIES = ["seq", "if-then", "if-else", "loop", "break", "join", "race"] +FAMILIES = ["seq", "if-then", "if-else", "loop", "break", "join", "race", "quoting"] # Exact marker counts per structural path while the instance is still suspended # ("before") and after it resumes and completes ("after"). Paths absent from a @@ -44,6 +44,9 @@ "before_values": {"r.0": [1]}, "after_values": {"r.0": [1, 2, 3]}}, "join": {"before": {"r.0": 1, "r.b": 1}, "after": {"r.0": 1, "r.b": 1, "r.1": 1}}, "race": {"before": {"r.w": 1}, "after": {"r.w": 1, "r.w2": 1}}, + "quoting": {"before": {"r.0": 1}, "after": {"r.0": 2, "r.1": 2}, + "before_values": {"r.0": [1]}, "after_values": {"r.0": [1, 1], "r.1": [1, 1]}, + "result": {"rows": [{"value": 1}], "row_count": 1}}, } @@ -364,7 +367,8 @@ def check(phase, binary, schema): cluster.release(f"{family}-b1") cluster.start_sequence("seq-done-b1") cluster.start_sequence("seq-binary-b2", waiting=True) - names += ["seq-done-b1", "seq-binary-b2"] + cluster.start_case("quoting-binary-b2", "quoting", True) + names += ["seq-done-b1", "seq-binary-b2", "quoting-binary-b2"] check("b1", current, previous) # B2: new binary, new schema. Resume every -b2 case and start a fresh one. @@ -373,6 +377,7 @@ def check(phase, binary, schema): for family in FAMILIES: cluster.release(f"{family}-b2") cluster.release("seq-binary-b2") + cluster.release("quoting-binary-b2") cluster.start_sequence("seq-done-b2") names += ["seq-done-b2"] check("b2", current, current) diff --git a/src/activities/execute_sql.rs b/src/activities/execute_sql.rs index 4cc71df0..e591f4ab 100644 --- a/src/activities/execute_sql.rs +++ b/src/activities/execute_sql.rs @@ -51,7 +51,9 @@ use sqlx::{Column, Row, TypeInfo}; use std::sync::Arc; use tokio::sync::Semaphore; -use crate::types::{connect_as_user, get_execution_acquire_timeout, get_max_user_connections}; +use crate::types::{ + connect_as_user, get_execution_acquire_timeout, get_max_user_connections, SqlQuotingVersion, +}; /// Activity name for registration and scheduling pub const NAME: &str = "pg_durable::activity::execute-sql"; @@ -61,9 +63,36 @@ pub const NAME: &str = "pg_durable::activity::execute-sql"; pub struct ExecuteSqlInput { pub query: String, pub submitted_by: String, - /// Target database (None = extension database) - #[serde(skip_serializing_if = "Option::is_none")] + /// Target database (None = extension database). Historical inputs include the null. pub database: Option, + #[serde(default, skip_serializing_if = "SqlQuotingVersion::is_legacy")] + pub sql_quoting_version: SqlQuotingVersion, +} + +pub(crate) async fn prepare_sql_quoting( + conn: &mut sqlx::PgConnection, + version: SqlQuotingVersion, +) -> Result<(), String> { + if version.is_legacy() { + // Pin an already-enabled setting so a configuration reload cannot make + // historical literals unsafe between this check and query parsing. + let setting = sqlx::query_scalar::<_, String>( + "SELECT pg_catalog.set_config('standard_conforming_strings', 'on', false) + WHERE pg_catalog.current_setting('standard_conforming_strings') + OPERATOR(pg_catalog.=) 'on'", + ) + .fetch_optional(conn) + .await + .map_err(|e| format!("Failed to enforce legacy SQL quoting: {e}"))?; + if setting.is_none() { + return Err( + "Legacy workflow SQL requires standard_conforming_strings=on; \ + replay-compatible literals are unsafe when it is off" + .to_string(), + ); + } + } + Ok(()) } /// Decode a single column value from a PostgreSQL row into a `serde_json::Value`. @@ -237,6 +266,7 @@ pub(crate) async fn execute_in_origin( if let Some(origin) = origin { crate::origin::validate_execution_connection(&mut conn, &origin).await?; } + prepare_sql_quoting(&mut conn, input.sql_quoting_version).await?; if let Some(source) = source { source.validate().await?; } @@ -283,3 +313,32 @@ async fn execute_query( } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn sql_quoting_version_preserves_legacy_activity_bytes() { + for legacy in [ + r#"{"query":"SELECT '\\path'","submitted_by":"reader","database":null}"#, + r#"{"query":"SELECT '\\path'","submitted_by":"reader","database":"other"}"#, + ] { + let input: ExecuteSqlInput = serde_json::from_str(legacy).unwrap(); + assert_eq!(input.sql_quoting_version, SqlQuotingVersion::Legacy); + assert_eq!(serde_json::to_string(&input).unwrap(), legacy); + } + } + + #[test] + fn sql_quoting_version_is_explicit_for_new_activities() { + let current = r#"{"query":"SELECT 1","submitted_by":"reader","database":null,"sql_quoting_version":"v1"}"#; + let input: ExecuteSqlInput = serde_json::from_str(current).unwrap(); + assert_eq!(input.sql_quoting_version, SqlQuotingVersion::V1); + assert_eq!(serde_json::to_string(&input).unwrap(), current); + assert!(serde_json::from_str::( + r#"{"query":"SELECT 1","submitted_by":"reader","database":null,"sql_quoting_version":"v2"}"# + ) + .is_err()); + } +} diff --git a/src/activities/load_function_graph.rs b/src/activities/load_function_graph.rs index 02f46749..e5156227 100644 --- a/src/activities/load_function_graph.rs +++ b/src/activities/load_function_graph.rs @@ -13,6 +13,7 @@ use std::time::Duration; use crate::types::{ is_role_superuser_name, superuser_instances_enabled, FunctionGraph, FunctionNode, + SqlQuotingVersion, }; /// Activity name for registration and scheduling @@ -370,6 +371,7 @@ async fn load_visible_graph( instance_id, root_node_id, nodes, + sql_quoting_version: SqlQuotingVersion::V1, }; ctx.trace_info(format!( diff --git a/src/lib.rs b/src/lib.rs index 9c3747a7..79ef04ac 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -2666,6 +2666,67 @@ mod tests { assert_eq!(Spi::get_one::(&query).unwrap(), Some(false)); } + #[pg_test] + fn test_legacy_sql_quoting_rejects_unsafe_connection_settings() { + use crate::activities::execute_sql::prepare_sql_quoting; + use crate::types::SqlQuotingVersion; + use sqlx::Connection; + + let connection_string = test_database_connection_string(); + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + rt.block_on(async { + let mut conn = sqlx::PgConnection::connect(&connection_string) + .await + .unwrap(); + let mut tx = conn.begin().await.unwrap(); + sqlx::raw_sql( + "CREATE SCHEMA quoting_guard_shadow; + CREATE FUNCTION quoting_guard_shadow.current_setting(text) + RETURNS text LANGUAGE sql AS $$ SELECT 'on'::text $$; + CREATE FUNCTION quoting_guard_shadow.set_config(text, text, boolean) + RETURNS text LANGUAGE sql AS $$ SELECT 'on'::text $$; + CREATE FUNCTION quoting_guard_shadow.equal(text, text) + RETURNS boolean LANGUAGE sql AS $$ SELECT true $$; + CREATE OPERATOR quoting_guard_shadow.= ( + FUNCTION = quoting_guard_shadow.equal, + LEFTARG = text, RIGHTARG = text); + SET LOCAL search_path = quoting_guard_shadow, pg_catalog, pg_temp", + ) + .execute(&mut *tx) + .await + .unwrap(); + + for version in [SqlQuotingVersion::Legacy, SqlQuotingVersion::V1] { + for setting in ["on", "off"] { + sqlx::query(&format!( + "SET LOCAL standard_conforming_strings = {setting}" + )) + .execute(&mut *tx) + .await + .unwrap(); + let result = prepare_sql_quoting(&mut tx, version).await; + if version.is_legacy() && setting == "off" { + assert!(result + .unwrap_err() + .contains("requires standard_conforming_strings=on")); + } else { + result.unwrap(); + } + let actual: String = sqlx::query_scalar("SHOW standard_conforming_strings") + .fetch_one(&mut *tx) + .await + .unwrap(); + assert_eq!(actual, setting, "must not silently change string semantics"); + } + } + tx.rollback().await.unwrap(); + conn.close().await.unwrap(); + }); + } + #[pg_test] fn test_literal_quoting_round_trips_all_result_forms() { use crate::types::{substitute_all, substitute_all_raw, SystemVars}; diff --git a/src/orchestrations/execute_function_graph.rs b/src/orchestrations/execute_function_graph.rs index 7c3d61fb..f24fb393 100644 --- a/src/orchestrations/execute_function_graph.rs +++ b/src/orchestrations/execute_function_graph.rs @@ -19,8 +19,8 @@ use duroxide::OrchestrationContext; use crate::activities; use crate::activities::load_function_graph::{TransactionAwareLoadInput, TransactionGraphProbe}; use crate::types::{ - evaluate_condition, string_map_to_json, substitute_all, substitute_all_raw, FunctionGraph, - FunctionInput, FunctionNode, LoopConfig, SystemVars, + evaluate_condition, string_map_to_json, substitute_all_raw, substitute_all_with_options, + FunctionGraph, FunctionInput, FunctionNode, LoopConfig, SqlQuotingVersion, SystemVars, }; /// Orchestration name for ExecuteFunctionGraph @@ -886,7 +886,18 @@ async fn execute_node_inner( }; match node.node_type.to_lowercase().as_str() { - "sql" => execute_sql_node(ctx, node, node_id, results, exec_ctx, &sys_vars).await, + "sql" => { + execute_sql_node( + ctx, + node, + node_id, + results, + exec_ctx, + &sys_vars, + graph.sql_quoting_version, + ) + .await + } "then" => execute_then_node(ctx, graph, node, node_id, results, exec_ctx).await, "sleep" => execute_sleep_node(ctx, node, node_id).await, "wait_schedule" => execute_wait_schedule_node(ctx, node, node_id).await, @@ -915,26 +926,35 @@ async fn execute_sql_node( results: &mut HashMap, exec_ctx: &ExecutionContext, sys_vars: &SystemVars, + sql_quoting_version: SqlQuotingVersion, ) -> NodeResult { let query = node .query .as_ref() .ok_or_else(|| format!("SQL node {node_id} has no query"))?; - let final_query = substitute_all(query, results, &exec_ctx.vars, sys_vars)?; + let final_query = substitute_all_with_options( + query, + results, + &exec_ctx.vars, + sys_vars, + Some(sql_quoting_version), + )?; // The substituted text is not traced here: the execute_sql activity logs it // under pg_durable.log_workflow_sql, and an orchestration must not read a // GUC to decide what to emit. ctx.trace_info(format!("Executing SQL node {node_id}")); - let input = serde_json::json!({ - "query": final_query, - "submitted_by": node.submitted_by, - "database": node.database, - }); + let input = serde_json::to_string(&activities::execute_sql::ExecuteSqlInput { + query: final_query, + submitted_by: node.submitted_by.clone(), + database: node.database.clone(), + sql_quoting_version, + }) + .map_err(|e| format!("Failed to serialize execute_sql input: {e}"))?; let result = ctx - .schedule_activity(activities::execute_sql::NAME, input.to_string()) + .schedule_activity(activities::execute_sql::NAME, input) .await .map_err(NodeError::Application)?; @@ -2460,11 +2480,12 @@ mod tests { } #[test] - fn subtree_input_is_byte_stable_across_result_insertion_order() { - let graph = FunctionGraph { + fn subtree_input_preserves_quoting_version_and_result_order() { + let mut graph = FunctionGraph { instance_id: "deadbeef".to_string(), root_node_id: "cafebabe".to_string(), nodes: std::collections::BTreeMap::new(), + sql_quoting_version: SqlQuotingVersion::Legacy, }; let exec_ctx = ExecutionContext { vars: HashMap::new(), @@ -2489,6 +2510,16 @@ mod tests { assert_eq!(first_input, expected); assert_eq!(second_input, expected); + + graph.sql_quoting_version = SqlQuotingVersion::V1; + let current_input = build_subtree_input(&graph, "deadbeef", &first, &exec_ctx).unwrap(); + assert_eq!( + current_input, + build_subtree_input(&graph, "deadbeef", &second, &exec_ctx).unwrap() + ); + let subtree: SubtreeInput = serde_json::from_str(¤t_input).unwrap(); + let inherited: FunctionGraph = serde_json::from_str(&subtree.graph).unwrap(); + assert_eq!(inherited.sql_quoting_version, SqlQuotingVersion::V1); } #[test] diff --git a/src/types.rs b/src/types.rs index 8abecf20..103c5dda 100644 --- a/src/types.rs +++ b/src/types.rs @@ -8,7 +8,7 @@ use pgrx::{pg_extern, Spi}; use chrono::{DateTime, Utc}; use cron::Schedule as CronSchedule; use percent_encoding::{utf8_percent_encode, NON_ALPHANUMERIC}; -use postgres_protocol::escape::escape_literal as quote_literal; +use postgres_protocol::escape::escape_literal; use serde::{Deserialize, Serialize}; use std::ffi::{CStr, CString}; use std::str::FromStr; @@ -721,6 +721,29 @@ pub fn is_truthy(value: &serde_json::Value) -> bool { } } +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum SqlQuotingVersion { + #[default] + Legacy, + V1, +} + +impl SqlQuotingVersion { + pub(crate) fn is_legacy(&self) -> bool { + matches!(self, Self::Legacy) + } + + fn quote_literal(self, value: &str) -> String { + match self { + // Historical activity inputs must remain byte-identical. The SQL + // activity permits this version only with standard strings enabled. + Self::Legacy => format!("'{}'", value.replace('\'', "''")), + Self::V1 => escape_literal(value), + } + } +} + /// System variables available during workflow execution pub struct SystemVars { pub instance_id: String, @@ -772,11 +795,11 @@ fn quote_identifier(name: &str) -> String { } /// Format a JSON value for use in a SQL or raw context. -fn format_value(val: &serde_json::Value, for_sql: bool) -> String { +fn format_value(val: &serde_json::Value, sql_quoting: Option) -> String { match val { serde_json::Value::String(s) => { - if for_sql { - quote_literal(s) + if let Some(quoting) = sql_quoting { + quoting.quote_literal(s) } else { s.clone() } @@ -784,8 +807,8 @@ fn format_value(val: &serde_json::Value, for_sql: bool) -> String { serde_json::Value::Number(n) => n.to_string(), serde_json::Value::Bool(b) => b.to_string(), _ => { - if for_sql { - quote_literal(&val.to_string()) + if let Some(quoting) = sql_quoting { + quoting.quote_literal(&val.to_string()) } else { val.to_string() } @@ -798,14 +821,14 @@ fn extract_first_column_value( name: &str, json_str: &str, null_safe: bool, - for_sql: bool, + sql_quoting: Option, ) -> Result { let json: serde_json::Value = match serde_json::from_str(json_str) { Ok(v) => v, Err(_) => { // Not JSON — return raw value (backward compat for HTTP responses etc.) - return Ok(if for_sql { - quote_literal(json_str) + return Ok(if let Some(quoting) = sql_quoting { + quoting.quote_literal(json_str) } else { json_str.to_string() }); @@ -839,9 +862,9 @@ fn extract_first_column_value( }; } - Ok(format_value(val, for_sql)) - } else if for_sql { - Ok(quote_literal(json_str)) + Ok(format_value(val, sql_quoting)) + } else if let Some(quoting) = sql_quoting { + Ok(quoting.quote_literal(json_str)) } else { Ok(json_str.to_string()) } @@ -860,7 +883,7 @@ fn extract_column_value( json_str: &str, col: &str, null_safe: bool, - for_sql: bool, + sql_quoting: Option, ) -> Result { let json: serde_json::Value = serde_json::from_str(json_str) .map_err(|_| format!("${name}.{col}: result is not valid JSON"))?; @@ -893,7 +916,7 @@ fn extract_column_value( // header, a multipart field — has no such parser, so a leftover // `$name.col` would be sent over the wire verbatim and the request // would fail somewhere far less obvious. Fail loudly instead. - if !for_sql && !null_safe { + if sql_quoting.is_none() && !null_safe { let mut available: Vec<&str> = fields.keys().map(String::as_str).collect(); available.sort_unstable(); return Err(format!( @@ -914,14 +937,18 @@ fn extract_column_value( }; } - Ok(format_value(val, for_sql)) + Ok(format_value(val, sql_quoting)) } /// Expand `$name.*` into an inline `VALUES` subquery (SQL) or JSON array (raw). /// /// Preserve the generated casts (including `::text`) for replay of unaffected /// activity inputs. This SQL executes as the submitting user, not the worker role. -fn expand_row_set(name: &str, json_str: &str, for_sql: bool) -> Result { +fn expand_row_set( + name: &str, + json_str: &str, + sql_quoting: Option, +) -> Result { /// Maximum number of rows allowed in `$name.*` expansion to prevent /// unbounded SQL string allocation from large result sets. const MAX_ROWSET_EXPANSION: usize = 10_000; @@ -943,9 +970,9 @@ fn expand_row_set(name: &str, json_str: &str, for_sql: bool) -> Result Result { - format!("{}::text", quote_literal(s)) + format!("{}::text", quoting.quote_literal(s)) } Some(serde_json::Value::Number(n)) => n.to_string(), Some(serde_json::Value::Bool(b)) => b.to_string(), Some(serde_json::Value::Null) | None => "NULL".to_string(), Some(other) => { - format!("{}::text", quote_literal(&other.to_string())) + format!("{}::text", quoting.quote_literal(&other.to_string())) } }) .collect(); @@ -1000,7 +1027,7 @@ fn expand_row_set(name: &str, json_str: &str, for_sql: bool) -> Result, - for_sql: bool, + sql_quoting: Option, ) -> Result { if results.is_empty() { return Ok(input.to_string()); @@ -1035,7 +1062,7 @@ fn substitute_results( // 1. $name.* — row-set expansion if after_name.starts_with(".*") { - let replacement = expand_row_set(name, json_str, for_sql)?; + let replacement = expand_row_set(name, json_str, sql_quoting)?; out.push_str(&replacement); i += 1 + name.len() + 2; // $ + name + .* matched = true; @@ -1049,7 +1076,7 @@ fn substitute_results( let after_col = &after_dot[col.len()..]; let null_safe = after_col.starts_with('?'); let replacement = - extract_column_value(name, json_str, col, null_safe, for_sql)?; + extract_column_value(name, json_str, col, null_safe, sql_quoting)?; out.push_str(&replacement); i += 1 + name.len() + 1 + col.len() + if null_safe { 1 } else { 0 }; matched = true; @@ -1060,7 +1087,7 @@ fn substitute_results( // 4. $name? — null-safe scalar if after_name.starts_with('?') { - let replacement = extract_first_column_value(name, json_str, true, for_sql)?; + let replacement = extract_first_column_value(name, json_str, true, sql_quoting)?; out.push_str(&replacement); i += 1 + name.len() + 1; // $ + name + ? matched = true; @@ -1069,7 +1096,7 @@ fn substitute_results( // 5. $name — strict scalar (with word-boundary check) if after_name.is_empty() || !is_ident_continue(after_name.as_bytes()[0]) { - let replacement = extract_first_column_value(name, json_str, false, for_sql)?; + let replacement = extract_first_column_value(name, json_str, false, sql_quoting)?; out.push_str(&replacement); i += 1 + name.len(); matched = true; @@ -1133,6 +1160,7 @@ fn substitute_braced_variables( /// /// User vars and system vars are substituted without quoting - the user should /// handle SQL escaping in the original query if needed. +/// `sql_quoting` selects the recorded SQL format; `None` leaves results unquoted. /// /// Returns `Err` if a strict (non-`?`) pattern references a result with no rows /// or a NULL value. @@ -1141,7 +1169,7 @@ pub fn substitute_all_with_options( results: &std::collections::HashMap, vars: &std::collections::HashMap, sys_vars: &SystemVars, - quote_results_for_sql: bool, + sql_quoting: Option, ) -> Result { // SECURITY: Raw substitution of user vars is by design — variables are // intended for SQL fragments (table names, expressions), not just values. @@ -1152,7 +1180,7 @@ pub fn substitute_all_with_options( let result = substitute_braced_variables(query, vars, sys_vars); // 3. Substitute results: $name with dot-notation, null-safe, and row-set support - substitute_results(&result, results, quote_results_for_sql) + substitute_results(&result, results, sql_quoting) } /// Substitute all variables with SQL quoting (default for SQL contexts) @@ -1162,7 +1190,7 @@ pub fn substitute_all( vars: &std::collections::HashMap, sys_vars: &SystemVars, ) -> Result { - substitute_all_with_options(query, results, vars, sys_vars, true) + substitute_all_with_options(query, results, vars, sys_vars, Some(SqlQuotingVersion::V1)) } /// Substitute all variables without SQL quoting (for URLs, headers, etc.) @@ -1172,7 +1200,7 @@ pub fn substitute_all_raw( vars: &std::collections::HashMap, sys_vars: &SystemVars, ) -> Result { - substitute_all_with_options(query, results, vars, sys_vars, false) + substitute_all_with_options(query, results, vars, sys_vars, None) } /// Report whether `value` consists of exactly one variable reference and nothing @@ -1459,6 +1487,8 @@ pub struct FunctionGraph { pub instance_id: String, pub root_node_id: String, pub nodes: std::collections::BTreeMap, + #[serde(default, skip_serializing_if = "SqlQuotingVersion::is_legacy")] + pub sql_quoting_version: SqlQuotingVersion, } fn is_zero_u32(value: &u32) -> bool { @@ -3150,6 +3180,84 @@ mod tests { assert_eq!(out, "SELECT 1, 'Bob'"); } + #[test] + fn sql_quoting_version_preserves_legacy_graph_bytes() { + let legacy = r#"{"instance_id":"instance","root_node_id":"root","nodes":{}}"#; + let mut graph: FunctionGraph = serde_json::from_str(legacy).unwrap(); + assert_eq!(graph.sql_quoting_version, SqlQuotingVersion::Legacy); + assert_eq!(serde_json::to_string(&graph).unwrap(), legacy); + + graph.sql_quoting_version = SqlQuotingVersion::V1; + let current = serde_json::to_string(&graph).unwrap(); + assert_eq!( + current, + r#"{"instance_id":"instance","root_node_id":"root","nodes":{},"sql_quoting_version":"v1"}"# + ); + assert_eq!( + serde_json::from_str::(¤t) + .unwrap() + .sql_quoting_version, + SqlQuotingVersion::V1 + ); + assert!(serde_json::from_str::( + r#"{"instance_id":"instance","root_node_id":"root","nodes":{},"sql_quoting_version":"v2"}"# + ) + .is_err()); + } + + #[test] + fn legacy_sql_quoting_preserves_recorded_substitutions() { + for (result, query, expected) in [ + ( + r#"{"rows":[{"value":"\\' OR true --"}]}"#, + "SELECT $payload", + r"SELECT '\'' OR true --'", + ), + ( + r#"{"rows":[{"value":"\\' OR true --"}]}"#, + "SELECT $payload.value?", + r"SELECT '\'' OR true --'", + ), + ( + r"\' OR true --", + "SELECT $payload", + r"SELECT '\'' OR true --'", + ), + ( + r#"{"body":"\\' OR true --"}"#, + "SELECT $payload", + r#"SELECT '{"body":"\\'' OR true --"}'"#, + ), + ( + r#"{"rows":[{"value":{"text":"\\"}}]}"#, + "SELECT $payload.value", + r#"SELECT '{"text":"\\"}'"#, + ), + ( + r#"{"rows":[{"text":"\\'","doc":{"text":"\\"},"arr":["\\'"]}]}"#, + "SELECT * FROM $payload.*", + r#"SELECT * FROM (VALUES ('\'''::text,'{"text":"\\"}'::text,'["\\''"]'::text)) AS "payload"("text", "doc", "arr")"#, + ), + ] { + let results = make_results(&[("payload", result)]); + assert_eq!( + substitute_all_with_options( + query, + &results, + &empty_vars(), + &sys_vars(), + Some(SqlQuotingVersion::Legacy), + ) + .unwrap(), + expected + ); + assert_ne!( + substitute_all(query, &results, &empty_vars(), &sys_vars()).unwrap(), + expected + ); + } + } + #[test] fn literal_quoting_preserves_unaffected_sql() { let results = make_results(&[( @@ -3249,7 +3357,7 @@ mod tests { fn test_expand_row_set_quoted_columns() { // Column names from PostgreSQL can contain special characters let json = r#"{"rows":[{"normal":1,"has space":2}],"row_count":1}"#; - let result = expand_row_set("tbl", json, true).unwrap(); + let result = expand_row_set("tbl", json, Some(SqlQuotingVersion::V1)).unwrap(); assert!(result.contains(r#""normal""#)); assert!(result.contains(r#""has space""#)); assert!(result.contains(r#"AS "tbl"("#)); @@ -3258,7 +3366,7 @@ mod tests { #[test] fn test_expand_row_set_empty_quoted_name() { let json = r#"{"rows":[],"row_count":0}"#; - let result = expand_row_set("batch", json, true).unwrap(); + let result = expand_row_set("batch", json, Some(SqlQuotingVersion::V1)).unwrap(); assert_eq!(result, r#"(SELECT NULL WHERE false) AS "batch""#); } #[test] diff --git a/tests/e2e/sql/97_literal_quoting.sql b/tests/e2e/sql/97_literal_quoting.sql index 260d5e06..87c7094a 100644 --- a/tests/e2e/sql/97_literal_quoting.sql +++ b/tests/e2e/sql/97_literal_quoting.sql @@ -69,7 +69,17 @@ SELECT df.start( ('nullable-column', $rows.id, $rows.value?, $rows.object_value?::jsonb, $rows.array_value?::jsonb, current_setting('standard_conforming_strings')) $query$, - 'SELECT $guard.matched AS matched' + df.seq( + df.loop( + $query$ + INSERT INTO test_literal_quoting.copied VALUES + ('loop', $rows.id, $rows.value, $rows.object_value::jsonb, + $rows.array_value::jsonb, current_setting('standard_conforming_strings')) + $query$, + 'SELECT count(*) < 2 FROM test_literal_quoting.copied WHERE variant = ''loop''' + ), + 'SELECT $guard.matched AS matched' + ) ) ) ), @@ -90,8 +100,8 @@ BEGIN IF (df.result(inst_id)::jsonb->'rows'->0->>'matched')::integer IS DISTINCT FROM 0 THEN RAISE EXCEPTION 'TEST FAILED: literal escaped into SQL expression'; END IF; - IF (SELECT count(*) FROM test_literal_quoting.copied) <> 9 THEN - RAISE EXCEPTION 'TEST FAILED: expected five row-set rows and four scalar rows'; + IF (SELECT count(*) FROM test_literal_quoting.copied) <> 11 THEN + RAISE EXCEPTION 'TEST FAILED: expected five row-set, four scalar, and two loop rows'; END IF; IF EXISTS ( SELECT FROM test_literal_quoting.copied c diff --git a/tests/upgrade/lifecycle.sql b/tests/upgrade/lifecycle.sql index 383296cf..e3812537 100644 --- a/tests/upgrade/lifecycle.sql +++ b/tests/upgrade/lifecycle.sql @@ -76,6 +76,8 @@ DECLARE gate text := df.wait_for_signal('resume'); graph text; instance text; + quoted_value text := $value$C:\path\O'Reilly$value$; + quoted_check text; BEGIN CASE shape WHEN 'seq' THEN @@ -88,6 +90,30 @@ BEGIN graph := df.seq(graph, public.upgrade_lifecycle_mark_sql( '{sys_label}', 'r.1', '$seed::integer + {lifecycle_increment}')); + WHEN 'quoting' THEN + graph := df.as(format( + 'SELECT %L::text AS value, %L::jsonb AS object_value, %L::jsonb AS array_value', + quoted_value, jsonb_build_object('value', quoted_value)::text, + jsonb_build_array(quoted_value)::text), 'quoted'); + quoted_check := format($check$( + SELECT CASE WHEN quoted.value = %L + AND quoted.value = $quoted AND quoted.value = $quoted? + AND quoted.value = $quoted.value AND quoted.value = $quoted.value? + AND quoted.object_value::jsonb = $quoted.object_value::jsonb + AND quoted.array_value::jsonb = $quoted.array_value::jsonb + THEN 1 ELSE 0 END FROM $quoted.* + )$check$, quoted_value); + -- The child records backslash-bearing inputs before suspension, + -- then must retain its quoting version across continue_as_new. + graph := df.seq(graph, df.loop( + df.seq( + public.upgrade_lifecycle_mark_sql(label, 'r.0', quoted_check), + df.seq( + df.if(public.upgrade_lifecycle_count_sql(label, 'r.0', '= 1'), + CASE WHEN waiting THEN gate ELSE 'SELECT 1' END, 'SELECT 1'), + public.upgrade_lifecycle_mark_sql(label, 'r.1', quoted_check))), + public.upgrade_lifecycle_count_sql(label, 'r.0', '< 2'))); + WHEN 'if-then' THEN -- Then branch suspends mid-branch; the else branch must stay unrun. graph := df.if('SELECT true', From ca3372bcc506acd9d836090ecedb0c1764f23523 Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Fri, 30 Oct 2026 11:31:07 +0000 Subject: [PATCH 3/3] Fix comment handling --- src/lib.rs | 89 ++++++++++++++++++++++++++++ src/types.rs | 53 ++++++++++++++++- tests/e2e/sql/97_literal_quoting.sql | 12 +++- 3 files changed, 151 insertions(+), 3 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index 79ef04ac..8270f91c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -2666,6 +2666,73 @@ mod tests { assert_eq!(Spi::get_one::(&query).unwrap(), Some(false)); } + #[pg_test] + fn test_literal_quoting_separates_negative_numeric_values() { + use crate::types::{ + substitute_all, substitute_all_with_options, SqlQuotingVersion, SystemVars, + }; + use std::collections::HashMap; + + let results = HashMap::from([( + "payload".to_string(), + serde_json::json!({"rows": [{"value": -1}]}).to_string(), + )]); + let vars = HashMap::new(); + let sys_vars = SystemVars { + instance_id: "negative-literal-quoting".to_string(), + label: None, + }; + let query = substitute_all( + "SELECT count(*) FROM (VALUES (1, false)) AS rows(id, allowed) + WHERE id = 1-$payload AND allowed", + &results, + &vars, + &sys_vars, + ) + .unwrap(); + assert_eq!(Spi::get_one::(&query).unwrap(), Some(0), "{query}"); + + for reference in ["$payload", "$payload?", "$payload.value", "$payload.value?"] { + let query = + substitute_all(&format!("SELECT 1-{reference}"), &results, &vars, &sys_vars) + .unwrap(); + assert_eq!(Spi::get_one::(&query).unwrap(), Some(2), "{query}"); + + let query = + substitute_all(&format!("SELECT 2^{reference}"), &results, &vars, &sys_vars) + .unwrap(); + assert_eq!(Spi::get_one::(&query).unwrap(), Some(0.5), "{query}"); + } + + let query = substitute_all( + "CREATE TEMP SEQUENCE literal_quoting_seq INCREMENT BY $payload", + &results, + &vars, + &sys_vars, + ) + .unwrap(); + Spi::run(&query).unwrap(); + assert_eq!( + Spi::get_one::( + "SELECT seqincrement FROM pg_catalog.pg_sequence + WHERE seqrelid = 'pg_temp.literal_quoting_seq'::pg_catalog.regclass" + ) + .unwrap(), + Some(-1) + ); + assert_eq!( + substitute_all_with_options( + "SELECT 1-$payload", + &results, + &vars, + &sys_vars, + Some(SqlQuotingVersion::Legacy), + ) + .unwrap(), + "SELECT 1--1" + ); + } + #[pg_test] fn test_legacy_sql_quoting_rejects_unsafe_connection_settings() { use crate::activities::execute_sql::prepare_sql_quoting; @@ -2723,6 +2790,28 @@ mod tests { } } tx.rollback().await.unwrap(); + + let source_query = "SELECT source FROM pg_catalog.pg_settings + WHERE name OPERATOR(pg_catalog.=) 'standard_conforming_strings'"; + let source: String = sqlx::query_scalar(source_query) + .fetch_one(&mut conn) + .await + .unwrap(); + assert_ne!( + source, "session", + "the control connection must not already be pinned" + ); + prepare_sql_quoting(&mut conn, SqlQuotingVersion::Legacy) + .await + .unwrap(); + let source: String = sqlx::query_scalar(source_query) + .fetch_one(&mut conn) + .await + .unwrap(); + assert_eq!( + source, "session", + "the pin must outlive the guard's transaction" + ); conn.close().await.unwrap(); }); } diff --git a/src/types.rs b/src/types.rs index 103c5dda..dd7da3f7 100644 --- a/src/types.rs +++ b/src/types.rs @@ -804,7 +804,14 @@ fn format_value(val: &serde_json::Value, sql_quoting: Option) s.clone() } } - serde_json::Value::Number(n) => n.to_string(), + serde_json::Value::Number(n) => { + let mut value = n.to_string(); + if sql_quoting == Some(SqlQuotingVersion::V1) && value.starts_with('-') { + // Keep a preceding operator from consuming the sign (e.g. 1--1). + value.insert(0, ' '); + } + value + } serde_json::Value::Bool(b) => b.to_string(), _ => { if let Some(quoting) = sql_quoting { @@ -3205,6 +3212,50 @@ mod tests { .is_err()); } + #[test] + fn v1_literal_quoting_preserves_recorded_bytes() { + for (value, expected) in [ + ("", "''"), + ("plain", "'plain'"), + ("O'Reilly", "'O''Reilly'"), + ("\\", r" E'\\'"), + (r"C:\path\O'Reilly", r" E'C:\\path\\O''Reilly'"), + ("\u{e9}\u{4e2d}\\'", " E'\u{e9}\u{4e2d}\\\\'''"), + ("line\nbreak", "'line\nbreak'"), + ] { + assert_eq!(SqlQuotingVersion::V1.quote_literal(value), expected); + } + + let results = make_results(&[( + "row", + r#"{"rows":[{"path":"C:\\path\\O'Reilly","negative":-1}]}"#, + )]); + assert_eq!( + substitute_all("SELECT * FROM $row.*", &results, &empty_vars(), &sys_vars()).unwrap(), + r#"SELECT * FROM (VALUES ( E'C:\\path\\O''Reilly'::text,-1)) AS "row"("path", "negative")"# + ); + } + + #[test] + fn v1_literal_quoting_separates_negative_numbers_only_in_sql() { + for (value, expected) in [ + (json!(-1), "-1"), + (json!(-1.5), "-1.5"), + (json!(-0.0), "-0.0"), + (json!(i64::MIN), "-9223372036854775808"), + ] { + assert_eq!( + format_value(&value, Some(SqlQuotingVersion::V1)), + format!(" {expected}") + ); + assert_eq!( + format_value(&value, Some(SqlQuotingVersion::Legacy)), + expected + ); + assert_eq!(format_value(&value, None), expected); + } + } + #[test] fn legacy_sql_quoting_preserves_recorded_substitutions() { for (result, query, expected) in [ diff --git a/tests/e2e/sql/97_literal_quoting.sql b/tests/e2e/sql/97_literal_quoting.sql index 87c7094a..dc1fe26e 100644 --- a/tests/e2e/sql/97_literal_quoting.sql +++ b/tests/e2e/sql/97_literal_quoting.sql @@ -11,6 +11,7 @@ END $setup$; -- Activities use separate connections, so a caller-session SET cannot test this. ALTER ROLE df_literal_quoting_user SET standard_conforming_strings = off; ALTER ROLE df_literal_quoting_user SET backslash_quote = on; +ALTER ROLE df_literal_quoting_user SET client_encoding = 'SJIS'; ALTER ROLE df_literal_quoting_user SET search_path = test_literal_quoting, pg_catalog, pg_temp; SELECT df.grant_usage('df_literal_quoting_user'); @@ -42,7 +43,7 @@ SELECT df.start( df.seq( df.seq( df.as( - 'SELECT value, id, object_value, array_value FROM test_literal_quoting.source ORDER BY id', + 'SELECT value, id, object_value, array_value, -1 AS negative_value FROM test_literal_quoting.source ORDER BY id', 'rows' ), df.as( @@ -78,7 +79,8 @@ SELECT df.start( $query$, 'SELECT count(*) < 2 FROM test_literal_quoting.copied WHERE variant = ''loop''' ), - 'SELECT $guard.matched AS matched' + 'SELECT $guard.matched AS matched, 1-$rows.negative_value AS difference, + current_setting(''client_encoding'') AS client_encoding' ) ) ) @@ -100,6 +102,12 @@ BEGIN IF (df.result(inst_id)::jsonb->'rows'->0->>'matched')::integer IS DISTINCT FROM 0 THEN RAISE EXCEPTION 'TEST FAILED: literal escaped into SQL expression'; END IF; + IF (df.result(inst_id)::jsonb->'rows'->0->>'difference')::integer IS DISTINCT FROM 2 THEN + RAISE EXCEPTION 'TEST FAILED: negative result merged with the preceding operator'; + END IF; + IF df.result(inst_id)::jsonb->'rows'->0->>'client_encoding' IS DISTINCT FROM 'UTF8' THEN + RAISE EXCEPTION 'TEST FAILED: activity did not override the role client encoding with UTF8'; + END IF; IF (SELECT count(*) FROM test_literal_quoting.copied) <> 11 THEN RAISE EXCEPTION 'TEST FAILED: expected five row-set, four scalar, and two loop rows'; END IF;