582 lines
25 KiB
Python
582 lines
25 KiB
Python
"""JARVIS message event mixin"""
|
|
import re
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from aiohttp import ClientSession
|
|
from beanie.operators import Inc, Set
|
|
from interactions import listen
|
|
from interactions.api.events.discord import MessageCreate, MessageDelete, MessageUpdate
|
|
from interactions.client.utils.misc_utils import find_all
|
|
from interactions.models.discord.channel import DMChannel, GuildText
|
|
from interactions.models.discord.components import ActionRow, Button
|
|
from interactions.models.discord.embed import EmbedField
|
|
from interactions.models.discord.enums import ButtonStyle, Permissions
|
|
from interactions.models.discord.message import Message
|
|
from interactions.models.discord.user import Member
|
|
from jarvis_core.db import q
|
|
from jarvis_core.db.models import (
|
|
Autopurge,
|
|
Autoreact,
|
|
Filter,
|
|
Mute,
|
|
Phishlist,
|
|
Roleping,
|
|
Setting,
|
|
Warning,
|
|
)
|
|
from jarvis_core.filters import invites, url
|
|
from statipy.db import Stat
|
|
|
|
from jarvis.branding import get_command_color
|
|
from jarvis.embeds.admin import warning_embed
|
|
from jarvis.utils import build_embed
|
|
from jarvis.tracking import WarningMetadata
|
|
|
|
|
|
class MessageEventMixin:
|
|
# Message
|
|
async def autopurge(self, message: Message) -> None:
|
|
"""Handle autopurge events."""
|
|
autopurge = await Autopurge.find_one(q(guild=message.guild.id, channel=message.channel.id))
|
|
if autopurge:
|
|
if not message.author.has_permission(Permissions.ADMINISTRATOR):
|
|
self.logger.debug(f"Autopurging message {message.guild.id}/{message.channel.id}/{message.id}")
|
|
await message.delete(delay=autopurge.delay)
|
|
|
|
async def autoreact(self, message: Message) -> None:
|
|
"""Handle autoreact events."""
|
|
autoreact = await Autoreact.find_one(
|
|
q(
|
|
guild=message.guild.id,
|
|
channel=message.channel.id,
|
|
)
|
|
)
|
|
if autoreact:
|
|
self.logger.debug(f"Autoreacting to message {message.guild.id}/{message.channel.id}/{message.id}")
|
|
for reaction in autoreact.reactions:
|
|
await message.add_reaction(reaction)
|
|
if autoreact.thread:
|
|
name = message.content.replace("\n", " ")
|
|
name = re.sub(r"<:\w+:(\d+)>", "", name)
|
|
if len(name) > 100:
|
|
name = name[:97] + "..."
|
|
await message.create_thread(name=message.content, reason="Autoreact")
|
|
|
|
async def checks(self, message: Message) -> None:
|
|
"""Other message checks."""
|
|
# #tech
|
|
# channel = find(lambda x: x.id == 599068193339736096, message._mention_ids)
|
|
# if channel and message.author.id == 293795462752894976:
|
|
# await channel.send(
|
|
# content="https://cdn.discordapp.com/attachments/664621130044407838/805218508866453554/tech.gif" # noqa: E501
|
|
# )
|
|
content = re.sub(r"\s+", "", message.content)
|
|
match = invites.search(content)
|
|
setting = await Setting.find_one(q(guild=message.guild.id, setting="noinvite"))
|
|
if not setting:
|
|
setting = Setting(guild=message.guild.id, setting="noinvite", value=True)
|
|
await setting.commit()
|
|
if match:
|
|
guild_invites = [x.code for x in await message.guild.fetch_invites()]
|
|
if message.guild.vanity_url_code:
|
|
guild_invites.append(message.guild.vanity_url_code)
|
|
allowed = guild_invites + ["dbrand", "VtgZntXcnZ", "gPfYGbvTCE", "interactions", "NTSHu97tHg"]
|
|
is_mod = message.author.has_permission(Permissions.MANAGE_GUILD) or message.author.has_permission(
|
|
Permissions.ADMINISTRATOR
|
|
)
|
|
if (m := match.group(1)) not in allowed and setting.value and not is_mod:
|
|
self.logger.debug(f"Removing non-allowed invite `{m}` from {message.guild.id}")
|
|
try:
|
|
await message.delete()
|
|
except Exception:
|
|
self.logger.debug("Message deleted before action taken")
|
|
|
|
expires_at = datetime.now(tz=timezone.utc) + timedelta(hours=24)
|
|
await Warning(
|
|
active=True,
|
|
admin=self.user.id,
|
|
duration=24,
|
|
expires_at=expires_at,
|
|
guild=message.guild.id,
|
|
reason="Sent an invite link",
|
|
user=message.author.id,
|
|
).commit()
|
|
md = WarningMetadata(
|
|
client_id=self.user.id,
|
|
client_name=self.client_name,
|
|
name="warning",
|
|
type="invite",
|
|
guild_id=message.guild.id,
|
|
guild_name=message.guild.name,
|
|
value=1,
|
|
)
|
|
await Stat(meta=md).insert()
|
|
embed = warning_embed(message.author, "Sent an invite link", self.user)
|
|
try:
|
|
await message.channel.send(embeds=embed)
|
|
except Exception:
|
|
self.logger.warn("Failed to send warning embed")
|
|
|
|
async def filters(self, message: Message) -> None:
|
|
"""Handle filter evennts."""
|
|
filters = await Filter.find(q(guild=message.guild.id)).to_list(None)
|
|
for item in filters:
|
|
for f in item.filters:
|
|
if re.search(f, message.content, re.IGNORECASE):
|
|
expires_at = datetime.now(tz=timezone.utc) + timedelta(hours=24)
|
|
await Warning(
|
|
active=True,
|
|
admin=self.user.id,
|
|
duration=24,
|
|
expires_at=expires_at,
|
|
guild=message.guild.id,
|
|
reason="Sent a message with a filtered word",
|
|
user=message.author.id,
|
|
).commit()
|
|
md = WarningMetadata(
|
|
client_id=self.user.id,
|
|
client_name=self.client_name,
|
|
name="warning",
|
|
type="filter",
|
|
guild_id=message.guild.id,
|
|
guild_name=message.guild.name,
|
|
value=1,
|
|
)
|
|
await Stat(meta=md).insert()
|
|
embed = warning_embed(message.author, "Sent a message with a filtered word", self.user)
|
|
try:
|
|
await message.reply(embeds=embed)
|
|
except Exception:
|
|
self.logger.warn("Failed to send warning embed")
|
|
|
|
try:
|
|
await message.delete()
|
|
except Exception:
|
|
self.logger.debug("Message deleted before action taken")
|
|
return
|
|
|
|
async def massmention(self, message: Message) -> None:
|
|
"""Handle massmention events."""
|
|
massmention = await Setting.find_one(
|
|
q(
|
|
guild=message.guild.id,
|
|
setting="massmention",
|
|
)
|
|
)
|
|
|
|
is_mod = message.author.has_permission(Permissions.MANAGE_GUILD) or message.author.has_permission(
|
|
Permissions.ADMINISTRATOR
|
|
)
|
|
|
|
if (
|
|
massmention
|
|
and massmention.value > 0 # noqa: W503
|
|
and len(message._mention_ids + message._mention_roles) # noqa: W503
|
|
- (1 if message.author.id in message._mention_ids else 0) # noqa: W503
|
|
> massmention.value # noqa: W503
|
|
and not is_mod # noqa: W503
|
|
):
|
|
self.logger.debug(f"Massmention threshold on {message.guild.id}/{message.channel.id}/{message.id}")
|
|
expires_at = datetime.now(tz=timezone.utc) + timedelta(hours=24)
|
|
await Warning(
|
|
active=True,
|
|
admin=self.user.id,
|
|
duration=24,
|
|
expires_at=expires_at,
|
|
guild=message.guild.id,
|
|
reason="Mass Mention",
|
|
user=message.author.id,
|
|
).commit()
|
|
md = WarningMetadata(
|
|
client_id=self.user.id,
|
|
client_name=self.client_name,
|
|
name="warning",
|
|
type="massmention",
|
|
guild_id=message.guild.id,
|
|
guild_name=message.guild.name,
|
|
value=1,
|
|
)
|
|
await Stat(meta=md).insert()
|
|
embed = warning_embed(message.author, "Mass Mention", self.user)
|
|
try:
|
|
await message.channel.send(embeds=embed)
|
|
except Exception:
|
|
self.logger.warn("Failed to send warning embed")
|
|
|
|
async def roleping(self, message: Message) -> None:
|
|
"""Handle roleping events."""
|
|
try:
|
|
if message.author.has_permission(Permissions.MANAGE_GUILD):
|
|
return
|
|
except Exception as e:
|
|
self.logger.error("Failed to get permissions, pretending check failed", exc_info=e)
|
|
if await Roleping.collection.count_documents(q(guild=message.guild.id, active=True)) == 0:
|
|
return
|
|
rolepings = await Roleping.find(q(guild=message.guild.id, active=True)).to_list(None)
|
|
|
|
# Get all role IDs involved with message
|
|
roles = [x.id async for x in message.mention_roles]
|
|
async for mention in message.mention_users:
|
|
roles += [x.id for x in mention.roles]
|
|
|
|
if not roles:
|
|
return
|
|
|
|
# Get all roles that are rolepinged
|
|
roleping_ids = [r.role for r in rolepings]
|
|
|
|
# Get roles in rolepings
|
|
role_in_rolepings = find_all(lambda x: x in roleping_ids, roles)
|
|
|
|
# Check if the user has the role, so they are allowed to ping it
|
|
user_missing_role = any(x.id not in roleping_ids for x in message.author.roles)
|
|
|
|
# Admins can ping whoever
|
|
user_is_admin = message.author.has_permission(Permissions.ADMINISTRATOR)
|
|
|
|
# Check if user in a bypass list
|
|
def check_has_role(roleping: Roleping) -> bool:
|
|
return any(role.id in roleping.bypass["roles"] for role in message.author.roles)
|
|
|
|
user_has_bypass = False
|
|
for roleping in rolepings:
|
|
if message.author.id in roleping.bypass["users"]:
|
|
user_has_bypass = True
|
|
break
|
|
if check_has_role(roleping):
|
|
user_has_bypass = True
|
|
break
|
|
|
|
if role_in_rolepings and user_missing_role and not user_is_admin and not user_has_bypass:
|
|
self.logger.debug(f"Rolepinged role in {message.guild.id}/{message.channel.id}/{message.id}")
|
|
expires_at = datetime.now(tz=timezone.utc) + timedelta(hours=24)
|
|
await Warning(
|
|
active=True,
|
|
admin=self.user.id,
|
|
duration=24,
|
|
expires_at=expires_at,
|
|
guild=message.guild.id,
|
|
reason="Pinged a blocked role/user with a blocked role",
|
|
user=message.author.id,
|
|
).commit()
|
|
md = WarningMetadata(
|
|
client_id=self.user.id,
|
|
client_name=self.client_name,
|
|
name="warning",
|
|
type="roleping",
|
|
guild_id=message.guild.id,
|
|
guild_name=message.guild.name,
|
|
value=1,
|
|
)
|
|
await Stat(meta=md).insert()
|
|
embed = warning_embed(message.author, "Pinged a blocked role/user with a blocked role", self.user)
|
|
try:
|
|
await message.channel.send(embeds=embed)
|
|
except Exception:
|
|
self.logger.warn("Failed to send warning embed")
|
|
|
|
async def phishing(self, message: Message) -> None:
|
|
"""Check if the message contains any known phishing domains."""
|
|
for match in url.finditer(message.content):
|
|
if (m := match.group("domain")) in self.phishing_domains:
|
|
pl = await Phishlist.find_one(q(url=m))
|
|
if pl and pl.confirmed and not pl.valid:
|
|
return False
|
|
self.logger.debug(
|
|
f"Phishing url `{m}` detected in {message.guild.id}/{message.channel.id}/{message.id}"
|
|
)
|
|
expires_at = datetime.now(tz=timezone.utc) + timedelta(hours=24)
|
|
await Warning(
|
|
active=True,
|
|
admin=self.user.id,
|
|
duration=24,
|
|
expires_at=expires_at,
|
|
guild=message.guild.id,
|
|
reason="Phishing URL",
|
|
user=message.author.id,
|
|
).commit()
|
|
md = WarningMetadata(
|
|
client_id=self.user.id,
|
|
client_name=self.client_name,
|
|
name="warning",
|
|
type="phishing",
|
|
guild_id=message.guild.id,
|
|
guild_name=message.guild.name,
|
|
value=1,
|
|
)
|
|
await Stat(meta=md).insert()
|
|
embed = warning_embed(message.author, "Phishing URL", self.user)
|
|
try:
|
|
await message.channel.send(embeds=embed)
|
|
except Exception:
|
|
self.logger.warn("Failed to send warning embed")
|
|
try:
|
|
await message.delete()
|
|
except Exception:
|
|
self.logger.warn("Failed to delete malicious message")
|
|
|
|
if not pl or not pl.confirmed:
|
|
if not pl:
|
|
pl = Phishlist(url=m)
|
|
await pl.commit()
|
|
|
|
embed = build_embed(
|
|
title="Phishing URL detected",
|
|
description="Please confirm that this is valid",
|
|
fields=[EmbedField(name="URL", value=m)],
|
|
)
|
|
|
|
valid_button = Button(style=ButtonStyle.GREEN, emoji="✔️", custom_id=f"pl|valid|{pl.id}")
|
|
invalid_button = Button(style=ButtonStyle.RED, emoji="✖️", custom_id=f"pl|invalid|{pl.id}")
|
|
|
|
channel = await self.fetch_channel(1026918337554423868)
|
|
|
|
components = [ActionRow(invalid_button, valid_button)]
|
|
|
|
await channel.send(embeds=embed, components=components)
|
|
|
|
return True
|
|
return False
|
|
|
|
async def malicious_url(self, message: Message) -> None:
|
|
"""Check if the message contains any known phishing domains."""
|
|
for match in url.finditer(message.content):
|
|
m = match.group("domain")
|
|
pl = await Phishlist.find_one(q(url=m))
|
|
if pl and pl.confirmed and not pl.valid:
|
|
return False
|
|
async with ClientSession() as session:
|
|
resp = await session.post(
|
|
"https://anti-fish.bitflow.dev/check",
|
|
json={"message": match.string},
|
|
headers={
|
|
"Application-Name": "JARVIS",
|
|
"Application-Link": "https://git.zevaryx.com/stark-industries/jarvis",
|
|
},
|
|
)
|
|
if resp.status != 200:
|
|
break
|
|
data = await resp.json()
|
|
if data["match"]:
|
|
self.logger.debug(
|
|
f"Scam url `{match.string}` detected in {message.guild.id}/{message.channel.id}/{message.id}"
|
|
)
|
|
expires_at = datetime.now(tz=timezone.utc) + timedelta(hours=24)
|
|
await Warning(
|
|
active=True,
|
|
admin=self.user.id,
|
|
duration=24,
|
|
expires_at=expires_at,
|
|
guild=message.guild.id,
|
|
reason="Unsafe URL",
|
|
user=message.author.id,
|
|
).commit()
|
|
md = WarningMetadata(
|
|
client_id=self.user.id,
|
|
client_name=self.client_name,
|
|
name="warning",
|
|
type="malicious",
|
|
guild_id=message.guild.id,
|
|
guild_name=message.guild.name,
|
|
value=1,
|
|
)
|
|
await Stat(meta=md).insert()
|
|
reasons = ", ".join(f"{m['source']}: {m['type']}" for m in data["matches"])
|
|
embed = warning_embed(message.author, reasons, self.user)
|
|
try:
|
|
await message.channel.send(embeds=embed)
|
|
except Exception:
|
|
self.logger.warn("Failed to send warning embed")
|
|
try:
|
|
await message.delete()
|
|
except Exception:
|
|
self.logger.warn("Failed to delete malicious message")
|
|
|
|
if not pl or not pl.confirmed:
|
|
if not pl:
|
|
pl = Phishlist(url=m)
|
|
await pl.commit()
|
|
|
|
embed = build_embed(
|
|
title="Malicious URL detected",
|
|
description="Please confirm that this is valid",
|
|
fields=[EmbedField(name="URL", value=m)],
|
|
)
|
|
|
|
valid_button = Button(style=ButtonStyle.GREEN, emoji="✔️", custom_id=f"pl|valid|{pl.id}")
|
|
invalid_button = Button(style=ButtonStyle.RED, emoji="✖️", custom_id=f"pl|invalid|{pl.id}")
|
|
|
|
channel = await self.fetch_channel(1026918337554423868)
|
|
|
|
components = [ActionRow(invalid_button, valid_button)]
|
|
|
|
await channel.send(embeds=embed, components=components)
|
|
|
|
return True
|
|
return False
|
|
|
|
async def timeout_user(self, user: Member, channel: GuildText) -> None:
|
|
"""Timeout a user."""
|
|
expires_at = datetime.now(tz=timezone.utc) + timedelta(minutes=30)
|
|
try:
|
|
await user.timeout(communication_disabled_until=expires_at, reason="Phishing link")
|
|
await Mute(
|
|
user=user.id,
|
|
reason="Auto mute for harmful link",
|
|
admin=self.user.id,
|
|
guild=user.guild.id,
|
|
duration=30,
|
|
active=True,
|
|
).commit()
|
|
ts = int(expires_at.timestamp())
|
|
embed = build_embed(
|
|
title="User Muted",
|
|
description=f"{user.mention} has been muted",
|
|
fields=[
|
|
EmbedField(name="Reason", value="Auto mute for harmful link"),
|
|
EmbedField(name="Until", value=f"<t:{ts}:F> <t:{ts}:R>"),
|
|
],
|
|
color=get_command_color("mute"),
|
|
)
|
|
embed.set_author(name=user.display_name, icon_url=user.display_avatar.url)
|
|
embed.set_thumbnail(url=user.display_avatar.url)
|
|
embed.set_footer(text=f"{user.username}#{user.discriminator} | {user.id}")
|
|
await channel.send(embeds=embed)
|
|
|
|
except Exception:
|
|
self.logger.warn("Failed to timeout user for phishing")
|
|
|
|
@listen()
|
|
async def on_message(self, event: MessageCreate) -> None:
|
|
"""Handle on_message event. Calls other event handlers."""
|
|
message = event.message
|
|
await self.autopurge(message)
|
|
if not isinstance(message.channel, DMChannel) and not message.author.bot:
|
|
await self.autoreact(message)
|
|
await self.massmention(message)
|
|
await self.roleping(message)
|
|
await self.checks(message)
|
|
await self.filters(message)
|
|
if not (phish := await self.phishing(message)):
|
|
malicious = await self.malicious_url(message)
|
|
if phish or malicious:
|
|
await self.timeout_user(message.author, message.channel)
|
|
|
|
@listen()
|
|
async def on_message_edit(self, event: MessageUpdate) -> None:
|
|
"""Process on_message_edit events."""
|
|
before = event.before
|
|
after = event.after
|
|
if not after.author.bot:
|
|
modlog = await Setting.find_one(q(guild=after.guild.id, setting="activitylog"))
|
|
ignore = await Setting.find_one(q(guild=after.guild.id, setting="log_ignore"))
|
|
if modlog and (not ignore or (ignore and after.channel.id not in ignore.value)):
|
|
if not before or before.content == after.content or before.content is None:
|
|
return
|
|
try:
|
|
channel = before.guild.get_channel(modlog.value)
|
|
fields = [
|
|
EmbedField(
|
|
"Original Message",
|
|
before.content if before.content else "N/A",
|
|
False,
|
|
),
|
|
EmbedField(
|
|
"New Message",
|
|
after.content if after.content else "N/A",
|
|
False,
|
|
),
|
|
]
|
|
embed = build_embed(
|
|
title="Message Edited",
|
|
description=f"{after.author.mention} edited a message in {before.channel.mention}",
|
|
fields=fields,
|
|
color="#fc9e3f",
|
|
timestamp=after.edited_timestamp,
|
|
url=after.jump_url,
|
|
)
|
|
embed.set_author(
|
|
name=after.author.username,
|
|
icon_url=after.author.display_avatar.url,
|
|
url=after.jump_url,
|
|
)
|
|
embed.set_footer(text=f"{after.author.username}#{after.author.discriminator} | {after.author.id}")
|
|
await channel.send(embeds=embed)
|
|
except Exception as e:
|
|
self.logger.warning(
|
|
f"Failed to process edit {before.guild.id}/{before.channel.id}/{before.id}: {e}"
|
|
)
|
|
if not isinstance(after.channel, DMChannel) and not after.author.bot:
|
|
await self.roleping(after)
|
|
await self.checks(after)
|
|
await self.roleping(after)
|
|
await self.checks(after)
|
|
await self.filters(after)
|
|
if not (phish := await self.phishing(after)):
|
|
malicious = await self.malicious_url(after)
|
|
if phish or malicious:
|
|
await self.timeout_user(after.author, after.channel)
|
|
|
|
@listen()
|
|
async def on_message_delete(self, event: MessageDelete) -> None:
|
|
"""Process on_message_delete events."""
|
|
message = event.message
|
|
modlog = await Setting.find_one(q(guild=message.guild.id, setting="activitylog"))
|
|
ignore = await Setting.find_one(q(guild=message.guild.id, setting="log_ignore"))
|
|
if modlog and (not ignore or (ignore and message.channel.id not in ignore.value)):
|
|
try:
|
|
content = message.content or "N/A"
|
|
except AttributeError:
|
|
content = "N/A"
|
|
fields = [EmbedField("Original Message", content, False)]
|
|
|
|
try:
|
|
if message.attachments:
|
|
value = "\n".join([f"[{x.filename}]({x.url})" for x in message.attachments])
|
|
fields.append(
|
|
EmbedField(
|
|
name="Attachments",
|
|
value=value,
|
|
inline=False,
|
|
)
|
|
)
|
|
|
|
if message.sticker_items:
|
|
value = "\n".join([f"Sticker: {x.name}" for x in message.sticker_items])
|
|
fields.append(
|
|
EmbedField(
|
|
name="Stickers",
|
|
value=value,
|
|
inline=False,
|
|
)
|
|
)
|
|
|
|
if message.embeds:
|
|
value = str(len(message.embeds)) + " embeds"
|
|
fields.append(
|
|
EmbedField(
|
|
name="Embeds",
|
|
value=value,
|
|
inline=False,
|
|
)
|
|
)
|
|
|
|
channel = message.guild.get_channel(modlog.value)
|
|
embed = build_embed(
|
|
title="Message Deleted",
|
|
description=f"{message.author.mention}'s message was deleted from {message.channel.mention}",
|
|
fields=fields,
|
|
color="#fc9e3f",
|
|
)
|
|
|
|
embed.set_author(
|
|
name=message.author.username,
|
|
icon_url=message.author.display_avatar.url,
|
|
url=message.jump_url,
|
|
)
|
|
embed.set_footer(
|
|
text=(f"{message.author.username}#{message.author.discriminator} | " f"{message.author.id}")
|
|
)
|
|
await channel.send(embeds=embed)
|
|
except Exception as e:
|
|
self.logger.warning(f"Failed to process edit {message.guild.id}/{message.channel.id}/{message.id}: {e}")
|