"istream",

"with_extended_output",

"with_exceptions",

"as_process",

"output_stream",

"stdout_as_string",

"kill_after_timeout",

"with_stdout",

"universal_newlines",

"shell",

"env",

"max_chunk_size",

"strip_newline_in_stdout",

"""Kill a POSIX process and its descendants, returning whether it was killed."""

# Collect descendants before signalling, while their parent PIDs still identify them.

pids = [pid]

for parent_pid in pids:

try:

try:

p = Popen(["pgrep", "-P", str(parent_pid)], stdout=PIPE)

except FileNotFoundError:

# POSIX ps does not support selecting by parent PID.

ps_args = ["ps", "-ef"] if sys.platform == "cygwin" else ["ps", "-A", "-o", "pid=", "-o", "ppid="]

with Popen(ps_args, stdout=PIPE) as p:

if p.stdout is not None:

for line in p.stdout:

fields = line.split()

if sys.platform == "cygwin":

fields = fields[1:3] # ps -ef starts with UID, PID, PPID.

if len(fields) == 2 and all(field.isdigit() for field in fields):

if int(fields[1]) == parent_pid:

pids.append(int(fields[0]))

else:

with p:

if p.stdout is not None:

for line in p.stdout:

if line.strip().isdigit():

pids.append(int(line))

except OSError as ex:

_logger.info("Unable to enumerate child processes: %r", ex)

killed = False

if sys.platform != "win32":

for process_pid in pids:

try:

os.kill(process_pid, signal.SIGKILL)

killed = killed or process_pid == pid

except OSError:

pass

process: Union["Git.AutoInterrupt", Popen],

stdout_handler: Union[

None,

Callable[[AnyStr], None],

Callable[[List[AnyStr]], None],

Callable[[bytes, "Repo", "DiffIndex"], None],

],

stderr_handler: Union[None, Callable[[AnyStr], None], Callable[[List[AnyStr]], None]],

finalizer: Union[None, Callable[[Union[Popen, "Git.AutoInterrupt"]], None]] = None,

decode_streams: bool = True,

kill_after_timeout: Union[None, float] = None,

) -> None:

R"""Register for notifications to learn that process output is ready to read, and

# Use 2 "pump" threads and wait for both to finish.

def pump_stream(

cmdline: List[str],

name: str,

stream: Union[BinaryIO, TextIO],

is_decode: bool,

handler: Union[None, Callable[[Union[bytes, str]], None]],

) -> None:

try:

for line in stream:

if handler:

if is_decode:

assert isinstance(line, bytes)

line_str = line.decode(defenc)

handler(line_str)

else:

handler(line)

except Exception as ex:

_logger.error(f"Pumping {name!r} of cmd({remove_password_if_present(cmdline)}) failed due to: {ex!r}")

if not (isinstance(ex, ValueError) and stream.closed):

# Only reraise if the error was not due to the stream closing.

raise CommandError([f"<{name}-pump>"] + remove_password_if_present(cmdline), ex) from ex

finally:

stream.close()

if hasattr(process, "proc"):

process = cast("Git.AutoInterrupt", process)

cmdline: str | Tuple[str, ...] | List[str] = getattr(process.proc, "args", "")

p_stdout = process.proc.stdout if process.proc else None

p_stderr = process.proc.stderr if process.proc else None

else:

process = cast(Popen, process) # type: ignore[redundant-cast]

cmdline = getattr(process, "args", "")

p_stdout = process.stdout

p_stderr = process.stderr

if not isinstance(cmdline, (tuple, list)):

cmdline = cmdline.split()

pumps: List[Tuple[str, IO, Callable[..., None] | None]] = []

if p_stdout:

pumps.append(("stdout", p_stdout, stdout_handler))

if p_stderr:

pumps.append(("stderr", p_stderr, stderr_handler))

threads: List[threading.Thread] = []

for name, stream, handler in pumps:

t = threading.Thread(target=pump_stream, args=(cmdline, name, stream, decode_streams, handler))

t.daemon = True

t.start()

threads.append(t)

# Wait for output handlers to finish before finalizing.

# If the child needs stdin, the caller must arrange to feed/close it

# before this call or concurrently; this function only drains output.

deadline = None if kill_after_timeout is None else time.monotonic() + kill_after_timeout

for t in threads:

t.join(timeout=None if deadline is None else max(0, deadline - time.monotonic()))

if t.is_alive():

if isinstance(process, Git.AutoInterrupt):

if sys.platform != "win32" and process.proc is not None and process.proc.poll() is None:

_kill_process(process.proc.pid)

process._terminate()

else: # Don't want to deal with the other case.

raise RuntimeError(

"Thread join() timed out in cmd.handle_process_output()."

f" kill_after_timeout={kill_after_timeout} seconds"

)

process._timeout_error = (

f"error: process killed because it timed out. kill_after_timeout={kill_after_timeout} seconds"

)

break

if finalizer:

command: Union[str, Sequence[Any]],

*,

shell: bool = False,

env: Optional[Mapping[str, str]] = None,

**kwargs: Any,

) -> Popen:

"""Call :class:`subprocess.Popen` on Windows but don't include a CWD in the

