Harden fenced markdown rendering against code-fence breakout and mention abuse

bot.py

@@ -27,14 +27,16 @@ async def do_cog_op(ctx: GitBotContext, cog: str, op: str) -> None:
                 getattr(bot, f'{op}_extension')(str(ext))
                 done += 1
         except commands.ExtensionError as e:
-            await ctx.error(f'**Exception during batch-{op}ing:**\n```{e}```')
+            error: str = bot.mgr.sanitize_codeblock_content(str(e))
+            await ctx.error(f'**Exception during batch-{op}ing:**\n```{error}```')
         else:
             await ctx.success(f'All extensions **successfully {op}ed.** ({done})')
     else:
         try:
             getattr(bot, f'{op}_extension')(cog)
         except commands.ExtensionError as e:
-            await ctx.error(f'**Exception while {op}ing** `{cog}`**:**\n```{e}```')
+            error: str = bot.mgr.sanitize_codeblock_content(str(e))
+            await ctx.error(f'**Exception while {op}ing** `{cog}`**:**\n```{error}```')
         else:
             await ctx.success(f'**Successfully {op}ed** `{cog}`.')
 

cogs/backend/handle/errors/_error_tools.py

@@ -40,15 +40,20 @@ async def log_error_in_discord(ctx: GitBotContext, error: Exception, _actual=Non
             color=0xda4353,
             title=f'Error in `{ctx.command}` command'
         )
-        embed.add_field(name='Message', value=f'```{error}```')
-        embed.add_field(name='Traceback', value=f'```{format_tb(error.__traceback__)}```')
+        message: str = ctx.bot.mgr.sanitize_codeblock_content(str(error))
+        tb: str = ctx.bot.mgr.sanitize_codeblock_content(format_tb(error.__traceback__))
+        sanitized_args: str = ctx.bot.mgr.sanitize_codeblock_content(format_args(ctx.args))
+        sanitized_kwargs: str = ctx.bot.mgr.sanitize_codeblock_content(format_kwargs(ctx.kwargs))
+        embed.add_field(name='Message', value=f'```{message}```')
+        embed.add_field(name='Traceback', value=f'```{tb}```')
         embed.add_field(name='Arguments',
-                        value=f'```properties\nargs={format_args(ctx.args)}\nkwargs={format_kwargs(ctx.kwargs)}```')
+                        value=f'```properties\nargs={sanitized_args}\nkwargs={sanitized_kwargs}```')
     elif isinstance(error, commands.CommandNotFound):
+        error_text: str = ctx.bot.mgr.sanitize_codeblock_content(str(error))
         embed: GitBotEmbed = GitBotEmbed(
             color=0x0384fc,
             title='Nonexistent command!',
-            description=f'```{(error := str(error))}```',
+            description=f'```{error_text}```',
             footer='Closest existing command: ' + closest_existing_command_from_error(ctx.bot, error)
         )
     elif isinstance(error, (BadRequest, QueryError)):
@@ -57,10 +62,13 @@ async def log_error_in_discord(ctx: GitBotContext, error: Exception, _actual=Non
             title='GitHub API Error!',
             footer='The logs may contain more information.'
         )
-        embed.add_field(name='API Response', value=f'```diff\n- {error}```')
-        embed.add_field(name='Code Location', value=f'```{ctx.gh_query_debug.code_location}```')
+        api_response: str = ctx.bot.mgr.sanitize_codeblock_content(str(error))
+        code_location: str = ctx.bot.mgr.sanitize_codeblock_content(ctx.gh_query_debug.code_location)
+        embed.add_field(name='API Response', value=f'```diff\n- {api_response}```')
+        embed.add_field(name='Code Location', value=f'```{code_location}```')
         if ctx.gh_query_debug.additional_info:
-            embed.add_field(name='Additional Info', value=f'```{ctx.gh_query_debug.additional_info}```')
+            additional_info: str = ctx.bot.mgr.sanitize_codeblock_content(ctx.gh_query_debug.additional_info)
+            embed.add_field(name='Additional Info', value=f'```{additional_info}```')
         if ctx.gh_query_debug.status_code is not None:
             embed.add_field(name='Status Code', value=f'```c\n{error.status_code}```')
         ping_owner: bool = True

cogs/backend/workers/release_feed.py

