import time from secrets import token_urlsafe from typing import Any, Callable import orjson from authlib.integrations.starlette_client import OAuth from authlib.oauth2.rfc7636 import create_s256_code_challenge from graphql import GraphQLResolveInfo from sqlalchemy.orm import Session from starlette.requests import Request from starlette.responses import JSONResponse, RedirectResponse from auth.tokens.storage import TokenStorage from orm.author import Author from orm.community import Community, CommunityAuthor, CommunityFollower from settings import ( FRONTEND_URL, OAUTH_CLIENTS, SESSION_COOKIE_HTTPONLY, SESSION_COOKIE_MAX_AGE, SESSION_COOKIE_NAME, SESSION_COOKIE_SAMESITE, SESSION_COOKIE_SECURE, ) from storage.db import local_session from storage.redis import redis from utils.generate_slug import generate_unique_slug from utils.logger import root_logger as logger # Type для dependency injection сессии SessionFactory = Callable[[], Session] class SessionManager: """Менеджер сессий для dependency injection с поддержкой тестирования""" def __init__(self) -> None: self._factory: SessionFactory = local_session def set_factory(self, factory: SessionFactory) -> None: """Устанавливает фабрику сессий для dependency injection""" self._factory = factory def get_session(self) -> Session: """Получает сессию БД через dependency injection""" return self._factory() # Глобальный менеджер сессий session_manager = SessionManager() def set_session_factory(factory: SessionFactory) -> None: """ Устанавливает фабрику сессий для dependency injection. Используется в тестах для подмены реальной БД на тестовую. """ session_manager.set_factory(factory) def get_session() -> Session: """ Получает сессию БД через dependency injection. Возвращает сессию которую нужно явно закрывать после использования. Внимание: не забывайте закрывать сессию после использования! Рекомендуется использовать try/finally блок. """ return session_manager.get_session() oauth = OAuth() # OAuth state management через Redis (TTL 10 минут) OAUTH_STATE_TTL = 600 # 10 минут # Конфигурация провайдеров для регистрации PROVIDER_CONFIGS = { "google": { "server_metadata_url": "https://accounts.google.com/.well-known/openid-configuration", "client_kwargs": { "scope": "openid email profile", }, }, "github": { "access_token_url": "https://github.com/login/oauth/access_token", "authorize_url": "https://github.com/login/oauth/authorize", "api_base_url": "https://api.github.com/", "client_kwargs": { "scope": "read:user user:email", }, }, "facebook": { "access_token_url": "https://graph.facebook.com/v18.0/oauth/access_token", "authorize_url": "https://www.facebook.com/v18.0/dialog/oauth", "api_base_url": "https://graph.facebook.com/", "scope": "email public_profile", # Явно указываем необходимые scope }, "x": { "access_token_url": "https://api.twitter.com/2/oauth2/token", "authorize_url": "https://twitter.com/i/oauth2/authorize", "api_base_url": "https://api.twitter.com/2/", "client_kwargs": { "scope": "tweet.read users.read", # Базовые scope для X API v2 }, }, "telegram": { "access_token_url": "https://oauth.telegram.org/auth/request", "authorize_url": "https://oauth.telegram.org/auth", "api_base_url": "https://api.telegram.org/", "client_kwargs": { "scope": "read", # Базовый scope для Telegram }, }, "vk": { "access_token_url": "https://oauth.vk.com/access_token", "authorize_url": "https://oauth.vk.com/authorize", "api_base_url": "https://api.vk.com/method/", "client_kwargs": { "scope": "email", # Минимальный scope для получения email }, }, "yandex": { "access_token_url": "https://oauth.yandex.ru/token", "authorize_url": "https://oauth.yandex.ru/authorize", "api_base_url": "https://login.yandex.ru/info", "client_kwargs": { "scope": "login:email login:info login:avatar", # Scope для получения профиля }, }, } # Константы для генерации временного email TEMP_EMAIL_SUFFIX = "@oauth.local" def _generate_temp_email(provider: str, user_id: str) -> str: """Генерирует временный email для OAuth провайдеров без email""" return f"{provider}_{user_id}@oauth.local" def _register_oauth_provider(provider: str, client_config: dict) -> None: """Регистрирует OAuth провайдер в зависимости от его типа""" try: provider_config = PROVIDER_CONFIGS.get(provider, {}) if not provider_config: logger.warning(f"Unknown OAuth provider: {provider}") return # 🔍 Отладочная информация logger.info( f"Registering OAuth provider {provider} with client_id: {client_config['id'][:8] if client_config['id'] else 'EMPTY'}..." ) # Базовые параметры для всех провайдеров register_params: dict[str, Any] = { "name": provider, "client_id": client_config["id"], "client_secret": client_config["key"], } # Добавляем конфигурацию провайдера с явной типизацией if isinstance(provider_config, dict): register_params.update(provider_config) # 🔒 Для Facebook добавляем дополнительные параметры безопасности if provider == "facebook": register_params.update( { "client_kwargs": { "scope": "email public_profile", "token_endpoint_auth_method": "client_secret_post", } } ) oauth.register(**register_params) logger.info(f"OAuth provider {provider} registered successfully") # 🔍 Проверяем что клиент действительно создался test_client = oauth.create_client(provider) if test_client: logger.info(f"OAuth client {provider} created successfully") else: logger.error(f"OAuth client {provider} failed to create after registration") except Exception as e: logger.error(f"Failed to register OAuth provider {provider}: {e}") # 🔍 Диагностика OAuth конфигурации logger.info(f"Available OAuth providers in config: {list(PROVIDER_CONFIGS.keys())}") logger.info(f"Available OAuth clients: {list(OAUTH_CLIENTS.keys())}") for provider in PROVIDER_CONFIGS: if provider.upper() in OAUTH_CLIENTS: client_config = OAUTH_CLIENTS[provider.upper()] # 🔍 Проверяем что id и key не пустые client_id = client_config.get("id", "").strip() client_key = client_config.get("key", "").strip() logger.info( f"OAuth provider {provider}: id={'SET' if client_id else 'EMPTY'}, key={'SET' if client_key else 'EMPTY'}" ) if client_id and client_key: _register_oauth_provider(provider, client_config) else: logger.warning(f"OAuth provider {provider} skipped: id={bool(client_id)}, key={bool(client_key)}") else: logger.warning(f"OAuth provider {provider} not found in OAUTH_CLIENTS") # Провайдеры со специальной обработкой данных PROVIDER_HANDLERS = { "google": lambda token, _: { "id": token.get("userinfo", {}).get("sub"), "email": token.get("userinfo", {}).get("email"), "name": token.get("userinfo", {}).get("name"), "picture": token.get("userinfo", {}).get("picture", "").replace("=s96", "=s600"), }, "telegram": lambda token, _: { "id": str(token.get("id", "")), "email": None, "phone": str(token.get("phone_number", "")), "name": token.get("first_name", "") + " " + token.get("last_name", ""), "picture": token.get("photo_url"), }, "x": lambda _, profile_data: { "id": profile_data.get("data", {}).get("id"), "email": None, "name": profile_data.get("data", {}).get("name") or profile_data.get("data", {}).get("username"), "picture": profile_data.get("data", {}).get("profile_image_url", "").replace("_normal", "_400x400"), }, } async def _fetch_github_profile(client: Any, token: Any) -> dict: """Получает профиль из GitHub API""" try: # Получаем основной профиль profile = await client.get("user", token=token) profile_data = profile.json() # Проверяем наличие ошибок в ответе GitHub if "message" in profile_data: logger.error(f"GitHub API error: {profile_data['message']}") return {} # Получаем email адреса (требует scope user:email) emails = await client.get("user/emails", token=token) emails_data = emails.json() # Ищем основной email primary_email = None if isinstance(emails_data, list): primary_email = next((email["email"] for email in emails_data if email.get("primary")), None) return { "id": str(profile_data.get("id", "")), "email": primary_email or profile_data.get("email"), "name": profile_data.get("name") or profile_data.get("login", ""), "picture": profile_data.get("avatar_url"), } except Exception as e: logger.error(f"Error fetching GitHub profile: {e}") return {} async def _fetch_facebook_profile(client: Any, token: Any) -> dict: """Получает профиль из Facebook API""" try: # Используем актуальную версию API v18.0+ и расширенные поля profile = await client.get("me?fields=id,name,email,picture.width(600).height(600)", token=token) profile_data = profile.json() # Проверяем наличие ошибок в ответе Facebook if "error" in profile_data: logger.error(f"Facebook API error: {profile_data['error']}") return {} return { "id": str(profile_data.get("id", "")), "email": profile_data.get("email"), # Может быть None если не предоставлен "name": profile_data.get("name", ""), "picture": profile_data.get("picture", {}).get("data", {}).get("url"), } except Exception as e: logger.error(f"Error fetching Facebook profile: {e}") return {} async def _fetch_x_profile(client: Any, token: Any) -> dict: """Получает профиль из X (Twitter) API""" try: # Используем правильный endpoint для X API v2 profile = await client.get("users/me?user.fields=id,name,username,profile_image_url", token=token) profile_data = profile.json() # Проверяем наличие ошибок в ответе X if "errors" in profile_data: logger.error(f"X API error: {profile_data['errors']}") return {} return PROVIDER_HANDLERS["x"](token, profile_data) except Exception as e: logger.error(f"Error fetching X profile: {e}") return {} async def _fetch_vk_profile(client: Any, token: Any) -> dict: """Получает профиль из VK API""" try: # Используем актуальную версию API v5.199+ profile = await client.get("users.get?fields=photo_400_orig,contacts&v=5.199", token=token) profile_data = profile.json() # Проверяем наличие ошибок в ответе VK if "error" in profile_data: logger.error(f"VK API error: {profile_data['error']}") return {} if profile_data.get("response"): user_data = profile_data["response"][0] return { "id": str(user_data["id"]), "email": user_data.get("contacts", {}).get("email"), "name": f"{user_data.get('first_name', '')} {user_data.get('last_name', '')}".strip(), "picture": user_data.get("photo_400_orig"), } return {} except Exception as e: logger.error(f"Error fetching VK profile: {e}") return {} async def _fetch_yandex_profile(client: Any, token: Any) -> dict: """Получает профиль из Yandex API""" profile = await client.get("?format=json", token=token) profile_data = profile.json() return { "id": profile_data.get("id"), "email": profile_data.get("default_email"), "name": profile_data.get("display_name") or profile_data.get("real_name"), "picture": f"https://avatars.yandex.net/get-yapic/{profile_data.get('default_avatar_id')}/islands-200" if profile_data.get("default_avatar_id") else None, } async def get_user_profile(provider: str, client: Any, token: Any) -> dict: """Получает профиль пользователя от провайдера OAuth""" # Простые провайдеры с обработкой через lambda if provider in PROVIDER_HANDLERS: return PROVIDER_HANDLERS[provider](token, None) # Провайдеры требующие API вызовов profile_fetchers = { "github": _fetch_github_profile, "facebook": _fetch_facebook_profile, "x": _fetch_x_profile, "vk": _fetch_vk_profile, "yandex": _fetch_yandex_profile, } if provider in profile_fetchers: return await profile_fetchers[provider](client, token) return {} async def oauth_login(_: None, _info: GraphQLResolveInfo, provider: str, callback_data: dict[str, Any]) -> JSONResponse: """ Обработка OAuth авторизации Args: provider: Провайдер OAuth (google, github, etc.) callback_data: Данные из callback-а Returns: dict: Результат авторизации с токеном или ошибкой """ if provider not in PROVIDER_CONFIGS: return JSONResponse({"error": "Invalid provider"}, status_code=400) client = oauth.create_client(provider) if not client: logger.error(f"OAuth client for {provider} not found. Available clients: {list(oauth._clients.keys())}") return JSONResponse({"error": "Provider not configured"}, status_code=400) # Получаем параметры из query string state = callback_data.get("state") redirect_uri = callback_data.get("redirect_uri", FRONTEND_URL) if not state: return JSONResponse({"error": "State parameter is required"}, status_code=400) # Генерируем PKCE challenge code_verifier = token_urlsafe(32) code_challenge = create_s256_code_challenge(code_verifier) # Сохраняем состояние OAuth в Redis oauth_data = { "code_verifier": code_verifier, "provider": provider, "redirect_uri": redirect_uri, "created_at": int(time.time()), } await store_oauth_state(state, oauth_data) # Используем URL из фронтенда для callback - для фронтенда oauth_callback_uri = f"{callback_data['base_url']}oauth/{provider}/callback" try: return await client.authorize_redirect( callback_data["request"], oauth_callback_uri, code_challenge=code_challenge, code_challenge_method="S256", state=state, ) except Exception as e: logger.error(f"OAuth redirect error for {provider}: {e!s}") return JSONResponse({"error": str(e)}, status_code=500) async def oauth_callback(request: Any) -> JSONResponse | RedirectResponse: """ Обработчик OAuth callback. Создает или обновляет пользователя и устанавливает сессионный токен. """ try: provider = request.path_params.get("provider") if not provider: return JSONResponse({"error": "Provider not specified"}, status_code=400) # Получаем OAuth клиента client = oauth.create_client(provider) if not client: return JSONResponse({"error": "Invalid provider"}, status_code=400) # Получаем токен token = await client.authorize_access_token(request) if not token: return JSONResponse({"error": "Failed to get access token"}, status_code=400) # Получаем профиль пользователя profile = await get_user_profile(provider, client, token) if not profile: return JSONResponse({"error": "Failed to get user profile"}, status_code=400) # Создаем или обновляем пользователя author = await _create_or_update_user(provider, profile) if not author: return JSONResponse({"error": "Failed to create/update user"}, status_code=500) # Создаем сессию session_token = await TokenStorage.create_session( str(author.id), auth_data={ "provider": provider, "profile": profile, }, username=author.name if isinstance(author.name, str) else str(author.name) if author.name is not None else None, device_info={ "user_agent": request.headers.get("user-agent"), "ip": request.client.host if hasattr(request, "client") and request.client else None, }, ) # Получаем state из Redis для редиректа state = request.query_params.get("state") state_data = await get_oauth_state(state) if state else None redirect_uri = state_data.get("redirect_uri") if state_data else FRONTEND_URL if not isinstance(redirect_uri, str) or not redirect_uri: redirect_uri = FRONTEND_URL # Создаем ответ с редиректом response = RedirectResponse(url=str(redirect_uri)) # Устанавливаем cookie с сессией response.set_cookie( SESSION_COOKIE_NAME, session_token, httponly=SESSION_COOKIE_HTTPONLY, secure=SESSION_COOKIE_SECURE, samesite=SESSION_COOKIE_SAMESITE, max_age=SESSION_COOKIE_MAX_AGE, path="/", # Важно: устанавливаем path="/" для доступности cookie во всех путях ) logger.info(f"OAuth успешно завершен для {provider}, user_id={author.id}") return response except Exception as e: logger.error(f"OAuth callback error: {e!s}") # В случае ошибки редиректим на фронтенд с ошибкой fallback_redirect = request.query_params.get("redirect_uri", FRONTEND_URL) return RedirectResponse(url=f"{fallback_redirect}?error=auth_failed") async def store_oauth_state(state: str, data: dict) -> None: """Сохраняет OAuth состояние в Redis с TTL""" key = f"oauth_state:{state}" await redis.execute("SETEX", key, OAUTH_STATE_TTL, orjson.dumps(data)) async def get_oauth_state(state: str) -> dict | None: """Получает и удаляет OAuth состояние из Redis (one-time use)""" key = f"oauth_state:{state}" data = await redis.execute("GET", key) if data: await redis.execute("DEL", key) # Одноразовое использование return orjson.loads(data) return None # HTTP handlers для тестирования async def oauth_login_http(request: Request) -> JSONResponse | RedirectResponse: """HTTP handler для OAuth login""" try: provider = request.path_params.get("provider") if not provider or provider not in PROVIDER_CONFIGS: return JSONResponse({"error": "Invalid provider"}, status_code=400) client = oauth.create_client(provider) if not client: logger.error(f"OAuth client for {provider} not found. Available clients: {list(oauth._clients.keys())}") return JSONResponse({"error": "Provider not configured"}, status_code=400) # Генерируем PKCE challenge code_verifier = token_urlsafe(32) code_challenge = create_s256_code_challenge(code_verifier) state = token_urlsafe(32) # 🔍 Сохраняем состояние OAuth только в Redis (убираем зависимость от request.session) oauth_data = { "code_verifier": code_verifier, "provider": provider, "redirect_uri": FRONTEND_URL, "created_at": int(time.time()), } await store_oauth_state(state, oauth_data) # URL для callback - для фронтенда callback_uri = f"{FRONTEND_URL}oauth/{provider}/callback" # 🔍 Создаем redirect URL вручную (обходим использование request.session в authlib) # VK не поддерживает PKCE, используем code_challenge только для поддерживающих провайдеров if provider in ["vk", "yandex", "telegram"]: # Провайдеры без PKCE поддержки authorization_url = await client.create_authorization_url( callback_uri, state=state, ) else: # Провайдеры с PKCE поддержкой (Google, GitHub, Facebook, X) authorization_url = await client.create_authorization_url( callback_uri, code_challenge=code_challenge, code_challenge_method="S256", state=state, ) return RedirectResponse(url=authorization_url["url"], status_code=302) except Exception as e: logger.error(f"OAuth login error: {e}") return JSONResponse({"error": "OAuth login failed"}, status_code=500) async def oauth_callback_http(request: Request) -> JSONResponse | RedirectResponse: """HTTP handler для OAuth callback""" try: # 🔍 Получаем состояние OAuth только из Redis (убираем зависимость от request.session) state = request.query_params.get("state") if not state: return JSONResponse({"error": "Missing OAuth state parameter"}, status_code=400) oauth_data = await get_oauth_state(state) if not oauth_data: return JSONResponse({"error": "Invalid or expired OAuth state"}, status_code=400) provider = oauth_data.get("provider") if not provider: return JSONResponse({"error": "No provider in OAuth state"}, status_code=400) # Дополнительная проверка провайдера из path параметров (для старого формата) provider_from_path = request.path_params.get("provider") if provider_from_path and provider_from_path != provider: return JSONResponse({"error": "Provider mismatch"}, status_code=400) # Используем существующую логику client = oauth.create_client(provider) if not client: return JSONResponse({"error": "Provider not configured"}, status_code=400) # Получаем authorization code из query параметров code = request.query_params.get("code") if not code: return JSONResponse({"error": "Missing authorization code"}, status_code=400) # 🔍 Обмениваем code на токен - с PKCE или без в зависимости от провайдера if provider in ["vk", "yandex", "telegram"]: # Провайдеры без PKCE поддержки token = await client.fetch_access_token( authorization_response=str(request.url), ) else: # Провайдеры с PKCE поддержкой code_verifier = oauth_data.get("code_verifier") if not code_verifier: return JSONResponse({"error": "Missing code verifier in OAuth state"}, status_code=400) token = await client.fetch_access_token( authorization_response=str(request.url), code_verifier=code_verifier, ) if not token: return JSONResponse({"error": "Failed to get access token"}, status_code=400) profile = await get_user_profile(provider, client, token) if not profile: return JSONResponse({"error": "Failed to get user profile"}, status_code=400) # Создаем или обновляем пользователя используя helper функцию author = await _create_or_update_user(provider, profile) if not author: return JSONResponse({"error": "Failed to create/update user"}, status_code=500) # Создаем токен сессии с полными данными session_token = await TokenStorage.create_session( str(author.id), auth_data={ "provider": provider, "profile": profile, }, username=author.name if isinstance(author.name, str) else str(author.name) if author.name is not None else None, device_info={ "user_agent": request.headers.get("user-agent"), "ip": request.client.host if hasattr(request, "client") and request.client else None, }, ) # Получаем redirect_uri из OAuth данных redirect_uri = oauth_data.get("redirect_uri", FRONTEND_URL) if not isinstance(redirect_uri, str) or not redirect_uri: redirect_uri = FRONTEND_URL # Возвращаем redirect с cookie response = RedirectResponse(url=str(redirect_uri), status_code=307) response.set_cookie( SESSION_COOKIE_NAME, session_token, httponly=SESSION_COOKIE_HTTPONLY, secure=SESSION_COOKIE_SECURE, samesite=SESSION_COOKIE_SAMESITE, max_age=SESSION_COOKIE_MAX_AGE, path="/", # Важно: устанавливаем path="/" для доступности cookie во всех путях ) logger.info(f"OAuth успешно завершен для {provider}, user_id={author.id}") return response except Exception as e: logger.error(f"OAuth callback error: {e!s}") # В случае ошибки редиректим на фронтенд с ошибкой fallback_redirect = request.query_params.get("redirect_uri", FRONTEND_URL) return RedirectResponse(url=f"{fallback_redirect}?error=auth_failed") async def _create_or_update_user(provider: str, profile: dict) -> Author: """ Создает или обновляет пользователя на основе OAuth профиля. Возвращает объект Author. """ # Для некоторых провайдеров (X, Telegram) email может отсутствовать email = profile.get("email") if not email: # Генерируем временный email на основе провайдера и ID email = _generate_temp_email(provider, profile.get("id", "unknown")) logger.info(f"Generated temporary email for {provider} user: {email}") # Создаем или обновляем пользователя session = get_session() try: # Сначала ищем пользователя по OAuth author = Author.find_by_oauth(provider, profile["id"], session) if author: # Пользователь найден по OAuth - обновляем данные author.set_oauth_account(provider, profile["id"], email=profile.get("email")) _update_author_profile(author, profile) else: # Ищем пользователя по email если есть настоящий email author = None if email and not email.endswith(TEMP_EMAIL_SUFFIX): author = session.query(Author).where(Author.email == email).first() if author: # Пользователь найден по email - добавляем OAuth данные author.set_oauth_account(provider, profile["id"], email=profile.get("email")) _update_author_profile(author, profile) else: # Создаем нового пользователя author = _create_new_oauth_user(provider, profile, email, session) session.commit() return author finally: session.close() def _update_author_profile(author: Author, profile: dict) -> None: """Обновляет профиль автора данными из OAuth""" if profile.get("name") and not author.name: author.name = profile["name"] # type: ignore[assignment] if profile.get("picture") and not author.pic: author.pic = profile["picture"] # type: ignore[assignment] author.updated_at = int(time.time()) # type: ignore[assignment] author.last_seen = int(time.time()) # type: ignore[assignment] def _create_new_oauth_user(provider: str, profile: dict, email: str, session: Any) -> Author: """Создает нового пользователя из OAuth профиля""" slug = generate_unique_slug(profile["name"] or f"{provider}_{profile.get('id', 'user')}") author = Author( email=email, name=profile["name"] or f"{provider.title()} User", slug=slug, pic=profile.get("picture"), email_verified=bool(profile.get("email")), created_at=int(time.time()), updated_at=int(time.time()), last_seen=int(time.time()), ) session.add(author) session.flush() # Получаем ID автора # Добавляем OAuth данные для нового пользователя author.set_oauth_account(provider, profile["id"], email=profile.get("email")) # Добавляем пользователя в основное сообщество с дефолтными ролями target_community_id = 1 # Основное сообщество # Получаем сообщество для назначения дефолтных ролей community = session.query(Community).where(Community.id == target_community_id).first() if community: default_roles = community.get_default_roles() # Проверяем, не существует ли уже запись CommunityAuthor existing_ca = ( session.query(CommunityAuthor).filter_by(community_id=target_community_id, author_id=author.id).first() ) if not existing_ca: # Создаем CommunityAuthor с дефолтными ролями community_author = CommunityAuthor( community_id=target_community_id, author_id=author.id, roles=",".join(default_roles) ) session.add(community_author) logger.info(f"Создана запись CommunityAuthor для OAuth пользователя {author.id} с ролями: {default_roles}") # Проверяем, не существует ли уже запись подписчика existing_follower = ( session.query(CommunityFollower).filter_by(community=target_community_id, follower=int(author.id)).first() ) if not existing_follower: # Добавляем пользователя в подписчики сообщества follower = CommunityFollower(community=target_community_id, follower=int(author.id)) session.add(follower) logger.info(f"OAuth пользователь {author.id} добавлен в подписчики сообщества {target_community_id}") return author