# コード例1: Argon2id によるパスワードハッシュ管理
import hashlib
import requests
from argon2 import PasswordHasher
from argon2.exceptions import VerifyMismatchError
 
 
class PasswordManager:
    """安全なパスワード管理クラス
 
    Argon2id を使用し、パスワードのハッシュ化・検証・ポリシーチェックを提供する。
    Argon2id は Argon2i（サイドチャネル耐性）と Argon2d（GPU耐性）を
    組み合わせたハイブリッドモードであり、OWASP が第一推奨とするアルゴリズムである。
    """
 
    def __init__(self):
        # OWASP 推奨パラメータ（2024年時点）
        # - time_cost: 反復回数（大きいほど低速=安全）
        # - memory_cost: メモリ使用量 KB（大きいほど GPU/ASIC 耐性向上）
        # - parallelism: 並列度（CPU コア数に合わせる）
        self.ph = PasswordHasher(
            time_cost=3,
            memory_cost=65536,  # 64 MB
            parallelism=4,
        )
 
    def hash_password(self, password: str) -> str:
        """パスワードを Argon2id でハッシュ化
 
        ソルトは自動生成・ハッシュ文字列に埋め込まれる。
        """
        return self.ph.hash(password)
 
    def verify_password(self, password: str, hashed: str) -> bool:
        """パスワードを検証（タイミング攻撃耐性あり）"""
        try:
            return self.ph.verify(hashed, password)
        except VerifyMismatchError:
            return False
 
    def needs_rehash(self, hashed: str) -> bool:
        """パラメータが古い場合に再ハッシュが必要か判定
 
        アルゴリズムのパラメータを強化した場合、
        ログイン成功時に自動的に再ハッシュを行うための判定。
        """
        return self.ph.check_needs_rehash(hashed)
 
    def validate_policy(self, password: str) -> list:
        """NIST SP 800-63B 準拠のパスワードポリシー検証
 
        NISTガイドラインのポイント:
        - 最小8文字（推奨12文字以上）
        - 最大64文字以上を許容
        - 文字種の強制は不要（ユーザビリティ低下のため）
        - 漏洩パスワードリストとの照合は必須
        """
        errors = []
        if len(password) < 12:
            errors.append("12文字以上必要です")
        if len(password) > 128:
            errors.append("128文字以下にしてください")
        if not any(c.isupper() for c in password):
            errors.append("大文字を1文字以上含めてください")
        if not any(c.islower() for c in password):
            errors.append("小文字を1文字以上含めてください")
        if not any(c.isdigit() for c in password):
            errors.append("数字を1文字以上含めてください")
        # 漏洩パスワードチェック（Have I Been Pwned API）
        if self._is_breached(password):
            errors.append("このパスワードは過去のデータ漏洩に含まれています")
        return errors
 
    def _is_breached(self, password: str) -> bool:
        """HIBPのk-匿名性APIでパスワード漏洩チェック
 
        パスワードの SHA-1 ハッシュの先頭5文字のみを送信するため、
        パスワード自体がネットワーク上に流れることはない。
        """
        sha1 = hashlib.sha1(password.encode()).hexdigest().upper()
        prefix, suffix = sha1[:5], sha1[5:]
        try:
            resp = requests.get(
                f"https://api.pwnedpasswords.com/range/{prefix}",
                timeout=5,
            )
            return suffix in resp.text
        except requests.RequestException:
            # API障害時は安全側に倒す（ブロックしない）
            return False
 
 
# 使用例
pm = PasswordManager()
 
# ポリシーチェック
errors = pm.validate_policy("MyS3cur3P@ssw0rd!")
if errors:
    print(f"ポリシー違反: {errors}")
else:
    # ハッシュ化と保存
    hashed = pm.hash_password("MyS3cur3P@ssw0rd!")
    print(f"ハッシュ値: {hashed}")
    # 例: $argon2id$v=19$m=65536,t=3,p=4$salt$hash
 
    # 検証
    print(pm.verify_password("MyS3cur3P@ssw0rd!", hashed))  # True
    print(pm.verify_password("wrong", hashed))               # False
 
    # ログイン時の再ハッシュチェック
    if pm.needs_rehash(hashed):
        new_hash = pm.hash_password("MyS3cur3P@ssw0rd!")
        print("パラメータが更新されたため再ハッシュしました")

