git-repo/git_command.py

# Copyright (C) 2008 The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import functools
import os
import sys
import subprocess
from typing import Any, Optional

from error import GitError
from git_refs import HEAD
import platform_utils
from repo_trace import REPO_TRACE, IsTrace, Trace
from wrapper import Wrapper

GIT = "git"
# NB: These do not need to be kept in sync with the repo launcher script.
# These may be much newer as it allows the repo launcher to roll between
# different repo releases while source versions might require a newer git.
#
# The soft version is when we start warning users that the version is old and
# we'll be dropping support for it.  We'll refuse to work with versions older
# than the hard version.
#
# git-1.7 is in (EOL) Ubuntu Precise.  git-1.9 is in Ubuntu Trusty.
MIN_GIT_VERSION_SOFT = (1, 9, 1)
MIN_GIT_VERSION_HARD = (1, 7, 2)
GIT_DIR = "GIT_DIR"

LAST_GITDIR = None
LAST_CWD = None


class _GitCall(object):
    @functools.lru_cache(maxsize=None)
    def version_tuple(self):
        ret = Wrapper().ParseGitVersion()
        if ret is None:
            print("fatal: unable to detect git version", file=sys.stderr)
            sys.exit(1)
        return ret

    def __getattr__(self, name):
        name = name.replace("_", "-")

        def fun(*cmdv):
            command = [name]
            command.extend(cmdv)
            return GitCommand(None, command).Wait() == 0

        return fun


git = _GitCall()


def RepoSourceVersion():
    """Return the version of the repo.git tree."""
    ver = getattr(RepoSourceVersion, "version", None)

    # We avoid GitCommand so we don't run into circular deps -- GitCommand needs
    # to initialize version info we provide.
    if ver is None:
        env = GitCommand._GetBasicEnv()

        proj = os.path.dirname(os.path.abspath(__file__))
        env[GIT_DIR] = os.path.join(proj, ".git")
        result = subprocess.run(
            [GIT, "describe", HEAD],
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            encoding="utf-8",
            env=env,
            check=False,
        )
        if result.returncode == 0:
            ver = result.stdout.strip()
            if ver.startswith("v"):
                ver = ver[1:]
        else:
            ver = "unknown"
        setattr(RepoSourceVersion, "version", ver)

    return ver


class UserAgent(object):
    """Mange User-Agent settings when talking to external services

    We follow the style as documented here:
    https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/User-Agent
    """

    _os = None
    _repo_ua = None
    _git_ua = None

    @property
    def os(self):
        """The operating system name."""
        if self._os is None:
            os_name = sys.platform
            if os_name.lower().startswith("linux"):
                os_name = "Linux"
            elif os_name == "win32":
                os_name = "Win32"
            elif os_name == "cygwin":
                os_name = "Cygwin"
            elif os_name == "darwin":
                os_name = "Darwin"
            self._os = os_name

        return self._os

    @property
    def repo(self):
        """The UA when connecting directly from repo."""
        if self._repo_ua is None:
            py_version = sys.version_info
            self._repo_ua = "git-repo/%s (%s) git/%s Python/%d.%d.%d" % (
                RepoSourceVersion(),
                self.os,
                git.version_tuple().full,
                py_version.major,
                py_version.minor,
                py_version.micro,
            )

        return self._repo_ua

    @property
    def git(self):
        """The UA when running git."""
        if self._git_ua is None:
            self._git_ua = "git/%s (%s) git-repo/%s" % (
                git.version_tuple().full,
                self.os,
                RepoSourceVersion(),
            )

        return self._git_ua


user_agent = UserAgent()


def git_require(min_version, fail=False, msg=""):
    git_version = git.version_tuple()
    if min_version <= git_version:
        return True
    if fail:
        need = ".".join(map(str, min_version))
        if msg:
            msg = " for " + msg
        print(
            "fatal: git %s or later required%s" % (need, msg), file=sys.stderr
        )
        sys.exit(1)
    return False


def _build_env(
    _kwargs_only=(),
    bare: Optional[bool] = False,
    disable_editor: Optional[bool] = False,
    ssh_proxy: Optional[Any] = None,
    gitdir: Optional[str] = None,
    objdir: Optional[str] = None,
):
    """Constucts an env dict for command execution."""

    assert _kwargs_only == (), "_build_env only accepts keyword arguments."

    env = GitCommand._GetBasicEnv()

    if disable_editor:
        env["GIT_EDITOR"] = ":"
    if ssh_proxy:
        env["REPO_SSH_SOCK"] = ssh_proxy.sock()
        env["GIT_SSH"] = ssh_proxy.proxy
        env["GIT_SSH_VARIANT"] = "ssh"
    if "http_proxy" in env and "darwin" == sys.platform:
        s = "'http.proxy=%s'" % (env["http_proxy"],)
        p = env.get("GIT_CONFIG_PARAMETERS")
        if p is not None:
            s = p + " " + s
        env["GIT_CONFIG_PARAMETERS"] = s
    if "GIT_ALLOW_PROTOCOL" not in env:
        env[
            "GIT_ALLOW_PROTOCOL"
        ] = "file:git:http:https:ssh:persistent-http:persistent-https:sso:rpc"
    env["GIT_HTTP_USER_AGENT"] = user_agent.git

    if objdir:
        # Set to the place we want to save the objects.
        env["GIT_OBJECT_DIRECTORY"] = objdir

        alt_objects = os.path.join(gitdir, "objects") if gitdir else None
        if alt_objects and os.path.realpath(alt_objects) != os.path.realpath(
            objdir
        ):
            # Allow git to search the original place in case of local or unique
            # refs that git will attempt to resolve even if we aren't fetching
            # them.
            env["GIT_ALTERNATE_OBJECT_DIRECTORIES"] = alt_objects
    if bare and gitdir is not None:
        env[GIT_DIR] = gitdir

    return env


