import re
 
# NG: ブラックリスト方式 — 回避が容易
def validate_input_blacklist(user_input: str) -> str:
    """危険な文字を除去する方式は常に不完全"""
    dangerous = ["<script>", "DROP TABLE", "OR 1=1"]
    for d in dangerous:
        user_input = user_input.replace(d, "")
    return user_input
    # 攻撃者は "<scr<script>ipt>" のような入れ子で回避可能
    # エンコーディング変換(%3Cscript%3E)でも回避可能
 
# OK: ホワイトリスト方式 — 許可パターンのみ受理
def validate_username(username: str) -> str:
    """ユーザ名は英数字とアンダースコアのみ許可"""
    if not re.match(r'^[a-zA-Z0-9_]{3,20}$', username):
        raise ValueError(
            "ユーザ名は3-20文字の英数字とアンダースコアのみ使用可能です"
        )
    return username

from pydantic import BaseModel, Field, field_validator, EmailStr
from datetime import date
from typing import Optional
import re
 
class UserRegistration(BaseModel):
    """型安全な入力バリデーション（Pydantic v2）"""
    username: str = Field(
        min_length=3,
        max_length=20,
        pattern=r'^[a-zA-Z0-9_]+$',
        description="英数字とアンダースコアのみ",
    )
    email: EmailStr
    password: str = Field(min_length=12, max_length=128)
    birth_date: date
    age: int = Field(ge=13, le=150)
    bio: Optional[str] = Field(default=None, max_length=500)
 
    @field_validator('password')
    @classmethod
    def validate_password_strength(cls, v: str) -> str:
        """パスワード強度の検証"""
        if not re.search(r'[A-Z]', v):
            raise ValueError('大文字を1文字以上含む必要があります')
        if not re.search(r'[a-z]', v):
            raise ValueError('小文字を1文字以上含む必要があります')
        if not re.search(r'\d', v):
            raise ValueError('数字を1文字以上含む必要があります')
        if not re.search(r'[!@#$%^&*(),.?":{}|<>]', v):
            raise ValueError('特殊文字を1文字以上含む必要があります')
        return v
 
    @field_validator('birth_date')
    @classmethod
    def validate_birth_date(cls, v: date) -> date:
        """未来の日付や非現実的な日付を拒否"""
        if v > date.today():
            raise ValueError('生年月日は未来の日付にできません')
        if v.year < 1900:
            raise ValueError('無効な生年月日です')
        return v
 
# 使用例
try:
    user = UserRegistration(
        username="john_doe",
        email="john@example.com",
        password="SecureP@ss123!",
        birth_date="1990-01-15",
        age=34,
    )
    print(f"検証成功: {user.username}")
except Exception as e:
    print(f"検証エラー: {e}")

import sqlite3
from typing import Optional
 
# NG: 文字列結合による SQL 構築
def get_user_bad(username: str) -> Optional[dict]:
    """脆弱なコード — SQL インジェクションが可能"""
    conn = sqlite3.connect('app.db')
    cursor = conn.cursor()
    query = f"SELECT * FROM users WHERE name = '{username}'"
    # 入力: ' OR '1'='1' --
    # 生成SQL: SELECT * FROM users WHERE name = '' OR '1'='1' --'
    # → 全ユーザの情報が取得される
    cursor.execute(query)
    return cursor.fetchone()
 
# OK: パラメータ化クエリ (プリペアドステートメント)
def get_user_good(username: str) -> Optional[dict]:
    """安全なコード — パラメータがクエリ構造と分離"""
    conn = sqlite3.connect('app.db')
    cursor = conn.cursor()
    # プレースホルダを使用 (DBドライバが安全にエスケープ)
    query = "SELECT * FROM users WHERE name = ?"
    cursor.execute(query, (username,))
    row = cursor.fetchone()
    if row:
        columns = [desc[0] for desc in cursor.description]
        return dict(zip(columns, row))
    return None
 
# OK: ORM を使用 (SQLAlchemy)
from sqlalchemy import select
from sqlalchemy.orm import Session
 
def get_user_orm(session: Session, username: str) -> Optional["User"]:
    """ORM は内部でパラメータ化クエリを生成"""
    stmt = select(User).where(User.name == username)
    return session.execute(stmt).scalar_one_or_none()
 
