Bài 83: OTP với Rate Limit & Chống Brute-force

Sau bài này bạn sẽ:

• Hiểu tại sao lưu hash OTP thay vì plaintext và cách dùng salt
• Triển khai rate limit 3 tầng cho send endpoint: cooldown, per-period, per-IP
• Viết Lua script atomic để check nhiều key cùng lúc
• Chặn brute-force verify bằng attempt counter + delete OTP khi quá limit
• Hiểu tradeoff OTP length (4/6/8 chữ số) và phone normalization

OTP là gì

OTP là mã ngắn (thường 4-8 chữ số) do server tạo ngẫu nhiên, gửi qua kênh out-of-band — tức kênh khác với kênh chính đang dùng. Nếu user đang ở web, OTP đến qua SMS hoặc email. Nếu user đang dùng app, OTP đến qua authenticator app.

Đặc điểm cốt lõi:

• Dùng một lần — sau khi verify thành công, OTP bị xóa ngay.
• TTL ngắn — thường 1-5 phút, server tự xóa khi hết hạn.
• Out-of-band — attacker phải kiểm soát thêm kênh thứ hai mới bypass được.

Use case phổ biến: login 2FA, xác thực số điện thoại khi đăng ký, password reset qua SMS.

Vấn đề 1 — Spam send endpoint

Kẻ tấn công gọi /send-otp liên tục cho nhiều số điện thoại. Phí SMS dao động 0.01–0.05 USD/tin nhắn. Nếu không có rate limit, một script đơn giản gửi 100.000 yêu cầu có thể tiêu tốn 1.000–5.000 USD phí SMS trong vài phút. Người dùng cũng bị spam inbox không có lý do.

Vấn đề 2 — Brute-force verify endpoint

OTP 6 chữ số có 10⁶ = 1.000.000 giá trị có thể. Nếu không giới hạn số lần thử, attacker gọi /verify-otp tự động với 1.000 req/s = 1.000.000 thử trong 1.000 giây (~17 phút) để đảm bảo match ít nhất một lần trước khi OTP hết hạn. Với OTP 4 chữ số (10.000 combination), chỉ cần 10 giây.

Hai vấn đề cần hai lớp bảo vệ khác nhau, nhưng đều dùng Redis counter làm nền.

Không lưu OTP plaintext. Nếu Redis bị dump (snapshot RDB rò rỉ, hoặc attacker đọc được memory), plaintext OTP lộ ngay. Thay vào đó, lưu SHA-256(salt + ":" + otp).

import secrets
import hashlib
import time
import redis

r = redis.Redis(decode_responses=True)

OTP_TTL = 300      # 5 phút
OTP_LENGTH = 6


def gen_otp() -> str:
    # secrets.randbelow sinh số ngẫu nhiên cryptographically secure
    return f"{secrets.randbelow(10 ** OTP_LENGTH):0{OTP_LENGTH}d}"


def hash_otp(otp: str, salt: str) -> str:
    return hashlib.sha256(f"{salt}:{otp}".encode()).hexdigest()


def send_otp(phone: str) -> None:
    otp = gen_otp()
    salt = secrets.token_urlsafe(16)  # 16 bytes = 128 bits entropy
    r.hset(f"otp:{phone}", mapping={
        "hash": hash_otp(otp, salt),
        "salt": salt,
        "attempts": 0,
        "created_at": int(time.time()),
    })
    r.expire(f"otp:{phone}", OTP_TTL)
    # sms_provider.send(phone, f"Your code: {otp}")
    print(f"[DEMO] OTP for {phone}: {otp}")

Lưu ý thiết kế:

• Salt ngẫu nhiên mỗi lần — nếu dùng salt cố định, attacker có thể precompute rainbow table cho 10⁶ giá trị.
• Hash (không encrypt) — server không cần giải mã; chỉ cần hash lại input rồi so sánh.
• HSET với mapping — lưu salt, attempts, created_at cùng key để tránh race condition giữa nhiều lần đọc.
• EXPIRE ngay sau HSET — nếu process crash giữa chừng, chỉ mất lần set expire, OTP cũ sẽ overwrite khi gọi lại.

