Reject submodule move destinations through intermediate symlinks

doc/source/changes.rst

@@ -2,6 +2,19 @@
 Changelog
 =========
 
+3.1.63
+======
+
+Security fixes for
+
+* https://github.com/gitpython-developers/GitPython/security/advisories/GHSA-gq48-pqfc-9p58
+
+If you can, also try and provide feedback on the upcoming v4 branch
+https://github.com/gitpython-developers/GitPython/pull/2177 - patches welcome.
+
+See the following for all changes.
+https://github.com/gitpython-developers/GitPython/releases/tag/3.1.63
+
 3.1.62
 ======
 

git/objects/submodule/base.py

@@ -9,6 +9,7 @@
 import ntpath
 import os
 import os.path as osp
+from pathlib import Path
 import shlex
 import stat
 import sys
@@ -255,7 +256,7 @@ def _config_parser(
         # END handle parent_commit
         fp_module: Union[str, BytesIO]
         if not repo.bare and parent_matches_head and repo.working_tree_dir:
-            fp_module = osp.join(repo.working_tree_dir, cls.k_modules_file)
+            fp_module = cls._checked_abspath(repo.working_tree_dir, cls.k_modules_file)
         else:
             assert parent_commit is not None, "need valid parent_commit in bare repositories"
             try:
@@ -322,7 +323,7 @@ def _module_abspath(cls, parent_repo: "Repo", path: PathLike, name: str) -> Path
         if cls._need_gitfile_submodules(parent_repo.git):
             return osp.join(parent_repo.git_dir, "modules", name)
         if parent_repo.working_tree_dir:
-            return osp.join(parent_repo.working_tree_dir, path)
+            return cls._checked_abspath(parent_repo.working_tree_dir, cls._to_relative_path(parent_repo, path))
         raise NotADirectoryError()
 
     @classmethod
@@ -361,8 +362,11 @@ def _clone_repo(
         :param kwargs:
             Additional arguments given to :manpage:`git-clone(1)`.
         """
+        path = cls._to_relative_path(repo, path)
+        if repo.working_tree_dir is None:
+            raise NotADirectoryError("Submodules require a working tree")
+        module_checkout_path = cls._checked_abspath(repo.working_tree_dir, path)
         module_abspath = cls._module_abspath(repo, path, name)
-        module_checkout_path = module_abspath
         if cls._need_gitfile_submodules(repo.git):
             if not allow_unsafe_options:
                 Git.check_unsafe_options(Git._option_candidates([], kwargs), repo.unsafe_git_clone_options)
@@ -373,11 +377,22 @@ def _clone_repo(
                         repo.unsafe_git_clone_options,
                     )
             allow_unsafe_options = True
+            if osp.islink(module_abspath):
+                # Clone into the target while retaining the metadata alias. Git for
+                # Windows cannot initialize through a dangling directory symlink.
+                # Read the link explicitly for Python 3.7, and remove the Windows
+                # namespace prefix returned by newer Python versions for Git.
+                target = os.readlink(module_abspath)
+                if sys.platform == "win32":
+                    if target.startswith("\\\\?\\UNC\\"):
+                        target = "\\\\" + target[8:]
+                    elif target.startswith("\\\\?\\"):
+                        target = target[4:]
+                module_abspath = to_native_path_linux(osp.join(osp.dirname(module_abspath), target))
             kwargs["separate_git_dir"] = module_abspath
             module_abspath_dir = osp.dirname(module_abspath)
             if not osp.isdir(module_abspath_dir):
                 os.makedirs(module_abspath_dir)
-            module_checkout_path = osp.join(repo.working_tree_dir, path)  # type: ignore[arg-type]
 
         if url.startswith("../"):
             remote_name = cast("RemoteReference", repo.active_branch.tracking_branch()).remote_name
@@ -419,13 +434,43 @@ def abspath(self) -> PathLike:
         root = self.repo.working_tree_dir
         if root is None:
             return super().abspath
-        path = root
-        for component in os.fspath(self._to_relative_path(self.repo, self.path)).split("/"):
-            path = join_path_native(path, component)
+        return self._checkout_abspath(self._to_relative_path(self.repo, self.path))
+
+    def _checkout_abspath(self, relative_path: PathLike, allow_final_symlink: bool = False) -> PathLike:
+        """Check a checkout path already normalized by :meth:`_to_relative_path`."""
+        return self._checked_abspath(self.repo.working_tree_dir, relative_path, allow_final_symlink)
+
+    @classmethod
+    def _checked_abspath(
+        cls, root: Union[PathLike, None], relative_path: PathLike, allow_final_symlink: bool = False
+    ) -> str:
+        """Reject symlinks below a trusted root before accessing submodule paths."""
+        if root is None:
+            raise NotADirectoryError("Submodules require a working tree")
+        path = os.fspath(root)
+        components = to_native_path_linux(relative_path).split("/")
+        for index, component in enumerate(components):
+            path = os.fspath(join_path_native(path, component))
+            if allow_final_symlink and index == len(components) - 1:
+                break
             if osp.islink(path):
-                raise ValueError("Submodule checkout path %r contains a symbolic link" % self.path)
+                raise ValueError("Submodule path %r contains a symbolic link" % relative_path)
         return path
 
+    @staticmethod
+    def _renames(source: PathLike, destination: PathLike) -> None:
+        os.makedirs(osp.dirname(destination), exist_ok=True)
+        os.rename(source, destination)
+        # Match renames() cleanup, but stop before directory symlinks: Windows
+        # rmdir() removes the link even when its target is nonempty.
+        parent = osp.dirname(source)
+        while parent and not osp.islink(parent):
+            try:
+                os.rmdir(parent)
+            except OSError:
+                break
+            parent = osp.dirname(parent)
+
     @classmethod
     def _write_git_file_and_module_config(cls, working_tree_dir: PathLike, module_abspath: PathLike) -> None:
         """Write a ``.git`` file containing a (preferably) relative path to the actual
@@ -449,14 +494,19 @@ def _write_git_file_and_module_config(cls, working_tree_dir: PathLike, module_ab
         :param module_abspath:
             Absolute path to the bare repository.
         """
+        # Git resolves metadata symlinks before interpreting core.worktree.
+        # Path.resolve() also handles Windows symlinks on Python 3.7.
+        module_abspath = str(Path(module_abspath).resolve())
+        working_tree_dir = str(Path(working_tree_dir).resolve())
         git_file = osp.join(working_tree_dir, ".git")
+        module_config = osp.join(module_abspath, "config")
         rela_path = osp.relpath(module_abspath, start=working_tree_dir)
         if sys.platform == "win32" and osp.isfile(git_file):
             os.remove(git_file)
         with open(git_file, "wb") as fp:
             fp.write(("gitdir: %s" % rela_path).encode(defenc))
 
-        with GitConfigParser(osp.join(module_abspath, "config"), read_only=False, merge_includes=False) as writer:
+        with GitConfigParser(module_config, read_only=False, merge_includes=False) as writer:
             writer.set_value(
                 "core",
                 "worktree",
@@ -567,6 +617,8 @@ def add(
             name,
             url="invalid-temporary",
         )
+        cls._checked_abspath(repo.working_tree_dir, cls.k_modules_file)
+        sm._checkout_abspath(path)
         if sm.exists():
             # Reretrieve submodule from tree.
             try:
@@ -661,7 +713,9 @@ def add(
 
         # We deliberately assume that our head matches our index!
         if mrepo:
-            sm.binsha = mrepo.head.commit.binsha
+            # Release cat-file processes before callers move the checkout on Windows.
+            with mrepo:
+                sm.binsha = mrepo.head.commit.binsha
         index.add([sm], write=True)
 
         return sm
@@ -1039,7 +1093,8 @@ def move(self, module_path: PathLike, configuration: bool = True, module: bool =
             self
 
         :raise ValueError:
-            If the module path existed and was not empty, or was a file.
+            If the module path existed and was not empty, was a file, or had a
+            symbolic link in an intermediate component.
 
         :note:
             Currently the method is not atomic, and it could leave the repository in an
@@ -1057,7 +1112,11 @@ def move(self, module_path: PathLike, configuration: bool = True, module: bool =
             return self
         # END handle no change
 
-        module_checkout_abspath = join_path_native(str(self.repo.working_tree_dir), module_checkout_path)
+        if configuration:
+            self._checked_abspath(self.repo.working_tree_dir, self.k_modules_file)
+        # Validate the source before removing the destination.
+        cur_path = self.abspath
+        module_checkout_abspath = self._checkout_abspath(module_checkout_path, allow_final_symlink=True)
         if osp.isfile(module_checkout_abspath):
             raise ValueError("Cannot move repository onto a file: %s" % module_checkout_abspath)
         # END handle target files
@@ -1089,10 +1148,9 @@ def move(self, module_path: PathLike, configuration: bool = True, module: bool =
         # END handle module
 
         # Move the module into place if possible.
-        cur_path = self.abspath
         renamed_module = False
         if module and osp.exists(cur_path):
-            os.renames(cur_path, module_checkout_abspath)
+            self._renames(cur_path, module_checkout_abspath)
             renamed_module = True
 
             if osp.isfile(osp.join(module_checkout_abspath, ".git")):
@@ -1123,7 +1181,7 @@ def move(self, module_path: PathLike, configuration: bool = True, module: bool =
             # END handle configuration flag
         except Exception:
             if renamed_module:
-                os.renames(module_checkout_abspath, cur_path)
+                self._renames(module_checkout_abspath, cur_path)
             # END undo module renaming
             raise
         # END handle undo rename
@@ -1180,6 +1238,12 @@ def remove(
             Doesn't work atomically, as failure to remove any part of the submodule will
             leave an inconsistent state.
 
+        :note:
+            Metadata-directory aliases under ``.git/modules`` are retained. A link
+            directly to the deleted repository becomes dangling; adding or initializing
+            the submodule again recreates its target. Linked parent directories remain
+            available to sibling submodules.
+
         :raise git.exc.InvalidGitRepositoryError:
             Thrown if the repository cannot be deleted.
 
@@ -1191,6 +1255,8 @@ def remove(
         # END handle parameters
 
         self._validated_name(self.name)
+        if configuration:
+            self._checked_abspath(self.repo.working_tree_dir, self.k_modules_file)
         # Recursively remove children of this submodule.
         nc = 0
         for csm in self.children():
@@ -1209,7 +1275,7 @@ def remove(
         ################################
         if module and self.module_exists():
             mod = self.module()
-            git_dir = mod.git_dir
+            git_dir = str(Path(mod.git_dir).resolve())
             if force:
                 # Take the fast lane and just delete everything in our module path.
                 # TODO: If we run into permission problems, we have a highly
@@ -1450,6 +1516,9 @@ def rename(self, new_name: str) -> "Submodule":
 
         self._validated_name(self.name)
         self._validated_name(new_name)
+        destination_module_abspath = self._module_abspath(self.repo, self.path, new_name)
+        mod = self.module()
+        self._checked_abspath(self.repo.working_tree_dir, self.k_modules_file)
 
         # .git/config
         with self.repo.config_writer() as pw:
@@ -1466,17 +1535,15 @@ def rename(self, new_name: str) -> "Submodule":
         self._name = new_name
 
         # .git/modules
-        mod = self.module()
         if mod.has_separate_working_tree():
-            destination_module_abspath = self._module_abspath(self.repo, self.path, new_name)
             source_dir = mod.git_dir
             # Let's be sure the submodule name is not so obviously tied to a directory.
             if str(destination_module_abspath).startswith(str(mod.git_dir)):
                 tmp_dir = self._module_abspath(self.repo, self.path, str(uuid.uuid4()))
-                os.renames(source_dir, tmp_dir)
+                self._renames(source_dir, tmp_dir)
                 source_dir = tmp_dir
             # END handle self-containment
-            os.renames(source_dir, destination_module_abspath)
+            self._renames(source_dir, destination_module_abspath)
             if mod.working_tree_dir:
                 self._write_git_file_and_module_config(mod.working_tree_dir, destination_module_abspath)
         # END move separate git repository

test/test_remote.py

@@ -1090,14 +1090,16 @@ def test_fetch_unsafe_branch_name(self, rw_repo, remote_repo):
 class TestTimeouts(TestBase):
     @with_rw_repo("HEAD", bare=False)
     def test_timeout_funcs(self, repo):
+        # Maintenance may outlive a timed-out fetch and race with fixture cleanup.
+        with repo.config_writer() as config:
+            config.set_value("maintenance", "auto", False)
+            config.set_value("gc", "auto", 0)  # Older Git versions use auto-gc.
+
         # Force error code to prevent a race condition if the python thread is slow.
-        default = Git.AutoInterrupt._status_code_if_terminate
-        Git.AutoInterrupt._status_code_if_terminate = -15
-        for function in ["pull", "fetch"]:  # Can't get push to time out.
-            f = getattr(repo.remotes.origin, function)
-            assert f is not None  # Make sure these functions exist.
-            _ = f()  # Make sure the function runs.
-            with pytest.raises(GitCommandError, match="kill_after_timeout=0 s"):
-                f(kill_after_timeout=0)
-
-        Git.AutoInterrupt._status_code_if_terminate = default
+        with mock.patch.object(Git.AutoInterrupt, "_status_code_if_terminate", -15):
+            for function in ["pull", "fetch"]:  # Can't get push to time out.
+                f = getattr(repo.remotes.origin, function)
+                assert f is not None  # Make sure these functions exist.
+                _ = f()  # Make sure the function runs.
+                with pytest.raises(GitCommandError, match="kill_after_timeout=0 s"):
+                    f(kill_after_timeout=0)