# 注意: ORM でも生SQLを使う場合は注意が必要
from sqlalchemy import text
 
def search_users_safe(session: Session, search_term: str) -> list:
    """ORM の text() でもバインドパラメータを使用する"""
    # NG: f"SELECT * FROM users WHERE name LIKE '%{search_term}%'"
    # OK: バインドパラメータ使用
    stmt = text("SELECT * FROM users WHERE name LIKE :term")
    result = session.execute(stmt, {"term": f"%{search_term}%"})
    return result.fetchall()

import subprocess
import re
import shlex
from pathlib import Path
 
# NG: シェルコマンドの文字列結合
def ping_bad(host: str) -> str:
    """脆弱なコード — 任意コマンド実行が可能"""
    import os
    os.system(f"ping -c 1 {host}")
    # 入力: "8.8.8.8; rm -rf /" → ping 実行後に全ファイル削除
    # 入力: "8.8.8.8 && cat /etc/passwd" → パスワードファイル読み取り
    return "done"
 
# OK: subprocess + リスト引数 (shell=False) + ホワイトリスト
def ping_good(host: str) -> str:
    """安全なコード — 引数が分離されシェル解釈されない"""
    # Step 1: ホワイトリストで入力を検証
    if not re.match(r'^(\d{1,3}\.){3}\d{1,3}$', host):
        raise ValueError(f"無効なIPアドレス形式: {host}")
 
    # Step 2: 各オクテットの範囲チェック
    octets = host.split('.')
    for octet in octets:
        if not 0 <= int(octet) <= 255:
            raise ValueError(f"無効なIPアドレス: {host}")
 
    # Step 3: shell=False でリスト引数として実行
    result = subprocess.run(
        ["ping", "-c", "1", "-W", "3", host],  # リスト形式
        capture_output=True,
        text=True,
        timeout=10,
    )
    return result.stdout
 
# OK: 外部コマンドの代わりに Python ライブラリを使用（最も推奨）
import socket
 
def check_host_reachable(host: str, port: int = 80, timeout: float = 3.0) -> bool:
    """外部コマンドを使わず Python で直接ネットワーク確認"""
    try:
        socket.create_connection((host, port), timeout=timeout)
        return True
    except (socket.timeout, ConnectionRefusedError, OSError):
        return False

import os
from pathlib import Path
 
# NG: ユーザ入力をパスに直接使用
def read_file_bad(filename: str) -> str:
    """脆弱なコード — 任意のファイルが読める"""
    path = f"/app/uploads/{filename}"
    # 入力: "../../etc/passwd" → /etc/passwd を読み取れる
    # 入力: "....//....//etc/passwd" → 正規化前に結合される
    return open(path).read()
 
# OK: パスの正規化と検証
def read_file_good(filename: str) -> str:
    """安全なコード — ベースディレクトリ外へのアクセスを防止"""
    base_dir = Path("/app/uploads").resolve()  # 絶対パスに正規化
    # ファイル名からディレクトリトラバーサル文字を除去
    safe_filename = Path(filename).name  # ディレクトリ部分を除去
    full_path = (base_dir / safe_filename).resolve()
 
    # ベースディレクトリ外へのアクセスを拒否
    if not str(full_path).startswith(str(base_dir)):
        raise PermissionError("アクセス拒否: パストラバーサルが検出されました")
 
    if not full_path.is_file():
        raise FileNotFoundError(f"ファイルが見つかりません: {safe_filename}")
 
    # ファイルサイズの制限チェック
    if full_path.stat().st_size > 10 * 1024 * 1024:  # 10MB
        raise ValueError("ファイルサイズが制限を超えています")
 
    return full_path.read_text(encoding='utf-8')

import ipaddress
import urllib.parse
from typing import Optional
 
