В админку можно войти через установленный Telegram

К виджету добавился deep-link: сайт создаёт одноразовый токен на 5 минут и открывает t.me/<bot>?start=login_<token>, бот проверяет администратора и подтверждает вход, а сайт забирает токен и открывает сессию. В БД хранится только SHA-256 токена; ссылка одноразовая. Если deep-link не настроен, остаётся обычный виджет.

README.md

@@ -68,6 +68,11 @@ APP_ENV=prod python -m bot    # prod
 HttpOnly-cookie (`ADMIN_SESSION_SECRET`), все формы защищены CSRF. Для входа нужны
 `BOT_TOKEN`, `BOT_USERNAME` (без @) и HTTPS-адрес `ADMIN_PUBLIC_URL`.
 
+Дополнительно — **вход через установленный Telegram**: кнопка «Открыть Telegram»
+создаёт одноразовый токен (5 минут) и открывает `t.me/<bot>?start=login_<token>`;
+бот подтверждает вход, сайт забирает токен и открывает сессию. Токен в БД хранится
+как SHA-256 и используется один раз.
+
 Раздел «Расписания» — CRUD рассылок (`bot.schedules` + история): время, дни недели,
 таймзона, шаблон и вкл/выкл. Бот раз в минуту читает активные расписания из БД; если
 база недоступна — использует расписание из кода (совпадает с сидом `V16`).
@@ -165,7 +170,7 @@ make backup E=test     # на домашнем сервере: ./scripts/clickho
 
 ## Версионирование
 
-- Единый источник версии — `common/version.py` (`__version__`), текущая — `0.5.4`.
+- Единый источник версии — `common/version.py` (`__version__`), текущая — `0.5.5`.
 - Схема — SemVer-подобная (`MAJOR.MINOR.PATCH`); git-тег = `v` + версия (`v0.1`).
 - Docker-образы тегируются версией (Блок 2), события ClickHouse получают `bot_version` (Блок 3).
 - Выпуск: изменить `common/version.py` → коммит → аннотированный тег `vX.Y` → `git push --tags`.

admin/middleware.py

@@ -15,7 +15,15 @@ from common.repositories import admins_repo
 logger = logging.getLogger(__name__)
 
 #: Эндпоинты без авторизации.
-PUBLIC_PATHS = frozenset({"/healthz", "/login", "/auth/telegram"})
+PUBLIC_PATHS = frozenset(
+    {
+        "/healthz",
+        "/login",
+        "/auth/telegram",
+        "/auth/deeplink/start",
+        "/auth/deeplink/status",
+    }
+)
 
 
 class AdminAuthMiddleware(BaseHTTPMiddleware):

admin/routes/auth.py

@@ -6,11 +6,11 @@ import logging
 import secrets
 
 from fastapi import APIRouter, Request, Response
-from fastapi.responses import HTMLResponse, RedirectResponse
+from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
 
 from admin.security import check_csrf, safe_redirect_target, verify_telegram_auth
 from common import __version__
-from common.repositories import admins_repo
+from common.repositories import admin_login_tokens_repo, admins_repo
 
 logger = logging.getLogger(__name__)
 
@@ -45,11 +45,81 @@ async def login_page(request: Request, next: str = "/") -> Response:
             "bot_username": settings.bot_username,
             "public_url": settings.admin_public_url.rstrip("/"),
             "configured": bool(settings.bot_token and settings.bot_username),
+            "deeplink": bool(settings.bot_username and settings.admin_public_url),
+            "deeplink_pending": bool(request.cookies.get("debot_login_token")),
+            "deeplink_error": request.query_params.get("error") == "deeplink",
             "next": target,
         },
     )
 
 
