Исходный код EJIO.user.session
## /user/session.py
import json
from pathlib import Path
from loguru import logger
from EJIO.utils import (
StaticClass,
Disposable,
encrypt_payload,
decrypt_payload,
generate_secure_key,
)
from .err import ManagerNotInitializedError
from .model import User
__all__ = [
'UserSessionManager',
]
[документация]
class UserSessionManager(StaticClass, Disposable):
"""
Статический менеджер пользовательских сессий ("Запомнить меня").
Записывает зашифрованный токен авторизации на жесткий диск.
"""
__slots__ = ()
_active_user_id: str | None = None
_is_remembered: bool = False
_is_initialized: bool = False
_SESSION_CRYPTO_KEY: str = "uXb9_Kjhd8234_llkAsD912384_jHGFD81239e12A34="
_session_file: Path | None = None
[документация]
@classmethod
def active_user(cls) -> User | None:
from .facade import UserFacade
return UserFacade.get_by_id(cls._active_user_id) if cls._active_user_id else None
[документация]
@classmethod
def initialize(cls, app_path: Path) -> None:
cls._session_file = app_path / "session.enc"
cls._is_initialized = True
[документация]
@classmethod
def save_session(cls, user_id: str, remember_me: bool) -> None:
"""Сохраняет токен сессии на диск, если пользователь поставил галочку."""
if not cls._is_initialized:
raise ManagerNotInitializedError("Cannot access not initialized manager.")
cls._active_user_id = user_id
cls._is_remembered = remember_me
session_file = cls._session_file
# Если галочка "Запомнить меня" не стоит, зачищаем старый файл с диска
if not remember_me:
session_file.unlink(missing_ok=True)
return
try:
# Генерируем случайный токен сессии (для будущих расширений валидации токенов)
session_token = generate_secure_key()
payload = {
"user_id": user_id,
"token": session_token
}
raw_bytes = json.dumps(payload).encode('utf-8')
encrypted = encrypt_payload(raw_bytes, cls._SESSION_CRYPTO_KEY)
from EJIO.utils import save_atomic
def _write(temp_path):
with open(temp_path, "wb") as f: f.write(encrypted)
save_atomic(str(session_file), _write)
logger.debug(f"Сессия пользователя ID '{user_id}' успешно сохранена на диск.")
except Exception as e:
logger.error(f"Не удалось записать файл сессии на диск: {e}")
[документация]
@classmethod
def get_stored_user_id(cls) -> str | None:
"""Считывает зашифрованный файл сессии и возвращает ID пользователя, если сессия валидна."""
if not cls._is_initialized:
raise ManagerNotInitializedError("Cannot access not initialized manager.")
session_file = cls._session_file
if not session_file.exists():
return None
try:
with open(session_file, "rb") as f:
encrypted_bytes = f.read()
if not encrypted_bytes:
return None
decrypted_bytes = decrypt_payload(encrypted_bytes, cls._SESSION_CRYPTO_KEY)
payload = json.loads(decrypted_bytes.decode('utf-8'))
cls._active_user_id = payload.get("user_id")
cls._is_remembered = True
return cls._active_user_id
except Exception as e:
logger.error(f"Файл сессии поврежден или неверный ключ: {e}")
session_file.unlink(missing_ok=True)
return None
[документация]
@classmethod
def logout(cls) -> None:
"""Разлогинивает пользователя и стирает сессию с диска."""
if not cls._is_initialized:
raise ManagerNotInitializedError("Cannot access not initialized manager.")
cls._active_user_id = None
cls._is_remembered = False
cls._session_file.unlink(missing_ok=True)
[документация]
@classmethod
def clear(cls) -> None:
cls._active_user_id = None
if not cls._is_remembered:
cls.logout()
cls._is_initialized = False
cls._session_file = None