0
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

[備忘録] Pythonで動的コード生成と実行を実装する

Posted at

はじめに

この記事は、Pythonで動的にコードを生成・実行する手法についての備忘録です。

image.png

目次

  1. Pythonで動的コードを実行する方法
  2. eval()exec()の基本
  3. compile()を使った高度な実行制御
  4. 動的コード生成のテクニック
  5. 実践的なユースケース
  6. セキュリティリスクと対策
  7. まとめと注意点

1. Pythonで動的コードを実行する方法

Pythonでは、主に以下の方法で動的にコードを実行できます:

  • eval(): 式を評価して結果を返す
  • exec(): 文(複数行のコードブロックも可)を実行する
  • compile(): コードをバイトコードにコンパイルする
  • 動的モジュールインポート: importlibを使用
  • AST(抽象構文木)による構文解析と実行

それぞれの特徴と用途を見ていきましょう。

2. eval()exec()の基本

eval()関数

eval()は**式(expression)**を評価し、その結果を返します。式とは値を生成するコードのことで、例えば 1 + 1"hello" + " world"len([1, 2, 3])などです。

# 基本的な使い方
result = eval('1 + 2 * 3')
print(result)  # 7

# 変数を使用
x = 10
result = eval('x * 2')
print(result)  # 20

# 関数呼び出し
result = eval('len("hello")')
print(result)  # 5

# リスト内包表記
result = eval('[i**2 for i in range(5)]')
print(result)  # [0, 1, 4, 9, 16]

image.png

exec()関数

exec()は**文(statement)**を実行します。文とは、代入、条件分岐、ループなど、値を返さないコードのことです。複数行のコードブロックも実行できます。

# 単純な文の実行
exec('x = 100')
print(x)  # 100

# 複数行のコードブロック
code = '''
def greet(name):
    return f"Hello, {name}!"

message = greet("Python")
'''
exec(code)
print(message)  # "Hello, Python!"

# 条件分岐やループの実行
exec('''
for i in range(3):
    print(f"Count: {i}")
''')
# Count: 0
# Count: 1
# Count: 2

image.png

名前空間の指定

eval()exec()は、オプションで名前空間(グローバル変数とローカル変数)を指定できます。これにより、実行環境を制御できます。

# グローバル名前空間を指定
global_ns = {'x': 10, 'y': 20}
result = eval('x + y', global_ns)
print(result)  # 30

# ローカル名前空間も指定
local_ns = {'y': 5}
result = eval('x + y', global_ns, local_ns)
print(result)  # 15(xはグローバルから、yはローカルから取得)

# exec()でも同様に指定可能
exec('z = x + y', global_ns, local_ns)
print(local_ns['z'])  # 15(zはローカル名前空間に追加される)

image.png

3. compile()を使った高度な実行制御

compile()関数を使うと、コードをコンパイルしてコードオブジェクトを作成できます。このオブジェクトはeval()exec()で実行できます。

# eval用にコンパイル(式モード)
expr_code = compile('x + y', '<string>', 'eval')
result = eval(expr_code, {'x': 10, 'y': 20})
print(result)  # 30

# exec用にコンパイル(文モード)
stmt_code = compile('for i in range(3): print(i)', '<string>', 'exec')
exec(stmt_code)
# 0
# 1
# 2

# 対話モード(複数の文と式を含むコードをコンパイル)
# 注意: 'single'モードは対話式インタプリタ用で、各文を別々にコンパイルする必要があります
exec_code = compile('x = 10\ny = 20', '<string>', 'exec')
exec(exec_code)
eval_code = compile('x + y', '<string>', 'eval')
result = eval(eval_code)
print(result)  # 30

image.png

compile()のメリットは、一度コンパイルしたコードを何度も実行できること、またコンパイル時にエラーチェックができることです。

# コードを一度コンパイルして複数回実行
calc_code = compile('a * b', '<string>', 'eval')

for a, b in [(2, 3), (5, 7), (10, 20)]:
    result = eval(calc_code, {'a': a, 'b': b})
    print(f"{a} * {b} = {result}")
