import kallithea.bin.kallithea_cli_ssh

# -*- coding: utf-8 -*-

# 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 3 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, see <http://www.gnu.org/licenses/>.

import click

import kallithea.bin.kallithea_cli_base as cli_base

import os

import sys

import re

import logging

import shlex

import kallithea

from kallithea.lib.utils2 import str2bool

from kallithea.lib.vcs.backends.git.ssh import GitSshHandler

from kallithea.lib.vcs.backends.hg.ssh import MercurialSshHandler

log = logging.getLogger(__name__)

@cli_base.register_command(config_file_initialize_app=True, hidden=True)

@click.argument('user-id', type=click.INT, required=True)

@click.argument('key-id', type=click.INT, required=True)

def ssh_serve(user_id, key_id):

    """Serve SSH repository protocol access.

    The trusted command that is invoked from .ssh/authorized_keys to serve SSH

    protocol access. The access will be granted as the specified user ID, and

    logged as using the specified key ID.

    """

    ssh_enabled = kallithea.CONFIG.get('ssh_enabled', False)

    if not str2bool(ssh_enabled):

        sys.stderr.write("SSH access is disabled.\n")

        return sys.exit(1)

    ssh_original_command = os.environ.get('SSH_ORIGINAL_COMMAND', '')

    connection = re.search('^([\d\.]+)', os.environ.get('SSH_CONNECTION', ''))

    client_ip = connection.group(1) if connection else '0.0.0.0'

    log.debug('ssh-serve was invoked for SSH command %r from %s', ssh_original_command, client_ip)

    if not ssh_original_command:

        if os.environ.get('SSH_CONNECTION'):

            sys.stderr.write("'kallithea-cli ssh-serve' can only provide protocol access over SSH. Interactive SSH login for this user is disabled.\n")

        else:

            sys.stderr.write("'kallithea-cli ssh-serve' cannot be called directly. It must be specified as command in an SSH authorized_keys file.\n")

        return sys.exit(1)

    try:

        ssh_command_parts = shlex.split(ssh_original_command)

    except ValueError as e:

        sys.stderr.write('Error parsing SSH command %r: %s\n' % (ssh_original_command, e))

        sys.exit(1)

    for VcsHandler in [MercurialSshHandler, GitSshHandler]:

        vcs_handler = VcsHandler.make(ssh_command_parts)

        if vcs_handler is not None:

            vcs_handler.serve(user_id, key_id, client_ip)

            assert False # serve is written so it never will terminate

    sys.stderr.write("This account can only be used for repository access. SSH command %r is not supported.\n" % ssh_original_command)

    sys.exit(1)

# Almost exactly like Mercurial contrib/hg-ssh:

def rejectpush(ui, **kwargs):

    """Mercurial hook to be installed as pretxnopen and prepushkey for read-only repos"""

    ex = get_hook_environment()

    ui.warn((b"Push access to %r denied\n") % safe_str(ex.repository))

    return 1

# -*- coding: utf-8 -*-

# 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 3 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, see <http://www.gnu.org/licenses/>.

import os

import logging

from kallithea.lib.hooks import log_pull_action

from kallithea.lib.utils import make_ui

from kallithea.lib.utils2 import safe_unicode, safe_str

from kallithea.lib.vcs.backends.ssh import BaseSshHandler

log = logging.getLogger(__name__)

