svn: fixed case of wrong extracted repository name for SSH backend. In cases

where we commited to a nested subdirs SVN reported the access path with the subdir paths in it.
We couldn't then match that extended name into proper rhodecode repository for ACL checks.

- Current implementation gives an slight overhead as we have to lookup all repositories
- fixes  #5606

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

@@ -1,223 +1,220 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2019 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 pyramid.compat import configparser
 
 from rhodecode.model.db import Session, User, UserSshKeys
 from rhodecode.model.scm import ScmModel
 
 from .hg import MercurialServer
 from .git import GitServer
 from .svn import SubversionServer
 log = logging.getLogger(__name__)
 
 
 class SshWrapper(object):
 
     def __init__(self, command, connection_info, mode,
                  user, user_id, key_id, shell, ini_path, env):
         self.command = command
         self.connection_info = connection_info
         self.mode = mode
         self.user = user
         self.user_id = user_id
         self.key_id = key_id
         self.shell = shell
         self.ini_path = ini_path
         self.env = env
 
         self.config = self.parse_config(ini_path)
         self.server_impl = None
 
     def parse_config(self, config_path):
         parser = configparser.ConfigParser()
         parser.read(config_path)
         return parser
 
     def update_key_access_time(self, key_id):
         key = UserSshKeys().query().filter(
             UserSshKeys.ssh_key_id == key_id).scalar()
         if key:
             key.accessed_on = datetime.datetime.utcnow()
             Session().add(key)
             Session().commit()
             log.debug('Update key id:`%s` fingerprint:`%s` access time',
                       key_id, key.ssh_key_fingerprint)
 
     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 get_repo_details(self, mode):
         vcs_type = mode if mode in ['svn', 'hg', 'git'] else None
-        mode = mode
         repo_name = None
 
         hg_pattern = r'^hg\s+\-R\s+(\S+)\s+serve\s+\-\-stdio$'
         hg_match = re.match(hg_pattern, self.command)
         if hg_match is not None:
             vcs_type = 'hg'
             repo_name = hg_match.group(1).strip('/')
             return vcs_type, repo_name, mode
 
-        git_pattern = (
-            r'^git-(receive-pack|upload-pack)\s\'[/]?(\S+?)(|\.git)\'$')
+        git_pattern = r'^git-(receive-pack|upload-pack)\s\'[/]?(\S+?)(|\.git)\'$'
         git_match = re.match(git_pattern, self.command)
         if git_match is not None:
             vcs_type = 'git'
             repo_name = git_match.group(2).strip('/')
             mode = git_match.group(1)
             return vcs_type, repo_name, mode
 
         svn_pattern = r'^svnserve -t'
         svn_match = re.match(svn_pattern, self.command)
 
         if svn_match is not None:
             vcs_type = 'svn'
-            # Repo name should be extracted from the input stream
+            # 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):
         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,
         }
 
         if vcs == 'hg':
             server = MercurialServer(
                 store=store, ini_path=self.ini_path,
                 repo_name=repo, user=user,
                 user_permissions=permissions, config=self.config, 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, config=self.config, 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, config=self.config, env=self.env)
             self.server_impl = server
             return server.run(tunnel_extras=extras)
 
         else:
             raise Exception('Unrecognised VCS: {}'.format(vcs))
 
     def wrap(self):
         mode = self.mode
         user = self.user
         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: `%s:%s` Shell: `%s` SSH Command: `\"%s\"` '
             'SCM_DETECTED: `%s` SCM Mode: `%s` SCM Repo: `%s`',
             mode, user, user_id, shell, self.command,
             scm_detected, scm_mode, scm_repo)
 
         # update last access time for this key
         self.update_key_access_time(key_id)
 
         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')
+            log.info('Dropping to shell, no command given and shell is allowed')
             os.execl('/bin/bash', '-l')
             exit_code = 1
 
         elif scm_detected:
             user = User.get(user_id)
             if not user:
                 log.warning('User with id %s not found', user_id)
                 exit_code = -1
                 return exit_code
 
             auth_user = user.AuthUser()
             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)
+            log.error('Unhandled Command: "%s" Aborting.', self.command)
             exit_code = -1
 
         return exit_code

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

