lux_fiscal/backend/app/services/auth.py

"""Логика аутентификации: вход, ротация refresh-токенов, выход."""

from __future__ import annotations

import uuid
from datetime import UTC, datetime

from fastapi import Request
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.config import settings
from app.core.logging import get_logger
from app.core.security import (
    create_access_token,
    generate_refresh_token,
    hash_password,
    hash_refresh_token,
    password_needs_rehash,
    refresh_token_expiry,
    verify_password,
)
from app.db.models.audit import AuditAction
from app.db.models.user import RefreshToken, User
from app.schemas.auth import TokenPair
from app.services import audit

log = get_logger(__name__)


class AuthError(Exception):
    """Вход или обновление токена отклонены."""


async def authenticate(
    session: AsyncSession, *, email: str, password: str, request: Request | None = None
) -> User:
    """Проверяет учётные данные.

    Все отказы — с одинаковым текстом: различающиеся сообщения позволили бы
    перебором выяснить, какие адреса зарегистрированы.
    """
    normalized = email.strip().lower()
    user = await session.scalar(select(User).where(User.email == normalized))

    if user is None or not verify_password(password, user.password_hash):
        await audit.record(
            session,
            action=AuditAction.LOGIN_FAILED,
            user=user,
            actor_label=normalized,
            payload={"reason": "bad_credentials"},
            request=request,
        )
        await session.commit()
        raise AuthError("Неверный email или пароль")

    if not user.is_active:
        await audit.record(
            session,
            action=AuditAction.LOGIN_FAILED,
            user=user,
            payload={"reason": "inactive"},
            request=request,
        )
        await session.commit()
        raise AuthError("Учётная запись отключена")

    # Параметры argon2 со временем ужесточаются — обновляем хеш на живом пароле.
    if password_needs_rehash(user.password_hash):
        user.password_hash = hash_password(password)

    user.last_login_at = datetime.now(UTC)
    return user


async def issue_token_pair(
    session: AsyncSession,
    user: User,
    *,
    request: Request | None = None,
    replaces: RefreshToken | None = None,
) -> TokenPair:
    """Выдаёт пару токенов и сохраняет хеш refresh-токена."""
    raw_refresh, token_hash = generate_refresh_token()

    stored = RefreshToken(
        user_id=user.id,
        token_hash=token_hash,
        expires_at=refresh_token_expiry(),
        user_agent=(request.headers.get("user-agent") if request else None),
        ip=audit.client_ip(request),
    )
    session.add(stored)
    await session.flush()  # нужен stored.id для ссылки replaced_by_id

    if replaces is not None:
        replaces.revoked_at = datetime.now(UTC)
        replaces.replaced_by_id = stored.id

    return TokenPair(
        access_token=create_access_token(user.id, user.role.value),
        refresh_token=raw_refresh,
        expires_in=settings.access_token_expire_minutes * 60,
    )


async def rotate_refresh_token(
    session: AsyncSession, *, raw_token: str, request: Request | None = None
) -> tuple[User, TokenPair]:
    """Обменивает refresh-токен на новую пару, отзывая предъявленный.

    Если предъявлен уже отозванный токен, это признак кражи: отзываем всю
    цепочку сессий пользователя и требуем полноценного входа.
    """
    token_hash = hash_refresh_token(raw_token)
    stored = await session.scalar(select(RefreshToken).where(RefreshToken.token_hash == token_hash))

    if stored is None:
        raise AuthError("Некорректный refresh-токен")

    if stored.revoked_at is not None:
        await revoke_all_for_user(session, stored.user_id)
        await audit.record(
            session,
            action=AuditAction.TOKEN_REUSE_DETECTED,
            entity_type="refresh_token",
            entity_id=stored.id,
            payload={"user_id": str(stored.user_id)},
            request=request,
        )
        await session.commit()
        log.warning("refresh_token_reuse_detected", user_id=str(stored.user_id))
        raise AuthError("Сессия отозвана, требуется повторный вход")

    if stored.expires_at <= datetime.now(UTC):
        raise AuthError("Срок действия refresh-токена истёк")

    user = await session.get(User, stored.user_id)
    if user is None or not user.is_active:
        raise AuthError("Учётная запись недоступна")

    pair = await issue_token_pair(session, user, request=request, replaces=stored)
    await audit.record(
        session, action=AuditAction.TOKEN_REFRESH, user=user, request=request
    )
    return user, pair


async def revoke_refresh_token(session: AsyncSession, *, raw_token: str) -> None:
    """Выход. Отсутствующий или уже отозванный токен ошибкой не считается."""
    token_hash = hash_refresh_token(raw_token)
    await session.execute(
        update(RefreshToken)
        .where(RefreshToken.token_hash == token_hash, RefreshToken.revoked_at.is_(None))
        .values(revoked_at=datetime.now(UTC))
    )


async def revoke_all_for_user(session: AsyncSession, user_id: uuid.UUID) -> None:
    """Отзывает все активные refresh-токены пользователя (выход со всех устройств)."""
    await session.execute(
        update(RefreshToken)
        .where(RefreshToken.user_id == user_id, RefreshToken.revoked_at.is_(None))
        .values(revoked_at=datetime.now(UTC))
    )