fix(security): resolve SQL injection risks in PostgreSQL backend

.github/workflows/test.yml

@@ -79,7 +79,7 @@ jobs:
 
       - name: Install activerecord and testsuite from PyPI/git
         run: |
-          pip install rhosocial-activerecord==1.0.0.dev25
+          pip install git+https://github.com/rhosocial/python-activerecord.git@release/v1.0.0.dev26#egg=rhosocial-activerecord
           pip install rhosocial-activerecord-testsuite==1.0.0.dev12
 
       - name: Create postgres scenario config

changelog.d/35.security.md

@@ -0,0 +1 @@
+**SECURITY**: Resolved SQL injection risks in PostgreSQL backend

src/rhosocial/activerecord/backend/impl/postgres/mixins/ddl/index.py

@@ -5,7 +5,7 @@
 PostgreSQL-specific index features and operations.
 """
 
-from typing import Any, Dict, Optional, Tuple, List, TYPE_CHECKING
+from typing import Any, Dict, Optional, Tuple, List, Union, TYPE_CHECKING
 
 from rhosocial.activerecord.backend.expression.bases import ToSQLProtocol
 
@@ -247,13 +247,20 @@ def format_create_index_pg_statement(
         include_columns: Optional[List[str]] = None,
         with_options: Optional[Dict[str, Any]] = None,
         tablespace: Optional[str] = None,
-        where_clause: Optional[str] = None,
+        where_clause: Optional[Union[str, ToSQLProtocol]] = None,
     ) -> Tuple[str, tuple]:
         """Format CREATE INDEX statement with PostgreSQL-specific options.
 
         This method provides a convenient way to create indexes with
         PostgreSQL-specific options without using expression objects.
+
+        Args:
+            where_clause: Accepts either a string (for hardcoded conditions)
+                or a ToSQLProtocol expression for parameterized queries.
+                WARNING: When using string, never pass user input directly;
+                always use ToSQLProtocol expressions for untrusted input.
         """
+        all_params = []
         parts = ["CREATE"]
 
         if unique:
@@ -312,6 +319,11 @@ def format_create_index_pg_statement(
 
         # WHERE clause for partial index
         if where_clause:
-            parts.append(f"WHERE {where_clause}")
+            if isinstance(where_clause, ToSQLProtocol):
+                where_sql, where_params = where_clause.to_sql()
+                parts.append(f"WHERE {where_sql}")
+                all_params = list(all_params) + list(where_params)
+            else:
+                parts.append(f"WHERE {where_clause}")
 
-        return (" ".join(parts), ())
+        return (" ".join(parts), tuple(all_params))

src/rhosocial/activerecord/backend/impl/postgres/mixins/ddl/trigger.py

@@ -66,7 +66,7 @@ def format_create_trigger_statement(self, expr) -> Tuple[str, tuple]:
             all_params.extend(cond_params)
 
         parts.append("EXECUTE FUNCTION")
-        parts.append(expr.function_name + "()")
+        parts.append(f"{self.format_identifier(expr.function_name)}()")
 
         return " ".join(parts), tuple(all_params)
 

src/rhosocial/activerecord/backend/impl/postgres/mixins/ddl/type.py

@@ -51,7 +51,7 @@ def format_create_type_enum_statement(
         if schema:
             type_name = f"{self.format_identifier(schema)}.{type_name}"
 
-        escaped_values = [f"'{v}'" for v in values]
+        escaped_values = [f"'{self._escape_sql_string(str(v))}'" for v in values]
         values_str = ", ".join(escaped_values)
 
         sql = f"CREATE TYPE {type_name} AS ENUM ({values_str})"

src/rhosocial/activerecord/backend/impl/postgres/mixins/dml/extended_statistics.py

@@ -46,8 +46,8 @@ def format_create_statistics_statement(self, expr: "PostgresCreateStatisticsExpr
         if not self.supports_create_statistics():
             raise ValueError("CREATE STATISTICS requires PostgreSQL 10+")
 
-        full_name = f"{expr.schema}.{expr.name}" if expr.schema else expr.name
-        table_full = f"{expr.schema}.{expr.table_name}" if expr.schema else expr.table_name
+        full_name = f"{self.format_identifier(expr.schema)}.{self.format_identifier(expr.name)}" if expr.schema else self.format_identifier(expr.name)
+        table_full = f"{self.format_identifier(expr.schema)}.{self.format_identifier(expr.table_name)}" if expr.schema else self.format_identifier(expr.table_name)
 
         exists_clause = "IF NOT EXISTS " if expr.if_not_exists else ""
 
@@ -78,7 +78,7 @@ def format_drop_statistics_statement(self, expr: "PostgresDropStatisticsExpressi
         Returns:
             Tuple of (SQL statement, parameters tuple)
         """
-        full_name = f"{expr.schema}.{expr.name}" if expr.schema else expr.name
+        full_name = f"{self.format_identifier(expr.schema)}.{self.format_identifier(expr.name)}" if expr.schema else self.format_identifier(expr.name)
         exists_clause = "IF EXISTS " if expr.if_exists else ""
 
         sql = f"DROP STATISTICS {exists_clause}{full_name}"

