Address SQL injection and other threats

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<String> = 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<String> = 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`)
 
 ---
 

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`.

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.
 
 

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;

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}")),
     };