def validate_url_for_ssrf(url: str) -> str:
    """SSRF を防止するURL検証"""
    parsed = urllib.parse.urlparse(url)
 
    # スキーマの制限
    if parsed.scheme not in ('http', 'https'):
        raise ValueError(f"許可されないスキーマ: {parsed.scheme}")
 
    # ホスト名の解決
    import socket
    try:
        resolved_ip = socket.gethostbyname(parsed.hostname)
    except socket.gaierror:
        raise ValueError(f"ホスト名を解決できません: {parsed.hostname}")
 
    # プライベートIPアドレスの拒否
    ip = ipaddress.ip_address(resolved_ip)
    if ip.is_private or ip.is_loopback or ip.is_link_local:
        raise ValueError(
            f"内部ネットワークへのアクセスは禁止されています: {resolved_ip}"
        )
 
    # AWS メタデータエンドポイントの拒否
    if resolved_ip == "169.254.169.254":
        raise ValueError("メタデータサービスへのアクセスは禁止されています")
 
    return url
 
# 使用例
import requests
 
def fetch_external_resource(url: str) -> Optional[str]:
    """安全な外部リソース取得"""
    validated_url = validate_url_for_ssrf(url)
    response = requests.get(
        validated_url,
        timeout=10,
        allow_redirects=False,  # リダイレクトでSSRFを回避されるのを防止
    )
    response.raise_for_status()
    return response.text

// ========================================
// パターン1: React (JSX の自動エスケープ)
// ========================================
function UserProfile({ user }) {
  return (
    <div>
      {/* OK: JSX は自動的に HTML エスケープする */}
      <h1>{user.name}</h1>
      <p>{user.bio}</p>
 
      {/* NG: dangerouslySetInnerHTML は XSS リスク */}
      <div dangerouslySetInnerHTML={{ __html: user.bio }} />
 
      {/* OK: サニタイズライブラリ (DOMPurify) を通す */}
      <div dangerouslySetInnerHTML={{
        __html: DOMPurify.sanitize(user.bio, {
          ALLOWED_TAGS: ['b', 'i', 'em', 'strong', 'a', 'p', 'br'],
          ALLOWED_ATTR: ['href', 'target', 'rel'],
          FORBID_ATTR: ['style', 'onerror', 'onclick'],
        })
      }} />
 
      {/* NG: href に javascript: プロトコルが入る可能性 */}
      <a href={user.website}>Website</a>
 
      {/* OK: URL スキーマを検証 */}
      <a href={sanitizeUrl(user.website)}>Website</a>
    </div>
  );
}
 
/**
 * URL のスキーマを検証して javascript: プロトコル等を排除
 */
function sanitizeUrl(url) {
  if (!url) return '#';
  try {
    const parsed = new URL(url);
    if (!['http:', 'https:', 'mailto:'].includes(parsed.protocol)) {
      return '#';
    }
    return url;
  } catch {
    return '#';
  }
}

# ========================================
# パターン2: サーバサイド (Python/Flask)
# ========================================
from markupsafe import escape, Markup
from flask import Flask, render_template_string
 
app = Flask(__name__)
 
@app.route('/profile/<username>')
def profile(username: str):
    # Jinja2 テンプレートは {{ }} 内を自動エスケープ
    template = """
    <h1>{{ username }}</h1>
    <p>{{ bio }}</p>
    """
    # username に <script>alert(1)</script> が入っても安全
    return render_template_string(template, username=username, bio="Hello")
 
# 手動エスケープが必要な場合
def escape_for_html(text: str) -> str:
    """手動 HTML エスケープ"""
    return str(escape(text))
 
# NG: |safe フィルターの誤用
# {{ user_input | safe }}  → エスケープされない！
# OK: サニタイズ済みデータにのみ |safe を使用
# {{ sanitized_html | safe }}

// ========================================
// パターン3: DOM ベース XSS の防止
// ========================================
 
// NG: DOM に直接 innerHTML を代入
function displaySearchResults_bad(query) {
  // URL: ?q=<img src=x onerror=alert(1)>
  document.getElementById('results').innerHTML =
    `検索結果: ${query}`;  // XSS!
}
 
// OK: textContent を使用（HTMLとして解釈されない）
function displaySearchResults_good(query) {
  const resultsEl = document.getElementById('results');
  resultsEl.textContent = `検索結果: ${query}`;
}
 
// OK: DOM API で安全に要素を構築
function displaySearchResults_best(query) {
  const resultsEl = document.getElementById('results');
  const textNode = document.createTextNode(`検索結果: ${query}`);
  resultsEl.replaceChildren(textNode);
}

