CVE-2020-15172: Deserialization of Untrusted Data (CWE-502)
value: The Act module for Red Discord Bot before commit 6b9f3b86 is vulnerable to Remote Code Execution. With this exploit, Discord users can use specially crafted messages to perform destructive actions and/or access sensitive information. Unloading the Act module with `unload act` can render this exploit inaccessible.
CWE description: The product deserializes untrusted data without sufficiently verifying that the resulting data will be valid.
CVSS v3 base score: 8.8
Published: 2020-09-15T19:15Z
Repository: https://github.com/zephyrkul/FluffyCogs
Language: Python
Files changed: act/act.py
Commit date: 2020-09-09 08:02:49 -0600
Commit message: [act] don't trigger on disabled cmds

Diff: diff --git a/act/act.py b/act/act.py
index 7d5aadc..49b68dc 100644
--- a/act/act.py
+++ b/act/act.py
@@ -1,18 +1,21 @@
-import itertools
+import functools
 import random
+import re
 from typing import Union
 
 import aiohttp
 import discord
 import inflection
-from redbot.core import Config, checks, commands
+from redbot.core import bot, Config, checks, commands
 from redbot.core.i18n import get_locale
 from redbot.core.utils.chat_formatting import italics
 
 from .helpers import *
 
+fmt_re = re.compile(r"{(?:0|user)(?:\.([^\{]+))?}")
+
 
 class Act(commands.Cog):
     """
     This cog makes all commands, e.g. [p]fluff, into valid commands if
     you command the bot to act on a user, e.g. [p]fluff [botname].
@@ -24,28 +27,40 @@ class Act(commands.Cog):
         return {}  # No data to get
 
     async def red_delete_data_for_user(self, *, requester, user_id):
         pass  # No data to delete
 
-    def __init__(self, bot):
+    def __init__(self, bot: bot.Red):
         super().__init__()
         self.bot = bot
         self.config = Config.get_conf(self, identifier=2_113_674_295, force_registration=True)
         self.config.register_global(custom={}, tenorkey=None)
         self.config.register_guild(custom={})
         self.try_after = None
 
-    async def initialize(self, bot):
+    async def initialize(self, bot: bot.Red):
         # temporary backwards compatibility
         key = await self.config.tenorkey()
         if not key:
             return
         await bot.set_shared_api_tokens("tenor", api_key=key)
         await self.config.tenorkey.clear()
 
+    @staticmethod
+    def repl(target: discord.Member, match: re.Match):
+        if attr := match.group(1):
+            print(attr)
+            if attr.startswith("_") or "." in attr:
+                return str(target)
+            try:
+                return str(getattr(target, attr))
+            except AttributeError:
+                return str(target)
+        return str(target)
+
     @commands.command(hidden=True)
-    async def act(self, ctx, *, target: Union[discord.Member, str] = None):
+    async def act(self, ctx: commands.Context, *, target: Union[discord.Member, str] = None):
         """
         Acts on the specified user.
         """
         if not target or isinstance(target, str):
             return  # no help text
@@ -85,11 +100,12 @@ class Act(commands.Cog):
             if iverb < 0:
                 return
             action.insert(iverb + 1, target.mention)
             message = italics(" ".join(action))
         else:
-            message = message.format(target, user=target)
+            assert isinstance(message, str)
+            message = fmt_re.sub(functools.partial(self.repl, target), message)
 
         # add reaction gif
         if self.try_after and ctx.message.created_at < self.try_after:
             return await ctx.send(message)
         if not await ctx.embed_requested():
@@ -129,34 +145,40 @@ class Act(commands.Cog):
             ),
         )
 
     @commands.group()
     @checks.is_owner()
-    async def actset(self, ctx):
+    async def actset(self, ctx: commands.Context):
         """
         Configure various settings for the act cog.
         """
 
     @actset.group(aliases=["custom"], invoke_without_command=True)
     @checks.admin_or_permissions(manage_guild=True)
     @commands.guild_only()
-    async def customize(self, ctx, command: str.lower, *, response: str = None):
+    async def customize(self, ctx: commands.Context, command: str.lower, *, response: str = None):
         """
         Customize the response to an action.
 
         You can use {0} or {user} to dynamically replace with the specified target of the action.
         Formats like {0.name} or {0.mention} can also be used.
         """
         if not response:
             await self.config.guild(ctx.guild).clear_raw("custom", command)
+            await ctx.tick()
         else:
             await self.config.guild(ctx.guild).set_raw("custom", command, value=response)
-        await ctx.tick()
+            await ctx.send(
+                fmt_re.sub(functools.partial(self.repl, ctx.author), response),
+                allowed_mentions=discord.AllowedMentions(users=False),
+            )
 
     @customize.command(name="global")
     @checks.is_owner()
-    async def customize_global(self, ctx, command: str.lower, *, response: str = None):
+    async def customize_global(
+        self, ctx: commands.Context, command: str.lower, *, response: str = None
+    ):
         """
         Globally customize the response to an action.
 
         You can use {0} or {user} to dynamically replace with the specified target of the action.
         Formats like {0.name} or {0.mention} can also be used.
@@ -168,11 +190,11 @@ class Act(commands.Cog):
         await ctx.tick()
 
     @actset.group(invoke_without_command=True)
     @checks.admin_or_permissions(manage_guild=True)
     @commands.guild_only()
