# コード例1: SQLインジェクションの攻撃パターンと防御
 
import sqlite3
 
# === 脆弱なコード ===
def login_vulnerable(username, password):
    """文字列連結によるSQL構築 -> SQLインジェクション脆弱"""
    conn = sqlite3.connect("app.db")
    query = f"SELECT * FROM users WHERE username='{username}' AND password='{password}'"
    # 攻撃例: username = "admin' --"
    # 生成SQL: SELECT * FROM users WHERE username='admin' --' AND password=''
    # -- 以降はコメント -> パスワード検証がスキップされる
    result = conn.execute(query).fetchone()
    return result is not None
 
# === 安全なコード: パラメータ化クエリ ===
def login_safe(username, password):
    """パラメータ化クエリで安全にSQLを実行"""
    conn = sqlite3.connect("app.db")
    query = "SELECT * FROM users WHERE username=? AND password=?"
    # ? はプレースホルダ -> 入力は常にデータとして扱われる
    result = conn.execute(query, (username, password)).fetchone()
    return result is not None
 
# === さらに安全: ORM使用 ===
from sqlalchemy.orm import Session
from sqlalchemy import select
 
def login_orm(session: Session, username: str, password_hash: str):
    """ORMを使用した安全なクエリ"""
    stmt = select(User).where(
        User.username == username,
        User.password_hash == password_hash,
    )
    return session.execute(stmt).scalar_one_or_none()

# コード例2: Blind SQLインジェクションの仕組みと対策
import time
 
# Blind (Boolean-based) の攻撃例
# 攻撃者のスクリプト（説明目的のみ）
def demonstrate_blind_sqli_concept():
    """
    Blind SQLiの原理を示す概念コード。
    実際のペネトレーションテストでは sqlmap 等のツールを使用する。
 
    脆弱なエンドポイント: /user?id=1
    正常: /user?id=1 → 200 OK (ユーザー情報表示)
    攻撃: /user?id=1 AND 1=1 → 200 OK (真)
    攻撃: /user?id=1 AND 1=2 → 404 Not Found (偽)
    → レスポンスの差異から情報を1ビットずつ抽出
 
    例: データベース名の1文字目を特定
    /user?id=1 AND SUBSTRING(database(),1,1)='a' → 404 (偽)
    /user?id=1 AND SUBSTRING(database(),1,1)='b' → 404 (偽)
    ...
    /user?id=1 AND SUBSTRING(database(),1,1)='m' → 200 (真!)
    → データベース名の1文字目は 'm'
    """
    pass
 
# Time-based Blind SQLiの原理
def demonstrate_time_based_concept():
    """
    レスポンスの有無ではなく、応答時間で真偽を判定する。
 
    /user?id=1; IF(SUBSTRING(database(),1,1)='m',
                    SLEEP(5), 0)
    → 5秒の遅延 = 真 (1文字目は 'm')
    → 即応答 = 偽
 
    対策: パラメータ化クエリを使用すればこれらの攻撃は全て防げる。
    """
    pass

# コード例3: Second-order SQLインジェクションの例と対策
 
# Second-order: 入力時ではなく、保存したデータの使用時に発動
 
# 脆弱なコード
def register_user(username, password):
    """ユーザー登録（パラメータ化されているので安全に見える）"""
    db.execute(
        "INSERT INTO users (username, password) VALUES (?, ?)",
        (username, password)  # ここは安全
    )
 
def change_password(username, new_password):
    """パスワード変更（ここが脆弱!）"""
    # usernameをDBから取得してSQLに埋め込む
    user = db.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone()
    # user["username"] = "admin'--" (登録時に仕込まれた値)
    db.execute(
        f"UPDATE users SET password='{new_password}' WHERE username='{user['username']}'"
    )
    # 結果: UPDATE users SET password='...' WHERE username='admin'--'
    # admin のパスワードが変更される!
 
