GIT_PROTO_PAT = re.compile(r'^/(.+)/(info/refs|git-upload-pack|git-receive-pack)$')

cmd_mapping = {

    'git-receive-pack': 'push',

    'git-upload-pack': 'pull',

}

class SimpleGit(BaseVCSController):

    scm_alias = 'git'

    @classmethod

    def parse_request(cls, environ):

        path_info = get_path_info(environ)

        m = GIT_PROTO_PAT.match(path_info)

        if m is None:

            return None

        class parsed_request(object):

            # See https://git-scm.com/book/en/v2/Git-Internals-Transfer-Protocols#_the_smart_protocol

            repo_name = m.group(1).rstrip('/')

            cmd = m.group(2)

            query_string = environ['QUERY_STRING']

            if cmd == 'info/refs' and query_string.startswith('service='):

                service = query_string.split('=', 1)[1]

                action = cmd_mapping.get(service)

            else:

                service = None

                action = cmd_mapping.get(cmd)

        return parsed_request

    def _make_app(self, parsed_request):

        """

        Return a pygrack wsgi application.

        """

        pygrack_app = make_wsgi_app(parsed_request.repo_name, self.basepath)

        def wrapper_app(environ, start_response):

            if (parsed_request.cmd == 'info/refs' and

                parsed_request.service == 'git-upload-pack'

            ):

                baseui = make_ui()

                repo = db.Repository.get_by_repo_name(parsed_request.repo_name)

                scm_repo = repo.scm_instance

                hooks.log_pull_action(ui=baseui, repo=scm_repo._repo)

            # Note: push hooks are handled by post-receive hook

            return pygrack_app(environ, start_response)

        return wrapper_app

        callback(deleted_by=deleted_by, **user_dict)

def _hook_environment(repo_path):

    """

    Create a light-weight environment for stand-alone scripts and return an UI and the

    db repository.

    Git hooks are executed as subprocess of Git while Kallithea is waiting, and

    they thus need enough info to be able to create an app environment and

    connect to the database.

    """

    import kallithea.config.application

    extras = get_hook_environment()

    path_to_ini_file = extras['config']

    config = paste.deploy.appconfig('config:' + path_to_ini_file)

    #logging.config.fileConfig(ini_file_path) # Note: we are in a different process - don't use configured logging

    kallithea.config.application.make_app(config.global_conf, **config.local_conf)

    # fix if it's not a bare repo

    if repo_path.endswith(os.sep + '.git'):

        repo_path = repo_path[:-5]

    repo = db.Repository.get_by_full_path(repo_path)

    if not repo:

        raise OSError('Repository %s not found in database' % repo_path)

    baseui = make_ui()

    return baseui, repo

def handle_git_pre_receive(repo_path, git_stdin_lines):

    """Called from Git pre-receive hook.

    The returned value is used as hook exit code and must be 0.

    """

    # Currently unused. TODO: remove?

    return 0

def handle_git_post_receive(repo_path, git_stdin_lines):

    """Called from Git post-receive hook.

    The returned value is used as hook exit code and must be 0.

    """

    try:

        baseui, repo = _hook_environment(repo_path)

    except HookEnvironmentError as e:

        return 0

    # the post push hook should never use the cached instance

    scm_repo = repo.scm_instance_no_cache()

    rev_data = []

    for l in git_stdin_lines:

        old_rev, new_rev, ref = l.strip().split(' ')

        _ref_data = ref.split('/')

        if _ref_data[1] in ['tags', 'heads']:

            rev_data.append({'old_rev': old_rev,

                             'new_rev': new_rev,

                             'ref': ref,

                             'type': _ref_data[1],

                             'name': '/'.join(_ref_data[2:])})

    git_revs = []

    for push_ref in rev_data:

        _type = push_ref['type']

        if _type == 'heads':

            if push_ref['old_rev'] == EmptyChangeset().raw_id:

                # update the symbolic ref if we push new repo

                if scm_repo.is_empty():

                    scm_repo._repo.refs.set_symbolic_ref(

                        b'HEAD',

                        b'refs/heads/%s' % safe_bytes(push_ref['name']))

                # build exclude list without the ref

                cmd = ['for-each-ref', '--format=%(refname)', 'refs/heads/*']

                stdout = scm_repo.run_git_command(cmd)

                ref = push_ref['ref']

                heads = [head for head in stdout.splitlines() if head != ref]

                # now list the git revs while excluding from the list

                cmd = ['log', push_ref['new_rev'], '--reverse', '--pretty=format:%H']

                cmd.append('--not')

                cmd.extend(heads) # empty list is ok

                stdout = scm_repo.run_git_command(cmd)

                git_revs += stdout.splitlines()

            elif push_ref['new_rev'] == EmptyChangeset().raw_id:

                # delete branch case

                git_revs += ['delete_branch=>%s' % push_ref['name']]

            else:

                cmd = ['log', '%(old_rev)s..%(new_rev)s' % push_ref,

                       '--reverse', '--pretty=format:%H']

                stdout = scm_repo.run_git_command(cmd)

                git_revs += stdout.splitlines()

