webmail-client/backend/app/identity/reset.py

"""Password reset tokens (SPEC §6.1).

Single-use, 30-minute, Redis-backed tokens, emailed as a link to the account's
recovery address through the system mailer (:mod:`app.identity.mailer`).

A token survives a wrong 2FA code (the page asks again) but not five: after
that it is burned, so the link cannot be used to guess codes. Requests are
capped per account so the reset form cannot be used to flood someone's inbox.
"""

from __future__ import annotations

import secrets
import uuid

from app.redis_client import aw, get_redis

_PREFIX = "pwreset:"
_FAILS = "pwreset-fail:"
_RATE = "pwreset-rate:"
_TTL = 30 * 60  # 30 minutes
MAX_TOTP_FAILURES = 5
MAX_REQUESTS_PER_HOUR = 3


async def issue_token(user_id: uuid.UUID) -> str:
    """Create a single-use reset token bound to ``user_id`` and return it."""
    token = secrets.token_urlsafe(32)
    await get_redis().set(_PREFIX + token, str(user_id), ex=_TTL)
    return token


async def request_allowed(user_id: uuid.UUID) -> bool:
    """Count a reset request; False once the account has had its hourly share."""
    redis = get_redis()
    key = _RATE + str(user_id)
    count = int(await redis.incr(key))
    if count == 1:
        await redis.expire(key, 3600)
    return count <= MAX_REQUESTS_PER_HOUR


def _user_id(raw: str | None) -> uuid.UUID | None:
    if raw is None:
        return None
    try:
        return uuid.UUID(raw)
    except ValueError:
        return None


async def peek_token(token: str) -> uuid.UUID | None:
    """The user a still-valid token belongs to, without using it up."""
    if not token:
        return None
    return _user_id(await get_redis().get(_PREFIX + token))


async def consume_token(token: str) -> uuid.UUID | None:
    """Atomically consume a reset token, returning the user_id (or None)."""
    if not token:
        return None
    redis = get_redis()
    raw = await aw(redis.getdel(_PREFIX + token))  # single-use: fetch and delete atomically
    await redis.delete(_FAILS + token)
    return _user_id(raw)


async def record_totp_failure(token: str) -> None:
    """Count a wrong 2FA code against the token; burn it at the limit."""
    redis = get_redis()
    fails = await redis.incr(_FAILS + token)
    if fails == 1:
        await redis.expire(_FAILS + token, _TTL)
    if fails >= MAX_TOTP_FAILURES:
        await redis.delete(_PREFIX + token, _FAILS + token)