# コード例1: AES-256-GCM による認証付き暗号
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
import os
 
 
class AESEncryptor:
    """AES-256-GCM による認証付き暗号化
 
    WHY AES-GCM を推奨するか:
    - 暗号化と認証を同時に行う（Authenticated Encryption with Associated Data）
    - 改ざん検知機能内蔵（認証タグによる完全性保証）
    - 高速（AES-NI ハードウェアアクセラレーション対応）
    - NIST 推奨、TLS 1.3 のデフォルト暗号スイート
 
    WHY ノンス（Nonce）が重要か:
    - 同じ鍵で同じノンスを使うと暗号文から平文の情報が漏洩する
    - GCM のノンスは同一鍵で絶対に再利用してはならない
    - 96ビットのランダムノンスなら 2^32 回まで安全に使用可能
    """
 
    KEY_SIZE = 32   # 256ビット
    NONCE_SIZE = 12  # 96ビット（GCM推奨）
 
    def __init__(self, key: bytes = None):
        if key is None:
            key = AESGCM.generate_key(bit_length=256)
        if len(key) != self.KEY_SIZE:
            raise ValueError(f"鍵長は{self.KEY_SIZE}バイトである必要があります")
        self._aesgcm = AESGCM(key)
        self._key = key
 
    def encrypt(self, plaintext: bytes,
                associated_data: bytes = None) -> bytes:
        """暗号化（ノンス + 暗号文 + 認証タグを返す）
 
        Args:
            plaintext: 暗号化するデータ
            associated_data: 追加認証データ（AAD）
                暗号化はしないが、改ざんを検知したいデータ。
                例: メタデータ、ヘッダ情報
 
        Returns:
            nonce(12バイト) + ciphertext + tag(16バイト)
        """
        nonce = os.urandom(self.NONCE_SIZE)
        ciphertext = self._aesgcm.encrypt(nonce, plaintext, associated_data)
        return nonce + ciphertext  # ノンスを先頭に付加
 
    def decrypt(self, data: bytes,
                associated_data: bytes = None) -> bytes:
        """復号（ノンスを分離してから復号）
 
        認証タグの検証に失敗した場合は例外が発生する。
        これにより改ざんされたデータの復号を防止する。
        """
        nonce = data[:self.NONCE_SIZE]
        ciphertext = data[self.NONCE_SIZE:]
        return self._aesgcm.decrypt(nonce, ciphertext, associated_data)
 
 
# 使用例
encryptor = AESEncryptor()
 
# 基本的な暗号化/復号
plaintext = b"This is a secret message"
encrypted = encryptor.encrypt(plaintext)
decrypted = encryptor.decrypt(encrypted)
assert decrypted == plaintext
print(f"暗号化成功: {len(encrypted)} バイト")
 
# AAD（追加認証データ）付き暗号化
# メタデータは暗号化しないが、改ざんを検知したい場合
aad = b"metadata:user_id=123"
encrypted_with_aad = encryptor.encrypt(plaintext, aad)
decrypted_with_aad = encryptor.decrypt(encrypted_with_aad, aad)
assert decrypted_with_aad == plaintext
 
# AADが異なると復号に失敗する（改ざん検知）
try:
    encryptor.decrypt(encrypted_with_aad, b"tampered_metadata")
except Exception as e:
    print(f"改ざん検知: {type(e).__name__}")