class GitSshHandler(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

        u'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

        u'bl\xe5b\xe6rgr\xf8d'

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

        u"foo'bar"

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

        u'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(safe_unicode(ssh_command_parts[1][1:]), ssh_command_parts[0])

        return None

    def __init__(self, repo_name, verb):

        self.repo_name = repo_name

        self.verb = verb

    def _serve(self):

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

            # base class called set_hook_environment - action is hardcoded to 'pull'

            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' % safe_str(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)

# -*- coding: utf-8 -*-

# 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 3 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, see <http://www.gnu.org/licenses/>.

import logging

from mercurial import hg

try:

    from mercurial.wireprotoserver import sshserver

except ImportError:

    from mercurial.sshserver import sshserver # moved in Mercurial 4.6 (1bf5263fe5cc)

from kallithea.lib.utils import make_ui

from kallithea.lib.utils2 import safe_unicode, safe_str

from kallithea.lib.vcs.backends.ssh import BaseSshHandler

log = logging.getLogger(__name__)

class MercurialSshHandler(BaseSshHandler):

    vcs_type = 'hg'

    @classmethod

    def make(cls, ssh_command_parts):

        r"""

        >>> import shlex

        >>> MercurialSshHandler.make(shlex.split('hg -R "foo bar" serve --stdio')).repo_name

        u'foo bar'

        >>> MercurialSshHandler.make(shlex.split(' hg -R blåbærgrød serve --stdio ')).repo_name

        u'bl\xe5b\xe6rgr\xf8d'

        >>> MercurialSshHandler.make(shlex.split('''hg -R 'foo"bar' serve --stdio''')).repo_name

        u'foo"bar'

        >>> MercurialSshHandler.make(shlex.split('/bin/hg -R "foo" serve --stdio'))

        >>> MercurialSshHandler.make(shlex.split('''hg -R "foo"bar" serve --stdio''')) # ssh-serve will report: Error parsing SSH command "...": invalid syntax

        Traceback (most recent call last):

        ValueError: No closing quotation

        >>> MercurialSshHandler.make(shlex.split('git-upload-pack "/foo"')) # not handled here

        """

        if ssh_command_parts[:2] == ['hg', '-R'] and ssh_command_parts[3:] == ['serve', '--stdio']:

            return cls(safe_unicode(ssh_command_parts[2]))

        return None

    def __init__(self, repo_name):

        self.repo_name = repo_name

    def _serve(self):

        # Note: we want a repo with config based on .hg/hgrc and can thus not use self.db_repo.scm_instance._repo.ui

        baseui = make_ui(repo_path=self.db_repo.repo_full_path)

        if not self.allow_push:

            baseui.setconfig('hooks', 'pretxnopen._ssh_reject', 'python:kallithea.lib.hooks.rejectpush')

            baseui.setconfig('hooks', 'prepushkey._ssh_reject', 'python:kallithea.lib.hooks.rejectpush')

        repo = hg.repository(baseui, safe_str(self.db_repo.repo_full_path))

        log.debug("Starting Mercurial sshserver for %s", self.db_repo.repo_full_path)

        sshserver(baseui, repo).serve_forever()

# -*- coding: utf-8 -*-

# 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 3 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, see <http://www.gnu.org/licenses/>.

"""

vcs.backends.ssh

~~~~~~~~~~~~~~~~~

SSH backend for all available SCMs

"""

import sys

import logging

from kallithea.model.db import Repository, User

from kallithea.lib.auth import HasPermissionAnyMiddleware, AuthUser

from kallithea.lib.utils2 import safe_str, set_hook_environment

log = logging.getLogger(__name__)

class BaseSshHandler(object):

    # Protocol for setting properties:

    # Set by sub class:

    #   vcs_type: 'hg' or 'git'

    # Set by make() / __init__():

    #   repo_name: requested repo name - only validated by serve()

    # Set by serve() - must not be accessed before:

    #   db_repo: repository db object

    #   authuser: user that has been authenticated - like request.authuser ... which isn't used here

    #   allow_push: false for read-only access to the repo

    # 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 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 = 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))

        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' % safe_str(self.repo_name))

        self.db_repo = 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.

        # If pull actions should be served, the actual hook invocation will be

        # hardcoded to 'pull' when log_pull_action is invoked (directly on Git,

        # or through the Mercurial 'outgoing' hook).

        # For push actions, the action in global hook environment is used (in

        # handle_git_post_receive when it is called as Git post-receive hook,

        # or in log_push_action through the Mercurial 'changegroup' hook).

        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)