# コード例1: 安全なアクセス制御の実装
from functools import wraps
from flask import Flask, request, abort, g
import uuid
 
app = Flask(__name__)
 
# ===== 悪い例: IDORの脆弱性 =====
@app.route("/api/orders/<int:order_id>")
def get_order_bad(order_id):
    # 誰でも任意のorder_idにアクセスできてしまう
    order = db.query("SELECT * FROM orders WHERE id = ?", order_id)
    return jsonify(order)
 
# ===== 良い例: オーナーシップチェック付き =====
@app.route("/api/orders/<int:order_id>")
@login_required
def get_order_good(order_id):
    order = db.query(
        "SELECT * FROM orders WHERE id = ? AND user_id = ?",
        order_id, g.current_user.id  # ユーザーIDでフィルタ
    )
    if not order:
        abort(404)  # 403ではなく404（情報漏洩防止）
    return jsonify(order)
 
 
# ===== ロールベースアクセス制御 (RBAC) =====
class Permission:
    """権限定義"""
    READ_OWN = "read:own"
    READ_ALL = "read:all"
    WRITE_OWN = "write:own"
    WRITE_ALL = "write:all"
    ADMIN = "admin"
 
ROLE_PERMISSIONS = {
    "user":    [Permission.READ_OWN, Permission.WRITE_OWN],
    "manager": [Permission.READ_OWN, Permission.WRITE_OWN, Permission.READ_ALL],
    "admin":   [Permission.READ_OWN, Permission.WRITE_OWN,
                Permission.READ_ALL, Permission.WRITE_ALL, Permission.ADMIN],
}
 
def require_permission(permission):
    """パーミッションベースの認可デコレータ"""
    def decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            if not g.current_user:
                abort(401)
            user_permissions = ROLE_PERMISSIONS.get(g.current_user.role, [])
            if permission not in user_permissions:
                # 監査ログを記録
                audit_log.warning(
                    f"Access denied: user={g.current_user.id}, "
                    f"permission={permission}, path={request.path}"
                )
                abort(403)
            return f(*args, **kwargs)
        return wrapper
    return decorator
 
 
def require_role(role):
    """ロールベースアクセス制御デコレータ"""
    def decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            if not g.current_user or g.current_user.role != role:
                abort(403)
            return f(*args, **kwargs)
        return wrapper
    return decorator
 
@app.route("/admin/users")
@login_required
@require_role("admin")
def admin_users():
    """管理者のみアクセス可能"""
    return jsonify(db.query("SELECT id, name FROM users"))
 
 
# ===== ABAC (属性ベースアクセス制御) の実装 =====
class ABACPolicy:
    """属性ベースアクセス制御"""
 
    def __init__(self):
        self.policies = []
 
    def add_policy(self, resource_type, action, condition_fn):
        self.policies.append({
            "resource_type": resource_type,
            "action": action,
            "condition": condition_fn,
        })
 
    def check(self, user, resource_type, action, resource=None):
        for policy in self.policies:
            if (policy["resource_type"] == resource_type and
                policy["action"] == action):
                if policy"condition":
                    return True
        return False
 
abac = ABACPolicy()
 
# ポリシー定義: ドキュメントの所有者のみ編集可能
abac.add_policy(
    "document", "edit",
    lambda user, doc: doc.owner_id == user.id
)
 
# ポリシー定義: 同じ部署のマネージャーは閲覧可能
abac.add_policy(
    "document", "view",
    lambda user, doc: (
        user.department == doc.department and
        user.role in ["manager", "admin"]
    )
)
 
# ポリシー定義: 公開ドキュメントは誰でも閲覧可能
abac.add_policy(
    "document", "view",
    lambda user, doc: doc.visibility == "public"
)

# コード例2: 推測困難なIDの使用
import uuid
 
