# 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 {}