+@router.get("/auth/deeplink/start")
+async def deeplink_start(request: Request) -> Response:
+    """Создать токен и открыть установленный Telegram с deep-link."""
+    settings = request.app.state.settings
+    username = settings.bot_username.lstrip("@")
+    if not username:
+        return RedirectResponse(url="/login?error=deeplink", status_code=303)
+
+    token = await admin_login_tokens_repo.create_token()
+    response = RedirectResponse(
+        url=f"https://t.me/{username}?start=login_{token}",
+        status_code=303,
+    )
+    response.set_cookie(
+        "debot_login_token",
+        token,
+        max_age=300,
+        httponly=True,
+        samesite="lax",
+        secure=settings.is_prod,
+    )
+    return response
+
+
+@router.get("/auth/deeplink/status")
+async def deeplink_status(request: Request) -> JSONResponse:
+    """Проверить, подтвердил ли пользователь вход в боте, и открыть сессию."""
+    token = request.cookies.get("debot_login_token")
+    if not token:
+        return JSONResponse({"status": "none"})
+
+    try:
+        telegram_id = await admin_login_tokens_repo.consume_token(token)
+    except Exception:
+        logger.exception("Не удалось проверить deep-link токен")
+        return JSONResponse({"status": "pending"})
+
+    if telegram_id is None:
+        return JSONResponse({"status": "pending"})
+
+    try:
+        admin = await admins_repo.get_admin(telegram_id)
+    except Exception:
+        logger.exception("Ошибка проверки администратора при deep-link входе")
+        return JSONResponse({"status": "pending"})
+
+    if admin is None or not admin["enabled"]:
+        response = JSONResponse({"status": "denied"})
+        response.delete_cookie("debot_login_token")
+        return response
+
+    request.session.clear()
+    request.session["admin_id"] = telegram_id
+    request.session["admin_username"] = admin["username"] 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("Вход в админку по deep-link: %s", telegram_id)
+    response = JSONResponse({"status": "ok"})
+    response.delete_cookie("debot_login_token")
+    return response
+
+
 @router.get("/auth/telegram")
 async def telegram_callback(request: Request) -> Response:
     """Проверить подпись Telegram, роль в БД и открыть сессию."""

admin/templates/auth/login.html

@@ -7,6 +7,12 @@
   <h1 class="text-2xl font-semibold">Вход в админку</h1>
   <p class="mt-2 text-sm text-slate-600">Доступ только администраторам клана.</p>
 
+  {% if deeplink_error %}
+  <p class="mt-4 rounded-lg border border-amber-200 bg-amber-50 p-3 text-sm text-amber-800">
+    Deep-link вход не настроен: проверьте <span class="font-mono">BOT_USERNAME</span>.
+  </p>
+  {% endif %}
+
   {% if configured %}
   <div class="mt-8 flex justify-center">
     <script
@@ -18,11 +24,51 @@
       data-request-access="write"
     ></script>
   </div>
-  {% else %}
+  {% endif %}
+
+  {% if deeplink %}
+  <div class="mt-6 border-t border-slate-100 pt-5">
+    <a
+      href="/auth/deeplink/start"
+      class="inline-block rounded-lg border border-slate-300 px-4 py-2 text-sm font-medium text-slate-700 hover:bg-slate-50"
+    >
+      Открыть Telegram
+    </a>
+    <p class="mt-2 text-xs text-slate-500">Подтвердите вход в установленном приложении Telegram.</p>
+    <p id="deeplink-status" class="mt-2 text-sm text-sky-700">
+      {% if deeplink_pending %}Ожидаем подтверждение в Telegram…{% endif %}
+    </p>
+  </div>
+  {% endif %}
+
+  {% if not configured and not deeplink %}
   <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>
+
+{% if deeplink_pending %}
+<script>
+  (function () {
+    var status = document.getElementById("deeplink-status");
+    var timer = setInterval(function () {
+      fetch("/auth/deeplink/status", { credentials: "same-origin" })
+        .then(function (response) { return response.json(); })
+        .then(function (data) {
+          if (data.status === "ok") {
+            clearInterval(timer);
+            window.location.href = "/";
+          } else if (data.status === "denied") {
+            clearInterval(timer);
+            if (status) status.textContent = "У вас нет доступа к админке.";
+          }
+        })
+        .catch(function () {});
+    }, 2000);
+    setTimeout(function () { clearInterval(timer); }, 300000);
+  })();
+</script>
+{% endif %}
 {% endblock %}