# コード例2: ChaCha20-Poly1305 による暗号化
from cryptography.hazmat.primitives.ciphers.aead import ChaCha20Poly1305
import os
 
 
class ChaChaEncryptor:
    """ChaCha20-Poly1305 による認証付き暗号化
 
    WHY ChaCha20 を検討するか:
    - AES-NI 非対応環境（古いCPU、ARM等）で AES-GCM より高速
    - ソフトウェア実装でもサイドチャネル攻撃に耐性がある
    - TLS 1.3 で AES-GCM と並ぶ標準暗号スイート
    - Google が多くのサービスで採用（特にモバイル向け）
    """
 
    NONCE_SIZE = 12  # 96ビット
 
    def __init__(self, key: bytes = None):
        if key is None:
            key = ChaCha20Poly1305.generate_key()
        self._cipher = ChaCha20Poly1305(key)
 
    def encrypt(self, plaintext: bytes,
                associated_data: bytes = None) -> bytes:
        """暗号化"""
        nonce = os.urandom(self.NONCE_SIZE)
        ciphertext = self._cipher.encrypt(nonce, plaintext, associated_data)
        return nonce + ciphertext
 
    def decrypt(self, data: bytes,
                associated_data: bytes = None) -> bytes:
        """復号"""
        nonce = data[:self.NONCE_SIZE]
        ciphertext = data[self.NONCE_SIZE:]
        return self._cipher.decrypt(nonce, ciphertext, associated_data)
 
 
# 使用例
chacha = ChaChaEncryptor()
encrypted = chacha.encrypt(b"Secret data for mobile app")
decrypted = chacha.decrypt(encrypted)
print(f"ChaCha20 復号結果: {decrypted}")

# コード例3: RSA と ECDSA の鍵生成・署名・検証
from cryptography.hazmat.primitives.asymmetric import rsa, ec, padding
from cryptography.hazmat.primitives import hashes, serialization
 
 
class AsymmetricCrypto:
    """非対称鍵暗号の操作
 
    RSA と ECC の使い分け:
    - RSA-4096: 互換性が最も高い、レガシーシステムとの接続
    - ECDSA P-256: 新規システムの標準、TLS 証明書
    - Ed25519: 最新の署名アルゴリズム、最高速度
    """
 
    @staticmethod
    def generate_rsa_keypair(key_size: int = 4096):
        """RSA 鍵ペアの生成
 
        公開指数 65537 (0x10001) は標準的な値で、
        暗号化の効率とセキュリティのバランスが良い。
        """
        private_key = rsa.generate_private_key(
            public_exponent=65537,
            key_size=key_size,
        )
        return private_key, private_key.public_key()
 
    @staticmethod
    def generate_ec_keypair(curve=None):
        """ECC 鍵ペアの生成
 
        P-256 (secp256r1): NIST 推奨、最も広くサポート
        P-384 (secp384r1): より高セキュリティ
        """
        if curve is None:
            curve = ec.SECP256R1()
        private_key = ec.generate_private_key(curve)
        return private_key, private_key.public_key()
 
    @staticmethod
    def rsa_sign(private_key, message: bytes) -> bytes:
        """RSA-PSS 署名
 
        WHY PKCS#1 v1.5 ではなく PSS を使うか:
        - PSS（Probabilistic Signature Scheme）はランダム性を含む
        - 同じメッセージに対しても毎回異なる署名が生成される
        - 数学的に証明されたセキュリティを持つ
        """
        return private_key.sign(
            message,
            padding.PSS(
                mgf=padding.MGF1(hashes.SHA256()),
                salt_length=padding.PSS.MAX_LENGTH,
            ),
            hashes.SHA256(),
        )
 
    @staticmethod
    def rsa_verify(public_key, message: bytes, signature: bytes) -> bool:
        """RSA-PSS 署名の検証"""
        try:
            public_key.verify(
                signature,
                message,
                padding.PSS(
                    mgf=padding.MGF1(hashes.SHA256()),
                    salt_length=padding.PSS.MAX_LENGTH,
                ),
                hashes.SHA256(),
            )
            return True
        except Exception:
            return False
 
    @staticmethod
    def ec_sign(private_key, message: bytes) -> bytes:
        """ECDSA 署名"""
        return private_key.sign(
            message,
            ec.ECDSA(hashes.SHA256()),
        )
 
    @staticmethod
    def ec_verify(public_key, message: bytes, signature: bytes) -> bool:
        """ECDSA 署名の検証"""
        try:
            public_key.verify(
                signature,
                message,
                ec.ECDSA(hashes.SHA256()),
            )
            return True
        except Exception:
            return False
 
    @staticmethod
    def export_public_key_pem(public_key) -> str:
        """公開鍵を PEM 形式でエクスポート"""
        return public_key.public_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PublicFormat.SubjectPublicKeyInfo,
        ).decode()
 
    @staticmethod
    def export_private_key_pem(private_key, passphrase: bytes = None) -> str:
        """秘密鍵を PEM 形式でエクスポート（暗号化オプション付き）"""
        encryption = (
            serialization.BestAvailableEncryption(passphrase)
            if passphrase
            else serialization.NoEncryption()
        )
        return private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=encryption,
        ).decode()
 
 
