/brz/remove-bazaar

# Copyright (C) 2007-2018 Jelmer Vernooij <jelmer@jelmer.uk>

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

"""Remote dirs, repositories and branches."""

from __future__ import absolute_import

import gzip

from io import BytesIO

import re

from .. import (

    config,

    debug,

    errors,

    osutils,

    trace,

    ui,

    urlutils,

    )

from ..push import (

    PushResult,

    )

from ..errors import (

    AlreadyBranchError,

    BzrError,

    DivergedBranches,

    InProcessTransport,

    InvalidRevisionId,

    NoSuchFile,

    NoSuchRevision,

    NoSuchTag,

    NotBranchError,

    NotLocalUrl,

    PermissionDenied,

    UninitializableFormat,

    )

from ..revision import NULL_REVISION

from ..revisiontree import RevisionTree

from ..sixish import (

    text_type,

    viewitems,

    )

from ..transport import (

    Transport,

    register_urlparse_netloc_protocol,

    )

from . import (

    lazy_check_versions,

    is_github_url,

    user_agent_for_github,

    )

lazy_check_versions()

from .branch import (

    GitBranch,

    GitBranchFormat,

    GitBranchPushResult,

    GitTags,

    _quick_lookup_revno,

    )

from .dir import (

    GitControlDirFormat,

    GitDir,

    )

from .errors import (

    GitSmartRemoteNotSupported,

    NoSuchRef,

    )

from .mapping import (

    encode_git_path,

    mapping_registry,

    )

from .object_store import (

    get_object_store,

    )

from .push import (

    remote_divergence,

    )

from .repository import (

    GitRepository,

    GitRepositoryFormat,

    )

from .refs import (

    branch_name_to_ref,

    is_peeled,

    ref_to_tag_name,

    tag_name_to_ref,

    )

import dulwich

import dulwich.client

from dulwich.errors import (

    GitProtocolError,

    HangupException,

    )

from dulwich.pack import (

    Pack,

    pack_objects_to_data,

    )

from dulwich.protocol import ZERO_SHA

from dulwich.refs import (

    DictRefsContainer,

    SYMREF,

    )

from dulwich.repo import (

    NotGitRepository,

    )

import os

import select

try:

    import urllib.parse as urlparse

    from urllib.parse import splituser

except ImportError:

    import urlparse

    from urllib import splituser

# urlparse only supports a limited number of schemes by default

register_urlparse_netloc_protocol('git')

register_urlparse_netloc_protocol('git+ssh')

from dulwich.pack import load_pack_index

class GitPushResult(PushResult):

    def _lookup_revno(self, revid):

        try:

            return _quick_lookup_revno(self.source_branch, self.target_branch,

                                       revid)

        except GitSmartRemoteNotSupported:

            return None

    @property

    def old_revno(self):

        return self._lookup_revno(self.old_revid)

    @property

    def new_revno(self):

        return self._lookup_revno(self.new_revid)

# Don't run any tests on GitSmartTransport as it is not intended to be

# a full implementation of Transport

def get_test_permutations():

    return []

def split_git_url(url):

    """Split a Git URL.

    :param url: Git URL

    :return: Tuple with host, port, username, path.

    """

    parsed_url = urlparse.urlparse(url)

    path = urlparse.unquote(parsed_url.path)

    if path.startswith("/~"):

        path = path[1:]

    return ((parsed_url.hostname or '', parsed_url.port, parsed_url.username, path))

class RemoteGitError(BzrError):

    _fmt = "Remote server error: %(msg)s"

class HeadUpdateFailed(BzrError):

    _fmt = ("Unable to update remote HEAD branch. To update the master "

            "branch, specify the URL %(base_url)s,branch=master.")

    def __init__(self, base_url):

        super(HeadUpdateFailed, self).__init__()

        self.base_url = base_url