bot/bot_factory.py

@@ -25,7 +25,7 @@ def create_dispatcher() -> Dispatcher:
     Порядок важен: сначала события участников и команды, затем магазин,
     и в конце — экономика (в ней есть catch-all обработчик сообщений).
     """
-    from bot.handlers import commands, economy, member_events, shop
+    from bot.handlers import commands, economy, login, member_events, shop
     from bot.handlers.errors import register_errors
     from bot.middlewares import AnalyticsMiddleware
 
@@ -33,6 +33,7 @@ def create_dispatcher() -> Dispatcher:
     # Внешний middleware: ловит все сообщения до фильтров (события `message`/`command`).
     dp.message.outer_middleware(AnalyticsMiddleware())
     dp.include_router(member_events.router)
+    dp.include_router(login.router)
     dp.include_router(commands.router)
     dp.include_router(shop.router)
     dp.include_router(economy.router)

bot/commands_menu.py

@@ -15,6 +15,7 @@ logger = logging.getLogger(__name__)
 
 #: Ожидаемый набор команд в меню (порядок = порядок в Telegram).
 BOT_COMMANDS: tuple[BotCommand, ...] = (
+    BotCommand(command="start", description="Начать"),
     BotCommand(command="help", description="Помощь"),
     BotCommand(command="info", description="Инфо"),
     BotCommand(command="nick", description="Сменить ник"),

bot/handlers/login.py

@@ -0,0 +1,50 @@
+"""Вход в админку через deep-link из установленного Telegram."""
+
+from __future__ import annotations
+
+from aiogram import Router
+from aiogram.filters import Command
+from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup, Message
+
+from common.config import get_settings
+from common.repositories import admin_login_tokens_repo, admins_repo
+from common.texts import system
+
+router = Router(name="login")
+
+
+@router.message(Command("start"))
+async def start_handler(message: Message) -> None:
+    """Обработать /start; payload ``login_<token>`` подтверждает вход в админку."""
+    if message.chat.type != "private" or message.from_user is None:
+        return
+
+    parts = (message.text or "").split(maxsplit=1)
+    payload = parts[1].strip() if len(parts) > 1 else ""
+    if not payload.startswith("login_"):
+        await message.answer(system.MSG_START)
+        return
+
+    admin = await admins_repo.get_admin(message.from_user.id)
+    if admin is None or not admin["enabled"]:
+        await message.answer(system.MSG_LOGIN_DENIED)
+        return
+
+    if not await admin_login_tokens_repo.confirm_token(payload[6:], message.from_user.id):
+        await message.answer(system.MSG_LOGIN_INVALID)
+        return
+
+    settings = get_settings()
+    markup = None
+    if settings.admin_public_url.startswith(("http://", "https://")):
+        markup = InlineKeyboardMarkup(
+            inline_keyboard=[
+                [
+                    InlineKeyboardButton(
+                        text="Вернуться в админку",
+                        url=settings.admin_public_url.rstrip("/") + "/login",
+                    )
+                ]
+            ]
+        )
+    await message.answer(system.MSG_LOGIN_CONFIRMED, reply_markup=markup)

common/config.py

@@ -57,6 +57,9 @@ class Settings(BaseSettings):
     #: Общий каталог картинок (Docker volume), который видят бот и админка.
     media_dir: str = "/app/media"
 
+    #: Публичный адрес админки (deep-link вход, ссылки из бота).
+    admin_public_url: str = "http://localhost:8000"
+
     #: ClickHouse (аналитика событий). CLICKHOUSE_ENABLED=false — полностью выключить.
     clickhouse_enabled: bool = True
     clickhouse_host: str = "localhost"
@@ -144,7 +147,7 @@ class AdminSettings(Settings):
     admin_port: int = 8000
 
     #: Публичный адрес админки (для Telegram Login Widget; нужен HTTPS).
-    admin_public_url: str = "http://localhost:8000"
+    #: Наследуется из базовых настроек: ADMIN_PUBLIC_URL.
     #: Username бота без @ — из него виджет рисует кнопку входа.
     bot_username: str = ""
     #: Секрет подписи cookie-сессии. В проде обязателен, иначе сессии слетают

common/repositories/admin_login_tokens_repo.py

@@ -0,0 +1,73 @@
+"""Одноразовые токены входа в админку через Telegram deep-link."""
+
+from __future__ import annotations
+
+import hashlib
+import secrets
+
+from common.db.pool import get_pool
+
+#: Время жизни токена.
+TTL_MINUTES = 5
+
+
+def _hash(token: str) -> str:
+    return hashlib.sha256(token.encode()).hexdigest()
+
+
+async def create_token() -> str:
+    """Создать токен, вернуть его в открытом виде (в БД — только хеш)."""
+    token = secrets.token_urlsafe(32)
+    async with get_pool().connection() as conn:
+        await conn.execute(
+            """
+            INSERT INTO bot.admin_login_tokens (token_hash, expires_at)
+            VALUES (%s, now() + make_interval(mins => %s))
+            """,
+            (_hash(token), TTL_MINUTES),
+        )
+        await conn.execute("DELETE FROM bot.admin_login_tokens WHERE expires_at < now() - interval '1 day'")
+    return token
+
+
+async def confirm_token(token: str, telegram_id: int) -> bool:
+    """Отметить токен подтверждённым ботом. False — если истёк/уже подтверждён."""
+    async with get_pool().connection() as conn:
+        cur = await conn.execute(
+            """
+            UPDATE bot.admin_login_tokens
+            SET confirmed_at = now(), telegram_id = %s
+            WHERE token_hash = %s
+              AND confirmed_at IS NULL
+              AND used_at IS NULL
+              AND expires_at > now()
+            """,
+            (telegram_id, _hash(token)),
+        )
+        return bool(cur.rowcount)
+
+
+async def consume_token(token: str) -> int | None:
+    """Забрать подтверждённый токен один раз. Возвращает Telegram-ID или None."""
+    async with get_pool().connection() as conn:
+        cur = await conn.execute(
+            """
+            SELECT telegram_id
+            FROM bot.admin_login_tokens
+            WHERE token_hash = %s
+              AND confirmed_at IS NOT NULL
+              AND used_at IS NULL
+              AND expires_at > now()
+            FOR UPDATE
+            """,
+            (_hash(token),),
+        )
+        row = await cur.fetchone()
+        if row is None or row["telegram_id"] is None:
+            return None
+
+        await conn.execute(
+            "UPDATE bot.admin_login_tokens SET used_at = now() WHERE token_hash = %s",
+            (_hash(token),),
+        )
+        return int(row["telegram_id"])

common/texts/system.py

@@ -27,6 +27,11 @@ MSG_DONAT_PROMO = "Назови этот код админам!\n`{promo}`"
 MSG_EGGS_TOO_EARLY = "Слишком рано..."
 MSG_EGGS_EVENT_OVER = "Событие с яичками уже завершилось — увидимся на следующую Пасху."
 
+MSG_START = "Привет! Это бот Dispel Evil."
+MSG_LOGIN_CONFIRMED = "Вход в админку подтверждён. Вернитесь на сайт."
+MSG_LOGIN_DENIED = "У вас нет доступа к админке."
+MSG_LOGIN_INVALID = "Ссылка для входа устарела. Начните вход заново на сайте."
+
 MSG_BOT_ADDED = "Я пришёл! Требую админские права!"
 
 MSG_BOT_FOREIGN_CHAT = "Создатель запретил ходить в другие чатики :с"

common/version.py

@@ -7,4 +7,4 @@
 
 from __future__ import annotations
 
-__version__ = "0.5.4"
+__version__ = "0.5.5"

sql/V23__admin_login_tokens.sql

@@ -0,0 +1,14 @@
+-- Одноразовые токены для входа в админку через установленный Telegram (deep-link).
+CREATE TABLE IF NOT EXISTS bot.admin_login_tokens (
+    token_hash  text PRIMARY KEY,
+    created_at  timestamptz NOT NULL DEFAULT now(),
+    expires_at  timestamptz NOT NULL,
+    confirmed_at timestamptz,
+    used_at     timestamptz,
+    telegram_id bigint
+);
+
+CREATE INDEX IF NOT EXISTS idx_admin_login_tokens_expires
+    ON bot.admin_login_tokens (expires_at);
+
+COMMENT ON TABLE bot.admin_login_tokens IS 'Одноразовые токены deep-link входа в админку (хранится SHA-256)';

tests/test_deeplink_login.py

@@ -0,0 +1,102 @@
+"""Тесты deep-link входа в админку через Telegram."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+
+from common.repositories import admin_login_tokens_repo
+
+
+@pytest.fixture
+def fake_tokens(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]:
+    state: dict[str, Any] = {"confirmed": None}
+
+    async def create_token() -> str:
+        return "test-token"
+
+    async def consume_token(token: str) -> int | None:
+        return state["confirmed"]
+
+    monkeypatch.setattr(admin_login_tokens_repo, "create_token", create_token)
+    monkeypatch.setattr(admin_login_tokens_repo, "consume_token", consume_token)
+    return state
+
+
+def _client(client_factory, settings_factory) -> TestClient:
+    return client_factory(
+        authenticated=False,
+        follow_redirects=False,
+        settings=settings_factory(
+            bot_username="debot_test_bot",
+            admin_public_url="https://debot-test.mpakosrv.ru",
+        ),
+    )
+
+
+def test_login_page_has_deeplink(fake_tokens: dict[str, Any], client_factory, settings_factory) -> None:
+    response = _client(client_factory, settings_factory).get("/login")
+
+    assert response.status_code == 200
+    assert "Открыть Telegram" in response.text
+    assert "/auth/deeplink/start" in response.text
+
+
+def test_deeplink_start_redirects_to_telegram(
+    fake_tokens: dict[str, Any],
+    client_factory,
+    settings_factory,
+) -> None:
+    client = _client(client_factory, settings_factory)
+
+    response = client.get("/auth/deeplink/start")
+
+    assert response.status_code == 303
+    assert response.headers["location"] == "https://t.me/debot_test_bot?start=login_test-token"
+    assert "debot_login_token" in response.headers["set-cookie"]
+
+
+def test_deeplink_status_without_cookie(fake_tokens: dict[str, Any], client_factory, settings_factory) -> None:
+    response = _client(client_factory, settings_factory).get("/auth/deeplink/status")
+
+    assert response.json() == {"status": "none"}
+
+
+def test_deeplink_status_pending(fake_tokens: dict[str, Any], client_factory, settings_factory) -> None:
+    client = _client(client_factory, settings_factory)
+    client.get("/auth/deeplink/start")
+
+    response = client.get("/auth/deeplink/status")
+
+    assert response.json() == {"status": "pending"}
+
+
+def test_deeplink_status_confirms_session(
+    fake_tokens: dict[str, Any],
+    client_factory,
+    settings_factory,
+) -> None:
+    fake_tokens["confirmed"] = 1
+    client = _client(client_factory, settings_factory)
+    client.get("/auth/deeplink/start")
+
+    response = client.get("/auth/deeplink/status")
+
+    assert response.json() == {"status": "ok"}
+    assert client.get("/").status_code == 200
+
+
+def test_deeplink_status_denied_for_unknown_admin(
+    fake_tokens: dict[str, Any],
+    client_factory,
+    settings_factory,
+) -> None:
+    fake_tokens["confirmed"] = 2
+    client = _client(client_factory, settings_factory)
+    client.get("/auth/deeplink/start")
+
+    response = client.get("/auth/deeplink/status")
+
+    assert response.json() == {"status": "denied"}

tests/test_version.py

@@ -7,4 +7,4 @@ from common import __version__
 
 def test_version_is_set():
     assert __version__
-    assert __version__ == "0.5.4"
+    assert __version__ == "0.5.5"