# 使用例
crypto = AsymmetricCrypto()
 
# RSA 署名
rsa_priv, rsa_pub = crypto.generate_rsa_keypair()
message = b"Important document"
rsa_sig = crypto.rsa_sign(rsa_priv, message)
print(f"RSA 署名検証: {crypto.rsa_verify(rsa_pub, message, rsa_sig)}")  # True
 
# ECDSA 署名
ec_priv, ec_pub = crypto.generate_ec_keypair()
ec_sig = crypto.ec_sign(ec_priv, message)
print(f"ECDSA 署名検証: {crypto.ec_verify(ec_pub, message, ec_sig)}")  # True
 
# 公開鍵のエクスポート
pem = crypto.export_public_key_pem(ec_pub)
print(f"公開鍵 PEM:\n{pem[:100]}...")

# コード例4: ハッシュ関数の安全な使い方
import hashlib
import hmac
 
 
class SecureHash:
    """安全なハッシュ操作
 
    ハッシュ関数は「元に戻せない」一方向関数であり、
    暗号化（元に戻せる）とは根本的に異なる。
    """
 
    @staticmethod
    def sha256(data: bytes) -> str:
        """SHA-256 ハッシュ（データの完全性検証用）
 
        用途: ファイルのチェックサム、データの改ざん検知
        用途外: パスワード保存（Argon2id を使うこと）
        """
        return hashlib.sha256(data).hexdigest()
 
    @staticmethod
    def sha3_256(data: bytes) -> str:
        """SHA-3-256 ハッシュ（SHA-2 の代替）
 
        SHA-3 は SHA-2 とは異なる数学的構造（Keccak）を持つ。
        SHA-2 に脆弱性が発見された場合のフォールバック。
        """
        return hashlib.sha3_256(data).hexdigest()
 
    @staticmethod
    def file_hash(filepath: str, algorithm: str = "sha256") -> str:
        """大きなファイルのハッシュをストリーミング計算
 
        WHY ストリーミングが必要か:
        - 数GBのファイルを一度にメモリに読み込むとOOMが発生する
        - 8KB ずつ読み込むことでメモリ使用量を一定に保つ
        """
        h = hashlib.new(algorithm)
        with open(filepath, "rb") as f:
            while chunk := f.read(8192):
                h.update(chunk)
        return h.hexdigest()
 
    @staticmethod
    def constant_time_compare(a: str, b: str) -> bool:
        """タイミング攻撃を防ぐ比較
 
        WHY 通常の == ではダメか:
        - a == b は最初の不一致バイトで比較を打ち切る
        - 攻撃者は応答時間の差から正しいバイトを推測できる
        - hmac.compare_digest は常に全バイトを比較する
        """
        return hmac.compare_digest(a.encode(), b.encode())
 
 
# 使用例
sh = SecureHash()
 
# データの完全性検証
data = b"Important document content"
hash_value = sh.sha256(data)
print(f"SHA-256: {hash_value}")
 
# 改ざん検知
modified_data = b"Important document contenT"  # 末尾1文字変更
modified_hash = sh.sha256(modified_data)
print(f"改ざん検知: {hash_value != modified_hash}")  # True
 
