Files
ForbiddenStarsApp/backend/app/services/achievement_service.py
T
NotBigGhostandClaude Opus 5 9d9e4a345a Ревью: безопасность и корректность в сервисах бэкенда
Находки прохода /code-review high по backend/app:

- achievement_service: slug из URL шёл в путь без проверки, из-за чего
  DELETE /api/admin/achievements/%2E%2E удалял rmtree'ом родительскую папку
  каталога ачивок (в проде это /data — БД, uploads, ачивки целиком).
- match_service/attachment_service: версия партии = updated_at, но onupdate
  срабатывает лишь при реальном UPDATE строки matches. Правка одних участников
  и работа с вложениями его не вызывали, и оптимистичная блокировка молча
  пропускала конкурентную запись — бампаем updated_at явно.
- admin_service: удаление группы с партиями упиралось в RESTRICT и уходило
  наружу голым 500; теперь понятная ошибка. Админское удаление партии не
  чистило файлы вложений с тома — они оставались навсегда.
- user_service: при повторной загрузке аватара с тем же расширением avatar_path
  не менялся, updated_at не двигался, и кэш-бастер оставлял старую картинку до
  часа. Плюс версия считалась из наивного времени как из локального и
  разъезжалась с лидербордом, где то же поле считает SQL.
- membership_service: единственный владелец мог разжаловать сам себя и группа
  оставалась без владельца навсегда.
- notification_service: mark_read не слал SSE-сигнал, и бейдж непрочитанных на
  других устройствах висел до перезагрузки.
- routers/admin: created_at после правки пользователя отдавался без смещения,
  и дата «создан» прыгала на часовой пояс до следующего обновления списка.

#8

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0186Fk74jkkszahEHSjBzTjD
2026-09-09 15:24:52 +03:00

