# コード例1: CSRF攻撃の例と防御
from flask import Flask, request, session, abort
import secrets
 
app = Flask(__name__)
app.secret_key = secrets.token_hex(32)
 
# 脆弱なコード: CSRFトークンなし
@app.route("/transfer", methods=["POST"])
def transfer_vulnerable():
    # Cookieだけで認証 -> CSRF攻撃可能
    to_account = request.form["to"]
    amount = request.form["amount"]
    execute_transfer(session["user_id"], to_account, amount)
    return "送金完了"
 
# 安全なコード: CSRFトークン検証付き
def generate_csrf_token() -> str:
    """セッションごとのCSRFトークンを生成"""
    if "csrf_token" not in session:
        session["csrf_token"] = secrets.token_hex(32)
    return session["csrf_token"]
 
def validate_csrf_token(token: str) -> bool:
    """CSRFトークンを検証"""
    expected = session.get("csrf_token")
    if not expected or not secrets.compare_digest(token, expected):
        return False
    return True
 
@app.route("/transfer", methods=["POST"])
def transfer_safe():
    # CSRFトークンの検証
    token = request.form.get("csrf_token", "")
    if not validate_csrf_token(token):
        abort(403, "Invalid CSRF token")
    to_account = request.form["to"]
    amount = request.form["amount"]
    execute_transfer(session["user_id"], to_account, amount)
    return "送金完了"

# コード例2: Double Submit Cookie パターン
import secrets
import hmac
 
class DoubleSubmitCSRF:
    """Double Submit Cookie方式のCSRF対策
 
    原理:
    - ランダムな値をCookieとフォームの両方に設定
    - 攻撃者はCookieは読み取れない（Same-Origin Policy）
    - したがってフォームに正しい値を設定できない
 
    注意: サブドメイン攻撃に対する耐性を高めるため
    HMAC署名を併用する
    """
 
    def __init__(self, secret_key: str):
        self.secret_key = secret_key
 
    def generate_token(self) -> tuple:
        """CookieとHTMLフォーム用のトークンペアを生成"""
        random_value = secrets.token_hex(32)
        # HMACで署名してCookie値を生成
        signed = hmac.new(
            self.secret_key.encode(),
            random_value.encode(),
            "sha256"
        ).hexdigest()
        return random_value, signed  # (cookie_value, form_value)
 
    def validate(self, cookie_value: str, form_value: str) -> bool:
        """Cookie値とフォーム値を照合"""
        expected_signed = hmac.new(
            self.secret_key.encode(),
            cookie_value.encode(),
            "sha256"
        ).hexdigest()
        return hmac.compare_digest(expected_signed, form_value)
 
csrf = DoubleSubmitCSRF("my-secret-key")
cookie_val, form_val = csrf.generate_token()
# cookie_val -> Set-Cookie: csrf=<random_value>
# form_val   -> <input type="hidden" name="csrf_token" value="<signed>">

# コード例3: 署名付きDouble Submit Cookie（タイムスタンプ含む）
import time
import hmac
import hashlib
import secrets
import json
import base64
 
class SignedDoubleSubmitCSRF:
    """改良版 Double Submit Cookie
 
    従来のDouble Submitの弱点を補完:
    - タイムスタンプで有効期限を設定
    - セッションIDとの紐付けでセッション固定攻撃を防止
    - HMAC-SHA256 で改ざん検知
    """
 
    def __init__(self, secret_key: str, max_age: int = 3600):
        self.secret_key = secret_key.encode()
        self.max_age = max_age  # トークンの有効期限（秒）
 
    def generate_token(self, session_id: str) -> str:
        """タイムスタンプ付きCSRFトークンを生成"""
        payload = {
            "sid": session_id,
            "ts": int(time.time()),
            "nonce": secrets.token_hex(16),
        }
        payload_b64 = base64.urlsafe_b64encode(
            json.dumps(payload).encode()
        ).decode()
 
        signature = hmac.new(
            self.secret_key,
            payload_b64.encode(),
            hashlib.sha256,
        ).hexdigest()
 
        return f"{payload_b64}.{signature}"
 
    def validate_token(self, token: str, session_id: str) -> bool:
        """CSRFトークンを検証"""
        try:
            parts = token.split(".")
            if len(parts) != 2:
                return False
            payload_b64, signature = parts
 
            # 署名検証
            expected_sig = hmac.new(
                self.secret_key,
                payload_b64.encode(),
                hashlib.sha256,
            ).hexdigest()
            if not hmac.compare_digest(signature, expected_sig):
                return False
 
            # ペイロード検証
            payload = json.loads(
                base64.urlsafe_b64decode(payload_b64)
            )
 
            # セッションID一致確認
            if payload.get("sid") != session_id:
                return False
 
            # 有効期限確認
            if int(time.time()) - payload.get("ts", 0) > self.max_age:
                return False
 
            return True
        except Exception:
            return False
 