@@ -1,163 +1,162 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2019 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 json
 import logging
 
 from rhodecode.lib.hooks_daemon import prepare_callback_daemon
 from rhodecode.lib.vcs.conf import settings as vcs_settings
 from rhodecode.model.scm import ScmModel
 
 log = logging.getLogger(__name__)
 
 
 class VcsServer(object):
     _path = None  # set executable path for hg/git/svn binary
     backend = None  # set in child classes
     tunnel = None  # subprocess handling tunnel
     write_perms = ['repository.admin', 'repository.write']
     read_perms = ['repository.read', 'repository.admin', 'repository.write']
 
     def __init__(self, user, user_permissions, config, env):
         self.user = user
         self.user_permissions = user_permissions
         self.config = config
         self.env = env
         self.stdin = sys.stdin
 
         self.repo_name = None
         self.repo_mode = None
         self.store = ''
         self.ini_path = ''
 
     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
         """
         ScmModel().mark_for_invalidation(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)
-        log.debug(
-            'permission for %s on %s are: %s',
+        log.debug('permission for %s on %s are: %s',
             self.user, self.repo_name, permission)
 
         if not permission:
             log.error('user `%s` permissions to repo:%s are empty. Forbidding access.',
                       self.user, 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"!',
                     self.user, self.repo_name)
                 return 0
         else:
             if permission in self.write_perms:
                 log.info(
                     'WRITE+ Permissions for User "%s" detected to repo "%s"!',
                     self.user, self.repo_name)
                 return 0
 
         log.error('Cannot properly fetch or verify user `%s` permissions. '
                   'Permissions: %s, vcs action: %s',
                   self.user, 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': 'ssh-user-agent',
             'hooks': ['push', 'pull'],
             'hooks_module': 'rhodecode.lib.hooks_daemon',
             '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))
 
     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['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"
 
     def run(self, tunnel_extras=None):
         tunnel_extras = tunnel_extras or {}
         extras = {}
         extras.update(tunnel_extras)
 
         callback_daemon, extras = prepare_callback_daemon(
             extras, protocol=vcs_settings.HOOKS_PROTOCOL,
             host=vcs_settings.HOOKS_HOST,
             use_direct_calls=False)
 
         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,75 +1,74 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2019 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 .base import VcsServer
 
 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
 
         self.server.update_environment(action=action, extras=extras)
         self.create_hooks_env()
         return os.system(self.command())
 
 
 class GitServer(VcsServer):
     backend = 'git'
 
     def __init__(self, store, ini_path, repo_name, repo_mode,
                  user, user_permissions, config, env):
         super(GitServer, self).\
             __init__(user, user_permissions, config, env)
 
         self.store = store
         self.ini_path = ini_path
         self.repo_name = repo_name
-        self._path = self.git_path = config.get(
-            'app:main', 'ssh.executable.git')
+        self._path = self.git_path = config.get('app:main', 'ssh.executable.git')
 
         self.repo_mode = repo_mode
         self.tunnel = GitTunnelWrapper(server=self)

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

@@ -1,228 +1,254 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2019 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 urlparse
 
 from .base import VcsServer
 
 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()
         command = [
             self.server.svn_path, '-t',
             '--config-file', self.svn_conf_path,
             '-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)
+        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(
+        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 '{length}:{string} '.format(length=len(str_), string=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
+        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 = urlparse.urlparse(first_response['url'])
-        self.server.repo_name = url_parts.path.strip('/')
+
+        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(VcsServer):
     backend = 'svn'
 
     def __init__(self, store, ini_path, repo_name,
                  user, user_permissions, config, env):
         super(SubversionServer, self)\
             .__init__(user, user_permissions, config, env)
         self.store = store
         self.ini_path = ini_path
-        # this is set in .run() from input stream
+        # 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 = config.get(
-            'app:main', 'ssh.executable.svn')
+        self._path = self.svn_path = config.get('app:main', '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/apps/ssh_support/tests/test_server_svn.py

@@ -1,124 +1,204 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2019 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 mock
 import pytest
 
 from rhodecode.apps.ssh_support.lib.backends.svn import SubversionServer
 from rhodecode.apps.ssh_support.tests.conftest import plain_dummy_env, plain_dummy_user
 
 
 class SubversionServerCreator(object):
     root = '/tmp/repo/path/'
     svn_path = '/usr/local/bin/svnserve'
     config_data = {
         'app:main': {
             'ssh.executable.svn': svn_path,
             'vcs.hooks.protocol': 'http',
         }
     }
     repo_name = 'test-svn'
     user = plain_dummy_user()
 
     def __init__(self):
         def config_get(part, key):
             return self.config_data.get(part, {}).get(key)
         self.config_mock = mock.Mock()
         self.config_mock.get = mock.Mock(side_effect=config_get)
 
     def create(self, **kwargs):
         parameters = {
             'store': self.root,
             'repo_name': self.repo_name,
             'ini_path': '',
             'user': self.user,
             'user_permissions': {
                 self.repo_name: 'repository.admin'
             },
             'config': self.config_mock,
             'env': plain_dummy_env()
         }
 
         parameters.update(kwargs)
         server = SubversionServer(**parameters)
         return server
 
 
 @pytest.fixture()
 def svn_server(app):
     return SubversionServerCreator()
 
 
 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, '-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),
 
     ])
     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'),
+    ])
+    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:
+            repo_name = server.tunnel._match_repo_name(path)
+            assert repo_name == expected_match
+
     def test_run_returns_executes_command(self, svn_server):
         server = svn_server.create()
         from rhodecode.apps.ssh_support.lib.backends.svn import SubversionTunnelWrapper
         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()
         # SVN has this differently configured, and we get in our mock env
         # None as return code
         assert exit_code == (None, False)
 
     def test_run_returns_executes_command_that_cannot_extract_repo_name(self, svn_server):
         server = svn_server.create()
         from rhodecode.apps.ssh_support.lib.backends.svn import SubversionTunnelWrapper
         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()
 
         assert exit_code == (1, False)