# 2 * 3 = 6
# 5 * 7 = 35
# 10 * 20 = 200

4. 動的コード生成のテクニック

実際にコードを動的に生成するテクニックを見ていきましょう。

テンプレートとしてのコード生成

def generate_function(name, params):
    param_list = ', '.join(params)
    body = ', '.join([f'{p}={{{p}}}' for p in params])
    
    code = f'''
def {name}({param_list}):
    print(f"Function {name} called with {body}")
    return {' + '.join(params) if params else 0}
'''
    return code

# 関数を動的に生成
func_code = generate_function('add_numbers', ['a', 'b', 'c'])
print("生成されたコード:")
print(func_code)

# 生成したコードを実行
exec(func_code)

# 生成した関数を呼び出す
result = add_numbers(10, 20, 30)
print(f"結果: {result}")

出力:
image.png

動的クラス生成

def generate_class(class_name, fields):
    field_defs = '\n    '.join([f'{field} = None' for field in fields])
    init_params = ', '.join(['self'] + fields)
    init_body = '\n        '.join([f'self.{field} = {field}' for field in fields])
    
    # __repr__メソッドを動的に生成
    repr_fields = []
    for field in fields:
        repr_fields.append(f"{field}={{self.{field}}}")
    
    repr_body = ', '.join(repr_fields)
    
    code = f'''
class {class_name}:
    {field_defs}
    
    def __init__({init_params}):
        {init_body}
    
    def __repr__(self):
        return f"{class_name}({repr_body})"
'''
    return code

# クラスを動的に生成
person_class_code = generate_class('Person', ['name', 'age', 'email'])
print("生成されたクラスコード:")
print(person_class_code)

# 生成したコードを実行
exec(person_class_code)

# 生成したクラスを使用
person = Person("山田太郎", 30, "yamada@example.com")
print(person)  # Person(name=山田太郎, age=30, email=yamada@example.com)

image.png

プログラム全体の動的生成と実行

複雑なプログラムを動的に生成し、実行する例です。

def generate_data_processor(data_type, operations):
    """データ処理プログラムを動的に生成する"""
    
    imports = []
    functions = []
    main_code = []
    
    # 必要なライブラリをインポート
    if data_type == 'csv':
        imports.append('import csv')
    elif data_type == 'json':
        imports.append('import json')
    
    if 'plot' in operations:
        imports.append('import matplotlib.pyplot as plt')
    
    # 関数を生成
    if 'read' in operations:
        if data_type == 'csv':
            functions.append('''
def read_data(file_path):
    data = []
    with open(file_path, 'r', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        for row in reader:
            data.append(row)
    return data
''')
        elif data_type == 'json':
            functions.append('''
def read_data(file_path):
    with open(file_path, 'r', encoding='utf-8') as f:
        return json.load(f)
''')
    
    if 'process' in operations:
        functions.append('''
def process_data(data):
    # データを加工する処理
    results = []
    for item in data:
        # 仮の処理:数値フィールドを抽出して合計
        num_values = [float(val) for val in item.values() if str(val).replace('.', '', 1).isdigit()]
        if num_values:
            results.append(sum(num_values))
    return results
''')
    
    if 'plot' in operations:
        functions.append('''
def plot_data(data):
    plt.figure(figsize=(10, 6))
    plt.bar(range(len(data)), data)
    plt.title('Data Visualization')
    plt.xlabel('Index')
    plt.ylabel('Value')
    plt.savefig('output_plot.png')
    plt.close()
''')
    
    # メインコード
    main_code.append('''
def main():
    # ファイルパスは実際の環境に合わせて変更してください
    file_path = 'sample_data.csv' if data_type == 'csv' else 'sample_data.json'
    
    print(f"Processing {data_type} data...")
''')
    
    if 'read' in operations:
        main_code.append('    data = read_data(file_path)')
    
    if 'process' in operations:
        main_code.append('    results = process_data(data)')
        main_code.append('    print(f"Processed {len(results)} data points")')
    
    if 'plot' in operations:
        main_code.append('    plot_data(results)')
        main_code.append('    print("Plot saved as output_plot.png")')
    
    main_code.append('''
    print("Done!")

if __name__ == "__main__":
    main()
''')
    
    # 全てのコードを結合
    full_code = '\n'.join(imports) + '\n\n' + '\n'.join(functions) + '\n'.join(main_code)
    full_code = full_code.replace('data_type', f"'{data_type}'")
    
    return full_code