204 lines
7.6 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Ачивки: определения хранятся файлами на диске (settings.achievements_dir).
Папка на ачивку: achievements/<slug>/
meta.json — {"name", "description", "created_at"}
condition.py — скрипт-условие выдачи (ЗАДЕЛ: только хранится как текст, НЕ исполняется)
icon.<ext> — картинка-герб (png/jpg/webp), опционально
Источник правды — файлы. В БД (user_achievements / users.title_achievement_slug) — только
будущая выдача игрокам (сейчас не используется). Выдача и исполнение condition.py — следующий
этап; здесь админ лишь создаёт/именует/описывает ачивки.
"""
from __future__ import annotations
import json
import re
import shutil
from datetime import datetime, timezone
from pathlib import Path
from app.core.config import settings
from app.core.errors import ConflictError, NotFoundError, ValidationError
from app.services.user_service import avatar_media_type, sniff_image_ext
_CONDITION_FILE = "condition.py"
_META_FILE = "meta.json"
_CONDITION_PLACEHOLDER = (
"# Условие выдачи ачивки (ЗАДЕЛ на будущее — пока не исполняется).\n"
"# Здесь будет описана логика автоматической выдачи этого титула игроку.\n"
)
# Транслитерация для slug из русского названия.
_TRANSLIT = {
"а": "a", "б": "b", "в": "v", "г": "g", "д": "d", "е": "e", "ё": "e", "ж": "zh",
"з": "z", "и": "i", "й": "i", "к": "k", "л": "l", "м": "m", "н": "n", "о": "o",
"п": "p", "р": "r", "с": "s", "т": "t", "у": "u", "ф": "f", "х": "h", "ц": "ts",
"ч": "ch", "ш": "sh", "щ": "sch", "ъ": "", "ы": "y", "ь": "", "э": "e", "ю": "yu",
"я": "ya",
}
def _root() -> Path:
return Path(settings.achievements_dir)
# Формат slug — то, что выдаёт _slugify: только латиница, цифры и дефис.
_SLUG_RE = re.compile(r"[a-z0-9][a-z0-9-]{0,63}")
def _slugify(name: str) -> str:
text = "".join(_TRANSLIT.get(ch, ch) for ch in (name or "").strip().lower())
slug = re.sub(r"[^a-z0-9]+", "-", text).strip("-")
return slug[:48]
def _dir(slug: str) -> Path:
"""Папка ачивки. Slug приходит из URL, поэтому формат проверяем здесь: без этого
`..` или `a/b` увели бы файловые операции (вплоть до rmtree в delete) за пределы
каталога ачивок."""
if not _SLUG_RE.fullmatch(slug or ""):
raise NotFoundError("Ачивка не найдена.")
return _root() / slug
def _icon_path(slug: str) -> Path | None:
folder = _dir(slug)
if not folder.is_dir():
return None
for f in sorted(folder.glob("icon.*")):
if f.is_file():
return f
return None
def _read(slug: str) -> dict:
folder = _dir(slug)
meta_file = folder / _META_FILE
if not meta_file.is_file():
raise NotFoundError("Ачивка не найдена.")
meta = json.loads(meta_file.read_text(encoding="utf-8"))
condition = ""
cond_file = folder / _CONDITION_FILE
if cond_file.is_file():
condition = cond_file.read_text(encoding="utf-8")
icon = _icon_path(slug)
icon_url = None
if icon is not None:
version = int(icon.stat().st_mtime)
icon_url = f"/api/achievements/{slug}/icon?v={version}"
return {
"slug": slug,
"name": meta.get("name", slug),
"description": meta.get("description", ""),
"condition": condition,
"has_condition": bool(condition.strip()),
"icon_url": icon_url,
}
def list_achievements() -> list[dict]:
root = _root()
if not root.is_dir():
return []
out = []
for folder in sorted(root.iterdir()):
if folder.is_dir() and (folder / _META_FILE).is_file():
out.append(_read(folder.name))
return out
def get(slug: str) -> dict:
return _read(slug)
def _validate_name(name: str) -> str:
name = (name or "").strip()
if not (1 <= len(name) <= 64):
raise ValidationError("Название ачивки: 1–64 символа.")
return name
def create(name: str, description: str | None, condition: str | None) -> dict:
name = _validate_name(name)
slug = _slugify(name)
if not slug:
raise ValidationError("Не удалось сформировать идентификатор из названия.")
folder = _dir(slug)
if folder.exists():
raise ConflictError("Ачивка с таким названием уже существует.")
folder.mkdir(parents=True, exist_ok=True)
_write_meta(slug, name, (description or "").strip())
(folder / _CONDITION_FILE).write_text(
condition if condition else _CONDITION_PLACEHOLDER, encoding="utf-8"
)
return _read(slug)
def update(
slug: str,
*,
name: str | None = None,
description: str | None = None,
condition: str | None = None,
) -> dict:
folder = _dir(slug)
meta_file = folder / _META_FILE
if not meta_file.is_file():
raise NotFoundError("Ачивка не найдена.")
meta = json.loads(meta_file.read_text(encoding="utf-8"))
# slug не меняем (он стабильный идентификатор/путь), даже если правят название.
new_name = _validate_name(name) if name is not None else meta.get("name", slug)
new_desc = description.strip() if description is not None else meta.get("description", "")
_write_meta(slug, new_name, new_desc, created_at=meta.get("created_at"))
if condition is not None:
(folder / _CONDITION_FILE).write_text(condition, encoding="utf-8")
return _read(slug)
def set_icon(slug: str, content: bytes, ext: str) -> dict:
folder = _dir(slug)
if not (folder / _META_FILE).is_file():
raise NotFoundError("Ачивка не найдена.")
old = _icon_path(slug)
if old is not None:
try:
old.unlink()
except OSError:
pass
(folder / f"icon.{ext}").write_bytes(content)
return _read(slug)
def icon_file(slug: str) -> tuple[Path, str]:
"""Путь к файлу иконки и его media_type; NotFound, если иконки нет."""
icon = _icon_path(slug)
if icon is None:
raise NotFoundError("Иконка не найдена.")
return icon, avatar_media_type(icon.suffix.lstrip("."))
def delete(slug: str) -> None:
folder = _dir(slug)
if not folder.is_dir():
raise NotFoundError("Ачивка не найдена.")
shutil.rmtree(folder)
def validate_icon(content: bytes) -> str:
"""Проверяет, что это поддерживаемая картинка, и возвращает расширение."""
ext = sniff_image_ext(content)
if ext is None:
raise ValidationError("Поддерживаются только изображения PNG, JPEG или WebP.")
return ext
def _write_meta(slug: str, name: str, description: str, created_at: str | None = None) -> None:
meta = {
"name": name,
"description": description,
"created_at": created_at or datetime.now(timezone.utc).isoformat(),
}
(_dir(slug) / _META_FILE).write_text(
json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8"
)