## /user/facade.py
import os
import json
from pathlib import Path
from threading import Lock
from typing import Callable
from loguru import logger
from EJIO.utils import (
StaticClass,
Disposable,
SystemPaths,
encrypt_payload,
decrypt_payload,
generate_secure_key,
)
from .model import User
from .flags import UserRole
from .repository import UserRepository
from .media import FileFacade
from .session import UserSessionManager
from .err import (
FacadeNotInitializedError,
UserAlreadyExistsError,
UserAlreadyAuthenticatedError,
UserNotAuthenticatedError,
UserSessionNotFoundError,
UserNotFoundError,
InvalidCredentialsError,
IncorrectPasswordError,
)
__all__ = [
'UserFacade',
]
[документация]
class UserFacade(StaticClass, Disposable):
"""
Центральный фасад управления пользователями.
Отвечает за генерацию ID, соли, авторизацию и атомарное сохранение базы.
"""
__slots__ = ()
_LOCK: Lock = Lock()
_repo: UserRepository = UserRepository()
_is_initialized: bool = False
_DB_SYSTEM_KEY: str = "uXb9_Kjhd8234_llkAsD912384_jHGFD81239e12A34="
_APP_NAME: str | None = None
avatar_service: FileFacade | None = None
_db_file: Path | None = None
@classmethod
def _generate_incremental_id(cls, fill_gaps: bool = False) -> str:
"""
Вычисляет следующий инкрементальный ID.
:param fill_gaps: Если True — найдет и займет первый пропущенный (удаленный) ID.
Если False — строго выдаст max_id + 1.
"""
all_users = cls._repo.get_all_raw()
if not all_users:
return "1"
# Превращаем строковые ID в числа, которые можно отсортировать
existing_ids = sorted([int(u["id"]) for u in all_users if u["id"].isdigit()])
if not existing_ids:
return "1"
if fill_gaps:
# Алгоритм поиска дыр в числовом ряду
# Если ряд начался не с единицы (например, 1 удалили), занимаем её
if existing_ids[0] > 1:
return "1"
for i in range(len(existing_ids) - 1):
if existing_ids[i+1] - existing_ids[i] > 1:
return str(existing_ids[i] + 1)
# Обычный строгий инкремент
return str(existing_ids[-1] + 1)
[документация]
@classmethod
def initialize(cls) -> None:
with cls._LOCK:
if cls._is_initialized:
return
try:
name = SETTINGS.get('APPLICATION').get('NAME')
except Exception:
name = 'DefaultEJIOApp'
cls._APP_NAME = name#ApplicationCore.settings.get('application', {}).get('name', 'DefaultEJIOApp')
app_data = SystemPaths.app_data
app_folder = app_data / cls._APP_NAME
app_folder.mkdir(parents=True, exist_ok=True)
UserSessionManager.initialize(app_folder)
db_file = app_folder / "users.enc"
cls._db_file = db_file
# Инициализируем наш универсальный файловый фасад для аватарок!
cls.avatar_service = FileFacade(
working_dir=app_folder / "media" / "avatars",
db_path=app_folder / "media" / "avatars_registry.json",
crypto_key=cls._DB_SYSTEM_KEY,
file_base_name="avatar",
)
if db_file.exists():
try:
with open(db_file, "rb") as f:
encrypted_bytes = f.read()
decrypted_bytes = decrypt_payload(encrypted_bytes, cls._DB_SYSTEM_KEY)
raw_list = json.loads(decrypted_bytes.decode('utf-8'))
cls._repo.load_raw_list(raw_list)
except Exception as e:
logger.error(f"User database decryption error: {e}")
cls._is_initialized = True
# Инициализация дефолтного root:root, если база пустая
if cls._repo.get_by_login("root") is None:
cls._LOCK.release()
# Создаем пользователя через нашу честную фабрику
root = cls.create_user(login="root", password="root")
root.role = UserRole.ROOT
cls.emit(root)
cls._LOCK.acquire()
[документация]
@classmethod
def create_user(cls, login: str, password: str, fill_gaps: bool = False) -> User:
"""
Фабрика: принимает строго список из логина и пароля.
Просит базу выдать уникальный ID, генерирует соль и ключ ровно ОДИН раз.
"""
if not cls._is_initialized:
raise FacadeNotInitializedError("Cannot access not initialized facade.")
# Проверяем уникальность логина
if cls._repo.get_by_login(login) is not None:
raise UserAlreadyExistsError(f"A user with the login '{login}' already exists in the database.")
# 1. Строим заготовку
user = User.create_bare(login=login)
# 2. Просим БАЗУ выдать новый уникальный ID
user.id = cls._generate_incremental_id(fill_gaps=fill_gaps)
# 3. Генерируем соль и крипто-ключ ровно ОДИН раз при создании
user.salt = os.urandom(16).hex()
user.crypto_key = generate_secure_key()
# 4. Прописываем пароль (внутри свойства сработает автоматическое соление)
user.password = password
return user
[документация]
@classmethod
def authenticate(cls, login: str | None = None, password: str | None = None, remember_me: bool = False) -> User:
"""
Синхронный метод авторизации.
Проверяет логин, берет соль из БД, солит введенный пароль и сверяет хэши.
В случае успеха возвращает готовый заполненный класс User.
"""
if not cls._is_initialized:
raise FacadeNotInitializedError("Cannot access not initialized facade.")
if UserSessionManager.active_user():
raise UserAlreadyAuthenticatedError("User session already active, logout.")
# СЦЕНАРИЙ 2: АВТ ОМАТИЧЕСКИЙ ВХ ОД ПО СЕССИИ ("Запомнить меня")
if login is None and password is None:
stored_id = UserSessionManager.get_stored_user_id()
if stored_id is None:
raise UserSessionNotFoundError(
"No active session found. Explicit entry of login and password is required.."
)
user = cls._repo.get_by_id(stored_id)
if user is None:
UserSessionManager.logout()
raise UserNotFoundError("The user from the saved session has been removed from the database.")
logger.info(f"User '{user.login}' successfully logged in using a saved session.")
return user
# СЦЕНАРИЙ 1: КЛАССИЧЕСКИЙ ВХ ОД
if not login or not password:
raise InvalidCredentialsError("A login requires both a username and a password.")
user_in_db = cls._repo.get_by_login(login)
if user_in_db is None:
raise UserNotFoundError(f"User '{login}' not found.")
import hashlib
salted_payload = password + user_in_db.salt
check_hash = hashlib.sha256(salted_payload.encode('utf-8')).hexdigest()
if user_in_db.password_hash != check_hash:
raise IncorrectPasswordError("Incorrect password. Access denied.")
# Фиксируем сессию на жестком диске (создаст или зачистит файл в зависимости от флага remember_me)
UserSessionManager.save_session(user_in_db.id, remember_me)
logger.info(f"User '{user_in_db.login}' successfully logged in.")
return user_in_db
[документация]
@classmethod
def logout(cls) -> None:
"""Принудительно разлогинивает пользователя и стирает файлы сессий."""
if not UserSessionManager.active_user:
raise UserNotAuthenticatedError("User session is not active, login.")
UserSessionManager.logout()
logger.info("The user has successfully logged out, session files have been deleted..")
[документация]
@classmethod
def emit(cls, user: User) -> None:
"""Фиксирует пользователя в репозитории и сохраняет зашифрованную базу на диск."""
if not cls._is_initialized:
raise FacadeNotInitializedError("Cannot access not initialized facade.")
cls._repo.add(user)
cls.commit()
[документация]
@classmethod
def commit(cls) -> None:
if not cls._is_initialized:
raise FacadeNotInitializedError("Cannot access not initialized facade.")
db_file = cls._db_file
flat_list = cls._repo.get_all_raw()
raw_bytes = json.dumps(flat_list, ensure_ascii=False, indent=2).encode('utf-8')
encrypted = encrypt_payload(raw_bytes, cls._DB_SYSTEM_KEY)
from EJIO.utils import save_atomic
def _write(temp_path):
with open(temp_path, "wb") as f: f.write(encrypted)
with cls._LOCK:
save_atomic(str(db_file), _write)
[документация]
@classmethod
def get_by_id(cls, user_id: str) -> User | None:
if not cls._is_initialized:
raise FacadeNotInitializedError("Cannot access not initialized facade.")
return cls._repo.get_by_id(user_id)
[документация]
@classmethod
def get_all_users(cls) -> list[User]:
if not cls._is_initialized:
raise FacadeNotInitializedError("Cannot access not initialized facade.")
return cls._repo.get_all()
[документация]
@classmethod
def filter_users(cls, predicate: Callable[[User], bool]) -> list[User]:
if not cls._is_initialized:
raise FacadeNotInitializedError("Cannot access not initialized facade.")
return cls._repo.filter(predicate)
[документация]
@classmethod
def clear(cls) -> None:
with cls._LOCK:
cls._repo.clear()
UserSessionManager.clear()
if cls.avatar_service:
cls.avatar_service = None # type: ignore[assignment]
cls._APP_NAME = None
cls._db_file = None
cls._is_initialized = False