class Order:
    def __init__(self, user_id, items):
        # 連番IDではなくUUIDv4を使用
        self.id = str(uuid.uuid4())  # "a3b8f9c2-1d4e-4a6b-8c3d-9e7f0a1b2c3d"
        self.user_id = user_id
        self.items = items
 
# 注意: UUIDの使用はIDORの根本対策ではない
# 推測を困難にするが、認可チェックは依然として必須
 
@app.route("/api/orders/<order_id>")
@login_required
def get_order(order_id):
    # UUIDでもオーナーシップチェックは必須
    order = Order.query.filter_by(
        id=order_id,
        user_id=g.current_user.id
    ).first_or_404()
    return jsonify(order.to_dict())

# コード例3: 適切な暗号化の実装
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
import bcrypt
import os
import base64
import hmac
import hashlib
 
class SecureCrypto:
    """安全な暗号化ユーティリティ"""
 
    @staticmethod
    def hash_password(password: str) -> str:
        """パスワードのハッシュ化（bcrypt使用）
 
        bcryptが推奨される理由:
        1. ソルトが自動生成・付与される
        2. コストファクター（rounds）で計算時間を調整可能
        3. GPU/ASIC による並列攻撃に対して耐性がある
        4. タイミング攻撃に対して一定時間比較を行う
        """
        # 悪い例: MD5やSHA-256の直接使用
        # hashlib.md5(password.encode()).hexdigest()  # NG!
        # hashlib.sha256(password.encode()).hexdigest()  # NG!
 
        # 良い例: bcryptによるソルト付きハッシュ
        # rounds=12 は2025年時点で推奨される最小値
        # サーバの性能に応じて13-14まで上げることが望ましい
        salt = bcrypt.gensalt(rounds=12)
        return bcrypt.hashpw(password.encode(), salt).decode()
 
    @staticmethod
    def verify_password(password: str, hashed: str) -> bool:
        """パスワードの検証（定数時間比較）"""
        return bcrypt.checkpw(password.encode(), hashed.encode())
 
    @staticmethod
    def hash_password_argon2(password: str) -> str:
        """Argon2idによるパスワードハッシュ（OWASP推奨）
 
        Argon2id は OWASP が最も推奨するパスワードハッシュアルゴリズム。
        メモリハード関数であり、GPU/ASIC攻撃に対してbcryptより強い耐性を持つ。
        """
        from argon2 import PasswordHasher
        # OWASP推奨パラメータ (2024):
        # memory_cost=65536 (64MB), time_cost=3, parallelism=4
        ph = PasswordHasher(
            memory_cost=65536,   # 64MB
            time_cost=3,         # 3回の反復
            parallelism=4,       # 4スレッド
            hash_len=32,         # 256bit出力
            salt_len=16,         # 128bitソルト
        )
        return ph.hash(password)
 
    @staticmethod
    def encrypt_sensitive_data(plaintext: str, master_key: bytes) -> str:
        """機密データの暗号化（AES-256-GCM）
 
        GCMモード（Galois/Counter Mode）を使用する理由:
        1. 認証付き暗号化（AEAD）: 暗号化と改竄検知を同時に実現
        2. 並列処理可能: CBCモードより高速
        3. IVの再利用が致命的: 12バイトのランダムノンスを毎回生成
        """
        # ノンスは毎回一意に生成（12バイトが推奨）
        nonce = os.urandom(12)
        key = AESGCM.generate_key(bit_length=256) if not master_key else master_key[:32]
 
        aesgcm = AESGCM(key)
        ciphertext = aesgcm.encrypt(nonce, plaintext.encode(), None)
 
        # ノンス + 暗号文を結合して返す
        return base64.b64encode(nonce + ciphertext).decode()
 
    @staticmethod
    def encrypt_with_kdf(plaintext: str, master_key: bytes) -> str:
        """KDFを使用した暗号化（鍵導出関数付き）"""
        # PBKDF2でマスターキーから暗号鍵を導出
        salt = os.urandom(16)
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            iterations=480000,  # OWASP推奨: 最低600,000（PBKDF2-SHA256の場合）
        )
        key = base64.urlsafe_b64encode(kdf.derive(master_key))
        f = Fernet(key)
        encrypted = f.encrypt(plaintext.encode())
        # ソルトと暗号文を結合して返す
        return base64.b64encode(salt + encrypted).decode()
 
    @staticmethod
    def constant_time_compare(a: str, b: str) -> bool:
        """定数時間文字列比較（タイミング攻撃対策）
 
        通常の == 比較は最初の不一致文字で即座に返すため、
        応答時間の差から正しい文字列を推測できてしまう。
        """
        return hmac.compare_digest(a.encode(), b.encode())
 
 