Ba tầng rate limit bổ sung cho nhau — mỗi tầng chặn một loại abuse khác nhau:

• Tầng 1 — Cooldown 60s per phone: ngăn user (hoặc attacker) resend liên tục.
• Tầng 2 — Max 5 OTP / hour per phone: giới hạn tổng số tin nhắn gửi đến một số.
• Tầng 3 — Max 50 OTP / hour per IP: chặn script abuse từ một địa chỉ IP dù thay đổi số điện thoại.

class TooManyRequests(Exception):
    pass


def send_otp_with_limits(phone: str, client_ip: str) -> None:
    # Tầng 1: cooldown 60s — SET NX để chỉ set nếu key chưa tồn tại
    if not r.set(f"otp:cooldown:{phone}", "1", nx=True, ex=60):
        raise TooManyRequests("Wait 60 seconds before resending")

    # Tầng 2: max 5 OTP / hour per phone
    hour_key = f"otp:limit:hour:{phone}"
    count = r.incr(hour_key)
    if count == 1:
        r.expire(hour_key, 3600)
    if count > 5:
        raise TooManyRequests("Max 5 OTP requests per hour for this number")

    # Tầng 3: max 50 OTP / hour per IP
    ip_key = f"otp:limit:ip:{client_ip}"
    count = r.incr(ip_key)
    if count == 1:
        r.expire(ip_key, 3600)
    if count > 50:
        raise TooManyRequests("IP rate limit exceeded")

    send_otp(phone)

Vấn đề với code này: ba tầng là ba round-trip Redis độc lập. Giữa các lệnh, process khác có thể chạy xen. Phần tiếp theo giải quyết bằng Lua.

Lưu ý về pattern INCR + EXPIRE: nếu process crash giữa INCR và EXPIRE, key sẽ không có TTL và tồn tại vĩnh viễn. Cách an toàn hơn là dùng Lua (xem bước 5) hoặc SET với giá trị mặc định.

Lua script chạy atomic trên Redis — toàn bộ script hoàn thành như một transaction, không có request nào chen vào giữa. Trả về giá trị âm khi bị reject, trả về 1 khi pass.

-- KEYS[1] = otp:cooldown:{phone}
-- KEYS[2] = otp:limit:hour:{phone}
-- KEYS[3] = otp:limit:day:{phone}
-- KEYS[4] = otp:limit:ip:{client_ip}

if redis.call("EXISTS", KEYS[1]) == 1 then
  return -1  -- cooldown chưa hết
end

local h = redis.call("INCR", KEYS[2])
if h == 1 then redis.call("EXPIRE", KEYS[2], 3600) end
if h > 5 then return -2 end  -- quá 5/hour per phone

local d = redis.call("INCR", KEYS[3])
if d == 1 then redis.call("EXPIRE", KEYS[3], 86400) end
if d > 10 then return -3 end  -- quá 10/day per phone

local ip = redis.call("INCR", KEYS[4])
if ip == 1 then redis.call("EXPIRE", KEYS[4], 3600) end
if ip > 50 then return -4 end  -- quá 50/hour per IP

redis.call("SET", KEYS[1], "1", "EX", 60)
return 1

RATE_LIMIT_SCRIPT = """
if redis.call("EXISTS", KEYS[1]) == 1 then
  return -1
end
local h = redis.call("INCR", KEYS[2])
if h == 1 then redis.call("EXPIRE", KEYS[2], 3600) end
if h > 5 then return -2 end
local d = redis.call("INCR", KEYS[3])
if d == 1 then redis.call("EXPIRE", KEYS[3], 86400) end
if d > 10 then return -3 end
local ip = redis.call("INCR", KEYS[4])
if ip == 1 then redis.call("EXPIRE", KEYS[4], 3600) end
if ip > 50 then return -4 end
redis.call("SET", KEYS[1], "1", "EX", 60)
return 1
"""

