anti-plagiarism/services/api/app/core/oauth.py

"""OAuth2 authorization-code flow для Google и Яндекс — без authlib, только httpx.

Единый интерфейс для обоих провайдеров: get_authorize_url() формирует ссылку на
экран согласия, exchange_code() меняет code на профиль пользователя. Асинхронно
(httpx.AsyncClient) — синхронные вызовы блокировали бы event loop всего API на
время внешнего запроса к Google/Яндекс.
"""

from dataclasses import dataclass
from urllib.parse import urlencode

import httpx

from app.config import settings

_TIMEOUT = 10.0


class OAuthNotConfigured(Exception):
    """Провайдер не настроен (пустой client_id/secret в конфиге)."""


@dataclass
class OAuthUserInfo:
    provider_id: str
    email: str
    name: str
    email_verified: bool


def _redirect_uri(provider: str) -> str:
    return f"{settings.APP_URL}/api/auth/{provider}/callback"


def _credentials(provider: str) -> tuple[str, str]:
    if provider == "google":
        client_id, secret = settings.GOOGLE_CLIENT_ID, settings.GOOGLE_CLIENT_SECRET
    elif provider == "yandex":
        client_id, secret = settings.YANDEX_CLIENT_ID, settings.YANDEX_CLIENT_SECRET
    else:
        raise ValueError(f"неизвестный OAuth-провайдер: {provider!r}")
    if not client_id or not secret:
        raise OAuthNotConfigured(provider)
    return client_id, secret


def get_authorize_url(provider: str, state: str) -> str:
    """Собрать ссылку на экран согласия провайдера."""
    client_id, _ = _credentials(provider)
    redirect_uri = _redirect_uri(provider)

    if provider == "google":
        params = {
            "client_id": client_id,
            "redirect_uri": redirect_uri,
            "response_type": "code",
            "scope": "openid email profile",
            "state": state,
            "prompt": "select_account",
        }
        return f"https://accounts.google.com/o/oauth2/v2/auth?{urlencode(params)}"

    params = {
        "response_type": "code",
        "client_id": client_id,
        "redirect_uri": redirect_uri,
        "state": state,
    }
    return f"https://oauth.yandex.ru/authorize?{urlencode(params)}"


async def exchange_code(provider: str, code: str) -> OAuthUserInfo:
    """Обменять authorization code на профиль пользователя у провайдера."""
    client_id, client_secret = _credentials(provider)
    redirect_uri = _redirect_uri(provider)

    async with httpx.AsyncClient(timeout=_TIMEOUT) as client:
        if provider == "google":
            token_resp = await client.post(
                "https://oauth2.googleapis.com/token",
                data={
                    "code": code,
                    "client_id": client_id,
                    "client_secret": client_secret,
                    "redirect_uri": redirect_uri,
                    "grant_type": "authorization_code",
                },
            )
            token_resp.raise_for_status()
            access_token = token_resp.json()["access_token"]

            info_resp = await client.get(
                "https://www.googleapis.com/oauth2/v3/userinfo",
                headers={"Authorization": f"Bearer {access_token}"},
            )
            info_resp.raise_for_status()
            info = info_resp.json()
            email = info["email"]
            return OAuthUserInfo(
                provider_id=info["sub"],
                email=email,
                name=info.get("name") or email.split("@")[0],
                email_verified=bool(info.get("email_verified", False)),
            )

        # yandex
        token_resp = await client.post(
            "https://oauth.yandex.ru/token",
            data={
                "grant_type": "authorization_code",
                "code": code,
                "client_id": client_id,
                "client_secret": client_secret,
            },
        )
        token_resp.raise_for_status()
        access_token = token_resp.json()["access_token"]

        info_resp = await client.get(
            "https://login.yandex.ru/info",
            params={"format": "json"},
            headers={"Authorization": f"OAuth {access_token}"},
        )
        info_resp.raise_for_status()
        info = info_resp.json()

        email = info.get("default_email") or next(iter(info.get("emails") or []), None)
        if not email:
            raise ValueError("Яндекс не вернул email — нужно разрешение на доступ к почте")

        return OAuthUserInfo(
            provider_id=str(info["id"]),
            email=email,
            name=info.get("real_name") or info.get("display_name") or email.split("@")[0],
            email_verified=True,  # Яндекс отдаёт только подтверждённые адреса
        )