-    async def ignore(self, ctx, command: str.lower):
+    async def ignore(self, ctx: commands.Context, command: str.lower):
         """
         Ignore or unignore the specified action.
 
         The bot will no longer respond to these actions.
         """
@@ -187,11 +209,11 @@ class Act(commands.Cog):
             await self.config.guild(ctx.guild).set_raw("custom", command, value=None)
             await ctx.send("I will now ignore the {command} action".format(command=command))
 
     @ignore.command(name="global")
     @checks.is_owner()
-    async def ignore_global(self, ctx, command: str.lower):
+    async def ignore_global(self, ctx: commands.Context, command: str.lower):
         """
         Globally ignore or unignore the specified action.
 
         The bot will no longer respond to these actions.
         """
@@ -203,11 +225,11 @@ class Act(commands.Cog):
             await self.config.clear_raw("custom", command)
         await ctx.tick()
 
     @actset.command()
     @checks.is_owner()
-    async def tenorkey(self, ctx):
+    async def tenorkey(self, ctx: commands.Context):
         """
         Sets a Tenor GIF API key to enable reaction gifs with act commands.
 
         You can obtain a key from here: https://tenor.com/developer/dashboard
         """
@@ -223,22 +245,16 @@ class Act(commands.Cog):
         ]
         instructions = [f"**{i}.** {v}" for i, v in enumerate(instructions, 1)]
         await ctx.maybe_send_embed("\n".join(instructions))
 
     @commands.Cog.listener()
-    async def on_message(self, message):
-        if message.author.bot:
+    async def on_command_error(
+        self, ctx: commands.Context, error: commands.CommandError, unhandled_by_cog: bool = False
+    ):
+        if ctx.command == self.act:
             return
-
-        ctx = await self.bot.get_context(message)
-        if ctx.prefix is None or not ctx.invoked_with.replace("_", "").isalpha():
+        if isinstance(error, commands.UserFeedbackCheckFailure):
+            # UserFeedbackCheckFailure inherits from CheckFailure
             return
-
-        if ctx.valid and ctx.command.enabled:
-            try:
-                if await ctx.command.can_run(ctx):
-                    return
-            except commands.errors.CheckFailure:
-                return
-
-        ctx.command = self.act
-        await self.bot.invoke(ctx)
+        elif isinstance(error, (commands.CheckFailure, commands.CommandNotFound)):
+            ctx.command = self.act
+            await ctx.bot.invoke(ctx)

Vulnerable code: import itertools
from redbot.core import Config, checks, commands
    def __init__(self, bot):
    async def initialize(self, bot):
    async def act(self, ctx, *, target: Union[discord.Member, str] = None):
            message = message.format(target, user=target)
    async def actset(self, ctx):
    async def customize(self, ctx, command: str.lower, *, response: str = None):
        await ctx.tick()
    async def customize_global(self, ctx, command: str.lower, *, response: str = None):
    async def ignore(self, ctx, command: str.lower):
    async def ignore_global(self, ctx, command: str.lower):
    async def tenorkey(self, ctx):
    async def on_message(self, message):
        if message.author.bot:

        ctx = await self.bot.get_context(message)
        if ctx.prefix is None or not ctx.invoked_with.replace("_", "").isalpha():

        if ctx.valid and ctx.command.enabled:
            try:
                if await ctx.command.can_run(ctx):
                    return
            except commands.errors.CheckFailure:
                return

        ctx.command = self.act
        await self.bot.invoke(ctx)
Fixed code: import functools
import re
from redbot.core import bot, Config, checks, commands
fmt_re = re.compile(r"{(?:0|user)(?:\.([^\{]+))?}")

    def __init__(self, bot: bot.Red):
    async def initialize(self, bot: bot.Red):
    @staticmethod
    def repl(target: discord.Member, match: re.Match):
        if attr := match.group(1):
            print(attr)
            if attr.startswith("_") or "." in attr:
                return str(target)
            try:
                return str(getattr(target, attr))
            except AttributeError:
                return str(target)
        return str(target)

    async def act(self, ctx: commands.Context, *, target: Union[discord.Member, str] = None):
            assert isinstance(message, str)
            message = fmt_re.sub(functools.partial(self.repl, target), message)
    async def actset(self, ctx: commands.Context):
    async def customize(self, ctx: commands.Context, command: str.lower, *, response: str = None):
            await ctx.tick()
            await ctx.send(
                fmt_re.sub(functools.partial(self.repl, ctx.author), response),
                allowed_mentions=discord.AllowedMentions(users=False),
            )
    async def customize_global(
        self, ctx: commands.Context, command: str.lower, *, response: str = None
    ):
    async def ignore(self, ctx: commands.Context, command: str.lower):
    async def ignore_global(self, ctx: commands.Context, command: str.lower):
    async def tenorkey(self, ctx: commands.Context):
    async def on_command_error(
        self, ctx: commands.Context, error: commands.CommandError, unhandled_by_cog: bool = False
    ):
        if ctx.command == self.act:
        if isinstance(error, commands.UserFeedbackCheckFailure):
            # UserFeedbackCheckFailure inherits from CheckFailure
        elif isinstance(error, (commands.CheckFailure, commands.CommandNotFound)):
            ctx.command = self.act
            await ctx.bot.invoke(ctx)