.build.yml

@@ -1,7 +0,0 @@
-image: alpine/3.20
-packages:
- - python3
-tasks:
- - test: |-
-    cd auto-ban-bot
-    python3 tests.py

.gitignore

@@ -1,3 +1 @@
 config.json
-venv/
-__pycache__/

autobanbot.py

@@ -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"] = []
-        retval["regexes"] = list(map(lambda r: re.compile(r, re.I), config["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):
+async def new_msg(update, context, regexes):
-    if update.message is None and update.edited_message is None:
+    is_spam = False
-        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
     for regex in regexes:
-        if message.text is not None and  regex.search(message.text) is not None:
+        if update.message.text is not None and regex.search(update.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}")
+            is_spam = True
-            await message.chat.ban_member(message.from_user.id)
+            break
-            await message.delete()
+    if is_spam:
-            return
+        logging.info(f"Banning: {update.message}")
-
+        update.message.chat.ban_member(update.message.from_user.id)
-async def handle_error(update, context):
+        update.message.delete()
-    logging.error("Exception while handling an update:", exc_info=context.error)
 
 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)

config.json.example

@@ -1,8 +1,7 @@
 {
     "telegramApiKey": "PUT YOUR KEY HERE",
-    "allowedChats": [123, 456],
     "regexes": [
-        "test123",
+        "ready-made telegram accounts",
-        "(?:@|(?:(?:(?:https?://)?t(?:elegram)?)\\.me\\/))(\\w{4,})bot"
+        "253239090.*473157472"
     ]
 }

fakes/telegram/__init__.py

@@ -1,4 +0,0 @@
-#!/ust/bin/env python3
-
-class Update:
-    pass

fakes/telegram/ext.py

@@ -1,12 +0,0 @@
-#!/usr/bin/env python3
-
-class Application:
-    pass
-
-
-class MessageHandler:
-    pass
-
-
-class filters:
-    pass

tests.py

@@ -1,134 +1,93 @@
 #!/usr/bin/env python
 
 import unittest
-from unittest.mock import AsyncMock
+from unittest.mock import Mock
-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)
 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
+class Member:
-DISALLOWED_CHAT_ID = 789
+    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"
+    def __init__(self, user_id, first_name, last_name, username):
-    delete = AsyncMock()
+        self.new_chat_members = [Member(user_id, first_name, last_name, username)]
-    chat = Chat()
+        self.message_id = MESSAGE_ID
-    from_user = User()
 
+class Chat:
+    def __init__(self):
+        self.id = CHAT_ID
 
-@dataclass
 class Update:
-    message = Message()
+    def __init__(self, user_id, first_name, last_name, username):
-    edited_message = None
+        self.message = Message(user_id, first_name, last_name, username)
+        self.effective_chat = Chat()
 
-
+class TestAutoBanBot(unittest.TestCase):
-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):
     def setUp(self):
-        config = autobanbot.load_config("config.json.example")
+        self.regexes = autobanbot.load_config()["regexes"]
-        self.regexes = config['regexes']
-        self.allowed_chats = config['allowedChats']
 
-    async def test_not_bannable(self):
+    def banned_join(self, user_id, first_name, last_name, username):
-        messages = ["not bannable message", "siematest_123elo", "i am not a bot", "t.me, i'm not a bot"]
+        bot = Mock(spec=BOT_SPEC)
-        for msg in messages:
+        self.assertTrue(autobanbot.hello(
-            with self.subTest("new message", message=msg):
+            bot, Update(user_id, first_name, last_name, username), self.regexes
-                update = make_update(msg, ALLOWED_CHAT_ID)
+        ))
-                self.assertIsNone(update.edited_message)
+        bot.kick_chat_member.assert_called_once_with(CHAT_ID, USER_ID)
-                update.message.delete.reset_mock()
+        bot.delete_message.assert_called_once_with(CHAT_ID, MESSAGE_ID)
-                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()
 
-    async def test_bannable(self):
+    def legit_join(self, user_id, first_name, last_name, username):
-        messages = [
+        bot = Mock(spec=BOT_SPEC)
-            "hitest123hello",
+        self.assertFalse(autobanbot.hello(
-            "HiTeSt123HeLlO",
+            bot, Update(user_id, first_name, last_name, username), self.regexes
-            "t.me/whatever_bot?not bannable message",
+        ))
-            "https://www.t.me/ohhaaaaaaaibot/siematest123elo",
+        bot.kick_chat_member.assert_not_called()
-            "@Hello_BoT"
+        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:
+        for update in updates:
-            with self.subTest("Allowed chat causes a ban", message=msg):
+            with self.subTest(first_name=update):
-                update = make_update(msg, ALLOWED_CHAT_ID)
+                self.banned_join(USER_ID, update, "", "")
-                self.assertIsNone(update.edited_message)
+            with self.subTest(last_name=update):
-                update.message.delete.reset_mock()
+                self.banned_join(USER_ID, "", update, "")
-                update.message.chat.ban_member.reset_mock()
+            with self.subTest(last_name=update, first_name=None):
-                await autobanbot.new_msg(update, None, self.regexes, self.allowed_chats)
+                self.banned_join(USER_ID, None, update, "")
-                update.message.delete.assert_called_once()
+            with self.subTest(last_name=None, first_name=update):
-                update.message.chat.ban_member.assert_called_once_with(update.message.from_user.id)
+                self.banned_join(USER_ID, update, None, "")
-            with self.subTest("Unknown chat is ignored", message=msg):
+            with self.subTest(last_name=update, first_name=update):
-                update = make_update(msg, DISALLOWED_CHAT_ID)
+                self.banned_join(USER_ID, update, update, "")
-                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()
 
+    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()