_rate_limit_sha = None


def _get_script_sha() -> str:
    global _rate_limit_sha
    if _rate_limit_sha is None:
        _rate_limit_sha = r.script_load(RATE_LIMIT_SCRIPT)
    return _rate_limit_sha


ERROR_MAP = {
    -1: "Wait 60 seconds before resending",
    -2: "Max 5 OTP requests per hour for this number",
    -3: "Max 10 OTP requests per day for this number",
    -4: "IP rate limit exceeded",
}


def check_send_limits(phone: str, client_ip: str) -> None:
    keys = [
        f"otp:cooldown:{phone}",
        f"otp:limit:hour:{phone}",
        f"otp:limit:day:{phone}",
        f"otp:limit:ip:{client_ip}",
    ]
    result = r.evalsha(_get_script_sha(), 4, *keys)
    if result != 1:
        raise TooManyRequests(ERROR_MAP.get(result, "Rate limit exceeded"))

Dùng EVALSHA + script_load thay vì EVAL trực tiếp để Redis cache compiled script — tiết kiệm bandwidth mỗi request.

Sau khi user nhập OTP, server so sánh hash. Mỗi lần sai tăng attempts. Khi đạt MAX_ATTEMPTS, xóa OTP ngay — user phải request OTP mới.

import hmac

MAX_ATTEMPTS = 5


class Unauthorized(Exception):
    pass


def verify_otp(phone: str, submitted: str) -> bool:
    key = f"otp:{phone}"
    data = r.hgetall(key)

    if not data:
        raise Unauthorized("OTP expired or not found")

    attempts = int(data["attempts"])
    if attempts >= MAX_ATTEMPTS:
        r.delete(key)
        raise Unauthorized("Too many attempts — request a new OTP")

    submitted_hash = hash_otp(submitted, data["salt"])

    # Timing-safe comparison — xem bước 8
    if not hmac.compare_digest(submitted_hash, data["hash"]):
        r.hincrby(key, "attempts", 1)
        remaining = MAX_ATTEMPTS - attempts - 1
        raise Unauthorized(f"Wrong OTP — {remaining} attempts remaining")

    # Verify thành công — xóa OTP (burn after use)
    r.delete(key)
    return True

Tại sao xóa OTP sau verify thành công?

Nếu không xóa, attacker intercept session sau khi user verify thành công có thể replay OTP trong khoảng thời gian còn lại. OTP chỉ có giá trị dùng một lần — xóa ngay khi dùng là bắt buộc.

Tại sao xóa OTP khi hết attempts?

Giữ OTP sau khi đã thử hết attempts không có ý nghĩa bảo mật. Nếu attacker biết OTP vẫn còn tồn tại trong Redis, có thể cố tìm cách reset counter. Xóa hẳn buộc user tạo session OTP mới, server biết attack đã xảy ra.

Phân tích xác suất

Với max 5 attempts trên 1 OTP 6 chữ số: xác suất đoán đúng = 5 / 1.000.000 = 0,0005%. Attacker cần tạo OTP mới sau mỗi 5 lần thử — mỗi lần tạo mới bị giới hạn bởi cooldown và rate limit của send endpoint.

Ngay cả khi đã limit 5 attempts per OTP, attacker vẫn có thể bypass nếu kiểm soát được send endpoint (ví dụ có nhiều SIM, hoặc có số điện thoại hợp lệ). Thêm rate limit tổng cho verify:

def rate_limit_verify(phone: str) -> None:
    """Max 10 verify attempts / hour per phone số."""
    key = f"otp:verify:limit:{phone}"
    count = r.incr(key)
    if count == 1:
        r.expire(key, 3600)
    if count > 10:
        raise TooManyRequests("Too many verify attempts — try again later")


