fix: encode or validate all user provided data passed to urls (#739)

Fix a path traversal attack with request handling where untrusted values
from users were passed directly into urls.

Author: onerandomusername

monty/exts/info/codesnippets.py

@@ -16,7 +16,7 @@
 from monty.bot import Monty
 from monty.log import get_logger
 from monty.utils import scheduling
-from monty.utils.helpers import EXPAND_BUTTON_PREFIX, decode_github_link
+from monty.utils.helpers import EXPAND_BUTTON_PREFIX, block_url_traversal, decode_github_link
 from monty.utils.markdown import remove_codeblocks
 from monty.utils.messages import DeleteButton, suppress_embeds
 
@@ -30,7 +30,8 @@
 
 # start_char, line_delimiter, and end_char are currently unused.
 GITHUB_RE = re.compile(
-    r"https?:\/\/github\.(?:com|dev)\/(?P<user>[a-zA-Z0-9-]+)\/(?P<repo>[a-zA-Z0-9-]+)\/(?:blob|tree)\/(?P<path>[^#>]+)(\?[^#>]+)?"
+    r"https?:\/\/github\.(?:com|dev)\/(?P<user>[a-zA-Z0-9-]+)\/(?P<repo>[a-zA-Z0-9-]+)\/(?:blob|tree)\/"
+    r"(?P<path>[^#>]+/[^#>]+)(\?[^#>]+)?"
     r"(?:(#L(?P<L>L)?(?P<start_line>\d+)(?(L)C(?P<start_char>\d+))(?:(?P<line_delimiter>[-~\:]"
     r"|(\.\.))L(?P<end_line>\d+)(?(L)C(?P<end_char>\d+)))?))"
 )
@@ -119,17 +120,22 @@ async def _fetch_github_snippet(
         repo: str,
         path: str,
         start_line: str,
-        end_line: str,
+        end_line: str | None,
         **kwargs: "NoReturn",
     ) -> str:
         """Fetches a snippet from a GitHub repo."""
+        user, repo, start_line = block_url_traversal(user, repo, start_line)
+        if end_line:
+            end_line = block_url_traversal(end_line)
+
         # Search the GitHub API for the specified branch
         r = await self.bot.github.rest.repos.async_list_branches(user, repo, per_page=100)
         branches = r.json()
         r = await self.bot.github.rest.repos.async_list_tags(user, repo, per_page=100)
         tags = r.json()
         refs = branches + tags
         ref, encoded_file_path = self._find_ref(path, refs)
+        ref, encoded_file_path = block_url_traversal(ref, encoded_file_path)
 
         r = await self.bot.github.rest.repos.async_get_content(
             user,
@@ -156,6 +162,9 @@ async def _fetch_github_gist_snippet(
         **kwargs: "NoReturn",
     ) -> str:
         """Fetches a snippet from a GitHub gist."""
+        gist_id, revision, file_path, start_line, end_line = block_url_traversal(
+            gist_id, revision, file_path, start_line, end_line
+        )
         if revision:
             endpoint = self.bot.github.rest.gists.async_get_revision(gist_id, revision)
         else:
@@ -186,20 +195,17 @@ async def _fetch_gitlab_snippet(
         **kwargs: "NoReturn",
     ) -> str:
         """Fetches a snippet from a GitLab repo."""
-        enc_repo = quote_plus(repo)
-
+        repo, start_line, end_line = block_url_traversal(repo, start_line, end_line)
         # Searches the GitLab API for the specified branch
-        branches = await self._fetch_response(
-            f"https://gitlab.com/api/v4/projects/{enc_repo}/repository/branches", "json"
-        )
-        tags = await self._fetch_response(f"https://gitlab.com/api/v4/projects/{enc_repo}/repository/tags", "json")
+        branches = await self._fetch_response(f"https://gitlab.com/api/v4/projects/{repo}/repository/branches", "json")
+        tags = await self._fetch_response(f"https://gitlab.com/api/v4/projects/{repo}/repository/tags", "json")
         refs = branches + tags
         ref, file_path = self._find_ref(path, refs)
-        enc_ref = quote_plus(ref)
-        enc_file_path = quote_plus(file_path)
+        ref = quote_plus(ref)
+        path = quote_plus(file_path)
 
         file_contents = await self._fetch_response(
-            f"https://gitlab.com/api/v4/projects/{enc_repo}/repository/files/{enc_file_path}/raw?ref={enc_ref}",
+            f"https://gitlab.com/api/v4/projects/{repo}/repository/files/{path}/raw?ref={ref}",
             "text",
         )
         return self._snippet_to_codeblock(file_contents, file_path, start_line, end_line)

monty/exts/info/github_info.py

@@ -27,14 +27,20 @@
 from monty.log import get_logger
 from monty.utils import responses, scheduling
 from monty.utils.extensions import invoke_help_command
-from monty.utils.helpers import fromisoformat, get_num_suffix
+from monty.utils.helpers import block_url_traversal, fromisoformat, get_num_suffix
 from monty.utils.markdown import DiscordRenderer, remove_codeblocks
 from monty.utils.messages import DeleteButton, extract_urls, suppress_embeds
 
 
 # Maximum number of issues in one message
 MAXIMUM_ISSUES = 6
 
+USERNAME_RE = re.compile(r"[a-zA-Z0-9][a-zA-Z0-9\-]{0,38}")
+REPO_RE = re.compile(
+    rf"(?:(?P<owner>{USERNAME_RE.pattern})\/)?"
+    r"(?P<repo>[\w\-\.]{1,100})"
+)
+
 # Regex used when looking for automatic linking in messages
 # regex101 of current regex https://regex101.com/r/V2ji8M/6
 AUTOMATIC_REGEX = re.compile(
@@ -285,6 +291,11 @@ async def github_group(self, ctx: commands.Context, user_or_repo: str = "") -> N
     @github_group.command(name="user", aliases=("userinfo",))
     async def github_user_info(self, ctx: commands.Context, username: str) -> None:
         """Fetches a user's GitHub information."""
+        if not (match := USERNAME_RE.fullmatch(username)):
+            msg = f"`{username}` is not a valid GitHub username."
+            raise commands.BadArgument(msg)
+        username = match.group(0)
+
         async with ctx.typing():
             try:
                 resp = await self.bot.github.rest.users.async_get_by_username(username)
@@ -375,18 +386,24 @@ async def github_repo_info(self, ctx: commands.Context, *repository: str) -> Non
         else:
             repo_name = ""
 
+        if not (match := REPO_RE.fullmatch(repo_name)):
+            msg = f"`{repo_name}` is not a valid GitHub repository name."
+            raise commands.BadArgument(msg)
+        repo_name = match.group(0)
+
         if not repo_name or repo_name.count("/") > 1:
             args = " ".join(repository[:2])
 
             raise commands.BadArgument(
                 "The repository should look like `user/reponame` or `user reponame`"
                 + (f", not `{args}`." if "`" not in args and len(args) < 20 else ".")
             )
-            return
+
+        owner, repo = block_url_traversal(*repo_name.split("/", 1))
 
         async with ctx.typing():
             try:
-                resp = await self.bot.github.rest.repos.async_get(*repo_name.split("/", 1))
+                resp = await self.bot.github.rest.repos.async_get(owner, repo)
                 repo_data = resp.json()
             except githubkit.exception.RequestFailed:
                 # There won't be a message key if this repo exists
@@ -461,6 +478,7 @@ async def fetch_issues(
         Returns IssueState on success, FetchError on failure.
         """
         json_data = None
+        user, repository = block_url_traversal(user, repository)
         if not is_discussion:  # not a discussion, or uncertain
             try:
                 r = await self.bot.github.rest.issues.async_get(user, repository, number)

monty/exts/python/pypi.py

@@ -151,6 +151,10 @@ async def fetch_package_list(self, *, use_cache: bool = True) -> None:
 
     async def fetch_package(self, package: str) -> dict[str, Any] | None:
         """Fetch a package from PyPI."""
+        if characters := self.check_characters(package):
+            msg = f"Illegal character(s) passed into command: '{disnake.utils.escape_markdown(characters.group(0))}'"
+            raise MontyCommandError(msg)
+
         async with self.bot.http_session.get(JSON_URL.format(package=package), headers=PYPI_API_HEADERS) as response:
             if response.status == 200 and response.content_type == "application/json":
                 return await response.json()
@@ -265,9 +269,6 @@ async def pypi_package(
         embed.set_thumbnail(url=PYPI_ICON)
 
         defer_task = None
-        if characters := self.check_characters(package):
-            msg = f"Illegal character(s) passed into command: '{disnake.utils.escape_markdown(characters.group(0))}'"
-            raise MontyCommandError(msg)
 
         response_json = await self.fetch_package(package)
         if not response_json:

monty/utils/helpers.py

@@ -4,7 +4,7 @@
 import datetime
 import ssl
 from typing import TYPE_CHECKING, Any, TypeVar, overload
-from urllib.parse import urlsplit, urlunsplit
+from urllib.parse import quote_plus, urlsplit, urlunsplit
 
 import base65536
 import dateutil.parser
@@ -76,6 +76,21 @@ def pad_base64(data: str) -> str:
     return data + "=" * (-len(data) % 4)
 
 
+@overload
+def block_url_traversal(arg: str) -> str: ...
+
+
+@overload
+def block_url_traversal(*args: str) -> tuple[str, ...]: ...
+
+
+def block_url_traversal(arg: str, *args: str) -> str | tuple[str, ...]:
+    """Normalise a string for use in a URL by fixing slashes and escaping `/`."""
+    if not args:
+        return quote_plus(arg)
+    return tuple(quote_plus(a) for a in (arg, *args))
+
+
 EXPAND_BUTTON_PREFIX = "ghexp-v1:"