core: run ruff format
This commit is contained in:
RhodeCode Admin
parent
7130effb58
commit
166db21812
821 changed files with 82905 additions and 84121 deletions
|
|
@ -31,18 +31,18 @@ def _sanitize_settings_and_apply_defaults(settings):
|
|||
"""
|
||||
settings_maker = SettingsMaker(settings)
|
||||
|
||||
settings_maker.make_setting(config_keys.generate_authorized_keyfile, False, parser='bool')
|
||||
settings_maker.make_setting(config_keys.wrapper_allow_shell, False, parser='bool')
|
||||
settings_maker.make_setting(config_keys.enable_debug_logging, False, parser='bool')
|
||||
settings_maker.make_setting(config_keys.ssh_key_generator_enabled, True, parser='bool')
|
||||
settings_maker.make_setting(config_keys.generate_authorized_keyfile, False, parser="bool")
|
||||
settings_maker.make_setting(config_keys.wrapper_allow_shell, False, parser="bool")
|
||||
settings_maker.make_setting(config_keys.enable_debug_logging, False, parser="bool")
|
||||
settings_maker.make_setting(config_keys.ssh_key_generator_enabled, True, parser="bool")
|
||||
|
||||
settings_maker.make_setting(config_keys.authorized_keys_file_path, '~/.ssh/authorized_keys_rhodecode')
|
||||
settings_maker.make_setting(config_keys.wrapper_cmd, '/usr/local/bin/rhodecode_bin/bin/rc-ssh-wrapper-v2')
|
||||
settings_maker.make_setting(config_keys.authorized_keys_line_ssh_opts, '')
|
||||
settings_maker.make_setting(config_keys.authorized_keys_file_path, "~/.ssh/authorized_keys_rhodecode")
|
||||
settings_maker.make_setting(config_keys.wrapper_cmd, "/usr/local/bin/rhodecode_bin/bin/rc-ssh-wrapper-v2")
|
||||
settings_maker.make_setting(config_keys.authorized_keys_line_ssh_opts, "")
|
||||
|
||||
settings_maker.make_setting(config_keys.ssh_hg_bin, '/usr/local/bin/rhodecode_bin/vcs_bin/hg')
|
||||
settings_maker.make_setting(config_keys.ssh_git_bin, '/usr/local/bin/rhodecode_bin/vcs_bin/git')
|
||||
settings_maker.make_setting(config_keys.ssh_svn_bin, '/usr/local/bin/rhodecode_bin/vcs_bin/svnserve')
|
||||
settings_maker.make_setting(config_keys.ssh_hg_bin, "/usr/local/bin/rhodecode_bin/vcs_bin/hg")
|
||||
settings_maker.make_setting(config_keys.ssh_git_bin, "/usr/local/bin/rhodecode_bin/vcs_bin/git")
|
||||
settings_maker.make_setting(config_keys.ssh_svn_bin, "/usr/local/bin/rhodecode_bin/vcs_bin/svnserve")
|
||||
|
||||
settings_maker.env_expand()
|
||||
|
||||
|
|
@ -56,5 +56,5 @@ def includeme(config):
|
|||
# lazy import here for faster code reading... via sshwrapper-v2 mode
|
||||
from .subscribers import generate_ssh_authorized_keys_file_subscriber
|
||||
from .events import SshKeyFileChangeEvent
|
||||
config.add_subscriber(
|
||||
generate_ssh_authorized_keys_file_subscriber, SshKeyFileChangeEvent)
|
||||
|
||||
config.add_subscriber(generate_ssh_authorized_keys_file_subscriber, SshKeyFileChangeEvent)
|
||||
|
|
|
|||
|
|
@ -19,14 +19,14 @@
|
|||
|
||||
# Definition of setting keys used to configure this module. Defined here to
|
||||
# avoid repetition of keys throughout the module.
|
||||
generate_authorized_keyfile = 'ssh.generate_authorized_keyfile'
|
||||
authorized_keys_file_path = 'ssh.authorized_keys_file_path'
|
||||
authorized_keys_line_ssh_opts = 'ssh.authorized_keys_ssh_opts'
|
||||
ssh_key_generator_enabled = 'ssh.enable_ui_key_generator'
|
||||
wrapper_cmd = 'ssh.wrapper_cmd.v2'
|
||||
wrapper_allow_shell = 'ssh.wrapper_cmd_allow_shell'
|
||||
enable_debug_logging = 'ssh.enable_debug_logging'
|
||||
generate_authorized_keyfile = "ssh.generate_authorized_keyfile"
|
||||
authorized_keys_file_path = "ssh.authorized_keys_file_path"
|
||||
authorized_keys_line_ssh_opts = "ssh.authorized_keys_ssh_opts"
|
||||
ssh_key_generator_enabled = "ssh.enable_ui_key_generator"
|
||||
wrapper_cmd = "ssh.wrapper_cmd.v2"
|
||||
wrapper_allow_shell = "ssh.wrapper_cmd_allow_shell"
|
||||
enable_debug_logging = "ssh.enable_debug_logging"
|
||||
|
||||
ssh_hg_bin = 'ssh.executable.hg'
|
||||
ssh_git_bin = 'ssh.executable.git'
|
||||
ssh_svn_bin = 'ssh.executable.svn'
|
||||
ssh_hg_bin = "ssh.executable.hg"
|
||||
ssh_git_bin = "ssh.executable.git"
|
||||
ssh_svn_bin = "ssh.executable.svn"
|
||||
|
|
|
|||
|
|
@ -25,5 +25,6 @@ class SshKeyFileChangeEvent(RhodecodeEvent):
|
|||
"""
|
||||
This event will be triggered on every modification of the stored SSH keys
|
||||
"""
|
||||
name = 'rhodecode-ssh-key-file-change'
|
||||
display_name = _('RhodeCode SSH Key files changed.')
|
||||
|
||||
name = "rhodecode-ssh-key-file-change"
|
||||
display_name = _("RhodeCode SSH Key files changed.")
|
||||
|
|
|
|||
|
|
@ -34,8 +34,8 @@ class SshVcsServer(object):
|
|||
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']
|
||||
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
|
||||
|
|
@ -46,8 +46,8 @@ class SshVcsServer(object):
|
|||
|
||||
self.repo_name = None
|
||||
self.repo_mode = None
|
||||
self.store = ''
|
||||
self.ini_path = ''
|
||||
self.store = ""
|
||||
self.ini_path = ""
|
||||
self.hooks_protocol = None
|
||||
|
||||
def _invalidate_cache(self, repo_name):
|
||||
|
|
@ -58,74 +58,69 @@ class SshVcsServer(object):
|
|||
"""
|
||||
# Todo: Leave only "celery" case after transition.
|
||||
match self.hooks_protocol:
|
||||
case 'http':
|
||||
case "http":
|
||||
from rhodecode.model.scm import ScmModel
|
||||
|
||||
ScmModel().mark_for_invalidation(repo_name)
|
||||
case 'celery':
|
||||
call_service_api(self.settings, {
|
||||
"method": "service_mark_for_invalidation",
|
||||
"args": {"repo_name": repo_name}
|
||||
})
|
||||
case "celery":
|
||||
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']:
|
||||
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)
|
||||
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)
|
||||
log.error("user `%s` permissions to repo:%s are empty. Forbidding access.", user_info, self.repo_name)
|
||||
return -2
|
||||
|
||||
if action == 'pull':
|
||||
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)
|
||||
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)
|
||||
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)
|
||||
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),
|
||||
"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)
|
||||
|
|
@ -134,30 +129,30 @@ class SshVcsServer(object):
|
|||
|
||||
def get_root_store(self):
|
||||
root_store = self.store
|
||||
if not root_store.endswith('/'):
|
||||
if not root_store.endswith("/"):
|
||||
# always append trailing slash
|
||||
root_store = root_store + '/'
|
||||
root_store = root_store + "/"
|
||||
return root_store
|
||||
|
||||
def _handle_tunnel(self, extras):
|
||||
# pre-auth
|
||||
action = 'pull'
|
||||
action = "pull"
|
||||
exit_code = self._check_permissions(action)
|
||||
if exit_code:
|
||||
return exit_code, False
|
||||
|
||||
req = self.env.get('request')
|
||||
req = self.env.get("request")
|
||||
if req:
|
||||
server_url = req.host_url + req.script_name
|
||||
extras['server_url'] = server_url
|
||||
extras["server_url"] = server_url
|
||||
|
||||
log.debug('Using %s binaries from path %s', self.backend, self._path)
|
||||
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.v2']
|
||||
self.hooks_protocol = self.settings["vcs.hooks.protocol.v2"]
|
||||
tunnel_extras = tunnel_extras or {}
|
||||
extras = {}
|
||||
extras.update(tunnel_extras)
|
||||
|
|
@ -168,6 +163,6 @@ class SshVcsServer(object):
|
|||
try:
|
||||
return self._handle_tunnel(extras)
|
||||
finally:
|
||||
log.debug('Running cleanup with cache invalidation')
|
||||
log.debug("Running cleanup with cache invalidation")
|
||||
if self.repo_name:
|
||||
self._invalidate_cache(self.repo_name)
|
||||
|
|
|
|||
|
|
@ -40,8 +40,8 @@ class GitTunnelWrapper(object):
|
|||
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)
|
||||
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
|
||||
|
||||
|
|
@ -71,8 +71,8 @@ class GitTunnelWrapper(object):
|
|||
|
||||
|
||||
class GitServer(SshVcsServer):
|
||||
backend = 'git'
|
||||
repo_user_agent = 'git'
|
||||
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)
|
||||
|
|
@ -80,7 +80,7 @@ class GitServer(SshVcsServer):
|
|||
self.store = store
|
||||
self.ini_path = ini_path
|
||||
self.repo_name = repo_name
|
||||
self._path = self.git_path = settings['ssh.executable.git']
|
||||
self._path = self.git_path = settings["ssh.executable.git"]
|
||||
|
||||
self.repo_mode = repo_mode
|
||||
self.tunnel = GitTunnelWrapper(server=self)
|
||||
|
|
|
|||
|
|
@ -37,31 +37,31 @@ class MercurialTunnelWrapper(object):
|
|||
self.server = server
|
||||
self.stdin = sys.stdin
|
||||
self.stdout = sys.stdout
|
||||
self.hooks_env_fd, self.hooks_env_path = tempfile.mkstemp(prefix='hgrc_rhodecode_')
|
||||
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))
|
||||
"""
|
||||
).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')
|
||||
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)
|
||||
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:
|
||||
with os.fdopen(self.hooks_env_fd, "w") as hooks_env_file:
|
||||
hooks_env_file.write(content)
|
||||
|
||||
return self.hooks_env_path
|
||||
|
|
@ -72,18 +72,16 @@ class MercurialTunnelWrapper(object):
|
|||
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))
|
||||
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 = '?'
|
||||
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()
|
||||
|
|
@ -95,9 +93,9 @@ class MercurialTunnelWrapper(object):
|
|||
|
||||
|
||||
class MercurialServer(SshVcsServer):
|
||||
backend = 'hg'
|
||||
repo_user_agent = 'mercurial'
|
||||
cli_flags = ['phases', 'largefiles', 'extensions', 'experimental', 'hooks']
|
||||
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)
|
||||
|
|
@ -105,35 +103,35 @@ class MercurialServer(SshVcsServer):
|
|||
self.store = store
|
||||
self.ini_path = ini_path
|
||||
self.repo_name = repo_name
|
||||
self._path = self.hg_path = settings['ssh.executable.hg']
|
||||
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':
|
||||
data = call_service_api(self.settings, {
|
||||
"method": "service_config_to_hgrc",
|
||||
"args": {"cli_flags": self.cli_flags, "repo_name": repo_name}
|
||||
})
|
||||
return data['flags']
|
||||
if self.hooks_protocol == "celery":
|
||||
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'),
|
||||
("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))
|
||||
ui_sections["hooks"].append((k, v))
|
||||
|
||||
for entry in ui:
|
||||
if not entry.active:
|
||||
|
|
@ -143,15 +141,15 @@ class MercurialServer(SshVcsServer):
|
|||
|
||||
if sec in self.cli_flags:
|
||||
# we want only custom hooks, so we skip builtins
|
||||
if sec == 'hooks' and key in RhodeCodeUi.HOOKS_BUILTIN:
|
||||
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}]')
|
||||
flags.append(" ")
|
||||
flags.append(f"[{_sec}]")
|
||||
for key, val in key_val:
|
||||
flags.append(f'{key}= {val}')
|
||||
flags.append(f"{key}= {val}")
|
||||
return flags
|
||||
|
|
|
|||
|
|
@ -45,19 +45,15 @@ class SubversionTunnelWrapper(object):
|
|||
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:
|
||||
content = ("[general]\nhooks-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')
|
||||
content = "[default]\nLANG = 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:
|
||||
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):
|
||||
|
|
@ -69,16 +65,21 @@ class SubversionTunnelWrapper(object):
|
|||
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))
|
||||
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)
|
||||
self.process = Popen(" ".join(command), stdin=PIPE, shell=True)
|
||||
|
||||
def sync(self):
|
||||
while self.process.poll() is None:
|
||||
|
|
@ -97,26 +98,19 @@ class SubversionTunnelWrapper(object):
|
|||
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)
|
||||
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()
|
||||
|
||||
version = response['version']
|
||||
capabilities = response['capabilities']
|
||||
client = response['client'] or b''
|
||||
version = response["version"]
|
||||
capabilities = response["capabilities"]
|
||||
client = response["client"] or b""
|
||||
|
||||
url = self._svn_bytes(response['url'])
|
||||
ra_client = self._svn_bytes(response['ra_client'])
|
||||
url = self._svn_bytes(response["url"])
|
||||
ra_client = self._svn_bytes(response["ra_client"])
|
||||
|
||||
buffer_ = b"( %b ( %b ) %b%b( %b) ) " % (
|
||||
version,
|
||||
capabilities,
|
||||
url,
|
||||
ra_client,
|
||||
client
|
||||
)
|
||||
buffer_ = b"( %b ( %b ) %b%b( %b) ) " % (version, capabilities, url, ra_client, client)
|
||||
self.process.stdin.write(buffer_)
|
||||
|
||||
def fail(self, message):
|
||||
|
|
@ -132,9 +126,9 @@ class SubversionTunnelWrapper(object):
|
|||
|
||||
def _svn_bytes(self, bytes_: bytes) -> bytes:
|
||||
if not bytes_:
|
||||
return b''
|
||||
return b""
|
||||
|
||||
return f'{len(bytes_)}:'.encode() + bytes_ + b' '
|
||||
return f"{len(bytes_)}:".encode() + bytes_ + b" "
|
||||
|
||||
def _read_first_client_response(self):
|
||||
buffer_ = b""
|
||||
|
|
@ -161,19 +155,14 @@ class SubversionTunnelWrapper(object):
|
|||
|
||||
Please check https://svn.apache.org/repos/asf/subversion/trunk/subversion/libsvn_ra_svn/protocol
|
||||
"""
|
||||
version_re = br'(?P<version>\d+)'
|
||||
capabilities_re = br'\(\s(?P<capabilities>[\w\d\-\ ]+)\s\)'
|
||||
url_re = br'\d+\:(?P<url>[\W\w]+)'
|
||||
ra_client_re = br'(\d+\:(?P<ra_client>[\W\w]+)\s)'
|
||||
client_re = br'(\d+\:(?P<client>[\W\w]+)\s)*'
|
||||
version_re = rb"(?P<version>\d+)"
|
||||
capabilities_re = rb"\(\s(?P<capabilities>[\w\d\-\ ]+)\s\)"
|
||||
url_re = rb"\d+\:(?P<url>[\W\w]+)"
|
||||
ra_client_re = rb"(\d+\:(?P<ra_client>[\W\w]+)\s)"
|
||||
client_re = rb"(\d+\:(?P<client>[\W\w]+)\s)*"
|
||||
regex = re.compile(
|
||||
br'^\(\s%b\s%b\s%b\s%b'
|
||||
br'\(\s%b\)\s\)\s*$' % (
|
||||
version_re,
|
||||
capabilities_re,
|
||||
url_re,
|
||||
ra_client_re,
|
||||
client_re)
|
||||
rb"^\(\s%b\s%b\s%b\s%b"
|
||||
rb"\(\s%b\)\s\)\s*$" % (version_re, capabilities_re, url_re, ra_client_re, client_re)
|
||||
)
|
||||
matcher = regex.match(buffer_)
|
||||
|
||||
|
|
@ -190,23 +179,22 @@ class SubversionTunnelWrapper(object):
|
|||
if url in self.server.user_permissions:
|
||||
return url
|
||||
|
||||
log.debug('Extracting repository name from subdir path %s', 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 + '/'
|
||||
repo_name_prefix = repo_name + "/"
|
||||
if url.startswith(repo_name_prefix):
|
||||
log.debug('Found prefix %s match, returning proper repository name',
|
||||
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'
|
||||
action = "pull"
|
||||
self.create_svn_config()
|
||||
self.start()
|
||||
|
||||
|
|
@ -214,9 +202,9 @@ class SubversionTunnelWrapper(object):
|
|||
if not first_response:
|
||||
return self.fail(b"Repository name cannot be extracted")
|
||||
|
||||
url_parts = urllib.parse.urlparse(first_response['url'])
|
||||
url_parts = urllib.parse.urlparse(first_response["url"])
|
||||
|
||||
self.server.repo_name = self._match_repo_name(safe_str(url_parts.path).strip('/'))
|
||||
self.server.repo_name = self._match_repo_name(safe_str(url_parts.path).strip("/"))
|
||||
|
||||
exit_code = self.server._check_permissions(action)
|
||||
if exit_code:
|
||||
|
|
@ -233,8 +221,8 @@ class SubversionTunnelWrapper(object):
|
|||
|
||||
|
||||
class SubversionServer(SshVcsServer):
|
||||
backend = 'svn'
|
||||
repo_user_agent = 'svn'
|
||||
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)
|
||||
|
|
@ -243,27 +231,24 @@ class SubversionServer(SshVcsServer):
|
|||
# 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._path = self.svn_path = settings["ssh.executable.svn"]
|
||||
|
||||
self.tunnel = SubversionTunnelWrapper(server=self)
|
||||
|
||||
def _handle_tunnel(self, extras):
|
||||
|
||||
# pre-auth
|
||||
action = 'pull'
|
||||
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.get('request')
|
||||
req = self.env.get("request")
|
||||
if req:
|
||||
server_url = req.host_url + req.script_name
|
||||
extras['server_url'] = server_url
|
||||
extras["server_url"] = server_url
|
||||
|
||||
log.debug('Using %s binaries from path %s', self.backend, self._path)
|
||||
log.debug("Using %s binaries from path %s", self.backend, self._path)
|
||||
exit_code = self.tunnel.run(extras)
|
||||
|
||||
return exit_code, action == "push"
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -32,41 +32,45 @@ log = logging.getLogger(__name__)
|
|||
|
||||
|
||||
@click.command()
|
||||
@click.argument('ini_path', type=click.Path(exists=True))
|
||||
@click.argument("ini_path", type=click.Path(exists=True))
|
||||
@click.option(
|
||||
'--mode', '-m', required=False, default='auto',
|
||||
type=click.Choice(['auto', 'vcs', 'git', 'hg', 'svn', 'test']),
|
||||
help='mode of operation')
|
||||
@click.option('--user', help='Username for which the command will be executed')
|
||||
@click.option('--user-id', help='User ID for which the command will be executed')
|
||||
@click.option('--key-id', help='ID of the key from the database')
|
||||
@click.option('--shell', '-s', is_flag=True, help='Allow Shell')
|
||||
@click.option('--debug', is_flag=True, help='Enabled detailed output logging')
|
||||
"--mode",
|
||||
"-m",
|
||||
required=False,
|
||||
default="auto",
|
||||
type=click.Choice(["auto", "vcs", "git", "hg", "svn", "test"]),
|
||||
help="mode of operation",
|
||||
)
|
||||
@click.option("--user", help="Username for which the command will be executed")
|
||||
@click.option("--user-id", help="User ID for which the command will be executed")
|
||||
@click.option("--key-id", help="ID of the key from the database")
|
||||
@click.option("--shell", "-s", is_flag=True, help="Allow Shell")
|
||||
@click.option("--debug", is_flag=True, help="Enabled detailed output logging")
|
||||
def main(ini_path, mode, user, user_id, key_id, shell, debug):
|
||||
setup_custom_logging(ini_path, debug)
|
||||
|
||||
command = os.environ.get('SSH_ORIGINAL_COMMAND', '')
|
||||
if not command and mode not in ['test']:
|
||||
command = os.environ.get("SSH_ORIGINAL_COMMAND", "")
|
||||
if not command and mode not in ["test"]:
|
||||
raise ValueError(
|
||||
'Unable to fetch SSH_ORIGINAL_COMMAND from environment.'
|
||||
'Please make sure this is set and available during execution '
|
||||
'of this script.')
|
||||
connection_info = os.environ.get('SSH_CONNECTION', '')
|
||||
"Unable to fetch SSH_ORIGINAL_COMMAND from environment."
|
||||
"Please make sure this is set and available during execution "
|
||||
"of this script."
|
||||
)
|
||||
connection_info = os.environ.get("SSH_CONNECTION", "")
|
||||
time_start = time.time()
|
||||
with bootstrap(ini_path, env={'RC_CMD_SSH_WRAPPER': '1'}) as env:
|
||||
settings = env['registry'].settings
|
||||
with bootstrap(ini_path, env={"RC_CMD_SSH_WRAPPER": "1"}) as env:
|
||||
settings = env["registry"].settings
|
||||
statsd = StatsdClient.statsd
|
||||
try:
|
||||
ssh_wrapper = SshWrapper(
|
||||
command, connection_info, mode,
|
||||
user, user_id, key_id, shell, ini_path, settings, env)
|
||||
command, connection_info, mode, user, user_id, key_id, shell, ini_path, settings, env
|
||||
)
|
||||
except Exception:
|
||||
log.exception('Failed to execute SshWrapper')
|
||||
log.exception("Failed to execute SshWrapper")
|
||||
sys.exit(-5)
|
||||
return_code = ssh_wrapper.wrap()
|
||||
operation_took = time.time() - time_start
|
||||
if statsd:
|
||||
operation_took_ms = round(1000.0 * operation_took)
|
||||
statsd.timing("rhodecode_ssh_wrapper_timing.histogram", operation_took_ms,
|
||||
use_decimals=False)
|
||||
statsd.timing("rhodecode_ssh_wrapper_timing.histogram", operation_took_ms, use_decimals=False)
|
||||
sys.exit(return_code)
|
||||
|
|
|
|||
|
|
@ -44,55 +44,62 @@ log = logging.getLogger(__name__)
|
|||
|
||||
|
||||
@click.command()
|
||||
@click.argument('ini_path', type=click.Path(exists=True))
|
||||
@click.argument("ini_path", type=click.Path(exists=True))
|
||||
@click.option(
|
||||
'--mode', '-m', required=False, default='auto',
|
||||
type=click.Choice(['auto', 'vcs', 'git', 'hg', 'svn', 'test']),
|
||||
help='mode of operation')
|
||||
@click.option('--user', help='Username for which the command will be executed')
|
||||
@click.option('--user-id', help='User ID for which the command will be executed')
|
||||
@click.option('--key-id', help='ID of the key from the database')
|
||||
@click.option('--shell', '-s', is_flag=True, help='Allow Shell')
|
||||
@click.option('--debug', is_flag=True, help='Enabled detailed output logging')
|
||||
"--mode",
|
||||
"-m",
|
||||
required=False,
|
||||
default="auto",
|
||||
type=click.Choice(["auto", "vcs", "git", "hg", "svn", "test"]),
|
||||
help="mode of operation",
|
||||
)
|
||||
@click.option("--user", help="Username for which the command will be executed")
|
||||
@click.option("--user-id", help="User ID for which the command will be executed")
|
||||
@click.option("--key-id", help="ID of the key from the database")
|
||||
@click.option("--shell", "-s", is_flag=True, help="Allow Shell")
|
||||
@click.option("--debug", is_flag=True, help="Enabled detailed output logging")
|
||||
def main(ini_path, mode, user, user_id, key_id, shell, debug):
|
||||
|
||||
time_start = time.time()
|
||||
setup_custom_logging(ini_path, debug)
|
||||
|
||||
command = os.environ.get('SSH_ORIGINAL_COMMAND', '')
|
||||
if not command and mode not in ['test']:
|
||||
command = os.environ.get("SSH_ORIGINAL_COMMAND", "")
|
||||
if not command and mode not in ["test"]:
|
||||
raise ValueError(
|
||||
'Unable to fetch SSH_ORIGINAL_COMMAND from environment.'
|
||||
'Please make sure this is set and available during execution '
|
||||
'of this script.')
|
||||
"Unable to fetch SSH_ORIGINAL_COMMAND from environment."
|
||||
"Please make sure this is set and available during execution "
|
||||
"of this script."
|
||||
)
|
||||
|
||||
# initialize settings and get defaults
|
||||
settings = get_app_config_lightweight(ini_path)
|
||||
settings = sanitize_settings_and_apply_defaults({'__file__': ini_path}, settings)
|
||||
settings = sanitize_settings_and_apply_defaults({"__file__": ini_path}, settings)
|
||||
|
||||
# init and bootstrap StatsdClient
|
||||
StatsdClient.setup(settings)
|
||||
statsd = StatsdClient.statsd
|
||||
|
||||
try:
|
||||
connection_info = os.environ.get('SSH_CONNECTION', '')
|
||||
request = Request.blank('/', base_url=settings['app.base_url'])
|
||||
request.user = AttributeDict({'username': user,
|
||||
'user_id': user_id,
|
||||
'ip_addr': connection_info.split(' ')[0] if connection_info else None})
|
||||
env = {'RC_CMD_SSH_WRAPPER': '1', 'request': request}
|
||||
connection_info = os.environ.get("SSH_CONNECTION", "")
|
||||
request = Request.blank("/", base_url=settings["app.base_url"])
|
||||
request.user = AttributeDict(
|
||||
{
|
||||
"username": user,
|
||||
"user_id": user_id,
|
||||
"ip_addr": connection_info.split(" ")[0] if connection_info else None,
|
||||
}
|
||||
)
|
||||
env = {"RC_CMD_SSH_WRAPPER": "1", "request": request}
|
||||
ssh_wrapper = SshWrapperStandalone(
|
||||
command, connection_info, mode,
|
||||
user, user_id, key_id, shell, ini_path, settings, env)
|
||||
command, connection_info, mode, user, user_id, key_id, shell, ini_path, settings, env
|
||||
)
|
||||
except Exception:
|
||||
log.exception('Failed to execute SshWrapper')
|
||||
log.exception("Failed to execute SshWrapper")
|
||||
sys.exit(-5)
|
||||
|
||||
return_code = ssh_wrapper.wrap()
|
||||
operation_took = time.time() - time_start
|
||||
if statsd:
|
||||
operation_took_ms = round(1000.0 * operation_took)
|
||||
statsd.timing("rhodecode_ssh_wrapper_timing.histogram", operation_took_ms,
|
||||
use_decimals=False)
|
||||
statsd.timing("rhodecode_ssh_wrapper_timing.histogram", operation_took_ms, use_decimals=False)
|
||||
|
||||
sys.exit(return_code)
|
||||
|
|
|
|||
|
|
@ -22,13 +22,14 @@ import logging
|
|||
def setup_custom_logging(ini_path, debug):
|
||||
if debug:
|
||||
from pyramid.paster import setup_logging # Lazy import
|
||||
|
||||
# enabled rhodecode.ini controlled logging setup
|
||||
setup_logging(ini_path)
|
||||
else:
|
||||
# configure logging in a mode that doesn't print anything.
|
||||
# in case of regularly configured logging it gets printed out back
|
||||
# to the client doing an SSH command.
|
||||
logger = logging.getLogger('')
|
||||
logger = logging.getLogger("")
|
||||
null = logging.NullHandler()
|
||||
# add the handler to the root logger
|
||||
logger.handlers = [null]
|
||||
|
|
|
|||
|
|
@ -27,23 +27,20 @@ from rhodecode.lib.utils2 import AttributeDict
|
|||
@pytest.fixture()
|
||||
def dummy_conf_file(tmpdir):
|
||||
conf = configparser.ConfigParser()
|
||||
conf.add_section('app:main')
|
||||
conf.set('app:main', 'ssh.executable.hg', '/usr/bin/hg')
|
||||
conf.set('app:main', 'ssh.executable.git', '/usr/bin/git')
|
||||
conf.set('app:main', 'ssh.executable.svn', '/usr/bin/svnserve')
|
||||
conf.add_section("app:main")
|
||||
conf.set("app:main", "ssh.executable.hg", "/usr/bin/hg")
|
||||
conf.set("app:main", "ssh.executable.git", "/usr/bin/git")
|
||||
conf.set("app:main", "ssh.executable.svn", "/usr/bin/svnserve")
|
||||
|
||||
f_path = os.path.join(str(tmpdir), 'ssh_wrapper_test.ini')
|
||||
with open(f_path, 'wt') as f:
|
||||
f_path = os.path.join(str(tmpdir), "ssh_wrapper_test.ini")
|
||||
with open(f_path, "wt") as f:
|
||||
conf.write(f)
|
||||
|
||||
return os.path.join(f_path)
|
||||
|
||||
|
||||
def plain_dummy_env():
|
||||
return {
|
||||
'request':
|
||||
AttributeDict(host_url='http://localhost', script_name='/')
|
||||
}
|
||||
return {"request": AttributeDict(host_url="http://localhost", script_name="/")}
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
|
|
@ -52,10 +49,7 @@ def dummy_env():
|
|||
|
||||
|
||||
def plain_dummy_user():
|
||||
return AttributeDict(
|
||||
user_id=1,
|
||||
username='test_user'
|
||||
)
|
||||
return AttributeDict(user_id=1, username="test_user")
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
|
|
@ -65,7 +59,16 @@ def dummy_user():
|
|||
|
||||
@pytest.fixture()
|
||||
def ssh_wrapper(app, dummy_conf_file, dummy_env):
|
||||
conn_info = '127.0.0.1 22 10.0.0.1 443'
|
||||
conn_info = "127.0.0.1 22 10.0.0.1 443"
|
||||
return SshWrapper(
|
||||
'random command', conn_info, 'auto', 'admin', '1', key_id='1',
|
||||
shell=False, ini_path=dummy_conf_file, settings={}, env=dummy_env)
|
||||
"random command",
|
||||
conn_info,
|
||||
"auto",
|
||||
"admin",
|
||||
"1",
|
||||
key_id="1",
|
||||
shell=False,
|
||||
ini_path=dummy_conf_file,
|
||||
settings={},
|
||||
env=dummy_env,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -27,19 +27,19 @@ from rhodecode.lib.ext_json import json
|
|||
|
||||
|
||||
class GitServerCreator(object):
|
||||
root = '/tmp/repo/path/'
|
||||
git_path = '/usr/local/bin/git'
|
||||
root = "/tmp/repo/path/"
|
||||
git_path = "/usr/local/bin/git"
|
||||
config_data = {
|
||||
'app:main': {
|
||||
'ssh.executable.git': git_path,
|
||||
'vcs.hooks.protocol.v2': 'celery',
|
||||
'app.service_api.host': 'http://localhost',
|
||||
'app.service_api.token': 'secret4',
|
||||
'rhodecode.api.url': '/_admin/api',
|
||||
"app:main": {
|
||||
"ssh.executable.git": git_path,
|
||||
"vcs.hooks.protocol.v2": "celery",
|
||||
"app.service_api.host": "http://localhost",
|
||||
"app.service_api.token": "secret4",
|
||||
"rhodecode.api.url": "/_admin/api",
|
||||
}
|
||||
}
|
||||
repo_name = 'test_git'
|
||||
repo_mode = 'receive-pack'
|
||||
repo_name = "test_git"
|
||||
repo_mode = "receive-pack"
|
||||
user = plain_dummy_user()
|
||||
|
||||
def __init__(self, service_api_url, ini_file):
|
||||
|
|
@ -47,28 +47,26 @@ class GitServerCreator(object):
|
|||
self.ini_file = ini_file
|
||||
|
||||
def create(self, **kwargs):
|
||||
self.config_data['app:main']['app.service_api.host'] = self.service_api_url
|
||||
self.config_data["app:main"]["app.service_api.host"] = self.service_api_url
|
||||
parameters = {
|
||||
'store': self.root,
|
||||
'ini_path': self.ini_file,
|
||||
'user': self.user,
|
||||
'repo_name': self.repo_name,
|
||||
'repo_mode': self.repo_mode,
|
||||
'user_permissions': {
|
||||
self.repo_name: 'repository.admin'
|
||||
},
|
||||
'settings': self.config_data['app:main'],
|
||||
'env': plain_dummy_env()
|
||||
"store": self.root,
|
||||
"ini_path": self.ini_file,
|
||||
"user": self.user,
|
||||
"repo_name": self.repo_name,
|
||||
"repo_mode": self.repo_mode,
|
||||
"user_permissions": {self.repo_name: "repository.admin"},
|
||||
"settings": self.config_data["app:main"],
|
||||
"env": plain_dummy_env(),
|
||||
}
|
||||
parameters.update(kwargs)
|
||||
server = GitServer(**parameters)
|
||||
return server
|
||||
|
||||
|
||||
@pytest.fixture(scope='module')
|
||||
@pytest.fixture(scope="module")
|
||||
def git_server(request, module_app, rhodecode_factory, available_port_factory):
|
||||
ini_file = module_app._pyramid_settings['__file__']
|
||||
vcsserver_host = module_app._pyramid_settings['vcs.server']
|
||||
ini_file = module_app._pyramid_settings["__file__"]
|
||||
vcsserver_host = module_app._pyramid_settings["vcs.server"]
|
||||
|
||||
store_dir = os.path.dirname(ini_file)
|
||||
|
||||
|
|
@ -78,47 +76,52 @@ def git_server(request, module_app, rhodecode_factory, available_port_factory):
|
|||
store_dir=store_dir,
|
||||
port=available_port_factory(),
|
||||
overrides=(
|
||||
{'handler_console': {'level': 'DEBUG'}},
|
||||
{'app:main': {'vcs.server': vcsserver_host}},
|
||||
{'app:main': {'repo_store.path': store_dir}}
|
||||
))
|
||||
{"handler_console": {"level": "DEBUG"}},
|
||||
{"app:main": {"vcs.server": vcsserver_host}},
|
||||
{"app:main": {"repo_store.path": store_dir}},
|
||||
),
|
||||
)
|
||||
|
||||
service_api_url = f'http://{rc.bind_addr}'
|
||||
service_api_url = f"http://{rc.bind_addr}"
|
||||
|
||||
return GitServerCreator(service_api_url, ini_file)
|
||||
|
||||
|
||||
class TestGitServer:
|
||||
|
||||
def test_command(self, git_server):
|
||||
server = git_server.create()
|
||||
expected_command = (
|
||||
'cd {root}; {git_path} {repo_mode} \'{root}{repo_name}\''.format(
|
||||
root=git_server.root, git_path=git_server.git_path,
|
||||
repo_mode=git_server.repo_mode, repo_name=git_server.repo_name)
|
||||
expected_command = "cd {root}; {git_path} {repo_mode} '{root}{repo_name}'".format(
|
||||
root=git_server.root,
|
||||
git_path=git_server.git_path,
|
||||
repo_mode=git_server.repo_mode,
|
||||
repo_name=git_server.repo_name,
|
||||
)
|
||||
assert expected_command == server.tunnel.command()
|
||||
|
||||
@pytest.mark.parametrize('permissions, action, code', [
|
||||
({}, 'pull', -2),
|
||||
({'test_git': 'repository.read'}, 'pull', 0),
|
||||
({'test_git': 'repository.read'}, 'push', -2),
|
||||
({'test_git': 'repository.write'}, 'push', 0),
|
||||
({'test_git': 'repository.admin'}, 'push', 0),
|
||||
|
||||
])
|
||||
@pytest.mark.parametrize(
|
||||
"permissions, action, code",
|
||||
[
|
||||
({}, "pull", -2),
|
||||
({"test_git": "repository.read"}, "pull", 0),
|
||||
({"test_git": "repository.read"}, "push", -2),
|
||||
({"test_git": "repository.write"}, "push", 0),
|
||||
({"test_git": "repository.admin"}, "push", 0),
|
||||
],
|
||||
)
|
||||
def test_permission_checks(self, git_server, permissions, action, code):
|
||||
server = git_server.create(user_permissions=permissions)
|
||||
result = server._check_permissions(action)
|
||||
assert result is code
|
||||
|
||||
@pytest.mark.parametrize('permissions, value', [
|
||||
({}, False),
|
||||
({'test_git': 'repository.read'}, False),
|
||||
({'test_git': 'repository.write'}, True),
|
||||
({'test_git': 'repository.admin'}, True),
|
||||
|
||||
])
|
||||
@pytest.mark.parametrize(
|
||||
"permissions, value",
|
||||
[
|
||||
({}, False),
|
||||
({"test_git": "repository.read"}, False),
|
||||
({"test_git": "repository.write"}, True),
|
||||
({"test_git": "repository.admin"}, True),
|
||||
],
|
||||
)
|
||||
def test_has_write_permissions(self, git_server, permissions, value):
|
||||
server = git_server.create(user_permissions=permissions)
|
||||
result = server.has_write_perm()
|
||||
|
|
@ -126,49 +129,46 @@ class TestGitServer:
|
|||
|
||||
def test_run_returns_executes_command(self, git_server):
|
||||
from rhodecode.apps.ssh_support.lib.backends.git import GitTunnelWrapper
|
||||
|
||||
server = git_server.create()
|
||||
|
||||
os.environ['SSH_CLIENT'] = '127.0.0.1'
|
||||
with mock.patch.object(GitTunnelWrapper, 'create_hooks_env') as _patch:
|
||||
os.environ["SSH_CLIENT"] = "127.0.0.1"
|
||||
with mock.patch.object(GitTunnelWrapper, "create_hooks_env") as _patch:
|
||||
_patch.return_value = 0
|
||||
with mock.patch.object(GitTunnelWrapper, 'command', return_value='date'):
|
||||
exit_code = server.run(tunnel_extras={'config': server.ini_path})
|
||||
with mock.patch.object(GitTunnelWrapper, "command", return_value="date"):
|
||||
exit_code = server.run(tunnel_extras={"config": server.ini_path})
|
||||
|
||||
assert exit_code == (0, False)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'repo_mode, action', [
|
||||
['receive-pack', 'push'],
|
||||
['upload-pack', 'pull']
|
||||
])
|
||||
@pytest.mark.parametrize("repo_mode, action", [["receive-pack", "push"], ["upload-pack", "pull"]])
|
||||
def test_update_environment(self, git_server, repo_mode, action):
|
||||
server = git_server.create(repo_mode=repo_mode)
|
||||
store = server.store
|
||||
|
||||
with mock.patch('os.environ', {'SSH_CLIENT': '10.10.10.10 b'}):
|
||||
with mock.patch('os.putenv') as putenv_mock:
|
||||
with mock.patch("os.environ", {"SSH_CLIENT": "10.10.10.10 b"}):
|
||||
with mock.patch("os.putenv") as putenv_mock:
|
||||
server.update_environment(action)
|
||||
|
||||
expected_data = {
|
||||
'username': git_server.user.username,
|
||||
'user_id': git_server.user.user_id,
|
||||
'scm': 'git',
|
||||
'repository': git_server.repo_name,
|
||||
'make_lock': None,
|
||||
'action': action,
|
||||
'ip': '10.10.10.10',
|
||||
'locked_by': [None, None],
|
||||
'config': git_server.ini_file,
|
||||
'repo_store': store,
|
||||
'server_url': None,
|
||||
'hooks': ['push', 'pull'],
|
||||
'is_shadow_repo': False,
|
||||
'hooks_module': 'rhodecode.lib.hook_daemon.hook_module',
|
||||
'check_branch_perms': False,
|
||||
'detect_force_push': False,
|
||||
'user_agent': u'git/ssh-user-agent',
|
||||
'SSH': True,
|
||||
'SSH_PERMISSIONS': 'repository.admin',
|
||||
"username": git_server.user.username,
|
||||
"user_id": git_server.user.user_id,
|
||||
"scm": "git",
|
||||
"repository": git_server.repo_name,
|
||||
"make_lock": None,
|
||||
"action": action,
|
||||
"ip": "10.10.10.10",
|
||||
"locked_by": [None, None],
|
||||
"config": git_server.ini_file,
|
||||
"repo_store": store,
|
||||
"server_url": None,
|
||||
"hooks": ["push", "pull"],
|
||||
"is_shadow_repo": False,
|
||||
"hooks_module": "rhodecode.lib.hook_daemon.hook_module",
|
||||
"check_branch_perms": False,
|
||||
"detect_force_push": False,
|
||||
"user_agent": "git/ssh-user-agent",
|
||||
"SSH": True,
|
||||
"SSH_PERMISSIONS": "repository.admin",
|
||||
}
|
||||
args, kwargs = putenv_mock.call_args
|
||||
assert json.loads(args[1]) == expected_data
|
||||
|
|
|
|||
|
|
@ -26,19 +26,19 @@ from rhodecode.apps.ssh_support.tests.conftest import plain_dummy_env, plain_dum
|
|||
|
||||
|
||||
class MercurialServerCreator(object):
|
||||
root = '/tmp/repo/path/'
|
||||
hg_path = '/usr/local/bin/hg'
|
||||
root = "/tmp/repo/path/"
|
||||
hg_path = "/usr/local/bin/hg"
|
||||
|
||||
config_data = {
|
||||
'app:main': {
|
||||
'ssh.executable.hg': hg_path,
|
||||
'vcs.hooks.protocol.v2': 'celery',
|
||||
'app.service_api.host': 'http://localhost',
|
||||
'app.service_api.token': 'secret4',
|
||||
'rhodecode.api.url': '/_admin/api',
|
||||
"app:main": {
|
||||
"ssh.executable.hg": hg_path,
|
||||
"vcs.hooks.protocol.v2": "celery",
|
||||
"app.service_api.host": "http://localhost",
|
||||
"app.service_api.token": "secret4",
|
||||
"rhodecode.api.url": "/_admin/api",
|
||||
}
|
||||
}
|
||||
repo_name = 'test_hg'
|
||||
repo_name = "test_hg"
|
||||
user = plain_dummy_user()
|
||||
|
||||
def __init__(self, service_api_url, ini_file):
|
||||
|
|
@ -46,27 +46,25 @@ class MercurialServerCreator(object):
|
|||
self.ini_file = ini_file
|
||||
|
||||
def create(self, **kwargs):
|
||||
self.config_data['app:main']['app.service_api.host'] = self.service_api_url
|
||||
self.config_data["app:main"]["app.service_api.host"] = self.service_api_url
|
||||
parameters = {
|
||||
'store': self.root,
|
||||
'ini_path': self.ini_file,
|
||||
'user': self.user,
|
||||
'repo_name': self.repo_name,
|
||||
'user_permissions': {
|
||||
self.repo_name: 'repository.admin'
|
||||
},
|
||||
'settings': self.config_data['app:main'],
|
||||
'env': plain_dummy_env()
|
||||
"store": self.root,
|
||||
"ini_path": self.ini_file,
|
||||
"user": self.user,
|
||||
"repo_name": self.repo_name,
|
||||
"user_permissions": {self.repo_name: "repository.admin"},
|
||||
"settings": self.config_data["app:main"],
|
||||
"env": plain_dummy_env(),
|
||||
}
|
||||
parameters.update(kwargs)
|
||||
server = MercurialServer(**parameters)
|
||||
return server
|
||||
|
||||
|
||||
@pytest.fixture(scope='module')
|
||||
@pytest.fixture(scope="module")
|
||||
def hg_server(request, module_app, rhodecode_factory, available_port_factory):
|
||||
ini_file = module_app._pyramid_settings['__file__']
|
||||
vcsserver_host = module_app._pyramid_settings['vcs.server']
|
||||
ini_file = module_app._pyramid_settings["__file__"]
|
||||
vcsserver_host = module_app._pyramid_settings["vcs.server"]
|
||||
|
||||
store_dir = os.path.dirname(ini_file)
|
||||
|
||||
|
|
@ -76,49 +74,51 @@ def hg_server(request, module_app, rhodecode_factory, available_port_factory):
|
|||
store_dir=store_dir,
|
||||
port=available_port_factory(),
|
||||
overrides=(
|
||||
{'handler_console': {'level': 'DEBUG'}},
|
||||
{'app:main': {'vcs.server': vcsserver_host}},
|
||||
{'app:main': {'repo_store.path': store_dir}}
|
||||
))
|
||||
{"handler_console": {"level": "DEBUG"}},
|
||||
{"app:main": {"vcs.server": vcsserver_host}},
|
||||
{"app:main": {"repo_store.path": store_dir}},
|
||||
),
|
||||
)
|
||||
|
||||
service_api_url = f'http://{rc.bind_addr}'
|
||||
service_api_url = f"http://{rc.bind_addr}"
|
||||
|
||||
return MercurialServerCreator(service_api_url, ini_file)
|
||||
|
||||
|
||||
class TestMercurialServer:
|
||||
|
||||
def test_command(self, hg_server, tmpdir):
|
||||
server = hg_server.create()
|
||||
custom_hgrc = os.path.join(str(tmpdir), 'hgrc')
|
||||
expected_command = (
|
||||
'cd {root}; HGRCPATH={custom_hgrc} {hg_path} -R {root}{repo_name} serve --stdio'.format(
|
||||
root=hg_server.root, custom_hgrc=custom_hgrc, hg_path=hg_server.hg_path,
|
||||
repo_name=hg_server.repo_name)
|
||||
custom_hgrc = os.path.join(str(tmpdir), "hgrc")
|
||||
expected_command = "cd {root}; HGRCPATH={custom_hgrc} {hg_path} -R {root}{repo_name} serve --stdio".format(
|
||||
root=hg_server.root, custom_hgrc=custom_hgrc, hg_path=hg_server.hg_path, repo_name=hg_server.repo_name
|
||||
)
|
||||
server_command = server.tunnel.command(custom_hgrc)
|
||||
assert expected_command == server_command
|
||||
|
||||
@pytest.mark.parametrize('permissions, action, code', [
|
||||
({}, 'pull', -2),
|
||||
({'test_hg': 'repository.read'}, 'pull', 0),
|
||||
({'test_hg': 'repository.read'}, 'push', -2),
|
||||
({'test_hg': 'repository.write'}, 'push', 0),
|
||||
({'test_hg': 'repository.admin'}, 'push', 0),
|
||||
|
||||
])
|
||||
@pytest.mark.parametrize(
|
||||
"permissions, action, code",
|
||||
[
|
||||
({}, "pull", -2),
|
||||
({"test_hg": "repository.read"}, "pull", 0),
|
||||
({"test_hg": "repository.read"}, "push", -2),
|
||||
({"test_hg": "repository.write"}, "push", 0),
|
||||
({"test_hg": "repository.admin"}, "push", 0),
|
||||
],
|
||||
)
|
||||
def test_permission_checks(self, hg_server, permissions, action, code):
|
||||
server = hg_server.create(user_permissions=permissions)
|
||||
result = server._check_permissions(action)
|
||||
assert result is code
|
||||
|
||||
@pytest.mark.parametrize('permissions, value', [
|
||||
({}, False),
|
||||
({'test_hg': 'repository.read'}, False),
|
||||
({'test_hg': 'repository.write'}, True),
|
||||
({'test_hg': 'repository.admin'}, True),
|
||||
|
||||
])
|
||||
@pytest.mark.parametrize(
|
||||
"permissions, value",
|
||||
[
|
||||
({}, False),
|
||||
({"test_hg": "repository.read"}, False),
|
||||
({"test_hg": "repository.write"}, True),
|
||||
({"test_hg": "repository.admin"}, True),
|
||||
],
|
||||
)
|
||||
def test_has_write_permissions(self, hg_server, permissions, value):
|
||||
server = hg_server.create(user_permissions=permissions)
|
||||
result = server.has_write_perm()
|
||||
|
|
@ -127,13 +127,11 @@ class TestMercurialServer:
|
|||
def test_run_returns_executes_command(self, hg_server):
|
||||
server = hg_server.create()
|
||||
from rhodecode.apps.ssh_support.lib.backends.hg import MercurialTunnelWrapper
|
||||
os.environ['SSH_CLIENT'] = '127.0.0.1'
|
||||
with mock.patch.object(MercurialTunnelWrapper, 'create_hooks_env') as _patch:
|
||||
|
||||
os.environ["SSH_CLIENT"] = "127.0.0.1"
|
||||
with mock.patch.object(MercurialTunnelWrapper, "create_hooks_env") as _patch:
|
||||
_patch.return_value = 0
|
||||
with mock.patch.object(MercurialTunnelWrapper, 'command', return_value='date'):
|
||||
exit_code = server.run(tunnel_extras={'config': server.ini_path})
|
||||
with mock.patch.object(MercurialTunnelWrapper, "command", return_value="date"):
|
||||
exit_code = server.run(tunnel_extras={"config": server.ini_path})
|
||||
|
||||
assert exit_code == (0, False)
|
||||
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -26,19 +26,19 @@ from rhodecode.apps.ssh_support.tests.conftest import plain_dummy_env, plain_dum
|
|||
|
||||
|
||||
class SubversionServerCreator(object):
|
||||
root = '/tmp/repo/path/'
|
||||
svn_path = '/usr/local/bin/svnserve'
|
||||
root = "/tmp/repo/path/"
|
||||
svn_path = "/usr/local/bin/svnserve"
|
||||
|
||||
config_data = {
|
||||
'app:main': {
|
||||
'ssh.executable.svn': svn_path,
|
||||
'vcs.hooks.protocol.v2': 'celery',
|
||||
'app.service_api.host': 'http://localhost',
|
||||
'app.service_api.token': 'secret4',
|
||||
'rhodecode.api.url': '/_admin/api',
|
||||
"app:main": {
|
||||
"ssh.executable.svn": svn_path,
|
||||
"vcs.hooks.protocol.v2": "celery",
|
||||
"app.service_api.host": "http://localhost",
|
||||
"app.service_api.token": "secret4",
|
||||
"rhodecode.api.url": "/_admin/api",
|
||||
}
|
||||
}
|
||||
repo_name = 'test-svn'
|
||||
repo_name = "test-svn"
|
||||
user = plain_dummy_user()
|
||||
|
||||
def __init__(self, service_api_url, ini_file):
|
||||
|
|
@ -46,27 +46,25 @@ class SubversionServerCreator(object):
|
|||
self.ini_file = ini_file
|
||||
|
||||
def create(self, **kwargs):
|
||||
self.config_data['app:main']['app.service_api.host'] = self.service_api_url
|
||||
self.config_data["app:main"]["app.service_api.host"] = self.service_api_url
|
||||
parameters = {
|
||||
'store': self.root,
|
||||
'ini_path': self.ini_file,
|
||||
'user': self.user,
|
||||
'repo_name': self.repo_name,
|
||||
'user_permissions': {
|
||||
self.repo_name: 'repository.admin'
|
||||
},
|
||||
'settings': self.config_data['app:main'],
|
||||
'env': plain_dummy_env()
|
||||
"store": self.root,
|
||||
"ini_path": self.ini_file,
|
||||
"user": self.user,
|
||||
"repo_name": self.repo_name,
|
||||
"user_permissions": {self.repo_name: "repository.admin"},
|
||||
"settings": self.config_data["app:main"],
|
||||
"env": plain_dummy_env(),
|
||||
}
|
||||
parameters.update(kwargs)
|
||||
server = SubversionServer(**parameters)
|
||||
return server
|
||||
|
||||
|
||||
@pytest.fixture(scope='module')
|
||||
@pytest.fixture(scope="module")
|
||||
def svn_server(request, module_app, rhodecode_factory, available_port_factory):
|
||||
ini_file = module_app._pyramid_settings['__file__']
|
||||
vcsserver_host = module_app._pyramid_settings['vcs.server']
|
||||
ini_file = module_app._pyramid_settings["__file__"]
|
||||
vcsserver_host = module_app._pyramid_settings["vcs.server"]
|
||||
|
||||
store_dir = os.path.dirname(ini_file)
|
||||
|
||||
|
|
@ -76,116 +74,123 @@ def svn_server(request, module_app, rhodecode_factory, available_port_factory):
|
|||
store_dir=store_dir,
|
||||
port=available_port_factory(),
|
||||
overrides=(
|
||||
{'handler_console': {'level': 'DEBUG'}},
|
||||
{'app:main': {'vcs.server': vcsserver_host}},
|
||||
{'app:main': {'repo_store.path': store_dir}}
|
||||
))
|
||||
{"handler_console": {"level": "DEBUG"}},
|
||||
{"app:main": {"vcs.server": vcsserver_host}},
|
||||
{"app:main": {"repo_store.path": store_dir}},
|
||||
),
|
||||
)
|
||||
|
||||
service_api_url = f'http://{rc.bind_addr}'
|
||||
service_api_url = f"http://{rc.bind_addr}"
|
||||
|
||||
return SubversionServerCreator(service_api_url, ini_file)
|
||||
|
||||
|
||||
class TestSubversionServer(object):
|
||||
|
||||
def test_command(self, svn_server):
|
||||
server = svn_server.create()
|
||||
expected_command = [
|
||||
svn_server.svn_path, '-t',
|
||||
'--config-file', server.tunnel.svn_conf_path,
|
||||
'--tunnel-user', svn_server.user.username,
|
||||
'-r', svn_server.root
|
||||
svn_server.svn_path,
|
||||
"-t",
|
||||
"--config-file",
|
||||
server.tunnel.svn_conf_path,
|
||||
"--tunnel-user",
|
||||
svn_server.user.username,
|
||||
"-r",
|
||||
svn_server.root,
|
||||
]
|
||||
|
||||
assert expected_command == server.tunnel.command()
|
||||
|
||||
@pytest.mark.parametrize('permissions, action, code', [
|
||||
({}, 'pull', -2),
|
||||
({'test-svn': 'repository.read'}, 'pull', 0),
|
||||
({'test-svn': 'repository.read'}, 'push', -2),
|
||||
({'test-svn': 'repository.write'}, 'push', 0),
|
||||
({'test-svn': 'repository.admin'}, 'push', 0),
|
||||
|
||||
])
|
||||
@pytest.mark.parametrize(
|
||||
"permissions, action, code",
|
||||
[
|
||||
({}, "pull", -2),
|
||||
({"test-svn": "repository.read"}, "pull", 0),
|
||||
({"test-svn": "repository.read"}, "push", -2),
|
||||
({"test-svn": "repository.write"}, "push", 0),
|
||||
({"test-svn": "repository.admin"}, "push", 0),
|
||||
],
|
||||
)
|
||||
def test_permission_checks(self, svn_server, permissions, action, code):
|
||||
server = svn_server.create(user_permissions=permissions)
|
||||
result = server._check_permissions(action)
|
||||
assert result is code
|
||||
|
||||
@pytest.mark.parametrize('permissions, access_paths, expected_match', [
|
||||
# not matched repository name
|
||||
({
|
||||
'test-svn': ''
|
||||
}, ['test-svn-1', 'test-svn-1/subpath'],
|
||||
None),
|
||||
|
||||
# exact match
|
||||
({
|
||||
'test-svn': ''
|
||||
},
|
||||
['test-svn'],
|
||||
'test-svn'),
|
||||
|
||||
# subdir commits
|
||||
({
|
||||
'test-svn': ''
|
||||
},
|
||||
['test-svn/foo',
|
||||
'test-svn/foo/test-svn',
|
||||
'test-svn/trunk/development.txt',
|
||||
],
|
||||
'test-svn'),
|
||||
|
||||
# subgroups + similar patterns
|
||||
({
|
||||
'test-svn': '',
|
||||
'test-svn-1': '',
|
||||
'test-svn-subgroup/test-svn': '',
|
||||
|
||||
},
|
||||
['test-svn-1',
|
||||
'test-svn-1/foo/test-svn',
|
||||
'test-svn-1/test-svn',
|
||||
],
|
||||
'test-svn-1'),
|
||||
|
||||
# subgroups + similar patterns
|
||||
({
|
||||
'test-svn-1': '',
|
||||
'test-svn-10': '',
|
||||
'test-svn-100': '',
|
||||
},
|
||||
['test-svn-10',
|
||||
'test-svn-10/foo/test-svn',
|
||||
'test-svn-10/test-svn',
|
||||
],
|
||||
'test-svn-10'),
|
||||
|
||||
# subgroups + similar patterns
|
||||
({
|
||||
'name': '',
|
||||
'nameContains': '',
|
||||
'nameContainsThis': '',
|
||||
},
|
||||
['nameContains',
|
||||
'nameContains/This',
|
||||
'nameContains/This/test-svn',
|
||||
],
|
||||
'nameContains'),
|
||||
|
||||
# subgroups + similar patterns
|
||||
({
|
||||
'test-svn': '',
|
||||
'test-svn-1': '',
|
||||
'test-svn-subgroup/test-svn': '',
|
||||
|
||||
},
|
||||
['test-svn-subgroup/test-svn',
|
||||
'test-svn-subgroup/test-svn/foo/test-svn',
|
||||
'test-svn-subgroup/test-svn/trunk/example.txt',
|
||||
],
|
||||
'test-svn-subgroup/test-svn'),
|
||||
])
|
||||
@pytest.mark.parametrize(
|
||||
"permissions, access_paths, expected_match",
|
||||
[
|
||||
# not matched repository name
|
||||
({"test-svn": ""}, ["test-svn-1", "test-svn-1/subpath"], None),
|
||||
# exact match
|
||||
({"test-svn": ""}, ["test-svn"], "test-svn"),
|
||||
# subdir commits
|
||||
(
|
||||
{"test-svn": ""},
|
||||
[
|
||||
"test-svn/foo",
|
||||
"test-svn/foo/test-svn",
|
||||
"test-svn/trunk/development.txt",
|
||||
],
|
||||
"test-svn",
|
||||
),
|
||||
# subgroups + similar patterns
|
||||
(
|
||||
{
|
||||
"test-svn": "",
|
||||
"test-svn-1": "",
|
||||
"test-svn-subgroup/test-svn": "",
|
||||
},
|
||||
[
|
||||
"test-svn-1",
|
||||
"test-svn-1/foo/test-svn",
|
||||
"test-svn-1/test-svn",
|
||||
],
|
||||
"test-svn-1",
|
||||
),
|
||||
# subgroups + similar patterns
|
||||
(
|
||||
{
|
||||
"test-svn-1": "",
|
||||
"test-svn-10": "",
|
||||
"test-svn-100": "",
|
||||
},
|
||||
[
|
||||
"test-svn-10",
|
||||
"test-svn-10/foo/test-svn",
|
||||
"test-svn-10/test-svn",
|
||||
],
|
||||
"test-svn-10",
|
||||
),
|
||||
# subgroups + similar patterns
|
||||
(
|
||||
{
|
||||
"name": "",
|
||||
"nameContains": "",
|
||||
"nameContainsThis": "",
|
||||
},
|
||||
[
|
||||
"nameContains",
|
||||
"nameContains/This",
|
||||
"nameContains/This/test-svn",
|
||||
],
|
||||
"nameContains",
|
||||
),
|
||||
# subgroups + similar patterns
|
||||
(
|
||||
{
|
||||
"test-svn": "",
|
||||
"test-svn-1": "",
|
||||
"test-svn-subgroup/test-svn": "",
|
||||
},
|
||||
[
|
||||
"test-svn-subgroup/test-svn",
|
||||
"test-svn-subgroup/test-svn/foo/test-svn",
|
||||
"test-svn-subgroup/test-svn/trunk/example.txt",
|
||||
],
|
||||
"test-svn-subgroup/test-svn",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_repo_extraction_on_subdir(self, svn_server, permissions, access_paths, expected_match):
|
||||
server = svn_server.create(user_permissions=permissions)
|
||||
for path in access_paths:
|
||||
|
|
@ -196,21 +201,14 @@ class TestSubversionServer(object):
|
|||
from rhodecode.apps.ssh_support.lib.backends.svn import SubversionTunnelWrapper
|
||||
|
||||
server = svn_server.create()
|
||||
os.environ['SSH_CLIENT'] = '127.0.0.1'
|
||||
os.environ["SSH_CLIENT"] = "127.0.0.1"
|
||||
with mock.patch.object(
|
||||
SubversionTunnelWrapper, 'get_first_client_response',
|
||||
return_value={'url': 'http://server/test-svn'}):
|
||||
with mock.patch.object(
|
||||
SubversionTunnelWrapper, 'patch_first_client_response',
|
||||
return_value=0):
|
||||
with mock.patch.object(
|
||||
SubversionTunnelWrapper, 'sync',
|
||||
return_value=0):
|
||||
with mock.patch.object(
|
||||
SubversionTunnelWrapper, 'command',
|
||||
return_value=['date']):
|
||||
|
||||
exit_code = server.run(tunnel_extras={'config': server.ini_path})
|
||||
SubversionTunnelWrapper, "get_first_client_response", return_value={"url": "http://server/test-svn"}
|
||||
):
|
||||
with mock.patch.object(SubversionTunnelWrapper, "patch_first_client_response", return_value=0):
|
||||
with mock.patch.object(SubversionTunnelWrapper, "sync", return_value=0):
|
||||
with mock.patch.object(SubversionTunnelWrapper, "command", return_value=["date"]):
|
||||
exit_code = server.run(tunnel_extras={"config": server.ini_path})
|
||||
# SVN has this differently configured, and we get in our mock env
|
||||
# None as return code
|
||||
assert exit_code == (None, False)
|
||||
|
|
@ -219,9 +217,8 @@ class TestSubversionServer(object):
|
|||
from rhodecode.apps.ssh_support.lib.backends.svn import SubversionTunnelWrapper
|
||||
|
||||
server = svn_server.create()
|
||||
with mock.patch.object(SubversionTunnelWrapper, 'command', return_value=['date']):
|
||||
with mock.patch.object(SubversionTunnelWrapper, 'get_first_client_response',
|
||||
return_value=None):
|
||||
exit_code = server.run(tunnel_extras={'config': server.ini_path})
|
||||
with mock.patch.object(SubversionTunnelWrapper, "command", return_value=["date"]):
|
||||
with mock.patch.object(SubversionTunnelWrapper, "get_first_client_response", return_value=None):
|
||||
exit_code = server.run(tunnel_extras={"config": server.ini_path})
|
||||
|
||||
assert exit_code == (1, False)
|
||||
|
|
|
|||
|
|
@ -25,27 +25,23 @@ from rhodecode.lib.utils2 import AttributeDict
|
|||
|
||||
|
||||
class TestSshKeyFileGeneration(object):
|
||||
@pytest.mark.parametrize('ssh_wrapper_cmd', ['/tmp/sshwrapper.py'])
|
||||
@pytest.mark.parametrize('allow_shell', [True, False])
|
||||
@pytest.mark.parametrize('debug', [True, False])
|
||||
@pytest.mark.parametrize('ssh_opts', [None, 'mycustom,option'])
|
||||
@pytest.mark.parametrize("ssh_wrapper_cmd", ["/tmp/sshwrapper.py"])
|
||||
@pytest.mark.parametrize("allow_shell", [True, False])
|
||||
@pytest.mark.parametrize("debug", [True, False])
|
||||
@pytest.mark.parametrize("ssh_opts", [None, "mycustom,option"])
|
||||
def test_write_keyfile(self, tmpdir, ssh_wrapper_cmd, allow_shell, debug, ssh_opts):
|
||||
|
||||
authorized_keys_file_path = os.path.join(str(tmpdir), 'authorized_keys')
|
||||
authorized_keys_file_path = os.path.join(str(tmpdir), "authorized_keys")
|
||||
|
||||
def keys():
|
||||
return [
|
||||
AttributeDict({'user': AttributeDict(username='admin'),
|
||||
'ssh_key_data': 'ssh-rsa ADMIN_KEY'}),
|
||||
AttributeDict({'user': AttributeDict(username='user'),
|
||||
'ssh_key_data': 'ssh-rsa USER_KEY'}),
|
||||
AttributeDict({"user": AttributeDict(username="admin"), "ssh_key_data": "ssh-rsa ADMIN_KEY"}),
|
||||
AttributeDict({"user": AttributeDict(username="user"), "ssh_key_data": "ssh-rsa USER_KEY"}),
|
||||
]
|
||||
with mock.patch('rhodecode.apps.ssh_support.utils.get_all_active_keys',
|
||||
return_value=keys()):
|
||||
with mock.patch.dict('rhodecode.CONFIG', {'__file__': '/tmp/file.ini'}):
|
||||
|
||||
with mock.patch("rhodecode.apps.ssh_support.utils.get_all_active_keys", return_value=keys()):
|
||||
with mock.patch.dict("rhodecode.CONFIG", {"__file__": "/tmp/file.ini"}):
|
||||
utils._generate_ssh_authorized_keys_file(
|
||||
authorized_keys_file_path, ssh_wrapper_cmd,
|
||||
allow_shell, ssh_opts, debug
|
||||
authorized_keys_file_path, ssh_wrapper_cmd, allow_shell, ssh_opts, debug
|
||||
)
|
||||
|
||||
assert os.path.isfile(authorized_keys_file_path)
|
||||
|
|
@ -53,17 +49,16 @@ class TestSshKeyFileGeneration(object):
|
|||
content = f.read()
|
||||
|
||||
assert 'command="/tmp/sshwrapper.py' in content
|
||||
assert 'This file is managed by RhodeCode, ' \
|
||||
'please do not edit it manually.' in content
|
||||
assert "This file is managed by RhodeCode, please do not edit it manually." in content
|
||||
|
||||
if allow_shell:
|
||||
assert '--shell' in content
|
||||
assert "--shell" in content
|
||||
|
||||
if debug:
|
||||
assert '--debug' in content
|
||||
assert "--debug" in content
|
||||
|
||||
assert '--user' in content
|
||||
assert '--user-id' in content
|
||||
assert "--user" in content
|
||||
assert "--user-id" in content
|
||||
|
||||
if ssh_opts:
|
||||
assert ssh_opts in content
|
||||
|
|
|
|||
|
|
@ -20,29 +20,32 @@ import pytest
|
|||
|
||||
|
||||
class TestSSHWrapper(object):
|
||||
|
||||
def test_serve_raises_an_exception_when_vcs_is_not_recognized(self, ssh_wrapper):
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
ssh_wrapper.serve(
|
||||
vcs='microsoft-tfs', repo='test-repo', mode=None, user='test',
|
||||
permissions={}, branch_permissions={})
|
||||
assert str(exc_info.value) == 'Unrecognised VCS: microsoft-tfs'
|
||||
vcs="microsoft-tfs", repo="test-repo", mode=None, user="test", permissions={}, branch_permissions={}
|
||||
)
|
||||
assert str(exc_info.value) == "Unrecognised VCS: microsoft-tfs"
|
||||
|
||||
def test_get_connection_info(self, ssh_wrapper):
|
||||
conn_info = ssh_wrapper.get_connection_info()
|
||||
assert {'client_ip': '127.0.0.1',
|
||||
'client_port': '22',
|
||||
'server_ip': '10.0.0.1',
|
||||
'server_port': '443'} == conn_info
|
||||
assert {
|
||||
"client_ip": "127.0.0.1",
|
||||
"client_port": "22",
|
||||
"server_ip": "10.0.0.1",
|
||||
"server_port": "443",
|
||||
} == conn_info
|
||||
|
||||
@pytest.mark.parametrize('command, vcs', [
|
||||
('xxx', None),
|
||||
('svnserve -t', 'svn'),
|
||||
('hg -R repo serve --stdio', 'hg'),
|
||||
('git-receive-pack \'repo.git\'', 'git'),
|
||||
|
||||
])
|
||||
@pytest.mark.parametrize(
|
||||
"command, vcs",
|
||||
[
|
||||
("xxx", None),
|
||||
("svnserve -t", "svn"),
|
||||
("hg -R repo serve --stdio", "hg"),
|
||||
("git-receive-pack 'repo.git'", "git"),
|
||||
],
|
||||
)
|
||||
def test_get_repo_details(self, ssh_wrapper, command, vcs):
|
||||
ssh_wrapper.command = command
|
||||
vcs_type, repo_name, mode = ssh_wrapper.get_repo_details(mode='auto')
|
||||
vcs_type, repo_name, mode = ssh_wrapper.get_repo_details(mode="auto")
|
||||
assert vcs_type == vcs
|
||||
|
|
|
|||
|
|
@ -28,72 +28,60 @@ from rhodecode.model.db import true, joinedload, User, UserSshKeys
|
|||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
HEADER = \
|
||||
"# This file is managed by RhodeCode, please do not edit it manually. # \n" \
|
||||
HEADER = (
|
||||
"# This file is managed by RhodeCode, please do not edit it manually. # \n"
|
||||
"# Current entries: {}, create date: UTC:{}.\n"
|
||||
)
|
||||
|
||||
# Default SSH options for authorized_keys file, can be override via .ini
|
||||
SSH_OPTS = 'no-pty,no-port-forwarding,no-X11-forwarding,no-agent-forwarding'
|
||||
SSH_OPTS = "no-pty,no-port-forwarding,no-X11-forwarding,no-agent-forwarding"
|
||||
|
||||
|
||||
def get_all_active_keys():
|
||||
result = UserSshKeys.query() \
|
||||
.join(User) \
|
||||
.filter(User != User.get_default_user()) \
|
||||
.filter(User.active == true()) \
|
||||
.all()
|
||||
result = UserSshKeys.query().join(User).filter(User != User.get_default_user()).filter(User.active == true()).all()
|
||||
return result
|
||||
|
||||
|
||||
def _generate_ssh_authorized_keys_file(
|
||||
authorized_keys_file_path, ssh_wrapper_cmd, allow_shell, ssh_opts, debug):
|
||||
def _generate_ssh_authorized_keys_file(authorized_keys_file_path, ssh_wrapper_cmd, allow_shell, ssh_opts, debug):
|
||||
import rhodecode
|
||||
|
||||
authorized_keys_file_path = os.path.abspath(
|
||||
os.path.expanduser(authorized_keys_file_path))
|
||||
authorized_keys_file_path = os.path.abspath(os.path.expanduser(authorized_keys_file_path))
|
||||
tmp_file_dir = os.path.dirname(authorized_keys_file_path)
|
||||
|
||||
if not os.path.exists(tmp_file_dir):
|
||||
log.debug('SSH authorized_keys file dir does not exist, creating one now...')
|
||||
log.debug("SSH authorized_keys file dir does not exist, creating one now...")
|
||||
os.makedirs(tmp_file_dir)
|
||||
|
||||
all_active_keys = get_all_active_keys()
|
||||
|
||||
if allow_shell:
|
||||
ssh_wrapper_cmd = ssh_wrapper_cmd + ' --shell'
|
||||
ssh_wrapper_cmd = ssh_wrapper_cmd + " --shell"
|
||||
if debug:
|
||||
ssh_wrapper_cmd = ssh_wrapper_cmd + ' --debug'
|
||||
ssh_wrapper_cmd = ssh_wrapper_cmd + " --debug"
|
||||
|
||||
if not os.path.isfile(authorized_keys_file_path):
|
||||
log.debug('Creating file at %s', authorized_keys_file_path)
|
||||
with open(authorized_keys_file_path, 'w'):
|
||||
log.debug("Creating file at %s", authorized_keys_file_path)
|
||||
with open(authorized_keys_file_path, "w"):
|
||||
# create a file with write access
|
||||
pass
|
||||
|
||||
if not os.access(authorized_keys_file_path, os.R_OK):
|
||||
raise OSError('Access to file {} is without read access'.format(
|
||||
authorized_keys_file_path))
|
||||
raise OSError("Access to file {} is without read access".format(authorized_keys_file_path))
|
||||
|
||||
line_tmpl = '{ssh_opts},command="{wrapper_command} {ini_path} --user-id={user_id} --user={user} --key-id={user_key_id}" {key}\n'
|
||||
|
||||
fd, tmp_authorized_keys = tempfile.mkstemp(
|
||||
'.authorized_keys_write_operation',
|
||||
dir=tmp_file_dir)
|
||||
fd, tmp_authorized_keys = tempfile.mkstemp(".authorized_keys_write_operation", dir=tmp_file_dir)
|
||||
|
||||
now = datetime.datetime.utcnow().isoformat()
|
||||
keys_file = os.fdopen(fd, 'wt')
|
||||
keys_file = os.fdopen(fd, "wt")
|
||||
keys_file.write(HEADER.format(len(all_active_keys), now))
|
||||
ini_path = rhodecode.CONFIG['__file__']
|
||||
ini_path = rhodecode.CONFIG["__file__"]
|
||||
|
||||
for user_key in all_active_keys:
|
||||
username = user_key.user.username
|
||||
user_id = user_key.user.user_id
|
||||
# replace all newline from ends and inside
|
||||
safe_key_data = user_key.ssh_key_data\
|
||||
.strip()\
|
||||
.replace('\n', ' ') \
|
||||
.replace('\t', ' ') \
|
||||
.replace('\r', ' ')
|
||||
safe_key_data = user_key.ssh_key_data.strip().replace("\n", " ").replace("\t", " ").replace("\r", " ")
|
||||
|
||||
line = line_tmpl.format(
|
||||
ssh_opts=ssh_opts or SSH_OPTS,
|
||||
|
|
@ -102,10 +90,11 @@ def _generate_ssh_authorized_keys_file(
|
|||
user_id=user_id,
|
||||
user=username,
|
||||
user_key_id=user_key.ssh_key_id,
|
||||
key=safe_key_data)
|
||||
key=safe_key_data,
|
||||
)
|
||||
|
||||
keys_file.write(line)
|
||||
log.debug('addkey: Key added for user: `%s`', username)
|
||||
log.debug("addkey: Key added for user: `%s`", username)
|
||||
keys_file.close()
|
||||
|
||||
# Explicitly setting read-only permissions to authorized_keys
|
||||
|
|
@ -115,22 +104,15 @@ def _generate_ssh_authorized_keys_file(
|
|||
|
||||
|
||||
def generate_ssh_authorized_keys_file(registry):
|
||||
log.info('Generating new authorized key file')
|
||||
log.info("Generating new authorized key file")
|
||||
|
||||
authorized_keys_file_path = registry.settings.get(
|
||||
config_keys.authorized_keys_file_path)
|
||||
authorized_keys_file_path = registry.settings.get(config_keys.authorized_keys_file_path)
|
||||
|
||||
ssh_wrapper_cmd = registry.settings.get(
|
||||
config_keys.wrapper_cmd)
|
||||
allow_shell = registry.settings.get(
|
||||
config_keys.wrapper_allow_shell)
|
||||
ssh_opts = registry.settings.get(
|
||||
config_keys.authorized_keys_line_ssh_opts)
|
||||
debug = registry.settings.get(
|
||||
config_keys.enable_debug_logging)
|
||||
ssh_wrapper_cmd = registry.settings.get(config_keys.wrapper_cmd)
|
||||
allow_shell = registry.settings.get(config_keys.wrapper_allow_shell)
|
||||
ssh_opts = registry.settings.get(config_keys.authorized_keys_line_ssh_opts)
|
||||
debug = registry.settings.get(config_keys.enable_debug_logging)
|
||||
|
||||
_generate_ssh_authorized_keys_file(
|
||||
authorized_keys_file_path, ssh_wrapper_cmd, allow_shell, ssh_opts,
|
||||
debug)
|
||||
_generate_ssh_authorized_keys_file(authorized_keys_file_path, ssh_wrapper_cmd, allow_shell, ssh_opts, debug)
|
||||
|
||||
return 0
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue