refactor(ssh-wrapper): changed SSHVcsServer to SshVcsServer, updated call_service_api method.

rhodecode/apps/ssh_support/lib/backends/__init__.py

@@ -1,417 +1,387 b''
 # Copyright (C) 2016-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import os
 import re
 import logging
 import datetime
 from sqlalchemy import Table
 
 from rhodecode.lib.api_utils import call_service_api
 from rhodecode.lib.utils2 import AttributeDict
-from rhodecode.lib.vcs.exceptions import ImproperlyConfiguredError
 
 from .hg import MercurialServer
 from .git import GitServer
 from .svn import SubversionServer
 log = logging.getLogger(__name__)
 
 
 class SshWrapper(object):
     hg_cmd_pat  = re.compile(r'^hg\s+\-R\s+(\S+)\s+serve\s+\-\-stdio$')
     git_cmd_pat = re.compile(r'^git-(receive-pack|upload-pack)\s\'[/]?(\S+?)(|\.git)\'$')
     svn_cmd_pat = re.compile(r'^svnserve -t')
 
     def __init__(self, command, connection_info, mode,
                  user, user_id, key_id: int, shell, ini_path: str, settings, env):
         self.command = command
         self.connection_info = connection_info
         self.mode = mode
         self.username = user
         self.user_id = user_id
         self.key_id = key_id
         self.shell = shell
         self.ini_path = ini_path
         self.env = env
         self.settings = settings
         self.server_impl = None
 
     def update_key_access_time(self, key_id):
         from rhodecode.model.meta import raw_query_executor, Base
 
         table = Table('user_ssh_keys', Base.metadata, autoload=False)
         atime = datetime.datetime.utcnow()
         stmt = (
             table.update()
             .where(table.c.ssh_key_id == key_id)
             .values(accessed_on=atime)
             # no MySQL Support for .returning :((
             #.returning(table.c.accessed_on, table.c.ssh_key_fingerprint)
         )
 
         res_count = None
         with raw_query_executor() as session:
             result = session.execute(stmt)
             if result.rowcount:
                 res_count = result.rowcount
 
         if res_count:
             log.debug('Update key id:`%s` access time', key_id)
 
     def get_user(self, user_id):
         user = AttributeDict()
         # lazy load db imports
         from rhodecode.model.db import User
         dbuser = User.get(user_id)
         if not dbuser:
             return None
         user.user_id = dbuser.user_id
         user.username = dbuser.username
         user.auth_user = dbuser.AuthUser()
         return user
 
     def get_connection_info(self):
         """
         connection_info
 
         Identifies the client and server ends of the connection.
         The variable contains four space-separated values: client IP address,
         client port number, server IP address, and server port number.
         """
         conn = dict(
             client_ip=None,
             client_port=None,
             server_ip=None,
             server_port=None,
         )
 
         info = self.connection_info.split(' ')
         if len(info) == 4:
             conn['client_ip'] = info[0]
             conn['client_port'] = info[1]
             conn['server_ip'] = info[2]
             conn['server_port'] = info[3]
 
         return conn
 
     def maybe_translate_repo_uid(self, repo_name):
         _org_name = repo_name
         if _org_name.startswith('_'):
             # remove format of _ID/subrepo
             _org_name = _org_name.split('/', 1)[0]
 
         if repo_name.startswith('_'):
             from rhodecode.model.repo import RepoModel
             org_repo_name = repo_name
             log.debug('translating UID repo %s', org_repo_name)
             by_id_match = RepoModel().get_repo_by_id(repo_name)
             if by_id_match:
                 repo_name = by_id_match.repo_name
                 log.debug('translation of UID repo %s got `%s`', org_repo_name, repo_name)
 
         return repo_name, _org_name
 
     def get_repo_details(self, mode):
         vcs_type = mode if mode in ['svn', 'hg', 'git'] else None
         repo_name = None
 
         hg_match = self.hg_cmd_pat.match(self.command)
         if hg_match is not None:
             vcs_type = 'hg'
             repo_id = hg_match.group(1).strip('/')
             repo_name, org_name = self.maybe_translate_repo_uid(repo_id)
             return vcs_type, repo_name, mode
 
         git_match = self.git_cmd_pat.match(self.command)
         if git_match is not None:
             mode = git_match.group(1)
             vcs_type = 'git'
             repo_id = git_match.group(2).strip('/')
             repo_name, org_name = self.maybe_translate_repo_uid(repo_id)
             return vcs_type, repo_name, mode
 
         svn_match = self.svn_cmd_pat.match(self.command)
         if svn_match is not None:
             vcs_type = 'svn'
             # Repo name should be extracted from the input stream, we're unable to
             # extract it at this point in execution
             return vcs_type, repo_name, mode
 
         return vcs_type, repo_name, mode
 
     def serve(self, vcs, repo, mode, user, permissions, branch_permissions):
         # TODO: remove this once we have .ini defined access path...
         from rhodecode.model.scm import ScmModel
 
         store = ScmModel().repos_path
 
         check_branch_perms = False
         detect_force_push = False
 
         if branch_permissions:
             check_branch_perms = True
             detect_force_push = True
 
         log.debug(
             'VCS detected:`%s` mode: `%s` repo_name: %s, branch_permission_checks:%s',
             vcs, mode, repo, check_branch_perms)
 
         # detect if we have to check branch permissions
         extras = {
             'detect_force_push': detect_force_push,
             'check_branch_perms': check_branch_perms,
             'config': self.ini_path
         }
 
         if vcs == 'hg':
             server = MercurialServer(
                 store=store, ini_path=self.ini_path,
                 repo_name=repo, user=user,
                 user_permissions=permissions, settings=self.settings, env=self.env)
             self.server_impl = server
             return server.run(tunnel_extras=extras)
 
         elif vcs == 'git':
             server = GitServer(
                 store=store, ini_path=self.ini_path,
                 repo_name=repo, repo_mode=mode, user=user,
                 user_permissions=permissions, settings=self.settings, env=self.env)
             self.server_impl = server
             return server.run(tunnel_extras=extras)
 
         elif vcs == 'svn':
             server = SubversionServer(
                 store=store, ini_path=self.ini_path,
                 repo_name=None, user=user,
                 user_permissions=permissions, settings=self.settings, env=self.env)
             self.server_impl = server
             return server.run(tunnel_extras=extras)
 
         else:
             raise Exception(f'Unrecognised VCS: {vcs}')
 
     def wrap(self):
         mode = self.mode
         username = self.username
         user_id = self.user_id
         key_id = self.key_id
         shell = self.shell
 
         scm_detected, scm_repo, scm_mode = self.get_repo_details(mode)
 
         log.debug(
             'Mode: `%s` User: `name:%s : id:%s` Shell: `%s` SSH Command: `\"%s\"` '
             'SCM_DETECTED: `%s` SCM Mode: `%s` SCM Repo: `%s`',
             mode, username, user_id, shell, self.command,
             scm_detected, scm_mode, scm_repo)
 
         log.debug('SSH Connection info %s', self.get_connection_info())
 
         # update last access time for this key
         if key_id:
             self.update_key_access_time(key_id)
 
         if shell and self.command is None:
             log.info('Dropping to shell, no command given and shell is allowed')
             os.execl('/bin/bash', '-l')
             exit_code = 1
 
         elif scm_detected:
             user = self.get_user(user_id)
             if not user:
                 log.warning('User with id %s not found', user_id)
                 exit_code = -1
                 return exit_code
 
             auth_user = user.auth_user
             permissions = auth_user.permissions['repositories']
             repo_branch_permissions = auth_user.get_branch_permissions(scm_repo)
             try:
                 exit_code, is_updated = self.serve(
                     scm_detected, scm_repo, scm_mode, user, permissions,
                     repo_branch_permissions)
             except Exception:
                 log.exception('Error occurred during execution of SshWrapper')
                 exit_code = -1
 
         elif self.command is None and shell is False:
             log.error('No Command given.')
             exit_code = -1
 
         else:
             log.error('Unhandled Command: "%s" Aborting.', self.command)
             exit_code = -1
 
         return exit_code
 
 
 class SshWrapperStandalone(SshWrapper):
     """
     New version of SshWrapper designed to be depended only on service API
     """
     repos_path = None
-    service_api_host: str
-    service_api_token: str
-    api_url: str
-
-    def __init__(self, command, connection_info, mode,
-                 user, user_id, key_id: int, shell, ini_path: str, settings, env):
-
-        # validate our settings for making a standalone calls
-        try:
-            self.service_api_host = settings['app.service_api.host']
-            self.service_api_token = settings['app.service_api.token']
-        except KeyError:
-            raise ImproperlyConfiguredError(
-                "app.service_api.host or app.service_api.token are missing. "
-                "Please ensure that app.service_api.host and app.service_api.token are "
-                "defined inside of .ini configuration file."
-            )
-
-        try:
-            self.api_url = settings['rhodecode.api.url']
-        except KeyError:
-            raise ImproperlyConfiguredError(
-                "rhodecode.api.url is missing. "
-                "Please ensure that rhodecode.api.url is "
-                "defined inside of .ini configuration file."
-            )
-
-        super(SshWrapperStandalone, self).__init__(
-            command, connection_info, mode, user, user_id, key_id, shell, ini_path, settings, env)
 
     @staticmethod
     def parse_user_related_data(user_data):
         user = AttributeDict()
         user.user_id = user_data['user_id']
         user.username = user_data['username']
         user.repo_permissions = user_data['repo_permissions']
         user.branch_permissions = user_data['branch_permissions']
         return user
 
     def wrap(self):
         mode = self.mode
         username = self.username
         user_id = self.user_id
         shell = self.shell
 
         scm_detected, scm_repo, scm_mode = self.get_repo_details(mode)
 
         log.debug(
             'Mode: `%s` User: `name:%s : id:%s` Shell: `%s` SSH Command: `\"%s\"` '
             'SCM_DETECTED: `%s` SCM Mode: `%s` SCM Repo: `%s`',
             mode, username, user_id, shell, self.command,
             scm_detected, scm_mode, scm_repo)
 
         log.debug('SSH Connection info %s', self.get_connection_info())
 
         if shell and self.command is None:
             log.info('Dropping to shell, no command given and shell is allowed')
             os.execl('/bin/bash', '-l')
             exit_code = 1
 
         elif scm_detected:
