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 = configparser.ConfigParser()
config_path = "/etc/gonets/config.ini"
if not os.path.exists(config_path):
raise RuntimeError(f"Configuration file {config_path} not found")
config.read(config_path)
SECRET_KEY = config["JWT"]["secret_key"]
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
[docs]
class AuthService:
"""Сервис для регистрации, авторизации и управления сессиями"""
def __init__(self, translations: Translations = Depends(get_translations())):
self.storage = JsonStorage()
self.session_storage = SessionStorage()
self.translations = translations
[docs]
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
[docs]
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
[docs]
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("Ошибка регистрации"))
[docs]
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("Ошибка авторизации"))
[docs]
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"])
[docs]
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
}
[docs]
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("Ошибка при выходе из системы"))