Prevent write operations in the SQL query

flux_sdk/etl/data_models/query.py

@@ -34,7 +34,7 @@ class SQLQuery:
 
     def __post_init__(self):
         """Perform validation."""
-        check_field(self, "text", str, required=True)
+        check_field(self, "text", str, required=True, block_list_regex=r'\b(?:INSERT\s+INTO|UPDATE|DELETE\s+FROM|DROP|ALTER|CREATE)\b')
         check_field(self, "args", dict[str, SQLQueryArg])
 
 

flux_sdk/etl/data_models/tests/test_query.py

@@ -24,6 +24,11 @@ def test_validate_args_wrong_type(self):
             with self.assertRaises(TypeError):
                 SQLQuery(text="select column from table", args=value)
 
+    def test_validate_text_contains_write_operation(self):
+        for value in ["insert into table", "update table", "delete from table", "drop table", "alter table", "create table"]:
+            with self.assertRaises(ValueError):
+                SQLQuery(text=value)
+
     def test_validate_success_minimal(self):
         SQLQuery(text="select column from table")
 

flux_sdk/flux_core/validation.py

@@ -1,7 +1,8 @@
+import re
 from typing import Any, Type, Union, get_args, get_origin
 
 
-def check_field(obj: Any, attr: str, desired_type: Type, required: bool = False):
+def check_field(obj: Any, attr: str, desired_type: Type, required: bool = False, block_list_regex: str = None):
     value = getattr(obj, attr)
 
     if required and not value:
@@ -10,6 +11,9 @@ def check_field(obj: Any, attr: str, desired_type: Type, required: bool = False)
     if value:
         _check_type(value, attr, desired_type)
 
+    if block_list_regex and value and re.search(block_list_regex, value, re.IGNORECASE):
+        raise ValueError(f"{attr} is not allowed")
+
 
 def _check_type(value: Any, attr: str, desired_type: Type):
     origin = get_origin(desired_type)