# コード例2: bcrypt によるパスワードハッシュ（レガシーシステム向け）
import bcrypt
 
 
class BcryptPasswordManager:
    """bcrypt によるパスワード管理
 
    Argon2id が使えない環境（古い Python/ライブラリ制約）向け。
    bcrypt は 1999 年から使用されている実績のあるアルゴリズム。
 
    WHY bcrypt の cost factor が重要か:
    - cost=10: 約 100ms/hash（2015 年の標準）
    - cost=12: 約 400ms/hash（2024 年の推奨）
    - cost=14: 約 1.6s/hash（高セキュリティ環境）
    ハードウェアの進化に合わせて cost を上げることで安全性を維持する。
    """
 
    DEFAULT_ROUNDS = 12  # 2024年推奨値
 
    def hash_password(self, password: str) -> str:
        """bcrypt ハッシュ化（ソルト自動生成）"""
        # bcrypt は内部で 72 バイトに切り詰めるため注意
        if len(password.encode("utf-8")) > 72:
            # 長いパスワードは事前に SHA-256 でハッシュ
            import hashlib
            password = hashlib.sha256(password.encode()).hexdigest()
 
        salt = bcrypt.gensalt(rounds=self.DEFAULT_ROUNDS)
        return bcrypt.hashpw(password.encode("utf-8"), salt).decode("utf-8")
 
    def verify_password(self, password: str, hashed: str) -> bool:
        """パスワードを検証"""
        if len(password.encode("utf-8")) > 72:
            import hashlib
            password = hashlib.sha256(password.encode()).hexdigest()
 
        return bcrypt.checkpw(
            password.encode("utf-8"),
            hashed.encode("utf-8"),
        )
 
 
# 使用例
bpm = BcryptPasswordManager()
hashed = bpm.hash_password("SecurePass123!")
print(bpm.verify_password("SecurePass123!", hashed))  # True

# コード例3: セキュアなセッション管理
import secrets
import time
import hashlib
from typing import Optional, Dict
 
 
class SecureSessionManager:
    """安全なセッション管理
 
    WHY 各設計判断が必要か:
    - SESSION_ID_LENGTH=32: 256ビットのエントロピーで推測攻撃を防止
      (2^256 の組み合わせ → ブルートフォース不可能)
    - SESSION_TIMEOUT=3600: セッション固定攻撃の時間窓を制限
    - IDLE_TIMEOUT=1800: 離席時のセッション悪用を防止
    - UA ハッシュ検証: セッションハイジャックの検知
    """
 
    SESSION_ID_LENGTH = 32   # 32バイト = 256ビットのエントロピー
    SESSION_TIMEOUT = 3600   # 絶対タイムアウト: 1時間
    IDLE_TIMEOUT = 1800      # アイドルタイムアウト: 30分
    MAX_SESSIONS_PER_USER = 5  # 同時セッション数制限
 
    def __init__(self):
        # 本番環境では Redis 等の外部ストアを使用
        self.sessions: Dict[str, dict] = {}
 
    def create_session(self, user_id: str, ip: str,
                       user_agent: str) -> str:
        """認証成功後にセッションを作成
 
        Returns:
            暗号学的に安全なセッションID文字列
        """
        # 同時セッション数の制限
        user_sessions = [
            sid for sid, data in self.sessions.items()
            if data["user_id"] == user_id
        ]
        if len(user_sessions) >= self.MAX_SESSIONS_PER_USER:
            # 最も古いセッションを削除
            oldest = min(user_sessions,
                        key=lambda s: self.sessions[s]["created_at"])
            self.destroy_session(oldest)
 
        session_id = secrets.token_hex(self.SESSION_ID_LENGTH)
        now = time.time()
        self.sessions[session_id] = {
            "user_id": user_id,
            "created_at": now,
            "last_activity": now,
            "ip": ip,
            "user_agent_hash": hashlib.sha256(
                user_agent.encode()
            ).hexdigest(),
        }
        return session_id
 
    def validate_session(self, session_id: str, ip: str,
                         user_agent: str) -> Optional[str]:
        """セッションの検証
 
        検証項目:
        1. セッションの存在確認
        2. 絶対タイムアウト（作成から一定時間経過）
        3. アイドルタイムアウト（最終アクティビティから一定時間経過）
        4. User-Agent の一貫性（セッションハイジャック検知）
        """
        session = self.sessions.get(session_id)
        if not session:
            return None
 
        now = time.time()
 
        # 絶対タイムアウトチェック
        if now - session["created_at"] > self.SESSION_TIMEOUT:
            self.destroy_session(session_id)
            return None
 
        # アイドルタイムアウトチェック
        if now - session["last_activity"] > self.IDLE_TIMEOUT:
            self.destroy_session(session_id)
            return None
 
        # User-Agent の変更を検出（セッションハイジャックの兆候）
        ua_hash = hashlib.sha256(user_agent.encode()).hexdigest()
        if session["user_agent_hash"] != ua_hash:
            self.destroy_session(session_id)
            return None
 
        # 最終アクティビティ更新
        session["last_activity"] = now
        return session["user_id"]
 
    def regenerate_session(self, old_session_id: str) -> Optional[str]:
        """セッションIDの再生成（権限昇格時に必須）
 
        WHY: ログイン前に攻撃者が設定したセッションIDが
        ログイン後も使い回されるとセッション固定攻撃が成立する。
        ログイン成功・権限変更時に必ずセッションIDを再生成する。
        """
        session = self.sessions.get(old_session_id)
        if not session:
            return None
        # 旧セッションを削除
        del self.sessions[old_session_id]
        # 新セッションIDで同じデータを登録
        new_session_id = secrets.token_hex(self.SESSION_ID_LENGTH)
        session["last_activity"] = time.time()
        self.sessions[new_session_id] = session
        return new_session_id
 
    def destroy_session(self, session_id: str) -> None:
        """セッションの破棄"""
        self.sessions.pop(session_id, None)
 
    def destroy_all_user_sessions(self, user_id: str) -> int:
        """特定ユーザーの全セッションを破棄
 
        用途: パスワード変更時、アカウント侵害検出時
        """
        to_delete = [
            sid for sid, data in self.sessions.items()
            if data["user_id"] == user_id
        ]
        for sid in to_delete:
            del self.sessions[sid]
        return len(to_delete)
 
    def get_active_sessions(self, user_id: str) -> list:
        """ユーザーのアクティブセッション一覧を取得
 
        ユーザーが自分の全セッションを確認・管理できるようにする。
        """
        now = time.time()
        return [
            {
                "session_id": sid[:8] + "...",  # 先頭のみ表示
                "created_at": data["created_at"],
                "last_activity": data["last_activity"],
                "ip": data["ip"],
            }
            for sid, data in self.sessions.items()
            if data["user_id"] == user_id
            and now - data["created_at"] <= self.SESSION_TIMEOUT
        ]
 
 
# 使用例
sm = SecureSessionManager()
 
# ログイン成功時
session_id = sm.create_session(
    "user123", "192.168.1.100", "Mozilla/5.0..."
)
print(f"セッション作成: {session_id[:16]}...")
 
# セッションID再生成（権限昇格時）
new_session_id = sm.regenerate_session(session_id)
print(f"セッション再生成: {new_session_id[:16]}...")
 
# セッション検証
user_id = sm.validate_session(
    new_session_id, "192.168.1.100", "Mozilla/5.0..."
)
print(f"認証ユーザー: {user_id}")  # user123
 
# ログアウト時
sm.destroy_session(new_session_id)

# コード例4: ブルートフォース対策の多層実装
import time
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict
 
 
@dataclass
class LoginAttempt:
    """ログイン試行の追跡データ"""
    count: int = 0
    first_attempt: float = 0
    last_attempt: float = 0
    locked_until: float = 0
 
 