# パスワードハッシュアルゴリズムの比較表
"""
+------------------------------------------------------------------+
|  パスワードハッシュアルゴリズム比較                                    |
|------------------------------------------------------------------|
|  アルゴリズム | GPU耐性 | メモリ使用 | OWASP推奨 | 備考             |
|  ------------|---------|----------|----------|------------------|
|  MD5          | 最弱    | 極小     | 非推奨   | 衝突攻撃が容易     |
|  SHA-256      | 弱      | 極小     | 非推奨   | 高速すぎる         |
|  bcrypt       | 中      | 4KB固定  | 推奨     | 72バイト制限あり    |
|  scrypt       | 強      | 可変     | 推奨     | パラメータ設定が複雑 |
|  Argon2id     | 最強    | 可変     | 最推奨   | PHC勝者(2015)      |
+------------------------------------------------------------------+
"""

# コード例4: 包括的なインジェクション対策
import sqlite3
import subprocess
import shlex
import re
 
# ===== SQLインジェクション対策 =====
 
# 悪い例: 文字列連結によるSQL構築
def search_users_bad(username):
    query = f"SELECT * FROM users WHERE name = '{username}'"
    return db.execute(query)  # ' OR '1'='1 で全件取得可能
 
# 良い例1: パラメータ化クエリ
def search_users_good(username):
    query = "SELECT * FROM users WHERE name = ?"
    return db.execute(query, (username,))
 
# 良い例2: ORMの使用（SQLAlchemy）
from sqlalchemy import select
from models import User
 
def search_users_orm(session, username):
    stmt = select(User).where(User.name == username)
    return session.execute(stmt).scalars().all()
 
# 良い例3: ストアドプロシージャの使用
def search_users_stored_proc(username):
    return db.execute("CALL search_users(?)", (username,))
 
 
# ===== OSコマンドインジェクション対策 =====
 
# 悪い例: os.system / shell=True
def ping_host_bad(hostname):
    os.system(f"ping -c 1 {hostname}")  # ; rm -rf / が挿入可能
 
# 良い例: subprocessでシェルを介さない
def ping_host_good(hostname):
    # ホワイトリスト検証
    if not re.match(r'^[a-zA-Z0-9\.\-]+$', hostname):
        raise ValueError("Invalid hostname")
    # shell=False（デフォルト）で実行
    result = subprocess.run(
        ["ping", "-c", "1", hostname],
        capture_output=True, text=True, timeout=10
    )
    return result.stdout
 
 
# ===== テンプレートインジェクション (SSTI) 対策 =====
from jinja2 import Environment, select_autoescape, sandbox
 
# 悪い例: ユーザ入力をテンプレートとして評価
def render_bad(user_input):
    from jinja2 import Template
    return Template(user_input).render()  # {{7*7}} → 49, RCE可能
 
# 良い例: サンドボックス環境を使用
def render_good(template_name, context):
    env = sandbox.SandboxedEnvironment(
        autoescape=select_autoescape(['html', 'xml'])
    )
    template = env.get_template(template_name)
    return template.render(**context)
 
 
# ===== LDAPインジェクション対策 =====
import ldap3
 
