diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 617111e1d..f3ab67035 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -7,7 +7,7 @@ repos: exclude: ^test/fixtures/ - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.15.8 + rev: v0.15.12 hooks: - id: ruff-check args: ["--fix"] diff --git a/VERSION b/VERSION index e1ace7c6e..0bc461141 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -3.1.47 +3.1.50 diff --git a/doc/source/changes.rst b/doc/source/changes.rst index 90b2e0739..b5152b3c5 100644 --- a/doc/source/changes.rst +++ b/doc/source/changes.rst @@ -2,6 +2,34 @@ Changelog ========= +3.1.50 +====== + +Save setting of configuration values, this time sections as well, as follow-up to 3.1.49. + +See the following for all changes. +https://github.com/gitpython-developers/GitPython/releases/tag/3.1.50 + +3.1.49 +====== + +Save setting of configuration values, +which could be used to inject other more configuration. + +Also more conforming `rev-parse` implementation. + +See the following for all changes. +https://github.com/gitpython-developers/GitPython/releases/tag/3.1.49 + +3.1.48 +====== + +Safe reference creation in the face of untrusted input. + +See the following for all changes. +https://github.com/gitpython-developers/GitPython/releases/tag/3.1.48 + + 3.1.47 ====== diff --git a/git/config.py b/git/config.py index c6eaf8f7b..82747eadd 100644 --- a/git/config.py +++ b/git/config.py @@ -72,6 +72,9 @@ See: https://git-scm.com/docs/git-config#_conditional_includes """ +UNSAFE_CONFIG_CHARS_RE = re.compile(r"[\r\n\x00]") +"""Characters that cannot be safely written in config names or values.""" + class MetaParserBuilder(abc.ABCMeta): # noqa: B024 """Utility class wrapping base-class methods into decorators that assure read-only @@ -778,6 +781,7 @@ def _assure_writable(self, method_name: str) -> None: def add_section(self, section: "cp._SectionName") -> None: """Assures added options will stay in order.""" + self._assure_config_name_safe(section, "section") return super().add_section(section) @property @@ -882,6 +886,30 @@ def _value_to_string(self, value: Union[str, bytes, int, float, bool]) -> str: return str(value) return force_text(value) + def _value_to_string_safe(self, value: Union[str, bytes, int, float, bool]) -> str: + value_str = self._value_to_string(value) + if UNSAFE_CONFIG_CHARS_RE.search(value_str): + raise ValueError("Git config values must not contain CR, LF, or NUL") + return value_str + + def _assure_config_name_safe(self, name: "cp._SectionName", label: str) -> None: + if isinstance(name, str) and UNSAFE_CONFIG_CHARS_RE.search(name): + raise ValueError("Git config %s names must not contain CR, LF, or NUL" % label) + + @needs_values + @set_dirty_and_flush_changes + def set( + self, + section: str, + option: str, + value: Union[str, bytes, int, float, bool, None] = None, + ) -> None: + self._assure_config_name_safe(section, "section") + self._assure_config_name_safe(option, "option") + if value is not None: + value = self._value_to_string_safe(value) + return super().set(section, option, value) + @needs_values @set_dirty_and_flush_changes def set_value(self, section: str, option: str, value: Union[str, bytes, int, float, bool]) -> "GitConfigParser": @@ -902,9 +930,12 @@ def set_value(self, section: str, option: str, value: Union[str, bytes, int, flo :return: This instance """ + self._assure_config_name_safe(section, "section") + self._assure_config_name_safe(option, "option") + value_str = self._value_to_string_safe(value) if not self.has_section(section): self.add_section(section) - self.set(section, option, self._value_to_string(value)) + super().set(section, option, value_str) return self @needs_values @@ -929,9 +960,12 @@ def add_value(self, section: str, option: str, value: Union[str, bytes, int, flo :return: This instance """ + self._assure_config_name_safe(section, "section") + self._assure_config_name_safe(option, "option") + value_str = self._value_to_string_safe(value) if not self.has_section(section): self.add_section(section) - self._sections[section].add(option, self._value_to_string(value)) + self._sections[section].add(option, value_str) return self def rename_section(self, section: str, new_name: str) -> "GitConfigParser": @@ -948,6 +982,7 @@ def rename_section(self, section: str, new_name: str) -> "GitConfigParser": """ if not self.has_section(section): raise ValueError("Source section '%s' doesn't exist" % section) + self._assure_config_name_safe(new_name, "section") if self.has_section(new_name): raise ValueError("Destination section '%s' already exists" % new_name) diff --git a/git/ext/gitdb b/git/ext/gitdb index 5c1b3036a..335c0f661 160000 --- a/git/ext/gitdb +++ b/git/ext/gitdb @@ -1 +1 @@ -Subproject commit 5c1b3036a6e34782e0ab6ce85e5ae64fe777fdbe +Subproject commit 335c0f66173eecdc7b2597c2b6c3d1fde795df30 diff --git a/git/refs/log.py b/git/refs/log.py index 4751cff99..fbbe66b22 100644 --- a/git/refs/log.py +++ b/git/refs/log.py @@ -4,7 +4,6 @@ __all__ = ["RefLog", "RefLogEntry"] from mmap import mmap -import os.path as osp import re import time as _time @@ -212,8 +211,11 @@ def path(cls, ref: "SymbolicReference") -> str: :param ref: :class:`~git.refs.symbolic.SymbolicReference` instance + + :raise ValueError: + If `ref.path` is invalid or escapes the repository's reflog directory. """ - return osp.join(ref.repo.git_dir, "logs", to_native_path(ref.path)) + return to_native_path(ref._get_validated_reflog_path(ref.repo, ref.path)) @classmethod def iter_entries(cls, stream: Union[str, "BytesIO", mmap]) -> Iterator[RefLogEntry]: diff --git a/git/refs/remote.py b/git/refs/remote.py index b4f4f7b36..e16ae70f8 100644 --- a/git/refs/remote.py +++ b/git/refs/remote.py @@ -58,17 +58,20 @@ def delete(cls, repo: "Repo", *refs: "RemoteReference", **kwargs: Any) -> None: `kwargs` are given for comparability with the base class method as we should not narrow the signature. """ + for ref in refs: + cls._check_ref_name_valid(ref.path) + repo.git.branch("-d", "-r", *refs) # The official deletion method will ignore remote symbolic refs - these are # generally ignored in the refs/ folder. We don't though and delete remainders # manually. for ref in refs: try: - os.remove(os.path.join(repo.common_dir, ref.path)) + os.remove(cls._get_validated_path(repo.common_dir, ref.path)) except OSError: pass try: - os.remove(os.path.join(repo.git_dir, ref.path)) + os.remove(cls._get_validated_path(repo.git_dir, ref.path)) except OSError: pass # END for each ref diff --git a/git/refs/symbolic.py b/git/refs/symbolic.py index 99af4f57c..020de5e13 100644 --- a/git/refs/symbolic.py +++ b/git/refs/symbolic.py @@ -110,6 +110,32 @@ def name(self) -> str: def abspath(self) -> PathLike: return join_path_native(_git_dir(self.repo, self.path), self.path) + @staticmethod + def _get_validated_path(base: PathLike, path: PathLike) -> str: + path = os.fspath(path) + base_path = os.path.realpath(os.fspath(base)) + abs_path = os.path.realpath(os.path.join(base_path, path)) + try: + common_path = os.path.commonpath([base_path, abs_path]) + except ValueError as e: + raise ValueError("Reference path %r escapes the repository" % path) from e + if os.path.normcase(common_path) != os.path.normcase(base_path): + raise ValueError("Reference path %r escapes the repository" % path) + return abs_path + + @classmethod + def _get_validated_ref_path(cls, repo: "Repo", path: PathLike) -> str: + """Return the absolute filesystem path for a ref after validating it.""" + cls._check_ref_name_valid(path) + ref_path = os.fspath(path) + return cls._get_validated_path(_git_dir(repo, ref_path), ref_path) + + @classmethod + def _get_validated_reflog_path(cls, repo: "Repo", path: PathLike) -> str: + """Return the absolute filesystem path for a reflog after validating it.""" + cls._check_ref_name_valid(path) + return cls._get_validated_path(os.path.join(repo.git_dir, "logs"), path) + @classmethod def _get_packed_refs_path(cls, repo: "Repo") -> str: return os.path.join(repo.common_dir, "packed-refs") @@ -485,7 +511,7 @@ def set_reference( # END handle non-existing # END retrieve old hexsha - fpath = self.abspath + fpath = self._get_validated_ref_path(self.repo, self.path) assure_directory_exists(fpath, is_file=True) lfd = LockedFD(fpath) @@ -632,7 +658,7 @@ def delete(cls, repo: "Repo", path: PathLike) -> None: Alternatively the symbolic reference to be deleted. """ full_ref_path = cls.to_full_path(path) - abs_path = os.path.join(repo.common_dir, full_ref_path) + abs_path = cls._get_validated_ref_path(repo, full_ref_path) if os.path.exists(abs_path): os.remove(abs_path) else: @@ -695,9 +721,8 @@ def _create( symbolic reference. Otherwise it will be resolved to the corresponding object and a detached symbolic reference will be created instead. """ - git_dir = _git_dir(repo, path) full_ref_path = cls.to_full_path(path) - abs_ref_path = os.path.join(git_dir, full_ref_path) + abs_ref_path = cls._get_validated_ref_path(repo, full_ref_path) # Figure out target data. target = reference @@ -789,8 +814,8 @@ def rename(self, new_path: PathLike, force: bool = False) -> "SymbolicReference" if self.path == new_path: return self - new_abs_path = os.path.join(_git_dir(self.repo, new_path), new_path) - cur_abs_path = os.path.join(_git_dir(self.repo, self.path), self.path) + new_abs_path = self._get_validated_ref_path(self.repo, new_path) + cur_abs_path = self._get_validated_ref_path(self.repo, self.path) if os.path.isfile(new_abs_path): if not force: # If they point to the same file, it's not an error. diff --git a/git/repo/base.py b/git/repo/base.py index 96c78df56..7579e326f 100644 --- a/git/repo/base.py +++ b/git/repo/base.py @@ -242,6 +242,28 @@ def __init__( # It's important to normalize the paths, as submodules will otherwise # initialize their repo instances with paths that depend on path-portions # that will not exist after being removed. It's just cleaner. + if ( + osp.isfile(osp.join(curpath, "gitdir")) + and osp.isfile(osp.join(curpath, "commondir")) + and osp.isfile(osp.join(curpath, "HEAD")) + ): + git_dir = curpath + + if "GIT_WORK_TREE" in os.environ: + self._working_tree_dir = os.getenv("GIT_WORK_TREE") + else: + # Linked worktree administrative directories store the path to the + # worktree's .git file in their gitdir file (without "gitdir: " prefix). + with open(osp.join(git_dir, "gitdir")) as fp: + worktree_gitfile = fp.read().strip() + + if not osp.isabs(worktree_gitfile): + worktree_gitfile = osp.normpath(osp.join(git_dir, worktree_gitfile)) + + self._working_tree_dir = osp.dirname(worktree_gitfile) + + break + if is_git_dir(curpath): git_dir = curpath # from man git-config : core.worktree diff --git a/git/repo/fun.py b/git/repo/fun.py index 3f00e60ea..66e7eba69 100644 --- a/git/repo/fun.py +++ b/git/repo/fun.py @@ -20,6 +20,7 @@ import os import os.path as osp from pathlib import Path +import re import stat from string import digits @@ -28,19 +29,21 @@ from git.cmd import Git from git.exc import WorkTreeRepositoryUnsupported from git.objects import Object +from git.objects.util import parse_date from git.refs import SymbolicReference from git.util import cygpath, bin_to_hex, hex_to_bin # Typing ---------------------------------------------------------------------- -from typing import Optional, TYPE_CHECKING, Union, cast, overload +from typing import Iterator, Optional, TYPE_CHECKING, Tuple, Union, cast, overload from git.types import AnyGitObject, Literal, PathLike if TYPE_CHECKING: from git.db import GitCmdObjectDB - from git.objects import Commit, TagObject + from git.objects import Commit from git.refs.reference import Reference + from git.refs.log import RefLog, RefLogEntry from git.refs.tag import Tag from .base import Repo @@ -139,6 +142,23 @@ def short_to_long(odb: "GitCmdObjectDB", hexsha: str) -> Optional[bytes]: # END exception handling +def _describe_to_long(repo: "Repo", name: str) -> Optional[bytes]: + """Resolve git-describe style names to the abbreviated object they contain.""" + match = re.match(r"^.+-\d+-g([0-9A-Fa-f]{4,40})(?:-dirty)?$", name) + if match is None: + match = re.match(r"^.+-g([0-9A-Fa-f]{4,40})(?:-dirty)?$", name) + if match is None: + match = re.match(r"^([0-9A-Fa-f]{4,40})-dirty$", name) + if match is None: + return None + # END handle match + + hexsha = match.group(1) + if len(hexsha) == 40: + return hexsha.encode("ascii") + return short_to_long(repo.odb, hexsha) + + @overload def name_to_object(repo: "Repo", name: str, return_ref: Literal[False] = ...) -> AnyGitObject: ... @@ -192,6 +212,10 @@ def name_to_object(repo: "Repo", name: str, return_ref: bool = False) -> Union[A # END for each base # END handle hexsha + if hexsha is None: + hexsha = _describe_to_long(repo, name) + # END handle describe output + # Didn't find any ref, this is an error. if return_ref: raise BadObject("Couldn't find reference named %r" % name) @@ -227,6 +251,361 @@ def to_commit(obj: Object) -> "Commit": return obj +def _object_from_hexsha(repo: "Repo", hexsha: str) -> AnyGitObject: + return Object.new_from_sha(repo, hex_to_bin(hexsha)) + + +def _current_reflog_ref(repo: "Repo") -> SymbolicReference: + try: + return repo.head.ref + except TypeError: + return repo.head + # END handle detached head + + +def _common_reflog_path(repo: "Repo", ref: SymbolicReference) -> Optional[str]: + if repo.common_dir == repo.git_dir: + return None + # END handle normal repository + return SymbolicReference._get_validated_path(osp.join(repo.common_dir, "logs"), ref.path) + + +def _ref_log(repo: "Repo", ref: SymbolicReference) -> "RefLog": + try: + return ref.log() + except FileNotFoundError: + common_path = _common_reflog_path(repo, ref) + if common_path and osp.isfile(common_path): + from git.refs.log import RefLog + + return RefLog.from_file(common_path) + # END handle linked-worktree branch logs + try: + if ref.path == repo.head.ref.path: + return repo.head.log() + # END handle linked-worktree current branch logs + except TypeError: + pass + # END handle detached head + raise + # END handle missing branch log + + +def _ref_log_entry(repo: "Repo", ref: SymbolicReference, index: int) -> "RefLogEntry": + try: + return ref.log_entry(index) + except FileNotFoundError: + common_path = _common_reflog_path(repo, ref) + if common_path and osp.isfile(common_path): + from git.refs.log import RefLog + + return RefLog.entry_at(common_path, index) + # END handle linked-worktree branch logs + try: + if ref.path == repo.head.ref.path: + return repo.head.log_entry(index) + # END handle linked-worktree current branch logs + except TypeError: + pass + # END handle detached head + raise + # END handle missing branch log + + +def _find_reflog_entry_by_date(repo: "Repo", ref: SymbolicReference, spec: str) -> str: + try: + timestamp, _offset = parse_date(spec) + except ValueError as e: + raise NotImplementedError("Support for additional @{...} modes not implemented") from e + # END handle unsupported dates + log = _ref_log(repo, ref) + if not log: + raise IndexError("Invalid revlog date: %s" % spec) + # END handle empty log + + for entry in reversed(log): + if entry.time[0] <= timestamp: + return entry.newhexsha + # END found candidate + # END for each entry + return log[0].newhexsha + + +def _previous_checked_out_branch(repo: "Repo", nth: int) -> AnyGitObject: + if nth <= 0: + raise ValueError("Invalid previous checkout selector: -%i" % nth) + # END handle invalid input + + seen = 0 + for entry in reversed(_ref_log(repo, repo.head)): + message = entry.message or "" + prefix = "checkout: moving from " + if not message.startswith(prefix): + continue + # END skip non-checkouts + + previous_branch = message[len(prefix) :].split(" to ", 1)[0] + seen += 1 + if seen == nth: + return name_to_object(repo, previous_branch) + # END found selector + # END for each entry + raise IndexError("Invalid previous checkout selector: -%i" % nth) + + +def _tracking_branch_object(repo: "Repo", ref: Optional[SymbolicReference]) -> AnyGitObject: + from git.refs.head import Head + + if ref is None: + try: + head = repo.active_branch + except TypeError as e: + raise BadName("@{upstream}") from e + elif isinstance(ref, Head): + head = ref + elif os.fspath(ref.path).startswith("refs/heads/"): + head = Head(repo, ref.path) + else: + raise BadName("%s@{upstream}" % ref.name) + # END handle head + + tracking_branch = head.tracking_branch() + if tracking_branch is None: + raise BadName("%s@{upstream}" % head.name) + # END handle missing upstream + return tracking_branch.commit + + +def _apply_reflog(repo: "Repo", ref: Optional[SymbolicReference], content: str) -> AnyGitObject: + if content.startswith("+"): + content = content[1:] + # END handle explicit positive sign + + if content.startswith("-"): + if ref is not None: + raise ValueError("Previous checkout selectors do not take an explicit ref") + if content == "-0": + raise ValueError("Negative zero is invalid in reflog selector") + # END handle invalid negative zero + try: + return _previous_checked_out_branch(repo, int(content[1:])) + except ValueError as e: + raise ValueError("Invalid previous checkout selector: %s" % content) from e + # END handle previous checkout branch + + content_lower = content.lower() + if content_lower in ("u", "upstream", "push"): + return _tracking_branch_object(repo, ref) + # END handle sibling branches + + ref = ref or _current_reflog_ref(repo) + try: + entry_no = int(content) + except ValueError: + hexsha = _find_reflog_entry_by_date(repo, ref, content) + else: + if entry_no >= 100000000: + hexsha = _find_reflog_entry_by_date(repo, ref, "%s +0000" % entry_no) + elif entry_no == 0: + return ref.commit + else: + try: + entry = _ref_log_entry(repo, ref, -(entry_no + 1)) + except IndexError as e: + raise IndexError("Invalid revlog index: %i" % entry_no) from e + # END handle index out of bound + hexsha = entry.newhexsha + # END handle offset or date-like timestamp + # END handle content + return _object_from_hexsha(repo, hexsha) + + +def _find_closing_brace(rev: str, start: int) -> int: + depth = 1 + escaped = False + for idx in range(start + 1, len(rev)): + char = rev[idx] + if escaped: + escaped = False + elif char == "\\": + escaped = True + elif char == "{": + depth += 1 + elif char == "}": + depth -= 1 + if depth == 0: + return idx + # END found end + # END handle char + # END for each char + raise ValueError("Missing closing brace to define type in %s" % rev) + + +def _parse_search(pattern: str) -> Tuple[str, bool]: + if not pattern: + raise ValueError("Revision search requires a pattern") + # END handle empty pattern + + if pattern.startswith("!-"): + return pattern[2:], True + if pattern.startswith("!!"): + return pattern[1:], False + if pattern.startswith("!"): + raise ValueError("Need one character after /!, typically -") + return pattern, False + + +def _unescape_braced_regex(pattern: str) -> str: + out = [] + idx = 0 + while idx < len(pattern): + char = pattern[idx] + if char == "\\" and idx + 1 < len(pattern): + next_char = pattern[idx + 1] + if next_char in "{}\\": + out.append(next_char) + else: + out.append(char) + out.append(next_char) + # END handle escaped char + idx += 2 + continue + # END handle backslash + out.append(char) + idx += 1 + # END for each char + return "".join(out) + + +def _find_commit_by_message( + repo: "Repo", rev: Optional[AnyGitObject], pattern: str, braced: bool = False +) -> AnyGitObject: + pattern, negated = _parse_search(_unescape_braced_regex(pattern) if braced else pattern) + try: + regex = re.compile(pattern) + except re.error as e: + raise ValueError("Invalid commit message regex %r" % pattern) from e + # END handle invalid regex + if rev is None: + commits = _all_ref_commits(repo) + else: + commits = _reachable_commits([to_commit(cast(Object, rev))]) + # END handle starting point + + for commit in commits: + message = commit.message + if isinstance(message, bytes): + message = message.decode(commit.encoding, "replace") + # END handle bytes message + matches = regex.search(message or "") is not None + if matches != negated: + return commit + # END found commit + # END for each commit + raise BadName("No commit found matching message pattern %r" % pattern) + + +def _all_ref_commits(repo: "Repo") -> Iterator["Commit"]: + starts = [] + for ref in repo.references: + try: + starts.append(to_commit(cast(Object, ref.object))) + except (BadName, ValueError): + pass + # END skip refs that do not point to commits + # END for each ref + try: + starts.append(repo.head.commit) + except ValueError: + pass + # END handle unborn head + return _reachable_commits(starts) + + +def _reachable_commits(starts: list["Commit"]) -> Iterator["Commit"]: + seen = set() + pending = starts[:] + while pending: + pending.sort(key=lambda commit: commit.committed_date, reverse=True) + commit = pending.pop(0) + if commit.binsha in seen: + continue + # END skip seen commit + seen.add(commit.binsha) + yield commit + pending.extend(commit.parents) + # END while commits remain + + +def _index_lookup(repo: "Repo", spec: str) -> AnyGitObject: + if not spec: + raise ValueError("':' must be followed by a path") + # END handle empty lookup + + stage = 0 + path = spec + if len(spec) >= 2 and spec[1] == ":" and spec[0] in "0123": + stage = int(spec[0]) + path = spec[2:] + # END handle stage + + try: + return repo.index.entries[(path, stage)].to_blob(repo) + except KeyError as e: + raise BadName("Path %r did not exist in the index at stage %i" % (path, stage)) from e + + +def _tree_lookup(obj: AnyGitObject, path: str) -> AnyGitObject: + if obj.type != "tree": + obj = to_commit(cast(Object, obj)).tree + # END get tree + if not path: + return obj + return obj[path] + + +def _peel(obj: AnyGitObject, output_type: str, repo: "Repo", rev: str) -> AnyGitObject: + if output_type.startswith("/"): + return _find_commit_by_message(repo, obj, output_type[1:], braced=True) + if output_type == "": + return deref_tag(obj) if obj.type == "tag" else obj + if output_type == "object": + return obj + if output_type == "commit": + return to_commit(cast(Object, obj)) + if output_type == "tree": + return to_commit(cast(Object, obj)).tree if obj.type != "tree" else obj + if output_type == "blob": + obj = deref_tag(obj) if obj.type == "tag" else obj + if obj.type == output_type: + return obj + # END handle matching type + raise ValueError("Could not accommodate requested object type %r, got %s" % (output_type, obj.type)) + if output_type == "tag": + if obj.type == output_type: + return obj + # END handle matching type + raise ValueError("Could not accommodate requested object type %r, got %s" % (output_type, obj.type)) + # END handle known types + raise ValueError("Invalid output type: %s ( in %s )" % (output_type, rev)) + + +def _first_rev_token(rev: str) -> Optional[int]: + for idx, char in enumerate(rev): + if char in "^~:": + return idx + if char == "@": + next_char = rev[idx + 1] if idx + 1 < len(rev) else None + if idx == 0 and next_char in (None, "^", "~", ":", "{"): + return idx + if next_char == "{": + return idx + # END handle reflog selector + # END handle at symbol + # END for each char + return None + + def rev_parse(repo: "Repo", rev: str) -> AnyGitObject: """Parse a revision string. Like :manpage:`git-rev-parse(1)`. @@ -253,137 +632,82 @@ def rev_parse(repo: "Repo", rev: str) -> AnyGitObject: :raise IndexError: If an invalid reflog index is specified. """ - # Are we in colon search mode? if rev.startswith(":/"): - # Colon search mode - raise NotImplementedError("commit by message search (regex)") - # END handle search + return _find_commit_by_message(repo, None, rev[2:]) + if rev.startswith(":"): + return _index_lookup(repo, rev[1:]) + # END handle top-level colon modes obj: Optional[AnyGitObject] = None ref = None - output_type = "commit" - start = 0 - parsed_to = 0 lr = len(rev) - while start < lr: - if rev[start] not in "^~:@": - start += 1 - continue - # END handle start + first_token = _first_rev_token(rev) + if first_token is None: + return name_to_object(repo, rev) + # END handle plain name + + if first_token == 0: + if rev[0] != "@": + raise ValueError("Revision specifier must start with an object name: %s" % rev) + # END handle invalid leading token + ref = _current_reflog_ref(repo) + obj = ref.commit + start = 0 if rev.startswith("@{") else 1 + else: + if rev[first_token] == "@": + ref = cast("Reference", name_to_object(repo, rev[:first_token], return_ref=True)) + obj = ref.commit + else: + obj = name_to_object(repo, rev[:first_token]) + # END handle anchor + start = first_token + # END initialize anchor + while start < lr: token = rev[start] - if obj is None: - # token is a rev name. - if start == 0: - ref = repo.head.ref - else: - if token == "@": - ref = cast("Reference", name_to_object(repo, rev[:start], return_ref=True)) - else: - obj = name_to_object(repo, rev[:start]) - # END handle token - # END handle refname - else: - if ref is not None: - obj = ref.commit - # END handle ref - # END initialize obj on first token + if token == "@": + if start + 1 >= lr or rev[start + 1] != "{": + raise ValueError("Invalid @ token in revision specifier: %s" % rev) + # END handle invalid @ + end = _find_closing_brace(rev, start + 1) + obj = _apply_reflog(repo, ref if first_token != 0 and start == first_token else None, rev[start + 2 : end]) + ref = None + start = end + 1 + continue + # END handle reflog - start += 1 + if token == ":": + return _tree_lookup(obj, rev[start + 1 :]) + # END handle path - # Try to parse {type}. - if start < lr and rev[start] == "{": - end = rev.find("}", start) - if end == -1: - raise ValueError("Missing closing brace to define type in %s" % rev) - output_type = rev[start + 1 : end] # Exclude brace. - - # Handle type. - if output_type == "commit": - obj = cast("TagObject", obj) - if obj and obj.type == "tag": - obj = deref_tag(obj) - else: - # Cannot do anything for non-tags. - pass - # END handle tag - elif output_type == "tree": - try: - obj = cast(AnyGitObject, obj) - obj = to_commit(obj).tree - except (AttributeError, ValueError): - pass # Error raised later. - # END exception handling - elif output_type in ("", "blob"): - obj = cast("TagObject", obj) - if obj and obj.type == "tag": - obj = deref_tag(obj) - else: - # Cannot do anything for non-tags. - pass - # END handle tag - elif token == "@": - # try single int - assert ref is not None, "Require Reference to access reflog" - revlog_index = None - try: - # Transform reversed index into the format of our revlog. - revlog_index = -(int(output_type) + 1) - except ValueError as e: - # TODO: Try to parse the other date options, using parse_date maybe. - raise NotImplementedError("Support for additional @{...} modes not implemented") from e - # END handle revlog index - - try: - entry = ref.log_entry(revlog_index) - except IndexError as e: - raise IndexError("Invalid revlog index: %i" % revlog_index) from e - # END handle index out of bound - - obj = Object.new_from_sha(repo, hex_to_bin(entry.newhexsha)) - - # Make it pass the following checks. - output_type = "" - else: - raise ValueError("Invalid output type: %s ( in %s )" % (output_type, rev)) - # END handle output type - - # Empty output types don't require any specific type, its just about - # dereferencing tags. - if output_type and obj and obj.type != output_type: - raise ValueError("Could not accommodate requested object type %r, got %s" % (output_type, obj.type)) - # END verify output type + start += 1 - start = end + 1 # Skip brace. - parsed_to = start + if token == "^" and start < lr and rev[start] == "{": + end = _find_closing_brace(rev, start) + obj = _peel(obj, rev[start + 1 : end], repo, rev) + ref = None + start = end + 1 continue # END parse type - # Try to parse a number. num = 0 - if token != ":": - found_digit = False - while start < lr: - if rev[start] in digits: - num = num * 10 + int(rev[start]) - start += 1 - found_digit = True - else: - break - # END handle number - # END number parse loop - - # No explicit number given, 1 is the default. It could be 0 though. - if not found_digit: - num = 1 - # END set default num - # END number parsing only if non-blob mode - - parsed_to = start - # Handle hierarchy walk. + found_digit = False + while start < lr: + if rev[start] in digits: + num = num * 10 + int(rev[start]) + start += 1 + found_digit = True + else: + break + # END handle number + # END number parse loop + + if not found_digit: + num = 1 + # END set default num + try: - obj = cast(AnyGitObject, obj) if token == "~": obj = to_commit(obj) for _ in range(num): @@ -391,15 +715,11 @@ def rev_parse(repo: "Repo", rev: str) -> AnyGitObject: # END for each history item to walk elif token == "^": obj = to_commit(obj) - # Must be n'th parent. - if num: + if num == 0: + pass + else: obj = obj.parents[num - 1] - elif token == ":": - if obj.type != "tree": - obj = obj.tree - # END get tree type - obj = obj[rev[start:]] - parsed_to = lr + # END handle parent else: raise ValueError("Invalid token: %r" % token) # END end handle tag @@ -410,16 +730,7 @@ def rev_parse(repo: "Repo", rev: str) -> AnyGitObject: # END exception handling # END parse loop - # Still no obj? It's probably a simple name. - if obj is None: - obj = name_to_object(repo, rev) - parsed_to = lr - # END handle simple name - if obj is None: raise ValueError("Revision specifier could not be parsed: %s" % rev) - if parsed_to != lr: - raise ValueError("Didn't consume complete rev spec %s, consumed part: %s" % (rev, rev[:parsed_to])) - return obj diff --git a/git/util.py b/git/util.py index c3ffdd62b..712fabe85 100644 --- a/git/util.py +++ b/git/util.py @@ -289,7 +289,7 @@ def join_path(a: PathLike, *p: PathLike) -> PathLike: if sys.platform == "win32": - def to_native_path_windows(path: PathLike) -> PathLike: + def to_native_path_windows(path: PathLike) -> str: path = os.fspath(path) return path.replace("/", "\\") diff --git a/test/test_config.py b/test/test_config.py index 11ea52d16..3ddaf0a4b 100644 --- a/test/test_config.py +++ b/test/test_config.py @@ -150,6 +150,70 @@ def test_config_value_with_trailing_new_line(self): git_config = GitConfigParser(config_file) git_config.read() # This should not throw an exception + @with_rw_directory + def test_set_value_rejects_config_injection(self, rw_dir): + config_path = osp.join(rw_dir, "config") + payload = "foo\n[core]\nhooksPath=/tmp/hooks" + + with GitConfigParser(config_path, read_only=False) as git_config: + with pytest.raises(ValueError, match="CR, LF, or NUL"): + git_config.set_value("user", "name", payload) + + with GitConfigParser(config_path, read_only=True) as git_config: + self.assertFalse(git_config.has_section("user")) + self.assertFalse(git_config.has_section("core")) + + @with_rw_directory + def test_set_value_rejects_unsafe_section_and_option_names(self, rw_dir): + config_path = osp.join(rw_dir, "config") + bad_keys = ("user]\n[core", "user]\r[core", "user]\x00[core") + + with GitConfigParser(config_path, read_only=False) as git_config: + git_config.add_section("user") + for bad_key in bad_keys: + with pytest.raises(ValueError, match="CR, LF, or NUL"): + git_config.add_section(bad_key) + with pytest.raises(ValueError, match="CR, LF, or NUL"): + git_config.set(bad_key, "hooksPath", "/tmp/hooks") + with pytest.raises(ValueError, match="CR, LF, or NUL"): + git_config.set("user", bad_key, "/tmp/hooks") + with pytest.raises(ValueError, match="CR, LF, or NUL"): + git_config.set_value(bad_key, "hooksPath", "/tmp/hooks") + with pytest.raises(ValueError, match="CR, LF, or NUL"): + git_config.set_value("user", bad_key, "/tmp/hooks") + with pytest.raises(ValueError, match="CR, LF, or NUL"): + git_config.add_value(bad_key, "hooksPath", "/tmp/hooks") + with pytest.raises(ValueError, match="CR, LF, or NUL"): + git_config.add_value("user", bad_key, "/tmp/hooks") + with pytest.raises(ValueError, match="CR, LF, or NUL"): + git_config.rename_section("user", bad_key) + + git_config.set_value("user", "name", "safe") + + with GitConfigParser(config_path, read_only=True) as git_config: + self.assertEqual(git_config.get_value("user", "name"), "safe") + self.assertFalse(git_config.has_section("core")) + + @with_rw_directory + def test_set_and_add_value_reject_unsafe_value_characters(self, rw_dir): + config_path = osp.join(rw_dir, "config") + bad_values = ("foo\rbar", "foo\nbar", "foo\x00bar", b"foo\nbar") + + with GitConfigParser(config_path, read_only=False) as git_config: + git_config.add_section("user") + for bad_value in bad_values: + with pytest.raises(ValueError, match="CR, LF, or NUL"): + git_config.set("user", "name", bad_value) + with pytest.raises(ValueError, match="CR, LF, or NUL"): + git_config.set_value("user", "name", bad_value) + with pytest.raises(ValueError, match="CR, LF, or NUL"): + git_config.add_value("user", "name", bad_value) + + git_config.set_value("user", "name", "safe") + + with GitConfigParser(config_path, read_only=True) as git_config: + self.assertEqual(git_config.get_value("user", "name"), "safe") + def test_base(self): path_repo = fixture_path("git_config") path_global = fixture_path("git_config_global") diff --git a/test/test_index.py b/test/test_index.py index 33490f907..f8280450a 100644 --- a/test/test_index.py +++ b/test/test_index.py @@ -172,6 +172,19 @@ def _decode(stdout): _win_bash_status = WinBashStatus.check() +def _windows_supports_symlinks(): + if sys.platform != "win32": + return False + + with tempfile.TemporaryDirectory(prefix="gitpython-symlink-check-") as temp_dir: + link_path = osp.join(temp_dir, "link") + try: + os.symlink("missing-target", link_path) + except (NotImplementedError, OSError): + return False + return S_ISLNK(os.lstat(link_path)[ST_MODE]) + + def _make_hook(git_dir, name, content, make_exec=True): """A helper to create a hook""" hp = hook_path(name, git_dir) @@ -553,9 +566,9 @@ def _count_existing(self, repo, files): # END num existing helper @pytest.mark.xfail( - sys.platform == "win32" and Git().config("core.symlinks") == "true", + sys.platform == "win32" and (Git().config("core.symlinks") == "true" or _windows_supports_symlinks()), reason="Assumes symlinks are not created on Windows and opens a symlink to a nonexistent target.", - raises=FileNotFoundError, + raises=(FileNotFoundError, GitCommandError), ) @with_rw_repo("0.1.6") def test_index_mutation(self, rw_repo): diff --git a/test/test_refs.py b/test/test_refs.py index 329515807..d77b34eba 100644 --- a/test/test_refs.py +++ b/test/test_refs.py @@ -3,6 +3,7 @@ # This module is part of GitPython and is released under the # 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/ +import contextlib from itertools import chain import os.path as osp from pathlib import Path @@ -18,6 +19,7 @@ RefLog, Reference, RemoteReference, + Repo, SymbolicReference, TagReference, ) @@ -29,6 +31,18 @@ class TestRefs(TestBase): + @contextlib.contextmanager + def _repo_with_initial_commit(self, base_dir): + repo_dir = base_dir / "repo" + repo = Repo.init(repo_dir) + (repo_dir / "file.txt").write_text("initial\n", encoding="utf-8") + repo.index.add(["file.txt"]) + repo.index.commit("initial") + try: + yield repo + finally: + repo.git.clear_cache() + def test_from_path(self): # Should be able to create any reference directly. for ref_type in (Reference, Head, TagReference, RemoteReference): @@ -648,6 +662,115 @@ def test_refs_outside_repo(self): ref_file_name = Path(ref_file.name).name self.assertRaises(BadName, self.rorepo.commit, f"../../{ref_file_name}") + def test_reference_create_rejects_path_traversal(self): + with tempfile.TemporaryDirectory() as tmp_dir: + base_dir = Path(tmp_dir) + with self._repo_with_initial_commit(base_dir) as repo: + outside_path = base_dir / "outside_write.txt" + + self.assertRaises(ValueError, Reference.create, repo, "../../../outside_write.txt", "HEAD") + assert not outside_path.exists() + + def test_symbolic_reference_create_rejects_path_traversal(self): + with tempfile.TemporaryDirectory() as tmp_dir: + base_dir = Path(tmp_dir) + with self._repo_with_initial_commit(base_dir) as repo: + outside_path = base_dir / "outside_write.txt" + + self.assertRaises(ValueError, SymbolicReference.create, repo, "../../outside_write.txt", "HEAD") + assert not outside_path.exists() + + def test_symbolic_reference_set_reference_rejects_path_traversal(self): + with tempfile.TemporaryDirectory() as tmp_dir: + base_dir = Path(tmp_dir) + with self._repo_with_initial_commit(base_dir) as repo: + outside_path = base_dir / "outside_write.txt" + + self.assertRaises(ValueError, SymbolicReference(repo, "../../outside_write.txt").set_reference, "HEAD") + assert not outside_path.exists() + + def test_symbolic_reference_rename_rejects_path_traversal(self): + with tempfile.TemporaryDirectory() as tmp_dir: + base_dir = Path(tmp_dir) + with self._repo_with_initial_commit(base_dir) as repo: + outside_path = base_dir / "outside_move.txt" + ref = SymbolicReference.create(repo, "SAFE_RENAME_SOURCE", "HEAD") + + self.assertRaises(ValueError, ref.rename, "../../outside_move.txt") + assert not outside_path.exists() + assert Path(ref.abspath).is_file() + + def test_symbolic_reference_delete_rejects_path_traversal(self): + with tempfile.TemporaryDirectory() as tmp_dir: + base_dir = Path(tmp_dir) + with self._repo_with_initial_commit(base_dir) as repo: + outside_path = base_dir / "outside_delete.txt" + outside_path.write_text("do not delete\n", encoding="utf-8") + + self.assertRaises(ValueError, SymbolicReference.delete, repo, "../../outside_delete.txt") + assert outside_path.read_text(encoding="utf-8") == "do not delete\n" + + def test_symbolic_reference_log_append_rejects_path_traversal(self): + with tempfile.TemporaryDirectory() as tmp_dir: + base_dir = Path(tmp_dir) + with self._repo_with_initial_commit(base_dir) as repo: + outside_path = base_dir / "outside_reflog.txt" + + ref = SymbolicReference(repo, "../../../outside_reflog.txt") + self.assertRaises( + ValueError, ref.log_append, Commit.NULL_BIN_SHA, "do not write", repo.head.commit.binsha + ) + assert not outside_path.exists() + + def test_symbolic_reference_set_reference_rejects_symlink_escape(self): + with tempfile.TemporaryDirectory() as tmp_dir: + base_dir = Path(tmp_dir) + with self._repo_with_initial_commit(base_dir) as repo: + outside_dir = base_dir / "outside_refs" + outside_dir.mkdir() + outside_path = outside_dir / "escaped" + + refs_heads_dir = Path(repo.common_dir) / "refs" / "heads" + refs_heads_dir.mkdir(parents=True, exist_ok=True) + symlink_path = refs_heads_dir / "link_out" + try: + symlink_path.symlink_to(outside_dir, target_is_directory=True) + except (OSError, NotImplementedError) as ex: + self.skipTest("symlinks unavailable on this platform: %s" % ex) + if osp.realpath(symlink_path / "escaped") == osp.abspath(symlink_path / "escaped"): + self.skipTest("realpath does not resolve directory symlinks on this platform") + + ref = SymbolicReference(repo, "refs/heads/link_out/escaped") + self.assertRaises(ValueError, ref.set_reference, "HEAD") + assert not outside_path.exists() + + def test_remote_reference_delete_cleanup_rejects_path_traversal(self): + with tempfile.TemporaryDirectory() as tmp_dir: + base_dir = Path(tmp_dir) + git_dir = base_dir / "repo" / ".git" + git_dir.mkdir(parents=True) + outside_path = base_dir / "outside_remote_delete.txt" + outside_path.write_text("do not delete\n", encoding="utf-8") + + class GitStub: + branch_called = False + + def branch(self, *args): + self.branch_called = True + + class RepoStub: + pass + + repo = RepoStub() + repo.git = GitStub() + repo.common_dir = str(git_dir) + repo.git_dir = str(git_dir) + ref = RemoteReference(repo, "../../outside_remote_delete.txt", check_path=False) + + self.assertRaises(ValueError, RemoteReference.delete, repo, ref) + assert not repo.git.branch_called + assert outside_path.read_text(encoding="utf-8") == "do not delete\n" + def test_validity_ref_names(self): """Ensure ref names are checked for validity. diff --git a/test/test_repo.py b/test/test_repo.py index 544b5c561..fae3dc0b9 100644 --- a/test/test_repo.py +++ b/test/test_repo.py @@ -146,6 +146,23 @@ def test_commit_from_revision(self): self.assertEqual(commit.type, "commit") self.assertEqual(self.rorepo.commit(commit), commit) + @with_rw_directory + def test_commit_from_tag_starting_with_at(self, rw_dir): + repo = Repo.init(rw_dir) + with repo.config_writer() as writer: + writer.set_value("user", "name", "GitPython Tests") + writer.set_value("user", "email", "gitpython@example.com") + + tracked_file = Path(rw_dir) / "hello.txt" + tracked_file.write_text("hello") + repo.index.add([str(tracked_file)]) + commit = repo.index.commit("init") + repo.create_tag("@foo") + + self.assertEqual(repo.tags["@foo"].commit, commit) + self.assertEqual(repo.commit("@"), commit) + self.assertEqual(repo.commit("@foo"), commit) + def test_commits(self): mc = 10 commits = list(self.rorepo.iter_commits("0.1.6", max_count=mc)) @@ -848,8 +865,13 @@ def test_rev_parse(self): # Currently, nothing more is supported. self.assertRaises(NotImplementedError, rev_parse, "@{1 week ago}") - # The last position. - assert rev_parse("@{1}") != head.commit + # The previous position, if this checkout has enough reflog history. + try: + previous = rev_parse("@{1}") + except IndexError: + pass + else: + self.assertNotEqual(previous, head.commit) def test_repo_odbtype(self): target_type = GitCmdObjectDB @@ -1126,6 +1148,42 @@ def test_git_work_tree_env(self, rw_dir): self.assertEqual(r.working_tree_dir, repo_dir) self.assertEqual(r.working_dir, repo_dir) + @with_rw_directory + def test_git_work_tree_env_in_linked_worktree(self, rw_dir): + """Check that Repo() autodiscovers a linked worktree when GIT_DIR is set.""" + git = Git(rw_dir) + if git.version_info[:3] < (2, 5, 1): + raise RuntimeError("worktree feature unsupported (test needs git 2.5.1 or later)") + + rw_master = self.rorepo.clone(join_path_native(rw_dir, "master_repo")) + branch = rw_master.create_head("bbbbbbbb") + worktree_path = join_path_native(rw_dir, "worktree_repo") + if Git.is_cygwin(): + worktree_path = cygpath(worktree_path) + + rw_master.git.worktree("add", worktree_path, branch.name) + + git_dir = Git(worktree_path).rev_parse("--git-dir") + + patched_env = dict(os.environ) + patched_env["GIT_DIR"] = git_dir + patched_env.pop("GIT_WORK_TREE", None) + patched_env.pop("GIT_COMMON_DIR", None) + + with mock.patch.dict(os.environ, patched_env, clear=True): + old_cwd = os.getcwd() + try: + os.chdir(worktree_path) + + explicit = Repo(os.getcwd()) + autodiscovered = Repo() + + self.assertTrue(osp.samefile(explicit.working_tree_dir, worktree_path)) + self.assertTrue(osp.samefile(autodiscovered.working_tree_dir, worktree_path)) + self.assertTrue(osp.samefile(autodiscovered.working_tree_dir, explicit.working_tree_dir)) + finally: + os.chdir(old_cwd) + @with_rw_directory def test_rebasing(self, rw_dir): r = Repo.init(rw_dir) diff --git a/test/test_rev_parse.py b/test/test_rev_parse.py new file mode 100644 index 000000000..b00347668 --- /dev/null +++ b/test/test_rev_parse.py @@ -0,0 +1,159 @@ +# Copyright (C) 2026 Michael Trier (mtrier@gmail.com) and contributors +# +# This module is part of GitPython and is released under the +# 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/ + +from pathlib import Path + +import pytest + +from git import Repo +from git.refs import RemoteReference +from git.refs import SymbolicReference +from gitdb.exc import BadName + + +def _write(repo, path, content): + full_path = Path(repo.working_tree_dir) / path + full_path.parent.mkdir(parents=True, exist_ok=True) + full_path.write_text(content) + repo.index.add([str(full_path)]) + + +@pytest.fixture +def rev_parse_repo(tmp_path): + repo = Repo.init(tmp_path) + with repo.config_writer() as writer: + writer.set_value("user", "name", "GitPython Tests") + writer.set_value("user", "email", "gitpython@example.com") + + _write(repo, "README.md", "root\n") + _write(repo, "CHANGES", "root changes\n") + _write(repo, "dir/file.txt", "root file\n") + root = repo.index.commit("root commit") + repo.create_tag("ann", ref=root, message="annotated tag") + + _write(repo, "README.md", "release\n") + release = repo.index.commit("release candidate") + repo.create_tag("v1.0", ref=release) + main = repo.active_branch + + _write(repo, "side.txt", "side\n") + side_commit = repo.index.commit("side branch", parent_commits=[root], head=False, skip_hooks=True) + repo.create_head("side", side_commit) + + merge = repo.index.commit("merge side", parent_commits=[release, side_commit], skip_hooks=True) + repo.head.log_append(side_commit.binsha, "checkout: moving from side to main", merge.binsha) + + repo.create_head("aaaaaaaa", merge) + repo.create_tag("@foo", ref=merge) + + return { + "repo": repo, + "root": root, + "release": release, + "side": side_commit, + "merge": merge, + "main": main, + } + + +def test_rev_parse_names_hex_and_describe_forms(rev_parse_repo): + repo = rev_parse_repo["repo"] + release = rev_parse_repo["release"] + merge = rev_parse_repo["merge"] + + assert repo.rev_parse("@") == merge + assert repo.rev_parse("@foo") == merge + assert repo.rev_parse("aaaaaaaa") == merge + assert repo.rev_parse(merge.hexsha[:7]) == merge + describe_name = "anything-9-g%s" % merge.hexsha[:7] + assert repo.rev_parse("v1.0-1-g%s" % merge.hexsha[:7]) == merge + assert repo.rev_parse(describe_name) == merge + assert repo.rev_parse("%s-dirty" % merge.hexsha[:7]) == merge + + repo.create_tag(describe_name, ref=release) + assert repo.rev_parse(describe_name) == release + + +def test_rev_parse_navigation_and_peeling(rev_parse_repo): + repo = rev_parse_repo["repo"] + root = rev_parse_repo["root"] + release = rev_parse_repo["release"] + side = rev_parse_repo["side"] + merge = rev_parse_repo["merge"] + tag = repo.rev_parse("ann") + + assert repo.rev_parse("HEAD^0") == merge + assert repo.rev_parse("HEAD~0") == merge + assert repo.rev_parse("HEAD^1") == release + assert repo.rev_parse("HEAD^2") == side + assert repo.rev_parse("HEAD~") == release + assert repo.rev_parse("HEAD^^") == root + + assert tag.type == "tag" + assert repo.rev_parse("ann^{object}") == tag + assert repo.rev_parse("ann^{tag}") == tag + assert repo.rev_parse("ann^{}") == root + assert repo.rev_parse("ann^{commit}") == root + assert repo.rev_parse("HEAD^{tree}") == merge.tree + with pytest.raises(ValueError): + repo.rev_parse("HEAD^{/}") + + +def test_rev_parse_tree_and_index_paths(rev_parse_repo): + repo = rev_parse_repo["repo"] + merge = rev_parse_repo["merge"] + + assert repo.rev_parse("HEAD:") == merge.tree + assert repo.rev_parse("HEAD:README.md") == merge.tree["README.md"] + assert repo.rev_parse("HEAD^{tree}:README.md") == merge.tree["README.md"] + assert repo.rev_parse(":README.md").binsha == merge.tree["README.md"].binsha + assert repo.rev_parse(":0:README.md").binsha == merge.tree["README.md"].binsha + + +def test_rev_parse_reflog_selectors(rev_parse_repo): + repo = rev_parse_repo["repo"] + merge = rev_parse_repo["merge"] + side = rev_parse_repo["side"] + main = rev_parse_repo["main"] + release = rev_parse_repo["release"] + + assert repo.rev_parse("@{0}") == merge + assert repo.rev_parse("@{+0}") == merge + assert repo.rev_parse("@{1}") == release + assert repo.rev_parse("%s@{0}" % main.name) == merge + assert repo.rev_parse("@{-1}") == side + + SymbolicReference.create(repo, "refs/remotes/origin/%s" % main.name, merge) + main.set_tracking_branch(RemoteReference(repo, "refs/remotes/origin/%s" % main.name)) + assert repo.rev_parse("%s@{upstream}" % main.name) == merge + + +def test_rev_parse_commit_message_search(rev_parse_repo): + repo = rev_parse_repo["repo"] + release = rev_parse_repo["release"] + merge = rev_parse_repo["merge"] + + assert repo.rev_parse(":/release") == release + assert repo.rev_parse("HEAD^{/release}") == release + assert repo.rev_parse("HEAD^{/!-release}") == merge + + +def test_rev_parse_rejects_invalid_object_specs(rev_parse_repo): + repo = rev_parse_repo["repo"] + + with pytest.raises(ValueError): + repo.rev_parse(":") + with pytest.raises(ValueError): + repo.rev_parse(":/") + with pytest.raises(ValueError): + repo.rev_parse(":/[") + with pytest.raises(ValueError): + repo.rev_parse("HEAD^{/[}") + with pytest.raises(ValueError): + repo.rev_parse("@{-0}") + with pytest.raises(ValueError): + repo.rev_parse("HEAD^{invalid}") + with pytest.raises(BadName): + repo.rev_parse(":missing")