Patched code sample

import pandas as pd
import ast
import subprocess

def safe_query(df, query_string):
    """
    Safely queries a Pandas DataFrame, mitigating command injection vulnerabilities.

    Args:
        df: The Pandas DataFrame to query.
        query_string: The query string.

    Returns:
        A new Pandas DataFrame representing the result of the query, or None if the query is unsafe.
    """

    # 1. Whitelist Allowed Functions and Attributes (AST-based)
    allowed_functions = {'abs', 'min', 'max', 'sum', 'len', 'mean'}  #  Example, extend as needed
    allowed_attributes = set(df.columns)  # Only column names allowed

    try:
        # Parse the query string into an Abstract Syntax Tree (AST)
        tree = ast.parse(query_string)

        # Walk through the AST to check for disallowed function calls or attributes
        for node in ast.walk(tree):
            if isinstance(node, ast.Call):
                if isinstance(node.func, ast.Name):
                    function_name = node.func.id
                    if function_name not in allowed_functions:
                        print(f"Error: Disallowed function '{function_name}' in query.")
                        return None  # Reject the query
            elif isinstance(node, ast.Attribute):
                if isinstance(node.value, ast.Name): # Checking if attribute belongs to df object
                    attribute_name = node.attr
                    if node.value.id != 'df' or attribute_name not in allowed_attributes:
                        print(f"Error: Disallowed attribute '{attribute_name}' in query.")
                        return None  # Reject the query
            elif isinstance(node, ast.Subscript):
                if isinstance(node.value, ast.Name) and node.value.id == 'df':  #Prevent df['os.system("evil")']
                    if isinstance(node.slice, ast.Constant):
                        if not isinstance(node.slice.value, str):
                            print(f"Error: Disallowed index type in query")
                            return None
                    elif isinstance(node.slice, ast.Slice) and all(isinstance(x, ast.Constant) or x is None for x in [node.slice.start, node.slice.stop, node.slice.step]):
                        pass
                    else:
                        print(f"Error: Complex slicing not allowed in safe query.")
                        return None
            elif isinstance(node, ast.Import) or isinstance(node, ast.ImportFrom):
                print("Error: Imports are not allowed in safe queries.")
                return None

            # Add more checks here to cover other potentially dangerous AST node types.

        # 2. Use a Safe Querying Method (e.g., numexpr or a custom evaluation)
        try:
            result = df.query(query_string, engine='numexpr') # Force the usage of numexpr if available
            return result
        except Exception as e:
            print(f"Query execution error: {e}")
            return None

    except SyntaxError as e:
        print(f"Syntax error in query: {e}")
        return None
    except Exception as e:
        print(f"Error during query validation: {e}")
        return None


if __name__ == '__main__':
    # Example usage
    data = {'col1': [1, 2, 3, 4, 5],
            'col2': [6, 7, 8, 9, 10]}
    df = pd.DataFrame(data)

    # Safe query
    safe_result = safe_query(df, "col1 > 2 and col2 < 9")
    if safe_result is not None:
        print("Safe Query Result:\n", safe_result)

    # Attempt to inject code (this will be blocked)
    unsafe_query = "col1 > 2 and df['col1'].abs().sum() < 15"
    unsafe_result = safe_query(df, unsafe_query)  # Example of using an allowed function
    if unsafe_result is not None:
        print("Unsafe Query Result:\n", unsafe_result)
    else:
        print("Unsafe query was blocked.")
    
    # Attempt to use an attribute that is not allowed
    unsafe_query2 = "col1 > 2 and df.index.name == 'evil'"
    unsafe_result2 = safe_query(df, unsafe_query2)  # Example of using an allowed function
    if unsafe_result2 is not None:
        print("Unsafe Query Result:\n", unsafe_result2)
    else:
        print("Unsafe query 2 was blocked.")

    #Example of blocked import.
    unsafe_query3 = "import os; col1 > 2"
    unsafe_result3 = safe_query(df, unsafe_query3)
    if unsafe_result3 is not None:
        print("Unsafe Query Result:\n", unsafe_result3)
    else:
        print("Unsafe query 3 was blocked.")