# 悪い例: 文字列連結
def search_ldap_bad(username):
    filter_str = f"(uid={username})"  # *)(uid=*))(|(uid=* で全件取得
    conn.search("dc=example,dc=com", filter_str)
 
# 良い例: エスケープ関数の使用
def search_ldap_good(username):
    from ldap3.utils.conv import escape_filter_chars
    safe_username = escape_filter_chars(username)
    filter_str = f"(uid={safe_username})"
    conn.search("dc=example,dc=com", filter_str)

# コード例5: XSS対策の実装
from markupsafe import escape
from flask import Flask, Markup
 
app = Flask(__name__)
 
# 悪い例: ユーザ入力を直接HTMLに埋め込み
@app.route("/profile")
def profile_bad():
    name = request.args.get("name")
    return f"<h1>Welcome, {name}</h1>"  # XSS脆弱
 
# 良い例1: エスケープ
@app.route("/profile")
def profile_good():
    name = escape(request.args.get("name", ""))
    return f"<h1>Welcome, {name}</h1>"
 
# 良い例2: テンプレートエンジンの自動エスケープ
# templates/profile.html: <h1>Welcome, {{ name }}</h1>
# Jinja2はデフォルトで自動エスケープ
 
# CSP (Content Security Policy) の設定
@app.after_request
def set_csp(response):
    """XSSの影響を最小化するCSPヘッダー"""
    response.headers["Content-Security-Policy"] = (
        "default-src 'self'; "
        "script-src 'self' 'nonce-{nonce}'; "  # nonceベースの許可
        "style-src 'self' 'unsafe-inline'; "
        "img-src 'self' data: https:; "
        "frame-ancestors 'none'; "
        "base-uri 'self'; "
        "form-action 'self'"
    )
    return response

# コード例6: ビジネスロジックのセキュリティ設計
from datetime import datetime, timedelta
from collections import defaultdict
 
class SecurePasswordReset:
    """安全なパスワードリセットの設計
 
    安全でない設計の例:
    - リセットリンクが予測可能（連番ID）
    - リセットトークンに有効期限がない
    - レートリミットがない（列挙攻撃可能）
    - 「ユーザが存在しません」というエラーメッセージ（情報漏洩）
    """
 
    def __init__(self):
        self.reset_tokens = {}
        self.attempt_counts = defaultdict(list)
        self.TOKEN_EXPIRY = timedelta(minutes=15)  # 短い有効期限
        self.MAX_ATTEMPTS = 3  # 1時間あたり3回まで
 
    def request_reset(self, email: str) -> dict:
        # レートリミットチェック
        now = datetime.utcnow()
        recent = [t for t in self.attempt_counts[email]
                  if t > now - timedelta(hours=1)]
        self.attempt_counts[email] = recent
 
        if len(recent) >= self.MAX_ATTEMPTS:
            # 同じレスポンスを返す（情報漏洩防止）
            return {"message": "If the email exists, a reset link has been sent."}
 
        self.attempt_counts[email].append(now)
 
        # ユーザの存在有無に関わらず同じレスポンス
        user = find_user_by_email(email)
        if user:
            token = secrets.token_urlsafe(32)  # 256bit のランダムトークン
            self.reset_tokens[token] = {
                "user_id": user.id,
                "expires_at": now + self.TOKEN_EXPIRY,
                "used": False,
            }
            send_reset_email(email, token)
 
        # 攻撃者にユーザの存在を知らせない
        return {"message": "If the email exists, a reset link has been sent."}
 
    def verify_reset(self, token: str, new_password: str) -> bool:
        data = self.reset_tokens.get(token)
        if not data:
            return False
        if data["used"]:
            return False  # ワンタイム使用
        if datetime.utcnow() > data["expires_at"]:
            del self.reset_tokens[token]
            return False
 
        # トークンを使用済みにマーク
        data["used"] = True
        update_password(data["user_id"], new_password)
 
        # 他のセッションを無効化
        invalidate_all_sessions(data["user_id"])
 
        return True

