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 защита).
  - Состояние доступа проверяется ТОЛЬКО ПОСЛЕ проверки пароля, и осмысленный
    ответ (403 «пробный доступ закончился») получает исключительно тот, кто
    пароль уже доказал. Ветвление ДО пароля превратило бы отдельный статус в
    оракул существования логина: перебором можно было бы перечислить аккаунты,
    не зная ни одного пароля (миграция data/sql/auth/004, WHY-2).
  - #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.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.identity_store import AccessState, get_identity_db
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 = "неверный логин или пароль"

# Единственный ответ логина, который НЕ generic 401: пароль верный, но пробный
# период истёк. `code` — машиночитаемый контракт для фронта (текст можно менять,
# ветку по нему — нет). Потребитель: `loginErrorMessage` в
# tradein-mvp/frontend/src/app/login/page.tsx — читает `detail.code` из
# `HTTPError.body` (frontend/src/lib/api.ts отдаёт тело ответа как есть) и
# показывает экран про пробный период вместо generic «Проверьте подключение».
# Меняешь значение здесь — меняй и там.
_ACCESS_EXPIRED_CODE = "access_expired"
_ACCESS_EXPIRED_MESSAGE = "Пробный доступ закончился"


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_identity_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)

    # Пароль проверен ВЫШЕ и безусловно — только теперь смотрим на состояние
    # доступа. Порядок несущий, а не стилистический: см. модульный docstring.
    if user is None or not password_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)

    access_state = user["access_state"]
    if access_state is AccessState.TRIAL_EXPIRED:
        # Пароль верный, сессия НЕ создаётся. Единственный не-generic ответ:
        # аккаунт существует и владелец это уже доказал паролем, так что
        # осмысленный текст ничего не раскрывает постороннему.
        # В режиме identity_store="tradein" эта ветка недостижима: булев
        # is_active даёт только active/disabled (identity_store.to_access_state).
        schedule_event(
            event_type="login_blocked_expired",
            username=user["username"],
            ip=ip,
            user_agent=user_agent,
            path="/api/v1/auth/login",
            method="POST",
        )
        raise HTTPException(
            status_code=403,
            detail={"code": _ACCESS_EXPIRED_CODE, "message": _ACCESS_EXPIRED_MESSAGE},
        )

    if not access_state.can_sign_in:
        # disabled (и любое нераспознанное состояние — to_access_state fail-closed)
        # → ТОТ ЖЕ generic 401 и то же событие, что при неверном пароле:
        # заблокированный аккаунт неотличим от несуществующего.
        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)

    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_identity_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}