gendesign/tradein-mvp/backend/app/api/v1/auth.py

"""POST /api/v1/auth/login + /logout — DB-backed session auth (#2552, эпик #2549).

Переходный механизм, параллельный legacy Caddy trusted-header auth (roles.yaml).
См. `app.core.rbac.rbac_guard` (dual-mode resolver) и `app.services.auth_session`
(session CRUD). Mounted at `/api/v1/auth`; через Caddy `uri strip_prefix /trade-in`
это `/trade-in/api/v1/auth/*` снаружи.

Security:
  - Неверные creds (неизвестный username / неактивен / password_hash NULL /
    неверный пароль) → ОДИНАКОВЫЙ 401 с generic сообщением — не раскрываем,
    существует ли username (user-enumeration защита).
  - #2552 post-review Medium 2: `verify_password` ВСЕГДА вызывается ровно
    один раз — для несуществующего username / NULL password_hash сверяем
    против статичного dummy-хеша (`_DUMMY_PASSWORD_HASH`, сгенерирован один
    раз на импорте модуля), результат игнорируется. Без этого короткое
    замыкание (`user is None → сразу 401`) давало наблюдаемую разницу во
    времени ответа (~1мс без bcrypt vs ~100-300мс с ним) — классический
    timing-oracle для user-enumeration, даже при одинаковом detail-сообщении.
  - Rate-limit по (username, IP) — ЖЁСТЧЕ общего `RateLimitMiddleware`
    (`/api/*`), т.к. login — типичная brute-force поверхность. Использует
    `SlidingWindowLimiter` (тот же примитив, что и общий rate-limit). Ключ
    length-prefixed (`len(username):username:ip`) — без этого произвольный
    username с `:` внутри мог бы схлопнуть бюджет с другой (username, ip)
    парой (IPv6-адреса тоже содержат `:`, так что просто эскейпить разделитель
    в username недостаточно — паразитная граница возможна с обеих сторон).
  - Raw-пароль НИКОГДА не логируется и не попадает в user_events payload —
    только username/ip/user_agent/path/method (см. schedule_event ниже).
"""

from __future__ import annotations

import logging
import secrets
from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, Request, Response
from pydantic import BaseModel
from sqlalchemy.orm import Session

from app.core.config import settings
from app.core.db import get_db
from app.core.password import hash_password, verify_password
from app.core.ratelimit import SlidingWindowLimiter, _client_ip
from app.services.auth_session import create_session, get_user_by_username, revoke_session
from app.services.user_events import schedule_event

logger = logging.getLogger(__name__)

router = APIRouter()

# Отдельный, более узкий бюджет чем общий per-user/per-IP `/api/*` лимит
# (см. app.core.ratelimit.SlidingWindowLimiter docstring — designed именно для
# такого случая). Ключ = username+IP: не даёт распределённому brute-force по
# ОДНОМУ аккаунту с разных IP уйти от лимита целиком (per-IP было бы недостаточно),
# и не блокирует ВЕСЬ IP из-за перебора чужих логинов одним же клиентом.
_LOGIN_LIMITER = SlidingWindowLimiter(
    limit=settings.login_rate_limit,
    window_s=settings.login_rate_limit_window_s,
)

# Timing-oracle защита (см. module docstring): bcrypt-хеш случайного пароля,
# сгенерированный ОДИН РАЗ на импорте модуля — используется вместо
# password_hash, когда юзер не найден/деактивирован/без пароля, чтобы
# `verify_password` (доминирующая по времени операция, ~100-300мс) всегда
# отрабатывала полный bcrypt-компар, независимо от того, существует ли аккаунт.
_DUMMY_PASSWORD_HASH = hash_password(secrets.token_urlsafe(16))

_INVALID_CREDENTIALS_DETAIL = "неверный логин или пароль"


class LoginRequest(BaseModel):
    username: str
    password: str


class LoginResponse(BaseModel):
    ok: bool = True


@router.post("/login", response_model=LoginResponse)
async def login(
    body: LoginRequest,
    request: Request,
    response: Response,
    db: Annotated[Session, Depends(get_db)],
) -> LoginResponse:
    ip = _client_ip(request)
    user_agent = request.headers.get("user-agent")
    rate_key = f"{len(body.username)}:{body.username}:{ip}"

    retry_after = _LOGIN_LIMITER.check(rate_key)
    if retry_after is not None:
        raise HTTPException(
            status_code=429,
            detail="слишком много попыток входа, попробуйте позже",
            headers={"Retry-After": str(int(retry_after) + 1)},
        )

    user = get_user_by_username(db, body.username)
    hash_to_check = (
        user["password_hash"]
        if user is not None and user["password_hash"] is not None
        else _DUMMY_PASSWORD_HASH
    )
    # ВСЕГДА вызывается — dummy-хеш при отсутствующем юзере/NULL password_hash
    # держит время ответа одинаковым независимо от существования аккаунта.
    password_ok = verify_password(body.password, hash_to_check)
    credentials_ok = user is not None and user["is_active"] and password_ok

    if not credentials_ok:
        schedule_event(
            event_type="login_failed",
            username=body.username,
            ip=ip,
            user_agent=user_agent,
            path="/api/v1/auth/login",
            method="POST",
        )
        raise HTTPException(status_code=401, detail=_INVALID_CREDENTIALS_DETAIL)

    assert user is not None  # narrowed by credentials_ok above
    token = create_session(db, user_id=user["user_id"], ip=ip, user_agent=user_agent)

    response.set_cookie(
        key=settings.session_cookie_name,
        value=token,
        max_age=settings.session_ttl_hours * 3600,
        httponly=True,
        secure=True,
        samesite="lax",
        path="/",
    )

    schedule_event(
        event_type="login_success",
        username=user["username"],
        ip=ip,
        user_agent=user_agent,
        path="/api/v1/auth/login",
        method="POST",
    )

    return LoginResponse(ok=True)


@router.post("/logout")
async def logout(
    request: Request,
    response: Response,
    db: Annotated[Session, Depends(get_db)],
) -> dict[str, bool]:
    token = request.cookies.get(settings.session_cookie_name)
    if token:
        revoke_session(db, token)
    response.delete_cookie(key=settings.session_cookie_name, path="/")
    return {"ok": True}