# コード例7: セキュリティヘッダーの包括的な設定
from flask import Flask, Response
 
app = Flask(__name__)
 
# 本番環境ではデバッグモードを無効化
app.config["DEBUG"] = False
app.config["TESTING"] = False
 
# セッションの安全な設定
app.config["SESSION_COOKIE_SECURE"] = True       # HTTPS必須
app.config["SESSION_COOKIE_HTTPONLY"] = True      # JavaScript からアクセス不可
app.config["SESSION_COOKIE_SAMESITE"] = "Lax"    # CSRF対策
app.config["PERMANENT_SESSION_LIFETIME"] = 1800   # 30分でタイムアウト
 
@app.after_request
def set_security_headers(response: Response) -> Response:
    """全レスポンスにセキュリティヘッダーを付与する"""
    # XSSフィルタ（CSPに委任）
    response.headers["X-XSS-Protection"] = "0"
 
    # MIMEタイプスニッフィング防止
    response.headers["X-Content-Type-Options"] = "nosniff"
 
    # クリックジャッキング防止
    response.headers["X-Frame-Options"] = "DENY"
 
    # Content Security Policy（XSS緩和の最重要ヘッダー）
    response.headers["Content-Security-Policy"] = (
        "default-src 'self'; "
        "script-src 'self'; "
        "style-src 'self' 'unsafe-inline'; "
        "img-src 'self' data: https:; "
        "font-src 'self' https://fonts.gstatic.com; "
        "connect-src 'self' https://api.example.com; "
        "frame-ancestors 'none'; "
        "base-uri 'self'; "
        "form-action 'self'; "
        "upgrade-insecure-requests"
    )
 
    # HTTPS強制（HSTS）
    response.headers["Strict-Transport-Security"] = (
        "max-age=31536000; includeSubDomains; preload"
    )
 
    # リファラー制御
    response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
 
    # ブラウザ機能の制限
    response.headers["Permissions-Policy"] = (
        "camera=(), microphone=(), geolocation=(), "
        "payment=(), usb=(), magnetometer=()"
    )
 
    # キャッシュ制御（機密データ）
    if "api" in request.path:
        response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate"
        response.headers["Pragma"] = "no-cache"
 
    return response
 
 
# エラーハンドラ（情報漏洩防止）
@app.errorhandler(500)
def internal_error(error):
    """本番環境ではスタックトレースを返さない"""
    app.logger.error(f"Internal error: {error}", exc_info=True)
    return {"error": "Internal server error"}, 500
 
@app.errorhandler(404)
def not_found(error):
    return {"error": "Resource not found"}, 404

# コード例8: 依存関係のセキュリティチェック
 
# Python: pip-audit
pip install pip-audit
pip-audit
 
# Python: Safety
pip install safety
safety check
 
# Node.js: npm audit
npm audit
npm audit fix
 
# Go: govulncheck
go install golang.org/x/vuln/cmd/govulncheck@latest
govulncheck ./...
 
# Rust: cargo-audit
cargo install cargo-audit
cargo audit
 
# マルチ言語対応: Trivy
trivy fs --scanners vuln .

# GitHub Actions: Dependabot設定
# .github/dependabot.yml
version: 2
updates:
  - package-ecosystem: "pip"
    directory: "/"
    schedule:
      interval: "weekly"
    open-pull-requests-limit: 10
    labels:
      - "dependencies"
      - "security"
    reviewers:
      - "security-team"

# コード例9: セッション管理のベストプラクティス
import secrets
from datetime import datetime, timedelta
from flask import Flask, session, request
 
app = Flask(__name__)
app.secret_key = secrets.token_hex(32)  # 256bit
 