# CSVデータを読み込み、処理し、プロットするプログラムを生成
program = generate_data_processor('csv', ['read', 'process', 'plot'])
print("生成されたプログラム:")
print(program)

# 実際の環境で実行する場合は以下のようにする
# with open('generated_program.py', 'w') as f:
#     f.write(program)
# import subprocess
# subprocess.run(['python', 'generated_program.py'])

image.png

5. 実践的なユースケース

ユースケース1: 数式の動的評価

ユーザーが入力した数式を安全に評価する例です。

def safe_eval_expression(expr, variables=None):
    """安全に数式を評価する関数"""
    if variables is None:
        variables = {}
    
    # 許可する演算子と関数のリスト
    allowed_names = {
        'abs': abs, 'round': round, 'min': min, 'max': max,
        'sum': sum, 'len': len, 'pow': pow, 'int': int, 'float': float
    }
    
    # 変数をallowed_namesに追加
    for name, value in variables.items():
        if isinstance(value, (int, float, list, tuple)):
            allowed_names[name] = value
    
    # evalを制限された名前空間で実行
    try:
        return eval(expr, {"__builtins__": {}}, allowed_names)
    except Exception as e:
        return f"エラー: {str(e)}"

# 使用例
expressions = [
    "2 + 2 * 3",
    "abs(-15) + max(4, 7, 2)",
    "pow(2, 8)",
    "round(3.14159, 2)"
]

for expr in expressions:
    result = safe_eval_expression(expr)
    print(f"{expr} = {result}")

# 変数を使った例
variables = {'x': 10, 'y': 5, 'values': [1, 2, 3, 4, 5]}
expr = "x * y + sum(values)"
result = safe_eval_expression(expr, variables)
print(f"{expr} = {result}")  # x * y + sum(values) = 65

image.png

ユースケース2: 動的なデータ検証ルール

データ検証ルールを動的に構築して実行する例です。

def create_validator(rules):
    """検証ルールから検証関数を動的に生成"""
    function_code = '''
def validate_data(data):
    errors = []
'''
    
    for field, rule in rules.items():
        condition = rule['condition']
        error_msg = rule['error']
        
        check_code = f'''
    # {field}のチェック
    if "{field}" in data:
        if not ({condition}):
            errors.append(f"{error_msg}")
    else:
        errors.append(f"{field}が存在しません")
'''
        function_code += check_code
    
    function_code += '''
    return errors
'''
    
    # 関数を生成して返す
    local_vars = {}
    exec(function_code, globals(), local_vars)
    return local_vars['validate_data']

# 検証ルールの定義
validation_rules = {
    'age': {
        'condition': 'isinstance(data["age"], int) and 0 <= data["age"] <= 120',
        'error': '年齢は0〜120の整数である必要があります'
    },
    'email': {
        'condition': 'isinstance(data["email"], str) and "@" in data["email"]',
        'error': 'メールアドレスの形式が不正です'
    },
    'username': {
        'condition': 'isinstance(data["username"], str) and 3 <= len(data["username"]) <= 20',
        'error': 'ユーザー名は3〜20文字である必要があります'
    }
}

# 検証関数を動的に生成
validator = create_validator(validation_rules)

# テストデータ
test_data = [
    {'username': 'john', 'age': 25, 'email': 'john@example.com'},  # 有効
    {'username': 'a', 'age': 150, 'email': 'invalid-email'},  # 無効
    {'username': 'alice', 'age': '30', 'email': 'alice@example.com'}  # 無効(ageが文字列)
]