class BruteForceProtection:
    """ブルートフォース攻撃対策
 
    WHY プログレッシブ遅延とロックアウトを組み合わせるか:
    - 遅延だけ: 攻撃を遅くするが止められない
    - ロックアウトだけ: 正規ユーザーを DoS できてしまう
    - 組み合わせ: 攻撃を遅くし、最終的に停止させつつ、
      自動解除で正規ユーザーへの影響を最小化する
    """
 
    MAX_ATTEMPTS = 5
    LOCKOUT_DURATION = 1800  # 30分
    WINDOW = 300             # 5分間のウィンドウ
    PROGRESSIVE_DELAYS = [0, 1, 2, 4, 8, 15]  # 秒
 
    def __init__(self):
        self.account_attempts: Dict[str, LoginAttempt] = defaultdict(
            LoginAttempt
        )
        self.ip_attempts: Dict[str, LoginAttempt] = defaultdict(LoginAttempt)
 
    def check_and_record(self, username: str, ip: str) -> dict:
        """ログイン試行を確認・記録する
 
        アカウント単位とIP単位の両方でチェックする。
        Credential Stuffing 攻撃（同一IPから多数のアカウントへの試行）
        を検知するため、IP単位のチェックも重要。
        """
        # アカウント単位のチェック
        account_result = self._check_identifier(
            self.account_attempts, username
        )
        if not account_result["allowed"]:
            return account_result
 
        # IP単位のチェック（より緩いレート）
        ip_result = self._check_identifier(
            self.ip_attempts, ip, max_attempts=20
        )
        if not ip_result["allowed"]:
            ip_result["reason"] = "ip_rate_limited"
            return ip_result
 
        return account_result
 
    def _check_identifier(self, attempts_store: dict,
                          identifier: str,
                          max_attempts: int = None) -> dict:
        """識別子に対するチェックと記録"""
        if max_attempts is None:
            max_attempts = self.MAX_ATTEMPTS
 
        attempt = attempts_store[identifier]
        now = time.time()
 
        # ロックアウト中かチェック
        if attempt.locked_until > now:
            remaining = int(attempt.locked_until - now)
            return {
                "allowed": False,
                "reason": "account_locked",
                "retry_after": remaining,
            }
 
        # ウィンドウのリセット
        if now - attempt.first_attempt > self.WINDOW:
            attempt.count = 0
            attempt.first_attempt = now
 
        # 試行回数の記録
        if attempt.count == 0:
            attempt.first_attempt = now
        attempt.count += 1
        attempt.last_attempt = now
 
        # ロックアウト判定
        if attempt.count >= max_attempts:
            attempt.locked_until = now + self.LOCKOUT_DURATION
            return {
                "allowed": False,
                "reason": "too_many_attempts",
                "retry_after": self.LOCKOUT_DURATION,
            }
 
        # プログレッシブ遅延
        delay_idx = min(attempt.count, len(self.PROGRESSIVE_DELAYS) - 1)
        delay = self.PROGRESSIVE_DELAYS[delay_idx]
 
        return {
            "allowed": True,
            "delay": delay,
            "attempts_remaining": max_attempts - attempt.count,
            "require_captcha": attempt.count >= 3,
        }
 
    def reset(self, username: str) -> None:
        """ログイン成功時にカウンターをリセット"""
        self.account_attempts.pop(username, None)
 
    def is_credential_stuffing(self, ip: str, threshold: int = 10) -> bool:
        """Credential Stuffing 攻撃の検知
 
        同一IPから短時間に多数の異なるアカウントへログインを
        試行するパターンを検知する。
        """
        ip_attempt = self.ip_attempts.get(ip)
        if ip_attempt and ip_attempt.count >= threshold:
            return True
        return False
 
 
# 使用例
protection = BruteForceProtection()
 
# ログイン試行
result = protection.check_and_record("admin", "192.168.1.100")
print(f"許可: {result['allowed']}")
print(f"残り試行: {result.get('attempts_remaining')}")
 
# 5回失敗後
for _ in range(5):
    result = protection.check_and_record("admin", "192.168.1.100")
 
print(f"ロックアウト: {not result['allowed']}")
print(f"解除まで: {result.get('retry_after')}秒")
 
# ログイン成功時のリセット
protection.reset("admin")