def verify_otp_safe(phone: str, submitted: str) -> bool:
    rate_limit_verify(phone)
    return verify_otp(phone, submitted)

Counter này đếm số lần gọi /verify-otp cho một số điện thoại trong 1 giờ, bất kể kết quả đúng hay sai. Limit 10 verify/hour đủ chặt để ngăn script tự động, nhưng vẫn để user thật có đủ cơ hội nhập lại.

Lưu ý: counter verify nên không reset khi verify thành công. Nếu reset, attacker có thể cố tình verify đúng một lần (dùng OTP hợp lệ họ có được từ nguồn khác) để xóa counter rồi tiếp tục brute-force lần sau.

So sánh string bằng == trả về False ngay khi gặp ký tự đầu tiên khác nhau. Thời gian so sánh thay đổi tùy theo số ký tự khớp từ đầu — đây gọi là timing attack. Attacker đo thời gian response có thể suy ra dần từng ký tự của hash.

Trong thực tế với SHA-256 hex (64 ký tự), timing attack rất khó do noise mạng, nhưng best practice là dùng hmac.compare_digest — so sánh trong thời gian hằng định bất kể vị trí khác nhau:

import hmac

# Nên dùng:
if hmac.compare_digest(submitted_hash, stored_hash):
    ...

# Không dùng:
# if submitted_hash == stored_hash:
#     ...

hmac.compare_digest có trong Python standard library từ 3.3. Không cần cài thêm dependency.

Kết hợp 6 chữ số với max 5 attempts per OTP và 10 verify/hour, xác suất brute-force thành công gần bằng 0 trong window 5 phút của OTP. Đây là lý do 6 chữ số là tiêu chuẩn phổ biến nhất.

Cùng một số điện thoại có thể được nhập theo nhiều format:

• 0912345678
• +84912345678
• 84912345678
• 0912 345 678 (có khoảng trắng)

Nếu không normalize trước khi tạo Redis key, rate limit bị bypass hoàn toàn: attacker gửi OTP 5 lần với format 0912..., sau đó chuyển sang +84912... để lách tầng 2.

import phonenumbers


def normalize_phone(raw: str, default_region: str = "VN") -> str:
    """
    Normalize phone number về E.164 format.
    "0912345678" -> "+84912345678"
    Raise ValueError nếu số không hợp lệ.
    """
    parsed = phonenumbers.parse(raw, default_region)
    if not phonenumbers.is_valid_number(parsed):
        raise ValueError(f"Invalid phone number: {raw}")
    return phonenumbers.format_number(
        parsed, phonenumbers.PhoneNumberFormat.E164
    )

Thư viện phonenumbers (Python) là wrapper của Google libphonenumber. Cài bằng pip install phonenumbers. Normalize một lần trước mọi thao tác tạo key Redis.

Tương tự với email: lowercase + strip whitespace trước khi dùng làm key.

TOTP — Time-based OTP

TOTP (RFC 6238) là cơ chế khác với SMS OTP. Server và client chia sẻ một secret key, cả hai generate OTP từ HMAC-SHA1(secret, floor(time / 30)) mà không cần giao tiếp. Google Authenticator, Authy, 1Password đều dùng TOTP.

Với TOTP, server không cần lưu OTP vào Redis — chỉ verify on-the-fly từ secret đã lưu trong DB. Redis vẫn cần cho rate limit verify và để track "used TOTP code" (prevent replay trong cùng 30s window):

def mark_totp_used(user_id: str, totp_code: str, window: int = 60) -> bool:
    """
    Ngăn replay attack: cùng TOTP code không được dùng 2 lần trong window.
    window = 60s để cover 2 time-step (30s mỗi step).
    """
    key = f"totp:used:{user_id}:{totp_code}"
    # SET NX: chỉ thành công nếu key chưa tồn tại
    return bool(r.set(key, "1", nx=True, ex=window))