class SecureSessionManager:
    """安全なセッション管理"""
 
    def __init__(self):
        self.sessions = {}
        self.MAX_SESSIONS_PER_USER = 5
 
    def create_session(self, user_id: str, ip: str, user_agent: str) -> str:
        """セッション作成"""
        # 既存セッション数チェック
        user_sessions = [
            s for s in self.sessions.values()
            if s["user_id"] == user_id and not s["expired"]
        ]
        if len(user_sessions) >= self.MAX_SESSIONS_PER_USER:
            # 最も古いセッションを無効化
            oldest = min(user_sessions, key=lambda s: s["created_at"])
            oldest["expired"] = True
 
        session_id = secrets.token_urlsafe(32)  # 256bitのランダムID
        self.sessions[session_id] = {
            "user_id": user_id,
            "created_at": datetime.utcnow(),
            "last_activity": datetime.utcnow(),
            "ip": ip,
            "user_agent": user_agent,
            "expired": False,
        }
        return session_id
 
    def validate_session(self, session_id: str, ip: str) -> bool:
        """セッション検証"""
        data = self.sessions.get(session_id)
        if not data or data["expired"]:
            return False
 
        # 絶対タイムアウト: 作成から24時間
        if datetime.utcnow() - data["created_at"] > timedelta(hours=24):
            data["expired"] = True
            return False
 
        # アイドルタイムアウト: 最終操作から30分
        if datetime.utcnow() - data["last_activity"] > timedelta(minutes=30):
            data["expired"] = True
            return False
 
        # IP変更の検知（セッションハイジャック対策）
        if data["ip"] != ip:
            audit_log.warning(
                f"Session IP mismatch: session={session_id}, "
                f"original={data['ip']}, current={ip}"
            )
            # 厳密モード: セッション無効化
            # data["expired"] = True
            # return False
 
        data["last_activity"] = datetime.utcnow()
        return True
 
    def destroy_session(self, session_id: str):
        """セッション破棄（ログアウト時）"""
        if session_id in self.sessions:
            self.sessions[session_id]["expired"] = True
 
    def destroy_all_user_sessions(self, user_id: str):
        """全セッション破棄（パスワード変更時）"""
        for session_data in self.sessions.values():
            if session_data["user_id"] == user_id:
                session_data["expired"] = True

# コード例10: 安全なデシリアライゼーション
import json
import hmac
import hashlib
 
# 悪い例: pickle の直接使用（任意コード実行の危険）
import pickle
def load_bad(data):
    return pickle.loads(data)  # 絶対にNG! RCE可能
 
# 良い例: JSONによるシリアライゼーション + 署名検証
class SecureSerializer:
    def __init__(self, secret_key: bytes):
        self.secret_key = secret_key
 
    def serialize(self, data: dict) -> str:
        """データをシリアライズして署名"""
        payload = json.dumps(data, sort_keys=True)
        signature = hmac.new(
            self.secret_key, payload.encode(), hashlib.sha256
        ).hexdigest()
        return json.dumps({"payload": payload, "signature": signature})
 
    def deserialize(self, raw: str) -> dict:
        """署名を検証してデシリアライズ"""
        container = json.loads(raw)
        expected_sig = hmac.new(
            self.secret_key,
            container["payload"].encode(),
            hashlib.sha256
        ).hexdigest()
        if not hmac.compare_digest(expected_sig, container["signature"]):
            raise ValueError("Signature verification failed")
        return json.loads(container["payload"])

# コード例11: 構造化セキュリティログ
import logging
import json
from datetime import datetime
 