# 使用例
csrf = SignedDoubleSubmitCSRF("super-secret-key", max_age=3600)
token = csrf.generate_token(session_id="sess_abc123")
is_valid = csrf.validate_token(token, session_id="sess_abc123")

# コード例4: SameSite Cookie の設定（各値の詳細）
from flask import Flask, make_response
 
app = Flask(__name__)
 
@app.route("/login", methods=["POST"])
def login():
    response = make_response("ログイン成功")
 
    # 推奨設定: SameSite=Lax
    response.set_cookie(
        "session_id",
        value=generate_session_id(),
        httponly=True,     # JavaScriptからアクセス不可
        secure=True,       # HTTPS通信でのみ送信
        samesite="Lax",    # クロスサイトのGETは許可、POSTは拒否
        max_age=3600,
        path="/",
    )
    return response
 
@app.route("/set-strict-cookie")
def set_strict():
    """最も厳格な設定（銀行サイト等向け）"""
    response = make_response("OK")
    response.set_cookie(
        "bank_session",
        value=generate_session_id(),
        httponly=True,
        secure=True,
        samesite="Strict",  # クロスサイトからは一切送信しない
        max_age=1800,
        path="/",
        domain=".bank.example.com",  # サブドメインにも適用
    )
    return response
 
@app.route("/set-none-cookie")
def set_none():
    """サードパーティCookie（OAuth等で必要な場合）"""
    response = make_response("OK")
    response.set_cookie(
        "third_party",
        value=generate_token(),
        httponly=True,
        secure=True,       # SameSite=NoneにはSecure必須
        samesite="None",   # クロスサイトでも送信
        max_age=3600,
        path="/",
    )
    return response

# コード例5: Origin/Referer ヘッダー検証の実装
from flask import Flask, request, abort
from urllib.parse import urlparse
 
app = Flask(__name__)
 
ALLOWED_ORIGINS = {
    "https://myapp.example.com",
    "https://www.myapp.example.com",
}
 
class OriginVerifier:
    """Origin/Referer ヘッダーによるCSRF防御（補助的対策）
 
    注意: Origin/Referer ヘッダーは以下の場合に欠落する可能性がある
    - プライバシー拡張機能が削除
    - Referrer-Policy: no-referrer が設定されている
    - 古いブラウザ
    - プロキシが除去
    したがって、CSRFトークンとの併用が推奨される
    """
 
    def __init__(self, allowed_origins: set):
        self.allowed_origins = allowed_origins
 
    def verify(self) -> bool:
        """Origin または Referer ヘッダーを検証"""
        origin = request.headers.get("Origin")
 
        if origin:
            return origin in self.allowed_origins
 
        # Origin がない場合は Referer をフォールバックとして使用
        referer = request.headers.get("Referer")
        if referer:
            parsed = urlparse(referer)
            referer_origin = f"{parsed.scheme}://{parsed.netloc}"
            return referer_origin in self.allowed_origins
 
        # 両方ない場合の判断
        # 厳格モード: 拒否（推奨）
        # 緩和モード: 許可（互換性重視だがリスクあり）
        return False  # 厳格モード
 
verifier = OriginVerifier(ALLOWED_ORIGINS)
 
@app.before_request
def check_origin():
    """状態変更リクエストのOriginを検証"""
    if request.method in ("GET", "HEAD", "OPTIONS"):
        return None
    if not verifier.verify():
        abort(403, "Origin verification failed")

# コード例6: SPA向けCSRF対策（FastAPI + React）
from fastapi import FastAPI, Request, Response, HTTPException
from fastapi.middleware.cors import CORSMiddleware
import secrets
 
app = FastAPI()
 
# CORS設定（CSRFと密接に関連）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.example.com"],  # 明示的に指定
    allow_credentials=True,   # Cookieを許可
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["X-CSRF-Token"],  # カスタムヘッダーを許可
)
 
@app.get("/api/csrf-token")
async def get_csrf_token(response: Response):
    """CSRFトークンを発行するエンドポイント"""
    token = secrets.token_hex(32)
    # Double Submit Cookie パターン
    response.set_cookie(
        key="csrf_token",
        value=token,
        httponly=False,  # JSから読み取り可能にする（フォーム送信用）
        secure=True,
        samesite="strict",
        max_age=3600,
    )
    return {"csrf_token": token}
 