# コード例5: TOTP（Time-based One-Time Password）の実装
import hmac
import hashlib
import struct
import time
import base64
import secrets
 
 
class TOTP:
    """RFC 6238 準拠の TOTP 実装
 
    WHY TOTP が広く使われるか:
    - サーバーとクライアントが時刻のみを共有すれば良い
    - 通信不要（オフラインで生成可能）
    - Google Authenticator 等の標準アプリで利用可能
    - SMS OTP と異なり SIM スワップ攻撃に耐性がある
    """
 
    def __init__(self, secret: bytes, digits: int = 6,
                 period: int = 30, algorithm=hashlib.sha1):
        """
        Args:
            secret: 共有シークレット（最低160ビット推奨）
            digits: OTP の桁数（6桁が標準）
            period: コード更新間隔（30秒が標準）
            algorithm: HMACアルゴリズム（SHA-1が標準、SHA-256も可）
        """
        self.secret = secret
        self.digits = digits
        self.period = period
        self.algorithm = algorithm
 
    @classmethod
    def generate_secret(cls) -> str:
        """新しい TOTP シークレットを生成
 
        Base32エンコードで返す（QRコード/手入力用）。
        20バイト = 160ビットのエントロピー。
        """
        return base64.b32encode(secrets.token_bytes(20)).decode()
 
    def generate_code(self, timestamp: float = None) -> str:
        """現在の TOTP コードを生成
 
        内部処理:
        1. 現在時刻を period で割ってカウンター値を算出
        2. カウンター値を HMAC でハッシュ
        3. Dynamic Truncation で 6 桁の数値を抽出
        """
        if timestamp is None:
            timestamp = time.time()
        counter = int(timestamp) // self.period
        counter_bytes = struct.pack(">Q", counter)
 
        mac = hmac.new(self.secret, counter_bytes, self.algorithm).digest()
        offset = mac[-1] & 0x0F
        truncated = struct.unpack(">I", mac[offset:offset + 4])[0]
        truncated &= 0x7FFFFFFF
        code = truncated % (10 ** self.digits)
        return str(code).zfill(self.digits)
 
    def verify(self, code: str, window: int = 1) -> bool:
        """TOTP コードを検証
 
        WHY window が必要か:
        サーバーとクライアントの時刻にわずかなズレがあっても
        認証が成功するよう、前後 window ステップ分を許容する。
        window=1 の場合、前後30秒（合計90秒間）有効。
        """
        now = time.time()
        for offset in range(-window, window + 1):
            check_time = now + (offset * self.period)
            if hmac.compare_digest(
                self.generate_code(check_time), code
            ):
                return True
        return False
 
    def get_provisioning_uri(self, account: str,
                             issuer: str) -> str:
        """QRコード用のプロビジョニングURIを生成
 
        Google Authenticator 等のアプリで読み取るためのURI。
        """
        secret_b32 = base64.b32encode(self.secret).decode()
        return (
            f"otpauth://totp/{issuer}:{account}"
            f"?secret={secret_b32}"
            f"&issuer={issuer}"
            f"&algorithm=SHA1"
            f"&digits={self.digits}"
            f"&period={self.period}"
        )
 
 
# 使用例
secret_b32 = TOTP.generate_secret()
print(f"シークレット: {secret_b32}")
 
secret_bytes = base64.b32decode(secret_b32)
totp = TOTP(secret_bytes)
 
# QRコード用URI
uri = totp.get_provisioning_uri("user@example.com", "MyApp")
print(f"プロビジョニングURI: {uri}")
 
# コード生成と検証
code = totp.generate_code()
print(f"現在のコード: {code}")
print(f"検証結果: {totp.verify(code)}")  # True

