Extract shared logic for using Popen safely on Windows

This creates git.util.safer_popen that, on non-Windows systems, is
bound to subprocess.Popen (to avoid introducing unnecessary
latency). On Windows, it is a function that wraps subprocess.Popen,
consolidating two pieces of logic that had previously been
duplicated:

1. Temporarily setting NoDefaultCurrentDirectoryInExePath in the
   calling environment and, when shell=True is used, setting it in
   the subprocess environment as well. This prevents executables
   specified as single names (which are mainly "git" and, for
   hooks, "bash.exe") from being searched for in the current
   working directory of GitPython or, when a shell is used, the
   current working directory of the shell used to run them.

2. Passing the CREATE_NO_WINDOW and CREATE_NEW_PROCESS_GROUP flags
   as creationflags. This is not a security measure. It is
   indirectly related to safety in that CREATE_NO_WINDOW eliminated
   at least some, and possibly all, cases where calling Git.execute
   (directly, or indirectly via a dynamic method) with shell=True
   conferred an advantage over the inherently more secure default
   of shell=False; and CREATE_NEW_PROCESS facilitates some ways of
   terminating subprocesses that would otherwise be unavailable,
   thereby making resource exhaustion less likely. But really the
   reason I included creationflags here is that it seems it should
   always be used in the same situations as preventing the current
   directory from being searched (and always was), and including it
   further reduces code duplication and simplifies calling code.

This commit does not improve security or robustness, because these
features were already present. Instead, this moves them to a single
location. It also documents them by giving the function bound to
safer_popen on Windows, _safer_popen_windows, a detailed docstring.

Because there would otherwise be potential for confusion on the
different ways to perform or customize path searches, I have also
added a doctring to py_where noting its limited use case and its
relationship to shutil.which and non-shell search.

(The search in _safer_popen_windows is typically a non-shell
search, which is why it cannot be reimplemented to do its own
lookup by calling an only slightly modified version of
shutil.which, without a risk of breaking some currently working
uses. It may, however, be possible to fix the race condition by
doing something analogous for Windows non-shell search behavior,
which is largely but not entirely described in the documentation
for CreateProcessW.)

git/cmd.py

@@ -29,8 +29,8 @@
     cygpath,
     expand_path,
     is_cygwin_git,
-    patch_env,
     remove_password_if_present,
+    safer_popen,
     stream_copy,
 )
 
@@ -225,14 +225,6 @@ def dict_to_slots_and__excluded_are_none(self: object, d: Mapping[str, Any], exc
 ## -- End Utilities -- @}
 
 
-if os.name == "nt":
-    # CREATE_NEW_PROCESS_GROUP is needed to allow killing it afterwards. See:
-    # https://docs.python.org/3/library/subprocess.html#subprocess.Popen.send_signal
-    PROC_CREATIONFLAGS = subprocess.CREATE_NO_WINDOW | subprocess.CREATE_NEW_PROCESS_GROUP
-else:
-    PROC_CREATIONFLAGS = 0
-
-
 class Git(LazyMixin):
     """The Git class manages communication with the Git binary.
 
@@ -985,28 +977,20 @@ def execute(
         if inline_env is not None:
             env.update(inline_env)
 
-        if shell is None:
-            shell = self.USE_SHELL
-
         if os.name == "nt":
             cmd_not_found_exception = OSError
             if kill_after_timeout is not None:
                 raise GitCommandError(
                     redacted_command,
                     '"kill_after_timeout" feature is not supported on Windows.',
                 )
-            # Search PATH but not CWD. The "1" can be any value. We'll patch just before
-            # the Popen call and unpatch just after, or we get a worse race condition.
-            maybe_patch_caller_env = patch_env("NoDefaultCurrentDirectoryInExePath", "1")
-            if shell:
-                # Modify the direct shell subprocess's own search behavior accordingly.
-                env["NoDefaultCurrentDirectoryInExePath"] = "1"
         else:
             cmd_not_found_exception = FileNotFoundError
-            maybe_patch_caller_env = contextlib.nullcontext()
         # END handle
 
         stdout_sink = PIPE if with_stdout else getattr(subprocess, "DEVNULL", None) or open(os.devnull, "wb")
+        if shell is None:
+            shell = self.USE_SHELL
         log.debug(
             "Popen(%s, cwd=%s, stdin=%s, shell=%s, universal_newlines=%s)",
             redacted_command,
@@ -1016,20 +1000,18 @@ def execute(
             universal_newlines,
         )
         try:
-            with maybe_patch_caller_env:
-                proc = Popen(
-                    command,
-                    env=env,
-                    cwd=cwd,
-                    bufsize=-1,
-                    stdin=(istream or DEVNULL),
-                    stderr=PIPE,
-                    stdout=stdout_sink,
-                    shell=shell,
-                    universal_newlines=universal_newlines,
-                    creationflags=PROC_CREATIONFLAGS,
-                    **subprocess_kwargs,
-                )
+            proc = safer_popen(
+                command,
+                env=env,
+                cwd=cwd,
+                bufsize=-1,
+                stdin=(istream or DEVNULL),
+                stderr=PIPE,
+                stdout=stdout_sink,
+                shell=shell,
+                universal_newlines=universal_newlines,
+                **subprocess_kwargs,
+            )
         except cmd_not_found_exception as err:
             raise GitCommandNotFound(redacted_command, err) from err
         else:

git/index/fun.py

@@ -3,7 +3,6 @@
 
 """Standalone functions to accompany the index implementation and make it more versatile."""
 
-import contextlib
 from io import BytesIO
 import os
 import os.path as osp
@@ -19,15 +18,15 @@
 )
 import subprocess
 
-from git.cmd import PROC_CREATIONFLAGS, handle_process_output
+from git.cmd import handle_process_output
 from git.compat import defenc, force_bytes, force_text, safe_decode
 from git.exc import HookExecutionError, UnmergedEntriesError
 from git.objects.fun import (
     traverse_tree_recursive,
     traverse_trees_recursive,
     tree_to_stream,
 )
-from git.util import IndexFileSHA1Writer, finalize_process, patch_env
+from git.util import IndexFileSHA1Writer, finalize_process, safer_popen
 from gitdb.base import IStream
 from gitdb.typ import str_tree_type
 
@@ -91,10 +90,6 @@ def run_commit_hook(name: str, index: "IndexFile", *args: str) -> None:
     env = os.environ.copy()
     env["GIT_INDEX_FILE"] = safe_decode(str(index.path))
     env["GIT_EDITOR"] = ":"
-    if os.name == "nt":
-        maybe_patch_caller_env = patch_env("NoDefaultCurrentDirectoryInExePath", "1")
-    else:
-        maybe_patch_caller_env = contextlib.nullcontext()
     cmd = [hp]
     try:
         if os.name == "nt" and not _has_file_extension(hp):
@@ -103,15 +98,13 @@ def run_commit_hook(name: str, index: "IndexFile", *args: str) -> None:
             relative_hp = Path(hp).relative_to(index.repo.working_dir).as_posix()
             cmd = ["bash.exe", relative_hp]
 
-        with maybe_patch_caller_env:
-            process = subprocess.Popen(
-                cmd + list(args),
-                env=env,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE,
-                cwd=index.repo.working_dir,
-                creationflags=PROC_CREATIONFLAGS,
-            )
+        process = safer_popen(
+            cmd + list(args),
+            env=env,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            cwd=index.repo.working_dir,
+        )
     except Exception as ex:
         raise HookExecutionError(hp, ex) from ex
     else:

git/util.py

@@ -33,6 +33,7 @@
     IO,
     Iterator,
     List,
+    Mapping,
     Optional,
     Pattern,
     Sequence,
@@ -327,6 +328,17 @@ def _get_exe_extensions() -> Sequence[str]:
 
 
 def py_where(program: str, path: Optional[PathLike] = None) -> List[str]:
+    """Perform a path search to assist :func:`is_cygwin_git`.
+
+    This is not robust for general use. It is an implementation detail of
+    :func:`is_cygwin_git`. When a search following all shell rules is needed,
+    :func:`shutil.which` can be used instead.
+
+    :note: Neither this function nor :func:`shutil.which` will predict the effect of an
+        executable search on a native Windows system due to a :class:`subprocess.Popen`
+        call without ``shell=True``, because shell and non-shell executable search on
+        Windows differ considerably.
+    """
     # From: http://stackoverflow.com/a/377028/548792
     winprog_exts = _get_exe_extensions()
 
@@ -524,6 +536,67 @@ def remove_password_if_present(cmdline: Sequence[str]) -> List[str]:
     return new_cmdline
 
 
+def _safer_popen_windows(
+    command: Union[str, Sequence[Any]],
+    *,
+    shell: bool = False,
+    env: Optional[Mapping[str, str]] = None,
+    **kwargs: Any,
+) -> subprocess.Popen:
+    """Call :class:`subprocess.Popen` on Windows but don't include a CWD in the search.
+
+    This avoids an untrusted search path condition where a file like ``git.exe`` in a
+    malicious repository would be run when GitPython operates on the repository. The
+    process using GitPython may have an untrusted repository's working tree as its
+    current working directory. Some operations may temporarily change to that directory
+    before running a subprocess. In addition, while by default GitPython does not run
+    external commands with a shell, it can be made to do so, in which case the CWD of
+    the subprocess, which GitPython usually sets to a repository working tree, can
+    itself be searched automatically by the shell. This wrapper covers all those cases.
+
+    :note: This currently works by setting the ``NoDefaultCurrentDirectoryInExePath``
+        environment variable during subprocess creation. It also takes care of passing
+        Windows-specific process creation flags, but that is unrelated to path search.
+
+    :note: The current implementation contains a race condition on :attr:`os.environ`.
+        GitPython isn't thread-safe, but a program using it on one thread should ideally
+        be able to mutate :attr:`os.environ` on another, without unpredictable results.
+        See comments in https://github.com/gitpython-developers/GitPython/pull/1650.
+    """
+    # CREATE_NEW_PROCESS_GROUP is needed for some ways of killing it afterwards. See:
+    # https://docs.python.org/3/library/subprocess.html#subprocess.Popen.send_signal
+    # https://docs.python.org/3/library/subprocess.html#subprocess.CREATE_NEW_PROCESS_GROUP
+    creationflags = subprocess.CREATE_NO_WINDOW | subprocess.CREATE_NEW_PROCESS_GROUP
+
+    # When using a shell, the shell is the direct subprocess, so the variable must be
+    # set in its environment, to affect its search behavior. (The "1" can be any value.)
+    if shell:
+        safer_env = {} if env is None else dict(env)
+        safer_env["NoDefaultCurrentDirectoryInExePath"] = "1"
+    else:
+        safer_env = env
+
+    # When not using a shell, the current process does the search in a CreateProcessW
+    # API call, so the variable must be set in our environment. With a shell, this is
+    # unnecessary, in versions where https://github.com/python/cpython/issues/101283 is
+    # patched. If not, in the rare case the ComSpec environment variable is unset, the
+    # shell is searched for unsafely. Setting NoDefaultCurrentDirectoryInExePath in all
+    # cases, as here, is simpler and protects against that. (The "1" can be any value.)
+    with patch_env("NoDefaultCurrentDirectoryInExePath", "1"):
+        return subprocess.Popen(
+            command,
+            shell=shell,
+            env=safer_env,
+            creationflags=creationflags,
+            **kwargs,
+        )
+
+
+if os.name == "nt":
+    safer_popen = _safer_popen_windows
+else:
+    safer_popen = subprocess.Popen
+
 # } END utilities
 
 # { Classes

test/test_git.py

@@ -25,7 +25,7 @@
 import ddt
 
 from git import Git, refresh, GitCommandError, GitCommandNotFound, Repo, cmd
-from git.util import cwd, finalize_process
+from git.util import cwd, finalize_process, safer_popen
 from test.lib import TestBase, fixture_path, with_rw_directory
 
 
@@ -115,28 +115,28 @@ def test_it_transforms_kwargs_into_git_command_arguments(self):
     def _do_shell_combo(self, value_in_call, value_from_class):
         with mock.patch.object(Git, "USE_SHELL", value_from_class):
             # git.cmd gets Popen via a "from" import, so patch it there.
-            with mock.patch.object(cmd, "Popen", wraps=cmd.Popen) as mock_popen:
+            with mock.patch.object(cmd, "safer_popen", wraps=cmd.safer_popen) as mock_safer_popen:
                 # Use a command with no arguments (besides the program name), so it runs
                 # with or without a shell, on all OSes, with the same effect.
                 self.git.execute(["git"], with_exceptions=False, shell=value_in_call)
 
-        return mock_popen
+        return mock_safer_popen
 
     @ddt.idata(_shell_cases)
     def test_it_uses_shell_or_not_as_specified(self, case):
         """A bool passed as ``shell=`` takes precedence over `Git.USE_SHELL`."""
         value_in_call, value_from_class, expected_popen_arg = case
-        mock_popen = self._do_shell_combo(value_in_call, value_from_class)
-        mock_popen.assert_called_once()
-        self.assertIs(mock_popen.call_args.kwargs["shell"], expected_popen_arg)
+        mock_safer_popen = self._do_shell_combo(value_in_call, value_from_class)
+        mock_safer_popen.assert_called_once()
+        self.assertIs(mock_safer_popen.call_args.kwargs["shell"], expected_popen_arg)
 
     @ddt.idata(full_case[:2] for full_case in _shell_cases)
     def test_it_logs_if_it_uses_a_shell(self, case):
         """``shell=`` in the log message agrees with what is passed to `Popen`."""
         value_in_call, value_from_class = case
         with self.assertLogs(cmd.log, level=logging.DEBUG) as log_watcher:
-            mock_popen = self._do_shell_combo(value_in_call, value_from_class)
-        self._assert_logged_for_popen(log_watcher, "shell", mock_popen.call_args.kwargs["shell"])
+            mock_safer_popen = self._do_shell_combo(value_in_call, value_from_class)
+        self._assert_logged_for_popen(log_watcher, "shell", mock_safer_popen.call_args.kwargs["shell"])
 
     @ddt.data(
         ("None", None),
@@ -405,13 +405,12 @@ def counter_stderr(line):
             fixture_path("cat_file.py"),
             str(fixture_path("issue-301_stderr")),
         ]
-        proc = subprocess.Popen(
+        proc = safer_popen(
             cmdline,
             stdin=None,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             shell=False,
-            creationflags=cmd.PROC_CREATIONFLAGS,
         )
 
         handle_process_output(proc, counter_stdout, counter_stderr, finalize_process)