gitutil.py

# Copyright (c) 2019 SAP SE or an SAP affiliate company. All rights reserved. This file is licensed
# under the Apache Software License, v. 2 except as noted otherwise in the LICENSE file
#
# 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 contextlib
import functools
import os
import subprocess
import tempfile
import urllib.parse

import git
import git.objects.util

from github.util import GitHubRepoBranch
from ci.util import not_empty, not_none, existing_dir, fail, info, random_str, urljoin
from model.github import (
    GithubConfig,
    Protocol,
)


def _ssh_auth_env(github_cfg):
    credentials = github_cfg.credentials()
    info(f'using github-credentials with uid: {credentials.username()}')

    tmp_id = tempfile.NamedTemporaryFile(mode='w', delete=False) # attention: callers must unlink
    tmp_id.write(credentials.private_key())
    tmp_id.flush()

    os.chmod(tmp_id.name, 0o400)
    suppress_hostcheck = '-o "StrictHostKeyChecking no"'
    id_only = '-o "IdentitiesOnly yes"'
    cmd_env = os.environ.copy()
    cmd_env['GIT_SSH_COMMAND'] = f'ssh -v -i {tmp_id.name} {suppress_hostcheck} {id_only}'
    return (cmd_env, tmp_id)


class GitHelper(object):
    def __init__(self, repo, github_cfg, github_repo_path):
        not_none(repo)
        if not isinstance(repo, git.Repo):
            # assume it's a file path if it's not already a git.Repo
            repo = git.Repo(str(repo))
        self.repo = repo
        self.github_cfg = github_cfg
        self.github_repo_path = github_repo_path

    @staticmethod
    def clone_into(
        target_directory: str,
        github_cfg: GithubConfig,
        github_repo_path: str,
        checkout_branch: str = None,
    ) -> 'GitHelper':

        protocol = github_cfg.preferred_protocol()
        if protocol is Protocol.SSH:
            cmd_env, tmp_id = _ssh_auth_env(github_cfg=github_cfg)
            url = urljoin(github_cfg.ssh_url(), github_repo_path)
        elif protocol is Protocol.HTTPS:
            url = url_with_credentials(github_cfg, github_repo_path)
        else:
            raise NotImplementedError

        args = ['--quiet']
        if checkout_branch is not None:
            args += ['--branch', checkout_branch, '--single-branch']
        args += [url, target_directory]

        repo = git.Git()
        if protocol is Protocol.SSH:
            with repo.custom_environment(**cmd_env):
                repo.clone(*args)
        else:
            repo.clone(*args)

        if protocol is Protocol.SSH:
            os.unlink(tmp_id.name)

        return GitHelper(
            repo = target_directory,
            github_cfg = github_cfg,
            github_repo_path = github_repo_path,
        )

    @staticmethod
    def from_githubrepobranch(
        githubrepobranch: GitHubRepoBranch,
        repo_path: str,
    ):
        return GitHelper(
            repo=repo_path,
            github_cfg=githubrepobranch.github_config(),
            github_repo_path=githubrepobranch.github_repo_path(),
        )

    def _changed_file_paths(self):
        lines = git.cmd.Git(self.repo.working_tree_dir).status('--porcelain=1', '-z').split('\x00')
        # output of git status --porcelain=1 and -z is guaranteed to not change in the future
        return [line[3:] for line in lines if line]

    @contextlib.contextmanager
    def _authenticated_remote(self):
        protocol = self.github_cfg.preferred_protocol()
        if protocol is Protocol.SSH:
            url = urljoin(self.github_cfg.ssh_url(), self.github_repo_path)
            cmd_env, tmp_id = _ssh_auth_env(github_cfg=self.github_cfg)
        elif protocol is Protocol.HTTPS:
            url = url_with_credentials(self.github_cfg, self.github_repo_path)
            cmd_env = os.environ
        else:
            raise NotImplementedError

        remote = git.remote.Remote.add(
            repo=self.repo,
            name=random_str(),
            url=url,
        )
        info(f'autenticated remote {remote.name} using {protocol}')

        try:
            yield (cmd_env, remote)
        finally:
            self.repo.delete_remote(remote)
            if protocol is Protocol.SSH:
                os.unlink(tmp_id.name)

    def index_to_commit(self, message, parent_commits=None):
        '''moves all diffs from worktree to a new commit without modifying branches.
        The worktree remains unchanged after the method returns.

        @param parent_commits: optional iterable of parent commits; head is used if absent
        @return the git.Commit object representing the newly created commit
        '''
        if not parent_commits:
            parent_commits = [self.repo.head.commit]
        # add all changes
        git.cmd.Git(self.repo.working_tree_dir).add('.')
        tree = self.repo.index.write_tree()

        if self.github_cfg:
            credentials = self.github_cfg.credentials()
            author = git.objects.util.Actor(credentials.username(), credentials.email_address())
            committer = git.objects.util.Actor(credentials.username(), credentials.email_address())

            create_commit = functools.partial(
                git.Commit.create_from_tree,
                author=author,
                committer=committer,
            )
        else:
            create_commit = git.Commit.create_from_tree

        commit = create_commit(
            repo=self.repo,
            tree=tree,
            parent_commits=parent_commits,
            message=message
        )
        self.repo.index.reset()
        return commit

    def _stash_changes(self):
        self.repo.git.stash('--include-untracked', '--quiet')

    def _has_stash(self):
        return bool(self.repo.git.stash('list'))

    def _pop_stash(self):
        self.repo.git.stash('pop', '--quiet')

    def push(self, from_ref, to_ref):
        with self._authenticated_remote() as (cmd_env, remote):
            with remote.repo.git.custom_environment(**cmd_env):
                remote.push(':'.join((from_ref, to_ref)))

    def rebase(self, commit_ish: str):
        self.repo.git.rebase('--quiet', commit_ish)

    def fetch_head(self, ref: str):
        with self._authenticated_remote() as (cmd_env, remote):
            with remote.repo.git.custom_environment(**cmd_env):
                fetch_result = remote.fetch(ref)[0]
                return fetch_result.commit