# CSP ヘッダーの実装 (Flask)
import secrets
 
@app.after_request
def add_security_headers(response):
    """セキュリティヘッダーを全レスポンスに付与"""
    nonce = secrets.token_urlsafe(32)
 
    csp_directives = [
        "default-src 'none'",
        f"script-src 'self' 'nonce-{nonce}'",
        "style-src 'self'",
        "img-src 'self' https://cdn.example.com",
        "font-src 'self'",
        "connect-src 'self' https://api.example.com",
        "frame-ancestors 'none'",
        "base-uri 'self'",
        "form-action 'self'",
        "upgrade-insecure-requests",
    ]
    response.headers['Content-Security-Policy'] = '; '.join(csp_directives)
 
    # その他のセキュリティヘッダー
    response.headers['X-Content-Type-Options'] = 'nosniff'
    response.headers['X-Frame-Options'] = 'DENY'
    response.headers['X-XSS-Protection'] = '0'  # 古いブラウザ対策は無効化
    response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin'
    response.headers['Permissions-Policy'] = (
        'camera=(), microphone=(), geolocation=()'
    )
 
    # nonce をテンプレートに渡す
    response.headers['X-Nonce'] = nonce
    return response

# CSP Report-Only モードで段階的導入
@app.after_request
def add_csp_report_only(response):
    """CSP を Report-Only モードで試験運用"""
    response.headers['Content-Security-Policy-Report-Only'] = (
        "default-src 'self'; "
        "report-uri /api/csp-report; "
        "report-to csp-endpoint"
    )
    return response
 
@app.route('/api/csp-report', methods=['POST'])
def csp_report():
    """CSP 違反レポートを収集"""
    import json
    import logging
    logger = logging.getLogger('csp')
 
    report = json.loads(request.data)
    logger.warning(
        "CSP Violation: blocked_uri=%s, violated_directive=%s, document_uri=%s",
        report.get('csp-report', {}).get('blocked-uri'),
        report.get('csp-report', {}).get('violated-directive'),
        report.get('csp-report', {}).get('document-uri'),
    )
    return '', 204

import logging
import uuid
import traceback
from flask import Flask, jsonify, request, g
from functools import wraps
from datetime import datetime
 
app = Flask(__name__)
logger = logging.getLogger(__name__)
 
# ========================================
# エラークラスの階層設計
# ========================================
class AppError(Exception):
    """アプリケーションエラー基底クラス"""
    def __init__(
        self,
        message: str,
        status_code: int = 500,
        error_code: str = "INTERNAL_ERROR",
        internal_message: str = None,
    ):
        self.message = message              # ユーザに返すメッセージ
        self.status_code = status_code
        self.error_code = error_code        # 機械判読可能なエラーコード
        self.internal_message = internal_message  # ログ用の詳細
 
class NotFoundError(AppError):
    def __init__(self, resource: str = "Resource", resource_id: str = None):
        internal_msg = f"{resource} not found (id={resource_id})" if resource_id else None
        super().__init__(
            message=f"{resource} not found",
            status_code=404,
            error_code="NOT_FOUND",
            internal_message=internal_msg,
        )
 
class ValidationError(AppError):
    def __init__(self, details: list):
        super().__init__(
            message="Validation failed",
            status_code=400,
            error_code="VALIDATION_ERROR",
        )
        self.details = details
 
class AuthenticationError(AppError):
    def __init__(self):
        super().__init__(
            message="Authentication required",
            status_code=401,
            error_code="UNAUTHORIZED",
        )
 
class RateLimitError(AppError):
    def __init__(self, retry_after: int = 60):
        super().__init__(
            message="Too many requests",
            status_code=429,
            error_code="RATE_LIMITED",
        )
        self.retry_after = retry_after
 
# ========================================
# リクエスト追跡ミドルウェア
# ========================================
@app.before_request
def assign_request_id():
    """各リクエストに一意の追跡IDを割り当て"""
    g.request_id = request.headers.get(
        'X-Request-ID',
        str(uuid.uuid4())[:8]
    )
    g.request_start = datetime.utcnow()
 