# コード例6: JWT の安全な実装
import jwt
import time
import secrets
from typing import Optional
 
 
class SecureJWT:
    """安全な JWT トークン管理
 
    WHY HS256 ではなく RS256 を推奨するか:
    - HS256: 署名と検証に同じ秘密鍵を使用
      → 検証側にも秘密鍵を共有する必要がある
      → マイクロサービスでは鍵の配布が問題になる
    - RS256: 秘密鍵で署名、公開鍵で検証
      → 検証側には公開鍵のみ配布すればよい
      → JWKS エンドポイントで公開鍵を自動配布可能
    """
 
    # トークンのブラックリスト（即時無効化用）
    _blacklist: set = set()
 
    def __init__(self, secret_key: str, algorithm: str = "HS256"):
        self.secret_key = secret_key
        # 安全でないアルゴリズムを拒否
        allowed_algorithms = {"HS256", "HS384", "HS512",
                              "RS256", "RS384", "RS512",
                              "ES256", "ES384", "ES512"}
        if algorithm not in allowed_algorithms:
            raise ValueError(
                f"Algorithm '{algorithm}' is not allowed. "
                f"Use one of: {allowed_algorithms}"
            )
        self.algorithm = algorithm
 
    def create_token(self, user_id: str, roles: list,
                     expires_in: int = 900) -> str:
        """アクセストークンを生成（デフォルト15分）
 
        最小権限の原則: トークンには必要最小限の情報のみ含める。
        個人情報（メールアドレス等）はトークンに含めない。
        """
        now = int(time.time())
        payload = {
            "sub": user_id,
            "roles": roles,
            "iat": now,
            "exp": now + expires_in,
            "nbf": now,                      # Not Before
            "jti": secrets.token_hex(16),    # JWT ID（リプレイ防止）
            "iss": "myapp",                  # 発行者
            "aud": "myapp-api",              # 対象者
        }
        return jwt.encode(payload, self.secret_key,
                          algorithm=self.algorithm)
 
    def create_refresh_token(self, user_id: str,
                             expires_in: int = 604800) -> str:
        """リフレッシュトークンを生成（デフォルト7日）
 
        リフレッシュトークンはアクセストークンの再発行にのみ使用。
        スコープを制限し、DBに保存して管理する。
        """
        now = int(time.time())
        payload = {
            "sub": user_id,
            "type": "refresh",
            "iat": now,
            "exp": now + expires_in,
            "jti": secrets.token_hex(16),
            "iss": "myapp",
        }
        return jwt.encode(payload, self.secret_key,
                          algorithm=self.algorithm)
 
    def verify_token(self, token: str) -> Optional[dict]:
        """トークンを検証
 
        検証項目:
        1. 署名の正当性（改竄検知）
        2. 有効期限（exp）
        3. 発行時刻（iat）
        4. Not Before（nbf）
        5. 発行者（iss）
        6. 対象者（aud）
        7. ブラックリストチェック
        """
        try:
            payload = jwt.decode(
                token,
                self.secret_key,
                algorithms=[self.algorithm],  # リスト形式で明示指定
                options={
                    "require": ["exp", "iat", "sub", "iss"],
                    "verify_exp": True,
                    "verify_iat": True,
                    "verify_nbf": True,
                },
                issuer="myapp",
                audience="myapp-api",
            )
 
            # ブラックリストチェック
            if payload.get("jti") in self._blacklist:
                return None
 
            return payload
        except jwt.ExpiredSignatureError:
            return None  # トークン期限切れ
        except jwt.InvalidTokenError:
            return None  # その他の検証エラー
 
    def revoke_token(self, token: str) -> bool:
        """トークンを即座に無効化（ブラックリスト方式）
 
        短い有効期限と組み合わせることで、
        ブラックリストのサイズを制限する。
        """
        try:
            # 署名検証なしでペイロードを取得（jti のみ必要）
            payload = jwt.decode(
                token,
                self.secret_key,
                algorithms=[self.algorithm],
                options={"verify_exp": False},
            )
            jti = payload.get("jti")
            if jti:
                self._blacklist.add(jti)
                return True
        except jwt.InvalidTokenError:
            pass
        return False
 
 
# 使用例
jwt_manager = SecureJWT("my-secret-key-at-least-256-bits-long!!")
 
# アクセストークン生成
access_token = jwt_manager.create_token("user123", ["viewer"])
print(f"アクセストークン: {access_token[:50]}...")
 
# リフレッシュトークン生成
refresh_token = jwt_manager.create_refresh_token("user123")
 
# トークン検証
payload = jwt_manager.verify_token(access_token)
print(f"ペイロード: {payload}")
 
# トークン無効化
jwt_manager.revoke_token(access_token)
revoked_result = jwt_manager.verify_token(access_token)
print(f"無効化後の検証: {revoked_result}")  # None