@app.middleware("http")
async def csrf_middleware(request: Request, call_next):
    """CSRF検証ミドルウェア"""
    if request.method in ("GET", "HEAD", "OPTIONS"):
        return await call_next(request)
 
    # カスタムヘッダーの存在確認
    # Simple Request ではカスタムヘッダーを送信できないため
    # これ自体がCSRF対策になる（CORSプリフライト必須）
    csrf_header = request.headers.get("X-CSRF-Token")
    csrf_cookie = request.cookies.get("csrf_token")
 
    if not csrf_header or not csrf_cookie:
        raise HTTPException(status_code=403, detail="CSRF token missing")
 
    if not secrets.compare_digest(csrf_header, csrf_cookie):
        raise HTTPException(status_code=403, detail="CSRF token mismatch")
 
    return await call_next(request)

// コード例7: React側のCSRFトークン管理
// csrf.js - CSRFトークンユーティリティ
 
class CSRFTokenManager {
  constructor() {
    this.token = null;
  }
 
  async fetchToken() {
    const response = await fetch('/api/csrf-token', {
      credentials: 'include',  // Cookieを含める
    });
    const data = await response.json();
    this.token = data.csrf_token;
    return this.token;
  }
 
  getToken() {
    if (!this.token) {
      // Cookieからトークンを読み取る（フォールバック）
      const cookies = document.cookie.split(';');
      for (const cookie of cookies) {
        const [name, value] = cookie.trim().split('=');
        if (name === 'csrf_token') {
          this.token = value;
          break;
        }
      }
    }
    return this.token;
  }
 
  /**
   * CSRF保護付きfetchラッパー
   */
  async secureFetch(url, options = {}) {
    if (!this.token) {
      await this.fetchToken();
    }
 
    const headers = {
      ...options.headers,
      'X-CSRF-Token': this.token,
      'Content-Type': 'application/json',
    };
 
    return fetch(url, {
      ...options,
      headers,
      credentials: 'include',
    });
  }
}
 
// 使用例
const csrfManager = new CSRFTokenManager();
 
async function transferFunds(toAccount, amount) {
  const response = await csrfManager.secureFetch('/api/transfer', {
    method: 'POST',
    body: JSON.stringify({ to: toAccount, amount }),
  });
 
  if (!response.ok) {
    throw new Error(`Transfer failed: ${response.status}`);
  }
  return response.json();
}

# コード例8: クリックジャッキング対策
from flask import Flask
 
app = Flask(__name__)
 
@app.after_request
def anti_clickjacking(response):
    """クリックジャッキング対策ヘッダーの設定"""
    # 方法1: X-Frame-Options（レガシーだが広くサポート）
    response.headers["X-Frame-Options"] = "DENY"
    # DENY: すべてのiframe埋め込みを拒否
    # SAMEORIGIN: 同一オリジンからのみ許可
    # ALLOW-FROM uri: 特定のオリジンからのみ許可（非推奨）
 
    # 方法2: CSP frame-ancestors（推奨、より柔軟）
    response.headers["Content-Security-Policy"] = (
        "frame-ancestors 'none'"
        # 'none': すべてのiframe埋め込みを拒否
        # 'self': 同一オリジンからのみ許可
        # https://trusted.com: 特定のオリジンからのみ許可
    )
    return response

// コード例9: JavaScriptによるフレーム脱出（補助的対策）
 
// 基本的なframe busting
if (window.top !== window.self) {
  window.top.location = window.self.location;
}
 
// より堅牢な方法
(function() {
  // sandbox属性による回避を防ぐ
  if (self === top) {
    // iframeに埋め込まれていない -> 正常
    document.documentElement.style.display = "block";
  } else {
    // iframeに埋め込まれている -> 脱出試行
    try {
      top.location = self.location;
    } catch (e) {
      // クロスオリジンでブロックされた場合
      document.body.innerHTML =
        "<h1>このページはiframe内では表示できません</h1>";
    }
  }
})();
 
// 最も堅牢な方法: CSSで初期非表示 + JSで表示
// HTML: <style>html { display: none; }</style>
(function() {
  if (self === top) {
    document.documentElement.style.display = "block";
  } else {
    // iframeに埋め込まれている場合は何も表示しない
    top.location = self.location;
  }
})();

# コード例10: CSRF + クリックジャッキング統合防御ミドルウェア
from flask import Flask, request, session, abort
from functools import wraps
import secrets
import time
import hashlib
import hmac
 
app = Flask(__name__)
 
class SecurityMiddleware:
    """CSRF・クリックジャッキング統合防御ミドルウェア"""
 
    SAFE_METHODS = {"GET", "HEAD", "OPTIONS"}
 
    def __init__(self, app, secret_key: str, csrf_header: str = "X-CSRF-Token"):
        self.app = app
        self.secret_key = secret_key
        self.csrf_header = csrf_header
        app.before_request(self._check_csrf)
        app.after_request(self._set_security_headers)
 
    def _check_csrf(self):
        """状態変更リクエストのCSRFトークンを検証"""
        if request.method in self.SAFE_METHODS:
            return None
 
        # AJAX リクエストの判定（補助的対策）
        if request.headers.get("X-Requested-With") == "XMLHttpRequest":
            # AJAX リクエストはCORSプリフライトが必要なため
            # 基本的にCSRF安全だが、追加検証も行う
            pass
 
        # Originヘッダーの検証（補助的対策）
        origin = request.headers.get("Origin")
        if origin and not self._is_same_origin(origin):
            abort(403, "Cross-origin request blocked")
 
        # CSRFトークンの検証
        token = (request.form.get("csrf_token") or
                 request.headers.get(self.csrf_header))
        if not token or not self._validate_token(token):
            abort(403, "Invalid CSRF token")
 
    def _set_security_headers(self, response):
        """セキュリティヘッダーの設定"""
        # クリックジャッキング対策
        response.headers["X-Frame-Options"] = "DENY"
        response.headers["Content-Security-Policy"] = (
            "frame-ancestors 'none'"
        )
 
        # その他のセキュリティヘッダー
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-XSS-Protection"] = "0"  # モダンブラウザでは無効化推奨
        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
 
        # CSRF Cookie の設定（Double Submit用）
        if "csrf_token" in session:
            response.set_cookie(
                "csrf_token",
                value=session["csrf_token"],
                httponly=False,  # JSから読み取り可能
                secure=True,
                samesite="Strict",
            )
 
        return response
 
    def _is_same_origin(self, origin: str) -> bool:
        return origin in ("https://myapp.example.com",)
 
    def _validate_token(self, token: str) -> bool:
        expected = session.get("csrf_token", "")
        return secrets.compare_digest(token, expected)
 
# ミドルウェアの適用
SecurityMiddleware(app, "my-secret-key")

# コード例11: Django の組み込みCSRF対策の活用
 
# settings.py
MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',  # CSRF ミドルウェア
    # ...
]
 
# CSRF設定
CSRF_COOKIE_SECURE = True        # HTTPS のみ
CSRF_COOKIE_HTTPONLY = False      # JSから読み取り可能（AJAX用）
CSRF_COOKIE_SAMESITE = "Lax"     # SameSite=Lax
CSRF_USE_SESSIONS = False        # CookieにCSRFトークンを保存（True でセッション）
CSRF_TRUSTED_ORIGINS = [         # 信頼するオリジン
    "https://myapp.example.com",
]
 
# セキュリティヘッダー設定
X_FRAME_OPTIONS = "DENY"
SECURE_CONTENT_TYPE_NOSNIFF = True
SECURE_BROWSER_XSS_FILTER = False  # X-XSS-Protection は無効化
 
# テンプレートでの使用
# <form method="post">
#   {% csrf_token %}
#   ...
# </form>
 
# AJAX での使用（Django公式パターン）
# views.py
from django.views.decorators.csrf import ensure_csrf_cookie
from django.http import JsonResponse
 
@ensure_csrf_cookie
def get_csrf(request):
    """CSRFクッキーを設定するエンドポイント"""
    return JsonResponse({"status": "ok"})

# コード例12: 正しいCORS設定
from flask import Flask
from flask_cors import CORS
 
app = Flask(__name__)
 
# NG: ワイルドカードオリジン + credentials
# CORS(app, origins="*", supports_credentials=True)  # 危険!
 
# OK: 明示的なオリジン指定
CORS(app,
    origins=["https://myapp.example.com"],
    supports_credentials=True,
    allow_headers=["Content-Type", "X-CSRF-Token"],
    methods=["GET", "POST", "PUT", "DELETE"],
    max_age=3600,
)

# コード例13: CSRF脆弱性のテスト
import requests
 
