252 lines
10 KiB
Python
252 lines
10 KiB
Python
"""Callbacks for matrix messages"""
|
||
|
||
import html
|
||
import time
|
||
import traceback
|
||
|
||
from mab import MatrixBot
|
||
from nio import MatrixRoom, RoomMessageText
|
||
|
||
import util
|
||
from database import Database
|
||
from datatypes import *
|
||
|
||
#
|
||
# PRIVATE
|
||
#
|
||
_bot: MatrixBot = None # type: ignore
|
||
_db: Database = None # type: ignore
|
||
|
||
def _generate_help_message() -> str:
|
||
"""Generates help message in HTML markup"""
|
||
result = "<strong><i>Как использовать</i></strong><br><ol>"
|
||
result += "<li>Используя <code>!auth</code>, создайте токен</li>"
|
||
result += "<li>Добавьте бота в комнату для уведомлений</li>"
|
||
result += "<li>Выполните <code>!info</code> в комнате, чтобы узнать код канала уведомений</li>"
|
||
result += "<li>Используя полученные токен и код канала, отправьте уведомление через веб-запрос</li>"
|
||
result += "</ol>"
|
||
|
||
result += "<br><br><strong><i>Команды</i></strong>"
|
||
for aliases in _COMMANDS:
|
||
result += f"<br><code>!{aliases[0]}</code> - <i>{html.escape(_COMMANDS[aliases][1])}</i>"
|
||
return result
|
||
|
||
#
|
||
# GENERIC EVENT HANDLERS
|
||
#
|
||
async def _on_text(room: MatrixRoom, event: RoomMessageText) -> None:
|
||
"""This callback is called when text message is received"""
|
||
if event.sender == _bot.get_client().user_id:
|
||
return
|
||
text = event.body
|
||
parts = [p for p in text.split() if p.strip()]
|
||
if len(parts) < 1:
|
||
return
|
||
cmd = parts[0].lower()
|
||
if not cmd.startswith("!"):
|
||
return
|
||
cb = None
|
||
for aliases in _COMMANDS:
|
||
for alias in aliases:
|
||
if alias == cmd.lstrip("!"):
|
||
cb = _COMMANDS[aliases][0]
|
||
break
|
||
if cb:
|
||
break
|
||
if cb is None:
|
||
await _bot.send_text_to_room(room.room_id, "<strong>Используйте <code>!help</code></strong>")
|
||
return
|
||
try:
|
||
await cb(room, parts[1:])
|
||
except:
|
||
traceback.print_exc()
|
||
|
||
#
|
||
# COMMAND HANDLERS
|
||
#
|
||
async def _on_cmd_help(room: MatrixRoom, args: list[str]) -> None:
|
||
"""!help"""
|
||
await _bot.send_text_to_room(room.room_id, _generate_help_message(), is_html=True)
|
||
|
||
async def _on_cmd_info(room: MatrixRoom, args: list[str]) -> None:
|
||
"""!info"""
|
||
# get room info (or add it)
|
||
room_info = await _db.room_get(room.room_id)
|
||
if room_info is None:
|
||
room_info = await _db.room_create(room.room_id)
|
||
# failure
|
||
if room_info is None:
|
||
await _bot.send_text_to_room(
|
||
room.room_id,
|
||
"<strong>Нет информации о комнате</strong>",
|
||
is_html=True
|
||
)
|
||
return
|
||
# respond
|
||
response = "<strong><i>Сведения о комнате</i></strong><br>"
|
||
response += f"<strong>Канал:</strong> <code>{html.escape(room_info.code)}</code>"
|
||
await _bot.send_text_to_room(room.room_id, response, is_html=True)
|
||
|
||
async def _on_cmd_tokens(room: MatrixRoom, args: list[str]) -> None:
|
||
"""!info"""
|
||
# get all tokens and check if there are none
|
||
tokens = await _db.token_get_all()
|
||
if not tokens:
|
||
await _bot.send_text_to_room(
|
||
room.room_id,
|
||
"<strong>Нет токенов, используйте <code>!auth</code></strong>",
|
||
is_html=True
|
||
)
|
||
return
|
||
# create the list of tokens
|
||
response = "<strong><i>Список токенов</i></strong>"
|
||
for token in tokens:
|
||
response += f"<br><strong><code>{html.escape(token.code)}</code></strong><br><ul>"
|
||
if token.name is not None:
|
||
response += f"<li><strong>Имя службы:</strong> <code>{token.name}</code></li>"
|
||
else:
|
||
response += f"<li><strong>Имя службы:</strong> <i>указывается в запросе</i></li>"
|
||
response += f"<li><strong>Создан:</strong> <i>{util.date_to_text(token.created_at)}</i></li>"
|
||
response += f"<li><strong>Последнее использование:</strong> <i>{util.date_to_text(token.last_access_at)}</i></li>"
|
||
response += "</ul>"
|
||
# respond
|
||
await _bot.send_text_to_room(room.room_id, response, is_html=True)
|
||
|
||
async def _on_cmd_auth(room: MatrixRoom, args: list[str]) -> None:
|
||
"""!auth"""
|
||
# create new token
|
||
token = await _db.token_create()
|
||
# set the name if it is provided
|
||
if args:
|
||
token.name = " ".join(args)
|
||
await _db.token_set_name(token.code, token.name)
|
||
# create the response
|
||
response = "<strong>Создан новый токен</strong>"
|
||
response += f"<br><strong>Код:</strong> <code>{token.code}</code>"
|
||
response += f"<br><strong>Имя сервиса:</strong> "
|
||
if token.name:
|
||
response += f"<code>{html.escape(token.name)}</code>"
|
||
else:
|
||
response += "указывается в запросе"
|
||
# respond
|
||
await _bot.send_text_to_room(room.room_id, response, is_html=True)
|
||
|
||
async def _on_cmd_deauth(room: MatrixRoom, args: list[str]) -> None:
|
||
"""!deauth"""
|
||
# no token provided
|
||
if len(args) != 1:
|
||
await _bot.send_text_to_room(room.room_id, "<strong>Укажите токен, который надо удалить (должен быть ровно один аргумент)</strong>", is_html=True)
|
||
return
|
||
token = args[0]
|
||
# check if token does not exist
|
||
if await _db.token_get(token) is None:
|
||
await _bot.send_text_to_room(room.room_id, "<strong>Токен не найден</strong>", is_html=True)
|
||
return
|
||
# remove the token
|
||
await _db.token_delete(token)
|
||
# respond
|
||
response = f"<strong>Удалён токен <code>{token}</code></strong>"
|
||
await _bot.send_text_to_room(room.room_id, response, is_html=True)
|
||
|
||
async def _on_cmd_name(room: MatrixRoom, args: list[str]) -> None:
|
||
"""!name"""
|
||
# check arguments
|
||
if len(args) < 1:
|
||
error = "<strong>Требуется указать как минимум токен. Если хотите убрать имя, то имя указывать не надо. Если имя нужно назначить или изменить, то после токена укажите новое имя.</strong>"
|
||
await _bot.send_text_to_room(room.room_id, error, is_html=True)
|
||
return
|
||
token = args[0]
|
||
new_name = " ".join(args[1:])
|
||
if not new_name.strip():
|
||
new_name = None
|
||
# check if token exists
|
||
if await _db.token_get(token) is None:
|
||
await _bot.send_text_to_room(room.room_id, "<strong>Токен не существует</strong>", is_html=True)
|
||
return
|
||
# set new name
|
||
await _db.token_set_name(token, new_name)
|
||
# prepare the response
|
||
if new_name:
|
||
response = f"<strong>Новое имя для токена <code>{token}</code>: <code>{new_name}</code></strong>"
|
||
else:
|
||
response = f"<strong>Удалено имя для токена <code>{token}</code></strong>"
|
||
# respond
|
||
await _bot.send_text_to_room(room.room_id, response, is_html=True)
|
||
|
||
async def _on_cmd_ban(room: MatrixRoom, args: list[str]) -> None:
|
||
"""!ban"""
|
||
# check arguments
|
||
if len(args) < 2:
|
||
error = "<strong>Формат: <code>!ban <IP> <SECONDS> [REASON]</code></strong>"
|
||
await _bot.send_text_to_room(room.room_id, error, is_html=True)
|
||
return
|
||
# get args
|
||
try:
|
||
ip = args[0]
|
||
duration = float(args[1])
|
||
reason = " ".join(args[2:]) if args[2:] else "Manual ban"
|
||
except:
|
||
error = "<strong>Возникла ошибка. Наверняка неправильно указаны секунды.</strong>"
|
||
await _bot.send_text_to_room(room.room_id, error, is_html=True)
|
||
return
|
||
# ban
|
||
try:
|
||
await _db.ban_create(ip, time.time() + duration, reason)
|
||
await _bot.send_text_to_room(room.room_id, "<strong>IP адрес заблокирован</strong>", is_html=True)
|
||
except:
|
||
await _bot.send_text_to_room(room.room_id, "<strong>Возникла ошибка</strong>", is_html=True)
|
||
|
||
async def _on_cmd_unban(room: MatrixRoom, args: list[str]) -> None:
|
||
"""!unban"""
|
||
# check arguments
|
||
if len(args) != 1:
|
||
error = "<strong>Формат: <code>!ban <IP></code></strong>"
|
||
await _bot.send_text_to_room(room.room_id, error, is_html=True)
|
||
return
|
||
# unban
|
||
try:
|
||
await _db.ban_delete(args[0])
|
||
await _bot.send_text_to_room(room.room_id, "<strong>IP адрес разблокирован (если он был заблокирован)</strong>", is_html=True)
|
||
except:
|
||
await _bot.send_text_to_room(room.room_id, "<strong>Возникла ошибка</strong>", is_html=True)
|
||
|
||
async def _on_cmd_bans(room: MatrixRoom, args: list[str]) -> None:
|
||
"""!bans"""
|
||
# list
|
||
bans = _db.ban_get_all()
|
||
if not bans:
|
||
await _bot.send_text_to_room(room.room_id, "<strong>Нет заблокированных IP адресов</strong>", is_html=True)
|
||
return
|
||
# create the response
|
||
result = "<strong>Список заблокированных IP</strong><br><ul>"
|
||
for ban in bans:
|
||
result += f"<li><strong>{html.escape(ban["ip"])}</strong><ul>"
|
||
result += f"<li><strong>Действует до:</strong> <code>{util.date_to_text(ban["expires_at"])}</code></li>"
|
||
result += f"<li><strong>Причина:</strong> <code>{html.escape(ban["reason"])}</code></li>"
|
||
result += "</ul></li>"
|
||
result += "</ul>"
|
||
await _bot.send_text_to_room(room.room_id, result, is_html=True)
|
||
|
||
|
||
_COMMANDS = {
|
||
("help", "h", "?"): (_on_cmd_help, "Получить справку"),
|
||
("info", "room", "i", "r"): (_on_cmd_info, "Получить сведения о комнате"),
|
||
("tokens", "t"): (_on_cmd_tokens, "Получить список токенов"),
|
||
("auth", "create", "a", "c"): (_on_cmd_auth, "Создать новый токен"),
|
||
("deauth", "delete", "d"): (_on_cmd_deauth, "Удалить существующий токен"),
|
||
("name", "n"): (_on_cmd_name, "Задать (или удалить) имя для токена"),
|
||
("ban", "b"): (_on_cmd_ban, "Забанить IP адрес"),
|
||
("unban", "u"): (_on_cmd_unban, "Разбанить IP адрес"),
|
||
("bans", "l"): (_on_cmd_bans, "Получить список забаненных IP адресов"),
|
||
}
|
||
|
||
#
|
||
# PUBLIC
|
||
#
|
||
def setup(bot: MatrixBot, db: Database) -> None:
|
||
"""Setup the callbacks"""
|
||
global _bot, _db
|
||
_db = db
|
||
_bot = bot
|
||
_bot.add_event_callback(_on_text, RoomMessageText) # type: ignore
|