Merge pull request #202 from PyDrocsid/feature/channel_hopping

Feature/channel hopping

Author: Defelo

moderation/spam_detection/cog.py

@@ -1,38 +1,53 @@
 import time
+from datetime import timedelta
 
-from discord import Embed, Member, VoiceState
+from discord import Embed, Forbidden, HTTPException, Member, VoiceState
 from discord.ext import commands
 from discord.ext.commands import Context, UserInputError, guild_only
 
 from PyDrocsid.cog import Cog
-from PyDrocsid.command import reply
-from PyDrocsid.config import Contributor
+from PyDrocsid.command import CommandError, docs, reply
 from PyDrocsid.redis import redis
 from PyDrocsid.translations import t
 
 from .colors import Colors
 from .permissions import SpamDetectionPermission
 from .settings import SpamDetectionSettings
+from ...contributor import Contributor
 from ...pubsub import send_alert, send_to_changelog
 
 
 tg = t.g
 t = t.spam_detection
 
 
+async def _send_changes(ctx: Context, amount: int, change_type: str, description):
+    description = description(amount, change_type) if amount > 0 else t.hop_detection_disabled(change_type)
+    embed = Embed(title=t.channel_hopping, description=description, colour=Colors.SpamDetection)
+
+    await reply(ctx, embed=embed)
+    await send_to_changelog(ctx.guild, description)
+
+
 class SpamDetectionCog(Cog, name="Spam Detection"):
-    CONTRIBUTORS = [Contributor.ce_phox, Contributor.Defelo]
+    CONTRIBUTORS = [Contributor.ce_phox, Contributor.Defelo, Contributor.NekoFanatic]
 
     async def on_voice_state_update(self, member: Member, before: VoiceState, after: VoiceState):
         """
         Checks for channel-hopping
         """
 
+        if await SpamDetectionPermission.bypass.check_permissions(member):
+            return
+
         if before.channel == after.channel:
             return
 
-        max_hops: int = await SpamDetectionSettings.max_hops.get()
-        if max_hops <= 0:
+        alert: int = await SpamDetectionSettings.max_hops_alert.get()
+        warning: int = await SpamDetectionSettings.max_hops_warning.get()
+        mute: int = await SpamDetectionSettings.max_hops_temp_mute.get()
+        duration: int = await SpamDetectionSettings.temp_mute_duration.get()
+        if alert == 0 and warning == 0 and mute == 0:
             return
 
         ts = time.time()
@@ -41,30 +56,38 @@ async def on_voice_state_update(self, member: Member, before: VoiceState, after:
         await redis.expire(key, 60)
         hops: int = await redis.zcount(key, "-inf", "inf")
 
-        if hops <= max_hops:
-            return
-
-        if await redis.exists(key := f"channel_hops_alert_sent:user={member.id}"):
-            return
-
-        await redis.setex(key, 10, 1)
-
-        embed = Embed(title=t.channel_hopping, color=Colors.SpamDetection, description=t.hops_in_last_minute(cnt=hops))
-        embed.add_field(name=tg.member, value=member.mention)
-        embed.add_field(name=t.member_id, value=member.id)
-        embed.set_author(name=str(member), icon_url=member.display_avatar.url)
-        if after.channel:
-            embed.add_field(name=t.current_channel, value=after.channel.name)
-
-        await send_alert(member.guild, embed)
+        if hops >= alert > 0 and not await redis.exists(key := f"channel_hops_alert_sent:user={member.id}"):
+            await redis.setex(key, 10, 1)
+            embed = Embed(
+                title=t.channel_hopping, color=Colors.SpamDetection, description=t.hops_in_last_minute(cnt=hops)
+            )
+            embed.add_field(name=tg.member, value=member.mention)
+            embed.add_field(name=t.member_id, value=str(member.id))
+            embed.set_author(name=str(member), icon_url=member.display_avatar.url)
+            if after.channel:
+                embed.add_field(name=t.current_channel, value=after.channel.name)
+            await send_alert(member.guild, embed)
+
+        if hops >= warning > 0 and not await redis.exists(key := f"channel_hops_warning_sent:user={member.id}"):
+            await redis.setex(key, 10, 1)
+            embed = Embed(title=t.channel_hopping_warning_sent, color=Colors.SpamDetection)
+            try:
+                await member.send(embed=embed)
+            except (HTTPException, Forbidden):
+                pass
+
+        if hops >= mute > 0 and not await redis.exists(key := f"channel_hops_mute:user={member.id}"):
+            try:
+                await member.timeout_for(duration=timedelta(seconds=duration), reason=t.reason)
+            except Forbidden:
+                await send_alert(member.guild, t.cant_mute(member.mention, member.id))
+            await redis.setex(key, 10, 1)
 
     @commands.group(aliases=["spam", "sd"])
     @SpamDetectionPermission.read.check
     @guild_only()
+    @docs(t.commands.spam_detection)
     async def spam_detection(self, ctx: Context):
-        """
-        view and change spam detection settings
-        """
 
         if ctx.subcommand_passed is not None:
             if ctx.invoked_subcommand is None:
@@ -73,26 +96,67 @@ async def spam_detection(self, ctx: Context):
 
         embed = Embed(title=t.spam_detection, color=Colors.SpamDetection)
 
-        if (max_hops := await SpamDetectionSettings.max_hops.get()) <= 0:
-            embed.add_field(name=t.channel_hopping, value=tg.disabled)
+        if (dm_warning := await SpamDetectionSettings.max_hops_warning.get()) <= 0:
+            embed.add_field(name=t.channel_hopping_warning, value=tg.disabled, inline=False)
+        else:
+            embed.add_field(name=t.channel_hopping_warning, value=t.max_x_hops(cnt=dm_warning), inline=False)
+
+        if (alert := await SpamDetectionSettings.max_hops_alert.get()) <= 0:
+            embed.add_field(name=t.channel_hopping_alert, value=tg.disabled, inline=False)
         else:
-            embed.add_field(name=t.channel_hopping, value=t.max_x_hops(cnt=max_hops))
+            embed.add_field(name=t.channel_hopping_alert, value=t.max_x_hops(cnt=alert), inline=False)
+
+        if (mute_hops := await SpamDetectionSettings.max_hops_temp_mute.get()) <= 0:
+            embed.add_field(name=t.channel_hopping_mute, value=tg.disabled, inline=False)
+        else:
+            embed.add_field(name=t.channel_hopping_mute, value=t.max_x_hops(cnt=mute_hops), inline=False)
+
+        mute_duration = await SpamDetectionSettings.temp_mute_duration.get()
+        embed.add_field(name=t.mute_duration, value=t.seconds_muted(cnt=mute_duration), inline=False)
 
         await reply(ctx, embed=embed)
 
-    @spam_detection.command(name="hops", aliases=["h"])
+    @spam_detection.group(name="channel_hopping", aliases=["ch", "h"])
     @SpamDetectionPermission.write.check
-    async def spam_detection_hops(self, ctx: Context, amount: int):
-        """
-        Changes the number of maximum channel hops per minute allowed before an alert is issued
-        set this to 0 to disable channel hopping alerts
-        """
+    @docs(t.commands.channel_hopping)
+    async def channel_hopping(self, ctx: Context):
 
-        await SpamDetectionSettings.max_hops.set(amount)
-        embed = Embed(
-            title=t.channel_hopping,
-            description=t.hop_amount_set(amount) if amount > 0 else t.hop_detection_disabled,
-            colour=Colors.SpamDetection,
-        )
-        await reply(ctx, embed=embed)
-        await send_to_changelog(ctx.guild, t.hop_amount_set(amount) if amount > 0 else t.hop_detection_disabled)
+        if not ctx.subcommand_passed or not ctx.invoked_subcommand:
+            raise UserInputError
+
+    @channel_hopping.command(name="alert", aliases=["a"])
+    @docs(t.commands.alert)
+    async def alert(self, ctx: Context, amount: int):
+
+        await SpamDetectionSettings.max_hops_alert.set(max(amount, 0))
+        await _send_changes(ctx, amount, t.change_types.alerts, t.hop_amount_set)
+
+    @channel_hopping.command(name="warning", aliases=["warn", "w", "dm"])
+    @docs(t.commands.warning)
+    async def warning(self, ctx: Context, amount: int):
+
+        await SpamDetectionSettings.max_hops_warning.set(max(amount, 0))
+        await _send_changes(ctx, amount, t.change_types.warnings, t.hop_amount_set)
+
+    @channel_hopping.group(name="mute", aliases=["m"])
+    @docs(t.commands.temp_mute)
+    async def mute(self, ctx: Context):
+
+        if not ctx.subcommand_passed or not ctx.invoked_subcommand:
+            raise UserInputError
+
+    @mute.command(name="hops", aliases=["h"])
+    @docs(t.commands.temp_mute_hops)
+    async def hops(self, ctx: Context, amount: int):
+
+        await SpamDetectionSettings.max_hops_temp_mute.set(max(amount, 0))
+        await _send_changes(ctx, amount, t.change_types.mutes, t.hop_amount_set)
+
+    @mute.command(name="duration", aliases=["d"])
+    @docs(t.commands.temp_mute_duration)
+    async def duration(self, ctx: Context, seconds: int):
+        if seconds not in range(1, 28 * 24 * 60 * 60):
+            raise CommandError(tg.invalid_duration)
+
+        await SpamDetectionSettings.temp_mute_duration.set(seconds)
+        await _send_changes(ctx, seconds, t.change_types.mutes, t.mute_time_set)

moderation/spam_detection/permissions.py

@@ -11,3 +11,4 @@ def description(self) -> str:
 
     read = auto()
     write = auto()
+    bypass = auto()

moderation/spam_detection/settings.py

@@ -2,4 +2,7 @@
 
 
 class SpamDetectionSettings(Settings):
-    max_hops = 0
+    max_hops_alert = 0
+    max_hops_warning = 0
+    max_hops_temp_mute = 0
+    temp_mute_duration = 10

moderation/spam_detection/translations/en.yml

@@ -1,18 +1,50 @@
+commands:
+  spam_detection: view and change spam detection settings
+  channel_hopping: edit spam detection settings
+  alert: set the number of allowed hops before a alert-channel notification is used (0 for disabling)
+  warning: set the number of allowed hops before a dm notification is used (0 for disabling)
+  temp_mute: edit the channel hopping mute settings
+  temp_mute_hops: set the number of allowed hops before a alert-channel notification is used (0 for disabling)
+  temp_mute_duration: set the number of seconds a user is muted (0 for disabling)
+
 permissions:
   read: read spam detection configuration
   write: write spam detection configuration
+  bypass: bypass the channel hopping check
 
 spam_detection: Spam Detection
+
 max_x_hops:
-  one: "Max.: {cnt} channel hop per minute"
-  many: "Max.: {cnt} channel hops per minute"
+  one: "Max.: `{cnt}` channel hop per minute"
+  many: "Max.: `{cnt}` channel hops per minute"
+seconds_muted:
+  one: "{cnt} second"
+  many: "{cnt} seconds"
+
 member_id: Member ID
-channel_hopping: Channel Hopping Detection
+
+channel_hopping: Channel hopping
+channel_hopping_warning: Warning via DM
+channel_hopping_alert: Alert message
+channel_hopping_mute: Temporare mute
+mute_duration: Mute duration
+reason: was timouted because of channelhopping
+
+cant_mute: "Cannot mute {} ({}) for channel hopping because of missing permissions"
+
 current_channel: Current Channel
-hop_amount_set: "The **maximum amount** of **channel hops per minute** has been **set to {}**."
-hop_detection_disabled: "**Channel Hopping Detection** has been **disabled**."
+
 new_amount: New amount
-max_hops: Maximum channel hops per minute
+hop_amount_set: "The **maximum amount** of **channel hops per minute** has been **set to {}** for **{}**."
+mute_time_set: "The **muted time** for channel hooping has been **set to {} seconds**."
+hop_detection_disabled: "**Channel Hopping Detection** has been **disabled** for **{}**."
+change_types:
+  warnings: warnings
+  alerts: alerts
+  mutes: mutes
+
 hops_in_last_minute:
   one: "{cnt} hop in the last minute"
   many: "{cnt} hops in the last minute"
+
+channel_hopping_warning_sent: "Please stop channel hopping!"