"""Постоянный роутер аутентификации: конфиг, вход по паролю, 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.password import login_player, throttle_register 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.core.security import client_ip 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 audit_service, 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("/register", response_model=s.MeRead) def password_register( body: s.PasswordRegister, request: Request, response: Response, session: Session = Depends(get_session), ) -> s.MeRead: """Регистрация по логину (нику) и паролю. Telegram привязывается позже в профиле.""" throttle_register(client_ip(request)) # против спама аккаунтов с одного IP (#62) user = user_service.register_local(session, body.nickname, body.password) audit_service.record( session, actor_id=user.id, action="create", entity_type="user", entity_id=user.id, payload={"provider": "local"}, ip=client_ip(request), user_agent=request.headers.get("user-agent"), ) establish_session(session, response, request, user, "local") return build_me(session, user) @router.post("/login", response_model=s.MeRead) def password_login( body: s.PasswordLogin, request: Request, response: Response, session: Session = Depends(get_session), ) -> s.MeRead: """Вход игрока по нику и паролю. Неверная пара — 401, перебор — 429.""" user = login_player(session, body.nickname, body.password, client_ip(request)) establish_session(session, response, request, user, "local") return build_me(session, user) @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(request: Request, response: Response) -> s.OkResponse: # Отзываем именно предъявленный токен (по jti) до его exp: украденная/оставленная # cookie перестаёт работать сразу, а не живёт до конца TTL (#57). security.revoke_session_token(request, security.USER_COOKIE, security.AUDIENCE_USER) security.clear_user_session(response) return s.OkResponse()