# CREATE_NEW_PROCESS_GROUP is needed for some ways of killing it afterwards.

# https://docs.python.org/3/library/subprocess.html#subprocess.Popen.send_signal

# https://docs.python.org/3/library/subprocess.html#subprocess.CREATE_NEW_PROCESS_GROUP

creationflags = subprocess.CREATE_NO_WINDOW | subprocess.CREATE_NEW_PROCESS_GROUP

# When using a shell, the shell is the direct subprocess, so the variable must

# be set in its environment, to affect its search behavior.

if shell:

# The original may be immutable, or the caller may reuse it. Mutate a copy.

env = {} if env is None else dict(env)

env["NoDefaultCurrentDirectoryInExePath"] = "1" # The "1" can be any value.

# When not using a shell, the current process does the search in a

# CreateProcessW API call, so the variable must be set in our environment. With

# a shell, that's unnecessary if https://github.com/python/cpython/issues/101283

# is patched. In Python versions where it is unpatched, in the rare case the

# ComSpec environment variable is unset, the search for the shell itself is

# unsafe. Setting NoDefaultCurrentDirectoryInExePath in all cases, as done here,

# is simpler and protects against that. (As above, the "1" can be any value.)

with patch_env("NoDefaultCurrentDirectoryInExePath", "1"):

