api/modules/auth/verification.py - master - systemsstrategyy/Treker

api/modules/auth/verification.py
93 строки4 KB
SystemsStrategy
Синхронизация с актуальной линией разработки (август 2026)
06 авг 2026, 12:52
06 авг 2026, 12:5219af6aa
"""Email-verification: запрос токена, отправка письма, consume.

Поток:
1. POST /auth/email-verification/request (auth required, без verified-email)
   → создаёт токен, отправляет email.
2. GET  /auth/email-verification/verify?token=...  (public)
   → consume токена, помечает user.email_verified_at.

Plain token живёт только в email; в БД — SHA256-хеш. TTL 24 часа.
"""

from __future__ import annotations

import hashlib
import secrets
from datetime import UTC, datetime, timedelta

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from api.core.errors import NotFoundError, ValidationError
from api.core.settings import settings
from api.modules.auth.models import EmailVerificationToken, User

TOKEN_TTL = timedelta(hours=24)


def _hash_token(plain: str) -> str:
    """SHA256 от plain token — храним только хеш в БД."""
    return hashlib.sha256(plain.encode("utf-8")).hexdigest()


class EmailVerificationService:
    @staticmethod
    async def request(db: AsyncSession, user: User) -> str:
        """Создаёт токен, отправляет email. Возвращает plain token
        (нужно тестам; в production не используется)."""
        if user.email_verified_at is not None:
            raise ValidationError("Email уже подтверждён")

        plain_token = secrets.token_urlsafe(32)
        token = EmailVerificationToken(
            user_id=user.id,
            token_hash=_hash_token(plain_token),
            expires_at=datetime.now(UTC) + TOKEN_TTL,
        )
        db.add(token)
        await db.flush()

        # Ленивая загрузка organization для шаблона.
        await db.refresh(user, attribute_names=["organization"])

        # Ссылка ведёт на ФРОНТ-страницу (не на API): юзер кликает → попадает
        # на /verify-email, которая дёргает GET .../email-verification/verify
        # и показывает результат. base_url = публичный origin приложения.
        verify_url = f"{settings.base_url}/verify-email?token={plain_token}"

        from workers.tasks.notifications import send_email_task

        await send_email_task.defer_async(
            to=user.email,
            subject="Подтверждение email",
            template="verify_email",
            context={
                "user": {"full_name": user.full_name},
                "verify_url": verify_url,
                "organization_name": user.organization.name,
            },
            user_uid=str(user.uid),
        )
        return plain_token

    @staticmethod
    async def verify(db: AsyncSession, plain_token: str) -> User:
        """Consume токена: помечает user.email_verified_at + consumed_at."""
        token_hash = _hash_token(plain_token)
        token = await db.scalar(
            select(EmailVerificationToken).where(EmailVerificationToken.token_hash == token_hash)
        )
        if token is None:
            raise NotFoundError("Токен не найден или уже использован")
        if token.consumed_at is not None:
            raise ValidationError("Токен уже использован")
        if token.expires_at < datetime.now(UTC):
            raise ValidationError("Токен истёк, запросите новый")

        user = await db.scalar(select(User).where(User.id == token.user_id))
        if user is None:
            raise NotFoundError("Пользователь не найден")

        token.consumed_at = datetime.now(UTC)
        user.email_verified_at = datetime.now(UTC)
        return user