diff --git a/docs/spec-security-model.md b/docs/spec-security-model.md index e18b369e..fd5cc8e8 100644 --- a/docs/spec-security-model.md +++ b/docs/spec-security-model.md @@ -3,7 +3,7 @@ **Status**: Implementation in progress **Authors**: pg_durable Team **Created**: 2025-12-25 -**Last Updated**: 2026-03-10 +**Last Updated**: 2026-03-11 --- @@ -102,7 +102,10 @@ The security guarantee is: **only superusers can install the extension**, theref | **T5**: Denial of Service | **MEDIUM** | Not implemented | Rate limiting; deferred | | **T6**: Worker Code Vulnerability | **MEDIUM** | Mitigated by design | Relies on code review | | **T0**: SECURITY DEFINER Misuse | **MEDIUM** | Documentation-only | Expected PG behavior | +| **T12**: SQL Injection in Internal SPI Queries | **CRITICAL** | **Implemented** | `df.status()` and `df.result()` now use parameterized SPI queries (`Spi::get_one_with_args`) | | **T7**: Extension Trustworthiness | **LOW** | Accepted | Standard PG trust model | +| **T13**: `search_path` Manipulation in PL/pgSQL Helpers | **LOW** | Implemented | Helper function definitions set `search_path = pg_catalog, df, pg_temp` | +| **T14**: Extension Object Pre-creation | **LOW** | Accepted | Requires operator error; superuser-only install | | **T1–T3**: Privilege Escalation | **CRITICAL** | Implemented | Per-user sqlx connections | --- @@ -299,6 +302,83 @@ See [spec-ssrf-protection.md](spec-ssrf-protection.md) for the full specificatio --- +#### T12: SQL Injection in Internal SPI Queries + +**Severity**: CRITICAL | **Status**: Implemented + +**Threat**: Two extension functions (`df.status()` and `df.result()`) previously interpolated user-supplied `instance_id` directly into SQL without escaping single quotes. That allowed SQL injection to bypass RLS and read other users' instance data. + +```sql +-- Attack: bypass RLS to read any user's instance status +SELECT df.status('x'' OR 1=1--'); + +-- Attack: read results from other users' workflows +SELECT df.result('x'' UNION SELECT secret FROM admin_table--'); +``` + +**Current mitigation** (`src/dsl.rs`): +```rust +// df.status() — parameterized SPI query +let status: Option = Spi::get_one_with_args( + "SELECT status FROM df.instances WHERE id = $1", + &[instance_id.into_datum()], +) +.expect("SPI query failed"); + +// df.result() — parameterized SPI query +let result: Option = Spi::get_one_with_args( + "SELECT result::text FROM df.nodes\n WHERE id = (SELECT root_node FROM df.instances WHERE id = $1)\n AND status = 'completed'", + &[instance_id.into_datum()], +) +.expect("SPI query failed"); +``` + +**Fix implemented**: Replaced string interpolation with parameterized SPI in `df.status()` and `df.result()` using `Spi::get_one_with_args()`. This removes quote-escaping as a correctness requirement for these call sites. Other internal SPI queries should also prefer parameterization wherever supported; manual escaping is fallback-only for cases where parameters are not available. + +**Note on variable substitution**: The `substitute_all_with_options()` function in `src/types.rs` inserts user variables (`{name}`) as-is into SQL without quoting. This is **by design** — variables are intended to be SQL fragments (e.g., table names, expressions). Users choose what to put in their own variables, and the SQL executes with their own privileges on a per-user connection. This is analogous to `psql` variable substitution (`:name`). Result substitution (`$name`) does properly quote string values. + +**Residual Risk**: Low. Remaining hardening is to continue migrating any remaining string-formatted SPI lookups to parameterized calls where possible. + +See this section and Appendix A checklist for ongoing SPI query hardening work. + +--- + +#### T13: `search_path` Manipulation in PL/pgSQL Helper Functions + +**Severity**: LOW | **Status**: Implemented + +**Threat**: The extension's PL/pgSQL helper functions (`df.if_then_op()`, `df.if_else_op()`, `df.ensure_durofut()`) and SQL wrapper functions (`df.as_op()`, `df.loop_prefix_op()`) do not set a fixed `search_path`. If an attacker can place a malicious function in a schema that appears earlier in `search_path`, they could shadow a built-in or extension function. + +**Mitigations (implemented)**: +- All function calls within the helpers are already schema-qualified: `df.ensure_durofut()`, `df.sql()`, `df.if()`, `df.loop()`, `df.as()` +- Built-in functions used (`jsonb_build_object`) are in `pg_catalog`, which is always implicitly first in `search_path` +- The functions are created in the `df` schema (owned by superuser) +- The extension requires superuser to install +- Helper definitions in both fresh install SQL and upgrade SQL include `SET search_path = pg_catalog, df, pg_temp` + +**Fix implemented**: Added `SET search_path = pg_catalog, df, pg_temp` to the helper function definitions (`df.if_then_op()`, `df.if_else_op()`, `df.ensure_durofut()`, `df.as_op()`, `df.loop_prefix_op()`) in both install and upgrade paths. + +**Residual Risk**: Low — current references are schema-qualified and helper search path is pinned as defense-in-depth. + +--- + +#### T14: Extension Object Pre-creation Attack + +**Severity**: LOW | **Status**: Accepted risk + +**Threat**: The extension uses `CREATE TABLE IF NOT EXISTS` and `CREATE OR REPLACE FUNCTION` patterns. An attacker could pre-create objects with the same names to retain ownership or inject malicious implementations. + +**Why this is low risk**: +- All DDL runs inside `extension_sql!()` blocks during `CREATE EXTENSION`, which requires superuser +- The `df` schema is created by pgrx during extension installation — it doesn't exist beforehand +- The `duroxide` schema is created by `sql/duroxide_install.sql` with `SET LOCAL search_path TO duroxide` +- An attacker would need `CREATE TABLE` privilege in a schema controlled by the extension before the extension is installed — this requires the superuser to have manually created the schema and granted access (operator error) +- The `CREATE OR REPLACE FUNCTION` for PL/pgSQL helpers is a standard pgrx convention; the `#[pg_extern]` macro also generates `CREATE OR REPLACE` + +**Residual Risk**: Low — requires operator error (manually creating `df` schema and granting usage before installing the extension). + +--- + ## 4. Functional Requirements ### 4.1 Overall Security Requirements @@ -1793,6 +1873,9 @@ These tests validate behavior when `execute_sql` fails due to expected errors an - [ ] Error messages don't leak other users' data - [ ] Logging includes effective user for audit trail - [ ] `SET df.in_workflow = 'true'` is set on user connections to prevent variable mutation during execution (future: could also guard against recursive `df.start()`) +- [ ] All SPI queries that accept user-supplied parameters prefer parameterized APIs (`Spi::get_one_with_args()` / `Spi::run_with_args()`). Use manual `.replace('\'', "''")` escaping only as fallback where parameters cannot be used. Cross-check: `df.status()`, `df.result()`, `df.cancel()`, `df.signal()`, monitoring functions +- [ ] PL/pgSQL and SQL helper functions include `SET search_path = pg_catalog, df, pg_temp` +- [ ] All function/table references in dynamic SQL are schema-qualified (`df.instances`, not `instances`) --- diff --git a/docs/upgrade-testing.md b/docs/upgrade-testing.md index a612bc2e..a6053f6b 100644 --- a/docs/upgrade-testing.md +++ b/docs/upgrade-testing.md @@ -189,6 +189,12 @@ what the upgrade script handles, and any backward compatibility considerations. ### v0.1.1 → v0.2.0 +#### #51 security hardening (helper `search_path` pinning + SPI parameterization) +- **DDL change:** Upgrade SQL now redefines helper SQL/PLpgSQL functions (`df.as_op()`, `df.if_then_op()`, `df.if_else_op()`, `df.ensure_durofut()`, `df.loop_prefix_op()`) with `SET search_path = pg_catalog, df, pg_temp` for defense-in-depth. +- **Scenario A considerations:** Schema comparison should verify helper function definitions match fresh-install SQL, including the `SET search_path` clause in `proconfig`/function definition text. +- **Scenario B1 considerations:** Runtime code moved key internal lookups to parameterized SPI/sqlx queries. This is backward compatible with prior schemas because query parameterization changed execution style, not table/column contracts. +- **Scenario B2 considerations:** Existing instances/graphs created pre-upgrade should remain readable and executable after `ALTER EXTENSION UPDATE`; tests should include status/result and graph loading paths to cover updated internal query call sites. + #### #53 per-user df.vars scoping via owner column + RLS - **DDL change:** `df.vars` adds `owner REGROLE NOT NULL DEFAULT current_user::regrole`, changes the primary key from `(name)` to `(owner, name)`, enables RLS, and adds the `vars_user_isolation` policy. - **Scenario A considerations:** The schema comparison must verify the new column, its default, the new primary key definition, RLS enabled state, the `vars_user_isolation` policy, and table grants. Because the upgrade script adds `owner` with `ALTER TABLE ... ADD COLUMN`, upgraded schemas place `owner` after the existing columns. Fresh-install DDL for v0.2.0 has been aligned to that order so Scenario A continues to compare `ordinal_position`. diff --git a/pg_durable.control b/pg_durable.control index b6e1a9e9..647f5fef 100644 --- a/pg_durable.control +++ b/pg_durable.control @@ -4,5 +4,10 @@ module_pathname = 'pg_durable' relocatable = false superuser = true trusted = false +# Note: 'schema' is intentionally omitted. This extension manages two schemas +# (df and duroxide), and PostgreSQL's control file only supports a single schema +# directive. The df schema is created by pgrx (#[pg_schema]), and the duroxide +# schema is created by sql/duroxide_install.sql. relocatable = false prevents +# schema relocation attacks. diff --git a/sql/pg_durable--0.1.1--0.2.0.sql b/sql/pg_durable--0.1.1--0.2.0.sql index 2fdf8029..ebf1feaa 100644 --- a/sql/pg_durable--0.1.1--0.2.0.sql +++ b/sql/pg_durable--0.1.1--0.2.0.sql @@ -36,3 +36,75 @@ CREATE POLICY vars_user_isolation ON df.vars FOR ALL USING (owner = current_user::regrole) WITH CHECK (owner = current_user::regrole); + +-- ============================================================================ +-- 3. Harden PL/pgSQL and SQL helper functions with SET search_path +-- (Defense-in-depth: all calls are already schema-qualified, but this +-- prevents future edits from accidentally introducing unqualified refs.) +-- ============================================================================ + +CREATE OR REPLACE FUNCTION df.as_op(fut text, name text) RETURNS text AS $$ + SELECT df.as(fut, name); +$$ LANGUAGE SQL IMMUTABLE SET search_path = pg_catalog, df, pg_temp; + +CREATE OR REPLACE FUNCTION df.if_then_op(condition text, then_branch text) RETURNS text AS $$ +DECLARE + cond_fut jsonb; + then_fut jsonb; + result_obj jsonb; +BEGIN + cond_fut := df.ensure_durofut(condition)::jsonb; + then_fut := df.ensure_durofut(then_branch)::jsonb; + result_obj := jsonb_build_object( + '_partial_if', true, + 'condition', cond_fut, + 'then_branch', then_fut + ); + RETURN result_obj::text; +END; +$$ LANGUAGE plpgsql IMMUTABLE SET search_path = pg_catalog, df, pg_temp; + +CREATE OR REPLACE FUNCTION df.if_else_op(partial_if text, else_branch text) RETURNS text AS $$ +DECLARE + partial jsonb; + else_fut text; + cond_text text; + then_text text; +BEGIN + partial := partial_if::jsonb; + IF partial->>'_partial_if' IS NULL THEN + RAISE EXCEPTION 'Invalid if-then-else: left side of !> must be a ?> expression'; + END IF; + cond_text := partial->'condition'::text; + then_text := partial->'then_branch'::text; + else_fut := df.ensure_durofut(else_branch); + RETURN df.if(cond_text, then_text, else_fut); +END; +$$ LANGUAGE plpgsql IMMUTABLE SET search_path = pg_catalog, df, pg_temp; + +CREATE OR REPLACE FUNCTION df.ensure_durofut(val text) RETURNS text AS $$ +DECLARE + node_type_val text; +BEGIN + BEGIN + node_type_val := (val::jsonb)->>'node_type'; + IF node_type_val IS NOT NULL THEN + IF node_type_val NOT IN ('SQL', 'THEN', 'IF', 'JOIN', 'LOOP', 'BREAK', 'RACE', 'SLEEP', 'WAIT_SCHEDULE', 'HTTP', 'SIGNAL') THEN + RAISE EXCEPTION 'Unknown node_type ''%''. Valid types: SQL, THEN, IF, JOIN, LOOP, BREAK, RACE, SLEEP, WAIT_SCHEDULE, HTTP, SIGNAL', node_type_val; + END IF; + RETURN val; + END IF; + EXCEPTION WHEN invalid_text_representation THEN + NULL; + WHEN raise_exception THEN + RAISE; + WHEN OTHERS THEN + NULL; + END; + RETURN df.sql(val); +END; +$$ LANGUAGE plpgsql IMMUTABLE SET search_path = pg_catalog, df, pg_temp; + +CREATE OR REPLACE FUNCTION df.loop_prefix_op(body text) RETURNS text AS $$ + SELECT df.loop(body); +$$ LANGUAGE SQL IMMUTABLE SET search_path = pg_catalog, df, pg_temp; diff --git a/src/activities/load_function_graph.rs b/src/activities/load_function_graph.rs index e4ea920e..df3e2a64 100644 --- a/src/activities/load_function_graph.rs +++ b/src/activities/load_function_graph.rs @@ -26,12 +26,13 @@ pub async fn execute( "Loading function graph for instance: {instance_id}" )); - let instance_query = format!("SELECT root_node FROM df.instances WHERE id = '{instance_id}'"); + let instance_query = "SELECT root_node FROM df.instances WHERE id = $1"; // Retry loop: wait for instance data to appear let start_time = std::time::Instant::now(); let root_node_id: String = loop { - match sqlx::query_scalar(&instance_query) + match sqlx::query_scalar::<_, String>(instance_query) + .bind(&instance_id) .fetch_one(pool.as_ref()) .await { @@ -54,16 +55,18 @@ pub async fn execute( } }; - let nodes_query = format!( - r#"SELECT id, node_type, query, result_name, + let nodes_query = r#"SELECT id, node_type, query, result_name, left_node, right_node, submitted_by::text AS submitted_by, login_role::text AS login_role, database - FROM df.nodes WHERE instance_id = '{instance_id}'"# - ); + FROM df.nodes WHERE instance_id = $1"#; - let rows = match sqlx::query(&nodes_query).fetch_all(pool.as_ref()).await { + let rows = match sqlx::query(nodes_query) + .bind(&instance_id) + .fetch_all(pool.as_ref()) + .await + { Ok(rows) => rows, Err(e) => return Err(format!("Failed to load function nodes: {e}")), }; diff --git a/src/dsl.rs b/src/dsl.rs index 0ecd562e..294024f8 100644 --- a/src/dsl.rs +++ b/src/dsl.rs @@ -2,6 +2,7 @@ use chrono::Utc; use cron::Schedule as CronSchedule; +use pgrx::datum::DatumWithOid; use pgrx::prelude::*; use std::str::FromStr; @@ -114,22 +115,14 @@ pub fn setvar(name: &str, value: &str) -> String { pgrx::error!("df.setvar() cannot be called inside a workflow - set variables before starting the workflow"); } - let escaped_name = name.replace('\'', "''"); - let escaped_value = value.replace('\'', "''"); let sql = if owner_scoped_vars_enabled() { - format!( - "INSERT INTO df.vars (name, value) VALUES ('{}', '{}') - ON CONFLICT (owner, name) DO UPDATE SET value = EXCLUDED.value", - escaped_name, escaped_value - ) + "INSERT INTO df.vars (name, value) VALUES ($1, $2) + ON CONFLICT (owner, name) DO UPDATE SET value = EXCLUDED.value" } else { - format!( - "INSERT INTO df.vars (name, value) VALUES ('{}', '{}') - ON CONFLICT (name) DO UPDATE SET value = EXCLUDED.value", - escaped_name, escaped_value - ) + "INSERT INTO df.vars (name, value) VALUES ($1, $2) + ON CONFLICT (name) DO UPDATE SET value = EXCLUDED.value" }; - if let Err(e) = Spi::run(&sql) { + if let Err(e) = Spi::run_with_args(sql, &[name.into(), value.into()]) { pgrx::error!("Failed to set variable: {:?}", e); } "OK".to_string() @@ -139,16 +132,14 @@ pub fn setvar(name: &str, value: &str) -> String { /// Returns the variable owned by the current user. #[pg_extern(schema = "df")] pub fn getvar(name: &str) -> Option { - let escaped_name = name.replace('\'', "''"); let sql = if owner_scoped_vars_enabled() { - format!( - "SELECT value FROM df.vars WHERE name = '{}' AND owner = current_user::regrole", - escaped_name - ) + "SELECT value FROM df.vars WHERE name = $1 AND owner = current_user::regrole" } else { - format!("SELECT value FROM df.vars WHERE name = '{}'", escaped_name) + "SELECT value FROM df.vars WHERE name = $1" }; - Spi::get_one::(&sql).ok().flatten() + Spi::get_one_with_args::(sql, &[name.into()]) + .ok() + .flatten() } /// Removes a workflow variable. @@ -160,16 +151,12 @@ pub fn unsetvar(name: &str) -> String { pgrx::error!("df.unsetvar() cannot be called inside a workflow - manage variables before starting the workflow"); } - let escaped_name = name.replace('\'', "''"); let sql = if owner_scoped_vars_enabled() { - format!( - "DELETE FROM df.vars WHERE name = '{}' AND owner = current_user::regrole", - escaped_name - ) + "DELETE FROM df.vars WHERE name = $1 AND owner = current_user::regrole" } else { - format!("DELETE FROM df.vars WHERE name = '{}'", escaped_name) + "DELETE FROM df.vars WHERE name = $1" }; - if let Err(e) = Spi::run(&sql) { + if let Err(e) = Spi::run_with_args(sql, &[name.into()]) { pgrx::error!("Failed to unset variable: {:?}", e); } "OK".to_string() @@ -547,10 +534,10 @@ pub fn signal(instance_id: &str, signal_name: &str, signal_data: default!(&str, // Ownership check: SPI goes through RLS, so this returns false for // non-owned instances (the row is invisible to the calling user). - let exists: bool = Spi::get_one(&format!( - "SELECT EXISTS(SELECT 1 FROM df.instances WHERE id = '{}')", - instance_id.replace('\'', "''") - )) + let exists: bool = Spi::get_one_with_args( + "SELECT EXISTS(SELECT 1 FROM df.instances WHERE id = $1)", + &[instance_id.into()], + ) .ok() .flatten() .unwrap_or(false); @@ -591,10 +578,10 @@ pub fn start( // Validate that the target database exists (if specified) if let Some(db) = database { - let exists: bool = match Spi::get_one(&format!( - "SELECT EXISTS(SELECT 1 FROM pg_database WHERE datname = '{}')", - db.replace('\'', "''") - )) { + let exists: bool = match Spi::get_one_with_args( + "SELECT EXISTS(SELECT 1 FROM pg_database WHERE datname = $1)", + &[db.into()], + ) { Ok(Some(v)) => v, Ok(None) => false, Err(e) => pgrx::error!("failed to check database existence: {}", e), @@ -608,19 +595,12 @@ pub fn start( let session_user_oid = unsafe { pgrx::pg_sys::GetSessionUserId() }; let outer_user_oid = unsafe { pgrx::pg_sys::GetOuterUserId() }; - let label_sql = label - .map(|l| format!("'{}'", l.replace('\'', "''"))) - .unwrap_or_else(|| "NULL".to_string()); - - let outer_oid_u32: u32 = outer_user_oid.into(); - let session_oid_u32: u32 = session_user_oid.into(); - // Insert all nodes from the nested graph into df.nodes, returning root node ID fn insert_nodes( node: &Durofut, instance_id: &str, - outer_user_oid: u32, - session_user_oid: u32, + outer_user_oid: pgrx::pg_sys::Oid, + session_user_oid: pgrx::pg_sys::Oid, database: Option<&str>, ) -> String { let node_id = short_id(); @@ -636,7 +616,7 @@ pub fn start( .map(|n| insert_nodes(n, instance_id, outer_user_oid, session_user_oid, database)); // Process config JSON to recursively insert embedded nodes and get their IDs - let query_escaped = match node.transform_config_children(|child| { + let query_val: Option = match node.transform_config_children(|child| { Ok(insert_nodes( child, instance_id, @@ -645,50 +625,49 @@ pub fn start( database, )) }) { - Ok(Some(updated_query)) => { - format!("'{}'", updated_query.replace('\'', "''")) - } - Ok(None) => "NULL".to_string(), + Ok(updated_query) => updated_query, Err(e) => pgrx::error!("Invalid config in {} node: {}", node.node_type, e), }; - let result_name_escaped = node - .result_name - .as_ref() - .map(|n| format!("'{}'", n.replace('\'', "''"))) - .unwrap_or_else(|| "NULL".to_string()); - - let left_node_escaped = left_id - .as_ref() - .map(|id| format!("'{id}'")) - .unwrap_or_else(|| "NULL".to_string()); - - let right_node_escaped = right_id - .as_ref() - .map(|id| format!("'{id}'")) - .unwrap_or_else(|| "NULL".to_string()); - - let database_escaped = database - .map(|db| format!("'{}'", db.replace('\'', "''"))) - .unwrap_or_else(|| "NULL".to_string()); + // Build parameterized args for the INSERT + let query_arg: DatumWithOid = match &query_val { + Some(q) => q.as_str().into(), + None => DatumWithOid::null::(), + }; + let result_name_arg: DatumWithOid = match &node.result_name { + Some(n) => n.as_str().into(), + None => DatumWithOid::null::(), + }; + let left_node_arg: DatumWithOid = match &left_id { + Some(id) => id.as_str().into(), + None => DatumWithOid::null::(), + }; + let right_node_arg: DatumWithOid = match &right_id { + Some(id) => id.as_str().into(), + None => DatumWithOid::null::(), + }; + let database_arg: DatumWithOid = match database { + Some(db) => db.into(), + None => DatumWithOid::null::(), + }; - // Insert this node with the generated ID - let insert_sql = format!( + // Insert this node with parameterized query + if let Err(e) = Spi::run_with_args( "INSERT INTO df.nodes (id, instance_id, node_type, query, result_name, left_node, right_node, submitted_by, login_role, database) - VALUES ('{}', '{}', '{}', {}, {}, {}, {}, {}::oid::regrole, {}::oid::regrole, {})", - node_id, - instance_id, - node.node_type.replace('\'', "''"), - query_escaped, - result_name_escaped, - left_node_escaped, - right_node_escaped, - outer_user_oid, - session_user_oid, - database_escaped - ); - - if let Err(e) = Spi::run(&insert_sql) { + VALUES ($1, $2, $3, $4, $5, $6, $7, $8::oid::regrole, $9::oid::regrole, $10)", + &[ + node_id.as_str().into(), + instance_id.into(), + node.node_type.as_str().into(), + query_arg, + result_name_arg, + left_node_arg, + right_node_arg, + outer_user_oid.into(), + session_user_oid.into(), + database_arg, + ], + ) { pgrx::error!("Failed to insert node {}: {:?}", node_id, e); } @@ -699,27 +678,33 @@ pub fn start( let root_node_id = insert_nodes( &durofut, &instance_id, - outer_oid_u32, - session_oid_u32, + outer_user_oid, + session_user_oid, database, ); - let database_sql = database - .map(|db| format!("'{}'", db.replace('\'', "''"))) - .unwrap_or_else(|| "NULL".to_string()); + // Build parameterized args for the instance INSERT + let label_arg: DatumWithOid = match label { + Some(l) => l.into(), + None => DatumWithOid::null::(), + }; + let database_arg: DatumWithOid = match database { + Some(db) => db.into(), + None => DatumWithOid::null::(), + }; // Create instance record with root node ID - let create_instance_sql = format!( - "INSERT INTO df.instances (id, label, root_node, status, submitted_by, login_role, database) VALUES ('{}', {}, '{}', 'pending', {}::oid::regrole, {}::oid::regrole, {})", - instance_id, - label_sql, - root_node_id, - outer_oid_u32, - session_oid_u32, - database_sql - ); - - if let Err(e) = Spi::run(&create_instance_sql) { + if let Err(e) = Spi::run_with_args( + "INSERT INTO df.instances (id, label, root_node, status, submitted_by, login_role, database) VALUES ($1, $2, $3, 'pending', $4::oid::regrole, $5::oid::regrole, $6)", + &[ + instance_id.as_str().into(), + label_arg, + root_node_id.as_str().into(), + outer_user_oid.into(), + session_user_oid.into(), + database_arg, + ], + ) { pgrx::error!("Failed to create instance: {:?}", e); } @@ -775,10 +760,10 @@ pub fn cancel(instance_id: &str, reason: default!(&str, "'Cancelled by user'")) // Ownership check: SPI goes through RLS, so this returns false for // non-owned instances (the row is invisible to the calling user). - let exists: bool = Spi::get_one(&format!( - "SELECT EXISTS(SELECT 1 FROM df.instances WHERE id = '{}')", - instance_id.replace('\'', "''") - )) + let exists: bool = Spi::get_one_with_args( + "SELECT EXISTS(SELECT 1 FROM df.instances WHERE id = $1)", + &[instance_id.into()], + ) .ok() .flatten() .unwrap_or(false); @@ -792,10 +777,10 @@ pub fn cancel(instance_id: &str, reason: default!(&str, "'Cancelled by user'")) // Update the instance status to 'cancelled' via SPI. // User has column-level UPDATE on (status, updated_at) with RLS restricting to own rows. - Spi::run(&format!( - "UPDATE df.instances SET status = 'cancelled', updated_at = now() WHERE id = '{}'", - instance_id.replace('\'', "''") - )) + Spi::run_with_args( + "UPDATE df.instances SET status = 'cancelled', updated_at = now() WHERE id = $1", + &[instance_id.into()], + ) .unwrap_or_else(|e| warning!("Failed to update instance status: {e}")); format!("Instance {instance_id} cancelled: {reason}") @@ -804,8 +789,12 @@ pub fn cancel(instance_id: &str, reason: default!(&str, "'Cancelled by user'")) /// Gets the status of a durable function instance. #[pg_extern(schema = "df")] pub fn status(instance_id: &str) -> Option { - let sql = format!("SELECT status FROM df.instances WHERE id = '{instance_id}'"); - Spi::get_one::(&sql).ok().flatten() + Spi::get_one_with_args::( + "SELECT status FROM df.instances WHERE id = $1", + &[instance_id.into()], + ) + .ok() + .flatten() } /// Manually runs pending durable functions. @@ -821,12 +810,14 @@ pub fn run(instance_id: default!(Option<&str>, "NULL")) -> String { /// Gets the result of a completed durable function. #[pg_extern(schema = "df")] pub fn result(instance_id: &str) -> Option { - let sql = format!( - r#"SELECT result::text FROM df.nodes - WHERE id = (SELECT root_node FROM df.instances WHERE id = '{instance_id}') - AND status = 'completed'"# - ); - Spi::get_one::(&sql).ok().flatten() + Spi::get_one_with_args::( + r#"SELECT result::text FROM df.nodes + WHERE id = (SELECT root_node FROM df.instances WHERE id = $1) + AND status = 'completed'"#, + &[instance_id.into()], + ) + .ok() + .flatten() } /// Waits for a durable function to complete, returning its final status. @@ -859,13 +850,11 @@ pub fn wait_for_completion( loop { // Query instance status - let sql = format!( - "SELECT status FROM df.instances WHERE id = '{}'", - instance_id.replace('\'', "''") - ); - - let status: Option = - Spi::get_one(&sql).map_err(|e| format!("Failed to query status: {:?}", e))?; + let status: Option = Spi::get_one_with_args( + "SELECT status FROM df.instances WHERE id = $1", + &[instance_id.into()], + ) + .map_err(|e| format!("Failed to query status: {:?}", e))?; if let Some(ref s) = status { let s_lower = s.to_lowercase(); diff --git a/src/explain.rs b/src/explain.rs index cccafb60..b1bf1483 100644 --- a/src/explain.rs +++ b/src/explain.rs @@ -48,11 +48,8 @@ pub fn explain(input: &str) -> String { fn explain_instance(instance_id: &str) -> String { // Get instance info from PostgreSQL let instance_info: Option<(String, Option, String)> = Spi::connect(|client| { - let sql = format!( - "SELECT root_node, label, status FROM df.instances WHERE id = '{}'", - instance_id.replace('\'', "''") - ); - if let Ok(table) = client.select(&sql, None, &[]) { + let sql = "SELECT root_node, label, status FROM df.instances WHERE id = $1"; + if let Ok(table) = client.select(sql, None, &[instance_id.into()]) { for row in table { let root_node: Option = row.get(1).ok().flatten(); let label: Option = row.get(2).ok().flatten(); @@ -296,23 +293,28 @@ fn collect_nodes( /// Load nodes from a table into a HashMap fn load_nodes_from_table(table: &str, instance_id: Option<&str>) -> HashMap { - let sql = if let Some(id) = instance_id { - format!( - r#"SELECT id, node_type, query, result_name, left_node, right_node, status, result::text - FROM {} WHERE instance_id = '{}'"#, - table, - id.replace('\'', "''") - ) - } else { - format!( - "SELECT id, node_type, query, result_name, left_node, right_node, status, result::text FROM {table}" - ) - }; - + // Note: table name is always a hardcoded value ("df.nodes") from internal callers, + // so it is safe to interpolate. Only instance_id is parameterized. let mut nodes = HashMap::new(); Spi::connect(|client| { - if let Ok(table_result) = client.select(&sql, None, &[]) { + let (sql, args): (String, Vec) = if let Some(id) = instance_id { + ( + format!( + "SELECT id, node_type, query, result_name, left_node, right_node, status, result::text FROM {} WHERE instance_id = $1", + table + ), + vec![id.into()], + ) + } else { + ( + format!( + "SELECT id, node_type, query, result_name, left_node, right_node, status, result::text FROM {table}" + ), + vec![], + ) + }; + if let Ok(table_result) = client.select(&sql, None, &args) { for row in table_result { if let Ok(Some(id)) = row.get::(1) { let node = ExplainNode { diff --git a/src/lib.rs b/src/lib.rs index da439c35..cf74623d 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -300,7 +300,7 @@ CREATE OPERATOR ~> ( -- Operator |=> for naming: fut |=> 'name' means "name this result as $name" CREATE OR REPLACE FUNCTION df.as_op(fut text, name text) RETURNS text AS $$ SELECT df.as(fut, name); -$$ LANGUAGE SQL IMMUTABLE; +$$ LANGUAGE SQL IMMUTABLE SET search_path = pg_catalog, df, pg_temp; CREATE OPERATOR |=> ( FUNCTION = df.as_op, @@ -344,7 +344,7 @@ BEGIN ); RETURN result_obj::text; END; -$$ LANGUAGE plpgsql IMMUTABLE; +$$ LANGUAGE plpgsql IMMUTABLE SET search_path = pg_catalog, df, pg_temp; -- Helper: partial_if !> else completes the if node CREATE OR REPLACE FUNCTION df.if_else_op(partial_if text, else_branch text) RETURNS text AS $$ @@ -368,7 +368,7 @@ BEGIN -- Now call the real df.if function RETURN df.if(cond_text, then_text, else_fut); END; -$$ LANGUAGE plpgsql IMMUTABLE; +$$ LANGUAGE plpgsql IMMUTABLE SET search_path = pg_catalog, df, pg_temp; -- Helper to ensure a value is a durofut (returns JSON string) -- Rejects JSON with unknown node_type values. @@ -402,7 +402,7 @@ BEGIN -- It's plain SQL, wrap it RETURN df.sql(val); END; -$$ LANGUAGE plpgsql IMMUTABLE; +$$ LANGUAGE plpgsql IMMUTABLE SET search_path = pg_catalog, df, pg_temp; CREATE OPERATOR ?> ( FUNCTION = df.if_then_op, @@ -420,7 +420,7 @@ CREATE OPERATOR !> ( -- This is a PREFIX operator with lowest precedence CREATE OR REPLACE FUNCTION df.loop_prefix_op(body text) RETURNS text AS $$ SELECT df.loop(body); -$$ LANGUAGE SQL IMMUTABLE; +$$ LANGUAGE SQL IMMUTABLE SET search_path = pg_catalog, df, pg_temp; CREATE OPERATOR @> ( FUNCTION = df.loop_prefix_op, diff --git a/src/monitoring.rs b/src/monitoring.rs index 55f9b86d..d8de2288 100644 --- a/src/monitoring.rs +++ b/src/monitoring.rs @@ -39,20 +39,21 @@ pub fn list_instances( // Query df.instances via SPI first — RLS filters to calling user's rows only. // This gives us the user's own instance IDs and labels. let user_instances: Vec<(String, Option)> = Spi::connect(|client| { - let sql = if let Some(status) = status_filter { - format!( - "SELECT id, label FROM df.instances WHERE status = '{}' ORDER BY created_at DESC LIMIT {}", - status.replace('\'', "''"), - limit_count + use pgrx::datum::DatumWithOid; + + let (sql, args): (&str, Vec) = if let Some(status) = status_filter { + ( + "SELECT id, label FROM df.instances WHERE status = $1 ORDER BY created_at DESC LIMIT $2", + vec![status.into(), (limit_count as i64).into()], ) } else { - format!( - "SELECT id, label FROM df.instances ORDER BY created_at DESC LIMIT {}", - limit_count + ( + "SELECT id, label FROM df.instances ORDER BY created_at DESC LIMIT $1", + vec![(limit_count as i64).into()], ) }; let mut instances = Vec::new(); - if let Ok(table) = client.select(&sql, None, &[]) { + if let Ok(table) = client.select(sql, None, &args) { for row in table { if let Ok(Some(id)) = row.get::(1) { let label: Option = row.get(2).ok().flatten(); @@ -125,18 +126,18 @@ pub fn instance_info( let instance_id_str = instance_id.to_string(); // Ownership check: SPI goes through RLS, returning NULL for non-owned instances. - let label: Option = Spi::get_one(&format!( - "SELECT label FROM df.instances WHERE id = '{}'", - instance_id.replace('\'', "''") - )) + let label: Option = Spi::get_one_with_args( + "SELECT label FROM df.instances WHERE id = $1", + &[instance_id.into()], + ) .ok() .flatten(); // Check if the instance exists for this user (RLS-filtered) - let exists: bool = Spi::get_one(&format!( - "SELECT EXISTS(SELECT 1 FROM df.instances WHERE id = '{}')", - instance_id.replace('\'', "''") - )) + let exists: bool = Spi::get_one_with_args( + "SELECT EXISTS(SELECT 1 FROM df.instances WHERE id = $1)", + &[instance_id.into()], + ) .ok() .flatten() .unwrap_or(false); @@ -199,10 +200,10 @@ pub fn instance_executions( let instance_id_owned = instance_id.to_string(); // Ownership check: SPI goes through RLS, so non-owned instances are invisible. - let exists: bool = Spi::get_one(&format!( - "SELECT EXISTS(SELECT 1 FROM df.instances WHERE id = '{}')", - instance_id.replace('\'', "''") - )) + let exists: bool = Spi::get_one_with_args( + "SELECT EXISTS(SELECT 1 FROM df.instances WHERE id = $1)", + &[instance_id.into()], + ) .ok() .flatten() .unwrap_or(false); @@ -347,12 +348,10 @@ pub fn instance_nodes( Option, Option, )> = Spi::connect(|client| { - let sql = format!( - r#"SELECT id, node_type, query, result_name, left_node, right_node, status, result::text, updated_at - FROM df.nodes WHERE instance_id = '{instance_id}'"# - ); + let sql = r#"SELECT id, node_type, query, result_name, left_node, right_node, status, result::text, updated_at + FROM df.nodes WHERE instance_id = $1"#; let mut nodes = Vec::new(); - if let Ok(table) = client.select(&sql, None, &[]) { + if let Ok(table) = client.select(sql, None, &[instance_id.as_str().into()]) { for row in table { if let Ok(Some(id)) = row.get::(1) { let node_type: String = row.get(2).ok().flatten().unwrap_or_default(); diff --git a/tests/e2e/sql/00_setup_playground.sql b/tests/e2e/sql/00_setup_playground.sql index c06357d7..a50f5832 100644 --- a/tests/e2e/sql/00_setup_playground.sql +++ b/tests/e2e/sql/00_setup_playground.sql @@ -50,6 +50,34 @@ BEGIN RAISE NOTICE 'Worker epoch sentinel detected — full restart cycle complete'; END $$; +-- --------------------------------------------------------------------------- +-- Reusable helper: DROP EXTENSION pg_durable with deadlock retry. +-- +-- After a durable function completes, the duroxide runtime may still be +-- acknowledging the orchestration item (ack_orchestration_item). If DROP +-- EXTENSION tries to take AccessExclusiveLock on those tables at the same +-- time, PostgreSQL detects a deadlock. This helper retries on deadlock. +-- --------------------------------------------------------------------------- +CREATE OR REPLACE FUNCTION public._e2e_drop_extension_safe() +RETURNS void +LANGUAGE plpgsql AS $$ +DECLARE + attempts INT := 0; +BEGIN + LOOP + BEGIN + EXECUTE 'DROP EXTENSION IF EXISTS pg_durable CASCADE'; + RETURN; + EXCEPTION WHEN deadlock_detected THEN + attempts := attempts + 1; + IF attempts >= 5 THEN + RAISE; + END IF; + PERFORM pg_sleep(1); + END; + END LOOP; +END $$; + -- Install extensions needed by tests (requires superuser) CREATE EXTENSION IF NOT EXISTS dblink; diff --git a/tests/e2e/sql/25_extension_creation_security.sql b/tests/e2e/sql/25_extension_creation_security.sql index dfcb26f0..9cce2eba 100644 --- a/tests/e2e/sql/25_extension_creation_security.sql +++ b/tests/e2e/sql/25_extension_creation_security.sql @@ -6,7 +6,7 @@ -- Note: This test drops and recreates the extension to test installation security -- Any running instances will be lost, but E2E tests are self-contained -DROP EXTENSION IF EXISTS pg_durable CASCADE; +SELECT public._e2e_drop_extension_safe(); -- ============================================================================ -- Test 1: Non-superuser cannot create extension diff --git a/tests/e2e/sql/28_bgw_lifecycle.sql b/tests/e2e/sql/28_bgw_lifecycle.sql index 1aacf56f..96336364 100644 --- a/tests/e2e/sql/28_bgw_lifecycle.sql +++ b/tests/e2e/sql/28_bgw_lifecycle.sql @@ -7,7 +7,7 @@ -- 4) After re-create, workflows execute again. -- Ensure a clean starting point -DROP EXTENSION IF EXISTS pg_durable CASCADE; +SELECT public._e2e_drop_extension_safe(); -- 1) Verify BGW does not create duroxide schema pre-extension DO $$ @@ -66,7 +66,7 @@ END $$; DROP TABLE _test_state; -- 3) Drop extension and verify schema removed -DROP EXTENSION pg_durable CASCADE; +SELECT public._e2e_drop_extension_safe(); DO $$ DECLARE diff --git a/tests/e2e/sql/29_database_validation.sql b/tests/e2e/sql/29_database_validation.sql index f73c2163..c0ca3e0c 100644 --- a/tests/e2e/sql/29_database_validation.sql +++ b/tests/e2e/sql/29_database_validation.sql @@ -38,7 +38,7 @@ END $$; -- ============================================================================ -- Test 1: CREATE EXTENSION should succeed in the correct database -- ============================================================================ -DROP EXTENSION IF EXISTS pg_durable CASCADE; +SELECT public._e2e_drop_extension_safe(); CREATE EXTENSION pg_durable; -- Wait for the background worker to fully reinitialize