# 検証実行
for i, data in enumerate(test_data):
    errors = validator(data)
    if errors:
        print(f"データ {i+1} は無効: {', '.join(errors)}")
    else:
        print(f"データ {i+1} は有効です")

image.png

ユースケース3: 単純なDSL(ドメイン特化言語)の実装

簡単なDSLを実装する例です。

class SimpleDSL:
    """シンプルなドメイン特化言語インタプリタ"""
    
    def __init__(self):
        self.variables = {}
    
    def execute(self, script):
        """DSLスクリプトを実行"""
        lines = script.strip().split('\n')
        result = None
        
        for i, line in enumerate(lines):
            line = line.strip()
            if not line or line.startswith('#'):
                continue
            
            try:
                if line.startswith('SET '):
                    # 変数設定: SET x = 10
                    parts = line[4:].split('=', 1)
                    var_name = parts[0].strip()
                    expr = parts[1].strip()
                    self.variables[var_name] = self._eval_expr(expr)
                
                elif line.startswith('PRINT '):
                    # 出力: PRINT x + 5
                    expr = line[6:].strip()
                    value = self._eval_expr(expr)
                    print(value)
                    result = value
                
                elif line.startswith('IF '):
                    # 条件: IF x > 10 THEN PRINT "Large"
                    parts = line[3:].split('THEN', 1)
                    condition = parts[0].strip()
                    action = parts[1].strip() if len(parts) > 1 else ""
                    
                    if self._eval_expr(condition):
                        result = self.execute(action)
                
                elif line.startswith('REPEAT '):
                    # ループ: REPEAT 5 TIMES PRINT i
                    parts = line.split('TIMES', 1)
                    count_expr = parts[0][7:].strip()
                    count = int(self._eval_expr(count_expr))
                    action = parts[1].strip()
                    
                    for i in range(count):
                        self.variables['i'] = i
                        result = self.execute(action)
            
            except Exception as e:
                print(f"エラー(行 {i+1}): {str(e)}")
        
        return result
    
    def _eval_expr(self, expr):
        """式を評価"""
        # DSL内の特殊構文を処理
        expr = expr.replace('"', "'")  # ダブルクォートをシングルクォートに変換
        
        # 安全な評価のために制限された名前空間を使用
        safe_globals = {"__builtins__": {}}
        safe_functions = {
            'len': len, 'str': str, 'int': int, 'float': float,
            'min': min, 'max': max, 'abs': abs, 'sum': sum
        }
        
        return eval(expr, safe_globals, {**safe_functions, **self.variables})

# DSLの使用例
dsl_script = '''
# 変数の設定
SET x = 10
SET y = 20
SET name = "Python"

# 計算と出力
PRINT x + y
PRINT "Hello, " + name

# 条件分岐
IF x > 5 THEN PRINT "x is greater than 5"

# ループ
REPEAT 3 TIMES PRINT "Count: " + str(i)
'''

dsl = SimpleDSL()
dsl.execute(dsl_script)

6. セキュリティリスクと対策

動的コード実行には重大なセキュリティリスクがあります。特に、ユーザー入力を直接eval()exec()に渡すことは、リモートコード実行攻撃の危険があります。

主なリスク

  1. 任意のコード実行: 悪意のあるユーザーがシステムコマンドを実行する可能性
  2. データアクセス: 機密情報への不正アクセス
  3. リソース消費: 無限ループなどによるリソース枯渇攻撃

安全対策

