diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 45115f60c..9471680e8 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -2,7 +2,7 @@ exclude: ^(?:gitdb|smmap)/ repos: - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.16.5 + rev: v0.16.9 hooks: - id: ruff-check args: ["--fix"] diff --git a/doc/source/changes.rst b/doc/source/changes.rst index 2f537c99d..8447228db 100644 --- a/doc/source/changes.rst +++ b/doc/source/changes.rst @@ -2,6 +2,19 @@ Changelog ========= +3.2.1 +===== + +Security fixes for + +* https://github.com/gitpython-developers/GitPython/security/advisories/GHSA-83vg-56qc-22m7 + +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.2.1 + 3.2.0 ===== diff --git a/git/index/base.py b/git/index/base.py index c9d99840d..4ecd0c5b1 100644 --- a/git/index/base.py +++ b/git/index/base.py @@ -1303,7 +1303,7 @@ def _flush_stdin_and_wait(self, proc: "Popen[bytes]", ignore_stdout: bool = Fals @default_index def checkout( self, - paths: Union[None, Iterable[PathLike]] = None, + paths: Union[None, PathLike, Iterable[PathLike]] = None, force: bool = False, fprogress: Callable = lambda *args: None, allow_unsafe_options: bool = False, @@ -1442,7 +1442,7 @@ def handle_stderr(proc: "Popen[bytes]", iter_checked_out_files: Iterable[PathLik handle_stderr(proc, rval_iter) return rval_iter else: - if isinstance(paths, str): + if isinstance(paths, (str, os.PathLike)): paths = [paths] # Make sure we have our entries loaded before we start checkout_index, which diff --git a/git/objects/commit.py b/git/objects/commit.py index 771f87976..307b8c59b 100644 --- a/git/objects/commit.py +++ b/git/objects/commit.py @@ -885,7 +885,7 @@ def _deserialize(self, stream: BytesIO) -> "Commit": if buf[0:10] == b"encoding ": self.encoding = buf[buf.find(b" ") + 1 :].decode(self.encoding, "ignore") elif buf[0:7] == b"gpgsig ": - sig = buf[buf.find(b" ") + 1 :] + b"\n" + sig_lines = [buf[buf.find(b" ") + 1 :] + b"\n"] is_next_header = False while True: sigbuf = readline() @@ -895,9 +895,9 @@ def _deserialize(self, stream: BytesIO) -> "Commit": buf = sigbuf.strip() is_next_header = True break - sig += sigbuf[1:] + sig_lines.append(sigbuf[1:]) # END read all signature - self.gpgsig = sig.rstrip(b"\n").decode(self.encoding, "ignore") + self.gpgsig = b"".join(sig_lines).rstrip(b"\n").decode(self.encoding, "ignore") if is_next_header: continue buf = readline().strip() diff --git a/git/objects/submodule/base.py b/git/objects/submodule/base.py index 15fe877e6..74fbd39fe 100644 --- a/git/objects/submodule/base.py +++ b/git/objects/submodule/base.py @@ -8,6 +8,7 @@ import ntpath import os import os.path as osp +import re import shlex import stat import sys @@ -47,6 +48,7 @@ IterableList, RemoteProgress, _to_relative_path, + _validate_repo_path, join_path_native, rmtree, to_native_path_linux, @@ -305,20 +307,55 @@ def _config_parser_constrained(self, read_only: bool) -> SectionConstraint: def _validated_name(cls, name: str) -> str: if ( not name + or "\0" in name or name.startswith(("/", "\\")) or ntpath.splitdrive(name)[0] or ".." in name.replace("\\", "/").split("/") ): raise ValueError("Invalid submodule name %r" % name) + cls._validate_windows_path(name) return name + @staticmethod + def _validate_windows_path(path: PathLike) -> None: + """Apply Git for Windows' filename checks before creating directories.""" + if sys.platform == "win32": + for component in ntpath.splitdrive(os.fspath(path))[1].replace("\\", "/").split("/"): + if component in (".", ".."): + continue + stem = component.split(".", 1)[0].rstrip(" ").upper() + if ( + component.endswith((" ", ".")) + or any(ord(char) < 32 or char in '<>:"|?*' for char in component) + or re.fullmatch(r"CON(?:IN\$|OUT\$)?|PRN|AUX|NUL|COM[1-9]|LPT[1-9]", stem) + ): + raise ValueError("Invalid submodule path on Windows: %r" % path) + @classmethod - def _module_abspath(cls, parent_repo: "Repo", path: PathLike, name: str) -> PathLike: + def _module_abspath( + cls, parent_repo: "Repo", path: PathLike, name: str, *, moving_from: Union[PathLike, None] = None + ) -> PathLike: + """Reject nested Git directories, allowing the source of a pending rename.""" + from git.repo.fun import is_git_dir + name = cls._validated_name(name) if cls._need_gitfile_submodules(parent_repo.git): + directory = osp.join(parent_repo.git_dir, "modules") + for component in to_native_path_linux(name).split("/")[:-1]: + directory = osp.join(directory, component) + if is_git_dir(directory) and ( + moving_from is None or Path(directory).resolve() != Path(moving_from).resolve() + ): + raise ValueError( + "Submodule metadata for %r is inside another Git directory: %r" % (name, directory) + ) return osp.join(parent_repo.git_dir, "modules", name) if parent_repo.working_tree_dir: - return cls._checked_abspath(parent_repo.working_tree_dir, cls._to_relative_path(parent_repo, path)) + return cls._checked_abspath( + parent_repo.working_tree_dir, + cls._to_relative_path(parent_repo, path), + git_dirs=(parent_repo.git_dir, parent_repo.common_dir), + ) raise NotADirectoryError() @classmethod @@ -360,7 +397,9 @@ def _clone_repo( 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_checkout_path = cls._checked_abspath( + repo.working_tree_dir, path, git_dirs=(repo.git_dir, repo.common_dir) + ) module_abspath = cls._module_abspath(repo, path, name) if cls._need_gitfile_submodules(repo.git): if not allow_unsafe_options: @@ -402,6 +441,16 @@ def _clone_repo( **kwargs, ) if cls._need_gitfile_submodules(repo.git): + # A concurrent clone may have turned a leading directory into a repository. + try: + cls._module_abspath(repo, path, name) + except ValueError: + clone.close() + try: + os.remove(osp.join(clone.git_dir, "HEAD")) + except FileNotFoundError: + pass + raise cls._write_git_file_and_module_config(module_checkout_path, module_abspath) return clone @@ -411,8 +460,9 @@ def _to_relative_path(cls, parent_repo: "Repo", path: PathLike) -> PathLike: """:return: A path guaranteed to be relative to the given parent repository :raise ValueError: - If path is not contained in the parent repository's working tree. + If path is outside the working tree or is unsafe as a submodule checkout. """ + cls._validate_windows_path(path) if parent_repo.working_tree_dir: path = _to_relative_path(parent_repo.working_tree_dir, path) else: @@ -422,6 +472,7 @@ def _to_relative_path(cls, parent_repo: "Repo", path: PathLike) -> PathLike: if not path or path == ".": raise ValueError("Submodule checkout path must not be the repository root") + _validate_repo_path(path) return path @property @@ -433,22 +484,35 @@ def abspath(self) -> PathLike: 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) + return self._checked_abspath( + self.repo.working_tree_dir, + relative_path, + allow_final_symlink, + git_dirs=(self.repo.git_dir, self.repo.common_dir), + ) @classmethod def _checked_abspath( - cls, root: Union[PathLike, None], relative_path: PathLike, allow_final_symlink: bool = False + cls, + root: Union[PathLike, None], + relative_path: PathLike, + allow_final_symlink: bool = False, + *, + git_dirs: Sequence[PathLike] = (), ) -> str: - """Reject symlinks below a trusted root before accessing submodule paths.""" + """Reject symlinks and checkout aliases of Git directories below a trusted root.""" if root is None: raise NotADirectoryError("Submodules require a working tree") path = os.fspath(root) + metadata_dirs = set(git_dirs) 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 metadata_dirs and osp.exists(path) and any(osp.samefile(path, directory) for directory in metadata_dirs): + raise ValueError("Submodule checkout path aliases Git metadata: %r" % relative_path) if osp.islink(path): + if allow_final_symlink and index == len(components) - 1: + break raise ValueError("Submodule path %r contains a symbolic link" % relative_path) return path @@ -1122,6 +1186,11 @@ def move(self, module_path: PathLike, configuration: bool = True, module: bool = self._checked_abspath(self.repo.working_tree_dir, self.k_modules_file) # Validate the source before removing the destination. cur_path = self.abspath + module_abspath = self._module_abspath(self.repo, self.path, self.name) + if self.path == self.name: + self._module_abspath( + self.repo, module_checkout_path, os.fspath(module_checkout_path), moving_from=module_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) @@ -1160,7 +1229,6 @@ def move(self, module_path: PathLike, configuration: bool = True, module: bool = renamed_module = True if osp.isfile(osp.join(module_checkout_abspath, ".git")): - module_abspath = self._module_abspath(self.repo, self.path, self.name) self._write_git_file_and_module_config(module_checkout_abspath, module_abspath) # END handle git file rewrite # END move physical module @@ -1522,8 +1590,8 @@ 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() + destination_module_abspath = self._module_abspath(self.repo, self.path, new_name, moving_from=mod.git_dir) self._checked_abspath(self.repo.working_tree_dir, self.k_modules_file) # .git/config @@ -1573,6 +1641,7 @@ def module(self) -> "Repo": """ self._validated_name(self.name) module_checkout_abspath = self.abspath + self._module_abspath(self.repo, self.path, self.name) try: repo = git.Repo(module_checkout_abspath) if repo != self.repo: diff --git a/git/objects/util.py b/git/objects/util.py index 64cdf71fd..1fa3b9df0 100644 --- a/git/objects/util.py +++ b/git/objects/util.py @@ -100,9 +100,11 @@ def mode_str_to_int(modestr: Union[bytes, str]) -> int: module regarding the rwx permissions for user, group and other, special flags and file system flags, such as whether it is a symlink. """ + if isinstance(modestr, bytes): + # A byte is an int. int(b"4"[0]) is 52, so b"100644" became 0o6767524. + modestr = modestr.decode("ascii") mode = 0 for iteration, char in enumerate(reversed(modestr[-6:])): - char = cast(Union[str, int], char) mode += int(char) << iteration * 3 # END for each char return mode diff --git a/git/util.py b/git/util.py index 6f1f38401..75677b7d3 100644 --- a/git/util.py +++ b/git/util.py @@ -658,10 +658,11 @@ def remove_password_if_present(cmdline: Sequence[str]) -> List[str]: if url.password is None and url.username is None: continue - if url.password is not None: - url = url._replace(netloc=url.netloc.replace(url.password, "*****")) - if url.username is not None: - url = url._replace(netloc=url.netloc.replace(url.username, "*****")) + # Match urllib.parse's userinfo boundary. Keeping the raw hostinfo + # preserves hostname case, IPv6 brackets, and port formatting. + _, _, hostinfo = url.netloc.rpartition("@") + redacted = "*****:*****" if url.password is not None else "*****" + url = url._replace(netloc=f"{redacted}@{hostinfo}") new_cmdline[index] = urlunsplit(url) except ValueError: # This is not a valid URL. @@ -1166,15 +1167,56 @@ def _obtain_lock_or_raise(self) -> None: if self._has_lock(): return lock_file = self._lock_file_path() - if osp.isfile(lock_file): + # Create the lock in one step. Checking for it first would allow another + # holder to create it between the check and the open. + try: + if sys.platform == "win32": + if "\0" in lock_file: + raise ValueError("embedded null character") + + import ctypes + from ctypes import wintypes + + # Unlike POSIX, Windows follows dangling symlinks even with O_EXCL. + # Open the reparse point itself so an existing link is rejected. + # Call the Unicode API directly: older _winapi.CreateFile wrappers + # use the ANSI API and can create a lock under the wrong filename. + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + create_file = kernel32.CreateFileW + create_file.argtypes = ( + wintypes.LPCWSTR, + wintypes.DWORD, + wintypes.DWORD, + wintypes.LPVOID, + wintypes.DWORD, + wintypes.DWORD, + wintypes.HANDLE, + ) + create_file.restype = wintypes.HANDLE + close_handle = kernel32.CloseHandle + close_handle.argtypes = (wintypes.HANDLE,) + close_handle.restype = wintypes.BOOL + handle = create_file( + lock_file, + 0x40000000, # GENERIC_WRITE + 0, + None, + 1, # CREATE_NEW + 0x00200000, # FILE_FLAG_OPEN_REPARSE_POINT + None, + ) + if handle == wintypes.HANDLE(-1).value: + raise ctypes.WinError(ctypes.get_last_error()) + if not close_handle(handle): + raise ctypes.WinError(ctypes.get_last_error()) + else: + fd = os.open(lock_file, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + os.close(fd) + except FileExistsError as e: raise OSError( "Lock for file %r did already exist, delete %r in case the lock is illegal" % (self._file_path, lock_file) - ) - - try: - with open(lock_file, mode="w"): - pass + ) from e except OSError as e: raise OSError(str(e)) from e diff --git a/pyproject.toml b/pyproject.toml index 4b8258fcb..ce7ed9cac 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,6 +38,7 @@ dynamic = ["license", "version", "dependencies", "optional-dependencies"] [project.urls] "Source code" = "https://github.com/gitpython-developers/GitPython" "Documentation" = "https://gitpython.readthedocs.io/en/stable/" +"Changelog" = "https://github.com/gitpython-developers/GitPython/blob/main/doc/source/changes.rst" [dependency-groups] doc = [ diff --git a/test/test_index.py b/test/test_index.py index 8c33fe29d..30cb54565 100644 --- a/test/test_index.py +++ b/test/test_index.py @@ -1731,6 +1731,32 @@ def test_index_file_v3_with_git_command(self, tmp_dir): assert "A file2.txt" in status_lines +class TestIndexCheckout: + @pytest.mark.parametrize("path_type", [str, Path, PathLikeMock]) + @pytest.mark.parametrize("absolute", [False, True]) + @pytest.mark.parametrize("directory", [False, True]) + @pytest.mark.parametrize("container", ["single", "list", "iterator"]) + def test_checkout_pathlike(self, tmp_path, path_type, absolute, directory, container): + with Repo.init(tmp_path) as repo: + nested = tmp_path / "nested" + nested.mkdir() + files = {"nested/first": b"first", "nested/second": b"second", "outside": b"outside"} + for name, data in files.items(): + (tmp_path / name).write_bytes(data) + repo.index.add(list(files)) + + path_name = "nested" if directory else "nested/first" + path = path_type(str(tmp_path / path_name) if absolute else path_name) + paths = path if container == "single" else [path] if container == "list" else iter([path]) + expected = {"nested/first", "nested/second"} if directory else {"nested/first"} + for name in expected: + (tmp_path / name).unlink() + + assert set(repo.index.checkout(paths)) == expected + for name, data in files.items(): + assert (tmp_path / name).read_bytes() == data + + class TestIndexUtils: @pytest.mark.parametrize("file_path_type", [str, Path]) def test_temporary_file_swap(self, tmp_path, file_path_type): diff --git a/test/test_submodule.py b/test/test_submodule.py index b1f4f5156..45b285c55 100644 --- a/test/test_submodule.py +++ b/test/test_submodule.py @@ -86,6 +86,7 @@ def test_submodule_update_preserves_literal_name(tmp_path, monkeypatch, caplog, def movable_submodule(tmp_path): """Create a committed local submodule whose logical name stays fixed when moved.""" with git.Repo.init(tmp_path / "source") as source, git.Repo.init(tmp_path / "parent") as parent: + source.git.symbolic_ref("HEAD", "refs/heads/master") (tmp_path / "source" / "file").write_text("content", encoding="utf-8") source.index.add(["file"]) source.index.commit("Create source") @@ -111,6 +112,207 @@ def _move_snapshot(submodule): ) +@pytest.mark.parametrize( + "path", + [ + ".git/child", + ".GiT/child", + "nested/.git/child", + "git~1/child", + "GIT~1 . /child", + ".git. /child", + ".git:stream/child", + ".git::$INDEX_ALLOCATION/child", + "nested\\.git\\child", + "C:relative", + "nul\0name", + ] + + [ + f".g{chr(codepoint)}it/child" + for codepoint in (*range(0x200C, 0x2010), *range(0x202A, 0x202F), *range(0x206A, 0x2070), 0xFEFF) + ], +) +@pytest.mark.parametrize("operation", ["add", "clone", "update", "move", "move-module", "move-config"]) +def test_submodule_rejects_unsafe_checkout_before_mutation(movable_submodule, tmp_path, path, operation): + sm = movable_submodule + root = Path(sm.repo.working_tree_dir) + before = _move_snapshot(sm) + paths = set(root.rglob("*")) + # Check the portable metadata aliases against Git's index validation as well. + if operation == "clone" and path not in ("C:relative", "nul\0name"): + with sm.repo.git.custom_environment(GIT_INDEX_FILE=str(tmp_path / "validation-index")): + with _patch_git_config("core.protectHFS", "true"), _patch_git_config("core.protectNTFS", "true"): + with pytest.raises(GitCommandError): + sm.repo.git.update_index("--add", "--cacheinfo", f"160000,{sm.hexsha},{path}") + with mock.patch.object(git.Repo, "clone_from", side_effect=AssertionError("clone attempted")): + with pytest.raises(ValueError): + if operation == "add": + sm.repo.create_submodule("new", path, sm.url) + elif operation == "clone": + Submodule._clone_repo(sm.repo, sm.url, path, "new") + elif operation == "update": + Submodule(sm.repo, sm.binsha, name="new", path=path, url=sm.url).update(init=True) + else: + sm.move(path, configuration=operation != "move-module", module=operation != "move-config") + assert _move_snapshot(sm) == before + assert set(root.rglob("*")) == paths + + +def test_add_rejects_metadata_checkout_without_writing_files(movable_submodule): + sm = movable_submodule + before = _move_snapshot(sm) + with pytest.raises(ValueError, match="Git metadata"): + sm.repo.create_submodule("new", ".git/new", sm.url) + assert not Path(sm.repo.git_dir, "new").exists() + assert not Path(sm.repo.git_dir, "modules/new").exists() + assert _move_snapshot(sm) == before + + +@pytest.mark.parametrize("absolute_path", [False, True]) +@pytest.mark.parametrize("metadata_name", ["metadata", "MeTaDaTa"]) +@pytest.mark.parametrize("operation", ["add", "clone", "update", "move"]) +def test_submodule_rejects_checkout_in_separate_metadata( + movable_submodule, tmp_path, absolute_path, metadata_name, operation +): + root = tmp_path / "separate" + with git.Repo.init(root, separate_git_dir=str(root / "metadata"), allow_unsafe_options=True) as parent: + if not (root / metadata_name).is_dir(): + pytest.skip("Requires a case-insensitive filesystem") + sm = parent.create_submodule("module", "module", movable_submodule.url) + parent.index.commit("Add submodule") + before = _move_snapshot(sm) + paths = set(root.rglob("*")) + path = root / metadata_name / "new" if absolute_path else f"{metadata_name}/new" + with pytest.raises(ValueError, match="Git metadata"): + if operation == "add": + parent.create_submodule("new", path, sm.url) + elif operation == "clone": + Submodule._clone_repo(parent, sm.url, path, "new") + elif operation == "update": + Submodule(parent, sm.binsha, name="new", path=path, url=sm.url).update(init=True) + else: + sm.move(path) + assert _move_snapshot(sm) == before + assert set(root.rglob("*")) == paths + + +@pytest.mark.parametrize("operation", ["add", "clone", "update", "rename", "move"]) +def test_submodule_rejects_nested_metadata_before_mutation(movable_submodule, operation): + sm = movable_submodule + other = sm.repo.create_submodule("other", "other", sm.url) + other.module().close() + root = Path(sm.repo.working_tree_dir) + before = _move_snapshot(sm), _move_snapshot(other) + paths = set(root.rglob("*")) + name = f"{sm.name}/child" + with pytest.raises(ValueError, match="inside.*Git directory"): + if operation == "add": + sm.repo.create_submodule(name, "new", sm.url) + elif operation == "clone": + Submodule._clone_repo(sm.repo, sm.url, "new", name) + elif operation == "update": + Submodule(sm.repo, sm.binsha, name=name, path="new", url=sm.url).update(init=True) + elif operation == "rename": + other.rename(name) + else: + other.move(name) + assert (_move_snapshot(sm), _move_snapshot(other)) == before + assert set(root.rglob("*")) == paths + + +@pytest.mark.parametrize("state", ["retained", "checked-out"]) +def test_update_rejects_existing_nested_metadata(movable_submodule, state): + sm = movable_submodule + name = f"{sm.name}/child" + checkout = Path(sm.repo.working_tree_dir, "new") + with git.Repo.clone_from( + sm.url, checkout, separate_git_dir=str(Path(sm.repo.git_dir, "modules", name)), allow_unsafe_options=True + ): + pass + if state == "retained": + shutil.rmtree(checkout) + before = _move_snapshot(sm) + paths = set(Path(sm.repo.working_tree_dir).rglob("*")) + with pytest.raises(ValueError, match="inside.*Git directory"): + Submodule(sm.repo, sm.binsha, name=name, path="new", url=sm.url).update(init=True, no_fetch=True) + assert _move_snapshot(sm) == before + assert set(Path(sm.repo.working_tree_dir).rglob("*")) == paths + + +@pytest.mark.parametrize( + "path", + [ + "CON", + "con.txt", + "CONIN$", + "conout$.txt", + "AUX .txt", + "PRN", + "NUL", + "COM9", + "LPT1", + "name:stream", + "space ", + "period.", + "line\nbreak", + "star*", + 'quote"', + "question?", + "angle<", + "angle>", + "pipe|", + ], +) +def test_submodule_rejects_windows_destination_names_before_mutation(movable_submodule, path): + sm = movable_submodule + before = _move_snapshot(sm) + paths = set(Path(sm.repo.working_tree_dir).rglob("*")) + with mock.patch("git.objects.submodule.base.sys", SimpleNamespace(platform="win32")): + with mock.patch.object(git.Repo, "clone_from", side_effect=AssertionError("clone attempted")): + with pytest.raises(ValueError): + sm.repo.create_submodule("new", f"nested/{path}", sm.url) + with pytest.raises(ValueError): + sm.repo.create_submodule(f"nested/{path}", "new", sm.url) + assert _move_snapshot(sm) == before + assert set(Path(sm.repo.working_tree_dir).rglob("*")) == paths + + +@pytest.mark.parametrize("path", ["nested/space ", "nested/period."]) +def test_windows_destination_validation_precedes_normalization(tmp_path, path): + parent = SimpleNamespace(working_tree_dir=str(tmp_path)) + # Windows' GetFullPathName removes trailing spaces and periods. + with mock.patch("git.objects.submodule.base._to_relative_path", return_value=path.rstrip(" .")): + with mock.patch("git.objects.submodule.base.sys", SimpleNamespace(platform="win32")): + with pytest.raises(ValueError, match="Invalid submodule path on Windows"): + Submodule._to_relative_path(parent, path) + + +def test_submodule_can_relocate_its_own_metadata(movable_submodule): + sm = movable_submodule + sm.rename(f"{sm.name}/child") + assert Path(sm.abspath, "file").read_text() == "content" + with sm.module() as module: + assert Path(module.git_dir) == Path(sm.repo.git_dir, "modules", sm.name) + + +def test_clone_disables_metadata_that_becomes_nested(movable_submodule, monkeypatch): + sm = movable_submodule + clone_from = git.Repo.clone_from + ancestor = Path(sm.repo.git_dir, "modules/new") + + def clone_and_create_ancestor(*args, **kwargs): + clone = clone_from(*args, **kwargs) + with git.Repo.init(ancestor, bare=True): + pass + return clone + + monkeypatch.setattr(git.Repo, "clone_from", clone_and_create_ancestor) + with pytest.raises(ValueError, match="inside.*Git directory"): + Submodule._clone_repo(sm.repo, sm.url, "new", "new/child") + assert not (ancestor / "child/HEAD").exists() + assert (ancestor / "HEAD").is_file() + + @pytest.mark.parametrize("target_kind", ["relative", "absolute", "internal", "dangling"]) @pytest.mark.parametrize("configuration,module", [(True, True), (False, True), (True, False)]) @pytest.mark.parametrize("absolute_path", [False, True]) @@ -167,6 +369,28 @@ def test_move_normal_destination(movable_submodule, absolute_path): assert _move_snapshot(submodule) == before +@pytest.mark.parametrize("metadata_dir", ["git_dir", "common_dir"]) +def test_move_rejects_leaf_symlink_to_metadata(movable_submodule, tmp_path, metadata_dir): + root = tmp_path / "worktree" + movable_submodule.repo.git.worktree("add", "--detach", str(root)) + with git.Repo(root) as parent: + assert not osp.samefile(parent.git_dir, parent.common_dir) + submodule = parent.submodules[0] + submodule.update(init=True) + target = Path(getattr(parent, metadata_dir)) + destination = root / "destination" + destination.symlink_to(target, target_is_directory=True) + before = _move_snapshot(submodule) + + with pytest.raises(ValueError, match="Git metadata"): + submodule.move("destination", module=False) + + assert _move_snapshot(submodule) == before + assert destination.is_symlink() + assert destination.samefile(target) + assert Path(submodule.abspath, "file").read_text(encoding="utf-8") == "content" + + @pytest.mark.parametrize("kind", ["empty", "nonempty", "file", "dangling"]) def test_move_leaf_symlink_compatibility(movable_submodule, tmp_path, kind): """Preserve leaf-symlink replacement without modifying the external target. @@ -1410,6 +1634,7 @@ def test_update_rejects_parent_component_in_name(self, rwdir): invalid_names = ( "", + "nul\0name", "..", "../module", R"..\module", diff --git a/test/test_util.py b/test/test_util.py index c6e68cf51..ca8e006f3 100644 --- a/test/test_util.py +++ b/test/test_util.py @@ -12,6 +12,7 @@ import subprocess import sys import tempfile +import threading import time from unittest import SkipTest, mock @@ -22,6 +23,7 @@ from git.objects.util import ( altz_to_utctz_str, from_timestamp, + mode_str_to_int, parse_actor_and_date, parse_date, tzoffset, @@ -417,9 +419,11 @@ def test_it_should_dashify(self): self.assertEqual("this-is-my-argument", dashify("this_is_my_argument")) self.assertEqual("foo", dashify("foo")) - def test_lock_file(self): + @ddt.data("my-lock-file", "my-lock-file-\u0394", "\u0394/my-lock-file", "\U0001f680/my-lock-file") + def test_lock_file(self, filename): with tempfile.TemporaryDirectory() as tdir: - my_file = os.path.join(tdir, "my-lock-file") + my_file = os.path.join(tdir, filename) + os.makedirs(os.path.dirname(my_file), exist_ok=True) lock_file = LockFile(my_file) assert not lock_file._has_lock() # Release lock we don't have - fine. @@ -428,6 +432,7 @@ def test_lock_file(self): # Get lock. lock_file._obtain_lock_or_raise() assert lock_file._has_lock() + assert os.path.isfile(my_file + ".lock") # Concurrent access. other_lock_file = LockFile(my_file) @@ -436,6 +441,7 @@ def test_lock_file(self): lock_file._release_lock() assert not lock_file._has_lock() + assert not os.path.exists(my_file + ".lock") other_lock_file._obtain_lock_or_raise() self.assertRaises(IOError, lock_file._obtain_lock_or_raise) @@ -445,6 +451,67 @@ def test_lock_file(self): lock_file._obtain_lock_or_raise() lock_file._release_lock() + def test_lock_file_rejects_embedded_nul(self): + with tempfile.TemporaryDirectory() as tdir: + my_file = os.path.join(tdir, "my-lock-file") + lock_file = LockFile(my_file + "\0suffix") + self.assertRaises(ValueError, lock_file._obtain_lock_or_raise) + assert not lock_file._has_lock() + assert not os.path.exists(my_file) + + @ddt.data(False, True) + @requires_symlinks + def test_lock_file_does_not_follow_a_symlink(self, target_exists): + with tempfile.TemporaryDirectory() as tdir: + my_file = os.path.join(tdir, "my-lock-file") + outside = os.path.join(tdir, "outside-the-lock") + content = b"Do not modify the symlink target." + if target_exists: + with open(outside, "wb") as stream: + stream.write(content) + os.symlink(outside, my_file + ".lock") + + lock_file = LockFile(my_file) + self.assertRaises(IOError, lock_file._obtain_lock_or_raise) + assert not lock_file._has_lock() + lock_file._release_lock() + assert os.path.islink(my_file + ".lock") + if target_exists: + with open(outside, "rb") as stream: + self.assertEqual(stream.read(), content) + else: + assert not os.path.exists(outside) + + def test_lock_file_is_obtained_by_a_single_holder(self): + with tempfile.TemporaryDirectory() as tdir: + my_file = os.path.join(tdir, "my-lock-file") + racers = 8 + at_the_line = threading.Barrier(racers) + holders = [] + guard = threading.Lock() + + def obtain(): + lock_file = LockFile(my_file) + at_the_line.wait() + try: + lock_file._obtain_lock_or_raise() + except OSError: + return + with guard: + holders.append(lock_file) + + threads = [threading.Thread(target=obtain) for _ in range(racers)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + try: + self.assertEqual(1, len(holders)) + finally: + for lock_file in holders: + lock_file._release_lock() + def test_blocking_lock_file(self): with tempfile.TemporaryDirectory() as tdir: my_file = os.path.join(tdir, "my-lock-file") @@ -760,3 +827,38 @@ def test_remove_password_from_command_line(self): redacted_cmd_6 = remove_password_if_present(cmd_6) assert authorization not in " ".join(redacted_cmd_6) assert "http.extraHeader=Authorization: *****" in redacted_cmd_6 + + def test_remove_password_keeps_host_intact(self): + """Redaction must not touch the host, even when it contains the username.""" + redacted = remove_password_if_present(["git", "clone", "https://git@github.com/user/repo.git"]) + assert redacted == ["git", "clone", "https://*****@github.com/user/repo.git"] + + redacted = remove_password_if_present(["git", "clone", "ssh://git@github.com/u/r.git"]) + assert redacted == ["git", "clone", "ssh://*****@github.com/u/r.git"] + + def test_remove_empty_password_keeps_host_intact(self): + """An empty password must not expand into every position of the netloc.""" + redacted = remove_password_if_present(["git", "clone", "https://:@fakerepo.example.com/testrepo"]) + assert redacted == ["git", "clone", "https://*****:*****@fakerepo.example.com/testrepo"] + + @ddt.data( + ( + "https://user%40example.com:p%40ss@GitHub.COM:00443/repo@name?q=a@b#c@d", + "https://*****:*****@GitHub.COM:00443/repo@name?q=a@b#c@d", + ), + ("//user:pass@[2001:db8::1]:0080/repo", "//*****:*****@[2001:db8::1]:0080/repo"), + ("https://user:p@ss@example.com/repo", "https://*****:*****@example.com/repo"), + ("https://user:@example.com/repo", "https://*****:*****@example.com/repo"), + ("https://@example.com/repo", "https://*****@example.com/repo"), + ("https://example.com/repo@name?q=a@b#c@d", "https://example.com/repo@name?q=a@b#c@d"), + ) + @ddt.unpack + def test_remove_password_preserves_url_components(self, url, expected): + assert remove_password_if_present([url]) == [expected] + + +def test_mode_str_to_int_accepts_bytes(): + assert mode_str_to_int("100644") == 0o100644 + assert mode_str_to_int(b"100644") == 0o100644 + assert mode_str_to_int("644") == 0o644 + assert mode_str_to_int(b"120000") == 0o120000