@@ -79,7 +79,8 @@ async def handle_feed_repo(self,
 
         if body := new_release['release']['descriptionHTML']:
             body: str = ' '.join(BeautifulSoup(body, features='html.parser').getText().split())
-            body: str = f"```{self.bot.mgr.truncate(body, 400, full_word=True)}```".strip()
+            body = self.bot.mgr.sanitize_codeblock_content(self.bot.mgr.truncate(body, 400, full_word=True))
+            body: str = f"```{body}```".strip()
 
         author: dict = new_release["release"]["author"]
         author: str = f'Created by [{author["login"]}]({author["url"]}) on ' \

cogs/github/base/org.py

@@ -54,7 +54,8 @@ async def org_info_command(self, ctx: GitBotContext, organization: Optional[GitH
             members: str = ctx.fmt('one_member', f"{org['html_url']}/people") + '\n'
         email: str = f"Email: {org['email']}\n" if 'email' in org and org["email"] is not None else '\n'
         if org['description'] is not None and len(org['description']) > 0:
-            embed.add_field(name=f":notepad_spiral: {ctx.l.org.info.glossary[0]}:", value=f"```{org['description']}```")
+            description: str = self.bot.mgr.sanitize_codeblock_content(org['description'])
+            embed.add_field(name=f":notepad_spiral: {ctx.l.org.info.glossary[0]}:", value=f"```{description}```")
         repos: str = f"{ctx.l.org.info.repos.no_repos}\n" if org['public_repos'] == 0 else ctx.fmt('repos plural',
                                                                                                    org['public_repos'],
                                                                                                    f"{org['url']}?tab=repositories") + '\n'

cogs/github/base/repo/repo.py

@@ -60,8 +60,9 @@ async def repo_info_command(self, ctx: GitBotContext, repo: Optional[GitHubRepos
         open_issues: int = r['issues']['totalCount']
 
         if r['description'] is not None and len(r['description']) != 0:
+            description: str = self.bot.mgr.sanitize_codeblock_content(re.sub(MARKDOWN_EMOJI_RE, '', r['description']).strip())
             embed.add_field(name=f":notepad_spiral: {ctx.l.repo.info.glossary[0]}:",
-                            value=f"```{re.sub(MARKDOWN_EMOJI_RE, '', r['description']).strip()}```")
+                            value=f"```{description}```")
 
         watchers: str = ctx.fmt('watchers plural', watch, f"{r['url']}/watchers") if watch != 1 else ctx.fmt(
             'watchers singular', f"{r['url']}/watchers")
@@ -162,12 +163,13 @@ async def repo_files_command(self, ctx: GitBotContext, repo_or_path: GitHubRepos
         embeds: list = []
 
         def make_embed(items: list, ftr: str | None = None) -> GitBotEmbed:
+            sanitized_path: str | None = self.bot.mgr.sanitize_codeblock_content(path) if path else None
             return GitBotEmbed(
                     color=self.bot.mgr.c.rounded,
                     title=f'{self.bot.mgr.e.branch}  `{repo}`' + (f' ({ref})' if ref else ''),
                     description='\n'.join(
                             f'{self.bot.mgr.e.file}  [`{f["name"]}`]({f["html_url"]})' if f['type'] == 'file' else
-                            f'{self.bot.mgr.e.folder}  [`{f["name"]}`]({f["html_url"]})' for f in items) + ('\n' + f'```{path}```' if path else ''),
+                            f'{self.bot.mgr.e.folder}  [`{f["name"]}`]({f["html_url"]})' for f in items) + ('\n' + f'```{sanitized_path}```' if sanitized_path else ''),
                     url=link,
                     footer=ftr
             )

cogs/github/base/user.py

@@ -50,7 +50,8 @@ async def user_info_command(self, ctx: GitBotContext, user: Optional[GitHubUser]
         contrib_count: Optional[tuple] = u['contributions']
         orgs_c: int = u['organizations_count']
         if "bio" in u and u['bio'] is not None and len(u['bio']) > 0:
-            embed.add_field(name=f":notepad_spiral: {ctx.l.user.info.glossary[0]}:", value=f"```{u['bio']}```")
+            bio: str = self.bot.mgr.sanitize_codeblock_content(u['bio'])
+            embed.add_field(name=f":notepad_spiral: {ctx.l.user.info.glossary[0]}:", value=f"```{bio}```")
         occupation: str = (ctx.l.user.info.company + '\n').format(u['company']) if 'company' in u and u[
             'company'] is not None else ctx.l.user.info.no_company + '\n'
         orgs: str = (ctx.l.user.info.orgs.plural.format(orgs_c) if orgs_c != 0 else ctx.l.user.info.orgs.no_orgs) + '\n'

cogs/github/numbered/commits.py

@@ -115,6 +115,8 @@ async def commit_command(self,
             message: str = (f"{self.bot.mgr.truncate(commit['messageBody'], 247, full_word=True)}"
                             if commit['messageBody'] and commit['messageBody'] != commit['messageHeadline']
                             else '')
+            full_headline = self.bot.mgr.sanitize_codeblock_content(full_headline)
+            message = self.bot.mgr.sanitize_codeblock_content(message)
             empty: str = ctx.l.commit.fields.message.empty if not full_headline and not message else ''
             message: str = '```' + full_headline + message + empty + '```'
             embed.add_field(name=f':notepad_spiral: {ctx.l.commit.fields.message.name}:', value=message)

cogs/github/numbered/gist.py

@@ -90,7 +90,8 @@ async def build_gist_embed(self,
 
         stargazers_and_comments = f'{stargazers} and {comments}'
         info: str = f'{created_at}{updated_at}{stargazers_and_comments}'
-        content: str = self.bot.mgr.truncate(first_file['text'], 749, ' [...]').replace('`', '\u200b`')
+        content: str = self.bot.mgr.sanitize_codeblock_content(self.bot.mgr.truncate(first_file['text'], 749, ' [...]'),
+                                                               neutralize_mentions=False)
         embed.add_field(name=f':notepad_spiral: {ctx.l.gist.glossary[0]}:', value=f"```{self.extension(first_file['extension'])}\n{content}```")
         embed.add_field(name=f":mag_right: {ctx.l.gist.glossary[1]}:", value=info)
 

cogs/github/numbered/issue.py

@@ -64,6 +64,7 @@ async def issue_command(self, ctx: GitBotContext, repo: GitHubRepository, issue_
         else:
             body = None
         if body:
+            body = self.bot.mgr.sanitize_codeblock_content(body)
             embed.add_field(name=f':notepad_spiral: {ctx.l.issue.glossary[0]}:', value=f"```{body}```", inline=False)
         user: str = ctx.fmt('created_at',
                             self.bot.mgr.to_github_hyperlink(issue['author']['login']),

cogs/github/numbered/pr.py

@@ -66,8 +66,9 @@ async def pull_request_command(self,
         )
         embed.set_thumbnail(url=pr['author']['avatarUrl'])
         if all(['bodyText' in pr and pr['bodyText'], len(pr['bodyText'])]):
+            body: str = self.bot.mgr.sanitize_codeblock_content(self.bot.mgr.truncate(pr['bodyText'], 387, full_word=True))
             embed.add_field(name=':notepad_spiral: Body:',
-                            value=f"```{self.bot.mgr.truncate(pr['bodyText'], 387, full_word=True)}```",
+                            value=f"```{body}```",
                             inline=False)
         user: str = ctx.fmt('created_at',
                             self.bot.mgr.to_github_hyperlink(pr['author']['login']),

cogs/github/other/snippets/_snippet_tools.py

@@ -48,7 +48,11 @@ async def get_text_from_url_and_data(ctx: 'GitBotContext',
     text: str = ''.join(lines)
     ctx.lines_total = len(lines_)
     if text:
-        return f"```{extension}\n{text.rstrip()}\n```" if wrap_in_codeblock else text.rstrip(), None
+        text = text.rstrip()
+        if wrap_in_codeblock:
+            text = ctx.bot.mgr.sanitize_codeblock_content(text, neutralize_mentions=False)
+            return f"```{extension}\n{text}\n```", None
+        return text, None
     return '', None
 
 

cogs/python/pypi.py

@@ -56,8 +56,9 @@ async def project_info_command(self, ctx: GitBotContext, project: PyPIProject) -
         )
 
         if data['info']['summary'] is not None and len(data['info']['summary']) != 0:
+            summary: str = self.bot.mgr.sanitize_codeblock_content(data['info']['summary'].strip())
             embed.add_field(name=f":notepad_spiral: {ctx.l.pypi.info.glossary[0]}:",
-                            value=f"```{data['info']['summary'].strip()}```")
+                            value=f"```{summary}```")
         author: str = ctx.fmt('author', f'[{(author := data["info"]["author"])}]'
                                         f'({await self.bot.mgr.ensure_http_status(f"https://pypi.org/user/{author}", alt="")})') + '\n'
 

cogs/rust/crates.py

@@ -53,8 +53,9 @@ async def crate_info_command(self, ctx: GitBotContext, crate: CratesIOCrate) ->
             )
 
             if (crate_desc := data['crate']['description']) is not None and len(crate_desc) != 0:
+                crate_desc = self.bot.mgr.sanitize_codeblock_content(crate_desc.strip())
                 embed.add_field(name=f":notepad_spiral: {ctx.l.crates.info.glossary[0]}:",
-                                value=f"```{crate_desc.strip()}```")
+                                value=f"```{crate_desc}```")
 
             more_authors: str = f' {ctx.fmt("more_authors", f"[{len(owners) - 5}]({crate_url})")}' if len(
                 owners) > 5 else ''

lib/manager.py

@@ -109,9 +109,9 @@
     @staticmethod
     def git_rev_parse_head() -> str | None:
         try:
             return subprocess.check_output(['git',
                                             'rev-parse',
                                             'HEAD']).decode('utf-8').strip()
         except subprocess.CalledProcessError:
             return None
 
@@ -231,6 +231,23 @@
             return str_[:length - len(ending)] + ending
         return str_
 
+    @staticmethod
+    def sanitize_codeblock_content(content: str | None, neutralize_mentions: bool = True) -> str:
+        """
+        Harden untrusted text for Discord fenced code blocks.
+
+        :param content: The text to sanitize
+        :param neutralize_mentions: Whether to prevent mention abuse
+        :return: The sanitized text
+        """
+        if not content:
+            return ''
+        content = content.replace('```', '`\u200b`\u200b`')
+        content = content.replace('~~~', '~\u200b~\u200b~')
+        if neutralize_mentions:
+            content = content.replace('@', '@\u200b')
+        return content
+
     @staticmethod
     def flatten(iterable: Iterable) -> Iterable:
         return list(iterable | traverse)