# ========================================
# エラーハンドラ
# ========================================
@app.errorhandler(AppError)
def handle_app_error(error):
    """アプリケーションエラーのハンドリング"""
    request_id = getattr(g, 'request_id', 'unknown')
 
    # 内部ログには詳細を記録
    logger.error(
        "AppError: error_code=%s message=%s request_id=%s path=%s "
        "method=%s internal=%s",
        error.error_code, error.message, request_id,
        request.path, request.method,
        error.internal_message or "N/A",
    )
 
    # ユーザには最小限の情報のみ返す
    response = {
        "error": error.message,
        "errorCode": error.error_code,
        "requestId": request_id,
    }
    if hasattr(error, 'details'):
        response["details"] = error.details
 
    resp = jsonify(response)
    resp.status_code = error.status_code
 
    if hasattr(error, 'retry_after'):
        resp.headers['Retry-After'] = str(error.retry_after)
 
    return resp
 
@app.errorhandler(Exception)
def handle_unexpected_error(error):
    """予期しないエラーのハンドリング"""
    request_id = getattr(g, 'request_id', 'unknown')
 
    # 予期しないエラーはスタックトレース付きでログ
    logger.exception(
        "Unexpected error: request_id=%s path=%s method=%s",
        request_id, request.path, request.method,
    )
 
    # ユーザには一般的なメッセージのみ
    return jsonify({
        "error": "Internal Server Error",
        "errorCode": "INTERNAL_ERROR",
        "requestId": request_id,
    }), 500

import json
import logging
from datetime import datetime
 
class StructuredFormatter(logging.Formatter):
    """JSON構造化ログフォーマッタ"""
 
    def format(self, record):
        log_entry = {
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno,
        }
 
        # request_id がある場合は追加
        if hasattr(record, 'request_id'):
            log_entry["request_id"] = record.request_id
 
        # 例外情報がある場合
        if record.exc_info:
            log_entry["exception"] = {
                "type": record.exc_info[0].__name__,
                "message": str(record.exc_info[1]),
                "traceback": self.formatException(record.exc_info),
            }
 
        # セキュリティ上の注意:
        # パスワード、トークン、個人情報をログに含めない
        return json.dumps(log_entry, ensure_ascii=False)
 
# 使用例
handler = logging.StreamHandler()
handler.setFormatter(StructuredFormatter())
logger = logging.getLogger('app')
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# ========================================
# 方法1: CSRF トークン (Flask-WTF)
# ========================================
from flask_wtf.csrf import CSRFProtect
 
app = Flask(__name__)
app.config['SECRET_KEY'] = os.environ['SECRET_KEY']
csrf = CSRFProtect(app)
 
# HTMLフォーム内にトークンを埋め込む
# <form method="POST">
#   <input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
#   <button type="submit">送信</button>
# </form>
 
# AJAX リクエストの場合
# <meta name="csrf-token" content="{{ csrf_token() }}">
# <script>
# fetch('/api/data', {
#   method: 'POST',
#   headers: {
#     'X-CSRFToken': document.querySelector('meta[name=csrf-token]').content,
#     'Content-Type': 'application/json',
#   },
#   body: JSON.stringify(data),
# })
# </script>
 
# ========================================
# 方法2: SameSite Cookie (最も簡単)
# ========================================
app.config.update(
    SESSION_COOKIE_SAMESITE='Lax',   # クロスサイトからの POST で Cookie を送信しない
    SESSION_COOKIE_SECURE=True,       # HTTPS のみ
    SESSION_COOKIE_HTTPONLY=True,      # JavaScript からアクセス不可
)
 
# ========================================
# 方法3: Double Submit Cookie パターン
# ========================================
import secrets
 
@app.before_request
def set_csrf_cookie():
    """CSRF トークンを Cookie とレスポンスの両方に設定"""
    if 'csrf_token' not in request.cookies:
        g.csrf_token = secrets.token_urlsafe(32)
    else:
        g.csrf_token = request.cookies['csrf_token']
 
@app.after_request
def add_csrf_cookie(response):
    """CSRF Cookie を設定"""
    response.set_cookie(
        'csrf_token',
        g.csrf_token,
        httponly=False,  # JS から読み取り可能にする
        secure=True,
        samesite='Lax',
    )
    return response

