84 lines
3.5 KiB
Python
84 lines
3.5 KiB
Python
"""Постоянный роутер аутентификации: конфиг, Telegram-вход, выход.
|
|
|
|
Stub-вход (по нику) физически вынесен в routers/dev_auth.py и доступен только в dev.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends, Request, Response
|
|
from sqlmodel import Session
|
|
|
|
from app.auth.login import establish_session
|
|
from app.auth.registry import enabled_methods
|
|
from app.auth.telegram import TelegramProvider
|
|
from app.core import security
|
|
from app.core.config import settings
|
|
from app.core.errors import TelegramNicknameRequiredError
|
|
from app.db.session import get_session
|
|
from app.routers.users import build_me
|
|
from app.schemas import api as s
|
|
from app.services import user_service
|
|
|
|
router = APIRouter(prefix="/auth", tags=["auth"])
|
|
|
|
|
|
@router.get("/config", response_model=s.AuthConfig)
|
|
def auth_config() -> s.AuthConfig:
|
|
return s.AuthConfig(
|
|
methods=enabled_methods(),
|
|
telegram_bot_username=settings.telegram_bot_username,
|
|
)
|
|
|
|
|
|
@router.post("/telegram", response_model=s.MeRead)
|
|
def telegram_login(
|
|
body: s.TelegramAuthPayload,
|
|
request: Request,
|
|
response: Response,
|
|
session: Session = Depends(get_session),
|
|
) -> s.MeRead:
|
|
"""Вход/регистрация через Telegram.
|
|
|
|
Возвращающийся пользователь (личность уже есть) — просто входит. Первый вход —
|
|
регистрация под тегом Telegram; если тег занят/некорректен, отдаём 409
|
|
TELEGRAM_NICKNAME_REQUIRED, и фронт повторяет регистрацию через /telegram/register
|
|
с выбранным ником.
|
|
"""
|
|
identity = TelegramProvider().authenticate(body.model_dump())
|
|
user = user_service.find_by_identity(session, identity)
|
|
if user is None:
|
|
suggested = (identity.suggested_nickname or "").strip()
|
|
if not user_service.nickname_format_ok(suggested) or not user_service.nickname_available(
|
|
session, suggested
|
|
):
|
|
raise TelegramNicknameRequiredError(suggested)
|
|
user = user_service.register_from_identity(session, identity, suggested)
|
|
establish_session(session, response, request, user, identity.provider)
|
|
return build_me(session, user)
|
|
|
|
|
|
@router.post("/telegram/register", response_model=s.MeRead)
|
|
def telegram_register(
|
|
body: s.TelegramRegister,
|
|
request: Request,
|
|
response: Response,
|
|
session: Session = Depends(get_session),
|
|
) -> s.MeRead:
|
|
"""Завершение регистрации через Telegram с выбранным ником.
|
|
|
|
Подпись виджета проверяется заново (поле nickname в HMAC не входит), поэтому личность
|
|
доверенная — подделать чужой telegram_id нельзя. Если пользователь уже успел
|
|
зарегистрироваться (гонка) — просто входим.
|
|
"""
|
|
identity = TelegramProvider().authenticate(body.model_dump(exclude={"nickname"}))
|
|
user = user_service.find_by_identity(session, identity)
|
|
if user is None:
|
|
user = user_service.register_from_identity(session, identity, body.nickname)
|
|
establish_session(session, response, request, user, identity.provider)
|
|
return build_me(session, user)
|
|
|
|
|
|
@router.post("/logout", response_model=s.OkResponse)
|
|
def logout(response: Response) -> s.OkResponse:
|
|
security.clear_user_session(response)
|
|
return s.OkResponse()
|