[DOP-31721] Validate SQL transformation FROM clause

Author: dolfinus

syncmaster/schemas/v1/transfers/transformations/sql.py

@@ -5,7 +5,8 @@
 
 from pydantic import BaseModel, Field, field_validator
 
-ALLOWED_SELECT_QUERY = re.compile(r"^\s*(SELECT|WITH)\b", re.IGNORECASE)
+ALLOWED_QUERY_TYPE = re.compile(r"^\s*(SELECT|WITH)\b", re.IGNORECASE)
+QUERY_CONTAINS_FROM_SOURCE = re.compile(r"\bFROM\s+source\b", re.IGNORECASE)
 
 
 class Sql(BaseModel):
@@ -16,8 +17,12 @@ class Sql(BaseModel):
     @field_validator("query", mode="after")
     @classmethod
     def _validate_query(cls, value: str) -> str:
-        if not ALLOWED_SELECT_QUERY.match(value):
-            msg = f"Query must be a SELECT statement, got '{value}'"
+        if not ALLOWED_QUERY_TYPE.match(value):
+            msg = "Query must be a SELECT statement"
+            raise ValueError(msg)
+
+        if not QUERY_CONTAINS_FROM_SOURCE.match(value):
+            msg = "Query must contain `FROM source` clause"
             raise ValueError(msg)
 
         return value

tests/test_unit/test_transfers/test_create_transfer.py

@@ -832,7 +832,7 @@ async def test_superuser_can_create_transfer(
                                 "sql",
                                 "query",
                             ],
-                            "message": "Value error, Query must be a SELECT statement, got 'INSERT INTO table1 VALUES (1, 2)'",
+                            "message": "Value error, Query must be a SELECT statement",
                             "code": "value_error",
                             "context": {},
                             "input": "INSERT INTO table1 VALUES (1, 2)",
@@ -842,6 +842,39 @@ async def test_superuser_can_create_transfer(
             },
             id="sql_non_select_query",
         ),
+        pytest.param(
+            {
+                "transformations": [
+                    {
+                        "type": "sql",
+                        "query": "SELECT * FROM table1",
+                        "dialect": "spark",
+                    },
+                ],
+            },
+            {
+                "error": {
+                    "code": "invalid_request",
+                    "message": "Invalid request",
+                    "details": [
+                        {
+                            "location": [
+                                "body",
+                                "transformations",
+                                0,
+                                "sql",
+                                "query",
+                            ],
+                            "message": "Value error, Query must contain `FROM source` clause",
+                            "code": "value_error",
+                            "context": {},
+                            "input": "SELECT * FROM table1",
+                        },
+                    ],
+                },
+            },
+            id="sql_invalid_select_query",
+        ),
         pytest.param(
             {
                 "transformations": [