Backup codes

User mất điện thoại → không dùng được TOTP app. Backup codes là tập 10 mã dùng một lần, user lưu ở nơi an toàn (in ra hoặc lưu password manager).

def gen_backup_codes(user_id: str) -> list[str]:
    """
    Tạo 10 backup code, lưu hash vào Redis Set.
    Không đặt TTL — backup codes vĩnh viễn cho đến khi dùng hết hoặc regenerate.
    """
    codes = [secrets.token_urlsafe(8) for _ in range(10)]
    hashed = [hashlib.sha256(c.encode()).hexdigest() for c in codes]
    pipe = r.pipeline()
    # Xóa set cũ trước khi tạo mới (regenerate)
    pipe.delete(f"backup:{user_id}")
    pipe.sadd(f"backup:{user_id}", *hashed)
    pipe.execute()
    return codes  # Trả về plaintext CHỈ một lần duy nhất cho user lưu


def verify_backup_code(user_id: str, code: str) -> bool:
    """
    SREM trả về số phần tử bị xóa: 1 = tồn tại và đã dùng, 0 = không tồn tại.
    Dùng một lần rồi xóa luôn khỏi Set.
    """
    h = hashlib.sha256(code.encode()).hexdigest()
    return bool(r.srem(f"backup:{user_id}", h))

Backup code không có TTL vì không gắn với session hay time window. Dùng Redis Set để lưu là hợp lý: SREM tự xóa code sau khi dùng trong một lệnh atomic.

Bảo vệ chi phí SMS

Rate limit per phone và per IP đã giảm thiểu đáng kể. Thêm các lớp nữa:

• Global daily quota: tổng số SMS gửi ra mỗi ngày trên toàn hệ thống. Nếu vượt quota, dừng hẳn và alert team.
• CAPTCHA tại /send-otp: Cloudflare Turnstile hoặc reCAPTCHA v3 để phân biệt human/bot trước khi gửi SMS.
• Phone allowlist: trong giai đoạn beta, chỉ cho phép số đã đăng ký trước.

DAILY_SMS_QUOTA = 10_000  # Điều chỉnh theo ngân sách


def check_global_sms_quota() -> None:
    from datetime import date
    key = f"sms:global:daily:{date.today().isoformat()}"
    count = r.incr(key)
    if count == 1:
        r.expire(key, 86400)
    if count > DAILY_SMS_QUOTA:
        raise TooManyRequests("Daily SMS quota exceeded — contact support")

Audit log

Log đủ để detect pattern tấn công:

• OTP sent: phone (masked), timestamp, IP, provider response code.
• OTP verified: phone (masked), timestamp, success/fail, attempt number.
• Rate limit hit: endpoint, phone/IP, limit type.

Mask phone khi log: +849****678 — đủ để debug, không leak số đầy đủ vào log file. Log về SIEM hoặc Elasticsearch để alert khi burst failed verify tăng đột biến.

Anti-patterns

• Lưu OTP plaintext — Redis snapshot (RDB) leak là vector tấn công thực tế; plaintext OTP trong dump = attacker có thể xác thực ngay.
• Không rate limit send — một request đơn giản với vòng lặp có thể tiêu hết $10.000 phí SMS trong vài giờ.
• Không limit verify attempts — với OTP 4 chữ số + unlimited verify, brute-force thành công trong dưới 10 giây.
• Resend giữ nguyên OTP cũ — nếu resend không gen OTP mới, attacker intercept "OTP cũ" vẫn hợp lệ trong toàn bộ TTL.
• Phone không normalize — ba format của cùng một số = bypass rate limit dễ dàng.
• Không xóa OTP sau verify thành công — replay attack trong khoảng TTL còn lại.
• Salt cố định — rainbow table 10⁶ entry là nhỏ, precompute trong vài giây.

Best practices

