app/api/endpoints/auth.py - master - pa1ch/FitAssistant

app/api/endpoints/auth.py
267 строк10 KB
Pavel Chugaev
Рефакторинг апи
14 июл 2026, 03:42
14 июл 2026, 03:42785d36b
"""Auth endpoints: Telegram Mini App initData, OIDC providers → session cookies."""

import hashlib
import hmac
import json
from urllib.parse import parse_qs, urlencode

import jwt
from fastapi import APIRouter, HTTPException, Request, Response, status
from fastapi.responses import RedirectResponse
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.api.deps import CurrentUser, SessionDep
from app.api.schemas import (
    IdentityOut,
    RefreshRequest,
    TelegramInitRequest,
    TokenResponse,
)
from app.config import settings
from app.models.user import User
from app.services import oidc
from app.services.auth import (
    ACCESS_COOKIE,
    REFRESH_COOKIE,
    IdentityConflictError,
    clear_auth_cookies,
    issue_access_token,
    issue_refresh_token,
    link_identity,
    resolve_login,
    revoke_refresh_token,
    rotate_refresh_token,
    set_auth_cookies,
)

router = APIRouter(prefix="/auth", tags=["auth"])


async def _login(session: SessionDep, response: Response, *, subject: int, name: str) -> str:
    """Resolve a Telegram login to a user, set session cookies, return the access token."""
    user = await resolve_login(session, "telegram", subject, name=name)
    access = issue_access_token(user.id)
    refresh = await issue_refresh_token(session, user.id)
    set_auth_cookies(response, access, refresh)
    return access


def _validate_init_data(init_data: str) -> dict:
    """Validate Telegram Mini App initData HMAC.

    Returns parsed data dict with at least 'user' key.
    Raises HTTPException on failure.
    """
    parsed = parse_qs(init_data, keep_blank_values=True)
    received_hash = parsed.pop("hash", [None])[0]
    if not received_hash:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing hash")

    # Build data-check-string: sorted key=value pairs separated by \n
    data_check = "\n".join(f"{k}={v[0]}" for k, v in sorted(parsed.items()))
    secret_key = hmac.new(
        b"WebAppData", settings.telegram_bot_token.encode(), hashlib.sha256
    ).digest()
    computed = hmac.new(secret_key, data_check.encode(), hashlib.sha256).hexdigest()

    if not hmac.compare_digest(computed, received_hash):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid initData signature",
        )

    return parsed


@router.post("/telegram-init", response_model=TokenResponse)
async def auth_telegram_init(body: TelegramInitRequest, session: SessionDep, response: Response):
    """Authenticate via Telegram Mini App initData."""
    parsed = _validate_init_data(body.init_data)
    user_json = parsed.get("user", [None])[0]
    if not user_json:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="No user in initData")

    tg_user = json.loads(user_json)
    telegram_id = tg_user["id"]
    name = tg_user.get("first_name", "") or tg_user.get("username", "User")

    access = await _login(session, response, subject=telegram_id, name=name)
    return TokenResponse(access_token=access)


@router.post("/refresh", response_model=TokenResponse)
async def auth_refresh(
    request: Request,
    session: SessionDep,
    response: Response,
    body: RefreshRequest | None = None,
):
    """Rotate the refresh token and issue a fresh access token.

    Web clients send no body and rely on the refresh cookie; native clients
    (no shared cookie jar with the backend) send their stored refresh token in
    the body and get both new tokens back in the JSON body instead of cookies.
    """
    from_native = bool(body and body.refresh_token)
    raw = (body.refresh_token if body else None) or request.cookies.get(REFRESH_COOKIE)
    if not raw:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="No refresh token")
    result = await rotate_refresh_token(session, raw)
    if result is None:
        clear_auth_cookies(response)
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid refresh token"
        )
    user, new_refresh = result
    access = issue_access_token(user.id)
    if from_native:
        return TokenResponse(access_token=access, refresh_token=new_refresh)
    set_auth_cookies(response, access, new_refresh)
    return TokenResponse(access_token=access)


@router.post("/logout", status_code=status.HTTP_204_NO_CONTENT)
async def auth_logout(request: Request, session: SessionDep, body: RefreshRequest | None = None):
    """Revoke the current refresh token and clear session cookies.

    Native clients pass their stored refresh token in the body (no cookie to
    read); the client-side token clearing itself happens on the device.
    """
    raw = (body.refresh_token if body else None) or request.cookies.get(REFRESH_COOKIE)
    if raw:
        await revoke_refresh_token(session, raw)
    response = Response(status_code=status.HTTP_204_NO_CONTENT)
    clear_auth_cookies(response)
    return response