# ========================================
# 推奨: bcrypt によるパスワードハッシュ化
# ========================================
import bcrypt
 
def hash_password(password: str) -> bytes:
    """パスワードを安全にハッシュ化"""
    # bcrypt は自動的にソルト(16バイト)を生成
    # rounds=12: 約0.3秒/ハッシュ (ブルートフォース対策)
    return bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt(rounds=12))
 
def verify_password(password: str, hashed: bytes) -> bool:
    """パスワードの検証（タイミング攻撃耐性あり）"""
    return bcrypt.checkpw(password.encode('utf-8'), hashed)
 
# ========================================
# より推奨: argon2 (Password Hashing Competition 優勝)
# ========================================
from argon2 import PasswordHasher
 
ph = PasswordHasher(
    time_cost=3,       # 反復回数
    memory_cost=65536,  # メモリ使用量(KB) = 64MB
    parallelism=4,      # 並列度
)
 
def hash_password_argon2(password: str) -> str:
    """Argon2id でパスワードをハッシュ化"""
    return ph.hash(password)
 
def verify_password_argon2(password: str, hashed: str) -> bool:
    """Argon2id でパスワードを検証"""
    try:
        return ph.verify(hashed, password)
    except Exception:
        return False
 
# ========================================
# NG: 以下は絶対に使用しない
# ========================================
# import hashlib
# hashlib.md5(password.encode()).hexdigest()      # NG: 高速すぎる + ソルトなし
# hashlib.sha256(password.encode()).hexdigest()    # NG: 高速すぎる + ソルトなし
# hashlib.sha256((salt + password).encode()).hexdigest()  # NG: 高速すぎる

from flask import Flask, session
import os
 
app = Flask(__name__)
 
# ========================================
# セッション設定のベストプラクティス
# ========================================
app.config.update(
    SECRET_KEY=os.environ['SESSION_SECRET'],   # 十分な長さのランダム値(32バイト以上)
    SESSION_COOKIE_SECURE=True,                # HTTPS でのみ送信
    SESSION_COOKIE_HTTPONLY=True,               # JavaScript からアクセス不可
    SESSION_COOKIE_SAMESITE='Lax',             # CSRF 防止
    PERMANENT_SESSION_LIFETIME=1800,           # 30分で期限切れ
    SESSION_COOKIE_NAME='__Host-session',      # __Host- プレフィックスで保護強化
)
 
# ========================================
# セッション固定攻撃の防止
# ========================================
@app.route('/login', methods=['POST'])
def login():
    """ログイン成功時にセッションIDを再生成"""
    # 認証処理...
    if authenticate(request.form['email'], request.form['password']):
        # 重要: ログイン成功後にセッションを再生成
        session.clear()  # 古いセッションデータを破棄
        session.regenerate()  # 新しいセッションIDを生成
        session['user_id'] = user.id
        session['login_time'] = datetime.utcnow().isoformat()
        return redirect('/dashboard')
    return render_template('login.html', error="Invalid credentials")
 
# ========================================
# セッションの妥当性検証
# ========================================
@app.before_request
def validate_session():
    """各リクエストでセッションの妥当性を確認"""
    if 'user_id' in session:
        login_time = datetime.fromisoformat(session.get('login_time', ''))
        # 長時間アイドルの場合はセッションを無効化
        if (datetime.utcnow() - login_time).total_seconds() > 3600:
            session.clear()
            return redirect('/login?reason=timeout')

# ========================================
# NG: pickle による安全でないデシリアライゼーション
# ========================================
import pickle
 
# 絶対にやってはいけない: 信頼できないデータを pickle.loads
def load_user_data_bad(serialized_data: bytes):
    """脆弱: pickle は任意のPythonオブジェクトを復元する"""
    return pickle.loads(serialized_data)
    # 攻撃者は __reduce__ メソッドを持つオブジェクトを送信し
    # os.system('rm -rf /') などの任意コマンドを実行できる
 
# ========================================
# OK: JSON を使用する（安全なシリアライゼーション）
# ========================================
import json
from typing import Any
 