class GitCommand(object):
    """Wrapper around a single git invocation."""

    def __init__(
        self,
        project,
        cmdv,
        bare=False,
        input=None,
        capture_stdout=False,
        capture_stderr=False,
        merge_output=False,
        disable_editor=False,
        ssh_proxy=None,
        cwd=None,
        gitdir=None,
        objdir=None,
    ):
        if project:
            if not cwd:
                cwd = project.worktree
            if not gitdir:
                gitdir = project.gitdir

        # Git on Windows wants its paths only using / for reliability.
        if platform_utils.isWindows():
            if objdir:
                objdir = objdir.replace("\\", "/")
            if gitdir:
                gitdir = gitdir.replace("\\", "/")

        env = _build_env(
            disable_editor=disable_editor,
            ssh_proxy=ssh_proxy,
            objdir=objdir,
            gitdir=gitdir,
            bare=bare,
        )

        command = [GIT]
        if bare:
            cwd = None
        command.append(cmdv[0])
        # Need to use the --progress flag for fetch/clone so output will be
        # displayed as by default git only does progress output if stderr is a
        # TTY.
        if sys.stderr.isatty() and cmdv[0] in ("fetch", "clone"):
            if "--progress" not in cmdv and "--quiet" not in cmdv:
                command.append("--progress")
        command.extend(cmdv[1:])

        stdin = subprocess.PIPE if input else None
        stdout = subprocess.PIPE if capture_stdout else None
        stderr = (
            subprocess.STDOUT
            if merge_output
            else (subprocess.PIPE if capture_stderr else None)
        )

        dbg = ""
        if IsTrace():
            global LAST_CWD
            global LAST_GITDIR

            if cwd and LAST_CWD != cwd:
                if LAST_GITDIR or LAST_CWD:
                    dbg += "\n"
                dbg += ": cd %s\n" % cwd
                LAST_CWD = cwd

            if GIT_DIR in env and LAST_GITDIR != env[GIT_DIR]:
                if LAST_GITDIR or LAST_CWD:
                    dbg += "\n"
                dbg += ": export GIT_DIR=%s\n" % env[GIT_DIR]
                LAST_GITDIR = env[GIT_DIR]

            if "GIT_OBJECT_DIRECTORY" in env:
                dbg += (
                    ": export GIT_OBJECT_DIRECTORY=%s\n"
                    % env["GIT_OBJECT_DIRECTORY"]
                )
            if "GIT_ALTERNATE_OBJECT_DIRECTORIES" in env:
                dbg += ": export GIT_ALTERNATE_OBJECT_DIRECTORIES=%s\n" % (
                    env["GIT_ALTERNATE_OBJECT_DIRECTORIES"]
                )

            dbg += ": "
            dbg += " ".join(command)
            if stdin == subprocess.PIPE:
                dbg += " 0<|"
            if stdout == subprocess.PIPE:
                dbg += " 1>|"
            if stderr == subprocess.PIPE:
                dbg += " 2>|"
            elif stderr == subprocess.STDOUT:
                dbg += " 2>&1"

        with Trace(
            "git command %s %s with debug: %s", LAST_GITDIR, command, dbg
        ):
            try:
                p = subprocess.Popen(
                    command,
                    cwd=cwd,
                    env=env,
                    encoding="utf-8",
                    errors="backslashreplace",
                    stdin=stdin,
                    stdout=stdout,
                    stderr=stderr,
                )
            except Exception as e:
                raise GitError("%s: %s" % (command[1], e))

            if ssh_proxy:
                ssh_proxy.add_client(p)

            self.process = p

            try:
                self.stdout, self.stderr = p.communicate(input=input)
            finally:
                if ssh_proxy:
                    ssh_proxy.remove_client(p)
            self.rc = p.wait()

    @staticmethod
    def _GetBasicEnv():
        """Return a basic env for running git under.

        This is guaranteed to be side-effect free.
        """
        env = os.environ.copy()
        for key in (
            REPO_TRACE,
            GIT_DIR,
            "GIT_ALTERNATE_OBJECT_DIRECTORIES",
            "GIT_OBJECT_DIRECTORY",
            "GIT_WORK_TREE",
            "GIT_GRAFT_FILE",
            "GIT_INDEX_FILE",
        ):
            env.pop(key, None)
        return env

    def Wait(self):
        return self.rc