From 82b5da41b87a48f5e60b88aae910f3595efb18bc Mon Sep 17 00:00:00 2001 From: goyban Date: Tue, 1 Sep 2026 21:50:01 +0000 Subject: [PATCH] Add public mode with channel membership requirements Completes TODO #2. PUBLIC_MODE swaps the ALLOWED_USERS allowlist for a membership check against REQUIRED_CHATS, with REQUIRE_MODE choosing between needing every chat or just one. - Stop is_admin() falling back to "everyone is an admin" when ADMIN_USERS is unset once PUBLIC_MODE is on. That fallback is the original single-owner convenience, but under public mode it handed moderation, /edit and /delete to anyone who found the bot. Public mode now grants admin to nobody until ADMIN_USERS is set, and says so at startup. Private deployments are unaffected. - Cache membership answers: inline search queries on every keystroke, so an uncached check meant several API calls per typed character. Grants last five minutes, refusals thirty seconds, so a user who has just joined is not locked out for the full window. - Fail closed when the membership check errors, and validate REQUIRED_CHATS at startup so a chat the bot cannot read is reported rather than silently refusing every user. - Refused users get a join prompt with real links instead of silence; inline shows a tappable bar that opens the bot and explains why. A private bot stays silent as before. Co-Authored-By: Claude Opus 5 --- .env.example | 11 +++ README.fa.md | 19 ++++++ README.md | 19 ++++++ VERSION | 2 +- bot.py | 180 ++++++++++++++++++++++++++++++++++++++++++++++--- test_access.py | 165 +++++++++++++++++++++++++++++++++++++++++++++ 6 files changed, 385 insertions(+), 11 deletions(-) create mode 100644 test_access.py diff --git a/.env.example b/.env.example index ff599b0..7e997fd 100644 --- a/.env.example +++ b/.env.example @@ -4,3 +4,14 @@ ADMIN_USERS=123456789 # Let non-admin users submit media for moderator approval (default: off) ACCEPT_SUBMISSIONS=false + +# Open the bot to anyone who meets the membership requirement below, +# instead of using the ALLOWED_USERS list (default: off) +PUBLIC_MODE=false + +# Chats a user must belong to. @username or numeric id, comma-separated. +# The bot must be an admin in each one. Leave empty for a fully open bot. +REQUIRED_CHATS= + +# all = must be in every chat listed above, any = one is enough +REQUIRE_MODE=all diff --git a/README.fa.md b/README.fa.md index 68bd6d6..3a22f93 100644 --- a/README.fa.md +++ b/README.fa.md @@ -63,6 +63,25 @@ docker compose up -d --build | `/delete ` | حذف یک مورد (ID را از `/list` بگیرید) | | `/cancel` | لغو عملیات جاری | +### عمومی‌کردن ربات + +با تنظیم `PUBLIC_MODE=true` ربات به‌جای فهرست `ALLOWED_USERS`، به هر کسی که شرط عضویت را داشته باشد اجازه دسترسی می‌دهد: + +```env +PUBLIC_MODE=true +ADMIN_USERS=123456789 +REQUIRED_CHATS=@mychannel,@mygroup +REQUIRE_MODE=all +``` + +در `REQUIRED_CHATS` می‌توانید `@username` یا شناسه عددی بنویسید. **ربات باید در هر کدام ادمین باشد**، وگرنه نمی‌تواند عضویت را بررسی کند و همه رد می‌شوند. ربات هنگام راه‌اندازی این را بررسی می‌کند و نام هر چتی را که نمی‌تواند بخواند در لاگ می‌نویسد. مقدار `all` یعنی عضویت در همه چت‌ها لازم است و `any` یعنی یکی کافی است. اگر `REQUIRED_CHATS` خالی باشد ربات برای همه باز است. + +کاربری که شرط را ندارد پیامی می‌بیند که فهرست چت‌های لازم را نشان می‌دهد. در جستجوی inline هم به‌جای نتیجه خالی و بدون توضیح، یک نوار قابل لمس می‌بیند که ربات را باز می‌کند و همین توضیح را می‌دهد. + +> **قبل از روشن‌کردن این گزینه حتماً `ADMIN_USERS` را تنظیم کنید.** وقتی `ADMIN_USERS` خالی باشد، ربات خصوصی کاربرانش را ادمین حساب می‌کند؛ این برای ربات تک‌نفره راحت است اما در حالت عمومی یعنی همه ادمین می‌شوند. به همین دلیل در حالت عمومی تا وقتی `ADMIN_USERS` تنظیم نشده باشد هیچ‌کس ادمین نیست و ربات هنگام راه‌اندازی خطا ثبت می‌کند. + +پاسخ بررسی عضویت پنج دقیقه کش می‌شود، چون جستجوی inline در غیر این صورت با هر بار فشردن کلید از تلگرام سؤال می‌کرد. رد شدن فقط سی ثانیه کش می‌شود تا کاربری که تازه عضو شده تقریباً بلافاصله دسترسی پیدا کند. + ### پذیرش ارسال کاربران اگر در فایل `.env` مقدار `ACCEPT_SUBMISSIONS=true` را تنظیم کنید، کاربران غیرادمین هم می‌توانند رسانه بفرستند. این موارد ذخیره می‌شوند اما تا وقتی یک ناظر تأییدشان نکند در جستجوی inline و `/list` دیده نمی‌شوند. رسانه‌ای که خود ادمین‌ها می‌فرستند مثل قبل مستقیم ذخیره می‌شود. diff --git a/README.md b/README.md index 9a76199..e1b2047 100644 --- a/README.md +++ b/README.md @@ -70,6 +70,25 @@ Telegram suggests these commands as you type `/`, with a short description besid | `/delete ` | Remove an entry (get ID from `/list`; admin only) | | `/cancel` | Cancel current add operation | +### Making the bot public + +Set `PUBLIC_MODE=true` to open the bot to anyone who meets a membership requirement, instead of the `ALLOWED_USERS` list: + +```env +PUBLIC_MODE=true +ADMIN_USERS=123456789 +REQUIRED_CHATS=@mychannel,@mygroup +REQUIRE_MODE=all +``` + +`REQUIRED_CHATS` accepts `@username` or numeric ids. **The bot must be an admin in each one**, otherwise it cannot check membership and everyone is refused. The bot verifies this at startup and logs an error naming any chat it cannot read. `REQUIRE_MODE=all` demands every chat; `any` accepts one. With `REQUIRED_CHATS` empty the bot is open to everyone. + +Users who don't qualify get a message listing what to join. In inline search they see a tappable bar that opens the bot and explains the same thing, rather than empty results with no reason given. + +> **Set `ADMIN_USERS` before turning this on.** With no `ADMIN_USERS`, a private bot treats its users as admins — convenient for a one-person bot, but in public mode that would mean everyone. Public mode therefore grants admin to nobody until `ADMIN_USERS` is set, and logs an error at startup if you forget. + +Membership answers are cached for five minutes, since inline search would otherwise query Telegram on every keystroke. Refusals are cached for only thirty seconds, so a user who has just joined gets in almost immediately. + ### Accepting submissions Set `ACCEPT_SUBMISSIONS=true` in `.env` to let non-admin users send media too. Their submissions are stored but stay invisible — they don't appear in inline search or `/list` until a moderator approves them. Admins' own media is still saved directly. diff --git a/VERSION b/VERSION index 26aaba0..f0bb29e 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.2.0 +1.3.0 diff --git a/bot.py b/bot.py index 8049060..253dcc2 100644 --- a/bot.py +++ b/bot.py @@ -1,9 +1,11 @@ import asyncio import logging import os +import time import uuid from aiogram import Bot, Dispatcher, F +from aiogram.exceptions import TelegramAPIError from aiogram.filters import Command, StateFilter from aiogram.fsm.context import FSMContext from aiogram.fsm.state import State, StatesGroup @@ -23,6 +25,7 @@ from aiogram.types import ( InlineQueryResultCachedVideo, InlineQueryResultCachedVoice, InlineQueryResultArticle, + InlineQueryResultsButton, InputTextMessageContent, Message, ) @@ -41,9 +44,33 @@ ADMIN_USERS: set[int] = { int(uid.strip()) for uid in os.environ.get("ADMIN_USERS", "").split(",") if uid.strip() } +def _env_flag(name: str) -> bool: + return os.environ.get(name, "").strip().lower() in {"1", "true", "yes"} + + # Off by default: with ALLOWED_USERS unset, is_allowed() lets everyone in, so # opening submissions implicitly would accept media from the whole world. -ACCEPT_SUBMISSIONS = os.environ.get("ACCEPT_SUBMISSIONS", "").strip().lower() in {"1", "true", "yes"} +ACCEPT_SUBMISSIONS = _env_flag("ACCEPT_SUBMISSIONS") + +# Public mode swaps the ALLOWED_USERS allowlist for "is this user in the chats +# I require?". With no REQUIRED_CHATS the bot is simply open to everyone. +PUBLIC_MODE = _env_flag("PUBLIC_MODE") + + +def _parse_chat(raw: str) -> int | str: + """Telegram takes either @username or a numeric id.""" + raw = raw.strip() + try: + return int(raw) + except ValueError: + return raw + + +REQUIRED_CHATS: list[int | str] = [ + _parse_chat(c) for c in os.environ.get("REQUIRED_CHATS", "").split(",") if c.strip() +] +# "all" (default): must be in every listed chat. "any": one is enough. +REQUIRE_MODE = os.environ.get("REQUIRE_MODE", "all").strip().lower() bot = Bot(token=BOT_TOKEN) dp = Dispatcher(storage=MemoryStorage()) @@ -69,12 +96,126 @@ def is_allowed(user_id: int) -> bool: def is_admin(user_id: int) -> bool: - # If no ADMIN_USERS set, fall back to ALLOWED_USERS behaviour (backward compat) if ADMIN_USERS: return user_id in ADMIN_USERS + # An unconfigured private bot treats its users as admins, which is the + # original single-owner behaviour. That fallback must not survive into + # public mode, where it would hand admin rights to anyone who found the + # bot. check_admin_config() flags the missing setting at startup. + if PUBLIC_MODE: + return False return not ALLOWED_USERS or user_id in ALLOWED_USERS +# Inline queries fire on every keystroke, so an uncached membership check would +# mean several API calls per typed character. Denials expire quickly so that a +# user who has just joined is not left waiting for the full window. +MEMBERSHIP_TTL_ALLOWED = 300.0 +MEMBERSHIP_TTL_DENIED = 30.0 + +ACCEPTED_MEMBER_STATUSES = {"creator", "administrator", "member"} + +_membership_cache: dict[tuple[int, int | str], tuple[bool, float]] = {} + + +async def _fetch_membership(user_id: int, chat: int | str) -> bool: + try: + member = await bot.get_chat_member(chat, user_id) + except TelegramAPIError: + # Usually the bot is not an admin in that chat, or the user is unknown + # to it. Fail closed; check_required_chats() reports setup problems at + # startup so this does not fail closed silently forever. + log.warning("membership check failed for user %s in chat %s", user_id, chat, exc_info=True) + return False + + if member.status in ACCEPTED_MEMBER_STATUSES: + return True + # A restricted user is still a member unless they have actually left. + if member.status == "restricted": + return bool(getattr(member, "is_member", False)) + return False + + +async def is_member_of(user_id: int, chat: int | str) -> bool: + key = (user_id, chat) + cached = _membership_cache.get(key) + now = time.monotonic() + if cached is not None and cached[1] > now: + return cached[0] + + result = await _fetch_membership(user_id, chat) + ttl = MEMBERSHIP_TTL_ALLOWED if result else MEMBERSHIP_TTL_DENIED + _membership_cache[key] = (result, now + ttl) + return result + + +async def passes_membership(user_id: int) -> bool: + if not REQUIRED_CHATS: + return True + if REQUIRE_MODE == "any": + for chat in REQUIRED_CHATS: + if await is_member_of(user_id, chat): + return True + return False + for chat in REQUIRED_CHATS: + if not await is_member_of(user_id, chat): + return False + return True + + +async def has_access(user_id: int) -> bool: + """Public mode replaces the allowlist with the membership requirement.""" + if is_admin(user_id): + return True + if PUBLIC_MODE: + return await passes_membership(user_id) + return is_allowed(user_id) + + +# Resolved once at startup so the join prompt can show real links. +REQUIRED_CHAT_LINKS: list[tuple[str, str | None]] = [] + + +def check_admin_config() -> None: + if PUBLIC_MODE and not ADMIN_USERS: + log.error( + "PUBLIC_MODE is on but ADMIN_USERS is empty — nobody can moderate, " + "add or delete. Set ADMIN_USERS to your Telegram user ID." + ) + + +async def check_required_chats() -> None: + """Verify the bot can actually see each required chat, and remember how to + link to it. A misconfigured chat would otherwise lock out every user with + nothing in the logs to explain why.""" + REQUIRED_CHAT_LINKS.clear() + for chat in REQUIRED_CHATS: + try: + info = await bot.get_chat(chat) + except TelegramAPIError: + log.error( + "REQUIRED_CHATS: cannot read %s — add the bot to that chat as an " + "admin, or users will not be able to pass the membership check.", + chat, exc_info=True, + ) + REQUIRED_CHAT_LINKS.append((str(chat), None)) + continue + url = f"https://t.me/{info.username}" if info.username else info.invite_link + REQUIRED_CHAT_LINKS.append((info.title or str(chat), url)) + if url is None: + log.warning("REQUIRED_CHATS: no public link for %s; users will only see its name", chat) + + +def join_prompt() -> str: + if not REQUIRED_CHAT_LINKS: + return "You do not have access to this bot." + joiner = "any one of" if REQUIRE_MODE == "any" else "all of" + lines = [f"To use this bot you need to join {joiner} these:"] + for title, url in REQUIRED_CHAT_LINKS: + lines.append(f"• {title}" if url else f"• {title}") + return "\n".join(lines) + + async def register_commands() -> None: """Publish Telegram's command menu with role-appropriate suggestions.""" if not ADMIN_USERS: @@ -120,11 +261,19 @@ def menu_text(user_id: int) -> str: ) +async def deny(msg: Message) -> None: + """A public bot tells the user how to get in; a private one stays silent, + as it always has, rather than confirming to strangers that it exists.""" + if PUBLIC_MODE: + await msg.answer(join_prompt(), parse_mode="HTML") + + # ── /start ─────────────────────────────────────────────────────────────────── @dp.message(Command("start")) async def cmd_start(msg: Message): - if not is_allowed(msg.from_user.id): + if not await has_access(msg.from_user.id): + await deny(msg) return await msg.answer( f"{menu_text(msg.from_user.id)}\n/menu — show this list again", @@ -136,7 +285,8 @@ async def cmd_start(msg: Message): @dp.message(Command("menu")) async def cmd_menu(msg: Message): - if not is_allowed(msg.from_user.id): + if not await has_access(msg.from_user.id): + await deny(msg) return await msg.answer(menu_text(msg.from_user.id), parse_mode="HTML") @@ -145,7 +295,7 @@ async def cmd_menu(msg: Message): @dp.message(Command("cancel")) async def cmd_cancel(msg: Message, state: FSMContext): - if not is_allowed(msg.from_user.id): + if not await has_access(msg.from_user.id): return await state.clear() await msg.answer("Cancelled.") @@ -205,7 +355,7 @@ async def on_media(msg: Message, state: FSMContext): user_id = msg.from_user.id if is_admin(user_id): status = "approved" - elif ACCEPT_SUBMISSIONS and is_allowed(user_id): + elif ACCEPT_SUBMISSIONS and await has_access(user_id): status = "pending" else: return @@ -289,7 +439,8 @@ async def list_page(page: int) -> tuple[str, InlineKeyboardMarkup | None]: @dp.message(Command("list")) async def cmd_list(msg: Message): - if not is_allowed(msg.from_user.id): + if not await has_access(msg.from_user.id): + await deny(msg) return text, keyboard = await list_page(0) await msg.answer(text, parse_mode="HTML", reply_markup=keyboard) @@ -297,7 +448,7 @@ async def cmd_list(msg: Message): @dp.callback_query(F.data.startswith("list:")) async def on_list_page(callback: CallbackQuery): - if not is_allowed(callback.from_user.id): + if not await has_access(callback.from_user.id): await callback.answer("Not allowed.", show_alert=True) return @@ -559,8 +710,15 @@ INLINE_BUILDERS = { @dp.inline_query() async def inline_handler(query: InlineQuery): - if not is_allowed(query.from_user.id): - await query.answer([], cache_time=1) + if not await has_access(query.from_user.id): + # A bar above the results beats silently returning nothing: tapping it + # opens the bot in private chat, where /start explains what to join. + button = None + if PUBLIC_MODE and REQUIRED_CHATS: + button = InlineQueryResultsButton( + text="Tap to see what you need to join", start_parameter="join" + ) + await query.answer([], cache_time=1, is_personal=True, button=button) return try: @@ -594,6 +752,8 @@ async def inline_handler(query: InlineQuery): async def main(): await db.init_db() + check_admin_config() + await check_required_chats() await register_commands() log.info("Bot starting…") await dp.start_polling(bot) diff --git a/test_access.py b/test_access.py new file mode 100644 index 0000000..de9c0f0 --- /dev/null +++ b/test_access.py @@ -0,0 +1,165 @@ +"""Access-control tests. Run with: python test_access.py + +Covers who gets into the bot in public mode, and the membership cache that +keeps inline search from hammering the Telegram API on every keystroke. +""" +import asyncio +import importlib +import logging +import os +import sys +import time + +os.environ.setdefault("BOT_TOKEN", "123456789:AAHtestTokenPlaceholder_ABCDEFGHIJKLMNOP") +os.environ.setdefault("DB_PATH", ":memory:") +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from aiogram.exceptions import TelegramAPIError + +# Several cases deliberately make the API fail; the bot logs those with a +# traceback, which is correct in production and only noise here. +logging.disable(logging.CRITICAL) + +CONFIG_KEYS = ("PUBLIC_MODE", "REQUIRED_CHATS", "REQUIRE_MODE", + "ALLOWED_USERS", "ADMIN_USERS", "ACCEPT_SUBMISSIONS") + + +class FakeMember: + def __init__(self, status: str, is_member: bool | None = None): + self.status = status + if is_member is not None: + self.is_member = is_member + + +def load_bot(**env): + """Reload the bot module with a fresh environment; config is read at import.""" + for key in CONFIG_KEYS: + os.environ.pop(key, None) + os.environ.update(env) + import bot + return importlib.reload(bot) + + +def stub_membership(B, members: dict, calls: list | None = None): + async def get_chat_member(chat, user_id): + if calls is not None: + calls.append((chat, user_id)) + if (chat, user_id) not in members: + raise TelegramAPIError(method=None, message="user not found") + return members[(chat, user_id)] + B.bot.get_chat_member = get_chat_member + + +async def test_member_statuses(): + B = load_bot(PUBLIC_MODE="true", REQUIRED_CHATS="@chan") + allowed = ["creator", "administrator", "member"] + denied = ["left", "kicked"] + + for status in allowed: + B._membership_cache.clear() + stub_membership(B, {("@chan", 1): FakeMember(status)}) + assert await B.has_access(1) is True, f"{status} should be allowed" + for status in denied: + B._membership_cache.clear() + stub_membership(B, {("@chan", 1): FakeMember(status)}) + assert await B.has_access(1) is False, f"{status} should be denied" + + # A restricted user is still a member unless they have actually left. + B._membership_cache.clear() + stub_membership(B, {("@chan", 1): FakeMember("restricted", is_member=True)}) + assert await B.has_access(1) is True, "restricted but still in the chat" + B._membership_cache.clear() + stub_membership(B, {("@chan", 1): FakeMember("restricted", is_member=False)}) + assert await B.has_access(1) is False, "restricted and gone" + print(" member statuses are classified correctly, including restricted") + + +async def test_require_all_and_any(): + members = {("@a", 1): FakeMember("member")} # in @a, not in @b + + B = load_bot(PUBLIC_MODE="true", REQUIRED_CHATS="@a,@b", REQUIRE_MODE="all") + stub_membership(B, members) + assert await B.has_access(1) is False, "all mode needs every chat" + + B = load_bot(PUBLIC_MODE="true", REQUIRED_CHATS="@a,@b", REQUIRE_MODE="any") + stub_membership(B, members) + assert await B.has_access(1) is True, "any mode needs just one" + print(" REQUIRE_MODE all and any behave differently as configured") + + +async def test_cache_absorbs_repeat_checks(): + B = load_bot(PUBLIC_MODE="true", REQUIRED_CHATS="@chan") + calls: list = [] + stub_membership(B, {("@chan", 1): FakeMember("member")}, calls) + + for _ in range(10): + assert await B.has_access(1) is True + assert len(calls) == 1, f"10 inline keystrokes caused {len(calls)} API calls" + print(" repeated checks hit the API once, not once per keystroke") + + +async def test_denials_expire_sooner_than_grants(): + B = load_bot(PUBLIC_MODE="true", REQUIRED_CHATS="@chan") + assert B.MEMBERSHIP_TTL_DENIED < B.MEMBERSHIP_TTL_ALLOWED + + stub_membership(B, {("@chan", 1): FakeMember("member")}) + await B.has_access(1) + await B.has_access(2) # unknown user -> denied + + now = time.monotonic() + granted = B._membership_cache[(1, "@chan")][1] - now + denied = B._membership_cache[(2, "@chan")][1] - now + assert denied < granted, "a user who just joined should not wait the full window" + print(" denials expire quickly so joining takes effect promptly") + + +async def test_admin_bypasses_membership(): + B = load_bot(PUBLIC_MODE="true", REQUIRED_CHATS="@chan", ADMIN_USERS="111") + calls: list = [] + stub_membership(B, {}, calls) # nobody is a member + + assert await B.has_access(111) is True, "admins must not be locked out" + assert calls == [], "admin check should not call the API at all" + assert await B.has_access(222) is False + print(" admins bypass the membership check without an API call") + + +async def test_private_mode_still_uses_the_allowlist(): + B = load_bot(ALLOWED_USERS="111,222") + calls: list = [] + stub_membership(B, {}, calls) + + assert await B.has_access(111) is True + assert await B.has_access(333) is False + assert calls == [], "a private bot should never call getChatMember" + print(" with PUBLIC_MODE off, the allowlist behaves exactly as before") + + +async def test_api_failure_fails_closed(): + B = load_bot(PUBLIC_MODE="true", REQUIRED_CHATS="@chan") + + async def always_fails(chat, user_id): + raise TelegramAPIError(method=None, message="bot is not an admin there") + B.bot.get_chat_member = always_fails + + assert await B.has_access(1) is False, "an unreadable chat must not open the gate" + print(" a failing membership check denies access rather than opening up") + + +async def test_open_public_bot(): + B = load_bot(PUBLIC_MODE="true") # no REQUIRED_CHATS + assert await B.has_access(999) is True + print(" public mode with no required chats is open to everyone") + + +async def main(): + for test in (test_member_statuses, test_require_all_and_any, + test_cache_absorbs_repeat_checks, test_denials_expire_sooner_than_grants, + test_admin_bypasses_membership, test_private_mode_still_uses_the_allowlist, + test_api_failure_fails_closed, test_open_public_bot): + await test() + print("\nAll access tests passed.") + + +if __name__ == "__main__": + asyncio.run(main())