from jose import JWTError, jwt from datetime import datetime, timedelta from fastapi import HTTPException, Depends, Request from fastapi.security import OAuth2PasswordBearer from models.auth import UserCreate, UserLogin, UserResponse, Token from storage.json_storage import JsonStorage from storage.session_storage import SessionStorage from typing import Optional import configparser import os import logging from utils.i18n import get_translations, _ from babel.support import Translations logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) CONFIG_PATH = os.getenv("GONETS_CONFIG_PATH", "/etc/gonets/config.ini") config = configparser.ConfigParser() read_files = config.read(CONFIG_PATH) if "JWT" in config: SECRET_KEY = config["JWT"]["secret_key"] elif os.getenv("SPHINX_BUILD"): logger.warning("JWT secret not found; using dummy value for documentation build") SECRET_KEY = "sphinx-placeholder-secret" else: raise RuntimeError(f"Configuration file {CONFIG_PATH} not found or missing [JWT] section") ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 30 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login") class AuthService: """Сервис для регистрации, авторизации и управления сессиями""" def __init__(self, translations: Translations = Depends(get_translations())): self.storage = JsonStorage() self.session_storage = SessionStorage() self.translations = translations def create_access_token(self, data: dict) -> str: """Создаёт токен""" to_encode = data.copy() expire = datetime.utcnow() + timedelta( minutes=ACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt def verify_token(self, token: str) -> Optional[dict]: """Проверяет токен""" try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) username: str = payload.get("sub") if username is None: return None return {"username": username} except JWTError: return None def register(self, user: UserCreate) -> UserResponse: """Регистрирует нового пользователя""" try: db_user = self.storage.create_user(user) return db_user except ValueError as e: logger.error( f"Error registering user {user.username}: {str(e)}" ) raise HTTPException(status_code=400, detail=self.translations.gettext(str(e))) except Exception as e: logger.error( f"Error registering user {user.username}: {type(e).__name__} - {str(e)}" ) raise HTTPException(status_code=500, detail=self.translations.gettext("Ошибка регистрации")) def login(self, user: UserLogin, client_id: str) -> dict: """Авторизует пользователя, создаёт сессию и возвращает токен""" try: db_user = self.storage.get_user_by_username(user.username) if not db_user: logger.error(f"User {user.username} not found") raise HTTPException(status_code=401, detail=self.translations.gettext("Неверный логин или пароль")) if not self.storage.verify_password(user.password, db_user["password_hash"]): logger.error( f"Invalid password for user {user.username}") raise HTTPException(status_code=401, detail=self.translations.gettext("Неверный логин или пароль")) access_token = self.create_access_token( data={"sub": user.username}) session_id = self.session_storage.create_session( user.username, access_token, client_id) logger.info(f"User {user.username} successfully logged in") return { "access_token": access_token, "token_type": "bearer", "session_id": session_id, "user": UserResponse(id=db_user["id"], username=db_user["username"], is_admin=db_user["is_admin"]), } except Exception as e: logger.error( f"Error logging in user {user.username}: {type(e).__name__} - {str(e)}" ) raise HTTPException(status_code=500, detail=self.translations.gettext("Ошибка авторизации")) def get_current_user( self, token: str = Depends(oauth2_scheme)) -> UserResponse: """Получает текущего пользователя по токену""" credentials = self.verify_token(token) if not credentials: logger.error("Invalid or expired token") raise HTTPException(status_code=401, detail=self.translations.gettext("Неверный токен")) username = credentials["username"] db_user = self.storage.get_user_by_username(username) if not db_user: logger.error(f"User {username} not found") raise HTTPException(status_code=401, detail=self.translations.gettext("Пользователь не найден")) return UserResponse(id=db_user["id"], username=db_user["username"], is_admin=db_user["is_admin"]) def get_user_by_session(self, client_id: str) -> dict: """Получает пользователя и токен по client_id""" session = self.session_storage.get_latest_session(client_id) if not session: logger.error( f"Session for client_id {client_id} not found or expired") raise HTTPException(status_code=401, detail=self.translations.gettext("Сессия недействительна")) username = session["username"] token = session["token"] db_user = self.storage.get_user_by_username(username) if not db_user: logger.error(f"User {username} not found") raise HTTPException(status_code=401, detail=self.translations.gettext("Пользователь не найден")) return { "user": UserResponse(id=db_user["id"], username=db_user["username"], is_admin=db_user["is_admin"]), "access_token": token } def logout(self, client_id: str): """Выполняет выход из системы, удаляя последнюю сессию для client_id""" try: session = self.session_storage.get_latest_session(client_id) if session: self.session_storage.delete_session(session["session_id"]) logger.info( f"User successfully logged out for client_id {client_id}" ) return {"message": self.translations.gettext("Успешный выход из системы")} except Exception as e: logger.error( f"Error logging out: {type(e).__name__} - {str(e)}") raise HTTPException(status_code=500, detail=self.translations.gettext("Ошибка при выходе из системы"))