# 安全な比較
print(f"一致: {sh.constant_time_compare(hash_value, hash_value)}")  # True

# コード例5: HMAC によるメッセージ認証
import hmac
import hashlib
import time
 
 
class MessageAuthenticator:
    """HMAC によるメッセージ認証
 
    HMAC の内部構造:
    HMAC(K, M) = H((K ^ opad) || H((K ^ ipad) || M))
    - 二重ハッシュ構造で Length Extension 攻撃を防止
    - 鍵をそのままハッシュに含めるとLength Extension攻撃に脆弱
    """
 
    def __init__(self, key: bytes):
        if len(key) < 32:
            raise ValueError("鍵は最低32バイト（256ビット）必要です")
        self.key = key
 
    def create_mac(self, message: bytes) -> str:
        """メッセージの MAC を計算"""
        return hmac.new(self.key, message, hashlib.sha256).hexdigest()
 
    def verify_mac(self, message: bytes, mac: str) -> bool:
        """MAC を検証（タイミング攻撃対策済み）"""
        expected = self.create_mac(message)
        return hmac.compare_digest(expected, mac)
 
    def create_signed_message(self, message: bytes) -> bytes:
        """タイムスタンプ付き署名メッセージを生成
 
        用途: 一時的な URL、API トークン、CSRF トークン
        """
        timestamp = str(int(time.time())).encode()
        payload = timestamp + b"." + message
        mac = self.create_mac(payload)
        return payload + b"." + mac.encode()
 
    def verify_signed_message(self, signed: bytes,
                               max_age: int = 300) -> bytes:
        """署名メッセージを検証（有効期限チェック付き）
 
        Args:
            signed: 署名付きメッセージ
            max_age: 最大有効期間（秒）
        """
        parts = signed.rsplit(b".", 1)
        if len(parts) != 2:
            raise ValueError("Invalid signed message format")
 
        payload, mac = parts[0], parts[1].decode()
        if not self.verify_mac(payload, mac):
            raise ValueError("MAC verification failed")
 
        timestamp_str, message = payload.split(b".", 1)
        timestamp = int(timestamp_str)
        if time.time() - timestamp > max_age:
            raise ValueError("Message expired")
 
        return message
 
 
# 使用例
auth = MessageAuthenticator(b"secret-key-32-bytes-long!!!!!!!!")
 
# 基本的な MAC の計算と検証
message = b"payment:100:USD"
mac_value = auth.create_mac(message)
print(f"MAC: {mac_value}")
print(f"検証: {auth.verify_mac(message, mac_value)}")  # True
print(f"改ざん検知: {auth.verify_mac(b'payment:999:USD', mac_value)}")  # False
 
# タイムスタンプ付き署名メッセージ
signed = auth.create_signed_message(b"action=approve&id=123")
original = auth.verify_signed_message(signed, max_age=60)
print(f"検証済みメッセージ: {original}")

# コード例6: ハイブリッド暗号（エンベロープ暗号化）
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
import os
 
 
class HybridEncryption:
    """ハイブリッド暗号方式（RSA + AES-GCM）
 
    TLS、PGP、S/MIME 等の実際のプロトコルと同じ設計パターン。
    AWS KMS のエンベロープ暗号化もこの原理に基づく。
    """
 
    def __init__(self):
        self.private_key = rsa.generate_private_key(
            public_exponent=65537, key_size=4096
        )
        self.public_key = self.private_key.public_key()
 
    def encrypt(self, plaintext: bytes) -> dict:
        """ハイブリッド暗号化
 
        ステップ:
        1. ランダムな AES 鍵（データ鍵）を生成
        2. データ鍵でデータを暗号化（AES-GCM: 高速）
        3. データ鍵を RSA 公開鍵で暗号化（RSA-OAEP: 安全な鍵配送）
        4. 平文のデータ鍵をメモリから消去
        """
        # 1. ランダムな AES 鍵を生成
        data_key = AESGCM.generate_key(bit_length=256)
 
        # 2. データ鍵でデータを暗号化
        nonce = os.urandom(12)
        aesgcm = AESGCM(data_key)
        encrypted_data = aesgcm.encrypt(nonce, plaintext, None)
 
        # 3. データ鍵を RSA 公開鍵で暗号化
        encrypted_key = self.public_key.encrypt(
            data_key,
            padding.OAEP(
                mgf=padding.MGF1(algorithm=hashes.SHA256()),
                algorithm=hashes.SHA256(),
                label=None,
            ),
        )
 
        # 4. 平文データ鍵をゼロクリア（セキュリティのため）
        data_key = b"\x00" * len(data_key)
 
        return {
            "encrypted_key": encrypted_key,
            "nonce": nonce,
            "encrypted_data": encrypted_data,
        }
 
    def decrypt(self, package: dict) -> bytes:
        """ハイブリッド復号"""
        # 1. RSA 秘密鍵でデータ鍵を復号
        data_key = self.private_key.decrypt(
            package["encrypted_key"],
            padding.OAEP(
                mgf=padding.MGF1(algorithm=hashes.SHA256()),
                algorithm=hashes.SHA256(),
                label=None,
            ),
        )
        # 2. データ鍵でデータを復号
        aesgcm = AESGCM(data_key)
        plaintext = aesgcm.decrypt(
            package["nonce"], package["encrypted_data"], None
        )
 
        # データ鍵をゼロクリア
        data_key = b"\x00" * len(data_key)
 
        return plaintext
 
 
# 使用例
hybrid = HybridEncryption()
plaintext = b"Sensitive financial data: account=123456, balance=1000000"
 
# 暗号化
package = hybrid.encrypt(plaintext)
print(f"暗号化データ鍵: {len(package['encrypted_key'])} バイト")
print(f"暗号化データ: {len(package['encrypted_data'])} バイト")
 
# 復号
decrypted = hybrid.decrypt(package)
assert decrypted == plaintext
print(f"復号成功: {decrypted.decode()}")