def obfuscate_url_pw(engine):

    try:

        _url = sa_url.make_url(engine or '')

    except ArgumentError:

        return engine

    if _url.password:

        _url.password = 'XXXXX'

    return str(_url)

class HookEnvironmentError(Exception): pass

def get_hook_environment():

    """

    Get hook context by deserializing the global KALLITHEA_EXTRAS environment

    variable.

    Called early in Git out-of-process hooks to get .ini config path so the

    basic environment can be configured properly. Also used in all hooks to get

    information about the action that triggered it.

    """

    try:

        kallithea_extras = os.environ['KALLITHEA_EXTRAS']

    except KeyError:

        raise HookEnvironmentError("Environment variable KALLITHEA_EXTRAS not found")

    extras = json.loads(kallithea_extras)

    for k in ['username', 'repository', 'scm', 'action', 'ip', 'config']:

        try:

            extras[k]

        except KeyError:

            raise HookEnvironmentError('Missing key %s in KALLITHEA_EXTRAS %s' % (k, extras))

    return AttributeDict(extras)

def set_hook_environment(username, ip_addr, repo_name, repo_alias, action=None):

    """Prepare global context for running hooks by serializing data in the

    global KALLITHEA_EXTRAS environment variable.

    Most importantly, this allow Git hooks to do proper logging and updating of

    caches after pushes.

    Must always be called before anything with hooks are invoked.

    """

    extras = {

        'username': username,

        'repository': repo_name,

        'config': kallithea.CONFIG['__file__'], # used by git hook to read config

    }

    os.environ['KALLITHEA_EXTRAS'] = json.dumps(extras)

def get_current_authuser():

    """

    Gets kallithea user from threadlocal tmpl_context variable if it's

    defined, else returns None.

    """

    try:

        return getattr(tmpl_context, 'authuser', None)

    except TypeError:  # No object (name: context) has been registered for this thread

        return None

def urlreadable(s, _cleanstringsub=re.compile('[^-a-zA-Z0-9./]+').sub):

    return _cleanstringsub('_', s).rstrip('_')

def recursive_replace(str_, replace=' '):

    """

    Recursive replace of given sign to just one instance

    :param str_: given string

    :param replace: char to find and replace multiple instances

    Examples::

    >>> recursive_replace("Mighty---Mighty-Bo--sstones",'-')

    'Mighty-Mighty-Bo-sstones'

    """

    if str_.find(replace * 2) == -1:

        return str_

    else:

        str_ = str_.replace(replace * 2, replace)

        return recursive_replace(str_, replace)

def repo_name_slug(value):

    """

    Return slug of name of repository

    This function is called on each creation/modification

    of repository to prevent bad names in repo

    """

    slug = remove_formatting(value)

    slug = strip_tags(slug)

    # Set defaults, in case .exit should be called early

    vcs_type = None

    repo_name = None

    @staticmethod

    def make(ssh_command):

        """Factory function. Given a command as invoked over SSH (and preserved

        in SSH_ORIGINAL_COMMAND when run as authorized_keys command), return a

        handler if the command looks ok, else return None.

        """

        raise NotImplementedError

    def __init__(self, repo_name):

        self.repo_name = repo_name.rstrip('/')

    def serve(self, user_id, key_id, client_ip):

        """Verify basic sanity of the repository, and that the user is

        valid and has access - then serve the native VCS protocol for

        repository access."""

        dbuser = db.User.get(user_id)

        if dbuser is None:

            self.exit('User %r not found' % user_id)

        self.authuser = AuthUser.make(dbuser=dbuser, ip_addr=client_ip)

        log.info('Authorized user %s from SSH %s trusting user id %s and key id %s for %r', dbuser, client_ip, user_id, key_id, self.repo_name)

        if self.authuser is None: # not ok ... but already kind of authenticated by SSH ... but not really not authorized ...

            self.exit('User %s from %s cannot be authorized' % (dbuser.username, client_ip))

        ssh_key = db.UserSshKeys.get(key_id)

        if ssh_key is None:

            self.exit('SSH key %r not found' % key_id)

        ssh_key.last_seen = datetime.datetime.now()

        meta.Session().commit()

        if HasPermissionAnyMiddleware('repository.write',

                                      'repository.admin')(self.authuser, self.repo_name):

            self.allow_push = True

        elif HasPermissionAnyMiddleware('repository.read')(self.authuser, self.repo_name):

            self.allow_push = False

        else:

            self.exit('Access to %r denied' % self.repo_name)

        self.db_repo = db.Repository.get_by_repo_name(self.repo_name)

        if self.db_repo is None:

            self.exit("Repository '%s' not found" % self.repo_name)

        assert self.db_repo.repo_name == self.repo_name

        # Set global hook environment up for 'push' actions.

        set_hook_environment(self.authuser.username, client_ip, self.repo_name, self.vcs_type, 'push')

        return self._serve()

    def _serve(self):

        """Serve the native protocol for repository access."""

        raise NotImplementedError

    def exit(self, error):

        log.info('abort serving %s %s: %s', self.vcs_type, self.repo_name, error)

        sys.stderr.write('abort: %s\n' % error)

        sys.exit(1)