def load_user_data_good(serialized_data: str) -> dict:
    """安全: JSON はデータのみを復元する（コード実行不可）"""
    data = json.loads(serialized_data)
    # さらに型を検証
    if not isinstance(data, dict):
        raise ValueError("Expected a JSON object")
    return data
 
# ========================================
# OK: YAML の安全な読み込み
# ========================================
import yaml
 
def load_config_safe(yaml_content: str) -> dict:
    """安全: safe_load は基本型のみ許可"""
    # NG: yaml.load(yaml_content, Loader=yaml.FullLoader)  # コード実行リスク
    # OK: yaml.safe_load は str, int, list, dict のみ
    return yaml.safe_load(yaml_content)

// NG: フロントエンドのみでバリデーション
async function submitForm() {
  const age = document.getElementById('age').value;
  if (age < 0 || age > 150) {
    alert('無効な年齢です');
    return;
  }
  // 攻撃者は DevTools, curl, Burp Suite で直接 API を叩ける
  await fetch('/api/users', {
    method: 'POST',
    body: JSON.stringify({ age: -999 }),  // バリデーションを完全にバイパス
  });
}

# OK: サーバ側でも必ず検証
from pydantic import BaseModel, Field
 
class UserUpdate(BaseModel):
    age: int = Field(ge=0, le=150)  # サーバ側で型+範囲を強制
 
@app.route('/api/users', methods=['POST'])
def create_user():
    try:
        data = UserUpdate(**request.json)  # 検証失敗は自動で422
    except ValueError as e:
        return jsonify({"error": "Validation failed", "details": str(e)}), 400
    # フロントエンド検証はUXのため、セキュリティはサーバ側の責任

# NG: ユーザ列挙攻撃を可能にするエラーメッセージ
@app.route('/login', methods=['POST'])
def login_bad():
    user = User.query.filter_by(email=request.form['email']).first()
    if not user:
        return {"error": "このメールアドレスは登録されていません"}, 401
    if not verify_password(request.form['password'], user.password_hash):
        return {"error": "パスワードが間違っています"}, 401
    # → 攻撃者はエラーメッセージからユーザの存在を確認できる
    # → メールアドレスのリストを作成し、パスワードスプレー攻撃に利用
 
# OK: 同一メッセージを返す + タイミング攻撃も防止
import time
import secrets
 
@app.route('/login', methods=['POST'])
def login_good():
    user = User.query.filter_by(email=request.form['email']).first()
    if not user:
        # ユーザが存在しなくてもパスワード検証と同等の時間をかける
        bcrypt.hashpw(b"dummy", bcrypt.gensalt(rounds=12))
        return {"error": "メールアドレスまたはパスワードが無効です"}, 401
 
    if not verify_password(request.form['password'], user.password_hash):
        return {"error": "メールアドレスまたはパスワードが無効です"}, 401
 
    # 成功
    return {"token": generate_jwt(user)}

# NG: ソースコードに秘密情報を直接記述
SECRET_KEY = "my-super-secret-key-12345"
DATABASE_URL = "postgresql://admin:password123@db.example.com/prod"
API_KEY = "sk-live-abcdef1234567890"
 
# OK: 環境変数やシークレットマネージャから取得
import os
 
SECRET_KEY = os.environ['SECRET_KEY']
DATABASE_URL = os.environ['DATABASE_URL']
 
# さらに安全: AWS Secrets Manager から取得
import boto3
import json
 
def get_secret(secret_name: str) -> dict:
    """AWS Secrets Manager からシークレットを取得"""
    client = boto3.client('secretsmanager')
    response = client.get_secret_value(SecretId=secret_name)
    return json.loads(response['SecretString'])

import re
 
# NG: 壊滅的バックトラッキングを引き起こす正規表現
# "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa!" のような入力で指数時間かかる
bad_pattern = re.compile(r'^(a+)+$')
 
# OK: 非バックトラッキングな正規表現、または入力長を制限
def validate_safe(text: str) -> bool:
    if len(text) > 100:  # 入力長を制限
        return False
    # 原子的グループやポゼッシブ量指定子を使えない場合は
    # パターンを単純化する
    return bool(re.match(r'^a+$', text))