def parse_git_error(url, message):

    """Parse a remote git server error and return a bzr exception.

    :param url: URL of the remote repository

    :param message: Message sent by the remote git server

    """

    message = str(message).strip()

    if (message.startswith("Could not find Repository ")

        or message == 'Repository not found.'

            or (message.startswith('Repository ') and

                message.endswith(' not found.'))):

        return NotBranchError(url, message)

    if message == "HEAD failed to update":

        base_url = urlutils.strip_segment_parameters(url)

        return HeadUpdateFailed(base_url)

    if message.startswith('access denied or repository not exported:'):

        extra, path = message.split(':', 1)

        return PermissionDenied(path.strip(), extra)

    if message.endswith('You are not allowed to push code to this project.'):

        return PermissionDenied(url, message)

    if message.endswith(' does not appear to be a git repository'):

        return NotBranchError(url, message)

    if message == 'pre-receive hook declined':

        return PermissionDenied(url, message)

    if re.match('(.+) is not a valid repository name',

                message.splitlines()[0]):

        return NotBranchError(url, message)

    if message == (

            'GitLab: You are not allowed to push code to protected branches '

            'on this project.'):

        return PermissionDenied(url, message)

    m = re.match(r'Permission to ([^ ]+) denied to ([^ ]+)\.', message)

    if m:

        return PermissionDenied(m.group(1), 'denied to %s' % m.group(2))

    # Don't know, just return it to the user as-is

    return RemoteGitError(message)

def parse_git_hangup(url, e):

    """Parse the error lines from a git servers stderr on hangup.

    :param url: URL of the remote repository

    :param e: A HangupException

    """

    stderr_lines = getattr(e, 'stderr_lines', None)

    if not stderr_lines:

        return e

    if all(line.startswith(b'remote: ') for line in stderr_lines):

        stderr_lines = [

            line[len(b'remote: '):] for line in stderr_lines]

    interesting_lines = [

        line for line in stderr_lines

        if line and line.replace(b'=', b'')]

    if len(interesting_lines) == 1:

        interesting_line = interesting_lines[0]

        return parse_git_error(

            url, interesting_line.decode('utf-8', 'surrogateescape'))

    return RemoteGitError(

        b'\n'.join(stderr_lines).decode('utf-8', 'surrogateescape'))

class GitSmartTransport(Transport):

    def __init__(self, url, _client=None):

        Transport.__init__(self, url)

        (self._host, self._port, self._username, self._path) = \

            split_git_url(url)

        if 'transport' in debug.debug_flags:

            trace.mutter('host: %r, user: %r, port: %r, path: %r',

                         self._host, self._username, self._port, self._path)

        self._client = _client

        self._stripped_path = self._path.rsplit(",", 1)[0]

    def external_url(self):

        return self.base

    def has(self, relpath):

        return False

    def _get_client(self):

        raise NotImplementedError(self._get_client)

    def _get_path(self):

        return self._stripped_path

    def get(self, path):

        raise NoSuchFile(path)

    def abspath(self, relpath):

        return urlutils.join(self.base, relpath)

    def clone(self, offset=None):

        """See Transport.clone()."""

        if offset is None:

            newurl = self.base

        else:

            newurl = urlutils.join(self.base, offset)

        return self.__class__(newurl, self._client)

class TCPGitSmartTransport(GitSmartTransport):

    _scheme = 'git'

    def _get_client(self):

        if self._client is not None:

            ret = self._client

            self._client = None

            return ret

        if self._host == '':

            # return dulwich.client.LocalGitClient()

            return dulwich.client.SubprocessGitClient()

        return dulwich.client.TCPGitClient(

            self._host, self._port, report_activity=self._report_activity)

class SSHSocketWrapper(object):

    def __init__(self, sock):

        self.sock = sock

    def read(self, len=None):

        return self.sock.recv(len)

    def write(self, data):

        return self.sock.write(data)

    def can_read(self):

        return len(select.select([self.sock.fileno()], [], [], 0)[0]) > 0

class DulwichSSHVendor(dulwich.client.SSHVendor):

    def __init__(self):

        from ..transport import ssh

        self.bzr_ssh_vendor = ssh._get_ssh_vendor()

    def run_command(self, host, command, username=None, port=None):

        connection = self.bzr_ssh_vendor.connect_ssh(

            username=username, password=None, port=port, host=host,

            command=command)

        (kind, io_object) = connection.get_sock_or_pipes()

        if kind == 'socket':

            return SSHSocketWrapper(io_object)

        else:

            raise AssertionError("Unknown io object kind %r'" % kind)

# dulwich.client.get_ssh_vendor = DulwichSSHVendor