@router.get("/identities", response_model=list[IdentityOut])
async def auth_identities(user: CurrentUser):
    """List the login methods linked to the current account."""
    return user.identities


@router.post("/dev-token", response_model=TokenResponse)
async def auth_dev_token(session: SessionDep, response: Response):
    """Return JWT for dev_user_id. Only works when DEV_USER_ID is set."""
    if not settings.dev_user_id:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Dev mode disabled")
    user = await session.scalar(select(User).where(User.id == settings.dev_user_id))
    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Dev user not found")
    access = issue_access_token(user.id)
    refresh = await issue_refresh_token(session, user.id)
    set_auth_cookies(response, access, refresh)
    return TokenResponse(access_token=access)


# ── OIDC / OAuth providers (Google, Yandex) ──────────────────────────────────


def _require_provider(provider: str) -> None:
    if oidc.get_client(provider) is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Unknown or unconfigured provider"
        )


def _callback_uri(provider: str) -> str:
    return f"{settings.public_base_url}/api/v1/auth/callback/{provider}"


async def _optional_current_user(request: Request, session: AsyncSession) -> User | None:
    """Resolve the logged-in user from the access cookie, if any (no error if absent)."""
    token = request.cookies.get(ACCESS_COOKIE)
    if not token:
        return None
    try:
        payload = jwt.decode(token, settings.jwt_secret, algorithms=["HS256"])
    except jwt.InvalidTokenError:
        return None
    return await session.get(User, payload.get("user_id"))


@router.get("/login/{provider}")
async def auth_oidc_login(provider: str, request: Request, client: str | None = None):
    """Start an OIDC/OAuth login (browser redirect to the provider).

    `?client=native` marks this as a Capacitor app login, opened in the system
    browser: the callback below then redirects to a deep link with the tokens
    instead of setting cookies (the system browser has no cookie jar in common
    with the app's own webview). The flag rides in the signed session cookie
    that Authlib already uses to carry the OAuth state across the redirect.
    """
    _require_provider(provider)
    request.session["oauth_native"] = client == "native"
    return await oidc.start_login(request, provider, _callback_uri(provider))


@router.get("/link/{provider}")
async def auth_oidc_link(provider: str, request: Request, user: CurrentUser):
    """Start linking another provider to the already-authenticated account."""
    _require_provider(provider)
    # Not a native login redirect — clears any stale flag from an abandoned attempt.
    request.session["oauth_native"] = False
    return await oidc.start_login(request, provider, _callback_uri(provider))


def _native_redirect(path: str, **params: str) -> RedirectResponse:
    query = urlencode(params)
    url = f"{settings.native_auth_redirect_scheme}://{path}"
    if query:
        url += f"?{query}"
    return RedirectResponse(url, status_code=303)


@router.get("/callback/{provider}")
async def auth_oidc_callback(provider: str, request: Request, session: SessionDep):
    """Handle the provider callback: link to the current user, or log in / sign up."""
    _require_provider(provider)
    # Consumed once — set by /login (or /link) at the start of this same flow.
    is_native = bool(request.session.pop("oauth_native", False))

    info = await oidc.complete_login(request, provider)
    if info is None:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="OAuth exchange failed")

    # If a valid session cookie is present, this is a "link another method" flow;
    # otherwise it's a plain login/sign-up.
    current = await _optional_current_user(request, session)
    if current is not None:
        try:
            await link_identity(
                session,
                current,
                provider,
                info.subject,
                email=info.email,
                email_verified=info.email_verified,
            )
        except IdentityConflictError:
            if is_native:
                return _native_redirect("auth-callback", error="already_linked")
            return RedirectResponse("/profile?error=already_linked", status_code=303)
        user = current
    else:
        user = await resolve_login(
            session,
            provider,
            info.subject,
            name=info.name,
            email=info.email,
            email_verified=info.email_verified,
        )

    access = issue_access_token(user.id)
    refresh = await issue_refresh_token(session, user.id)

    # The system browser that ran this flow shares no cookie jar with the
    # app's own webview, so native clients get both tokens via a deep link
    # and store them themselves instead of relying on cookies.
    if is_native:
        return _native_redirect("auth-callback", access_token=access, refresh_token=refresh)

    redirect = RedirectResponse("/", status_code=303)
    set_auth_cookies(redirect, access, refresh)
    return redirect