gonets/SRV/services/auth_service.py

# services/auth_service.py
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

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

config = configparser.ConfigParser()
config_path = "/etc/gonets/config.ini"
if not os.path.exists(config_path):
    raise RuntimeError(f"Конфигурационный файл {config_path} не найден")
config.read(config_path)
SECRET_KEY = config["JWT"]["secret_key"]

ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")


class AuthService:
    """Сервис для регистрации, авторизации и управления сессиями"""

    def __init__(self):
        self.storage = JsonStorage()
        self.session_storage = SessionStorage()

    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"Ошибка при регистрации пользователя {user.username}: {str(e)}"
            )
            raise HTTPException(status_code=400, detail=str(e))
        except Exception as e:
            logger.error(
                f"Ошибка при регистрации пользователя {user.username}: {type(e).__name__} - {str(e)}"
            )
            raise HTTPException(status_code=500, detail="Ошибка регистрации")

    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.username} не найден")
                raise HTTPException(status_code=401,
                                    detail="Неверный логин или пароль")
            if not self.storage.verify_password(user.password,
                                                db_user["password_hash"]):
                logger.error(
                    f"Неверный пароль для пользователя {user.username}")
                raise HTTPException(status_code=401,
                                    detail="Неверный логин или пароль")
            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.username} успешно авторизован")
            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"Ошибка при авторизации пользователя {user.username}: {type(e).__name__} - {str(e)}"
            )
            raise HTTPException(status_code=500, detail="Ошибка авторизации")

    def get_current_user(
        self, token: str = Depends(oauth2_scheme)) -> UserResponse:
        """Получает текущего пользователя по токену"""
        credentials = self.verify_token(token)
        if not credentials:
            logger.error("Неверный или истёкший токен")
            raise HTTPException(status_code=401, detail="Неверный токен")
        username = credentials["username"]
        db_user = self.storage.get_user_by_username(username)
        if not db_user:
            logger.error(f"Пользователь {username} не найден")
            raise HTTPException(status_code=401,
                                detail="Пользователь не найден")
        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"Сессия для client_id {client_id} не найдена или истекла")
            raise HTTPException(status_code=401,
                                detail="Сессия недействительна")
        username = session["username"]
        token = session["token"]
        db_user = self.storage.get_user_by_username(username)
        if not db_user:
            logger.error(f"Пользователь {username} не найден")
            raise HTTPException(status_code=401,
                                detail="Пользователь не найден")
        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"Пользователь успешно вышел из системы для client_id {client_id}"
            )
            return {"message": "Успешный выход из системы"}
        except Exception as e:
            logger.error(
                f"Ошибка при выходе из системы: {type(e).__name__} - {str(e)}")
            raise HTTPException(status_code=500,
                                detail="Ошибка при выходе из системы")