class SecurityLogger:
    """セキュリティイベント専用ロガー"""
 
    def __init__(self):
        self.logger = logging.getLogger("security")
        handler = logging.FileHandler("/var/log/app/security.json")
        handler.setFormatter(logging.Formatter("%(message)s"))
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)
 
    def log_event(self, event_type: str, details: dict):
        """構造化されたセキュリティイベントを記録"""
        event = {
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "event_type": event_type,
            "details": details,
            "source_ip": request.remote_addr if request else None,
            "user_agent": request.headers.get("User-Agent") if request else None,
        }
        self.logger.info(json.dumps(event))
 
    def log_auth_failure(self, username: str, reason: str):
        self.log_event("AUTH_FAILURE", {
            "username": username,
            "reason": reason,
        })
 
    def log_access_denied(self, user_id: str, resource: str):
        self.log_event("ACCESS_DENIED", {
            "user_id": user_id,
            "resource": resource,
        })
 
    def log_suspicious_activity(self, user_id: str, activity: str):
        self.log_event("SUSPICIOUS", {
            "user_id": user_id,
            "activity": activity,
            "severity": "HIGH",
        })
 
# ログすべきイベント一覧:
# - 認証の成功/失敗
# - アクセス制御の拒否
# - 入力検証の失敗
# - セッションの作成/破棄
# - 権限の変更
# - 管理操作（ユーザ作成、設定変更）
# - 高額取引/重要操作

# コード例12: SSRF対策の完全な実装
import ipaddress
import socket
from urllib.parse import urlparse
import re
 
class SSRFProtection:
    """SSRF攻撃を防止するURL検証
 
    SSRFの攻撃シナリオ:
    1. 内部メタデータAPIへのアクセス
       (AWS: http://169.254.169.254/latest/meta-data/)
    2. 内部サービスへのポートスキャン
    3. クラウドプロバイダの認証情報の窃取
    4. 内部ネットワークのファイル読み取り (file://)
    """
 
    BLOCKED_NETWORKS = [
        ipaddress.ip_network("10.0.0.0/8"),
        ipaddress.ip_network("172.16.0.0/12"),
        ipaddress.ip_network("192.168.0.0/16"),
        ipaddress.ip_network("127.0.0.0/8"),
        ipaddress.ip_network("169.254.0.0/16"),  # リンクローカル/メタデータ
        ipaddress.ip_network("100.64.0.0/10"),   # CGNATレンジ
        ipaddress.ip_network("::1/128"),          # IPv6ループバック
        ipaddress.ip_network("fc00::/7"),         # IPv6 ULA
        ipaddress.ip_network("fe80::/10"),        # IPv6 リンクローカル
    ]
 
    ALLOWED_SCHEMES = {"http", "https"}
    ALLOWED_PORTS = {80, 443, 8080, 8443}
 
    @classmethod
    def validate_url(cls, url: str) -> bool:
        """外部アクセスに使用するURLを検証する"""
        parsed = urlparse(url)
 
        # スキームチェック
        if parsed.scheme not in cls.ALLOWED_SCHEMES:
            return False
 
        # ポートチェック
        port = parsed.port or (443 if parsed.scheme == "https" else 80)
        if port not in cls.ALLOWED_PORTS:
            return False
 
        # ホスト名の検証
        hostname = parsed.hostname
        if not hostname:
            return False
 
        # DNS rebinding 対策: ドット区切りの数値IPを直接チェック
        try:
            ip = ipaddress.ip_address(hostname)
            return not cls._is_blocked(ip)
        except ValueError:
            pass  # ホスト名の場合はDNS解決が必要
 
        # ホスト解決とプライベートIP検出
        try:
            # 全てのアドレスを解決（CNAMEチェーン対策）
            addrinfos = socket.getaddrinfo(hostname, port)
            for family, _, _, _, sockaddr in addrinfos:
                ip = ipaddress.ip_address(sockaddr[0])
                if cls._is_blocked(ip):
                    return False
        except (socket.gaierror, ValueError):
            return False
 
        return True
 
    @classmethod
    def _is_blocked(cls, ip: ipaddress.IPv4Address) -> bool:
        for network in cls.BLOCKED_NETWORKS:
            if ip in network:
                return True
        return False
 
    @classmethod
    def safe_fetch(cls, url: str, timeout: int = 10) -> bytes:
        """安全な外部URLフェッチ"""
        if not cls.validate_url(url):
            raise ValueError(f"Blocked URL: {url}")
 
        import requests
        response = requests.get(
            url,
            timeout=timeout,
            allow_redirects=False,  # リダイレクトを手動で処理
            headers={"User-Agent": "MyApp/1.0"},
        )
 
        # リダイレクト先も検証
        if response.is_redirect:
            redirect_url = response.headers.get("Location")
            if redirect_url and cls.validate_url(redirect_url):
                return cls.safe_fetch(redirect_url, timeout)
            raise ValueError(f"Blocked redirect: {redirect_url}")
 
        return response.content

