gonets/SRV/services/auth_service.py

from jose import JWTError, jwt
from datetime import datetime, timedelta
from fastapi import HTTPException, Depends
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 _, _load_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: Optional[Translations] = None):
        self.storage = JsonStorage()
        self.session_storage = SessionStorage()
        self.translations = translations or _load_translations("ru")

    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("Ошибка при выходе из системы"))