# NG: 独自の暗号化アルゴリズム（XOR暗号の例）
def my_encrypt(data: bytes, key: bytes) -> bytes:
    """独自暗号: 単純なXOR"""
    return bytes(a ^ b for a, b in zip(data, key * (len(data) // len(key) + 1)))
 
# 問題点:
# - 鍵の再利用でパターンが漏洩する
# - 暗号学的な安全性が証明されていない
# - 認証機能がない（改ざん検知不可）
# - 既知平文攻撃で鍵が復元される
 
# OK: 標準アルゴリズムを使用
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
key = AESGCM.generate_key(bit_length=256)
aesgcm = AESGCM(key)
nonce = os.urandom(12)
ciphertext = aesgcm.encrypt(nonce, data, None)

# NG: AES-ECB モード
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
cipher = Cipher(algorithms.AES(key), modes.ECB())
# → 同一平文ブロック = 同一暗号文ブロック → パターン漏洩
 
# OK: AES-GCM（認証付き暗号モード）
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
aesgcm = AESGCM(key)
ciphertext = aesgcm.encrypt(nonce, plaintext, aad)
# → パターン隠蔽 + 改ざん検知

# NG: 固定ノンスの使用
nonce = b"\x00" * 12  # 固定ノンス
for message in messages:
    ciphertext = aesgcm.encrypt(nonce, message, None)  # 同じノンスを再利用
 
# OK: 毎回ランダムなノンスを生成
import os
for message in messages:
    nonce = os.urandom(12)  # 毎回新しいノンス
    ciphertext = aesgcm.encrypt(nonce, message, None)

# NG: CBC モードで暗号化のみ（認証なし）
cipher = Cipher(algorithms.AES(key), modes.CBC(iv))
encryptor = cipher.encryptor()
ciphertext = encryptor.update(plaintext) + encryptor.finalize()
# → 暗号文の改ざんを検知できない（Padding Oracle Attack のリスク）
 
# OK: GCM（認証付き暗号）
aesgcm = AESGCM(key)
ciphertext = aesgcm.encrypt(nonce, plaintext, None)
# → 暗号文が改ざんされると復号時に例外が発生

# デバッグ用ユーティリティ
import logging
import traceback
from functools import wraps
 
# ロガーの設定
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s [%(levelname)s] %(name)s: %(message)s'
)
logger = logging.getLogger(__name__)
 
def debug_decorator(func):
    """関数の入出力をログ出力するデコレータ"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        logger.debug(f"呼び出し: {func.__name__}(args={args}, kwargs={kwargs})")
        try:
            result = func(*args, **kwargs)
            logger.debug(f"戻り値: {func.__name__} -> {result}")
            return result
        except Exception as e:
            logger.error(f"例外発生: {func.__name__}: {e}")
            logger.error(traceback.format_exc())
            raise
    return wrapper
 
@debug_decorator
def process_data(items):
    """データ処理（デバッグ対象）"""
    if not items:
        raise ValueError("空のデータ")
    return [item * 2 for item in items]

# 設計判断の記録テンプレート
class ArchitectureDecisionRecord:
    """ADR (Architecture Decision Record) の作成"""
 
    def __init__(self, title: str):
        self.title = title
        self.context = ""
        self.decision = ""
        self.consequences = []
        self.alternatives = []
 
    def set_context(self, context: str):
        """背景と課題の記述"""
        self.context = context
        return self
 
    def set_decision(self, decision: str):
        """決定内容の記述"""
        self.decision = decision
        return self
 
    def add_consequence(self, consequence: str, positive: bool = True):
        """結果の追加"""
        self.consequences.append({
            'description': consequence,
            'type': 'positive' if positive else 'negative'
        })
        return self
 
    def add_alternative(self, name: str, reason_rejected: str):
        """却下した代替案の追加"""
        self.alternatives.append({
            'name': name,
            'reason_rejected': reason_rejected
        })
        return self
 
    def to_markdown(self) -> str:
        """Markdown形式で出力"""
        md = f"# ADR: {self.title}\n\n"
        md += f"## 背景\n{self.context}\n\n"
        md += f"## 決定\n{self.decision}\n\n"
        md += "## 結果\n"
        for c in self.consequences:
            icon = "✅" if c['type'] == 'positive' else "⚠️"
            md += f"- {icon} {c['description']}\n"
        md += "\n## 却下した代替案\n"
        for a in self.alternatives:
            md += f"- **{a['name']}**: {a['reason_rejected']}\n"
        return md

# チーム間のAPI契約定義
from dataclasses import dataclass
from typing import List, Optional
from enum import Enum
 
class Priority(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"
 
@dataclass
class APIContract:
    """チーム間のAPI契約"""
    endpoint: str
    method: str
    owner_team: str
    consumers: List[str]
    sla_ms: int  # レスポンスタイムSLA
    priority: Priority
 
    def validate_sla(self, actual_ms: int) -> bool:
        """SLA準拠の確認"""
        return actual_ms <= self.sla_ms
 
    def to_openapi(self) -> dict:
        """OpenAPI形式で出力"""
        return {
            'path': self.endpoint,
            'method': self.method,
            'x-owner': self.owner_team,
            'x-consumers': self.consumers,
            'x-sla-ms': self.sla_ms
        }
 
# 使用例
contracts = [
    APIContract(
        endpoint="/api/v1/users",
        method="GET",
        owner_team="user-team",
        consumers=["order-team", "notification-team"],
        sla_ms=200,
        priority=Priority.HIGH
    ),
    APIContract(
        endpoint="/api/v1/orders",
        method="POST",
        owner_team="order-team",
        consumers=["payment-team", "inventory-team"],
        sla_ms=500,
        priority=Priority.CRITICAL
    )
]