class CSRFTester:
    """CSRF脆弱性の自動テストツール"""
 
    def __init__(self, base_url: str):
        self.base_url = base_url
        self.session = requests.Session()
 
    def test_missing_token(self, endpoint: str, method: str = "POST",
                            data: dict = None) -> dict:
        """CSRFトークンなしでリクエストを送信"""
        response = getattr(self.session, method.lower())(
            f"{self.base_url}{endpoint}",
            data=data or {},
        )
        return {
            "test": "missing_token",
            "status_code": response.status_code,
            "vulnerable": response.status_code != 403,
        }
 
    def test_invalid_token(self, endpoint: str, method: str = "POST",
                            data: dict = None) -> dict:
        """無効なCSRFトークンでリクエストを送信"""
        test_data = data or {}
        test_data["csrf_token"] = "invalid_token_12345"
        response = getattr(self.session, method.lower())(
            f"{self.base_url}{endpoint}",
            data=test_data,
        )
        return {
            "test": "invalid_token",
            "status_code": response.status_code,
            "vulnerable": response.status_code != 403,
        }
 
    def test_cross_origin(self, endpoint: str, method: str = "POST",
                           data: dict = None) -> dict:
        """異なるOriginヘッダーでリクエストを送信"""
        headers = {"Origin": "https://evil.example.com"}
        response = getattr(self.session, method.lower())(
            f"{self.base_url}{endpoint}",
            data=data or {},
            headers=headers,
        )
        return {
            "test": "cross_origin",
            "status_code": response.status_code,
            "vulnerable": response.status_code != 403,
        }
 
    def test_clickjacking(self) -> dict:
        """X-Frame-Options / CSP frame-ancestors の確認"""
        response = self.session.get(self.base_url)
        xfo = response.headers.get("X-Frame-Options", "")
        csp = response.headers.get("Content-Security-Policy", "")
 
        vulnerable = True
        if xfo.upper() in ("DENY", "SAMEORIGIN"):
            vulnerable = False
        if "frame-ancestors" in csp:
            vulnerable = False
 
        return {
            "test": "clickjacking",
            "x_frame_options": xfo,
            "csp_frame_ancestors": csp,
            "vulnerable": vulnerable,
        }
 
# 使用例
tester = CSRFTester("https://myapp.example.com")
results = [
    tester.test_missing_token("/transfer", data={"to": "test", "amount": "1"}),
    tester.test_invalid_token("/transfer", data={"to": "test", "amount": "1"}),
    tester.test_cross_origin("/transfer", data={"to": "test", "amount": "1"}),
    tester.test_clickjacking(),
]
for r in results:
    status = "VULNERABLE" if r["vulnerable"] else "SAFE"
    print(f"[{status}] {r['test']}: {r.get('status_code', 'N/A')}")

# NG: GETで状態変更
@app.route("/delete-account")  # GETでアカウント削除
def delete_account():
    db.delete_user(session["user_id"])
    return "削除完了"
# 攻撃: <img src="/delete-account"> をメール等に埋め込むだけ
 
# OK: POSTで状態変更 + CSRF対策
@app.route("/delete-account", methods=["POST"])
def delete_account():
    validate_csrf_token(request.form["csrf_token"])
    db.delete_user(session["user_id"])
    return "削除完了"

# NG: 固定トークン
CSRF_TOKEN = "fixed-token-for-all-users"  # 全ユーザー共通!
 
# NG: 有効期限なし
session["csrf_token"] = secrets.token_hex(32)
# → セッションが続く限り同じトークンが使われる
 
# OK: セッションごとにユニーク + 有効期限付き
def generate_csrf_token():
    token_data = {
        "value": secrets.token_hex(32),
        "created_at": time.time(),
    }
    session["csrf_token"] = token_data
    return token_data["value"]

# NG: ワイルドカードオリジン + credentials
@app.after_request
def add_cors(response):
    response.headers["Access-Control-Allow-Origin"] = "*"
    response.headers["Access-Control-Allow-Credentials"] = "true"
    # → ブラウザはこの組み合わせを拒否するが、
    #   Originをエコーバックするサーバーは攻撃を許す
 
# NG: Originヘッダーをそのまま反映
@app.after_request
def add_cors_echo(response):
    origin = request.headers.get("Origin", "*")
    response.headers["Access-Control-Allow-Origin"] = origin  # 危険!
    response.headers["Access-Control-Allow-Credentials"] = "true"
 
# OK: ホワイトリストで検証
ALLOWED_ORIGINS = {"https://myapp.example.com", "https://admin.example.com"}
 
@app.after_request
def add_cors_safe(response):
    origin = request.headers.get("Origin")
    if origin in ALLOWED_ORIGINS:
        response.headers["Access-Control-Allow-Origin"] = origin
        response.headers["Access-Control-Allow-Credentials"] = "true"
    return response