Исходный код 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