Compare commits
No commits in common. "9a52cdf3cfc25340e4a492926cd10cde13b8009a" and "51f34324fbdc9426c75ecfe4572d970ca5fc1d43" have entirely different histories.
9a52cdf3cf
...
51f34324fb
|
|
@ -1,7 +0,0 @@
|
|||
image: alpine/3.20
|
||||
packages:
|
||||
- python3
|
||||
tasks:
|
||||
- test: |-
|
||||
cd auto-ban-bot
|
||||
python3 tests.py
|
||||
|
|
@ -1,3 +1 @@
|
|||
config.json
|
||||
venv/
|
||||
__pycache__/
|
||||
|
|
|
|||
|
|
@ -6,46 +6,35 @@ import re
|
|||
import logging
|
||||
import json
|
||||
|
||||
def load_config(filename: str = "config.json"):
|
||||
def load_config():
|
||||
retval = {}
|
||||
with open(filename) as config_file:
|
||||
with open("config.json") as config_file:
|
||||
config = json.load(config_file)
|
||||
retval["telegramApiKey"] = config["telegramApiKey"]
|
||||
retval["allowedChats"] = config["allowedChats"]
|
||||
retval["regexes"] = list(map(lambda r: re.compile(r, re.I), config["regexes"]))
|
||||
retval["regexes"] = []
|
||||
for regex in config["regexes"]:
|
||||
retval["regexes"].append(re.compile(regex, re.I))
|
||||
return retval
|
||||
|
||||
async def new_msg(update, context, regexes, allowed_chats):
|
||||
if update.message is None and update.edited_message is None:
|
||||
logging.info(f"Got following unknown update: {update}")
|
||||
return
|
||||
message = update.message
|
||||
if update.edited_message is not None:
|
||||
message = update.edited_message
|
||||
if message is None:
|
||||
return
|
||||
if message.chat.id not in allowed_chats:
|
||||
return
|
||||
async def new_msg(update, context, regexes):
|
||||
is_spam = False
|
||||
for regex in regexes:
|
||||
if message.text is not None and regex.search(message.text) is not None:
|
||||
logging.info(f"Banning {message.from_user.name} from {message.chat.effective_name} for posting a message matching {regex.pattern}")
|
||||
await message.chat.ban_member(message.from_user.id)
|
||||
await message.delete()
|
||||
return
|
||||
|
||||
async def handle_error(update, context):
|
||||
logging.error("Exception while handling an update:", exc_info=context.error)
|
||||
if update.message.text is not None and regex.search(update.message.text) is not None:
|
||||
is_spam = True
|
||||
break
|
||||
if is_spam:
|
||||
logging.info(f"Banning: {update.message}")
|
||||
update.message.chat.ban_member(update.message.from_user.id)
|
||||
update.message.delete()
|
||||
|
||||
if __name__ == '__main__':
|
||||
logging.basicConfig(format='%(asctime)s|%(levelname)s|%(message)s', level=logging.INFO)
|
||||
logging.getLogger("httpx").setLevel(logging.WARNING)
|
||||
logging.info("Starting")
|
||||
config = load_config()
|
||||
application = Application.builder().token(config["telegramApiKey"]).build()
|
||||
|
||||
application.add_handler(MessageHandler(filters.ALL, (
|
||||
lambda update, context: new_msg(update, context, config['regexes'], config['allowedChats'])
|
||||
lambda update, context: new_msg(update, context, config['regexes'])
|
||||
)))
|
||||
application.add_error_handler(handle_error, block=False)
|
||||
|
||||
application.run_polling(allowed_updates=Update.ALL_TYPES)
|
||||
|
|
|
|||
|
|
@ -1,8 +1,7 @@
|
|||
{
|
||||
"telegramApiKey": "PUT YOUR KEY HERE",
|
||||
"allowedChats": [123, 456],
|
||||
"regexes": [
|
||||
"test123",
|
||||
"(?:@|(?:(?:(?:https?://)?t(?:elegram)?)\\.me\\/))(\\w{4,})bot"
|
||||
"ready-made telegram accounts",
|
||||
"253239090.*473157472"
|
||||
]
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,4 +0,0 @@
|
|||
#!/ust/bin/env python3
|
||||
|
||||
class Update:
|
||||
pass
|
||||
|
|
@ -1,12 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
|
||||
class Application:
|
||||
pass
|
||||
|
||||
|
||||
class MessageHandler:
|
||||
pass
|
||||
|
||||
|
||||
class filters:
|
||||
pass
|
||||
187
tests.py
187
tests.py
|
|
@ -1,134 +1,93 @@
|
|||
#!/usr/bin/env python
|
||||
|
||||
import unittest
|
||||
from unittest.mock import AsyncMock
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
fakes_dir = os.path.join(os.path.dirname(__file__), 'fakes')
|
||||
assert(os.path.exists(fakes_dir))
|
||||
sys.path.insert(0, fakes_dir)
|
||||
from unittest.mock import Mock
|
||||
import autobanbot
|
||||
|
||||
CHAT_ID = "Chat ID"
|
||||
USER_ID = "User ID"
|
||||
MESSAGE_ID = "Message ID"
|
||||
BOT_SPEC = ["kick_chat_member", "delete_message"]
|
||||
|
||||
ALLOWED_CHAT_ID = 123
|
||||
DISALLOWED_CHAT_ID = 789
|
||||
class Member:
|
||||
def __init__(self, user_id, first_name, last_name, username):
|
||||
self.id = user_id
|
||||
self.first_name = first_name
|
||||
self.last_name = last_name
|
||||
self.username = username
|
||||
|
||||
|
||||
@dataclass
|
||||
class Chat:
|
||||
id: int = 0
|
||||
ban_member = AsyncMock()
|
||||
effective_name = "effective_chat_name"
|
||||
|
||||
|
||||
class User:
|
||||
id_count = 0
|
||||
|
||||
def __init__(self):
|
||||
self.id = User.id_count
|
||||
self.name = "User name"
|
||||
User.id_count += 1
|
||||
|
||||
|
||||
@dataclass
|
||||
class Message:
|
||||
text = "Message text"
|
||||
delete = AsyncMock()
|
||||
chat = Chat()
|
||||
from_user = User()
|
||||
def __init__(self, user_id, first_name, last_name, username):
|
||||
self.new_chat_members = [Member(user_id, first_name, last_name, username)]
|
||||
self.message_id = MESSAGE_ID
|
||||
|
||||
class Chat:
|
||||
def __init__(self):
|
||||
self.id = CHAT_ID
|
||||
|
||||
@dataclass
|
||||
class Update:
|
||||
message = Message()
|
||||
edited_message = None
|
||||
def __init__(self, user_id, first_name, last_name, username):
|
||||
self.message = Message(user_id, first_name, last_name, username)
|
||||
self.effective_chat = Chat()
|
||||
|
||||
|
||||
def make_update(msg: str, chat_id: int) -> Update:
|
||||
update = Update()
|
||||
update.message.text = msg
|
||||
update.message.chat.id = chat_id
|
||||
return update
|
||||
|
||||
def make_edited_message(msg: str, chat_id: int) -> Update:
|
||||
update = Update()
|
||||
update.message = None
|
||||
update.edited_message = Message()
|
||||
update.edited_message.text = msg
|
||||
update.edited_message.chat.id = chat_id
|
||||
return update
|
||||
|
||||
|
||||
class TestAutoBanBot(unittest.IsolatedAsyncioTestCase):
|
||||
class TestAutoBanBot(unittest.TestCase):
|
||||
def setUp(self):
|
||||
config = autobanbot.load_config("config.json.example")
|
||||
self.regexes = config['regexes']
|
||||
self.allowed_chats = config['allowedChats']
|
||||
self.regexes = autobanbot.load_config()["regexes"]
|
||||
|
||||
async def test_not_bannable(self):
|
||||
messages = ["not bannable message", "siematest_123elo", "i am not a bot", "t.me, i'm not a bot"]
|
||||
for msg in messages:
|
||||
with self.subTest("new message", message=msg):
|
||||
update = make_update(msg, ALLOWED_CHAT_ID)
|
||||
self.assertIsNone(update.edited_message)
|
||||
update.message.delete.reset_mock()
|
||||
update.message.chat.ban_member.reset_mock()
|
||||
await autobanbot.new_msg(update, None, self.regexes, self.allowed_chats)
|
||||
update.message.delete.assert_not_called()
|
||||
update.message.chat.ban_member.assert_not_called()
|
||||
with self.subTest("message edit", message=msg):
|
||||
update = make_edited_message(msg, ALLOWED_CHAT_ID)
|
||||
self.assertIsNone(update.message)
|
||||
update.edited_message.delete.reset_mock()
|
||||
update.edited_message.chat.ban_member.reset_mock()
|
||||
await autobanbot.new_msg(update, None, self.regexes, self.allowed_chats)
|
||||
update.edited_message.delete.assert_not_called()
|
||||
update.edited_message.chat.ban_member.assert_not_called()
|
||||
def banned_join(self, user_id, first_name, last_name, username):
|
||||
bot = Mock(spec=BOT_SPEC)
|
||||
self.assertTrue(autobanbot.hello(
|
||||
bot, Update(user_id, first_name, last_name, username), self.regexes
|
||||
))
|
||||
bot.kick_chat_member.assert_called_once_with(CHAT_ID, USER_ID)
|
||||
bot.delete_message.assert_called_once_with(CHAT_ID, MESSAGE_ID)
|
||||
|
||||
async def test_bannable(self):
|
||||
messages = [
|
||||
"hitest123hello",
|
||||
"HiTeSt123HeLlO",
|
||||
"t.me/whatever_bot?not bannable message",
|
||||
"https://www.t.me/ohhaaaaaaaibot/siematest123elo",
|
||||
"@Hello_BoT"
|
||||
def legit_join(self, user_id, first_name, last_name, username):
|
||||
bot = Mock(spec=BOT_SPEC)
|
||||
self.assertFalse(autobanbot.hello(
|
||||
bot, Update(user_id, first_name, last_name, username), self.regexes
|
||||
))
|
||||
bot.kick_chat_member.assert_not_called()
|
||||
bot.delete_message.assert_not_called()
|
||||
|
||||
def test_spam_joins(self):
|
||||
updates = [
|
||||
"READY-MADE TELEGRAM ACCOUNTS",
|
||||
"ready-made TeLeGrAm AcCoUnTs",
|
||||
"there are ready-made TeLeGrAm AcCoUnTs for sale!",
|
||||
"253239090 hi hello 473157472"
|
||||
]
|
||||
for msg in messages:
|
||||
with self.subTest("Allowed chat causes a ban", message=msg):
|
||||
update = make_update(msg, ALLOWED_CHAT_ID)
|
||||
self.assertIsNone(update.edited_message)
|
||||
update.message.delete.reset_mock()
|
||||
update.message.chat.ban_member.reset_mock()
|
||||
await autobanbot.new_msg(update, None, self.regexes, self.allowed_chats)
|
||||
update.message.delete.assert_called_once()
|
||||
update.message.chat.ban_member.assert_called_once_with(update.message.from_user.id)
|
||||
with self.subTest("Unknown chat is ignored", message=msg):
|
||||
update = make_update(msg, DISALLOWED_CHAT_ID)
|
||||
self.assertIsNone(update.edited_message)
|
||||
update.message.delete.reset_mock()
|
||||
update.message.chat.ban_member.reset_mock()
|
||||
await autobanbot.new_msg(update, None, self.regexes, self.allowed_chats)
|
||||
update.message.delete.assert_not_called()
|
||||
update.message.chat.ban_member.assert_not_called()
|
||||
with self.subTest("Edit in allowed chat causes a ban", message=msg):
|
||||
update = make_edited_message(msg, ALLOWED_CHAT_ID)
|
||||
self.assertIsNone(update.message)
|
||||
update.edited_message.delete.reset_mock()
|
||||
update.edited_message.chat.ban_member.reset_mock()
|
||||
await autobanbot.new_msg(update, None, self.regexes, self.allowed_chats)
|
||||
update.edited_message.delete.assert_called_once()
|
||||
update.edited_message.chat.ban_member.assert_called_once_with(update.edited_message.from_user.id)
|
||||
with self.subTest("Edit in unknown chat is ignored", message=msg):
|
||||
update = make_edited_message(msg, DISALLOWED_CHAT_ID)
|
||||
self.assertIsNone(update.message)
|
||||
update.edited_message.delete.reset_mock()
|
||||
update.edited_message.chat.ban_member.reset_mock()
|
||||
await autobanbot.new_msg(update, None, self.regexes, self.allowed_chats)
|
||||
update.edited_message.delete.assert_not_called()
|
||||
update.edited_message.chat.ban_member.assert_not_called()
|
||||
for update in updates:
|
||||
with self.subTest(first_name=update):
|
||||
self.banned_join(USER_ID, update, "", "")
|
||||
with self.subTest(last_name=update):
|
||||
self.banned_join(USER_ID, "", update, "")
|
||||
with self.subTest(last_name=update, first_name=None):
|
||||
self.banned_join(USER_ID, None, update, "")
|
||||
with self.subTest(last_name=None, first_name=update):
|
||||
self.banned_join(USER_ID, update, None, "")
|
||||
with self.subTest(last_name=update, first_name=update):
|
||||
self.banned_join(USER_ID, update, update, "")
|
||||
|
||||
def test_non_spam_joins(self):
|
||||
updates = [
|
||||
"telegram account",
|
||||
"abc",
|
||||
"ready-made telegram naccounts",
|
||||
"completely legit name",
|
||||
""
|
||||
]
|
||||
for update in updates:
|
||||
with self.subTest(first_name=update):
|
||||
self.legit_join(USER_ID, update, "", "")
|
||||
with self.subTest(last_name=update):
|
||||
self.legit_join(USER_ID, "", update, "")
|
||||
with self.subTest(first_name=update, last_name=None):
|
||||
self.legit_join(USER_ID, update, None, "")
|
||||
with self.subTest(last_name=update, first_name=None):
|
||||
self.legit_join(USER_ID, None, update, "")
|
||||
with self.subTest(last_name=update, first_name=update):
|
||||
self.legit_join(USER_ID, update, update, "")
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
|
|
|||
Loading…
Reference in New Issue