Potential fix for code scanning alert no. 27: Code injection

security code fix

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>
Signed-off-by: Kumar Vel <11884941+Kumarvels@users.noreply.github.com>

Author: Kumarvels

data_engineering/dataset_integration.py

@@ -15,6 +15,7 @@
 from datetime import datetime
 import hashlib
 
+import re
 # Optional imports for advanced features
 try:
     import plotly.express as px
@@ -298,6 +299,28 @@ def create_quality_filtered_dataset(self, dataset_id: str, min_trust_score: floa
             self.logger.error(f"Error creating quality-filtered dataset: {e}")
             raise
 
+    def is_safe_query(self, query_str: str, allowed_columns) -> bool:
+        """
+        Check if the query string is safe: only allowed column names, numbers, and safe operators.
+        """
+        # Only allow column names, numbers, whitespace, and safe operators
+        # Disallow parentheses, function calls, __import__, etc.
+        # Allowed operators: ==, !=, <, >, <=, >=, and, or, not
+        # Build regex for allowed columns
+        col_pattern = r'|'.join([re.escape(col) for col in allowed_columns])
+        # Full pattern: allowed columns, numbers, operators, whitespace
+        safe_pattern = rf'^([\s\d\.\'"]*({col_pattern})[\s\d\.\'"]*(==|!=|<=|>=|<|>|and|or|not|&|\||\s)*[\s\d\.\'"]*)+$'
+        # Disallow suspicious keywords
+        forbidden = ['__import__', 'os.', 'sys.', 'eval', 'exec', 'open(', '(', ')', '[', ']', '{', '}', ';']
+        lowered = query_str.lower()
+        for word in forbidden:
+            if word in lowered:
+                return False
+        # Check regex
+        if re.match(safe_pattern, query_str):
+            return True
+        return False
+
     def process_dataset(self, dataset_id: str, transformations: List[Dict]) -> str:
         """
         Apply transformations to a dataset
@@ -322,7 +345,10 @@ def process_dataset(self, dataset_id: str, transformations: List[Dict]) -> str:
             elif operation == 'rename_columns':
                 df = df.rename(columns=params['mapping'])
             elif operation == 'filter':
-                df = df.query(params['condition'])
+                condition = params['condition']
+                if not self.is_safe_query(condition, df.columns):
+                    raise ValueError("Unsafe filter condition detected. Only simple column comparisons are allowed.")
+                df = df.query(condition)
             elif operation == 'sort':
                 df = df.sort_values(by=params['columns'], ascending=params.get('ascending', True))
             elif operation == 'groupby':