def very_safe_eval(expr, allowed_names=None):
    """
    非常に制限された環境で式を評価する
    allowed_names: 許可する変数や関数の辞書
    """
    if allowed_names is None:
        allowed_names = {}
    
    # 組み込み関数へのアクセスを完全に遮断
    forbidden = {'__import__', 'eval', 'exec', 'compile', 'globals', 'locals', 'open', 
                'getattr', 'setattr', 'delattr', '__class__', '__base__', '__subclasses__'}
    
    # 禁止された単語がないか確認
    for word in forbidden:
        if word in expr:
            raise ValueError(f"式に禁止された単語が含まれています: {word}")
    
    # AST(抽象構文木)を使って式を解析し、安全かチェック
    import ast
    
    try:
        # 式を解析
        parsed = ast.parse(expr, mode='eval')
        
        # 式の内容を検証する関数
        class SafetyChecker(ast.NodeVisitor):
            def visit_Name(self, node):
                if node.id not in allowed_names:
                    raise NameError(f"変数 '{node.id}' は許可されていません")
                return self.generic_visit(node)
            
            def visit_Attribute(self, node):
                raise AttributeError("属性アクセスは許可されていません")
            
            def visit_Call(self, node):
                if isinstance(node.func, ast.Name):
                    if node.func.id not in allowed_names:
                        raise NameError(f"関数 '{node.func.id}' は許可されていません")
                else:
                    raise TypeError("関数呼び出しの形式が不正です")
                return self.generic_visit(node)
        
        # 安全性チェック実行
        checker = SafetyChecker()
        checker.visit(parsed)
        
        # 式をコンパイルして実行
        compiled_expr = compile(parsed, '<string>', 'eval')
        return eval(compiled_expr, {"__builtins__": {}}, allowed_names)
    
    except Exception as e:
        return f"エラー: {str(e)}"

# 使用例
allowed_vars = {
    'x': 10,
    'y': 20,
    'add': lambda a, b: a + b,
    'multiply': lambda a, b: a * b
}

expressions = [
    "x + y",                          # 安全: 許可された変数を使用
    "add(x, y)",                      # 安全: 許可された関数を使用
    "multiply(x, 5)",                 # 安全: 許可された関数と定数を使用
    "__import__('os').system('ls')",  # 危険: 禁止された関数
    "open('/etc/passwd').read()",     # 危険: ファイルアクセス
    "globals()",                      # 危険: グローバル変数へのアクセス
    "x.__class__",                    # 危険: 属性アクセス
]

print("安全な式の評価:")
for expr in expressions[:3]:  # 安全な式だけを評価
    result = very_safe_eval(expr, allowed_vars)
    print(f"{expr} => {result}")

print("\n危険な式のチェック:")
for expr in expressions[3:]:  # 危険な式
    try:
        result = very_safe_eval(expr, allowed_vars)
        print(f"{expr} => {result}")
    except ValueError as e:
        print(f"{expr} => エラー: {e}")

image.png

追加のセキュリティ対策

  1. タイムアウト設定: 実行時間に制限を設ける

    import signal
    
    def timeout_handler(signum, frame):
        raise TimeoutError("コードの実行がタイムアウトしました")
    
    def run_with_timeout(code, timeout=1):
        """タイムアウト付きでコードを実行"""
        signal.signal(signal.SIGALRM, timeout_handler)
        signal.alarm(timeout)
        
        try:
            result = eval(code)
            signal.alarm(0)  # タイマーをリセット
            return result
        except TimeoutError as e:
            return str(e)
    
  2. サンドボックス環境の使用: 別プロセスやコンテナでコードを実行

  3. ホワイトリスト方式: 許可する操作だけを明示的に定義

⚠️ 注意事項: この記事に含まれるコードのうち、signal.alarm()によるタイムアウト処理はUnix/Linux環境専用です。Windows環境では動作しないため、別のタイムアウト方法(例:threading.Timer)を検討してください。

7. まとめと注意点

動的コードの生成・実行は非常に柔軟で強力な手法ですが、慎重な取り扱いが求められます。

メリット:動作の変更やコード生成が実行時に可能で、DSLやプラグイン構築にも応用できます。

注意点:セキュリティリスクやデバッグの難しさ、可読性の低下、パフォーマンスへの影響が懸念されます。

推奨対策:必要最小限の範囲で使用し、入力の検証やサンドボックス化、タイムアウト・例外処理を必ず導入することが重要です。

おわりに

Pythonの動的コード生成・実行機能は諸刃の剣です。適切な場面で慎重に使えば強力なツールになりますが、安全性を最優先に考える必要があります。

参考リンク

0
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
0
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?