src/rhosocial/activerecord/backend/impl/postgres/mixins/dml/stored_procedure.py

@@ -63,7 +63,7 @@ def format_create_procedure_statement(
         if not self.supports_call_statement():
             raise ValueError("CREATE PROCEDURE requires PostgreSQL 11+")
 
-        full_name = f"{schema}.{name}" if schema else name
+        full_name = f"{self.format_identifier(schema)}.{self.format_identifier(name)}" if schema else self.format_identifier(name)
 
         # OR REPLACE clause
         replace_clause = "OR REPLACE " if or_replace else ""
@@ -126,7 +126,7 @@ def format_drop_procedure_statement(
         Returns:
             Tuple of (SQL statement, parameters tuple)
         """
-        full_name = f"{schema}.{name}" if schema else name
+        full_name = f"{self.format_identifier(schema)}.{self.format_identifier(name)}" if schema else self.format_identifier(name)
 
         exists_clause = "IF EXISTS " if if_exists else ""
 
@@ -158,7 +158,7 @@ def format_call_statement(
         if not self.supports_call_statement():
             raise ValueError("CALL statement requires PostgreSQL 11+")
 
-        full_name = f"{schema}.{name}" if schema else name
+        full_name = f"{self.format_identifier(schema)}.{self.format_identifier(name)}" if schema else self.format_identifier(name)
 
         if arguments:
             placeholders = ", ".join(["%s"] * len(arguments))

tests/rhosocial/activerecord_postgres_test/feature/backend/postgres/test_extended_statistics.py

@@ -84,9 +84,9 @@ def test_create_statistics_basic(self):
             table_name="test_table"
         )
         sql, params = dialect.format_create_statistics_statement(expr)
-        assert "CREATE STATISTICS test_stats" in sql
+        assert 'CREATE STATISTICS "test_stats"' in sql
         assert "ON col1, col2" in sql
-        assert "FROM test_table" in sql
+        assert 'FROM "test_table"' in sql
         assert params == ()
 
     def test_create_statistics_with_schema(self):
@@ -100,8 +100,8 @@ def test_create_statistics_with_schema(self):
             schema="public"
         )
         sql, params = dialect.format_create_statistics_statement(expr)
-        assert "public.test_stats" in sql
-        assert "FROM public.test_table" in sql
+        assert '"public"."test_stats"' in sql
+        assert 'FROM "public"."test_table"' in sql
 
     def test_create_statistics_if_not_exists(self):
         """Test CREATE STATISTICS IF NOT EXISTS."""
@@ -114,7 +114,7 @@ def test_create_statistics_if_not_exists(self):
             if_not_exists=True
         )
         sql, params = dialect.format_create_statistics_statement(expr)
-        assert "CREATE STATISTICS IF NOT EXISTS test_stats" in sql
+        assert 'CREATE STATISTICS IF NOT EXISTS "test_stats"' in sql
 
     def test_create_statistics_with_ndistinct(self):
         """Test CREATE STATISTICS with ndistinct type."""
@@ -127,7 +127,7 @@ def test_create_statistics_with_ndistinct(self):
             statistics_type="ndistinct"
         )
         sql, params = dialect.format_create_statistics_statement(expr)
-        assert "test_stats(ndistinct)" in sql
+        assert '"test_stats"(ndistinct)' in sql
 
     def test_create_statistics_with_dependencies(self):
         """Test CREATE STATISTICS with dependencies type."""
@@ -140,7 +140,7 @@ def test_create_statistics_with_dependencies(self):
             statistics_type="dependencies"
         )
         sql, params = dialect.format_create_statistics_statement(expr)
-        assert "test_stats(dependencies)" in sql
+        assert '"test_stats"(dependencies)' in sql
 
     def test_create_statistics_with_mcv_pg12(self):
         """Test CREATE STATISTICS with MCV type on PostgreSQL 12."""
@@ -153,7 +153,7 @@ def test_create_statistics_with_mcv_pg12(self):
             statistics_type="mcv"
         )
         sql, params = dialect.format_create_statistics_statement(expr)
-        assert "test_stats(mcv)" in sql
+        assert '"test_stats"(mcv)' in sql
 
     def test_create_statistics_with_mcv_pg11_raises_error(self):
         """Test CREATE STATISTICS with MCV type on PostgreSQL 11 raises error."""
@@ -193,7 +193,7 @@ def test_drop_statistics_basic(self):
             name="test_stats"
         )
         sql, params = dialect.format_drop_statistics_statement(expr)
-        assert sql == "DROP STATISTICS test_stats"
+        assert sql == 'DROP STATISTICS "test_stats"'
         assert params == ()
 
     def test_drop_statistics_with_schema(self):
@@ -205,7 +205,7 @@ def test_drop_statistics_with_schema(self):
             schema="public"
         )
         sql, params = dialect.format_drop_statistics_statement(expr)
-        assert "DROP STATISTICS public.test_stats" in sql
+        assert 'DROP STATISTICS "public"."test_stats"' in sql
 
     def test_drop_statistics_if_exists(self):
         """Test DROP STATISTICS IF EXISTS."""
@@ -216,4 +216,4 @@ def test_drop_statistics_if_exists(self):
             if_exists=True
         )
         sql, params = dialect.format_drop_statistics_statement(expr)
-        assert "DROP STATISTICS IF EXISTS test_stats" in sql
+        assert 'DROP STATISTICS IF EXISTS "test_stats"' in sql

tests/rhosocial/activerecord_postgres_test/feature/backend/postgres/test_stored_procedure.py

@@ -65,7 +65,7 @@ def test_create_procedure_basic(self):
             name="test_proc",
             body="BEGIN NULL; END"
         )
-        assert "CREATE PROCEDURE test_proc" in sql
+        assert "CREATE PROCEDURE \"test_proc\"" in sql
         assert "LANGUAGE plpgsql" in sql
         assert "BEGIN NULL; END" in sql
         assert params == ()
@@ -78,7 +78,7 @@ def test_create_procedure_with_schema(self):
             schema="public",
             body="BEGIN NULL; END"
         )
-        assert "public.test_proc" in sql
+        assert '"public"."test_proc"' in sql
 
     def test_create_procedure_with_or_replace(self):
         """Test CREATE OR REPLACE PROCEDURE statement."""
@@ -181,7 +181,7 @@ def test_drop_procedure_basic(self):
         sql, params = dialect.format_drop_procedure_statement(
             name="test_proc"
         )
-        assert sql == "DROP PROCEDURE test_proc"
+        assert sql == "DROP PROCEDURE \"test_proc\""
         assert params == ()
 
     def test_drop_procedure_with_schema(self):
@@ -191,7 +191,7 @@ def test_drop_procedure_with_schema(self):
             name="test_proc",
             schema="public"
         )
-        assert "DROP PROCEDURE public.test_proc" in sql
+        assert 'DROP PROCEDURE "public"."test_proc"' in sql
 
     def test_drop_procedure_if_exists(self):
         """Test DROP PROCEDURE IF EXISTS."""
@@ -200,7 +200,7 @@ def test_drop_procedure_if_exists(self):
             name="test_proc",
             if_exists=True
         )
-        assert "DROP PROCEDURE IF EXISTS test_proc" in sql
+        assert "DROP PROCEDURE IF EXISTS \"test_proc\"" in sql
 
     def test_drop_procedure_cascade(self):
         """Test DROP PROCEDURE CASCADE."""
@@ -209,7 +209,7 @@ def test_drop_procedure_cascade(self):
             name="test_proc",
             cascade=True
         )
-        assert "DROP PROCEDURE test_proc CASCADE" in sql
+        assert "DROP PROCEDURE \"test_proc\" CASCADE" in sql
 
     def test_drop_procedure_with_parameter_types(self):
         """Test DROP PROCEDURE with parameter types for overload resolution."""
@@ -221,7 +221,7 @@ def test_drop_procedure_with_parameter_types(self):
                 {"type": "varchar"}
             ]
         )
-        assert "test_proc(integer, varchar)" in sql
+        assert '"test_proc"(integer, varchar)' in sql
 
     def test_drop_procedure_with_parameter_type_strings(self):
         """Test DROP PROCEDURE with parameter types as strings."""
@@ -230,7 +230,7 @@ def test_drop_procedure_with_parameter_type_strings(self):
             name="test_proc",
             parameters=["integer", "varchar"]
         )
-        assert "test_proc(integer, varchar)" in sql
+        assert '"test_proc"(integer, varchar)' in sql
 
 
 class TestFormatCallStatement:
@@ -246,7 +246,7 @@ def test_call_statement_no_args(self):
         """Test CALL statement without arguments."""
         dialect = PostgresDialect((14, 0, 0))
         sql, params = dialect.format_call_statement(name="test_proc")
-        assert sql == "CALL test_proc()"
+        assert sql == 'CALL "test_proc"()'
         assert params == ()
 
     def test_call_statement_with_args(self):
@@ -256,7 +256,7 @@ def test_call_statement_with_args(self):
             name="test_proc",
             arguments=[1, "test", None]
         )
-        assert "CALL test_proc(%s, %s, %s)" in sql
+        assert 'CALL "test_proc"(%s, %s, %s)' in sql
         assert params == (1, "test", None)
 
     def test_call_statement_with_schema(self):
@@ -266,4 +266,4 @@ def test_call_statement_with_schema(self):
             name="test_proc",
             schema="public"
         )
-        assert "CALL public.test_proc()" in sql
+        assert 'CALL "public"."test_proc"()' in sql

tests/rhosocial/activerecord_postgres_test/feature/backend/postgres/test_trigger.py

@@ -75,7 +75,7 @@ def test_create_trigger_basic_before(self):
         assert 'CREATE TRIGGER "update_timestamp"' in sql
         assert "BEFORE UPDATE" in sql
         assert 'ON "users"' in sql
-        assert "EXECUTE FUNCTION update_updated_at_column()" in sql
+        assert 'EXECUTE FUNCTION "update_updated_at_column"()' in sql
 
     def test_create_trigger_basic_after(self):
         """Test basic CREATE TRIGGER AFTER statement."""