return Popen(

command,

shell=shell,

env=env,

creationflags=creationflags,

**kwargs,

for k, v in d.items():

setattr(self, k, v)

for k in excluded:

"""Process wrapper that terminates the wrapped process on finalization.

__slots__ = ("proc", "args", "status", "_timeout_error")

# If this is non-zero it will override any status code during _terminate, used

# to prevent race conditions in testing.

_status_code_if_terminate: int = 0

def __init__(self, proc: Union[None, subprocess.Popen], args: Any) -> None:

self.proc = proc

self.args = args

self.status: Union[int, None] = None

self._timeout_error: Optional[str] = None

def _terminate(self) -> None:

"""Terminate the underlying process."""

if self.proc is None:

return

proc = self.proc

self.proc = None

try:

if proc.stdin:

# A timed-out process may already have exited before input is flushed.

with contextlib.suppress(BrokenPipeError):

proc.stdin.close()

# Did the process finish already so we have a return code?

try:

if proc.poll() is not None:

self.status = self._status_code_if_terminate or proc.poll()

return

except OSError as ex:

_logger.info("Ignored error after process had died: %r", ex)

# It can be that nothing really exists anymore...

if getattr(os, "kill", None) is None:

return

# Try to kill it.

try:

proc.terminate()

except (OSError, AttributeError) as ex:

# On interpreter shutdown (notably on Windows), parts of the stdlib used by

# subprocess can already be torn down (e.g. `subprocess._winapi` becomes None),

# which can cause AttributeError during terminate(). In that case, we prefer

# to silently ignore to avoid noisy "Exception ignored in: __del__" messages.

_logger.info("Ignored error while terminating process: %r", ex)

return

# END exception handling

finally:

if proc.stdout:

proc.stdout.close()

if proc.stderr:

proc.stderr.close()

try:

status = proc.wait()

self.status = self._status_code_if_terminate or status

except (OSError, AttributeError) as ex:

_logger.info("Ignored error while waiting for terminated process: %r", ex)

def __del__(self) -> None:

self._terminate()

def __getattr__(self, attr: str) -> Any:

return getattr(self.proc, attr)

# TODO: Bad choice to mimic `proc.wait()` but with different args.

def wait(self, stderr: Union[None, str, bytes] = b"") -> int:

"""Wait for the process and return its status code.

status: Union[int, None]

if self.proc is not None:

status = self.proc.wait()

p_stderr = self.proc.stderr

else: # Assume the underlying proc was killed earlier or never existed.

status = self.status

p_stderr = None

stderr_b = force_bytes(data=stderr, encoding="utf-8") or b""

if self._timeout_error is not None:

stderr_b += (b"\n" if stderr_b else b"") + self._timeout_error.encode("utf-8")

status = status or 1

def read_all_from_possibly_closed_stream(stream: Union[IO[bytes], None]) -> bytes:

if stream:

try:

return stderr_b + force_bytes(stream.read())

except (OSError, ValueError):

return stderr_b or b""

else:

return stderr_b or b""

# END status handling

if status != 0:

errstr = read_all_from_possibly_closed_stream(p_stderr)

_logger.debug("AutoInterrupt wait stderr: %r" % (errstr,))

raise GitCommandError(remove_password_if_present(self.args), status, errstr)

"""Object representing a sized read-only stream returning the contents of

__slots__ = ("_stream", "_nbr", "_size")

def __init__(self, size: int, stream: IO[bytes]) -> None:

self._stream = stream

self._size = size

self._nbr = 0 # Number of bytes read.

# Special case: If the object is empty, has null bytes, get the final

# newline right away.

if size == 0:

stream.read(1)

# END handle empty streams

def read(self, size: int = -1) -> bytes:

bytes_left = self._size - self._nbr

if bytes_left == 0:

return b""

if size > -1:

# Ensure we don't try to read past our limit.

size = min(bytes_left, size)

else:

# They try to read all, make sure it's not more than what remains.

size = bytes_left

# END check early depletion

data = self._stream.read(size)

self._nbr += len(data)

# Check for depletion, read our final byte to make the stream usable by

# others.

if self._size - self._nbr == 0:

self._stream.read(1) # final newline

# END finish reading

return data

def readline(self, size: int = -1) -> bytes:

if self._nbr == self._size:

return b""

# Clamp size to lowest allowed value.

bytes_left = self._size - self._nbr

if size > -1:

size = min(bytes_left, size)

else:

size = bytes_left

# END handle size

data = self._stream.readline(size)

self._nbr += len(data)

# Handle final byte.

if self._size - self._nbr == 0:

self._stream.read(1)

# END finish reading

return data

def readlines(self, size: int = -1) -> List[bytes]:

if self._nbr == self._size:

return []

# Leave all additional logic to our readline method, we just check the size.

out = []

nbr = 0

while True:

line = self.readline()

if not line:

break

out.append(line)

if size > -1:

nbr += len(line)

if nbr > size:

break

# END handle size constraint

# END readline loop

return out

# skipcq: PYL-E0301

def __iter__(self) -> "Git.CatFileContentStream":

return self

def __next__(self) -> bytes:

line = self.readline()

if not line:

raise StopIteration

return line

next = __next__

def __del__(self) -> None:

bytes_left = self._size - self._nbr

if bytes_left:

# Read and discard - seeking is impossible within a stream.

# This includes any terminating newline.

self._stream.read(bytes_left + 1)

"""The Git class manages communication with the Git binary.

__slots__ = (

"_working_dir",

"cat_file_all",

"cat_file_header",

"_version_info",

"_version_info_token",

"_git_options",

"_persistent_git_options",

"_environment",

"_repo",

)

_excluded_ = (

"cat_file_all",

"cat_file_header",

"_version_info",

"_version_info_token",

"_repo",

)

_version_info: Optional[Tuple[int, ...]]

_version_info_token: object

# Match Git's leading transport selector, including an empty helper name.

# Git also selects the command-executing ext helper for an ext:// URL.

re_unsafe_protocol = re.compile(r"([A-Za-z0-9][A-Za-z0-9+.-]*|)::|ext://")

unsafe_git_ls_remote_options = [

# This option allows arbitrary command execution in git-ls-remote.

"--upload-pack",

"--exec",

]

unsafe_git_pathspec_from_file_options = [

# Reads pathspecs from a caller-controlled file. Some commands include an

# unmatched pathspec in their error output, which can disclose the file.

"--pathspec-from-file",

]

def __getstate__(self) -> Dict[str, Any]:

return slots_to_dict(self, exclude=self._excluded_)

def __setstate__(self, d: Dict[str, Any]) -> None:

dict_to_slots_and__excluded_are_none(self, d, excluded=self._excluded_)

# CONFIGURATION

git_exec_name = "git"

"""Default git command that should work on Linux, Windows, and other systems."""

GIT_PYTHON_TRACE = os.environ.get("GIT_PYTHON_TRACE", False)

"""Enables debugging of GitPython's git commands."""

_git_exec_env_var = "GIT_PYTHON_GIT_EXECUTABLE"

_refresh_env_var = "GIT_PYTHON_REFRESH"

GIT_PYTHON_GIT_EXECUTABLE = None

"""Provide the full path to the git executable. Otherwise it assumes git is in the

_refresh_token = object() # Since None would match an initial _version_info_token.

_version_check_cache: Dict[Tuple[Any, ...], Tuple[int, ...]] = {}

@classmethod

def refresh(cls, path: Union[None, PathLike] = None) -> bool:

"""Update information about the git executable :class:`Git` objects will use.

# Discern which path to refresh with.

if path is not None:

new_git = os.path.expanduser(path)

new_git = os.path.abspath(new_git)

else:

new_git = os.environ.get(cls._git_exec_env_var, cls.git_exec_name)

# Keep track of the old and new git executable path.

old_git = cls.GIT_PYTHON_GIT_EXECUTABLE

old_refresh_token = cls._refresh_token

cls.GIT_PYTHON_GIT_EXECUTABLE = new_git

cls._refresh_token = object()

# Test if the new git executable path is valid. A GitCommandNotFound error is

# raised by us. A PermissionError is raised if the git executable cannot be

# executed for whatever reason.

has_git = False

try:

cls().version()

has_git = True

except (GitCommandNotFound, PermissionError):

pass

# Warn or raise exception if test failed.

if not has_git:

err = (

dedent(

"""\

)

% cls._git_exec_env_var

)

# Revert to whatever the old_git was.

cls.GIT_PYTHON_GIT_EXECUTABLE = old_git

cls._refresh_token = old_refresh_token

if old_git is None:

# On the first refresh (when GIT_PYTHON_GIT_EXECUTABLE is None) we only

# are quiet, warn, or error depending on the GIT_PYTHON_REFRESH value.

# Determine what the user wants to happen during the initial refresh. We

# expect GIT_PYTHON_REFRESH to either be unset or be one of the

# following values:

#

# 0|q|quiet|s|silence|silent|n|none

# 1|w|warn|warning|l|log

# 2|r|raise|e|error|exception

mode = os.environ.get(cls._refresh_env_var, "raise").lower()

quiet = ["quiet", "q", "silence", "s", "silent", "none", "n", "0"]

warn = ["warn", "w", "warning", "log", "l", "1"]

error = ["error", "e", "exception", "raise", "r", "2"]

if mode in quiet:

pass

elif mode in warn or mode in error:

err = dedent(

"""\

) % (

err,

cls._refresh_env_var,

"|".join(quiet),

"|".join(warn),

"|".join(error),

cls._refresh_env_var,

quiet[0],

)

if mode in warn:

_logger.critical(err)

else:

raise ImportError(err)

else:

err = dedent(

"""\

) % (

cls._refresh_env_var,

"|".join(quiet),

"|".join(warn),

"|".join(error),

)

raise ImportError(err)

# We get here if this was the initial refresh and the refresh mode was

# not error. Go ahead and set the GIT_PYTHON_GIT_EXECUTABLE such that we

# discern the difference between the first refresh at import time

# and subsequent calls to git.refresh or this refresh method.

cls.GIT_PYTHON_GIT_EXECUTABLE = cls.git_exec_name

else:

# After the first refresh (when GIT_PYTHON_GIT_EXECUTABLE is no longer

# None) we raise an exception.

raise GitCommandNotFound(new_git, err)

return has_git

@classmethod

def is_cygwin(cls) -> bool:

return is_cygwin_git(cls.GIT_PYTHON_GIT_EXECUTABLE)

@overload

@classmethod

def polish_url(cls, url: str, is_cygwin: Literal[False] = ...) -> str: ...

@overload

@classmethod

def polish_url(cls, url: str, is_cygwin: Union[None, bool] = None) -> str: ...

@classmethod

def polish_url(cls, url: str, is_cygwin: Union[None, bool] = None) -> PathLike:

"""Remove any backslashes from URLs to be written in config files.

if is_cygwin is None:

is_cygwin = cls.is_cygwin()

if is_cygwin:

url = cygpath(url)

else:

if url.startswith("~"):

url = os.path.expanduser(url)

url = url.replace("\\\\", "\\").replace("\\", "/")

return url

@classmethod

def check_unsafe_protocols(cls, url: str) -> None:

"""Check for unsafe protocols.

match = cls.re_unsafe_protocol.match(url)

if match:

protocol = match.group(0)

raise UnsafeProtocolError(

f"The `{protocol}` protocol looks suspicious, use `allow_unsafe_protocols=True` to allow it."

)

def _check_unsafe_protocols_in_args(self, args: Sequence[Any], kwargs: Mapping[str, Any]) -> None:

"""Check positional operands and standalone values in rendered command options.

for arg in self._unpack_args(args) + self.transform_kwargs(**kwargs):

self.check_unsafe_protocols(arg)

@classmethod

def _canonicalize_option_name(cls, option: str) -> str:

"""Return the option name used for unsafe-option checks.

option_name = option.lstrip("-").split("=", 1)[0]

option_tokens = option_name.split(None, 1)

if not option_tokens:

return ""

return dashify(option_tokens[0])

@classmethod

def check_unsafe_options(

cls, options: List[str], unsafe_options: List[str], clusterable_short_options: str = "46flnqsv"

) -> None:

"""Raise :class:`~git.exc.UnsafeOptionError` for blocked option spellings.

# Options can be of the form `foo`, `--foo`, `--foo bar`, or `--foo=bar`.

# Git accepts any unambiguous prefix of a long option, so an abbreviated

# spelling such as `--upl` for `--upload-pack` must be rejected too. An

# option is unsafe if its canonical name is a prefix of any blocked

# option's canonical name. Only long options and multi-character kwargs

# can be abbreviations; single-character short options remain exact-match

# only.

canonical_unsafe_options = {cls._canonicalize_option_name(option): option for option in unsafe_options}

unsafe_short_options = {

canonical: option

for canonical, option in canonical_unsafe_options.items()

if option.startswith("-") and not option.startswith("--") and len(canonical) == 1

}

# These value-less Git flags can be clustered before another short option

# (for example, ``-fuVALUE``). Stop at any other character because it may

# begin an attached value, as ``o`` does in the safe option ``-oupstream``.

clusterable_short_options_set = frozenset(clusterable_short_options)

options_are_kwargs = all(not option.startswith("-") for option in options)

for option in options:

candidate = cls._canonicalize_option_name(option)

if not candidate:

continue

unsafe_option = canonical_unsafe_options.get(candidate)

if unsafe_option is not None:

raise UnsafeOptionError(f"{unsafe_option} is not allowed, use `allow_unsafe_options=True` to allow it.")

option_token = option.split("=", 1)[0].split(None, 1)[0]

if option_token.startswith("-") and not option_token.startswith("--"):

for option_char in option_token[1:]:

unsafe_option = unsafe_short_options.get(option_char)

if unsafe_option is not None:

raise UnsafeOptionError(

f"{unsafe_option} is not allowed, use `allow_unsafe_options=True` to allow it."

)

if option_char not in clusterable_short_options_set:

break

if not (option.startswith("--") or (options_are_kwargs and len(candidate) > 1)):

continue

for canonical, unsafe_option in canonical_unsafe_options.items():

if canonical.startswith(candidate):

raise UnsafeOptionError(

f"{unsafe_option} is not allowed, use `allow_unsafe_options=True` to allow it."

)

@classmethod

def _option_candidates(cls, args: Sequence[Any] = (), kwargs: Optional[Mapping[str, Any]] = None) -> List[str]:

"""Collect possible option spellings before command-line transformation."""

options = [

option for option in cls._unpack_args([arg for arg in args if arg is not None]) if option.startswith("-")

]

if kwargs:

split_single_char_options = kwargs.get("split_single_char_options", True)

for key, value in kwargs.items():

values = value if isinstance(value, (list, tuple)) else (value,)

if any(value is True or (value is not False and value is not None) for value in values):

key = str(key)

if len(key) != 1:

options.append(f"--{dashify(key)}")

elif split_single_char_options:

options.append(f"-{key}")