# Code Injection via eval() on Untrusted User Input Language: Python Severity: Critical CWE: CWE-95 ## Source 9 ## Flow 9-11 ## Sink 11 ## Vulnerable Code ```python def apply_iot_device_filter(sensor_data, filter_expression): filtered_results = [] for device in sensor_data: try: if eval(filter_expression, {'device': device, '__builtins__': {}}): filtered_results.append(device) except: pass return filtered_results ``` ## Explanation The function accepts untrusted user input via the filter_expression parameter and directly passes it to eval() on line 5. Despite attempting to restrict __builtins__, attackers can bypass this sandboxing through Python introspection and attribute access (e.g., accessing __class__.__bases__ chains) to achieve arbitrary code execution. ## Remediation The fix replaces the dangerous eval() call with a custom AST-based expression evaluator that parses the filter expression into an abstract syntax tree and only allows safe operations (comparisons, boolean logic, subscript access on dictionaries, and literal values). Function calls, attribute access, lambda expressions, comprehensions, and all other potentially dangerous constructs are explicitly rejected during AST validation before evaluation begins. ## Secure Code ```python import ast import operator SAFE_OPERATORS = { ast.Eq: operator.eq, ast.NotEq: operator.ne, ast.Lt: operator.lt, ast.LtE: operator.le, ast.Gt: operator.gt, ast.GtE: operator.ge, ast.And: lambda *args: all(args), ast.Or: lambda *args: any(args), ast.Not: operator.not_, ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, ast.In: lambda a, b: a in b, ast.NotIn: lambda a, b: a not in b, } def _safe_eval_node(node, device): """Recursively evaluate an AST node with only safe operations allowed.""" if isinstance(node, ast.Expression): return _safe_eval_node(node.body, device) elif isinstance(node, ast.BoolOp): op = SAFE_OPERATORS.get(type(node.op)) if op is None: raise ValueError(f"Unsupported boolean operator: {type(node.op).__name__}") values = [_safe_eval_node(v, device) for v in node.values] return op(*values) elif isinstance(node, ast.UnaryOp): if isinstance(node.op, ast.Not): return not _safe_eval_node(node.operand, device) raise ValueError(f"Unsupported unary operator: {type(node.op).__name__}") elif isinstance(node, ast.BinOp): op = SAFE_OPERATORS.get(type(node.op)) if op is None: raise ValueError(f"Unsupported binary operator: {type(node.op).__name__}") left = _safe_eval_node(node.left, device) right = _safe_eval_node(node.right, device) return op(left, right) elif isinstance(node, ast.Compare): left = _safe_eval_node(node.left, device) for op_node, comparator in zip(node.ops, node.comparators): op = SAFE_OPERATORS.get(type(op_node)) if op is None: raise ValueError(f"Unsupported comparison operator: {type(op_node).__name__}") right = _safe_eval_node(comparator, device) if not op(left, right): return False left = right return True elif isinstance(node, ast.Subscript): value = _safe_eval_node(node.value, device) if not isinstance(value, dict): raise ValueError("Subscript access is only allowed on device dictionaries") if isinstance(node.slice, ast.Constant): key = node.slice.value elif isinstance(node.slice, ast.Index): key = _safe_eval_node(node.slice.value, device) else: raise ValueError("Only constant subscript keys are allowed") if not isinstance(key, (str, int)): raise ValueError("Subscript key must be a string or integer") return value[key] elif isinstance(node, ast.Name): if node.id == 'device': return device elif node.id in ('True', 'true'): return True elif node.id in ('False', 'false'): return False elif node.id == 'None': return None raise ValueError(f"Access to variable '{node.id}' is not allowed") elif isinstance(node, ast.Constant): if isinstance(node.value, (int, float, str, bool, type(None))): return node.value raise ValueError(f"Unsupported constant type: {type(node.value).__name__}") elif isinstance(node, ast.Num): return node.n elif isinstance(node, ast.Str): return node.s elif isinstance(node, ast.NameConstant): return node.value elif isinstance(node, ast.List): return [_safe_eval_node(elt, device) for elt in node.elts] elif isinstance(node, ast.Tuple): return tuple(_safe_eval_node(elt, device) for elt in node.elts) else: raise ValueError(f"Unsupported expression type: {type(node).__name__}") def safe_evaluate_filter(filter_expression, device): """Safely evaluate a filter expression against a device dictionary.""" try: tree = ast.parse(filter_expression, mode='eval') except SyntaxError as e: raise ValueError(f"Invalid filter expression syntax: {e}") for node in ast.walk(tree): if isinstance(node, (ast.Call, ast.Attribute, ast.Import, ast.ImportFrom, ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Lambda, ast.GeneratorExp, ast.ListComp, ast.SetComp, ast.DictComp, ast.Await, ast.Yield, ast.YieldFrom, ast.Global, ast.Nonlocal, ast.Delete, ast.Assign, ast.AugAssign, ast.FormattedValue, ast.JoinedStr)): raise ValueError(f"Disallowed expression type: {type(node).__name__}. " "Only comparisons, boolean logic, and dictionary access are allowed.") return _safe_eval_node(tree, device) def apply_iot_device_filter(sensor_data, filter_expression): """Filter IoT sensor data based on a safe filter expression. Supported expressions: - device["temperature"] > 25 - device["status"] == "active" - device["temperature"] > 20 and device["humidity"] < 80 - device["type"] in ["sensor", "actuator"] """ if not isinstance(filter_expression, str): raise ValueError("Filter expression must be a string") if len(filter_expression) > 500: raise ValueError("Filter expression is too long (max 500 characters)") filtered_results = [] for device in sensor_data: try: if safe_evaluate_filter(filter_expression, device): filtered_results.append(device) except (ValueError, KeyError, TypeError, IndexError): pass return filtered_results ```