# 安全なコード: すべてのSQL文でパラメータ化を徹底
def change_password_safe(username, new_password):
    db.execute(
        "UPDATE users SET password=? WHERE username=?",
        (new_password, username)  # 常にパラメータ化
    )

# コード例4: 各データベース/言語でのパラメータ化クエリ
 
# --- Python ---
 
# SQLite3
import sqlite3
conn = sqlite3.connect("app.db")
conn.execute("SELECT * FROM users WHERE id=?", (user_id,))
 
# MySQL (mysql-connector-python)
import mysql.connector
conn = mysql.connector.connect(host="localhost", database="myapp")
cursor = conn.cursor(prepared=True)
cursor.execute("SELECT * FROM users WHERE id=%s", (user_id,))
 
# PostgreSQL (psycopg2)
import psycopg2
conn = psycopg2.connect("dbname=myapp")
cursor = conn.cursor()
cursor.execute("SELECT * FROM users WHERE id=%s", (user_id,))
 
# SQLAlchemy ORM
from sqlalchemy import select, text
# ORM クエリ（自動的にパラメータ化）
stmt = select(User).where(User.id == user_id)
# text() を使う場合（バインドパラメータ指定）
stmt = text("SELECT * FROM users WHERE id = :id").bindparams(id=user_id)
 
# --- Java ---
# PreparedStatement ps = conn.prepareStatement(
#     "SELECT * FROM users WHERE id = ?");
# ps.setInt(1, userId);
# ResultSet rs = ps.executeQuery();
 
# --- Node.js (pg) ---
# const result = await pool.query(
#     'SELECT * FROM users WHERE id = $1',
#     [userId]
# );
 
# --- Go ---
# rows, err := db.Query(
#     "SELECT * FROM users WHERE id = $1",
#     userId,
# )

# コード例5: NoSQLインジェクション（MongoDB）の攻撃と対策
from pymongo import MongoClient
 
client = MongoClient("mongodb://localhost:27017")
db = client["myapp"]
 
# === 脆弱なコード ===
def find_user_vulnerable(request_data):
    """JSONデータをそのままクエリに使用 -> NoSQL injection"""
    username = request_data["username"]
    password = request_data["password"]
    # 攻撃: {"username": "admin", "password": {"$ne": ""}}
    # $ne (not equal) で空文字以外 -> 任意のパスワードで認証成功
    user = db.users.find_one({"username": username, "password": password})
    return user
 
# === 安全なコード ===
def find_user_safe(request_data):
    """入力をバリデーションしてからクエリに使用"""
    username = request_data.get("username", "")
    password = request_data.get("password", "")
 
    # 型チェック: 文字列のみ許可（オブジェクトを拒否）
    if not isinstance(username, str) or not isinstance(password, str):
        raise ValueError("Invalid input type")
 
    # 長さ制限
    if len(username) > 100 or len(password) > 200:
        raise ValueError("Input too long")
 
    # MongoDB演算子の除去
    if any(key.startswith("$") for key in [username, password]
           if isinstance(key, str) and key.startswith("$")):
        raise ValueError("Invalid characters in input")
 
    user = db.users.find_one({
        "username": str(username),  # 明示的に文字列に変換
        "password_hash": hash_password(str(password)),
    })
    return user

# コード例6: MongoDB向け包括的なインジェクション防御
import re
from typing import Any
 
class MongoSanitizer:
    """MongoDB クエリの入力サニタイゼーション"""
 
    MONGO_OPERATORS = {
        "$gt", "$gte", "$lt", "$lte", "$ne", "$in", "$nin",
        "$and", "$or", "$not", "$nor", "$exists", "$type",
        "$regex", "$where", "$text", "$search", "$mod",
        "$all", "$elemMatch", "$size", "$slice",
    }
 
    @staticmethod
    def sanitize_value(value: Any) -> Any:
        """クエリ値をサニタイズ"""
        if isinstance(value, str):
            # 文字列はそのまま（安全）
            return value
        elif isinstance(value, (int, float, bool)):
            # プリミティブ型は安全
            return value
        elif isinstance(value, dict):
            # 辞書型（MongoDB演算子の可能性）
            for key in value:
                if key.startswith("$"):
                    raise ValueError(
                        f"MongoDB operator not allowed: {key}"
                    )
            return value
        elif isinstance(value, list):
            # リスト型は各要素をサニタイズ
            return [MongoSanitizer.sanitize_value(v) for v in value]
        else:
            raise ValueError(f"Unsupported type: {type(value)}")
 
    @staticmethod
    def sanitize_query(query: dict) -> dict:
        """クエリ全体をサニタイズ"""
        sanitized = {}
        for key, value in query.items():
            if key.startswith("$"):
                raise ValueError(f"Top-level operator not allowed: {key}")
            sanitized[key] = MongoSanitizer.sanitize_value(value)
        return sanitized
 
# 使用例
sanitizer = MongoSanitizer()
try:
    # 正常なクエリ
    safe_query = sanitizer.sanitize_query({
        "username": "alice",
        "age": 25,
    })
    result = db.users.find_one(safe_query)
 
    # 攻撃クエリ → 例外が発生
    malicious = sanitizer.sanitize_query({
        "username": "admin",
        "password": {"$ne": ""},  # ValueError!
    })
except ValueError as e:
    print(f"Blocked malicious query: {e}")

# コード例7: コマンドインジェクションの攻撃と対策
import subprocess
import shlex
import re
 
# === 脆弱なコード ===
def ping_host_vulnerable(host):
    """os.systemやshell=Trueでのコマンド実行 -> コマンドインジェクション"""
    import os
    os.system(f"ping -c 3 {host}")
    # 攻撃: host = "google.com; cat /etc/passwd"
    # 実行: ping -c 3 google.com; cat /etc/passwd
 
# === 安全なコード ===
def ping_host_safe(host: str) -> str:
    """安全なコマンド実行"""
    # Step 1: 入力バリデーション（ホワイトリスト方式）
    if not re.match(r'^[a-zA-Z0-9.\-]+$', host):
        raise ValueError(f"Invalid hostname: {host}")
 
    # Step 2: shell=Falseでリスト形式で引数を渡す
    result = subprocess.run(
        ["ping", "-c", "3", host],  # リスト形式 -> シェル解釈されない
        capture_output=True,
        text=True,
        timeout=10,
        shell=False,  # 明示的にFalse（デフォルトだが明示する）
    )
    return result.stdout
 
# === より安全: 外部コマンドを使わない ===
import socket
 
def check_host_reachable(host: str) -> bool:
    """外部コマンドを使わずにホストの到達性を確認"""
    if not re.match(r'^[a-zA-Z0-9.\-]+$', host):
        raise ValueError(f"Invalid hostname: {host}")
    try:
        socket.create_connection((host, 80), timeout=5)
        return True
    except (socket.timeout, socket.error):
        return False

# コード例8: 安全なサブプロセス実行のラッパー
import subprocess
import re
from typing import Optional
 