• 6 chữ số, TTL 5 phút, max 5 verify attempts per OTP.
• Hash + random salt lưu Redis — không bao giờ plaintext.
• Rate limit send: cooldown 60s + max 5/hour + max 10/day per phone + max 50/hour per IP.
• Rate limit verify: max 10/hour per phone.
• Lua atomic cho multi-key check tầng send.
• CAPTCHA tại /send-otp endpoint.
• Normalize phone number (E.164) trước khi tạo bất kỳ Redis key nào.
• Timing-safe comparison với hmac.compare_digest.
• Audit log toàn bộ sự kiện OTP (với masked phone).
• Global daily SMS quota + alert khi vượt ngưỡng.

Tổng kết

OTP an toàn với Redis cần giải quyết hai threat model độc lập: spam send (tốn tiền) và brute-force verify (mất xác thực). Lưu hash+salt thay vì plaintext là baseline không thể bỏ. Rate limit send dùng 3-4 counter Redis với Lua atomic; rate limit verify dùng thêm attempt counter trực tiếp trên hash key của OTP. Phone normalization trước mọi thao tác key là điều kiện tiên quyết để rate limit có hiệu quả.

Quiz

1. Giải thích tại sao lưu hash+salt tốt hơn hash đơn thuần (không salt) khi bảo vệ OTP trong Redis dump.
2. Tầng cooldown 60s và tầng max 5/hour giải quyết hai vấn đề khác nhau như thế nào? Có thể dùng chỉ một trong hai không?
3. Attacker có 5 lần thử cho OTP 6 chữ số. Xác suất đoán đúng là bao nhiêu? Nếu attacker có thể tạo OTP mới 10 lần/giờ, xác suất thành công trong 1 giờ là bao nhiêu?
4. Tại sao counter verify không nên reset khi verify thành công?
5. TOTP (Google Authenticator) không cần Redis lưu OTP nhưng vẫn cần Redis cho hai mục đích. Đó là gì?

Đáp án gợi ý

1. SHA-256 của 10⁶ OTP có thể precompute thành rainbow table trong vài giây (dữ liệu nhỏ). Salt ngẫu nhiên per OTP khiến mỗi hash phụ thuộc vào salt cụ thể — không thể dùng bảng precomputed nào.
2. Cooldown 60s chặn resend liên tiếp nhanh từ một người dùng (kể cả user thật nhấn "gửi lại" nhiều lần). Max 5/hour chặn tổng lượng OTP gửi đến một số trong khoảng dài hơn, kể cả khi attacker chờ hết cooldown. Chỉ dùng một trong hai thì hở lỗ hổng còn lại.
3. 5 / 1.000.000 = 0,0005% per OTP. Trong 1 giờ, với 10 OTP mới tạo được: 10 × 5 = 50 thử / 1.000.000 = 0,005%. Rate limit verify per phone giới hạn thêm: nếu max 10 verify/hour thì chỉ có 10 thử tổng cộng dù tạo được bao nhiêu OTP, xác suất còn 0,001%.
4. Nếu reset khi thành công, attacker dùng một OTP hợp lệ có được từ nguồn khác để reset counter, sau đó tiếp tục brute-force lần sau với counter về 0.
5. (1) Rate limit verify (max N lần/giờ per user); (2) Track used TOTP codes để ngăn replay attack trong cùng 30s window.

Bài tiếp theo

Bài 84 đi vào Password Reset Token: thiết kế flow an toàn từ request reset đến đặt password mới — token generation, hash lưu Redis, one-time use, expiry, và các lỗ hổng phổ biến trong flow reset.

Tham khảo

• Redis Docs — Lệnh SET (NX, EX options)
• Redis Docs — EVALSHA
• Redis Docs — HSET
• Redis Docs — SREM
• RFC 6238 — TOTP: Time-Based One-Time Password Algorithm
• Python Docs — hmac.compare_digest
• python-phonenumbers — Google libphonenumber wrapper
• redis-py — Redis client cho Python