# GitHub Actions: 包括的なセキュリティテスト
name: Security Pipeline
on:
  pull_request:
    branches: [main]
 
jobs:
  sast:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Semgrep SAST
        uses: returntocorp/semgrep-action@v1
        with:
          config: "p/owasp-top-ten"
 
  dependency-check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: pip-audit
        run: |
          pip install pip-audit
          pip-audit -r requirements.txt
 
  secret-scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: TruffleHog
        uses: trufflesecurity/trufflehog@main
        with:
          extra_args: --only-verified

# 対策: アルゴリズムを明示的に指定し、ヘッダーの値を信用しない
import jwt
 
# NG: ヘッダーのalgを信用
payload = jwt.decode(token, key, algorithms=jwt.get_unverified_header(token)["alg"])
 
# OK: 許可するアルゴリズムを固定
payload = jwt.decode(token, public_key, algorithms=["RS256"])

# TOCTOU (Time of Check to Time of Use) 問題
# Step 1: 残高チェック（残高: 100円）
# Step 2: 引き出し処理（100円引き出し）
# 並行リクエスト: Step 1とStep 2の間に同じリクエストが通ると二重引き出し
 
# 対策: データベースレベルのロック
def withdraw(user_id, amount):
    with db.transaction():
        balance = db.execute(
            "SELECT balance FROM accounts WHERE user_id = ? FOR UPDATE",
            (user_id,)
        ).fetchone()
        if balance[0] >= amount:
            db.execute(
                "UPDATE accounts SET balance = balance - ? WHERE user_id = ?",
                (amount, user_id)
            )

# NG: スタックトレースを返す
@app.errorhandler(Exception)
def handle_error(e):
    return {"error": str(e), "traceback": traceback.format_exc()}, 500
 
# OK: 一般的なメッセージのみ
@app.errorhandler(Exception)
def handle_error(e):
    error_id = str(uuid.uuid4())
    app.logger.error(f"Error {error_id}: {e}", exc_info=True)
    return {"error": "Internal server error", "reference": error_id}, 500

// NG: フロントエンドのみでバリデーション
// → ブラウザの開発者ツールやcurlで簡単にバイパス可能
 
// OK: フロントエンドは UX 向上のため、サーバサイドが本質的な防御
// フロントエンド: 即時フィードバック用
// バックエンド: セキュリティ用（必須）

def login(username, password):
    query = f"""
        SELECT * FROM users
        WHERE username = '{username}'
        AND password = '{hashlib.md5(password.encode()).hexdigest()}'
    """
    result = db.execute(query)
    if result:
        return create_session(result[0])
    return None

from flask import Flask, request, jsonify
import sqlite3
 
app = Flask(__name__)
app.secret_key = "secret123"
 
@app.route("/api/user/<user_id>")
def get_user(user_id):
    conn = sqlite3.connect("app.db")
    cursor = conn.execute(f"SELECT * FROM users WHERE id = {user_id}")
    user = cursor.fetchone()
    conn.close()
    return jsonify({"user": user})
 
@app.route("/api/search")
def search():
    q = request.args.get("q")
    return f"<h1>Results for: {q}</h1>"
 
@app.route("/api/upload", methods=["POST"])
def upload():
    file = request.files["file"]
    file.save(f"/uploads/{file.filename}")
    return jsonify({"status": "uploaded"})