def url_with_credentials(github_cfg, github_repo_path):
    base_url = urllib.parse.urlparse(github_cfg.http_url())
    credentials = github_cfg.credentials()
    credentials_str = ':'.join((credentials.username(), credentials.passwd()))
    url = urllib.parse.urlunparse((
        base_url.scheme,
        '@'.join((credentials_str, base_url.hostname)),
        github_repo_path,
        '',
        '',
        ''
    ))
    return url


def update_submodule(
    repo_path: str,
    tree_ish: str,
    submodule_path: str,
    commit_hash: str,
    author: str,
    email: str,
):
    '''Update the submodule of a git-repository to a specific commit.

    Create a new commit, with the passed tree-ish as parent, in the given repository.

    Note that this implementation only supports toplevel submodules. To be removed in a
    future version.

    Parameters
    ------
    repo_path : str
        Path to a directory containing an intialised git-repo with a submodule to update.
    tree_ish : str
        Valid tree-ish to use as base for creating the new commit. Used as parent for the
        commit to be created
        Example: 'master' for the head of the master-branch.
    submodule_path : str
        Path (relative to the repository root) to the submodule. Must be immediately below the root
        of the repository.
    commit_hash : str
        The hash the submodule should point to in the created commit. This should be a valid commit-
        hash in the submodule's repository.
    author : str,
        Will be set as author of the created commit
    email : str
        Will be set for the author of the created commit

    Returns
    ------
    str
        The hexadecimal SHA-1 hash of the created commit
    '''
    repo_path = existing_dir(os.path.abspath(repo_path))

    not_empty(submodule_path)
    if '/' in submodule_path:
        fail('This implementation only supports toplevel submodules: {s}'.format(s=submodule_path))

    not_empty(tree_ish)
    not_empty(commit_hash)
    not_empty(author)
    not_empty(email)

    repo = git.Repo(repo_path)
    _ensure_submodule_exists(repo, submodule_path)

    # Create mk-tree-parseable string-representation from given tree-ish.
    tree = repo.tree(tree_ish)
    tree_representation = _serialise_and_update_submodule(tree, submodule_path, commit_hash)

    # Pass the patched tree to git mk-tree using GitPython. We cannot do this in GitPython
    # directly as it does not support arbitrary tree manipulation.
    # We must keep a reference to auto_interrupt as it closes all streams to the subprocess
    # on finalisation
    auto_interrupt = repo.git.mktree(istream = subprocess.PIPE, as_process=True)
    process = auto_interrupt.proc
    stdout, _= process.communicate(input=tree_representation.encode())

    # returned string is byte-encoded and newline-terminated
    new_sha = stdout.decode('utf-8').strip()

    # Create a new commit in the repo's object database from the newly created tree.
    actor = git.Actor(author, email)
    parent_commit = repo.commit(tree_ish)
    commit = git.Commit.create_from_tree(
      repo = repo,
      tree = new_sha,
      parent_commits = [parent_commit],
      message='Upgrade submodule {s} to commit {c}'.format(
          s=submodule_path,
          c=commit_hash,
      ),
      author=actor,
      committer=actor,
    )

    return commit.hexsha


def _serialise_and_update_submodule(
    tree: git.Tree,
    submodule_path: str,
    commit_hash: str,
):
    '''Return a modified, serialised tree-representation in which the given submodule's entry is
    altered such that it points to the specified commit hash.
    The returned serialisation  format is understood by git mk-tree.

    Returns
    ------
        str
            An updated serialised git-tree with the updated submodule entry
    '''
    # GitPython offers no API to retrieve ls-tree representation
    return '\n'.join([
        _serialise_object_replace_submodule(
            tree_element=tree_element,
            submodule_path=submodule_path,
            commit_hash=commit_hash,
        ) for tree_element in tree]
    )


def _serialise_object_replace_submodule(tree_element, submodule_path, commit_hash):
    # GitPython uses the special type 'submodule' for submodules whereas git uses 'commit'.
    if tree_element.type == 'submodule':
        element_type = 'commit'
        # Replace the hash the of the 'commit'-tree with the passed value if the submodule
        # is at the specified path
        if tree_element.path == submodule_path:
            element_sha = commit_hash
    else:
        element_type = tree_element.type
        element_sha = tree_element.hexsha

    return '{mode} {type} {sha}\t{path}'.format(
        sha=element_sha,
        type=element_type,
        # mode is a number in octal representation WITHOUT '0o' prefix
        mode=format(tree_element.mode, 'o'),
        path=tree_element.path,
    )


def _ensure_submodule_exists(repo: git.Repo, path: str):
    '''Use GitPython to verify that a submodule with the given path exists in the repository.'''
    for submodule in repo.submodules:
        if submodule.path == path:
            return
    fail('No submodule with path {p} exists in the repository.'.format(p=path))