path: root/lorrycontroller/gitano.py

# Copyright (C) 2014-2020  Codethink Limited
#
# 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; version 2 of the License.
#
# 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.


import collections
import logging
import re
import urllib.request, urllib.error, urllib.parse

import cliapp
import requests

from . import hosts


class _GitanoCommandFailure(Exception):

    def __init__(self, trovehost, command, stderr):
        Exception.__init__(
            self,
            'Failed to run "%s" on Gitano on %s\n%s' %
            (command, trovehost, stderr))


class _GitanoCommand(object):

    '''Run a Gitano command on a Trove.'''

    def __init__(self, trovehost, protocol, username, password):
        self.trovehost = trovehost
        self.protocol = protocol
        self.username = username
        self.password = password

        if protocol == 'ssh':
            self._command = self._ssh_command
        elif protocol in ('http', 'https'):
            self._command = self._http_command
        else:
            raise _GitanoCommandFailure(
                self.trovehost, '__init__', 'unknown protocol %s' % protocol)

    def whoami(self):
        return self._command(['whoami'])

    def create(self, repo_path):
        self._command(['create', repo_path])

    def get_gitano_config(self, repo_path):
        stdout = self._command(['config', repo_path, 'show'])

        # "config REPO show" outputs a sequence of lines of the form "key: value".
        # Extract those into a collections.defaultdict.

        result = collections.defaultdict(str)
        for line in stdout.splitlines():
            m = re.match(r'^([^:])+:\s*(.*)$', line)
            if m:
                result[m.group(0)] = m.group(1).strip()

        return result

    def set_gitano_config(self, path, key, value):
        self._command(['config', path, 'set', key, value])

    def ls(self):
        return self._command(['ls'])

    def _ssh_command(self, gitano_args):
        quoted_args = [cliapp.shell_quote(x) for x in gitano_args]

        base_argv = [
            'ssh',
            '-oStrictHostKeyChecking=no',
            '-oBatchMode=yes',
             'git@%s' % self.trovehost,
            ]

        exit, stdout, stderr = cliapp.runcmd_unchecked(
            base_argv + quoted_args)
        if isinstance(stdout, bytes):
            stdout = stdout.decode('utf-8', errors='replace')
            stderr = stderr.decode('utf-8', errors='replace')

        if exit != 0:
            logging.error(
                'Failed to run "%s" for %s:\n%s',
                quoted_args, self.trovehost, stdout + stderr)
            raise _GitanoCommandFailure(
                self.trovehost,
                ' '.join(gitano_args),
                stdout + stderr)

        return stdout

    def _http_command(self, gitano_args):
        quoted_args = urllib.parse.quote(' '.join(gitano_args))
        url = urllib.parse.urlunsplit((
            self.protocol,
            self.trovehost,
            '/gitano-command.cgi',
            'cmd=%s' % quoted_args,
            ''))
        logging.debug('url=%r', url)

        try:
            if self.username and self.password:
                response = requests.get(url, auth=(self.username,
                                                   self.password))
            else:
                response = requests.get(url)
        except (requests.exceptions.RequestException) as e:
            raise _GitanoCommandFailure(
                self.trovehost, ' '.join(gitano_args), str(e))

        return response.text


class _LocalTroveGitanoCommand(_GitanoCommand):

    '''Run commands on the local Trove's Gitano.

    This is a version of the GitanoCommand class specifically for
    accessing the local Trove's Gitano.

    '''

    def __init__(self):
        _GitanoCommand.__init__(self, 'localhost', 'ssh', '', '')



class GitanoDownstream(hosts.DownstreamHost):
    @staticmethod
    def check_app_settings(app_settings):
        if app_settings['downstream-visibility'] != 'private':
            raise cliapp.ApplicationError(
                'Cannot create non-private repositories in Gitano')

    def __init__(self, app_settings):
        self._gitano = _LocalTroveGitanoCommand()

    def prepare_repo(self, repo_path, metadata):
        # Create repository on local Trove. If it fails, assume
        # it failed because the repository already existed, and
        # ignore the failure (but log message).

        try:
            self._gitano.create(repo_path)
        except _GitanoCommandFailure as e:
            logging.debug(
                'Ignoring error creating %s on local Trove: %s',
                repo_path, e)
        else:
            logging.info('Created %s on local repo', repo_path)

        try:
            local_config = self._gitano.get_gitano_config(repo_path)
            if 'head' in metadata \
               and metadata['head'] != local_config['project.head']:
                self._gitano.set_gitano_config(repo_path,
                                               'project.head',
                                               metadata['head'])
            if 'description' in metadata \
               and metadata['description'] != \
                   local_config['project.description']:
                self._gitano.set_gitano_config(repo_path,
                                               'project.description',
                                               metadata['description'])
        except _GitanoCommandFailure as e:
            logging.error('ERROR: %s' % str(e))
            # FIXME: We need a good way to report these errors to the
            # user. However, we probably don't want to fail the
            # request, so that's not the way to do this. Needs
            # thinking.


class TroveUpstream(hosts.UpstreamHost):
    def __init__(self, host_info):
        self._host_info = host_info
        self._gitano = _GitanoCommand(host_info['host'],
                                      host_info['protocol'],
                                      host_info['username'],
                                      host_info['password'])

    def list_repos(self):
        ls_output = self._gitano.ls()
        repo_paths = []
        for line in ls_output.splitlines():
            words = line.split(None, 1)
            if words[0].startswith('R') and len(words) == 2:
                repo_paths.append(words[1])
        return repo_paths

    def get_repo_url(self, remote_path):
        vars = dict(self._host_info)
        vars['remote_path'] = remote_path

        patterns = {
            'ssh':  'ssh://git@{host}/{remote_path}',
            'https':'https://{username}:{password}@{host}/git/{remote_path}',
            'http': 'http://{host}/git/{remote_path}',
            }

        return patterns[self._host_info['protocol']].format(**vars)

    def get_repo_metadata(self, repo_path):
        try:
            remote_config = self._gitano.get_gitano_config(repo_path)
            return {
                'head':        remote_config['project.head'],
                'description': remote_config['project.description'],
            }
        except _GitanoCommandFailure as e:
            logging.error('ERROR: %s' % str(e))
            # FIXME: We need a good way to report these errors to the
            # user. However, we probably don't want to fail the
            # request, so that's not the way to do this. Needs
            # thinking.
            return {}