class SafeCommandRunner:
    """安全なコマンド実行ラッパー
 
    原則:
    1. shell=False を常に使用
    2. 引数はリスト形式で渡す
    3. 入力をホワイトリストで検証
    4. タイムアウトを設定
    5. 実行可能コマンドを制限
    """
 
    ALLOWED_COMMANDS = {
        "ping": {
            "path": "/usr/bin/ping",
            "allowed_args": ["-c", "-W"],
            "input_pattern": r'^[a-zA-Z0-9.\-]+$',
        },
        "dig": {
            "path": "/usr/bin/dig",
            "allowed_args": ["+short", "+timeout"],
            "input_pattern": r'^[a-zA-Z0-9.\-]+$',
        },
        "nslookup": {
            "path": "/usr/bin/nslookup",
            "allowed_args": [],
            "input_pattern": r'^[a-zA-Z0-9.\-]+$',
        },
    }
 
    def run(self, command: str, args: list, user_input: str,
            timeout: int = 10) -> Optional[str]:
        """安全にコマンドを実行"""
        # コマンドのホワイトリストチェック
        if command not in self.ALLOWED_COMMANDS:
            raise ValueError(f"Command not allowed: {command}")
 
        cmd_config = self.ALLOWED_COMMANDS[command]
 
        # 引数のホワイトリストチェック
        for arg in args:
            if arg not in cmd_config["allowed_args"]:
                raise ValueError(f"Argument not allowed: {arg}")
 
        # ユーザー入力のバリデーション
        if not re.match(cmd_config["input_pattern"], user_input):
            raise ValueError(f"Invalid input: {user_input}")
 
        # コマンド実行（フルパスを使用）
        cmd = [cmd_config["path"]] + args + [user_input]
 
        try:
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=timeout,
                shell=False,
                env={},  # 環境変数を空にする（PATH injection防止）
            )
            return result.stdout
        except subprocess.TimeoutExpired:
            return None
 
# 使用例
runner = SafeCommandRunner()
output = runner.run("ping", ["-c", "3"], "example.com")

# コード例9: LDAPインジェクションの攻撃と対策
 
# === 脆弱なコード ===
def search_user_vulnerable(username):
    """文字列連結によるLDAPフィルタ構築"""
    ldap_filter = f"(&(uid={username})(objectClass=person))"
    # 攻撃: username = "*)(uid=*))(|(uid=*"
    # 生成: (&(uid=*)(uid=*))(|(uid=*)(objectClass=person))
    # -> 全ユーザーが返される
    return ldap_conn.search_s(base_dn, ldap.SCOPE_SUBTREE, ldap_filter)
 
# === 安全なコード ===
def ldap_escape(value: str) -> str:
    """LDAP特殊文字をエスケープする（RFC 4515準拠）"""
    escape_chars = {
        '\\': r'\5c',
        '*': r'\2a',
        '(': r'\28',
        ')': r'\29',
        '\x00': r'\00',
    }
    result = value
    # バックスラッシュを最初にエスケープする（順序重要）
    for char, replacement in escape_chars.items():
        result = result.replace(char, replacement)
    return result
 
def search_user_safe(username: str):
    """エスケープ済みのLDAPフィルタを構築"""
    # 入力バリデーション
    if not username or len(username) > 100:
        raise ValueError("Invalid username")
 
    safe_username = ldap_escape(username)
    ldap_filter = f"(&(uid={safe_username})(objectClass=person))"
    return ldap_conn.search_s(base_dn, ldap.SCOPE_SUBTREE, ldap_filter)

# コード例10: Server-Side Template Injection (SSTI)
from flask import Flask, request, render_template_string
 
app = Flask(__name__)
 
# === 脆弱なコード ===
@app.route("/greet")
def greet_vulnerable():
    name = request.args.get("name", "")
    # ユーザー入力をテンプレートとして解釈 → SSTI!
    template = f"<h1>Hello, {name}!</h1>"
    return render_template_string(template)
    # 攻撃: name = {{7*7}} → "Hello, 49!"
    # 攻撃: name = {{config.SECRET_KEY}} → 秘密鍵が漏洩
    # 攻撃: name = {{''.__class__.__mro__[1].__subclasses__()}}
    #   → Python のクラス階層を辿ってRCE（リモートコード実行）
 
# === 安全なコード ===
@app.route("/greet")
def greet_safe():
    name = request.args.get("name", "")
    # テンプレートを固定し、変数として渡す
    return render_template_string("<h1>Hello, {{ name }}!</h1>", name=name)
    # {{ name }} はデータとして扱われ、テンプレート構文として解釈されない
 
# === さらに安全: Jinja2 サンドボックス ===
from jinja2.sandbox import SandboxedEnvironment
 
sandbox = SandboxedEnvironment()
 
@app.route("/greet-sandbox")
def greet_sandbox():
    name = request.args.get("name", "")
    template = sandbox.from_string("<h1>Hello, {{ name }}!</h1>")
    return template.render(name=name)

# コード例11: XPath インジェクションの攻撃と対策
from lxml import etree
 
# XML データ
xml_data = """
<users>
    <user>
        <username>admin</username>
        <password>secret123</password>
        <role>admin</role>
    </user>
    <user>
        <username>alice</username>
        <password>pass456</password>
        <role>user</role>
    </user>
</users>
"""
 
tree = etree.fromstring(xml_data.encode())
 
# === 脆弱なコード ===
def auth_vulnerable(username, password):
    """文字列連結によるXPath構築"""
    xpath = f"//user[username='{username}' and password='{password}']"
    # 攻撃: username = "' or '1'='1' or '"
    # XPath: //user[username='' or '1'='1' or '' and password='']
    # → 全ユーザーにマッチ
    result = tree.xpath(xpath)
    return len(result) > 0
 
# === 安全なコード ===
def auth_safe(username, password):
    """XPath 変数を使用した安全なクエリ"""
    # lxml の XPath 変数バインディング
    xpath = "//user[username=$username and password=$password]"
    result = tree.xpath(
        xpath,
        username=username,
        password=password,
    )
    return len(result) > 0

-- コード例12: DB権限の最小化（MySQL）
 
-- アプリケーション用ユーザーの作成（最小権限）
CREATE USER 'app_user'@'%' IDENTIFIED BY 'strong_password';
 
-- 必要なテーブルへの SELECT/INSERT/UPDATE のみ付与
GRANT SELECT, INSERT, UPDATE ON myapp.users TO 'app_user'@'%';
GRANT SELECT, INSERT ON myapp.orders TO 'app_user'@'%';
 
-- DELETE は特定の条件でのみ許可（ストアドプロシージャ経由）
-- 直接の DELETE 権限は付与しない
 
-- 危険な権限を付与しない
-- GRANT FILE ON *.* → ファイル読み書きが可能になる
-- GRANT PROCESS ON *.* → プロセスリストが見える
-- GRANT SUPER ON *.* → 管理者操作が可能
 
-- 読み取り専用ユーザー（レポート用）
CREATE USER 'report_user'@'%' IDENTIFIED BY 'another_password';
GRANT SELECT ON myapp.* TO 'report_user'@'%';
 
-- 権限の確認
SHOW GRANTS FOR 'app_user'@'%';

# コード例13: SQLAlchemy でのデータベース接続とエラーハンドリング
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
from sqlalchemy.exc import SQLAlchemyError
import logging
 
logger = logging.getLogger(__name__)
 
# エンジン作成（接続プールの設定）
engine = create_engine(
    "postgresql://app_user:password@localhost/myapp",
    pool_size=10,
    max_overflow=20,
    pool_recycle=3600,
    echo=False,  # 本番ではSQLログを出力しない（情報漏洩防止）
)
 
Session = sessionmaker(bind=engine)
 
class UserRepository:
    """安全なデータアクセス層"""
 
    def find_by_username(self, username: str):
        """ORMを使用した安全なクエリ"""
        with Session() as session:
            try:
                return session.query(User).filter(
                    User.username == username
                ).first()
            except SQLAlchemyError as e:
                # エラーメッセージにSQL詳細を含めない
                logger.error(f"Database error: {type(e).__name__}")
                raise ApplicationError("データの取得に失敗しました")
 
    def search_users(self, keyword: str, limit: int = 50):
        """LIKE検索の安全な実装"""
        with Session() as session:
            # LIKE のワイルドカード文字をエスケープ
            safe_keyword = keyword.replace('%', r'\%').replace('_', r'\_')
            return session.query(User).filter(
                User.username.ilike(f"%{safe_keyword}%", escape='\\')
            ).limit(min(limit, 100)).all()  # 上限を設定
 
    def execute_raw_query(self, query_template: str, params: dict):
        """生SQLが必要な場合の安全な実行"""
        with Session() as session:
            # text() + バインドパラメータを必ず使用
            stmt = text(query_template)
            return session.execute(stmt, params).fetchall()
 
