fix(api): [security] use psycopg2.sql to safely compose DDL in PostgresEnumMigration (#10166)

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-authored-by: Adrián Peña <adrianjpr@gmail.com>
Co-authored-by: Pepe Fagoaga <pepe@prowler.com>

Author: 4 people

api/CHANGELOG.md

@@ -13,6 +13,10 @@ All notable changes to the **Prowler API** are documented in this file.
 - Attack Paths: Complete migration to private graph labels and properties, removing deprecated dual-write support [(#10268)](https://github.com/prowler-cloud/prowler/pull/10268)
 - Attack Paths: Added tenant and provider related labels to the nodes so they can be easily filtered on custom queries [(#10308)](https://github.com/prowler-cloud/prowler/pull/10308)
 
+### 🔐 Security
+
+- Use `psycopg2.sql` to safely compose DDL in `PostgresEnumMigration`, preventing SQL injection via f-string interpolation [(#10166)](https://github.com/prowler-cloud/prowler/pull/10166)
+
 ---
 
 ## [1.21.0] (Prowler v5.20.0)

api/src/backend/api/db_utils.py

@@ -18,6 +18,7 @@
 )
 from django_celery_beat.models import PeriodicTask
 from psycopg2 import connect as psycopg2_connect
+from psycopg2 import sql as psycopg2_sql
 from psycopg2.extensions import AsIs, new_type, register_adapter, register_type
 from rest_framework_json_api.serializers import ValidationError
 
@@ -280,15 +281,23 @@ def __init__(self, enum_name: str, enum_values: tuple):
         self.enum_values = enum_values
 
     def create_enum_type(self, apps, schema_editor):  # noqa: F841
-        string_enum_values = ", ".join([f"'{value}'" for value in self.enum_values])
         with schema_editor.connection.cursor() as cursor:
             cursor.execute(
-                f"CREATE TYPE {self.enum_name} AS ENUM ({string_enum_values});"
+                psycopg2_sql.SQL("CREATE TYPE {} AS ENUM ({})").format(
+                    psycopg2_sql.Identifier(self.enum_name),
+                    psycopg2_sql.SQL(", ").join(
+                        psycopg2_sql.Literal(v) for v in self.enum_values
+                    ),
+                )
             )
 
     def drop_enum_type(self, apps, schema_editor):  # noqa: F841
         with schema_editor.connection.cursor() as cursor:
-            cursor.execute(f"DROP TYPE {self.enum_name};")
+            cursor.execute(
+                psycopg2_sql.SQL("DROP TYPE {}").format(
+                    psycopg2_sql.Identifier(self.enum_name)
+                )
+            )
 
 
 class PostgresEnumField(models.Field):

api/src/backend/api/tests/test_db_utils.py

@@ -6,10 +6,12 @@
 from django.conf import settings
 from django.db import DEFAULT_DB_ALIAS, OperationalError
 from freezegun import freeze_time
+from psycopg2 import sql as psycopg2_sql
 from rest_framework_json_api.serializers import ValidationError
 
 from api.db_utils import (
     POSTGRES_TENANT_VAR,
+    PostgresEnumMigration,
     _should_create_index_on_partition,
     batch_delete,
     create_objects_in_batches,
@@ -910,3 +912,61 @@ def test_rls_transaction_cursor_yielded_correctly(self, tenants_fixture):
             cursor.execute("SELECT 1")
             result = cursor.fetchone()
             assert result[0] == 1
+
+
+class TestPostgresEnumMigration:
+    """
+    Verify that PostgresEnumMigration builds DDL statements via psycopg2.sql
+    so that enum type names and values are always properly quoted — preventing
+    SQL injection through f-string interpolation.
+    """
+
+    def _make_mock_schema_editor(self):
+        mock_cursor = MagicMock()
+        mock_conn = MagicMock()
+        mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
+        mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
+        mock_schema_editor = MagicMock()
+        mock_schema_editor.connection = mock_conn
+        return mock_schema_editor, mock_cursor
+
+    def test_create_enum_type_generates_correct_sql(self):
+        """create_enum_type builds a proper CREATE TYPE … AS ENUM via psycopg2.sql."""
+        migration = PostgresEnumMigration("my_enum", ("val_a", "val_b"))
+        schema_editor, mock_cursor = self._make_mock_schema_editor()
+
+        migration.create_enum_type(apps=None, schema_editor=schema_editor)
+
+        mock_cursor.execute.assert_called_once()
+        query_arg = mock_cursor.execute.call_args[0][0]
+        assert isinstance(
+            query_arg, psycopg2_sql.Composable
+        ), "create_enum_type must pass a psycopg2.sql.Composable, not a raw string."
+        # Verify the composed SQL structure: CREATE TYPE <Identifier> AS ENUM (<Literals>)
+        parts = query_arg.seq
+        assert parts[0] == psycopg2_sql.SQL("CREATE TYPE ")
+        assert isinstance(parts[1], psycopg2_sql.Identifier)
+        assert parts[1].strings == ("my_enum",)
+        assert parts[2] == psycopg2_sql.SQL(" AS ENUM (")
+        # The enum values are a Composed of Literal items joined by ", "
+        enum_literals = [p for p in parts[3].seq if isinstance(p, psycopg2_sql.Literal)]
+        assert [lit._wrapped for lit in enum_literals] == ["val_a", "val_b"]
+        assert parts[4] == psycopg2_sql.SQL(")")
+
+    def test_drop_enum_type_generates_correct_sql(self):
+        """drop_enum_type builds a proper DROP TYPE via psycopg2.sql."""
+        migration = PostgresEnumMigration("my_enum", ("val_a",))
+        schema_editor, mock_cursor = self._make_mock_schema_editor()
+
+        migration.drop_enum_type(apps=None, schema_editor=schema_editor)
+
+        mock_cursor.execute.assert_called_once()
+        query_arg = mock_cursor.execute.call_args[0][0]
+        assert isinstance(
+            query_arg, psycopg2_sql.Composable
+        ), "drop_enum_type must pass a psycopg2.sql.Composable, not a raw string."
+        # Verify the composed SQL structure: DROP TYPE <Identifier>
+        parts = query_arg.seq
+        assert parts[0] == psycopg2_sql.SQL("DROP TYPE ")
+        assert isinstance(parts[1], psycopg2_sql.Identifier)
+        assert parts[1].strings == ("my_enum",)

api/src/backend/api/tests/test_views.py

@@ -7858,8 +7858,12 @@ def test_create_relationship(
         assert response.status_code == status.HTTP_204_NO_CONTENT
         relationships = UserRoleRelationship.objects.filter(user=create_test_user.id)
         assert relationships.count() == 4
-        for relationship in relationships[2:]:  # Skip admin role
-            assert relationship.role.id in [r.id for r in roles_fixture[:2]]
+        # Use set membership instead of positional slicing — QuerySet ordering is
+        # non-deterministic without an explicit order_by, which makes slice-based
+        # checks intermittently fail.
+        added_role_ids = {r.id for r in roles_fixture[:2]}
+        relationship_role_ids = {rel.role.id for rel in relationships}
+        assert added_role_ids.issubset(relationship_role_ids)
 
     def test_create_relationship_already_exists(
         self, authenticated_client, roles_fixture, create_test_user