Auto mute on harmful link

jarvis/client/events/message.py

@@ -4,16 +4,18 @@ from datetime import datetime, timedelta, timezone
 
 from aiohttp import ClientSession
 from jarvis_core.db import q
-from jarvis_core.db.models import Autopurge, Autoreact, Roleping, Setting, Warning
+from jarvis_core.db.models import Autopurge, Autoreact, Mute, Roleping, Setting, Warning
 from jarvis_core.filters import invites, url
 from naff import listen
 from naff.api.events.discord import MessageCreate, MessageDelete, MessageUpdate
 from naff.client.utils.misc_utils import find_all
-from naff.models.discord.channel import DMChannel
+from naff.models.discord.channel import DMChannel, GuildText
 from naff.models.discord.embed import EmbedField
 from naff.models.discord.enums import Permissions
 from naff.models.discord.message import Message
+from naff.models.discord.user import Member
 
+from jarvis.branding import get_command_color
 from jarvis.tracking import malicious_tracker, warnings_tracker
 from jarvis.utils import build_embed
 from jarvis.utils.embeds import warning_embed
@@ -274,6 +276,37 @@ class MessageEventMixin:
                 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."""
@@ -284,8 +317,10 @@ class MessageEventMixin:
             await self.roleping(message)
             await self.autopurge(message)
             await self.checks(message)
-            if not await self.phishing(message):
-                await self.malicious_url(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:
@@ -337,8 +372,10 @@ class MessageEventMixin:
             await self.checks(after)
             await self.roleping(after)
             await self.checks(after)
-            if not await self.phishing(after):
-                await self.malicious_url(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: