import json
import os
import logging
from passlib.context import CryptContext
from typing import Optional
from models.auth import UserCreate, UserResponse
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
DATA_DIR = os.getenv("GONETS_DATA_DIR", "/etc/gonets")
DEFAULT_USERS_PATH = os.getenv("GONETS_USERS_PATH",
os.path.join(DATA_DIR, "users.json"))
[docs]
class JsonStorage:
"""Хранилище пользователей в JSON-файле.
Назначение:
- Обеспечивает простое файловое хранилище учетных записей без полноценной СУБД.
- Подходит для встраиваемых окружений, где критична минимальная зависимость от внешних сервисов.
Формат данных::
{
"users": [
{"id": int, "username": str, "password_hash": str, "is_admin": bool},
...
],
"next_id": int
}
Безопасность:
- Пароли никогда не сохраняются в открытом виде: используется `passlib.context.CryptContext` со схемой `bcrypt`.
- Файл создаётся с правами доступа ``0o600`` (чтение/запись только владельцу).
Ограничения:
- Нет блокировок для конкурентного доступа — в многопроцессной среде требуется внешняя синхронизация.
- Значение пути по умолчанию ``/etc/gonets/users.json`` предполагает наличие прав на запись и предварительную настройку окружения.
"""
def __init__(self, file_path: Optional[str] = None):
"""Инициализирует хранилище пользователей.
Args:
file_path (str, optional): Путь к JSON-файлу для хранения данных.
По умолчанию "/etc/gonets/users.json".
Создаёт объект CryptContext для хеширования паролей и проверяет/создаёт
файл, если он отсутствует.
"""
self.file_path = file_path or DEFAULT_USERS_PATH
self.pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
self._ensure_file_exists()
def _ensure_file_exists(self):
"""Создаёт JSON-файл, если он не существует.
Создаёт родительские директории и файл с начальной структурой данных,
устанавливая права доступа 600.
Raises:
Exception: Если произошла ошибка при создании файла или директории.
"""
try:
logger.info(f"Проверка существования файла {self.file_path}")
if not os.path.exists(self.file_path):
logger.info(f"Создание директории {os.path.dirname(self.file_path)}")
os.makedirs(os.path.dirname(self.file_path), exist_ok=True)
logger.info(f"Создание файла {self.file_path}")
with open(self.file_path, "w") as f:
json.dump({"users": [], "next_id": 1}, f)
os.chmod(self.file_path, 0o600)
logger.info(f"Создан файл {self.file_path}")
else:
logger.info(f"Файл {self.file_path} уже существует")
except Exception as e:
logger.error(
f"Ошибка при создании файла {self.file_path}: {type(e).__name__} - {str(e)}"
)
raise
def _read_data(self) -> dict:
"""Читает данные из JSON-файла.
Returns:
dict: Словарь с данными пользователей и следующим ID.
Raises:
FileNotFoundError: Если файл отсутствует (возвращает пустую структуру).
Exception: Если произошла ошибка при чтении файла.
"""
try:
with open(self.file_path, "r") as f:
return json.load(f)
except FileNotFoundError:
logger.info(
f"Файл {self.file_path} не существует, возвращаем пустую структуру"
)
return {"users": [], "next_id": 1}
except Exception as e:
logger.error(
f"Ошибка при чтении файла {self.file_path}: {type(e).__name__} - {str(e)}"
)
raise
def _write_data(self, data: dict):
"""Записывает данные в JSON-файл.
Args:
data (dict): Словарь с данными для записи (пользователи и next_id).
Raises:
Exception: Если произошла ошибка при записи в файл.
"""
try:
with open(self.file_path, "w") as f:
json.dump(data, f, indent=2)
except Exception as e:
logger.error(
f"Ошибка при записи файла {self.file_path}: {type(e).__name__} - {str(e)}"
)
raise
[docs]
def create_user(self, user: UserCreate) -> UserResponse:
"""Создаёт нового пользователя с хешированием пароля.
Args:
user (UserCreate): Объект с данными нового пользователя (username, password, is_admin).
Returns:
UserResponse: Объект с данными созданного пользователя (id, username, is_admin).
Raises:
ValueError: Если пользователь с таким именем уже существует.
Exception: Если произошла ошибка при записи данных.
"""
try:
data = self._read_data()
users = data["users"]
next_id = data["next_id"]
if any(u["username"] == user.username for u in users):
logger.error(f"Пользователь {user.username} уже существует")
raise ValueError("Пользователь уже существует")
password_hash = self.pwd_context.hash(user.password)
new_user = {
"id": next_id,
"username": user.username,
"password_hash": password_hash,
"is_admin": user.is_admin,
}
users.append(new_user)
data["next_id"] = next_id + 1
self._write_data(data)
logger.info(
f"Пользователь {user.username} успешно зарегистрирован")
return UserResponse(
id=new_user["id"],
username=new_user["username"],
is_admin=new_user["is_admin"],
)
except Exception as e:
logger.error(
f"Ошибка при создании пользователя {user.username}: {type(e).__name__} - {str(e)}"
)
raise
[docs]
def get_user_by_username(self, username: str) -> Optional[dict]:
"""Получает данные пользователя по имени.
Args:
username (str): Имя пользователя для поиска.
Returns:
Optional[dict]: Словарь с данными пользователя или None, если пользователь не найден.
Raises:
Exception: Если произошла ошибка при чтении файла.
"""
try:
if not os.path.exists(self.file_path):
logger.info(
f"Файл {self.file_path} не существует, пользователь {username} не найден"
)
return None
data = self._read_data()
for user in data["users"]:
if user["username"] == username:
return user
return None
except Exception as e:
logger.error(
f"Ошибка при получении пользователя {username}: {type(e).__name__} - {str(e)}"
)
raise
[docs]
def verify_password(self, plain_password: str, hashed_password: str) -> bool:
"""Проверяет соответствие введённого пароля хэшированному.
Args:
plain_password (str): Введённый пользователем пароль.
hashed_password (str): Хэшированный пароль из хранилища.
Returns:
bool: True, если пароль верен, False в противном случае.
"""
return self.pwd_context.verify(plain_password, hashed_password)