ElebobaandClaude Opus 5.5
219bebd5a2
- AccessService: одноразовые приглашения (128 бит, в БД только sha256, TTL 24 ч), атомарный захват приглашения (условный UPDATE — из двух сканов одного QR проходит один), антиперебор (> 5 неудач в час — молчание), отзыв (черновик и FSM удаляются), повторное приглашение - ADMIN_IDS заводятся в БД при старте; дальше — только по приглашению - AccessMiddleware: участники из БД (+ last_seen раз в 5 мин), новичкам — только /start <токен> - «👥 Участники» для админов: список, QR (segno) + ссылка, отзыв с подтверждением - один ответ на неверный/истёкший/использованный токен — без подсказок Refs #25 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
70 lines
2.5 KiB
Python
70 lines
2.5 KiB
Python
"""Access gate on every update — messages and button presses alike (SPEC A-5, ADR-011).
|
|
|
|
Members (active users in the DB) pass with `member` injected into handler data. Outsiders pass only with
|
|
`/start <token>` — the invite deep link; everything else gets one "invite only" reply a day, then silence.
|
|
"""
|
|
|
|
import time
|
|
from collections.abc import Awaitable, Callable
|
|
from typing import Any
|
|
|
|
from aiogram import BaseMiddleware
|
|
from aiogram.types import CallbackQuery, Message, TelegramObject, User
|
|
|
|
from app.access.service import AccessService, TgUser
|
|
from app.texts import Texts
|
|
|
|
NO_ACCESS_REPLY_EVERY = 24 * 3600.0
|
|
|
|
|
|
def tg_user(user: User) -> TgUser:
|
|
return TgUser(tg_id=user.id, full_name=user.full_name, username=user.username)
|
|
|
|
|
|
def is_invite_link(event: TelegramObject) -> bool:
|
|
if not isinstance(event, Message) or not event.text:
|
|
return False
|
|
parts = event.text.split(maxsplit=1)
|
|
return len(parts) == 2 and parts[0].split("@")[0] == "/start"
|
|
|
|
|
|
class AccessMiddleware(BaseMiddleware):
|
|
def __init__(self, access: AccessService, texts: Texts, clock: Callable[[], float] = time.monotonic) -> None:
|
|
self._access = access
|
|
self._texts = texts
|
|
self._clock = clock
|
|
# tg_id -> when we last told this outsider "invite only"; answering every message would let them spam us.
|
|
self._last_refusal: dict[int, float] = {}
|
|
|
|
async def __call__(
|
|
self,
|
|
handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]],
|
|
event: TelegramObject,
|
|
data: dict[str, Any],
|
|
) -> Any:
|
|
user: User | None = data.get("event_from_user")
|
|
if user is None:
|
|
return None
|
|
member = await self._access.get_member(user.id)
|
|
if member is not None:
|
|
await self._access.touch(member, tg_user(user))
|
|
data["member"] = member
|
|
return await handler(event, data)
|
|
if is_invite_link(event):
|
|
data["member"] = None
|
|
return await handler(event, data)
|
|
await self._refuse(event, user.id)
|
|
return None
|
|
|
|
async def _refuse(self, event: TelegramObject, tg_id: int) -> None:
|
|
if isinstance(event, CallbackQuery):
|
|
await event.answer() # stop the spinner on a stale button, say nothing
|
|
return
|
|
now = self._clock()
|
|
last = self._last_refusal.get(tg_id)
|
|
if last is not None and now - last < NO_ACCESS_REPLY_EVERY:
|
|
return
|
|
self._last_refusal[tg_id] = now
|
|
if isinstance(event, Message):
|
|
await event.answer(self._texts.start.no_access)
|