# 使用例
repo = UserRepository()
user = repo.find_by_username("alice")  # パラメータ化される
results = repo.execute_raw_query(
    "SELECT * FROM users WHERE created_at > :date AND status = :status",
    {"date": "2024-01-01", "status": "active"}
)

-- ストアドプロシージャでも動的SQLは危険
 
-- NG: ストアドプロシージャ内で文字列連結
CREATE PROCEDURE search_users(IN search_term VARCHAR(100))
BEGIN
    SET @sql = CONCAT('SELECT * FROM users WHERE name LIKE "%',
                       search_term, '%"');
    PREPARE stmt FROM @sql;
    EXECUTE stmt;
END;
 
-- OK: パラメータバインディングを使用
CREATE PROCEDURE search_users_safe(IN search_term VARCHAR(100))
BEGIN
    SET @search = CONCAT('%', search_term, '%');
    PREPARE stmt FROM 'SELECT * FROM users WHERE name LIKE ?';
    EXECUTE stmt USING @search;
    DEALLOCATE PREPARE stmt;
END;

# ORM を使っていても安全でないパターン
 
from sqlalchemy import text
 
# NG: text() 内で文字列連結
def search_unsafe(keyword):
    query = text(f"SELECT * FROM users WHERE name LIKE '%{keyword}%'")
    return db.execute(query).fetchall()
 
# NG: filter() 内で文字列結合
def search_unsafe2(column_name, value):
    # カラム名を動的に指定 → インジェクション可能
    query = text(f"SELECT * FROM users WHERE {column_name} = :value")
    return db.execute(query, {"value": value}).fetchall()
 
# OK: カラム名もホワイトリストで検証
ALLOWED_COLUMNS = {"username", "email", "status"}
 
def search_safe(column_name, value):
    if column_name not in ALLOWED_COLUMNS:
        raise ValueError(f"Invalid column: {column_name}")
    query = text(f"SELECT * FROM users WHERE {column_name} = :value")
    return db.execute(query, {"value": value}).fetchall()

# コード例14: インジェクション脆弱性のテスト
import requests
from typing import List, Dict
 
class InjectionTester:
    """インジェクション脆弱性の基本テスト"""
 
    SQL_PAYLOADS = [
        "' OR '1'='1",
        "' OR '1'='1' --",
        "'; DROP TABLE users; --",
        "1 UNION SELECT NULL,NULL,NULL",
        "1' AND SLEEP(5) --",
        "admin'--",
    ]
 
    NOSQL_PAYLOADS = [
        '{"$ne": ""}',
        '{"$gt": ""}',
        '{"$regex": ".*"}',
    ]
 
    COMMAND_PAYLOADS = [
        "; ls -la",
        "| cat /etc/passwd",
        "$(whoami)",
        "`id`",
    ]
 
    def test_sql_injection(self, url: str, param: str) -> List[Dict]:
        """SQLインジェクションの基本テスト"""
        results = []
        baseline = requests.get(url, params={param: "normal_value"})
 
        for payload in self.SQL_PAYLOADS:
            response = requests.get(url, params={param: payload})
            suspicious = False
 
            # 異常な応答の検出
            if response.status_code == 500:
                suspicious = True  # SQLエラー
            if "sql" in response.text.lower() or "syntax" in response.text.lower():
                suspicious = True  # エラーメッセージ漏洩
            if len(response.text) != len(baseline.text):
                suspicious = True  # レスポンスサイズの変化
 
            results.append({
                "payload": payload,
                "status": response.status_code,
                "suspicious": suspicious,
                "response_length": len(response.text),
            })
 
        return results
 
    def test_error_disclosure(self, url: str, param: str) -> Dict:
        """エラーメッセージにDB情報が含まれていないか確認"""
        error_indicators = [
            "mysql", "postgresql", "sqlite", "oracle",
            "syntax error", "unclosed quotation",
            "unterminated string", "SQL",
        ]
        response = requests.get(url, params={param: "'"})
        found = [ind for ind in error_indicators
                 if ind.lower() in response.text.lower()]
        return {
            "error_disclosure": len(found) > 0,
            "indicators_found": found,
        }
 
# 使用例（必ず許可された環境でのみ実行すること）
# tester = InjectionTester()
# results = tester.test_sql_injection("http://localhost:8080/search", "q")

# 書き換え対象:
def search_products(category, min_price, max_price, sort_by):
    query = f"""
    SELECT * FROM products
    WHERE category = '{category}'
    AND price BETWEEN {min_price} AND {max_price}
    ORDER BY {sort_by}
    """
    return db.execute(query).fetchall()
 
# ヒント:
# - category, min_price, max_price はパラメータ化
# - sort_by はホワイトリストで検証（パラメータ化できない部分）

# NG: ブラックリスト方式
BLACKLIST = ["SELECT", "DROP", "DELETE", "UNION", "INSERT", "--", ";"]
 
def sanitize_input_bad(value):
    for keyword in BLACKLIST:
        value = value.replace(keyword, "")
    return value
# バイパス: "SELSELECTECT" → "SELECT" (除去後に元に戻る)
# バイパス: "sel/**/ect" → ブラックリストにマッチしない
 
# OK: パラメータ化クエリ
def query_safe(value):
    return db.execute("SELECT * FROM users WHERE name = ?", (value,))

# NG: クライアントサイドのみ
# JavaScript: if (input.includes("'")) { alert("不正な文字"); }
# → curl で直接 API にリクエスト送信されたら無意味
 
# OK: サーバーサイドで必ずバリデーション
from pydantic import BaseModel, validator
 
class SearchInput(BaseModel):
    keyword: str
 
    @validator("keyword")
    def validate_keyword(cls, v):
        if len(v) > 100:
            raise ValueError("Keyword too long")
        if not v.isalnum() and not v.replace(" ", "").isalnum():
            raise ValueError("Invalid characters")
        return v

# NG: SQLエラーをそのままユーザーに返す
@app.route("/search")
def search():
    try:
        result = db.execute(f"SELECT * FROM users WHERE id={request.args['id']}")
    except Exception as e:
        return f"Error: {str(e)}", 500
    # → "Error: near "OR": syntax error" のようなメッセージが
    #   攻撃者にDBの種類やクエリ構造のヒントを与える
 
# OK: 汎用的なエラーメッセージ + 内部ログ
@app.route("/search")
def search():
    try:
        result = db.execute("SELECT * FROM users WHERE id=?",
                           (request.args['id'],))
    except Exception as e:
        logger.error(f"Database error: {e}")  # 詳細は内部ログのみ
        return {"error": "リクエストの処理に失敗しました"}, 500

# 演習1: 基本実装のテンプレート
class Exercise1:
    """基本的な実装パターンの演習"""
 
    def __init__(self):
        self.data = []
 
    def validate_input(self, value):
        """入力値の検証"""
        if value is None:
            raise ValueError("入力値がNoneです")
        return True
 
    def process(self, value):
        """データ処理のメインロジック"""
        self.validate_input(value)
        self.data.append(value)
        return self.data
 
    def get_results(self):
        """処理結果の取得"""
        return {
            'count': len(self.data),
            'data': self.data
        }
 
# テスト
def test_exercise1():
    ex = Exercise1()
    assert ex.process(1) == [1]
    assert ex.process(2) == [1, 2]
    assert ex.get_results()['count'] == 2
 
    try:
        ex.process(None)
        assert False, "例外が発生するべき"
    except ValueError:
        pass
 
    print("全テスト合格!")
 
test_exercise1()

