[Downloader] Use shlex for subprocesses (#2421)

redbot/cogs/downloader/repo_manager.py

@@ -2,13 +2,15 @@ import asyncio
 import functools
 import os
 import pkgutil
+import shlex
 import shutil
 import re
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 from subprocess import run as sp_run, PIPE
+from string import Formatter
 from sys import executable
-from typing import Tuple, MutableMapping, Union, Optional
+from typing import List, Tuple, Iterable, MutableMapping, Union, Optional
 
 from redbot.core import data_manager, commands
 from redbot.core.utils import safe_delete
@@ -22,6 +24,17 @@ from .log import log
 _ = Translator("RepoManager", __file__)
 
 
+class ProcessFormatter(Formatter):
+    def vformat(self, format_string, args, kwargs):
+        return shlex.split(super().vformat(format_string, args, kwargs))
+
+    def get_value(self, key, args, kwargs):
+        obj = super().get_value(key, args, kwargs)
+        if isinstance(obj, str) or not isinstance(obj, Iterable):
+            return shlex.quote(str(obj))
+        return " ".join(shlex.quote(str(o)) for o in obj)
+
+
 class Repo(RepoJSONMixin):
     GIT_CLONE = "git clone --recurse-submodules -b {branch} {url} {folder}"
     GIT_CLONE_NO_BRANCH = "git clone --recurse-submodules {url} {folder}"
@@ -94,8 +107,11 @@ class Repo(RepoJSONMixin):
         :return: Mapping of filename -> status_letter
         """
         p = await self._run(
-            self.GIT_DIFF_FILE_STATUS.format(
+            ProcessFormatter().format(
-                path=self.folder_path, old_hash=old_hash, new_hash=new_hash
+                self.GIT_DIFF_FILE_STATUS,
+                path=self.folder_path,
+                old_hash=old_hash,
+                new_hash=new_hash,
             )
         )
 
@@ -124,7 +140,8 @@ class Repo(RepoJSONMixin):
         :return: Git commit note log
         """
         p = await self._run(
-            self.GIT_LOG.format(
+            ProcessFormatter().format(
+                self.GIT_LOG,
                 path=self.folder_path,
                 old_hash=old_commit_hash,
                 relative_file_path=relative_file_path,
@@ -190,13 +207,15 @@ class Repo(RepoJSONMixin):
 
         if self.branch is not None:
             p = await self._run(
-                self.GIT_CLONE.format(
+                ProcessFormatter().format(
-                    branch=self.branch, url=self.url, folder=self.folder_path
+                    self.GIT_CLONE, branch=self.branch, url=self.url, folder=self.folder_path
-                ).split()
+                )
             )
         else:
             p = await self._run(
-                self.GIT_CLONE_NO_BRANCH.format(url=self.url, folder=self.folder_path).split()
+                ProcessFormatter().format(
+                    self.GIT_CLONE_NO_BRANCH, url=self.url, folder=self.folder_path
+                )
             )
 
         if p.returncode:
@@ -226,7 +245,9 @@ class Repo(RepoJSONMixin):
                 "A git repo does not exist at path: {}".format(self.folder_path)
             )
 
-        p = await self._run(self.GIT_CURRENT_BRANCH.format(path=self.folder_path).split())
+        p = await self._run(
+            ProcessFormatter().format(self.GIT_CURRENT_BRANCH, path=self.folder_path)
+        )
 
         if p.returncode != 0:
             raise errors.GitException(
@@ -259,7 +280,7 @@ class Repo(RepoJSONMixin):
             )
 
         p = await self._run(
-            self.GIT_LATEST_COMMIT.format(path=self.folder_path, branch=branch).split()
+            ProcessFormatter().format(self.GIT_LATEST_COMMIT, path=self.folder_path, branch=branch)
         )
 
         if p.returncode != 0:
@@ -290,7 +311,7 @@ class Repo(RepoJSONMixin):
         if folder is None:
             folder = self.folder_path
 
-        p = await self._run(Repo.GIT_DISCOVER_REMOTE_URL.format(path=folder).split())
+        p = await self._run(ProcessFormatter().format(Repo.GIT_DISCOVER_REMOTE_URL, path=folder))
 
         if p.returncode != 0:
             raise errors.NoRemoteURL("Unable to discover a repo URL.")
@@ -316,7 +337,7 @@ class Repo(RepoJSONMixin):
             )
 
         p = await self._run(
-            self.GIT_HARD_RESET.format(path=self.folder_path, branch=branch).split()
+            ProcessFormatter().format(self.GIT_HARD_RESET, path=self.folder_path, branch=branch)
         )
 
         if p.returncode != 0:
@@ -340,7 +361,7 @@ class Repo(RepoJSONMixin):
 
         await self.hard_reset(branch=curr_branch)
 
-        p = await self._run(self.GIT_PULL.format(path=self.folder_path).split())
+        p = await self._run(ProcessFormatter().format(self.GIT_PULL, path=self.folder_path))
 
         if p.returncode != 0:
             raise errors.UpdateError(
@@ -463,9 +484,9 @@ class Repo(RepoJSONMixin):
         # TODO: Check and see if any of these modules are already available
 
         p = await self._run(
-            self.PIP_INSTALL.format(
+            ProcessFormatter().format(
-                python=executable, target_dir=target_dir, reqs=" ".join(requirements)
+                self.PIP_INSTALL, python=executable, target_dir=target_dir, reqs=requirements
-            ).split()
+            )
         )
 
         if p.returncode != 0:
@@ -509,7 +530,7 @@ class Repo(RepoJSONMixin):
 
 class RepoManager:
 
-    GITHUB_OR_GITLAB_RE = re.compile("https?://git(?:hub)|(?:lab)\.com/")
+    GITHUB_OR_GITLAB_RE = re.compile(r"https?://git(?:hub)|(?:lab)\.com/")
     TREE_URL_RE = re.compile(r"(?P<tree>/tree)/(?P<branch>\S+)$")
 
     def __init__(self):