class SSHGitSmartTransport(GitSmartTransport):

    _scheme = 'git+ssh'

    def _get_path(self):

        path = self._stripped_path

        if path.startswith("/~/"):

            return path[3:]

        return path

    def _get_client(self):

        if self._client is not None:

            ret = self._client

            self._client = None

            return ret

        location_config = config.LocationConfig(self.base)

        client = dulwich.client.SSHGitClient(

            self._host, self._port, self._username,

            report_activity=self._report_activity)

        # Set up alternate pack program paths

        upload_pack = location_config.get_user_option('git_upload_pack')

        if upload_pack:

            client.alternative_paths["upload-pack"] = upload_pack

        receive_pack = location_config.get_user_option('git_receive_pack')

        if receive_pack:

            client.alternative_paths["receive-pack"] = receive_pack

        return client

class RemoteGitBranchFormat(GitBranchFormat):

    def get_format_description(self):

        return 'Remote Git Branch'

    @property

    def _matchingcontroldir(self):

        return RemoteGitControlDirFormat()

    def initialize(self, a_controldir, name=None, repository=None,

                   append_revisions_only=None):

        raise UninitializableFormat(self)

class DefaultProgressReporter(object):

    _GIT_PROGRESS_PARTIAL_RE = re.compile(r"(.*?): +(\d+)% \((\d+)/(\d+)\)")

    _GIT_PROGRESS_TOTAL_RE = re.compile(r"(.*?): (\d+)")

    def __init__(self, pb):

        self.pb = pb

    def progress(self, text):

        text = text.rstrip(b"\r\n")

        text = text.decode('utf-8')

        if text.lower().startswith('error: '):

            trace.show_error('git: %s', text[len(b'error: '):])

        else:

            trace.mutter("git: %s", text)

            g = self._GIT_PROGRESS_PARTIAL_RE.match(text)

            if g is not None:

                (text, pct, current, total) = g.groups()

                self.pb.update(text, int(current), int(total))

            else:

                g = self._GIT_PROGRESS_TOTAL_RE.match(text)

                if g is not None:

                    (text, total) = g.groups()

                    self.pb.update(text, None, int(total))

                else:

                    trace.note("%s", text)

