Админка закрылась входом через Telegram
Подпись Telegram-логина проверяется токеном бота, а доступ есть только у записей в bot.admin_users: роль перечитывается из БД на каждом запросе, поэтому выключение пользователя сразу закрывает доступ. Сессия — подписанная HttpOnly-cookie на 12 часов, все формы получили CSRF-токен, первые администраторы засеваются из ADMIN_IDS при старте. Домены в BotFather уже заданы, но NPM и публикацию не включаем — снаружи админка по-прежнему только на loopback.
This commit is contained in:
parent
d695f1991f
commit
f345844dfe
|
|
@ -5,6 +5,12 @@ APP_ENV=dev
|
|||
|
||||
# Токен бота (получается у @BotFather)
|
||||
BOT_TOKEN=
|
||||
# Username бота без @ (для кнопки Telegram-входа в админку)
|
||||
BOT_USERNAME=
|
||||
# Публичный HTTPS-адрес админки (для Telegram Login Widget)
|
||||
ADMIN_PUBLIC_URL=http://localhost:8000
|
||||
# Секрет подписи cookie-сессии: openssl rand -hex 32 (в проде обязателен)
|
||||
ADMIN_SESSION_SECRET=
|
||||
|
||||
# Postgres
|
||||
DB_HOST=localhost
|
||||
|
|
|
|||
19
DEPLOY.md
19
DEPLOY.md
|
|
@ -17,6 +17,8 @@ cd DispelEvilBot
|
|||
cp env/test.env.example env/test.env
|
||||
# заполнить: BOT_TOKEN, DB_PASSWORD, CHAT_ID (основной чат), ALERT_CHAT_ID (опц.),
|
||||
# ADMIN_IDS (опц.), CLICKHOUSE_PASSWORD (аналитика событий)
|
||||
# плюс для админки: BOT_USERNAME, ADMIN_PUBLIC_URL (HTTPS-домен),
|
||||
# ADMIN_SESSION_SECRET (openssl rand -hex 32)
|
||||
```
|
||||
|
||||
Порт ClickHouse на хосте (`CLICKHOUSE_HTTP_PORT`) у контуров должен различаться:
|
||||
|
|
@ -100,6 +102,23 @@ curl -s http://127.0.0.1:8101/healthz # prod
|
|||
Локально без Docker: `pip install -r requirements-admin.txt && python -m admin`
|
||||
(по умолчанию `0.0.0.0:8000`, переопределяется `ADMIN_HOST`/`ADMIN_PORT`).
|
||||
|
||||
### Вход через Telegram
|
||||
|
||||
Админка закрыта Telegram Login Widget. Что нужно на контуре:
|
||||
|
||||
- `BOT_TOKEN` — токен бота этого контура (им же проверяется подпись `hash`);
|
||||
- `BOT_USERNAME` — username бота без `@` (виджет рисует кнопку);
|
||||
- `ADMIN_PUBLIC_URL` — публичный HTTPS-адрес (`https://debot-test.mpakosrv.ru`);
|
||||
- `ADMIN_SESSION_SECRET` — `openssl rand -hex 32`;
|
||||
- `ADMIN_IDS` — Telegram-ID администраторов: ими засевается `bot.admin_users` при старте
|
||||
(`ON CONFLICT DO NOTHING`, существующих не трогает).
|
||||
|
||||
Домен должен быть привязан к боту в `@BotFather` (`/setdomain`), а TLS — настроен на
|
||||
reverse proxy. Сессия живёт 12 часов (`ADMIN_SESSION_MAX_AGE`), cookie подписана и
|
||||
`HttpOnly`; в проде ставится флаг `Secure`. Все POST-формы проверяют CSRF-токен.
|
||||
Роль администратора перечитывается из БД на каждом запросе — отключение пользователя
|
||||
(`enabled = false`) или удаление записи сразу закрывает доступ.
|
||||
|
||||
## Бэкап событий (забирает домашний сервер)
|
||||
|
||||
Сырые события живут на VPS ограниченное время: раз в сутки домашний сервер **сам забирает**
|
||||
|
|
|
|||
|
|
@ -62,6 +62,12 @@ APP_ENV=prod python -m bot # prod
|
|||
правка применяется без перезапуска; при недоступной БД бот использует тексты из кода.
|
||||
Предпросмотр использует тот же `render()`, что и бот.
|
||||
|
||||
**Вход** — через Telegram Login Widget: подпись `hash` проверяется токеном бота, доступ
|
||||
есть только у записей в `bot.admin_users` (роль проверяется в БД на каждом запросе).
|
||||
Первые администраторы засеваются из `ADMIN_IDS` при старте. Сессия — подписанная
|
||||
HttpOnly-cookie (`ADMIN_SESSION_SECRET`), все формы защищены CSRF. Для входа нужны
|
||||
`BOT_TOKEN`, `BOT_USERNAME` (без @) и HTTPS-адрес `ADMIN_PUBLIC_URL`.
|
||||
|
||||
Раздел «Расписания» — CRUD рассылок (`bot.schedules` + история): время, дни недели,
|
||||
таймзона, шаблон и вкл/выкл. Бот раз в минуту читает активные расписания из БД; если
|
||||
база недоступна — использует расписание из кода (совпадает с сидом `V16`).
|
||||
|
|
@ -127,7 +133,7 @@ make backup E=test # на домашнем сервере: ./scripts/clickho
|
|||
|
||||
## Версионирование
|
||||
|
||||
- Единый источник версии — `common/version.py` (`__version__`), текущая — `0.4.4`.
|
||||
- Единый источник версии — `common/version.py` (`__version__`), текущая — `0.4.5`.
|
||||
- Схема — SemVer-подобная (`MAJOR.MINOR.PATCH`); git-тег = `v` + версия (`v0.1`).
|
||||
- Docker-образы тегируются версией (Блок 2), события ClickHouse получают `bot_version` (Блок 3).
|
||||
- Выпуск: изменить `common/version.py` → коммит → аннотированный тег `vX.Y` → `git push --tags`.
|
||||
|
|
|
|||
35
admin/app.py
35
admin/app.py
|
|
@ -2,22 +2,29 @@
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import secrets
|
||||
from collections.abc import AsyncIterator
|
||||
from contextlib import asynccontextmanager
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import FastAPI
|
||||
from fastapi.templating import Jinja2Templates
|
||||
from starlette.middleware.sessions import SessionMiddleware
|
||||
|
||||
from admin.routes import health, messages, pages, schedules
|
||||
from admin.middleware import AdminAuthMiddleware
|
||||
from admin.routes import auth, health, messages, pages, schedules
|
||||
from common import __version__
|
||||
from common.config import AdminSettings, get_admin_settings
|
||||
from common.db.pool import close_pool, init_pool
|
||||
from common.logging_config import setup_logging
|
||||
from common.repositories import admins_repo
|
||||
|
||||
BASE_DIR = Path(__file__).resolve().parent
|
||||
TEMPLATES_DIR = BASE_DIR / "templates"
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def create_app(settings: AdminSettings | None = None, *, init_db: bool = True) -> FastAPI:
|
||||
"""Собрать приложение: настройки, шаблоны, роутеры.
|
||||
|
|
@ -32,6 +39,15 @@ def create_app(settings: AdminSettings | None = None, *, init_db: bool = True) -
|
|||
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
|
||||
if init_db:
|
||||
await init_pool(app.state.settings)
|
||||
try:
|
||||
inserted = await admins_repo.ensure_admins(app.state.settings.admin_ids)
|
||||
if inserted:
|
||||
logger.info("Добавлены администраторы из ADMIN_IDS: %s", inserted)
|
||||
except Exception:
|
||||
logger.warning(
|
||||
"Не удалось засеять администраторов (применены ли миграции?)",
|
||||
exc_info=True,
|
||||
)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
|
|
@ -50,7 +66,24 @@ def create_app(settings: AdminSettings | None = None, *, init_db: bool = True) -
|
|||
app.state.settings = settings
|
||||
app.state.templates = Jinja2Templates(directory=str(TEMPLATES_DIR))
|
||||
|
||||
# Порядок важен: SessionMiddleware добавляется последним и оборачивает auth,
|
||||
# поэтому request.session доступен в проверке администратора.
|
||||
app.add_middleware(AdminAuthMiddleware)
|
||||
session_secret = settings.admin_session_secret
|
||||
if not session_secret:
|
||||
session_secret = secrets.token_urlsafe(32)
|
||||
logger.warning("ADMIN_SESSION_SECRET не задан — сессии слетят при перезапуске")
|
||||
app.add_middleware(
|
||||
SessionMiddleware,
|
||||
secret_key=session_secret,
|
||||
session_cookie="debot_admin",
|
||||
max_age=settings.admin_session_max_age,
|
||||
same_site="lax",
|
||||
https_only=settings.is_prod,
|
||||
)
|
||||
|
||||
app.include_router(health.router)
|
||||
app.include_router(auth.router)
|
||||
app.include_router(pages.router)
|
||||
app.include_router(messages.router)
|
||||
app.include_router(schedules.router)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,47 @@
|
|||
"""Middleware админки: пускает дальше только вошедших администраторов."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import secrets
|
||||
from urllib.parse import quote
|
||||
|
||||
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
|
||||
from starlette.requests import Request
|
||||
from starlette.responses import RedirectResponse, Response
|
||||
|
||||
from common.repositories import admins_repo
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
#: Эндпоинты без авторизации.
|
||||
PUBLIC_PATHS = frozenset({"/healthz", "/login", "/auth/telegram"})
|
||||
|
||||
|
||||
class AdminAuthMiddleware(BaseHTTPMiddleware):
|
||||
"""Сессия + роль из БД на каждом запросе (кроме служебных и логина)."""
|
||||
|
||||
async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
|
||||
if request.url.path in PUBLIC_PATHS:
|
||||
return await call_next(request)
|
||||
|
||||
admin_id = request.session.get("admin_id") if "session" in request.scope else None
|
||||
if not admin_id:
|
||||
target = quote(request.url.path)
|
||||
return RedirectResponse(url=f"/login?next={target}", status_code=303)
|
||||
|
||||
try:
|
||||
admin = await admins_repo.get_admin(int(admin_id))
|
||||
except Exception:
|
||||
logger.exception("Не удалось проверить администратора")
|
||||
return RedirectResponse(url="/login", status_code=303)
|
||||
|
||||
if admin is None or not admin["enabled"]:
|
||||
request.session.clear()
|
||||
return RedirectResponse(url="/login", status_code=303)
|
||||
|
||||
request.state.admin = admin
|
||||
# CSRF-токен живёт в сессии; если её создали раньше — добираем.
|
||||
if not request.session.get("csrf"):
|
||||
request.session["csrf"] = secrets.token_urlsafe(32)
|
||||
return await call_next(request)
|
||||
|
|
@ -0,0 +1,98 @@
|
|||
"""Вход в админку через Telegram Login Widget."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import secrets
|
||||
|
||||
from fastapi import APIRouter, Request, Response
|
||||
from fastapi.responses import HTMLResponse, RedirectResponse
|
||||
|
||||
from admin.security import check_csrf, safe_redirect_target, verify_telegram_auth
|
||||
from common import __version__
|
||||
from common.repositories import admins_repo
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter(tags=["auth"])
|
||||
|
||||
#: Поля, которые Telegram включает в подпись (остальные в query игнорируем).
|
||||
TELEGRAM_FIELDS = frozenset({"id", "first_name", "last_name", "username", "photo_url", "auth_date", "hash"})
|
||||
|
||||
|
||||
def _deny(request: Request, message: str, status_code: int = 403) -> HTMLResponse:
|
||||
return request.app.state.templates.TemplateResponse(
|
||||
request=request,
|
||||
name="auth/denied.html",
|
||||
context={"version": __version__, "message": message},
|
||||
status_code=status_code,
|
||||
)
|
||||
|
||||
|
||||
@router.get("/login", response_class=HTMLResponse)
|
||||
async def login_page(request: Request, next: str = "/") -> Response:
|
||||
"""Страница входа с кнопкой Telegram."""
|
||||
target = safe_redirect_target(next)
|
||||
if request.session.get("admin_id"):
|
||||
return RedirectResponse(url=target, status_code=303)
|
||||
|
||||
settings = request.app.state.settings
|
||||
return request.app.state.templates.TemplateResponse(
|
||||
request=request,
|
||||
name="auth/login.html",
|
||||
context={
|
||||
"version": __version__,
|
||||
"bot_username": settings.bot_username,
|
||||
"public_url": settings.admin_public_url.rstrip("/"),
|
||||
"configured": bool(settings.bot_token and settings.bot_username),
|
||||
"next": target,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@router.get("/auth/telegram")
|
||||
async def telegram_callback(request: Request) -> Response:
|
||||
"""Проверить подпись Telegram, роль в БД и открыть сессию."""
|
||||
settings = request.app.state.settings
|
||||
data = {key: value for key, value in request.query_params.items() if key in TELEGRAM_FIELDS}
|
||||
|
||||
if not verify_telegram_auth(data, settings.bot_token):
|
||||
logger.warning("Отклонён Telegram-логин: неверная подпись или устаревший auth_date")
|
||||
return _deny(request, "Не удалось проверить вход через Telegram. Попробуйте ещё раз.")
|
||||
|
||||
try:
|
||||
telegram_id = int(str(data.get("id")))
|
||||
except (TypeError, ValueError):
|
||||
return _deny(request, "Telegram не передал идентификатор пользователя.", status_code=400)
|
||||
|
||||
try:
|
||||
admin = await admins_repo.get_admin(telegram_id)
|
||||
except Exception:
|
||||
logger.exception("Ошибка проверки администратора при входе")
|
||||
return _deny(request, "Не удалось проверить доступ. Попробуйте позже.", status_code=503)
|
||||
|
||||
if admin is None or not admin["enabled"]:
|
||||
logger.info("Отказ во входе: %s нет среди администраторов", telegram_id)
|
||||
return _deny(request, "У вас нет доступа к админке. Обратитесь к владельцу.")
|
||||
|
||||
request.session.clear()
|
||||
request.session["admin_id"] = telegram_id
|
||||
request.session["admin_username"] = admin["username"] or data.get("username") or data.get("first_name") or ""
|
||||
request.session["csrf"] = secrets.token_urlsafe(32)
|
||||
|
||||
try:
|
||||
await admins_repo.touch_login(telegram_id)
|
||||
except Exception:
|
||||
logger.warning("Не удалось обновить last_login_at", exc_info=True)
|
||||
|
||||
logger.info("Вход в админку: %s", telegram_id)
|
||||
return RedirectResponse(url=safe_redirect_target(request.query_params.get("next")), status_code=303)
|
||||
|
||||
|
||||
@router.post("/logout")
|
||||
async def logout(request: Request) -> Response:
|
||||
"""Выйти: очистить сессию."""
|
||||
form = await request.form()
|
||||
check_csrf(request, str(form.get("csrf_token", "")))
|
||||
request.session.clear()
|
||||
return RedirectResponse(url="/login", status_code=303)
|
||||
|
|
@ -8,6 +8,7 @@ from fastapi import APIRouter, HTTPException, Request, Response
|
|||
from fastapi.responses import HTMLResponse, RedirectResponse
|
||||
|
||||
from admin.preview import SAMPLE_CONTEXT
|
||||
from admin.security import check_csrf
|
||||
from common import __version__
|
||||
from common.repositories import templates_repo
|
||||
from common.texts.render import MAX_MESSAGE_LENGTH, TAG_REGISTRY, TemplateError, render, validate
|
||||
|
|
@ -56,6 +57,7 @@ async def preview_message(request: Request) -> HTMLResponse:
|
|||
но скрипты из введённого текста не выполняются.
|
||||
"""
|
||||
form = await request.form()
|
||||
check_csrf(request, str(form.get("csrf_token", "")))
|
||||
text = str(form.get("text", ""))
|
||||
|
||||
unknown = validate(text)
|
||||
|
|
@ -85,6 +87,7 @@ async def preview_message(request: Request) -> HTMLResponse:
|
|||
async def save_message(request: Request, key: str) -> Response:
|
||||
"""Сохранить текст шаблона; при ошибках вернуть редактор с подсветкой."""
|
||||
form = await request.form()
|
||||
check_csrf(request, str(form.get("csrf_token", "")))
|
||||
text = str(form.get("text", ""))
|
||||
|
||||
template = await templates_repo.get_template(key)
|
||||
|
|
@ -109,7 +112,8 @@ async def save_message(request: Request, key: str) -> Response:
|
|||
status_code=400,
|
||||
)
|
||||
|
||||
await templates_repo.update_template(key, text)
|
||||
editor_id = int(request.state.admin["telegram_id"])
|
||||
await templates_repo.update_template(key, text, editor_id=editor_id)
|
||||
return RedirectResponse(url=f"/messages/{key}?saved=1", status_code=303)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -9,12 +9,18 @@ from fastapi import APIRouter, HTTPException, Request, Response
|
|||
from fastapi.responses import HTMLResponse, RedirectResponse
|
||||
|
||||
from admin.forms import format_weekdays, parse_schedule_fields
|
||||
from admin.security import check_csrf
|
||||
from common import __version__
|
||||
from common.repositories import schedules_repo, templates_repo
|
||||
|
||||
router = APIRouter(prefix="/schedules", tags=["schedules"])
|
||||
|
||||
|
||||
def _editor_id(request: Request) -> int:
|
||||
"""Telegram-ID текущего администратора (ставит middleware)."""
|
||||
return int(request.state.admin["telegram_id"])
|
||||
|
||||
|
||||
def _raw_from_form(form: Any) -> dict[str, Any]:
|
||||
return {
|
||||
"template_key": str(form.get("template_key", "")),
|
||||
|
|
@ -107,6 +113,7 @@ async def new_schedule(request: Request) -> HTMLResponse:
|
|||
async def create_schedule(request: Request) -> Response:
|
||||
"""Создать расписание (с проверкой полей и существования шаблона)."""
|
||||
form = await request.form()
|
||||
check_csrf(request, str(form.get("csrf_token", "")))
|
||||
raw = _raw_from_form(form)
|
||||
data, errors = parse_schedule_fields(**raw)
|
||||
message_templates = await templates_repo.list_templates()
|
||||
|
|
@ -123,7 +130,7 @@ async def create_schedule(request: Request) -> Response:
|
|||
status_code=400,
|
||||
)
|
||||
|
||||
await schedules_repo.create_schedule(data)
|
||||
await schedules_repo.create_schedule(data, editor_id=_editor_id(request))
|
||||
return RedirectResponse(url="/schedules", status_code=303)
|
||||
|
||||
|
||||
|
|
@ -149,6 +156,7 @@ async def update_schedule(request: Request, schedule_id: int) -> Response:
|
|||
raise HTTPException(status_code=404, detail="Расписание не найдено")
|
||||
|
||||
form = await request.form()
|
||||
check_csrf(request, str(form.get("csrf_token", "")))
|
||||
raw = _raw_from_form(form)
|
||||
data, errors = parse_schedule_fields(**raw)
|
||||
message_templates = await templates_repo.list_templates()
|
||||
|
|
@ -167,24 +175,30 @@ async def update_schedule(request: Request, schedule_id: int) -> Response:
|
|||
status_code=400,
|
||||
)
|
||||
|
||||
await schedules_repo.update_schedule(schedule_id, data)
|
||||
await schedules_repo.update_schedule(schedule_id, data, editor_id=_editor_id(request))
|
||||
return RedirectResponse(url="/schedules", status_code=303)
|
||||
|
||||
|
||||
@router.post("/{schedule_id}/toggle")
|
||||
async def toggle_schedule(schedule_id: int) -> Response:
|
||||
async def toggle_schedule(request: Request, schedule_id: int) -> Response:
|
||||
"""Включить или выключить расписание."""
|
||||
form = await request.form()
|
||||
check_csrf(request, str(form.get("csrf_token", "")))
|
||||
|
||||
schedule = await schedules_repo.get_schedule(schedule_id)
|
||||
if schedule is None:
|
||||
raise HTTPException(status_code=404, detail="Расписание не найдено")
|
||||
|
||||
await schedules_repo.set_enabled(schedule_id, not schedule["enabled"])
|
||||
await schedules_repo.set_enabled(schedule_id, not schedule["enabled"], editor_id=_editor_id(request))
|
||||
return RedirectResponse(url="/schedules", status_code=303)
|
||||
|
||||
|
||||
@router.post("/{schedule_id}/delete")
|
||||
async def delete_schedule(schedule_id: int) -> Response:
|
||||
async def delete_schedule(request: Request, schedule_id: int) -> Response:
|
||||
"""Удалить расписание (история при этом сохраняется)."""
|
||||
if not await schedules_repo.delete_schedule(schedule_id):
|
||||
form = await request.form()
|
||||
check_csrf(request, str(form.get("csrf_token", "")))
|
||||
|
||||
if not await schedules_repo.delete_schedule(schedule_id, editor_id=_editor_id(request)):
|
||||
raise HTTPException(status_code=404, detail="Расписание не найдено")
|
||||
return RedirectResponse(url="/schedules", status_code=303)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,69 @@
|
|||
"""Безопасность админки: проверка Telegram-логина, сессия и CSRF."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import hmac
|
||||
import secrets
|
||||
import time
|
||||
from collections.abc import Mapping
|
||||
|
||||
from fastapi import HTTPException, Request
|
||||
|
||||
#: Сколько живёт подпись Telegram-логина (секунды).
|
||||
TELEGRAM_AUTH_MAX_AGE = 24 * 60 * 60
|
||||
|
||||
#: Допустимое расхождение часов с Telegram (секунды).
|
||||
FUTURE_SKEW = 300
|
||||
|
||||
|
||||
def telegram_check_string(data: Mapping[str, object]) -> str:
|
||||
"""Строка для проверки подписи: отсортированные ``key=value`` без ``hash``."""
|
||||
pairs = sorted((key, str(value)) for key, value in data.items() if key != "hash")
|
||||
return "\n".join(f"{key}={value}" for key, value in pairs)
|
||||
|
||||
|
||||
def verify_telegram_auth(
|
||||
data: Mapping[str, object],
|
||||
bot_token: str,
|
||||
*,
|
||||
max_age: int = TELEGRAM_AUTH_MAX_AGE,
|
||||
now: float | None = None,
|
||||
) -> bool:
|
||||
"""Проверить подпись Telegram Login Widget и свежесть ``auth_date``.
|
||||
|
||||
Секрет — ``SHA256(bot_token)``, подпись — HMAC-SHA256 от check-string.
|
||||
"""
|
||||
received = str(data.get("hash") or "")
|
||||
if not received or not bot_token:
|
||||
return False
|
||||
|
||||
expected = hmac.new(
|
||||
hashlib.sha256(bot_token.encode()).digest(),
|
||||
telegram_check_string(data).encode(),
|
||||
hashlib.sha256,
|
||||
).hexdigest()
|
||||
if not hmac.compare_digest(expected, received):
|
||||
return False
|
||||
|
||||
try:
|
||||
auth_date = int(str(data.get("auth_date") or ""))
|
||||
except (TypeError, ValueError):
|
||||
return False
|
||||
|
||||
moment = time.time() if now is None else now
|
||||
return -FUTURE_SKEW <= moment - auth_date <= max_age
|
||||
|
||||
|
||||
def check_csrf(request: Request, token: str) -> None:
|
||||
"""Проверить CSRF-токен формы; при несовпадении — 400."""
|
||||
expected = request.session.get("csrf")
|
||||
if not expected or not token or not secrets.compare_digest(str(expected), str(token)):
|
||||
raise HTTPException(status_code=400, detail="CSRF-токен неверен или устарел")
|
||||
|
||||
|
||||
def safe_redirect_target(value: str | None) -> str:
|
||||
"""Разрешить редирект только на внутренний путь (защита от open redirect)."""
|
||||
if value and value.startswith("/") and not value.startswith("//"):
|
||||
return value
|
||||
return "/"
|
||||
|
|
@ -0,0 +1,13 @@
|
|||
{% extends "base.html" %}
|
||||
|
||||
{% block title %}Доступ закрыт{% endblock %}
|
||||
|
||||
{% block content %}
|
||||
<div class="mx-auto mt-10 max-w-md rounded-lg border border-red-200 bg-white p-8 text-center">
|
||||
<h1 class="text-2xl font-semibold text-red-700">Доступ закрыт</h1>
|
||||
<p class="mt-3 text-sm text-slate-700">{{ message }}</p>
|
||||
<p class="mt-6">
|
||||
<a href="/login" class="text-sky-700 hover:underline">Вернуться к входу</a>
|
||||
</p>
|
||||
</div>
|
||||
{% endblock %}
|
||||
|
|
@ -0,0 +1,28 @@
|
|||
{% extends "base.html" %}
|
||||
|
||||
{% block title %}Вход{% endblock %}
|
||||
|
||||
{% block content %}
|
||||
<div class="mx-auto mt-10 max-w-md rounded-lg border border-slate-200 bg-white p-8 text-center">
|
||||
<h1 class="text-2xl font-semibold">Вход в админку</h1>
|
||||
<p class="mt-2 text-sm text-slate-600">Доступ только администраторам клана.</p>
|
||||
|
||||
{% if configured %}
|
||||
<div class="mt-8 flex justify-center">
|
||||
<script
|
||||
async
|
||||
src="https://telegram.org/js/telegram-widget.js?22"
|
||||
data-telegram-login="{{ bot_username }}"
|
||||
data-size="large"
|
||||
data-auth-url="{{ public_url }}/auth/telegram?next={{ next | urlencode }}"
|
||||
data-request-access="write"
|
||||
></script>
|
||||
</div>
|
||||
{% else %}
|
||||
<p class="mt-6 rounded-lg border border-amber-200 bg-amber-50 p-3 text-sm text-amber-800">
|
||||
Вход не настроен: задайте <span class="font-mono">BOT_TOKEN</span>,
|
||||
<span class="font-mono">BOT_USERNAME</span> и <span class="font-mono">ADMIN_PUBLIC_URL</span>.
|
||||
</p>
|
||||
{% endif %}
|
||||
</div>
|
||||
{% endblock %}
|
||||
|
|
@ -15,6 +15,13 @@
|
|||
<a href="/" class="text-slate-600 hover:text-slate-900">Главная</a>
|
||||
<a href="/messages" class="text-slate-600 hover:text-slate-900">Сообщения</a>
|
||||
<a href="/schedules" class="text-slate-600 hover:text-slate-900">Расписания</a>
|
||||
{% if request.session.get('admin_id') %}
|
||||
<span class="text-slate-500">{{ request.session.get('admin_username') or 'админ' }}</span>
|
||||
<form method="post" action="/logout">
|
||||
<input type="hidden" name="csrf_token" value="{{ request.session.get('csrf', '') }}" />
|
||||
<button type="submit" class="text-slate-600 hover:text-slate-900">Выйти</button>
|
||||
</form>
|
||||
{% endif %}
|
||||
<span class="text-slate-500">админка · v{{ version }}</span>
|
||||
</nav>
|
||||
</div>
|
||||
|
|
|
|||
|
|
@ -24,6 +24,7 @@
|
|||
{% endif %}
|
||||
|
||||
<form method="post" action="/messages/{{ template.key }}" class="mt-6 space-y-4">
|
||||
<input type="hidden" name="csrf_token" value="{{ request.session.get('csrf', '') }}" />
|
||||
<div>
|
||||
<label for="text" class="text-sm font-medium text-slate-700">Текст шаблона</label>
|
||||
<textarea
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@
|
|||
{% endif %}
|
||||
|
||||
<form method="post" action="{{ schedule.action }}" class="mt-6 max-w-3xl space-y-5">
|
||||
<input type="hidden" name="csrf_token" value="{{ request.session.get('csrf', '') }}" />
|
||||
<div>
|
||||
<label for="template_key" class="text-sm font-medium text-slate-700">Шаблон сообщения</label>
|
||||
<select
|
||||
|
|
|
|||
|
|
@ -48,11 +48,13 @@
|
|||
<div class="flex items-center justify-end gap-2">
|
||||
<a href="/schedules/{{ schedule.id }}" class="text-sky-700 hover:underline">Изменить</a>
|
||||
<form method="post" action="/schedules/{{ schedule.id }}/toggle">
|
||||
<input type="hidden" name="csrf_token" value="{{ request.session.get('csrf', '') }}" />
|
||||
<button type="submit" class="text-slate-600 hover:underline">
|
||||
{% if schedule.enabled %}Выключить{% else %}Включить{% endif %}
|
||||
</button>
|
||||
</form>
|
||||
<form method="post" action="/schedules/{{ schedule.id }}/delete" onsubmit="return confirm('Удалить расписание?');">
|
||||
<input type="hidden" name="csrf_token" value="{{ request.session.get('csrf', '') }}" />
|
||||
<button type="submit" class="text-red-600 hover:underline">Удалить</button>
|
||||
</form>
|
||||
</div>
|
||||
|
|
|
|||
|
|
@ -127,6 +127,16 @@ class AdminSettings(Settings):
|
|||
admin_host: str = "0.0.0.0"
|
||||
admin_port: int = 8000
|
||||
|
||||
#: Публичный адрес админки (для Telegram Login Widget; нужен HTTPS).
|
||||
admin_public_url: str = "http://localhost:8000"
|
||||
#: Username бота без @ — из него виджет рисует кнопку входа.
|
||||
bot_username: str = ""
|
||||
#: Секрет подписи cookie-сессии. В проде обязателен, иначе сессии слетают
|
||||
#: при перезапуске (генерируется временный).
|
||||
admin_session_secret: str = ""
|
||||
#: Время жизни сессии, секунды (по умолчанию 12 часов).
|
||||
admin_session_max_age: int = 12 * 60 * 60
|
||||
|
||||
|
||||
@lru_cache
|
||||
def get_admin_settings() -> AdminSettings:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,53 @@
|
|||
"""Репозиторий администраторов веб-панели (Блок 4.2)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from common.db.pool import get_pool
|
||||
|
||||
|
||||
async def get_admin(telegram_id: int) -> dict[str, Any] | None:
|
||||
"""Вернуть администратора по Telegram-ID (для проверки доступа)."""
|
||||
async with get_pool().connection() as conn:
|
||||
cur = await conn.execute(
|
||||
"""
|
||||
SELECT telegram_id, username, role, enabled, created_at, last_login_at
|
||||
FROM bot.admin_users
|
||||
WHERE telegram_id = %s
|
||||
""",
|
||||
(telegram_id,),
|
||||
)
|
||||
return await cur.fetchone()
|
||||
|
||||
|
||||
async def ensure_admins(telegram_ids: list[int]) -> int:
|
||||
"""Засеять администраторов из ADMIN_IDS (существующих не трогаем).
|
||||
|
||||
Возвращает количество вставленных строк.
|
||||
"""
|
||||
if not telegram_ids:
|
||||
return 0
|
||||
|
||||
inserted = 0
|
||||
async with get_pool().connection() as conn:
|
||||
for telegram_id in dict.fromkeys(telegram_ids):
|
||||
cur = await conn.execute(
|
||||
"""
|
||||
INSERT INTO bot.admin_users (telegram_id, role)
|
||||
VALUES (%s, 'admin')
|
||||
ON CONFLICT (telegram_id) DO NOTHING
|
||||
""",
|
||||
(telegram_id,),
|
||||
)
|
||||
inserted += cur.rowcount or 0
|
||||
return inserted
|
||||
|
||||
|
||||
async def touch_login(telegram_id: int) -> None:
|
||||
"""Отметить успешный вход (не критично, ошибки не должны ломать логин)."""
|
||||
async with get_pool().connection() as conn:
|
||||
await conn.execute(
|
||||
"UPDATE bot.admin_users SET last_login_at = now() WHERE telegram_id = %s",
|
||||
(telegram_id,),
|
||||
)
|
||||
|
|
@ -7,4 +7,4 @@
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
__version__ = "0.4.4"
|
||||
__version__ = "0.4.5"
|
||||
|
|
|
|||
|
|
@ -12,6 +12,12 @@ REGISTRY=git.mpakosrv.ru
|
|||
IMAGE_TAG=latest
|
||||
|
||||
BOT_TOKEN=
|
||||
# Username бота без @ (кнопка Telegram-входа в админку)
|
||||
BOT_USERNAME=
|
||||
# Публичный HTTPS-адрес админки (Telegram Login Widget)
|
||||
ADMIN_PUBLIC_URL=https://debot.mpakosrv.ru
|
||||
# Секрет подписи cookie-сессии: openssl rand -hex 32
|
||||
ADMIN_SESSION_SECRET=
|
||||
|
||||
# Postgres (для контейнера бота host = имя сервиса compose)
|
||||
DB_HOST=postgres
|
||||
|
|
|
|||
|
|
@ -11,6 +11,12 @@ REGISTRY=git.mpakosrv.ru
|
|||
IMAGE_TAG=latest
|
||||
|
||||
BOT_TOKEN=
|
||||
# Username бота без @ (кнопка Telegram-входа в админку)
|
||||
BOT_USERNAME=
|
||||
# Публичный HTTPS-адрес админки (Telegram Login Widget)
|
||||
ADMIN_PUBLIC_URL=https://debot-test.mpakosrv.ru
|
||||
# Секрет подписи cookie-сессии: openssl rand -hex 32
|
||||
ADMIN_SESSION_SECRET=
|
||||
|
||||
DB_HOST=postgres
|
||||
DB_PORT=5432
|
||||
|
|
|
|||
|
|
@ -8,6 +8,9 @@ jinja2>=3.1,<4.0
|
|||
pydantic>=2.7,<3.0
|
||||
pydantic-settings>=2.3,<3.0
|
||||
|
||||
# Подписанные cookie-сессии (Starlette SessionMiddleware).
|
||||
itsdangerous>=2.2,<3.0
|
||||
|
||||
# Таймзоны для проверки расписаний (в slim-образах нет системного tzdata).
|
||||
tzdata>=2024.1
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,15 @@
|
|||
-- Администраторы веб-панели (Блок 4.2): вход через Telegram Login Widget.
|
||||
-- Первых админов засевает приложение из ADMIN_IDS (идемпотентно при старте).
|
||||
|
||||
CREATE TABLE IF NOT EXISTS bot.admin_users (
|
||||
telegram_id bigint PRIMARY KEY,
|
||||
username text,
|
||||
role text NOT NULL DEFAULT 'admin' CHECK (role IN ('admin', 'moderator')),
|
||||
enabled boolean NOT NULL DEFAULT true,
|
||||
created_at timestamptz NOT NULL DEFAULT now(),
|
||||
last_login_at timestamptz
|
||||
);
|
||||
|
||||
COMMENT ON TABLE bot.admin_users IS 'Администраторы веб-панели (доступ через Telegram-логин)';
|
||||
COMMENT ON COLUMN bot.admin_users.role IS 'admin — полный доступ; moderator — задел на ограниченные права';
|
||||
COMMENT ON COLUMN bot.admin_users.last_login_at IS 'Время последнего успешного входа';
|
||||
|
|
@ -1,20 +1,124 @@
|
|||
"""Общие фикстуры тестов."""
|
||||
"""Общие приспособления тестов админки: Telegram-вход и CSRF-токен."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
import hashlib
|
||||
import hmac
|
||||
import re
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from common.config import Settings
|
||||
from admin.app import create_app
|
||||
from common.config import AdminSettings
|
||||
from common.repositories import admins_repo
|
||||
|
||||
BOT_TOKEN = "123:TEST"
|
||||
SESSION_SECRET = "test-secret"
|
||||
ADMIN: dict[str, Any] = {
|
||||
"telegram_id": 1,
|
||||
"username": "admin",
|
||||
"role": "admin",
|
||||
"enabled": True,
|
||||
"created_at": None,
|
||||
"last_login_at": None,
|
||||
}
|
||||
|
||||
_CSRF_RE = re.compile(r'name="csrf_token" value="([^"]+)"')
|
||||
|
||||
|
||||
def admin_settings(**overrides: Any) -> AdminSettings:
|
||||
values: dict[str, Any] = {
|
||||
"_env_file": None,
|
||||
"bot_token": BOT_TOKEN,
|
||||
"bot_username": "debot_test_bot",
|
||||
"admin_public_url": "https://debot-test.mpakosrv.ru",
|
||||
"admin_session_secret": SESSION_SECRET,
|
||||
"db_password": "test",
|
||||
"app_env": "dev",
|
||||
}
|
||||
values.update(overrides)
|
||||
return AdminSettings(**values)
|
||||
|
||||
|
||||
def telegram_params(
|
||||
*,
|
||||
token: str = BOT_TOKEN,
|
||||
telegram_id: int = 1,
|
||||
auth_date: int | None = None,
|
||||
) -> dict[str, str]:
|
||||
"""Собрать query как от Telegram Login Widget (с валидной подписью)."""
|
||||
data = {
|
||||
"id": str(telegram_id),
|
||||
"first_name": "Admin",
|
||||
"username": "admin",
|
||||
"auth_date": str(auth_date if auth_date is not None else int(time.time())),
|
||||
}
|
||||
check_string = "\n".join(f"{key}={value}" for key, value in sorted(data.items()))
|
||||
secret = hashlib.sha256(token.encode()).digest()
|
||||
data["hash"] = hmac.new(secret, check_string.encode(), hashlib.sha256).hexdigest()
|
||||
return data
|
||||
|
||||
|
||||
def csrf_from(html: str) -> str:
|
||||
"""Достать CSRF-токен из HTML формы."""
|
||||
match = _CSRF_RE.search(html)
|
||||
assert match, "в странице нет csrf_token"
|
||||
return match.group(1)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def settings_factory() -> Callable[..., Settings]:
|
||||
"""Фабрика ``Settings`` с обязательным ``BOT_TOKEN`` и без чтения ``.env``."""
|
||||
def bot_token() -> str:
|
||||
"""Токен тестового бота (подпись Telegram-логина)."""
|
||||
return BOT_TOKEN
|
||||
|
||||
def _factory(**overrides: Any) -> Settings:
|
||||
return Settings(_env_file=None, bot_token="1:a", **overrides)
|
||||
|
||||
return _factory
|
||||
@pytest.fixture
|
||||
def csrf_token():
|
||||
"""Функция: достать CSRF-токен из HTML."""
|
||||
return csrf_from
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def telegram_query():
|
||||
"""Функция: собрать подписанный query Telegram-логина."""
|
||||
return telegram_params
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def settings_factory():
|
||||
"""Фабрика настроек админки для тестов."""
|
||||
return admin_settings
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client_factory(monkeypatch: pytest.MonkeyPatch):
|
||||
"""Фабрика TestClient: подменяет проверку администратора и умеет логиниться."""
|
||||
|
||||
async def fake_get_admin(telegram_id: int) -> dict[str, Any] | None:
|
||||
return ADMIN if int(telegram_id) == ADMIN["telegram_id"] else None
|
||||
|
||||
async def fake_touch_login(telegram_id: int) -> None:
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(admins_repo, "get_admin", fake_get_admin)
|
||||
monkeypatch.setattr(admins_repo, "touch_login", fake_touch_login)
|
||||
|
||||
def make(
|
||||
*,
|
||||
authenticated: bool = True,
|
||||
follow_redirects: bool = True,
|
||||
settings: AdminSettings | None = None,
|
||||
) -> TestClient:
|
||||
client = TestClient(
|
||||
create_app(settings or admin_settings(), init_db=False),
|
||||
follow_redirects=follow_redirects,
|
||||
)
|
||||
if authenticated:
|
||||
response = client.get("/auth/telegram", params=telegram_params())
|
||||
assert response.status_code in {200, 303}
|
||||
return client
|
||||
|
||||
return make
|
||||
|
|
|
|||
|
|
@ -8,8 +8,6 @@ from typing import Any
|
|||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from admin.app import create_app
|
||||
from common.config import AdminSettings
|
||||
from common.repositories import templates_repo
|
||||
|
||||
TEMPLATE: dict[str, Any] = {
|
||||
|
|
@ -22,42 +20,7 @@ TEMPLATE: dict[str, Any] = {
|
|||
}
|
||||
|
||||
|
||||
def _client() -> TestClient:
|
||||
settings = AdminSettings(_env_file=None, bot_token="", db_password="test", app_env="dev")
|
||||
return TestClient(create_app(settings, init_db=False))
|
||||
|
||||
|
||||
def test_healthz() -> None:
|
||||
response = _client().get("/healthz")
|
||||
|
||||
assert response.status_code == 200
|
||||
payload = response.json()
|
||||
assert payload["status"] == "ok"
|
||||
assert payload["app_env"] == "dev"
|
||||
|
||||
|
||||
def test_index_renders_sections() -> None:
|
||||
response = _client().get("/")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "Панель администрирования" in response.text
|
||||
assert "Расписания" in response.text
|
||||
|
||||
|
||||
def test_messages_list(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
async def fake_list() -> list[dict[str, Any]]:
|
||||
return [{**TEMPLATE, "text_length": len(TEMPLATE["text"])}]
|
||||
|
||||
monkeypatch.setattr(templates_repo, "list_templates", fake_list)
|
||||
|
||||
response = _client().get("/messages")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "info.help" in response.text
|
||||
assert "Справка (/помощь)" in response.text
|
||||
|
||||
|
||||
def test_message_editor_and_history(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
def _patch_template(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
async def fake_get(key: str) -> dict[str, Any] | None:
|
||||
return TEMPLATE if key == TEMPLATE["key"] else None
|
||||
|
||||
|
|
@ -67,72 +30,115 @@ def test_message_editor_and_history(monkeypatch: pytest.MonkeyPatch) -> None:
|
|||
monkeypatch.setattr(templates_repo, "get_template", fake_get)
|
||||
monkeypatch.setattr(templates_repo, "template_history", fake_history)
|
||||
|
||||
response = _client().get("/messages/info.help")
|
||||
|
||||
def test_healthz(client_factory) -> None:
|
||||
response = client_factory(authenticated=False).get("/healthz")
|
||||
|
||||
assert response.status_code == 200
|
||||
payload = response.json()
|
||||
assert payload["status"] == "ok"
|
||||
assert payload["app_env"] == "dev"
|
||||
|
||||
|
||||
def test_index_renders_sections(client_factory) -> None:
|
||||
response = client_factory().get("/")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "Панель администрирования" in response.text
|
||||
assert "Расписания" in response.text
|
||||
|
||||
|
||||
def test_messages_list(client_factory, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
async def fake_list() -> list[dict[str, Any]]:
|
||||
return [{**TEMPLATE, "text_length": len(TEMPLATE["text"])}]
|
||||
|
||||
monkeypatch.setattr(templates_repo, "list_templates", fake_list)
|
||||
|
||||
response = client_factory().get("/messages")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "info.help" in response.text
|
||||
assert "Справка (/помощь)" in response.text
|
||||
|
||||
|
||||
def test_message_editor_and_history(client_factory, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_patch_template(monkeypatch)
|
||||
|
||||
response = client_factory().get("/messages/info.help")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "Справка по командам" in response.text
|
||||
assert "Доступные теги" in response.text
|
||||
|
||||
|
||||
def test_message_editor_404(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
def test_message_editor_404(client_factory, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
async def fake_get(key: str) -> dict[str, Any] | None:
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(templates_repo, "get_template", fake_get)
|
||||
|
||||
assert _client().get("/messages/unknown").status_code == 404
|
||||
assert client_factory().get("/messages/unknown").status_code == 404
|
||||
|
||||
|
||||
def test_preview_renders_tags() -> None:
|
||||
response = _client().post("/messages/preview", data={"text": "Привет, {nick}!"})
|
||||
def test_preview_renders_tags(client_factory, monkeypatch: pytest.MonkeyPatch, csrf_token) -> None:
|
||||
_patch_template(monkeypatch)
|
||||
client: TestClient = client_factory()
|
||||
token = csrf_token(client.get("/messages/info.help").text)
|
||||
|
||||
response = client.post("/messages/preview", data={"text": "Привет, {nick}!", "csrf_token": token})
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "ТестовыйНик" in response.text
|
||||
assert "<iframe" in response.text
|
||||
|
||||
|
||||
def test_preview_reports_unknown_tag() -> None:
|
||||
response = _client().post("/messages/preview", data={"text": "Привет, {unknown_tag}!"})
|
||||
def test_preview_reports_unknown_tag(client_factory, monkeypatch: pytest.MonkeyPatch, csrf_token) -> None:
|
||||
_patch_template(monkeypatch)
|
||||
client: TestClient = client_factory()
|
||||
token = csrf_token(client.get("/messages/info.help").text)
|
||||
|
||||
response = client.post("/messages/preview", data={"text": "Привет, {unknown_tag}!", "csrf_token": token})
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "Неизвестные теги" in response.text
|
||||
|
||||
|
||||
def test_save_rejects_unknown_tag(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
async def fake_get(key: str) -> dict[str, Any] | None:
|
||||
return TEMPLATE
|
||||
def test_preview_requires_csrf(client_factory, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_patch_template(monkeypatch)
|
||||
|
||||
async def fake_history(key: str, *, limit: int = 10) -> list[dict[str, Any]]:
|
||||
return []
|
||||
response = client_factory().post("/messages/preview", data={"text": "Привет"})
|
||||
|
||||
monkeypatch.setattr(templates_repo, "get_template", fake_get)
|
||||
monkeypatch.setattr(templates_repo, "template_history", fake_history)
|
||||
assert response.status_code == 400
|
||||
|
||||
response = _client().post("/messages/info.help", data={"text": "текст {unknown_tag}"})
|
||||
|
||||
def test_save_rejects_unknown_tag(client_factory, monkeypatch: pytest.MonkeyPatch, csrf_token) -> None:
|
||||
_patch_template(monkeypatch)
|
||||
client: TestClient = client_factory()
|
||||
token = csrf_token(client.get("/messages/info.help").text)
|
||||
|
||||
response = client.post(
|
||||
"/messages/info.help",
|
||||
data={"text": "текст {unknown_tag}", "csrf_token": token},
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "Неизвестные теги" in response.text
|
||||
|
||||
|
||||
def test_save_updates_template(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
def test_save_updates_template(client_factory, monkeypatch: pytest.MonkeyPatch, csrf_token) -> None:
|
||||
saved: dict[str, Any] = {}
|
||||
|
||||
async def fake_get(key: str) -> dict[str, Any] | None:
|
||||
return TEMPLATE
|
||||
|
||||
async def fake_history(key: str, *, limit: int = 10) -> list[dict[str, Any]]:
|
||||
return []
|
||||
|
||||
async def fake_update(key: str, text: str, *, editor_id: int | None = None) -> dict[str, Any]:
|
||||
saved.update(key=key, text=text)
|
||||
saved.update(key=key, text=text, editor_id=editor_id)
|
||||
return {"key": key, "text": text, "old_text": TEMPLATE["text"], "changed": True}
|
||||
|
||||
monkeypatch.setattr(templates_repo, "get_template", fake_get)
|
||||
monkeypatch.setattr(templates_repo, "template_history", fake_history)
|
||||
_patch_template(monkeypatch)
|
||||
monkeypatch.setattr(templates_repo, "update_template", fake_update)
|
||||
|
||||
response = _client().post("/messages/info.help", data={"text": "Новая справка"})
|
||||
client: TestClient = client_factory()
|
||||
token = csrf_token(client.get("/messages/info.help").text)
|
||||
response = client.post("/messages/info.help", data={"text": "Новая справка", "csrf_token": token})
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "Текст сохранён" in response.text
|
||||
assert saved == {"key": "info.help", "text": "Новая справка"}
|
||||
assert saved == {"key": "info.help", "text": "Новая справка", "editor_id": 1}
|
||||
|
|
|
|||
|
|
@ -8,8 +8,6 @@ from typing import Any
|
|||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from admin.app import create_app
|
||||
from common.config import AdminSettings
|
||||
from common.repositories import schedules_repo, templates_repo
|
||||
|
||||
TEMPLATE: dict[str, Any] = {
|
||||
|
|
@ -36,11 +34,6 @@ SCHEDULE: dict[str, Any] = {
|
|||
}
|
||||
|
||||
|
||||
def _client() -> TestClient:
|
||||
settings = AdminSettings(_env_file=None, bot_token="", db_password="test", app_env="dev")
|
||||
return TestClient(create_app(settings, init_db=False))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake_repos(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]:
|
||||
state: dict[str, Any] = {}
|
||||
|
|
@ -56,6 +49,7 @@ def fake_repos(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]:
|
|||
|
||||
async def create_schedule(data: dict[str, Any], *, editor_id: int | None = None) -> int:
|
||||
state["created"] = data
|
||||
state["editor_id"] = editor_id
|
||||
return 2
|
||||
|
||||
async def update_schedule(schedule_id: int, data: dict[str, Any], *, editor_id: int | None = None) -> bool:
|
||||
|
|
@ -88,8 +82,8 @@ def fake_repos(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]:
|
|||
return state
|
||||
|
||||
|
||||
def test_schedules_list(fake_repos: dict[str, Any]) -> None:
|
||||
response = _client().get("/schedules")
|
||||
def test_schedules_list(fake_repos: dict[str, Any], client_factory) -> None:
|
||||
response = client_factory().get("/schedules")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "17:00" in response.text
|
||||
|
|
@ -97,16 +91,19 @@ def test_schedules_list(fake_repos: dict[str, Any]) -> None:
|
|||
assert "включено" in response.text
|
||||
|
||||
|
||||
def test_new_schedule_form(fake_repos: dict[str, Any]) -> None:
|
||||
response = _client().get("/schedules/new")
|
||||
def test_new_schedule_form(fake_repos: dict[str, Any], client_factory) -> None:
|
||||
response = client_factory().get("/schedules/new")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "Новое расписание" in response.text
|
||||
assert "notify.blago" in response.text
|
||||
|
||||
|
||||
def test_create_schedule(fake_repos: dict[str, Any]) -> None:
|
||||
response = _client().post(
|
||||
def test_create_schedule(fake_repos: dict[str, Any], client_factory, csrf_token) -> None:
|
||||
client: TestClient = client_factory()
|
||||
token = csrf_token(client.get("/schedules/new").text)
|
||||
|
||||
response = client.post(
|
||||
"/schedules",
|
||||
data={
|
||||
"template_key": "notify.blago",
|
||||
|
|
@ -116,6 +113,7 @@ def test_create_schedule(fake_repos: dict[str, Any]) -> None:
|
|||
"timezone": "Europe/Moscow",
|
||||
"description": "Благо",
|
||||
"enabled": "on",
|
||||
"csrf_token": token,
|
||||
},
|
||||
)
|
||||
|
||||
|
|
@ -129,12 +127,22 @@ def test_create_schedule(fake_repos: dict[str, Any]) -> None:
|
|||
"enabled": True,
|
||||
"description": "Благо",
|
||||
}
|
||||
assert fake_repos["editor_id"] == 1
|
||||
|
||||
|
||||
def test_create_schedule_rejects_bad_hour(fake_repos: dict[str, Any]) -> None:
|
||||
response = _client().post(
|
||||
def test_create_schedule_rejects_bad_hour(fake_repos: dict[str, Any], client_factory, csrf_token) -> None:
|
||||
client: TestClient = client_factory()
|
||||
token = csrf_token(client.get("/schedules/new").text)
|
||||
|
||||
response = client.post(
|
||||
"/schedules",
|
||||
data={"template_key": "notify.blago", "hour": "25", "minute": "0", "timezone": "Europe/Moscow"},
|
||||
data={
|
||||
"template_key": "notify.blago",
|
||||
"hour": "25",
|
||||
"minute": "0",
|
||||
"timezone": "Europe/Moscow",
|
||||
"csrf_token": token,
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
|
|
@ -142,29 +150,50 @@ def test_create_schedule_rejects_bad_hour(fake_repos: dict[str, Any]) -> None:
|
|||
assert "created" not in fake_repos
|
||||
|
||||
|
||||
def test_create_schedule_rejects_unknown_template(fake_repos: dict[str, Any]) -> None:
|
||||
response = _client().post(
|
||||
def test_create_schedule_rejects_unknown_template(fake_repos: dict[str, Any], client_factory, csrf_token) -> None:
|
||||
client: TestClient = client_factory()
|
||||
token = csrf_token(client.get("/schedules/new").text)
|
||||
|
||||
response = client.post(
|
||||
"/schedules",
|
||||
data={"template_key": "unknown.key", "hour": "17", "minute": "0", "timezone": "Europe/Moscow"},
|
||||
data={
|
||||
"template_key": "unknown.key",
|
||||
"hour": "17",
|
||||
"minute": "0",
|
||||
"timezone": "Europe/Moscow",
|
||||
"csrf_token": token,
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "Шаблон не найден" in response.text
|
||||
|
||||
|
||||
def test_edit_schedule(fake_repos: dict[str, Any]) -> None:
|
||||
response = _client().get("/schedules/1")
|
||||
def test_create_requires_csrf(fake_repos: dict[str, Any], client_factory) -> None:
|
||||
response = client_factory().post(
|
||||
"/schedules",
|
||||
data={"template_key": "notify.blago", "hour": "17", "minute": "0", "timezone": "Europe/Moscow"},
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_edit_schedule(fake_repos: dict[str, Any], client_factory) -> None:
|
||||
response = client_factory().get("/schedules/1")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "Расписание #1" in response.text
|
||||
|
||||
|
||||
def test_edit_missing_schedule_404(fake_repos: dict[str, Any]) -> None:
|
||||
assert _client().get("/schedules/99").status_code == 404
|
||||
def test_edit_missing_schedule_404(fake_repos: dict[str, Any], client_factory) -> None:
|
||||
assert client_factory().get("/schedules/99").status_code == 404
|
||||
|
||||
|
||||
def test_update_schedule(fake_repos: dict[str, Any]) -> None:
|
||||
response = _client().post(
|
||||
def test_update_schedule(fake_repos: dict[str, Any], client_factory, csrf_token) -> None:
|
||||
client: TestClient = client_factory()
|
||||
token = csrf_token(client.get("/schedules/1").text)
|
||||
|
||||
response = client.post(
|
||||
"/schedules/1",
|
||||
data={
|
||||
"template_key": "notify.blago",
|
||||
|
|
@ -174,6 +203,7 @@ def test_update_schedule(fake_repos: dict[str, Any]) -> None:
|
|||
"timezone": "Europe/Moscow",
|
||||
"description": "",
|
||||
"enabled": "on",
|
||||
"csrf_token": token,
|
||||
},
|
||||
)
|
||||
|
||||
|
|
@ -184,13 +214,19 @@ def test_update_schedule(fake_repos: dict[str, Any]) -> None:
|
|||
assert data["minute"] == 30
|
||||
|
||||
|
||||
def test_toggle_schedule(fake_repos: dict[str, Any]) -> None:
|
||||
_client().post("/schedules/1/toggle")
|
||||
def test_toggle_schedule(fake_repos: dict[str, Any], client_factory, csrf_token) -> None:
|
||||
client: TestClient = client_factory()
|
||||
token = csrf_token(client.get("/schedules").text)
|
||||
|
||||
client.post("/schedules/1/toggle", data={"csrf_token": token})
|
||||
|
||||
assert fake_repos["enabled"] == (1, False)
|
||||
|
||||
|
||||
def test_delete_schedule(fake_repos: dict[str, Any]) -> None:
|
||||
_client().post("/schedules/1/delete")
|
||||
def test_delete_schedule(fake_repos: dict[str, Any], client_factory, csrf_token) -> None:
|
||||
client: TestClient = client_factory()
|
||||
token = csrf_token(client.get("/schedules").text)
|
||||
|
||||
client.post("/schedules/1/delete", data={"csrf_token": token})
|
||||
|
||||
assert fake_repos["deleted"] == 1
|
||||
|
|
|
|||
|
|
@ -0,0 +1,108 @@
|
|||
"""Тесты входа в админку через Telegram: подпись, сессия и CSRF."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from admin.security import verify_telegram_auth
|
||||
|
||||
|
||||
def test_healthz_is_public(client_factory) -> None:
|
||||
assert client_factory(authenticated=False).get("/healthz").status_code == 200
|
||||
|
||||
|
||||
def test_protected_page_redirects_to_login(client_factory) -> None:
|
||||
response = client_factory(authenticated=False, follow_redirects=False).get("/")
|
||||
|
||||
assert response.status_code == 303
|
||||
assert response.headers["location"].startswith("/login")
|
||||
|
||||
|
||||
def test_login_page_has_widget(client_factory) -> None:
|
||||
response = client_factory(authenticated=False).get("/login")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "telegram-widget.js" in response.text
|
||||
assert "debot_test_bot" in response.text
|
||||
assert "https://debot-test.mpakosrv.ru/auth/telegram" in response.text
|
||||
|
||||
|
||||
def test_login_page_without_config(client_factory, settings_factory) -> None:
|
||||
client = client_factory(
|
||||
authenticated=False,
|
||||
settings=settings_factory(bot_token="", bot_username=""),
|
||||
)
|
||||
|
||||
response = client.get("/login")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "Вход не настроен" in response.text
|
||||
|
||||
|
||||
def test_login_with_valid_signature(client_factory, telegram_query) -> None:
|
||||
client: TestClient = client_factory(authenticated=False, follow_redirects=False)
|
||||
|
||||
response = client.get("/auth/telegram", params=telegram_query())
|
||||
|
||||
assert response.status_code == 303
|
||||
assert response.headers["location"] == "/"
|
||||
assert client.get("/").status_code == 200
|
||||
|
||||
|
||||
def test_login_rejects_bad_hash(client_factory, telegram_query) -> None:
|
||||
params = telegram_query()
|
||||
params["hash"] = "0" * 64
|
||||
|
||||
response = client_factory(authenticated=False).get("/auth/telegram", params=params)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert "Не удалось проверить вход" in response.text
|
||||
|
||||
|
||||
def test_login_rejects_stale_auth_date(client_factory, telegram_query) -> None:
|
||||
params = telegram_query(auth_date=int(time.time()) - 90_000)
|
||||
|
||||
response = client_factory(authenticated=False).get("/auth/telegram", params=params)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
def test_login_rejects_non_admin(client_factory, telegram_query) -> None:
|
||||
response = client_factory(authenticated=False).get("/auth/telegram", params=telegram_query(telegram_id=2))
|
||||
|
||||
assert response.status_code == 403
|
||||
assert "нет доступа" in response.text
|
||||
|
||||
|
||||
def test_logout_requires_csrf(client_factory) -> None:
|
||||
response = client_factory().post("/logout")
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_logout_clears_session(client_factory, csrf_token) -> None:
|
||||
client: TestClient = client_factory()
|
||||
token = csrf_token(client.get("/").text)
|
||||
|
||||
response = client.post("/logout", data={"csrf_token": token})
|
||||
|
||||
assert response.status_code == 200
|
||||
assert client.get("/", follow_redirects=False).status_code == 303
|
||||
|
||||
|
||||
def test_verify_signature(bot_token: str) -> None:
|
||||
"""Юнит-проверка подписи без HTTP (данные собираем тем же алгоритмом)."""
|
||||
import hashlib
|
||||
import hmac
|
||||
|
||||
data = {"id": "1", "first_name": "Admin", "auth_date": str(int(time.time()))}
|
||||
check = "\n".join(f"{key}={value}" for key, value in sorted(data.items()))
|
||||
secret = hashlib.sha256(bot_token.encode()).digest()
|
||||
data["hash"] = hmac.new(secret, check.encode(), hashlib.sha256).hexdigest()
|
||||
|
||||
assert verify_telegram_auth(data, bot_token)
|
||||
assert not verify_telegram_auth({**data, "id": "2"}, bot_token)
|
||||
assert not verify_telegram_auth(data, "999:OTHER")
|
||||
assert not verify_telegram_auth(data, bot_token, now=time.time() + 90_000)
|
||||
|
|
@ -7,4 +7,4 @@ from common import __version__
|
|||
|
||||
def test_version_is_set():
|
||||
assert __version__
|
||||
assert __version__ == "0.4.4"
|
||||
assert __version__ == "0.4.5"
|
||||
|
|
|
|||
Loading…
Reference in New Issue