from kallithea.lib.vcs.ssh import base

log = logging.getLogger(__name__)

class GitSshHandler(base.BaseSshHandler):

    vcs_type = 'git'

    @classmethod

    def make(cls, ssh_command_parts):

        r"""

        >>> import shlex

        >>> GitSshHandler.make(shlex.split("git-upload-pack '/foo bar'")).repo_name

        'foo bar'

        >>> GitSshHandler.make(shlex.split("git-upload-pack '/foo bar'")).verb

        'git-upload-pack'

        >>> GitSshHandler.make(shlex.split(" git-upload-pack /blåbærgrød ")).repo_name # might not be necessary to support no quoting ... but we can

        'bl\xe5b\xe6rgr\xf8d'

        >>> GitSshHandler.make(shlex.split('''git-upload-pack "/foo'bar"''')).repo_name

        "foo'bar"

        >>> GitSshHandler.make(shlex.split("git-receive-pack '/foo'")).repo_name

        'foo'

        >>> GitSshHandler.make(shlex.split("git-receive-pack '/foo'")).verb

        'git-receive-pack'

        >>> GitSshHandler.make(shlex.split("/bin/git-upload-pack '/foo'")) # ssh-serve will report 'SSH command %r is not supported'

        >>> GitSshHandler.make(shlex.split('''git-upload-pack /foo bar''')) # ssh-serve will report 'SSH command %r is not supported'

        >>> shlex.split("git-upload-pack '/foo'bar' x") # ssh-serve will report: Error parsing SSH command "...": No closing quotation

        Traceback (most recent call last):

        ValueError: No closing quotation

        >>> GitSshHandler.make(shlex.split('hg -R foo serve --stdio')) # not handled here

        """

        if (len(ssh_command_parts) == 2 and

            ssh_command_parts[0] in ['git-upload-pack', 'git-receive-pack'] and

            ssh_command_parts[1].startswith('/')

        ):

            return cls(ssh_command_parts[1][1:], ssh_command_parts[0])

        return None

    def __init__(self, repo_name, verb):

        base.BaseSshHandler.__init__(self, repo_name)

        self.verb = verb

    def _serve(self):

        if self.verb == 'git-upload-pack': # action 'pull'

            hooks.log_pull_action(ui=make_ui(), repo=self.db_repo.scm_instance._repo)

        else: # probably verb 'git-receive-pack', action 'push'

            if not self.allow_push:

                self.exit('Push access to %r denied' % self.repo_name)

            # Note: push logging is handled by Git post-receive hook

        # git shell is not a real shell but use shell inspired quoting *inside* the argument.

        # Per https://github.com/git/git/blob/v2.22.0/quote.c#L12 :

        # The path must be "'" quoted, but "'" and "!" must exit the quoting and be "\" escaped

        quoted_abspath = "'%s'" % self.db_repo.repo_full_path.replace("'", r"'\''").replace("!", r"'\!'")

        newcmd = ['git', 'shell', '-c', "%s %s" % (self.verb, quoted_abspath)]

        log.debug('Serving: %s', newcmd)

        os.execvp(newcmd[0], newcmd)

        self.exit("Failed to exec 'git' as %s" % newcmd)