mirror of
https://github.com/PaiGramTeam/PaiGram.git
synced 2024-11-25 18:04:10 +00:00
446 lines
22 KiB
Python
446 lines
22 KiB
Python
import asyncio
|
|
import random
|
|
import time
|
|
from typing import Tuple, Union, Dict, Optional
|
|
|
|
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup, ChatPermissions, ChatMember, Message, User
|
|
from telegram.constants import ParseMode
|
|
from telegram.error import BadRequest
|
|
from telegram.ext import CallbackContext, CallbackQueryHandler, ChatMemberHandler, filters
|
|
from telegram.helpers import escape_markdown
|
|
|
|
from core.config import config
|
|
from core.dependence.mtproto import MTProto
|
|
from core.dependence.redisdb import RedisDB
|
|
from core.plugin import Plugin, handler
|
|
from core.services.quiz.services import QuizService
|
|
from utils.chatmember import extract_status_change
|
|
from utils.log import logger
|
|
|
|
try:
|
|
from pyrogram.errors import BadRequest as MTPBadRequest, FloodWait as MTPFloodWait
|
|
|
|
PYROGRAM_AVAILABLE = True
|
|
except ImportError:
|
|
MTPBadRequest = ValueError
|
|
MTPFloodWait = IndexError
|
|
PYROGRAM_AVAILABLE = False
|
|
|
|
try:
|
|
import ujson as jsonlib
|
|
|
|
except ImportError:
|
|
import json as jsonlib
|
|
|
|
FullChatPermissions = ChatPermissions(
|
|
can_send_messages=True,
|
|
can_send_media_messages=True,
|
|
can_send_polls=True,
|
|
can_send_other_messages=True,
|
|
can_add_web_page_previews=True,
|
|
can_change_info=True,
|
|
can_invite_users=True,
|
|
can_pin_messages=True,
|
|
)
|
|
|
|
|
|
class GroupCaptcha(Plugin):
|
|
"""群验证模块"""
|
|
|
|
def __init__(self, quiz_service: QuizService = None, mtp: MTProto = None, redis: RedisDB = None):
|
|
self.quiz_service = quiz_service
|
|
self.time_out = 120
|
|
self.kick_time = 120
|
|
self.lock = asyncio.Lock()
|
|
self.chat_administrators_cache: Dict[Union[str, int], Tuple[float, Tuple[ChatMember]]] = {}
|
|
self.is_refresh_quiz = False
|
|
self.mtp = mtp.client
|
|
self.redis = redis.client
|
|
|
|
async def initialize(self):
|
|
logger.info("群验证模块正在刷新问题列表")
|
|
await self.refresh_quiz()
|
|
logger.success("群验证模块刷新问题列表成功")
|
|
|
|
async def refresh_quiz(self):
|
|
async with self.lock:
|
|
if not self.is_refresh_quiz:
|
|
await self.quiz_service.refresh_quiz()
|
|
self.is_refresh_quiz = True
|
|
|
|
async def get_chat_administrators(self, context: CallbackContext, chat_id: Union[str, int]) -> Tuple[ChatMember]:
|
|
async with self.lock:
|
|
cache_data = self.chat_administrators_cache.get(f"{chat_id}")
|
|
if cache_data is not None:
|
|
cache_time, chat_administrators = cache_data
|
|
if time.time() >= cache_time + 360:
|
|
return chat_administrators
|
|
chat_administrators = await context.bot.get_chat_administrators(chat_id)
|
|
self.chat_administrators_cache[f"{chat_id}"] = (time.time(), chat_administrators)
|
|
return chat_administrators
|
|
|
|
@staticmethod
|
|
def is_admin(chat_administrators: Tuple[ChatMember], user_id: int) -> bool:
|
|
return any(admin.user.id == user_id for admin in chat_administrators)
|
|
|
|
async def kick_member_job(self, context: CallbackContext):
|
|
job = context.job
|
|
logger.info("踢出用户 user_id[%s] 在 chat_id[%s]", job.user_id, job.chat_id)
|
|
try:
|
|
await context.bot.ban_chat_member(
|
|
chat_id=job.chat_id, user_id=job.user_id, until_date=int(time.time()) + self.kick_time
|
|
)
|
|
except BadRequest as exc:
|
|
logger.error("GroupCaptcha插件在 chat_id[%s] user_id[%s] 执行kick失败", job.chat_id, job.user_id, exc_info=exc)
|
|
|
|
@staticmethod
|
|
async def clean_message_job(context: CallbackContext):
|
|
job = context.job
|
|
logger.debug("删除消息 chat_id[%s] 的 message_id[%s]", job.chat_id, job.data)
|
|
try:
|
|
await context.bot.delete_message(chat_id=job.chat_id, message_id=job.data)
|
|
except BadRequest as exc:
|
|
if "not found" in exc.message:
|
|
logger.warning("GroupCaptcha插件删除消息 chat_id[%s] message_id[%s]失败 消息不存在", job.chat_id, job.data)
|
|
elif "Message can't be deleted" in exc.message:
|
|
logger.warning("GroupCaptcha插件删除消息 chat_id[%s] message_id[%s]失败 消息无法删除 可能是没有授权", job.chat_id, job.data)
|
|
else:
|
|
logger.error("GroupCaptcha插件删除消息 chat_id[%s] message_id[%s]失败", job.chat_id, job.data, exc_info=exc)
|
|
|
|
@staticmethod
|
|
async def restore_member(context: CallbackContext, chat_id: int, user_id: int):
|
|
logger.debug("重置用户权限 user_id[%s] 在 chat_id[%s]", chat_id, user_id)
|
|
try:
|
|
await context.bot.restrict_chat_member(chat_id=chat_id, user_id=user_id, permissions=FullChatPermissions)
|
|
except BadRequest as exc:
|
|
logger.error("GroupCaptcha插件在 chat_id[%s] user_id[%s] 执行restore失败", chat_id, user_id, exc_info=exc)
|
|
|
|
async def get_new_chat_members_message(self, user: User, context: CallbackContext) -> Optional[Message]:
|
|
qname = f"plugin:auth:new_chat_members_message:{user.id}"
|
|
result = await self.redis.get(qname)
|
|
if result:
|
|
data = jsonlib.loads(str(result, encoding="utf-8"))
|
|
return Message.de_json(data, context.bot)
|
|
return None
|
|
|
|
async def set_new_chat_members_message(self, user: User, message: Message):
|
|
qname = f"plugin:auth:new_chat_members_message:{user.id}"
|
|
await self.redis.set(qname, message.to_json(), ex=60)
|
|
|
|
@handler(CallbackQueryHandler, pattern=r"^auth_admin\|", block=False)
|
|
async def admin(self, update: Update, context: CallbackContext) -> None:
|
|
async def admin_callback(callback_query_data: str) -> Tuple[str, int]:
|
|
_data = callback_query_data.split("|")
|
|
_result = _data[1]
|
|
_user_id = int(_data[2])
|
|
logger.debug("admin_callback函数返回 result[%s] user_id[%s]", _result, _user_id)
|
|
return _result, _user_id
|
|
|
|
callback_query = update.callback_query
|
|
user = callback_query.from_user
|
|
message = callback_query.message
|
|
chat = message.chat
|
|
logger.info("用户 %s[%s] 在群 %s[%s] 点击Auth管理员命令", user.full_name, user.id, chat.title, chat.id)
|
|
chat_administrators = await self.get_chat_administrators(context, chat_id=chat.id)
|
|
if not self.is_admin(chat_administrators, user.id):
|
|
logger.debug("用户 %s[%s] 在群 %s[%s] 非群管理", user.full_name, user.id, chat.title, chat.id)
|
|
await callback_query.answer(text="你不是管理！\n" + config.notice.user_mismatch, show_alert=True)
|
|
return
|
|
result, user_id = await admin_callback(callback_query.data)
|
|
try:
|
|
member_info = await context.bot.get_chat_member(chat.id, user_id)
|
|
except BadRequest as error:
|
|
logger.warning("获取用户 %s 在群 %s[%s] 信息失败 \n %s", user_id, chat.title, chat.id, error.message)
|
|
user_info = f"{user_id}"
|
|
else:
|
|
user_info = member_info.user.mention_markdown_v2()
|
|
|
|
if result == "pass":
|
|
await callback_query.answer(text="放行", show_alert=False)
|
|
await self.restore_member(context, chat.id, user_id)
|
|
if schedule := context.job_queue.scheduler.get_job(f"{chat.id}|{user_id}|auth_kick"):
|
|
schedule.remove()
|
|
await message.edit_text(f"{user_info} 被 {user.mention_markdown_v2()} 放行", parse_mode=ParseMode.MARKDOWN_V2)
|
|
logger.info("用户 %s[%s] 在群 %s[%s] 被管理放行", user.full_name, user.id, chat.title, chat.id)
|
|
elif result == "kick":
|
|
await callback_query.answer(text="驱离", show_alert=False)
|
|
await context.bot.ban_chat_member(chat.id, user_id)
|
|
await message.edit_text(f"{user_info} 被 {user.mention_markdown_v2()} 驱离", parse_mode=ParseMode.MARKDOWN_V2)
|
|
logger.info("用户 %s[%s] 在群 %s[%s] 被管理踢出", user.full_name, user.id, chat.title, chat.id)
|
|
elif result == "unban":
|
|
await callback_query.answer(text="解除驱离", show_alert=False)
|
|
await self.restore_member(context, chat.id, user_id)
|
|
if schedule := context.job_queue.scheduler.get_job(f"{chat.id}|{user_id}|auth_kick"):
|
|
schedule.remove()
|
|
await message.edit_text(
|
|
f"{user_info} 被 {user.mention_markdown_v2()} 解除驱离", parse_mode=ParseMode.MARKDOWN_V2
|
|
)
|
|
logger.info("用户 user_id[%s] 在群 %s[%s] 被管理解除封禁", user_id, chat.title, chat.id)
|
|
else:
|
|
logger.warning("auth 模块 admin 函数 发现未知命令 result[%s]", result)
|
|
await context.bot.send_message(chat.id, "派蒙这边收到了错误的消息！请检查详细日记！")
|
|
if schedule := context.job_queue.scheduler.get_job(f"{chat.id}|{user_id}|auth_kick"):
|
|
schedule.remove()
|
|
|
|
@handler(CallbackQueryHandler, pattern=r"^auth_challenge\|", block=False)
|
|
async def query(self, update: Update, context: CallbackContext) -> None:
|
|
async def query_callback(callback_query_data: str) -> Tuple[int, bool, str, str]:
|
|
_data = callback_query_data.split("|")
|
|
_user_id = int(_data[1])
|
|
_question_id = int(_data[2])
|
|
_answer_id = int(_data[3])
|
|
_answer = await self.quiz_service.get_answer(_answer_id)
|
|
_question = await self.quiz_service.get_question(_question_id)
|
|
_result = _answer.is_correct
|
|
_answer_encode = _answer.text
|
|
_question_encode = _question.text
|
|
logger.debug(
|
|
"query_callback函数返回 user_id[%s] result[%s] \nquestion_encode[%s] answer_encode[%s]",
|
|
_user_id,
|
|
_result,
|
|
_question_encode,
|
|
_answer_encode,
|
|
)
|
|
return _user_id, _result, _question_encode, _answer_encode
|
|
|
|
callback_query = update.callback_query
|
|
user = callback_query.from_user
|
|
message = callback_query.message
|
|
chat = message.chat
|
|
user_id, result, question, answer = await query_callback(callback_query.data)
|
|
logger.info("用户 %s[%s] 在群 %s[%s] 点击Auth认证命令", user.full_name, user.id, chat.title, chat.id)
|
|
if user.id != user_id:
|
|
await callback_query.answer(text="这不是你的验证！\n" + config.notice.user_mismatch, show_alert=True)
|
|
return
|
|
logger.info(
|
|
"用户 %s[%s] 在群 %s[%s] 认证结果为 %s", user.full_name, user.id, chat.title, chat.id, "通过" if result else "失败"
|
|
)
|
|
if result:
|
|
buttons = [[InlineKeyboardButton("驱离", callback_data=f"auth_admin|kick|{user.id}")]]
|
|
await callback_query.answer(text="验证成功", show_alert=False)
|
|
await self.restore_member(context, chat.id, user_id)
|
|
if schedule := context.job_queue.scheduler.get_job(f"{chat.id}|{user.id}|auth_kick"):
|
|
schedule.remove()
|
|
text = (
|
|
f"{user.mention_markdown_v2()} 验证成功，向着星辰与深渊！\n"
|
|
f"问题：{escape_markdown(question, version=2)} \n"
|
|
f"回答：{escape_markdown(answer, version=2)}"
|
|
)
|
|
logger.info("用户 user_id[%s] 在群 %s[%s] 验证成功", user_id, chat.title, chat.id)
|
|
else:
|
|
buttons = [
|
|
[
|
|
InlineKeyboardButton("驱离", callback_data=f"auth_admin|kick|{user.id}"),
|
|
InlineKeyboardButton("撤回驱离", callback_data=f"auth_admin|unban|{user.id}"),
|
|
]
|
|
]
|
|
await callback_query.answer(text=f"验证失败，请在 {self.time_out} 秒后重试", show_alert=True)
|
|
await asyncio.sleep(3)
|
|
await context.bot.ban_chat_member(
|
|
chat_id=chat.id, user_id=user_id, until_date=int(time.time()) + self.kick_time
|
|
)
|
|
text = (
|
|
f"{user.mention_markdown_v2()} 验证失败，已经赶出提瓦特大陆！\n"
|
|
f"问题：{escape_markdown(question, version=2)} \n"
|
|
f"回答：{escape_markdown(answer, version=2)}"
|
|
)
|
|
logger.info("用户 user_id[%s] 在群 %s[%s] 验证失败", user_id, chat.title, chat.id)
|
|
try:
|
|
await message.edit_text(text, reply_markup=InlineKeyboardMarkup(buttons), parse_mode=ParseMode.MARKDOWN_V2)
|
|
except BadRequest as exc:
|
|
if "are exactly the same as " in exc.message:
|
|
logger.warning("编辑消息发生异常，可能为用户点按多次键盘导致")
|
|
else:
|
|
raise exc
|
|
if schedule := context.job_queue.scheduler.get_job(f"{chat.id}|{user.id}|auth_kick"):
|
|
schedule.remove()
|
|
|
|
@handler.message(filters=filters.StatusUpdate.NEW_CHAT_MEMBERS, block=False)
|
|
async def new_mem(self, update: Update, context: CallbackContext) -> None:
|
|
message = update.effective_message
|
|
chat = message.chat
|
|
if len(config.verify_groups) >= 1:
|
|
for verify_group in config.verify_groups:
|
|
if verify_group == chat.id:
|
|
break
|
|
else:
|
|
return
|
|
else:
|
|
return
|
|
for user in message.new_chat_members:
|
|
if user.id == context.bot.id:
|
|
return
|
|
logger.debug("用户 %s[%s] 加入群 %s[%s]", user.full_name, user.id, chat.title, chat.id)
|
|
await self.set_new_chat_members_message(user, message)
|
|
|
|
@handler.chat_member(chat_member_types=ChatMemberHandler.CHAT_MEMBER, block=False)
|
|
async def track_users(self, update: Update, context: CallbackContext) -> None:
|
|
chat = update.effective_chat
|
|
if len(config.verify_groups) >= 1:
|
|
for verify_group in config.verify_groups:
|
|
if verify_group == chat.id:
|
|
break
|
|
else:
|
|
return
|
|
else:
|
|
return
|
|
new_chat_member = update.chat_member.new_chat_member
|
|
from_user = update.chat_member.from_user
|
|
user = new_chat_member.user
|
|
result = extract_status_change(update.chat_member)
|
|
if result is None:
|
|
return
|
|
was_member, is_member = result
|
|
if was_member and not is_member:
|
|
logger.info("用户 %s[%s] 退出群聊 %s[%s]", user.full_name, user.id, chat.title, chat.id)
|
|
return
|
|
if not was_member and is_member:
|
|
logger.info("用户 %s[%s] 尝试加入群 %s[%s]", user.full_name, user.id, chat.title, chat.id)
|
|
if user.is_bot:
|
|
return
|
|
chat_administrators = await self.get_chat_administrators(context, chat_id=chat.id)
|
|
if self.is_admin(chat_administrators, from_user.id):
|
|
await chat.send_message("派蒙检测到管理员邀请，自动放行了！")
|
|
return
|
|
question_id_list = await self.quiz_service.get_question_id_list()
|
|
if len(question_id_list) == 0:
|
|
await chat.send_message("旅行者！！！派蒙的问题清单你还没给我！！快去私聊我给我问题！")
|
|
return
|
|
try:
|
|
await chat.restrict_member(user_id=user.id, permissions=ChatPermissions(can_send_messages=False))
|
|
except BadRequest as exc:
|
|
if "Not enough rights" in exc.message:
|
|
logger.warning("%s[%s] 权限不够", chat.title, chat.id)
|
|
await chat.send_message(
|
|
f"派蒙无法修改 {user.mention_html()} 的权限！请检查是否给派蒙授权管理了",
|
|
parse_mode=ParseMode.HTML,
|
|
)
|
|
return
|
|
raise exc
|
|
new_chat_members_message = await self.get_new_chat_members_message(user, context)
|
|
question_id = random.choice(question_id_list) # nosec
|
|
question = await self.quiz_service.get_question(question_id)
|
|
buttons = [
|
|
[
|
|
InlineKeyboardButton(
|
|
answer.text,
|
|
callback_data=f"auth_challenge|{user.id}|{question.question_id}|{answer.answer_id}",
|
|
)
|
|
]
|
|
for answer in question.answers
|
|
]
|
|
random.shuffle(buttons)
|
|
buttons.append(
|
|
[
|
|
InlineKeyboardButton(
|
|
"放行",
|
|
callback_data=f"auth_admin|pass|{user.id}",
|
|
),
|
|
InlineKeyboardButton(
|
|
"驱离",
|
|
callback_data=f"auth_admin|kick|{user.id}",
|
|
),
|
|
]
|
|
)
|
|
if new_chat_members_message:
|
|
reply_message = (
|
|
f"*欢迎来到「提瓦特」世界！* \n"
|
|
f"问题: {escape_markdown(question.text, version=2)} \n"
|
|
f"请在*{self.time_out}*秒内回答问题"
|
|
)
|
|
else:
|
|
reply_message = (
|
|
f"*欢迎 {user.mention_markdown_v2()} 来到「提瓦特」世界！* \n"
|
|
f"问题: {escape_markdown(question.text, version=2)} \n"
|
|
f"请在*{self.time_out}*秒内回答问题"
|
|
)
|
|
logger.debug(
|
|
"发送入群验证问题 %s[%s] \n给%s[%s] 在 %s[%s]",
|
|
question.text,
|
|
question.question_id,
|
|
user.full_name,
|
|
user.id,
|
|
chat.title,
|
|
chat.id,
|
|
)
|
|
try:
|
|
if new_chat_members_message:
|
|
question_message = await new_chat_members_message.reply_markdown_v2(
|
|
reply_message, reply_markup=InlineKeyboardMarkup(buttons), allow_sending_without_reply=True
|
|
)
|
|
else:
|
|
question_message = await chat.send_message(
|
|
reply_message,
|
|
reply_markup=InlineKeyboardMarkup(buttons),
|
|
parse_mode=ParseMode.MARKDOWN_V2,
|
|
)
|
|
except BadRequest as exc:
|
|
await chat.send_message("派蒙分心了一下，不小心忘记你了，你只能先退出群再重新进来吧。")
|
|
raise exc
|
|
context.job_queue.run_once(
|
|
callback=self.kick_member_job,
|
|
when=self.time_out,
|
|
name=f"{chat.id}|{user.id}|auth_kick",
|
|
chat_id=chat.id,
|
|
user_id=user.id,
|
|
job_kwargs={"replace_existing": True, "id": f"{chat.id}|{user.id}|auth_kick"},
|
|
)
|
|
if new_chat_members_message:
|
|
context.job_queue.run_once(
|
|
callback=self.clean_message_job,
|
|
when=self.time_out,
|
|
data=new_chat_members_message.message_id,
|
|
name=f"{chat.id}|{user.id}|auth_clean_join_message",
|
|
chat_id=chat.id,
|
|
user_id=user.id,
|
|
job_kwargs={"replace_existing": True, "id": f"{chat.id}|{user.id}|auth_clean_join_message"},
|
|
)
|
|
context.job_queue.run_once(
|
|
callback=self.clean_message_job,
|
|
when=self.time_out,
|
|
data=question_message.message_id,
|
|
name=f"{chat.id}|{user.id}|auth_clean_question_message",
|
|
chat_id=chat.id,
|
|
user_id=user.id,
|
|
job_kwargs={"replace_existing": True, "id": f"{chat.id}|{user.id}|auth_clean_question_message"},
|
|
)
|
|
if PYROGRAM_AVAILABLE and self.mtp:
|
|
try:
|
|
if new_chat_members_message:
|
|
if question_message.id - new_chat_members_message.id - 1:
|
|
message_ids = list(range(new_chat_members_message.id + 1, question_message.id))
|
|
else:
|
|
return
|
|
else:
|
|
message_ids = [question_message.id - 3, question_message.id]
|
|
messages_list = await self.mtp.get_messages(chat.id, message_ids=message_ids)
|
|
for find_message in messages_list:
|
|
if find_message.empty:
|
|
continue
|
|
if find_message.from_user and find_message.from_user.id == user.id:
|
|
await self.mtp.delete_messages(chat_id=chat.id, message_ids=find_message.id)
|
|
text: Optional[str] = None
|
|
if find_message.text and "@" in find_message.text:
|
|
text = f"{user.full_name} 由于加入群组后，在验证缝隙间发送了带有 @(Mention) 的消息，已被踢出群组，并加入了封禁列表。"
|
|
elif find_message.caption and "@" in find_message.caption:
|
|
text = f"{user.full_name} 由于加入群组后，在验证缝隙间发送了带有 @(Mention) 的消息，已被踢出群组，并加入了封禁列表。"
|
|
elif find_message.forward_from_chat:
|
|
text = f"{user.full_name} 由于加入群组后，在验证缝隙间发送了带有 Forward 的消息，已被踢出群组，并加入了封禁列表。"
|
|
if text is not None:
|
|
await context.bot.ban_chat_member(chat.id, user.id)
|
|
button = [[InlineKeyboardButton("解除封禁", callback_data=f"auth_admin|pass|{user.id}")]]
|
|
await question_message.edit_text(text, reply_markup=InlineKeyboardMarkup(button))
|
|
if schedule := context.job_queue.scheduler.get_job(f"{chat.id}|{user.id}|auth_kick"):
|
|
schedule.remove()
|
|
logger.info(
|
|
"用户 %s[%s] 在群 %s[%s] 验证缝隙间发送消息 现已删除", user.full_name, user.id, chat.title, chat.id
|
|
)
|
|
except BadRequest as exc:
|
|
logger.error("后验证处理中发生错误 %s", exc.message)
|
|
logger.exception(exc)
|
|
except MTPFloodWait:
|
|
logger.warning("调用 mtp 触发洪水限制")
|
|
except MTPBadRequest as exc:
|
|
logger.error("调用 mtp 请求错误")
|
|
logger.exception(exc)
|