-            data = call_service_api(self.service_api_host, self.service_api_token, self.api_url, {
+            data = call_service_api(self.settings, {
                 "method": "service_get_data_for_ssh_wrapper",
                 "args": {"user_id": user_id, "repo_name": scm_repo, "key_id": self.key_id}
             })
             user = self.parse_user_related_data(data)
             if not user:
                 log.warning('User with id %s not found', user_id)
                 exit_code = -1
                 return exit_code
             self.repos_path = data['repos_path']
             permissions = user.repo_permissions
             repo_branch_permissions = user.branch_permissions
             try:
                 exit_code, is_updated = self.serve(
                     scm_detected, scm_repo, scm_mode, user, permissions,
                     repo_branch_permissions)
             except Exception:
                 log.exception('Error occurred during execution of SshWrapper')
                 exit_code = -1
 
         elif self.command is None and shell is False:
             log.error('No Command given.')
             exit_code = -1
 
         else:
             log.error('Unhandled Command: "%s" Aborting.', self.command)
             exit_code = -1
 
         return exit_code
 
     def maybe_translate_repo_uid(self, repo_name):
         _org_name = repo_name
         if _org_name.startswith('_'):
             _org_name = _org_name.split('/', 1)[0]
 
         if repo_name.startswith('_'):
             org_repo_name = repo_name
             log.debug('translating UID repo %s', org_repo_name)
-            by_id_match = call_service_api(self.service_api_host, self.service_api_token, self.api_url, {
+            by_id_match = call_service_api(self.settings, {
                 'method': 'service_get_repo_name_by_id',
                 "args": {"repo_id": repo_name}
             })
             if by_id_match:
                 repo_name = by_id_match['repo_name']
                 log.debug('translation of UID repo %s got `%s`', org_repo_name, repo_name)
 
         return repo_name, _org_name
 
     def serve(self, vcs, repo, mode, user, permissions, branch_permissions):
         store = self.repos_path
 
         check_branch_perms = False
         detect_force_push = False
 
         if branch_permissions:
             check_branch_perms = True
             detect_force_push = True
 
         log.debug(
             'VCS detected:`%s` mode: `%s` repo_name: %s, branch_permission_checks:%s',
             vcs, mode, repo, check_branch_perms)
 
         # detect if we have to check branch permissions
         extras = {
             'detect_force_push': detect_force_push,
             'check_branch_perms': check_branch_perms,
             'config': self.ini_path
         }
 
         match vcs:
             case 'hg':
                 server = MercurialServer(
                     store=store, ini_path=self.ini_path,
                     repo_name=repo, user=user,
                     user_permissions=permissions, settings=self.settings, env=self.env)
             case 'git':
                 server = GitServer(
                     store=store, ini_path=self.ini_path,
                     repo_name=repo, repo_mode=mode, user=user,
                     user_permissions=permissions, settings=self.settings, env=self.env)
             case 'svn':
                 server = SubversionServer(
                     store=store, ini_path=self.ini_path,
                     repo_name=None, user=user,
                     user_permissions=permissions, settings=self.settings, env=self.env)
             case _:
                 raise Exception(f'Unrecognised VCS: {vcs}')
         self.server_impl = server
         return server.run(tunnel_extras=extras)

rhodecode/apps/ssh_support/lib/backends/base.py

@@ -1,179 +1,175 b''
 # Copyright (C) 2016-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import os
 import sys
 import logging
 
 from rhodecode.lib.hook_daemon.base import prepare_callback_daemon
 from rhodecode.lib.ext_json import sjson as json
 from rhodecode.lib.vcs.conf import settings as vcs_settings
 from rhodecode.lib.api_utils import call_service_api
 
 log = logging.getLogger(__name__)
 
 
-class SSHVcsServer(object):
+class SshVcsServer(object):
     repo_user_agent = None  # set in child classes
     _path = None  # set executable path for hg/git/svn binary
     backend = None  # set in child classes
     tunnel = None  # subprocess handling tunnel
     settings = None  # parsed settings module
     write_perms = ['repository.admin', 'repository.write']
     read_perms = ['repository.read', 'repository.admin', 'repository.write']
 
     def __init__(self, user, user_permissions, settings, env):
         self.user = user
         self.user_permissions = user_permissions
         self.settings = settings
         self.env = env
         self.stdin = sys.stdin
 
         self.repo_name = None
         self.repo_mode = None
         self.store = ''
         self.ini_path = ''
         self.hooks_protocol = None
 
     def _invalidate_cache(self, repo_name):
         """
         Set's cache for this repository for invalidation on next access
 
         :param repo_name: full repo name, also a cache key
         """
         # Todo: Leave only "celery" case after transition.
         match self.hooks_protocol:
             case 'http':
                 from rhodecode.model.scm import ScmModel
                 ScmModel().mark_for_invalidation(repo_name)
             case 'celery':
-                service_api_host = self.settings['app.service_api.host']
-                service_api_token = self.settings['app.service_api.token']
-                api_url = self.settings['rhodecode.api.url']
-
-                call_service_api(service_api_host, service_api_token, api_url, {
+                call_service_api(self.settings, {
                     "method": "service_mark_for_invalidation",
                     "args": {"repo_name": repo_name}
                 })
 
     def has_write_perm(self):
         permission = self.user_permissions.get(self.repo_name)
         if permission in ['repository.write', 'repository.admin']:
             return True
 
         return False
 
     def _check_permissions(self, action):
         permission = self.user_permissions.get(self.repo_name)
         user_info = f'{self.user["user_id"]}:{self.user["username"]}'
         log.debug('permission for %s on %s are: %s',
                   user_info, self.repo_name, permission)
 
         if not permission:
             log.error('user `%s` permissions to repo:%s are empty. Forbidding access.',
                       user_info, self.repo_name)
             return -2
 
         if action == 'pull':
             if permission in self.read_perms:
                 log.info(
                     'READ Permissions for User "%s" detected to repo "%s"!',
                     user_info, self.repo_name)
                 return 0
         else:
             if permission in self.write_perms:
                 log.info(
                     'WRITE, or Higher Permissions for User "%s" detected to repo "%s"!',
                     user_info, self.repo_name)
                 return 0
 
         log.error('Cannot properly fetch or verify user `%s` permissions. '
                   'Permissions: %s, vcs action: %s',
                   user_info, permission, action)
         return -2
 
     def update_environment(self, action, extras=None):
 
         scm_data = {
             'ip': os.environ['SSH_CLIENT'].split()[0],
             'username': self.user.username,
             'user_id': self.user.user_id,
             'action': action,
             'repository': self.repo_name,
             'scm': self.backend,
             'config': self.ini_path,
             'repo_store': self.store,
             'make_lock': None,
             'locked_by': [None, None],
             'server_url': None,
             'user_agent': f'{self.repo_user_agent}/ssh-user-agent',
             'hooks': ['push', 'pull'],
             'hooks_module': 'rhodecode.lib.hook_daemon.hook_module',
             'is_shadow_repo': False,
             'detect_force_push': False,
             'check_branch_perms': False,
 
             'SSH': True,
             'SSH_PERMISSIONS': self.user_permissions.get(self.repo_name),
         }
         if extras:
             scm_data.update(extras)
         os.putenv("RC_SCM_DATA", json.dumps(scm_data))
         return scm_data
 
     def get_root_store(self):
         root_store = self.store
         if not root_store.endswith('/'):
             # always append trailing slash
             root_store = root_store + '/'
         return root_store
 
     def _handle_tunnel(self, extras):
         # pre-auth
         action = 'pull'
         exit_code = self._check_permissions(action)
         if exit_code:
             return exit_code, False
 
         req = self.env.get('request')
         if req:
             server_url = req.host_url + req.script_name
             extras['server_url'] = server_url
 
         log.debug('Using %s binaries from path %s', self.backend, self._path)
         exit_code = self.tunnel.run(extras)
 
         return exit_code, action == "push"
 
     def run(self, tunnel_extras=None):
         self.hooks_protocol = self.settings['vcs.hooks.protocol']
         tunnel_extras = tunnel_extras or {}
         extras = {}
         extras.update(tunnel_extras)
 
         callback_daemon, extras = prepare_callback_daemon(
             extras, protocol=self.hooks_protocol,
             host=vcs_settings.HOOKS_HOST)
 
         with callback_daemon:
             try:
                 return self._handle_tunnel(extras)
             finally:
                 log.debug('Running cleanup with cache invalidation')
                 if self.repo_name:
                     self._invalidate_cache(self.repo_name)

rhodecode/apps/ssh_support/lib/backends/git.py

@@ -1,86 +1,86 b''
 # Copyright (C) 2016-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import sys
 import logging
 import subprocess
 
 from vcsserver import hooks
-from .base import SSHVcsServer
+from .base import SshVcsServer
 
 log = logging.getLogger(__name__)
 
 
 class GitTunnelWrapper(object):
     process = None
 
     def __init__(self, server):
         self.server = server
         self.stdin = sys.stdin
         self.stdout = sys.stdout
 
     def create_hooks_env(self):
         pass
 
     def command(self):
         root = self.server.get_root_store()
         command = "cd {root}; {git_path} {mode} '{root}{repo_name}'".format(
             root=root, git_path=self.server.git_path,
             mode=self.server.repo_mode, repo_name=self.server.repo_name)
         log.debug("Final CMD: %s", command)
         return command
 
     def run(self, extras):
         action = "push" if self.server.repo_mode == "receive-pack" else "pull"
         exit_code = self.server._check_permissions(action)
         if exit_code:
             return exit_code
 
         scm_extras = self.server.update_environment(action=action, extras=extras)
 
         if action == "pull":
             hook_response = hooks.git_pre_pull(scm_extras)
             pre_pull_messages = hook_response.output
             sys.stdout.write(pre_pull_messages)
 
         self.create_hooks_env()
         result = subprocess.run(self.command(), shell=True)
         result = result.returncode
 
         # Upload-pack == clone
         if action == "pull":
             hook_response = hooks.git_post_pull(scm_extras)
             post_pull_messages = hook_response.output
             sys.stderr.write(post_pull_messages)
         return result
 
 
-class GitServer(SSHVcsServer):
+class GitServer(SshVcsServer):
     backend = 'git'
     repo_user_agent = 'git'
 
     def __init__(self, store, ini_path, repo_name, repo_mode, user, user_permissions, settings, env):
         super().__init__(user, user_permissions, settings, env)
 
         self.store = store
         self.ini_path = ini_path
         self.repo_name = repo_name
         self._path = self.git_path = settings['ssh.executable.git']
 
         self.repo_mode = repo_mode
         self.tunnel = GitTunnelWrapper(server=self)

rhodecode/apps/ssh_support/lib/backends/hg.py

@@ -1,160 +1,157 b''
 # Copyright (C) 2016-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import os
 import sys
 import logging
 import tempfile
 import textwrap
 import collections
 
-from .base import SSHVcsServer
+from .base import SshVcsServer
 
 from rhodecode.lib.api_utils import call_service_api
 
 log = logging.getLogger(__name__)
 
 
 class MercurialTunnelWrapper(object):
     process = None
 
     def __init__(self, server):
         self.server = server
         self.stdin = sys.stdin
         self.stdout = sys.stdout
         self.hooks_env_fd, self.hooks_env_path = tempfile.mkstemp(prefix='hgrc_rhodecode_')
 
     def create_hooks_env(self):
         repo_name = self.server.repo_name
         hg_flags = self.server.config_to_hgrc(repo_name)
 
         content = textwrap.dedent(
             '''
             # RhodeCode SSH hooks version=2.0.0
             {custom}
             '''
         ).format(custom='\n'.join(hg_flags))
 
         root = self.server.get_root_store()
         hgrc_custom = os.path.join(root, repo_name, '.hg', 'hgrc_rhodecode')
         hgrc_main = os.path.join(root, repo_name, '.hg', 'hgrc')
 
         # cleanup custom hgrc file
         if os.path.isfile(hgrc_custom):
             with open(hgrc_custom, 'wb') as f:
                 f.write(b'')
             log.debug('Cleanup custom hgrc file under %s', hgrc_custom)
 
         # write temp
         with os.fdopen(self.hooks_env_fd, 'w') as hooks_env_file:
             hooks_env_file.write(content)
 
         return self.hooks_env_path
 
     def remove_configs(self):
         os.remove(self.hooks_env_path)
 
     def command(self, hgrc_path):
         root = self.server.get_root_store()
 
         command = (
             "cd {root}; HGRCPATH={hgrc} {hg_path} -R {root}{repo_name} "
             "serve --stdio".format(
                 root=root, hg_path=self.server.hg_path,
                 repo_name=self.server.repo_name, hgrc=hgrc_path))
         log.debug("Final CMD: %s", command)
         return command
 
     def run(self, extras):
         # at this point we cannot tell, we do further ACL checks
         # inside the hooks
         action = '?'
         # permissions are check via `pre_push_ssh_auth` hook
         self.server.update_environment(action=action, extras=extras)
         custom_hgrc_file = self.create_hooks_env()
 
         try:
             return os.system(self.command(custom_hgrc_file))
         finally:
             self.remove_configs()
 
 
-class MercurialServer(SSHVcsServer):
+class MercurialServer(SshVcsServer):
     backend = 'hg'
     repo_user_agent = 'mercurial'
     cli_flags = ['phases', 'largefiles', 'extensions', 'experimental', 'hooks']
 
     def __init__(self, store, ini_path, repo_name, user, user_permissions, settings, env):
         super().__init__(user, user_permissions, settings, env)
 
         self.store = store
         self.ini_path = ini_path
         self.repo_name = repo_name
         self._path = self.hg_path = settings['ssh.executable.hg']
         self.tunnel = MercurialTunnelWrapper(server=self)
 
     def config_to_hgrc(self, repo_name):
         # Todo: once transition is done only call to service api should exist
         if self.hooks_protocol == 'celery':
-            service_api_host = self.settings['app.service_api.host']
-            service_api_token = self.settings['app.service_api.token']
-            api_url = self.settings['rhodecode.api.url']
-            data = call_service_api(service_api_host, service_api_token, api_url, {
+            data = call_service_api(self.settings, {
                 "method": "service_config_to_hgrc",
                 "args": {"cli_flags": self.cli_flags, "repo_name": repo_name}
             })
             return data['flags']
         else:
             from rhodecode.model.db import RhodeCodeUi
             from rhodecode.model.settings import VcsSettingsModel
             ui_sections = collections.defaultdict(list)
             ui = VcsSettingsModel(repo=repo_name).get_ui_settings(section=None, key=None)
 
             # write default hooks
             default_hooks = [
                 ('pretxnchangegroup.ssh_auth', 'python:vcsserver.hooks.pre_push_ssh_auth'),
                 ('pretxnchangegroup.ssh', 'python:vcsserver.hooks.pre_push_ssh'),
                 ('changegroup.ssh', 'python:vcsserver.hooks.post_push_ssh'),
 
                 ('preoutgoing.ssh', 'python:vcsserver.hooks.pre_pull_ssh'),
                 ('outgoing.ssh', 'python:vcsserver.hooks.post_pull_ssh'),
             ]
 
             for k, v in default_hooks:
                 ui_sections['hooks'].append((k, v))
 
             for entry in ui:
                 if not entry.active:
                     continue
                 sec = entry.section
                 key = entry.key
 
                 if sec in self.cli_flags:
                     # we want only custom hooks, so we skip builtins
                     if sec == 'hooks' and key in RhodeCodeUi.HOOKS_BUILTIN:
                         continue
 
                     ui_sections[sec].append([key, entry.value])
 
             flags = []
             for _sec, key_val in ui_sections.items():
                 flags.append(' ')
                 flags.append(f'[{_sec}]')
                 for key, val in key_val:
                     flags.append(f'{key}= {val}')
             return flags

rhodecode/apps/ssh_support/lib/backends/svn.py

@@ -1,254 +1,254 b''
 # Copyright (C) 2016-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import os
 import re
 import sys
 import logging
 import signal
 import tempfile
 from subprocess import Popen, PIPE
 import urllib.parse
 
-from .base import SSHVcsServer
+from .base import SshVcsServer
 
 log = logging.getLogger(__name__)
 
 
 class SubversionTunnelWrapper(object):
     process = None
 
     def __init__(self, server):
         self.server = server
         self.timeout = 30
         self.stdin = sys.stdin
         self.stdout = sys.stdout
         self.svn_conf_fd, self.svn_conf_path = tempfile.mkstemp()
         self.hooks_env_fd, self.hooks_env_path = tempfile.mkstemp()
 
         self.read_only = True  # flag that we set to make the hooks readonly
 
     def create_svn_config(self):
         content = (
             '[general]\n'
             'hooks-env = {}\n').format(self.hooks_env_path)
         with os.fdopen(self.svn_conf_fd, 'w') as config_file:
             config_file.write(content)
 
     def create_hooks_env(self):
         content = (
             '[default]\n'
             'LANG = en_US.UTF-8\n')
         if self.read_only:
             content += 'SSH_READ_ONLY = 1\n'
         with os.fdopen(self.hooks_env_fd, 'w') as hooks_env_file:
             hooks_env_file.write(content)
 
     def remove_configs(self):
         os.remove(self.svn_conf_path)
         os.remove(self.hooks_env_path)
 
     def command(self):
         root = self.server.get_root_store()
         username = self.server.user.username
 
         command = [
             self.server.svn_path, '-t',
             '--config-file', self.svn_conf_path,
             '--tunnel-user', username,
             '-r', root]
         log.debug("Final CMD: %s", ' '.join(command))
         return command
 
     def start(self):
         command = self.command()
         self.process = Popen(' '.join(command), stdin=PIPE, shell=True)
 
     def sync(self):
         while self.process.poll() is None:
             next_byte = self.stdin.read(1)
             if not next_byte:
                 break
             self.process.stdin.write(next_byte)
         self.remove_configs()
 
     @property
     def return_code(self):
         return self.process.returncode
 
     def get_first_client_response(self):
         signal.signal(signal.SIGALRM, self.interrupt)
         signal.alarm(self.timeout)
         first_response = self._read_first_client_response()
         signal.alarm(0)
         return (self._parse_first_client_response(first_response)
                 if first_response else None)
 
     def patch_first_client_response(self, response, **kwargs):
         self.create_hooks_env()
         data = response.copy()
         data.update(kwargs)
         data['url'] = self._svn_string(data['url'])
         data['ra_client'] = self._svn_string(data['ra_client'])
         data['client'] = data['client'] or ''
         buffer_ = (
             "( {version} ( {capabilities} ) {url}{ra_client}"
             "( {client}) ) ".format(**data))
         self.process.stdin.write(buffer_)
 
     def fail(self, message):
         print("( failure ( ( 210005 {message} 0: 0 ) ) )".format(
             message=self._svn_string(message)))
         self.remove_configs()
         self.process.kill()
         return 1
 
     def interrupt(self, signum, frame):
         self.fail("Exited by timeout")
 
     def _svn_string(self, str_):
         if not str_:
             return ''
         return f'{len(str_)}:{str_} '
 
     def _read_first_client_response(self):
         buffer_ = ""
         brackets_stack = []
         while True:
             next_byte = self.stdin.read(1)
             buffer_ += next_byte
             if next_byte == "(":
                 brackets_stack.append(next_byte)
             elif next_byte == ")":
                 brackets_stack.pop()
             elif next_byte == " " and not brackets_stack:
                 break
 
         return buffer_
 
     def _parse_first_client_response(self, buffer_):
         """
         According to the Subversion RA protocol, the first request
         should look like:
 
         ( version:number ( cap:word ... ) url:string ? ra-client:string
            ( ? client:string ) )
 
         Please check https://svn.apache.org/repos/asf/subversion/trunk/subversion/libsvn_ra_svn/protocol
         """
         version_re = r'(?P<version>\d+)'
         capabilities_re = r'\(\s(?P<capabilities>[\w\d\-\ ]+)\s\)'
         url_re = r'\d+\:(?P<url>[\W\w]+)'
         ra_client_re = r'(\d+\:(?P<ra_client>[\W\w]+)\s)'
         client_re = r'(\d+\:(?P<client>[\W\w]+)\s)*'
         regex = re.compile(
             r'^\(\s{version}\s{capabilities}\s{url}\s{ra_client}'
             r'\(\s{client}\)\s\)\s*$'.format(
                 version=version_re, capabilities=capabilities_re,
                 url=url_re, ra_client=ra_client_re, client=client_re))
         matcher = regex.match(buffer_)
 
         return matcher.groupdict() if matcher else None
 
     def _match_repo_name(self, url):
         """
         Given an server url, try to match it against ALL known repository names.
         This handles a tricky SVN case for SSH and subdir commits.
         E.g if our repo name is my-svn-repo, a svn commit on file in a subdir would
         result in the url with this subdir added.
         """
         # case 1 direct match, we don't do any "heavy" lookups
         if url in self.server.user_permissions:
             return url
 
         log.debug('Extracting repository name from subdir path %s', url)
         # case 2 we check all permissions, and match closes possible case...
         # NOTE(dan): In this case we only know that url has a subdir parts, it's safe
         # to assume that it will have the repo name as prefix, we ensure the prefix
         # for similar repositories isn't matched by adding a /
         # e.g subgroup/repo-name/ and subgroup/repo-name-1/ would work correct.
         for repo_name in self.server.user_permissions:
             repo_name_prefix = repo_name + '/'
             if url.startswith(repo_name_prefix):
                 log.debug('Found prefix %s match, returning proper repository name',
                           repo_name_prefix)
                 return repo_name
 
         return
 
     def run(self, extras):
         action = 'pull'
         self.create_svn_config()
         self.start()
 
         first_response = self.get_first_client_response()
         if not first_response:
             return self.fail("Repository name cannot be extracted")
 
         url_parts = urllib.parse.urlparse(first_response['url'])
 
         self.server.repo_name = self._match_repo_name(url_parts.path.strip('/'))
 
         exit_code = self.server._check_permissions(action)
         if exit_code:
             return exit_code
 
         # set the readonly flag to False if we have proper permissions
         if self.server.has_write_perm():
             self.read_only = False
         self.server.update_environment(action=action, extras=extras)
 
         self.patch_first_client_response(first_response)
         self.sync()
         return self.return_code
 
 
-class SubversionServer(SSHVcsServer):
+class SubversionServer(SshVcsServer):
     backend = 'svn'
     repo_user_agent = 'svn'
 
     def __init__(self, store, ini_path, repo_name, user, user_permissions, settings, env):
         super().__init__(user, user_permissions, settings, env)
         self.store = store
         self.ini_path = ini_path
         # NOTE(dan): repo_name at this point is empty,
         # this is set later in .run() based from parsed input stream
         self.repo_name = repo_name
         self._path = self.svn_path = settings['ssh.executable.svn']
 
         self.tunnel = SubversionTunnelWrapper(server=self)
 
     def _handle_tunnel(self, extras):
 
         # pre-auth
         action = 'pull'
         # Special case for SVN, we extract repo name at later stage
         # exit_code = self._check_permissions(action)
         # if exit_code:
         #     return exit_code, False
 
         req = self.env['request']
         server_url = req.host_url + req.script_name
         extras['server_url'] = server_url
 
         log.debug('Using %s binaries from path %s', self.backend, self._path)
         exit_code = self.tunnel.run(extras)
 
         return exit_code, action == "push"
 
 

rhodecode/lib/api_utils.py

@@ -1,38 +1,47 b''
 # Copyright (C) 2010-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import urllib.parse
 
 from rhodecode.lib.vcs import CurlSession
 from rhodecode.lib.ext_json import json
+from rhodecode.lib.vcs.exceptions import ImproperlyConfiguredError
 
 
-def call_service_api(service_api_host, service_api_token, api_url, payload):
-
+def call_service_api(settings, payload):
+    try:
+        api_host = settings['app.service_api.host']
+        api_token = settings['app.service_api.token']
+        api_url = settings['rhodecode.api.url']
+    except KeyError as exc:
+        raise ImproperlyConfiguredError(
+            f"{str(exc)} is missing. "
+            "Please ensure that app.service_api.host, app.service_api.token and rhodecode.api.url are "
+            "defined inside of .ini configuration file."
+        )
     payload.update({
         'id': 'service',
-        'auth_token': service_api_token
+        'auth_token': api_token
     })
-
-    service_api_url = urllib.parse.urljoin(service_api_host, api_url)
+    service_api_url = urllib.parse.urljoin(api_host, api_url)
     response = CurlSession().post(service_api_url, json.dumps(payload))
 
     if response.status_code != 200:
         raise Exception(f"Service API at {service_api_url} responded with error: {response.status_code}")
 
     return json.loads(response.content)['result']