# 演習2: 応用パターン
from typing import List, Dict, Optional
from datetime import datetime
 
class AdvancedExercise:
    """応用パターンの演習"""
 
    def __init__(self, max_size: int = 100):
        self._items: List[Dict] = []
        self._max_size = max_size
        self._created_at = datetime.now()
 
    def add(self, key: str, value: any) -> bool:
        """アイテムの追加（サイズ制限付き）"""
        if len(self._items) >= self._max_size:
            return False
        self._items.append({
            'key': key,
            'value': value,
            'timestamp': datetime.now().isoformat()
        })
        return True
 
    def find(self, key: str) -> Optional[Dict]:
        """キーによる検索"""
        for item in reversed(self._items):
            if item['key'] == key:
                return item
        return None
 
    def remove(self, key: str) -> bool:
        """キーによる削除"""
        for i, item in enumerate(self._items):
            if item['key'] == key:
                self._items.pop(i)
                return True
        return False
 
    def stats(self) -> Dict:
        """統計情報"""
        return {
            'total_items': len(self._items),
            'max_size': self._max_size,
            'usage_percent': len(self._items) / self._max_size * 100,
            'uptime': str(datetime.now() - self._created_at)
        }
 
# テスト
def test_advanced():
    ex = AdvancedExercise(max_size=3)
    assert ex.add("a", 1) == True
    assert ex.add("b", 2) == True
    assert ex.add("c", 3) == True
    assert ex.add("d", 4) == False  # サイズ制限
    assert ex.find("b")['value'] == 2
    assert ex.remove("b") == True
    assert ex.find("b") is None
    stats = ex.stats()
    assert stats['total_items'] == 2
    print("応用テスト全合格!")
 
test_advanced()

# 演習3: パフォーマンス最適化
import time
from functools import lru_cache
 
# 最適化前（O(n^2)）
def slow_search(data: list, target: int) -> int:
    """非効率な検索"""
    for i in range(len(data)):
        for j in range(i + 1, len(data)):
            if data[i] + data[j] == target:
                return (i, j)
    return (-1, -1)
 
# 最適化後（O(n)）
def fast_search(data: list, target: int) -> tuple:
    """ハッシュマップを使った効率的な検索"""
    seen = {}
    for i, num in enumerate(data):
        complement = target - num
        if complement in seen:
            return (seen[complement], i)
        seen[num] = i
    return (-1, -1)
 
# ベンチマーク
def benchmark():
    import random
    data = list(range(5000))
    random.shuffle(data)
    target = data[100] + data[4000]
 
    start = time.time()
    result1 = slow_search(data, target)
    slow_time = time.time() - start
 
    start = time.time()
    result2 = fast_search(data, target)
    fast_time = time.time() - start
 
    print(f"非効率版: {slow_time:.4f}秒")
    print(f"効率版:   {fast_time:.6f}秒")
    print(f"高速化率: {slow_time/fast_time:.0f}倍")
 
benchmark()

# NG: IN句を文字列連結で構築
ids = [1, 2, 3]
query = f"SELECT * FROM users WHERE id IN ({','.join(map(str, ids))})"
 
# OK: プレースホルダを動的生成
ids = [1, 2, 3]
placeholders = ','.join(['?'] * len(ids))
query = f"SELECT * FROM users WHERE id IN ({placeholders})"
result = db.execute(query, ids)
 
# OK: SQLAlchemy
from sqlalchemy import select
stmt = select(User).where(User.id.in_([1, 2, 3]))

# NG: カラム名を直接埋め込み
column = request.args["sort_by"]
query = f"SELECT * FROM users ORDER BY {column}"
 
# OK: ホワイトリストで検証
ALLOWED_SORT_COLUMNS = {"username", "email", "created_at"}
 
column = request.args["sort_by"]
if column not in ALLOWED_SORT_COLUMNS:
    column = "created_at"  # デフォルト値
query = f"SELECT * FROM users ORDER BY {column}"