diff --git a/api/CHANGELOG.md b/api/CHANGELOG.md index 023abaeb40..8c5a47ad6b 100644 --- a/api/CHANGELOG.md +++ b/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) diff --git a/api/src/backend/api/db_utils.py b/api/src/backend/api/db_utils.py index 7a71084ccd..e3b11d7084 100644 --- a/api/src/backend/api/db_utils.py +++ b/api/src/backend/api/db_utils.py @@ -18,6 +18,7 @@ from django.db import ( ) 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 @@ class PostgresEnumMigration: 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): diff --git a/api/src/backend/api/tests/test_db_utils.py b/api/src/backend/api/tests/test_db_utils.py index f52bb349aa..18935b9a3e 100644 --- a/api/src/backend/api/tests/test_db_utils.py +++ b/api/src/backend/api/tests/test_db_utils.py @@ -6,10 +6,12 @@ import pytest 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 @@ class TestRlsTransaction: 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 AS ENUM () + 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 + 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",) diff --git a/api/src/backend/api/tests/test_views.py b/api/src/backend/api/tests/test_views.py index 5aa7ae143d..d857b2bc07 100644 --- a/api/src/backend/api/tests/test_views.py +++ b/api/src/backend/api/tests/test_views.py @@ -7858,8 +7858,12 @@ class TestUserRoleRelationshipViewSet: 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