# コード例7: セキュアなパスワードリセットの実装
import secrets
import time
import hashlib
from typing import Optional, Tuple
 
 
class PasswordResetManager:
    """安全なパスワードリセット管理
 
    パスワードリセットの脆弱性パターン:
    1. 推測可能なリセットトークン（連番、タイムスタンプベース）
    2. トークンの有効期限なし
    3. トークンの使い回し（リプレイ攻撃）
    4. ユーザーの存在有無が推測可能なエラーメッセージ
    """
 
    TOKEN_EXPIRY = 3600  # 1時間
    TOKEN_LENGTH = 32    # 256ビット
 
    def __init__(self):
        self.reset_tokens: dict = {}
 
    def request_reset(self, email: str) -> Tuple[str, str]:
        """パスワードリセットをリクエスト
 
        セキュリティ上の注意:
        - ユーザーの存在有無に関わらず同じレスポンスを返す
        - レートリミットを適用（同一メールへの連続リクエストを制限）
        """
        # 暗号学的に安全なトークンを生成
        token = secrets.token_urlsafe(self.TOKEN_LENGTH)
 
        # トークンをハッシュ化して保存（DB漏洩時の保護）
        token_hash = hashlib.sha256(token.encode()).hexdigest()
 
        self.reset_tokens[token_hash] = {
            "email": email,
            "created_at": time.time(),
            "used": False,
        }
 
        return token, token_hash
 
    def verify_and_consume(self, token: str) -> Optional[str]:
        """リセットトークンを検証して消費（一回限り使用）"""
        token_hash = hashlib.sha256(token.encode()).hexdigest()
        data = self.reset_tokens.get(token_hash)
 
        if not data:
            return None
 
        # 有効期限チェック
        if time.time() - data["created_at"] > self.TOKEN_EXPIRY:
            del self.reset_tokens[token_hash]
            return None
 
        # 使用済みチェック
        if data["used"]:
            # トークンの再利用はセキュリティインシデントの可能性
            del self.reset_tokens[token_hash]
            return None
 
        # トークンを使用済みとしてマーク
        data["used"] = True
 
        return data["email"]
 
 
# 使用例
reset_mgr = PasswordResetManager()
token, _ = reset_mgr.request_reset("user@example.com")
email = reset_mgr.verify_and_consume(token)
print(f"リセット対象: {email}")
 
# 2回目の使用は失敗
email2 = reset_mgr.verify_and_consume(token)
print(f"再利用: {email2}")  # None

# NG: パスワードを平文で保存
def save_user(username, password):
    db.execute(
        "INSERT INTO users (username, password) VALUES (?, ?)",
        (username, password)  # 平文！
    )
 
# NG: AES等の可逆暗号化で保存
from cryptography.fernet import Fernet
key = Fernet.generate_key()
def save_user_encrypted(username, password):
    encrypted = Fernet(key).encrypt(password.encode())
    db.execute(
        "INSERT INTO users (username, password) VALUES (?, ?)",
        (username, encrypted)  # 復号可能 = 鍵が漏洩すれば全滅
    )
 
# OK: Argon2id で一方向ハッシュ化
from argon2 import PasswordHasher
ph = PasswordHasher()
def save_user_secure(username, password):
    hashed = ph.hash(password)  # 復元不可能
    db.execute(
        "INSERT INTO users (username, password_hash) VALUES (?, ?)",
        (username, hashed)
    )

# NG: アルゴリズムを検証しない
import jwt
payload = jwt.decode(token, secret, algorithms=None)  # 全アルゴリズム許可
 
# NG: デフォルト設定のまま使用（ライブラリによっては none を許可）
payload = jwt.decode(token, secret)
 
# OK: 許可するアルゴリズムを明示的に指定
payload = jwt.decode(
    token,
    secret,
    algorithms=["HS256"],  # HS256 のみ許可
    options={"require": ["exp", "iss"]},
)

# NG: ログイン前後でセッションIDが変わらない
@app.route("/login", methods=["POST"])
def login():
    if authenticate(request.form["username"], request.form["password"]):
        session["user_id"] = get_user_id(request.form["username"])
        return redirect("/dashboard")
        # セッションIDは変わらない！攻撃者が事前に仕込んだIDがそのまま使われる
 
# OK: ログイン成功時にセッションIDを再生成
@app.route("/login", methods=["POST"])
def login_secure():
    if authenticate(request.form["username"], request.form["password"]):
        session.regenerate()  # セッションIDを再生成
        session["user_id"] = get_user_id(request.form["username"])
        return redirect("/dashboard")

# NG: ユーザーの存在を推測可能なエラーメッセージ
def login(username, password):
    user = find_user(username)
    if not user:
        return {"error": "ユーザーが見つかりません"}  # ユーザー存在が分かる
    if not verify_password(password, user.password_hash):
        return {"error": "パスワードが間違っています"}  # ユーザーは存在する
 
# OK: 同一のエラーメッセージ
def login_secure(username, password):
    user = find_user(username)
    if not user or not verify_password(password, user.password_hash if user else "dummy"):
        return {"error": "ユーザー名またはパスワードが間違っています"}