A+B 複合 — 目的別モデリング × 単一責任 × Userクラスの落とし穴（会員(Member)管理 Bad→Good Ch14）× member-auth-api Cloud Armor クレデンシャルスタッフィング対策（LB+Serverless NEG統一 × rate_based_ban × Adaptive Protection × 脅威インテリジェンス × OWASP CRS段階導入 × アカウント列挙対策 × ログマスキング）

📋 コピー# tests/test_member.py
import pytest
import hashlib
from member import (
    AuthCredential, MemberProfile, PointBalance, MembershipStatus,
    Member, MemberRank, resolve_rank, InsufficientPointsError,
    MemberLifetimeValueCalculator,
)


class TestResolveRank:
    def test_gold_threshold(self):
        assert resolve_rank(120_000) == MemberRank.GOLD

    def test_bronze_below_all_thresholds(self):
        assert resolve_rank(1_000) == MemberRank.BRONZE


class TestPointBalance:
    def test_add_then_use(self):
        balance = PointBalance()
        balance.add(500)
        balance.use(200)
        assert balance.amount == 300

    def test_use_more_than_balance_raises(self):
        balance = PointBalance()
        balance.add(500)
        with pytest.raises(InsufficientPointsError):
            balance.use(10_000)


class TestAuthCredential:
    def test_verify_password_success(self):
        credential = AuthCredential(
            member_id="M001",
            password_hash=hashlib.sha256(b"secret").hexdigest(),
        )
        assert credential.verify_password("secret") is True
        assert credential.verify_password("wrong") is False


class TestMemberLifetimeValueCalculator:
    def test_calculate_delegates_to_repository(self):
        class FakeRepo:
            def sum_order_amount(self, member_id: str) -> int:
                return 42_000

        calculator = MemberLifetimeValueCalculator(order_repository=FakeRepo())
        assert calculator.calculate("M001") == 42_000

📋 コピー"""login_response.py — アカウント列挙防止(修正6) × ログイン試行ログのマスキング(修正7)"""
from __future__ import annotations

import hashlib
import logging
from typing import Protocol

logger = logging.getLogger(__name__)

# 修正6: 理由によらず同一メッセージ(存在しないメール/パスワード不一致を区別させない)
_GENERIC_AUTH_FAILURE_MESSAGE = "メールアドレスまたはパスワードが正しくありません"


class AuthenticationFailedError(Exception):
    """認証失敗(理由の詳細はログにのみ残し、レスポンスには含めない)。"""


class CredentialRepository(Protocol):
    def find_by_email(self, email: str) -> "AuthCredential | None": ...


def _hash_email(email: str) -> str:
    """メールアドレスをハッシュ化してログに残す(生のPIIをログに出さない)。"""
    return hashlib.sha256(email.encode()).hexdigest()[:12]


def authenticate(email: str, raw_password: str, credential_repo: CredentialRepository) -> "AuthCredential":
    """メールアドレス・パスワードで認証する。

    Args:
        email: 入力されたメールアドレス。
        raw_password: 入力された平文パスワード。
        credential_repo: 認証情報リポジトリ。

    Returns:
        認証に成功した AuthCredential。

    Raises:
        AuthenticationFailedError: メール未登録・パスワード不一致のいずれの場合も
            同一メッセージで送出する(アカウント列挙対策)。
    """
    credential = credential_repo.find_by_email(email)
    if credential is None:
        logger.info("login failed: email not found", extra={"email_hash": _hash_email(email)})
        raise AuthenticationFailedError(_GENERIC_AUTH_FAILURE_MESSAGE)  # 修正6: メール未登録も同一メッセージ

    if not credential.verify_password(raw_password):
        logger.info("login failed: password mismatch", extra={"email_hash": _hash_email(email)})
        raise AuthenticationFailedError(_GENERIC_AUTH_FAILURE_MESSAGE)  # 修正6: 不一致も同一メッセージ・同一例外型

    return credential


class PasswordMaskingFilter(logging.Filter):
    """修正7: ログレコードからraw_password等の機密フィールドをマスキングするフィルタ。"""

    _SENSITIVE_KEYS = frozenset({"raw_password", "password", "password_hash"})

    def filter(self, record: logging.LogRecord) -> bool:
        for key in self._SENSITIVE_KEYS:
            if hasattr(record, key):
                setattr(record, key, "***MASKED***")
        return True

Bad vs Good 設計比較

確認コマンド

📋 コピー# 1. Adaptive Protectionが有効化されていることを確認
gcloud compute security-policies describe member-auth-waf-policy --format="yaml(adaptiveProtectionConfig)"
# Expected: layer7DdosDefenseConfig.enable: true

# 2. レートベースルールが/api/v1/loginに適用されていることを確認
gcloud compute security-policies rules list --security-policy member-auth-waf-policy \
  --format="table(priority,action,description)"
# Expected: priority=3000, action=rate_based_ban が一覧に含まれる

# 3. レートベースbanの動作確認(同一IPから25回連続ログイン試行)
for i in $(seq 1 25); do
  curl -s -o /dev/null -w "%{http_code}\n" -X POST https://member-auth.example.com/api/v1/login \
    -d '{"email":"a@example.com","password":"x"}'
done
# Expected: 20回目付近までは401、それ以降は429(ban)が返る

# 4. アカウント列挙対策の確認(未登録メール・パスワード不一致でメッセージが一致すること)
curl -s -X POST https://member-auth.example.com/api/v1/login \
  -d '{"email":"not-exist@example.com","password":"x"}' | jq .message
curl -s -X POST https://member-auth.example.com/api/v1/login \
  -d '{"email":"real@example.com","password":"wrong"}' | jq .message
# Expected: 両方とも "メールアドレスまたはパスワードが正しくありません" で一致

# 5. OWASP CRSがpreviewモード(まだblockしていない)であることを確認
gcloud compute security-policies rules describe 2000 --security-policy member-auth-waf-policy \
  --format="value(preview)"
# Expected: True(段階導入中)