class RemoteGitDir(GitDir):

    def __init__(self, transport, format, client, client_path):

        self._format = format

        self.root_transport = transport

        self.transport = transport

        self._mode_check_done = None

        self._client = client

        self._client_path = client_path

        self.base = self.root_transport.base

        self._refs = None

    @property

    def _gitrepository_class(self):

        return RemoteGitRepository

    def archive(self, format, committish, write_data, progress=None,

                write_error=None, subdirs=None, prefix=None):

        if progress is None:

            pb = ui.ui_factory.nested_progress_bar()

            progress = DefaultProgressReporter(pb).progress

        else:

            pb = None

        def progress_wrapper(message):

            if message.startswith(b"fatal: Unknown archive format \'"):

                format = message.strip()[len(b"fatal: Unknown archive format '"):-1]

                raise errors.NoSuchExportFormat(format.decode('ascii'))

            return progress(message)

        try:

            self._client.archive(

                self._client_path, committish, write_data, progress_wrapper,

                write_error,

                format=(format.encode('ascii') if format else None),

                subdirs=subdirs,

                prefix=(encode_git_path(prefix) if prefix else None))

        except HangupException as e:

            raise parse_git_hangup(self.transport.external_url(), e)

        except GitProtocolError as e:

            raise parse_git_error(self.transport.external_url(), e)

        finally:

            if pb is not None:

                pb.finished()

    def fetch_pack(self, determine_wants, graph_walker, pack_data,

                   progress=None):

        if progress is None:

            pb = ui.ui_factory.nested_progress_bar()

            progress = DefaultProgressReporter(pb).progress

        else:

            pb = None

        try:

            result = self._client.fetch_pack(

                self._client_path, determine_wants, graph_walker, pack_data,

                progress)

            if result.refs is None:

                result.refs = {}

            self._refs = remote_refs_dict_to_container(

                result.refs, result.symrefs)

            return result

        except HangupException as e:

            raise parse_git_hangup(self.transport.external_url(), e)

        except GitProtocolError as e:

            raise parse_git_error(self.transport.external_url(), e)

        finally:

            if pb is not None:

                pb.finished()

    def send_pack(self, get_changed_refs, generate_pack_data, progress=None):

        if progress is None:

            pb = ui.ui_factory.nested_progress_bar()

            progress = DefaultProgressReporter(pb).progress

        else:

            pb = None

        def get_changed_refs_wrapper(remote_refs):

            if self._refs is not None:

                update_refs_container(self._refs, remote_refs)

            return get_changed_refs(remote_refs)

        try:

            return self._client.send_pack(

                self._client_path, get_changed_refs_wrapper,

                generate_pack_data, progress)

        except HangupException as e:

            raise parse_git_hangup(self.transport.external_url(), e)

        except GitProtocolError as e:

            raise parse_git_error(self.transport.external_url(), e)

        finally:

            if pb is not None:

                pb.finished()

    def create_branch(self, name=None, repository=None,

                      append_revisions_only=None, ref=None):

        refname = self._get_selected_ref(name, ref)

        if refname != b'HEAD' and refname in self.get_refs_container():

            raise AlreadyBranchError(self.user_url)

        ref_chain, unused_sha = self.get_refs_container().follow(

            self._get_selected_ref(name))

        if ref_chain and ref_chain[0] == b'HEAD':

            refname = ref_chain[1]

        repo = self.open_repository()

        return RemoteGitBranch(self, repo, refname)

    def destroy_branch(self, name=None):

        refname = self._get_selected_ref(name)

        def get_changed_refs(old_refs):

            ret = {}

            if refname not in old_refs:

                raise NotBranchError(self.user_url)

            ret[refname] = dulwich.client.ZERO_SHA

            return ret

        def generate_pack_data(have, want, ofs_delta=False):

            return pack_objects_to_data([])

        result = self.send_pack(get_changed_refs, generate_pack_data)

        if result is not None and not isinstance(result, dict):

            error = result.ref_status.get(refname)

            if error:

                raise RemoteGitError(error)

    @property

    def user_url(self):

        return self.control_url

    @property

    def user_transport(self):

        return self.root_transport

    @property

    def control_url(self):

        return self.control_transport.base

    @property

    def control_transport(self):

        return self.root_transport

    def open_repository(self):

        return RemoteGitRepository(self)

    def get_branch_reference(self, name=None):

        ref = branch_name_to_ref(name)

        val = self.get_refs_container().read_ref(ref)

        if val.startswith(SYMREF):

            return val[len(SYMREF):]

        return None

    def open_branch(self, name=None, unsupported=False,

                    ignore_fallbacks=False, ref=None, possible_transports=None,

                    nascent_ok=False):

        repo = self.open_repository()

        ref = self._get_selected_ref(name, ref)

        try:

            if not nascent_ok and ref not in self.get_refs_container():

                raise NotBranchError(

                    self.root_transport.base, controldir=self)

        except NotGitRepository:

            raise NotBranchError(self.root_transport.base,

                                 controldir=self)

        ref_chain, unused_sha = self.get_refs_container().follow(ref)

        return RemoteGitBranch(self, repo, ref_chain[-1])

    def open_workingtree(self, recommend_upgrade=False):

        raise NotLocalUrl(self.transport.base)

    def has_workingtree(self):

        return False

    def get_peeled(self, name):

        return self.get_refs_container().get_peeled(name)

    def get_refs_container(self):

        if self._refs is not None:

            return self._refs

        result = self.fetch_pack(lambda x: None, None,

                                 lambda x: None,

                                 lambda x: trace.mutter("git: %s" % x))

        self._refs = remote_refs_dict_to_container(

            result.refs, result.symrefs)

        return self._refs

    def push_branch(self, source, revision_id=None, overwrite=False,

                    remember=False, create_prefix=False, lossy=False,

                    name=None, tag_selector=None):

        """Push the source branch into this ControlDir."""

        if revision_id is None:

            # No revision supplied by the user, default to the branch

            # revision

            revision_id = source.last_revision()

        else:

            if not source.repository.has_revision(revision_id):

                raise NoSuchRevision(source, revision_id)

        push_result = GitPushResult()

        push_result.workingtree_updated = None

        push_result.master_branch = None