python3: fixed urllib usage

rhodecode/apps/login/views.py

@@ -1,470 +1,470 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2020 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 time
 import collections
 import datetime
 import formencode
 import formencode.htmlfill
 import logging
 import urllib.parse
 import requests
 
 from pyramid.httpexceptions import HTTPFound
 
 
 from rhodecode.apps._base import BaseAppView
 from rhodecode.authentication.base import authenticate, HTTP_TYPE
 from rhodecode.authentication.plugins import auth_rhodecode
 from rhodecode.events import UserRegistered, trigger
 from rhodecode.lib import helpers as h
 from rhodecode.lib import audit_logger
 from rhodecode.lib.auth import (
     AuthUser, HasPermissionAnyDecorator, CSRFRequired)
 from rhodecode.lib.base import get_ip_addr
 from rhodecode.lib.exceptions import UserCreationError
 from rhodecode.lib.utils2 import safe_str
 from rhodecode.model.db import User, UserApiKeys
 from rhodecode.model.forms import LoginForm, RegisterForm, PasswordResetForm
 from rhodecode.model.meta import Session
 from rhodecode.model.auth_token import AuthTokenModel
 from rhodecode.model.settings import SettingsModel
 from rhodecode.model.user import UserModel
 from rhodecode.translation import _
 
 
 log = logging.getLogger(__name__)
 
 CaptchaData = collections.namedtuple(
     'CaptchaData', 'active, private_key, public_key')
 
 
 def store_user_in_session(session, username, remember=False):
     user = User.get_by_username(username, case_insensitive=True)
     auth_user = AuthUser(user.user_id)
     auth_user.set_authenticated()
     cs = auth_user.get_cookie_store()
     session['rhodecode_user'] = cs
     user.update_lastlogin()
     Session().commit()
 
     # If they want to be remembered, update the cookie
     if remember:
         _year = (datetime.datetime.now() +
                  datetime.timedelta(seconds=60 * 60 * 24 * 365))
         session._set_cookie_expires(_year)
 
     session.save()
 
     safe_cs = cs.copy()
     safe_cs['password'] = '****'
     log.info('user %s is now authenticated and stored in '
              'session, session attrs %s', username, safe_cs)
 
     # dumps session attrs back to cookie
     session._update_cookie_out()
     # we set new cookie
     headers = None
     if session.request['set_cookie']:
         # send set-cookie headers back to response to update cookie
         headers = [('Set-Cookie', session.request['cookie_out'])]
     return headers
 
 
 def get_came_from(request):
     came_from = safe_str(request.GET.get('came_from', ''))
-    parsed = urllib.parse.urlparse.urlparse(came_from)
+    parsed = urllib.parse.urlparse(came_from)
     allowed_schemes = ['http', 'https']
     default_came_from = h.route_path('home')
     if parsed.scheme and parsed.scheme not in allowed_schemes:
         log.error('Suspicious URL scheme detected %s for url %s',
                   parsed.scheme, parsed)
         came_from = default_came_from
     elif parsed.netloc and request.host != parsed.netloc:
         log.error('Suspicious NETLOC detected %s for url %s server url '
                   'is: %s', parsed.netloc, parsed, request.host)
         came_from = default_came_from
     elif any(bad_str in parsed.path for bad_str in ('\r', '\n')):
         log.error('Header injection detected `%s` for url %s server url ',
                   parsed.path, parsed)
         came_from = default_came_from
 
     return came_from or default_came_from
 
 
 class LoginView(BaseAppView):
 
     def load_default_context(self):
         c = self._get_local_tmpl_context()
         c.came_from = get_came_from(self.request)
         return c
 
     def _get_captcha_data(self):
         settings = SettingsModel().get_all_settings()
         private_key = settings.get('rhodecode_captcha_private_key')
         public_key = settings.get('rhodecode_captcha_public_key')
         active = bool(private_key)
         return CaptchaData(
             active=active, private_key=private_key, public_key=public_key)
 
     def validate_captcha(self, private_key):
 
         captcha_rs = self.request.POST.get('g-recaptcha-response')
         url = "https://www.google.com/recaptcha/api/siteverify"
         params = {
             'secret': private_key,
             'response': captcha_rs,
             'remoteip': get_ip_addr(self.request.environ)
         }
         verify_rs = requests.get(url, params=params, verify=True, timeout=60)
         verify_rs = verify_rs.json()
         captcha_status = verify_rs.get('success', False)
         captcha_errors = verify_rs.get('error-codes', [])
         if not isinstance(captcha_errors, list):
             captcha_errors = [captcha_errors]
         captcha_errors = ', '.join(captcha_errors)
         captcha_message = ''
         if captcha_status is False:
             captcha_message = "Bad captcha. Errors: {}".format(
                 captcha_errors)
 
         return captcha_status, captcha_message
 
     def login(self):
         c = self.load_default_context()
         auth_user = self._rhodecode_user
 
         # redirect if already logged in
         if (auth_user.is_authenticated and
                 not auth_user.is_default and auth_user.ip_allowed):
             raise HTTPFound(c.came_from)
 
         # check if we use headers plugin, and try to login using it.
         try:
             log.debug('Running PRE-AUTH for headers based authentication')
             auth_info = authenticate(
                 '', '', self.request.environ, HTTP_TYPE, skip_missing=True)
             if auth_info:
                 headers = store_user_in_session(
                     self.session, auth_info.get('username'))
                 raise HTTPFound(c.came_from, headers=headers)
         except UserCreationError as e:
             log.error(e)
             h.flash(e, category='error')
 
         return self._get_template_context(c)
 
     def login_post(self):
         c = self.load_default_context()
 
         login_form = LoginForm(self.request.translate)()
 
         try:
             self.session.invalidate()
             form_result = login_form.to_python(self.request.POST)
             # form checks for username/password, now we're authenticated
             headers = store_user_in_session(
                 self.session,
                 username=form_result['username'],
                 remember=form_result['remember'])
             log.debug('Redirecting to "%s" after login.', c.came_from)
 
             audit_user = audit_logger.UserWrap(
                 username=self.request.POST.get('username'),
                 ip_addr=self.request.remote_addr)
             action_data = {'user_agent': self.request.user_agent}
             audit_logger.store_web(
                 'user.login.success', action_data=action_data,
                 user=audit_user, commit=True)
 
             raise HTTPFound(c.came_from, headers=headers)
         except formencode.Invalid as errors:
             defaults = errors.value
             # remove password from filling in form again
             defaults.pop('password', None)
             render_ctx = {
                 'errors': errors.error_dict,
                 'defaults': defaults,
             }
 
             audit_user = audit_logger.UserWrap(
                 username=self.request.POST.get('username'),
                 ip_addr=self.request.remote_addr)
             action_data = {'user_agent': self.request.user_agent}
             audit_logger.store_web(
                 'user.login.failure', action_data=action_data,
                 user=audit_user, commit=True)
             return self._get_template_context(c, **render_ctx)
 
         except UserCreationError as e:
             # headers auth or other auth functions that create users on
             # the fly can throw this exception signaling that there's issue
             # with user creation, explanation should be provided in
             # Exception itself
             h.flash(e, category='error')
             return self._get_template_context(c)
 
     @CSRFRequired()
     def logout(self):
         auth_user = self._rhodecode_user
         log.info('Deleting session for user: `%s`', auth_user)
 
         action_data = {'user_agent': self.request.user_agent}
         audit_logger.store_web(
             'user.logout', action_data=action_data,
             user=auth_user, commit=True)
         self.session.delete()
         return HTTPFound(h.route_path('home'))
 
     @HasPermissionAnyDecorator(
         'hg.admin', 'hg.register.auto_activate', 'hg.register.manual_activate')
     def register(self, defaults=None, errors=None):
         c = self.load_default_context()
         defaults = defaults or {}
         errors = errors or {}
 
         settings = SettingsModel().get_all_settings()
         register_message = settings.get('rhodecode_register_message') or ''
         captcha = self._get_captcha_data()
         auto_active = 'hg.register.auto_activate' in User.get_default_user()\
             .AuthUser().permissions['global']
 
         render_ctx = self._get_template_context(c)
         render_ctx.update({
             'defaults': defaults,
             'errors': errors,
             'auto_active': auto_active,
             'captcha_active': captcha.active,
             'captcha_public_key': captcha.public_key,
             'register_message': register_message,
         })
         return render_ctx
 
     @HasPermissionAnyDecorator(
         'hg.admin', 'hg.register.auto_activate', 'hg.register.manual_activate')
     def register_post(self):
         from rhodecode.authentication.plugins import auth_rhodecode
 
         self.load_default_context()
         captcha = self._get_captcha_data()
         auto_active = 'hg.register.auto_activate' in User.get_default_user()\
             .AuthUser().permissions['global']
 
         extern_name = auth_rhodecode.RhodeCodeAuthPlugin.uid
         extern_type = auth_rhodecode.RhodeCodeAuthPlugin.uid
 
         register_form = RegisterForm(self.request.translate)()
         try:
 
             form_result = register_form.to_python(self.request.POST)
             form_result['active'] = auto_active
             external_identity = self.request.POST.get('external_identity')
 
             if external_identity:
                 extern_name = external_identity
                 extern_type = external_identity
 
             if captcha.active:
                 captcha_status, captcha_message = self.validate_captcha(
                     captcha.private_key)
 
                 if not captcha_status:
                     _value = form_result
                     _msg = _('Bad captcha')
                     error_dict = {'recaptcha_field': captcha_message}
                     raise formencode.Invalid(
                         _msg, _value, None, error_dict=error_dict)
 
             new_user = UserModel().create_registration(
                 form_result, extern_name=extern_name, extern_type=extern_type)
 
             action_data = {'data': new_user.get_api_data(),
                            'user_agent': self.request.user_agent}
 
             if external_identity:
                 action_data['external_identity'] = external_identity
 
             audit_user = audit_logger.UserWrap(
                 username=new_user.username,
                 user_id=new_user.user_id,
                 ip_addr=self.request.remote_addr)
 
             audit_logger.store_web(
                 'user.register', action_data=action_data,
                 user=audit_user)
 
             event = UserRegistered(user=new_user, session=self.session)
             trigger(event)
             h.flash(
                 _('You have successfully registered with RhodeCode. You can log-in now.'),
                 category='success')
             if external_identity:
                 h.flash(
                     _('Please use the {identity} button to log-in').format(
                         identity=external_identity),
                     category='success')
             Session().commit()
 
             redirect_ro = self.request.route_path('login')
             raise HTTPFound(redirect_ro)
 
         except formencode.Invalid as errors:
             errors.value.pop('password', None)
             errors.value.pop('password_confirmation', None)
             return self.register(
                 defaults=errors.value, errors=errors.error_dict)
 
         except UserCreationError as e:
             # container auth or other auth functions that create users on
             # the fly can throw this exception signaling that there's issue
             # with user creation, explanation should be provided in
             # Exception itself
             h.flash(e, category='error')
             return self.register()
 
     def password_reset(self):
         c = self.load_default_context()
         captcha = self._get_captcha_data()
 
         template_context = {
             'captcha_active': captcha.active,
             'captcha_public_key': captcha.public_key,
             'defaults': {},
             'errors': {},
         }
 
         # always send implicit message to prevent from discovery of
         # matching emails
         msg = _('If such email exists, a password reset link was sent to it.')
 
         def default_response():
             log.debug('faking response on invalid password reset')
             # make this take 2s, to prevent brute forcing.
             time.sleep(2)
             h.flash(msg, category='success')
             return HTTPFound(self.request.route_path('reset_password'))
 
         if self.request.POST:
             if h.HasPermissionAny('hg.password_reset.disabled')():
                 _email = self.request.POST.get('email', '')
                 log.error('Failed attempt to reset password for `%s`.', _email)
                 h.flash(_('Password reset has been disabled.'), category='error')
                 return HTTPFound(self.request.route_path('reset_password'))
 
             password_reset_form = PasswordResetForm(self.request.translate)()
             description = u'Generated token for password reset from {}'.format(
                 datetime.datetime.now().isoformat())
 
             try:
                 form_result = password_reset_form.to_python(
                     self.request.POST)
                 user_email = form_result['email']
 
                 if captcha.active:
                     captcha_status, captcha_message = self.validate_captcha(
                         captcha.private_key)
 
                     if not captcha_status:
                         _value = form_result
                         _msg = _('Bad captcha')
                         error_dict = {'recaptcha_field': captcha_message}
                         raise formencode.Invalid(
                             _msg, _value, None, error_dict=error_dict)
 
                 # Generate reset URL and send mail.
                 user = User.get_by_email(user_email)
 
                 # only allow rhodecode based users to reset their password
                 # external auth shouldn't allow password reset
                 if user and user.extern_type != auth_rhodecode.RhodeCodeAuthPlugin.uid:
                     log.warning('User %s with external type `%s` tried a password reset. '
                                 'This try was rejected', user, user.extern_type)
                     return default_response()
 
                 # generate password reset token that expires in 10 minutes
                 reset_token = UserModel().add_auth_token(
                     user=user, lifetime_minutes=10,
                     role=UserModel.auth_token_role.ROLE_PASSWORD_RESET,
                     description=description)
                 Session().commit()
 
                 log.debug('Successfully created password recovery token')
                 password_reset_url = self.request.route_url(
                     'reset_password_confirmation',
                     _query={'key': reset_token.api_key})
                 UserModel().reset_password_link(
                     form_result, password_reset_url)
 
                 action_data = {'email': user_email,
                                'user_agent': self.request.user_agent}
                 audit_logger.store_web(
                     'user.password.reset_request', action_data=action_data,
                     user=self._rhodecode_user, commit=True)
 
                 return default_response()
 
             except formencode.Invalid as errors:
                 template_context.update({
                     'defaults': errors.value,
                     'errors': errors.error_dict,
                 })
                 if not self.request.POST.get('email'):
                     # case of empty email, we want to report that
                     return self._get_template_context(c, **template_context)
 
                 if 'recaptcha_field' in errors.error_dict:
                     # case of failed captcha
                     return self._get_template_context(c, **template_context)
 
             return default_response()
 
         return self._get_template_context(c, **template_context)
 
     def password_reset_confirmation(self):
         self.load_default_context()
         if self.request.GET and self.request.GET.get('key'):
             # make this take 2s, to prevent brute forcing.
             time.sleep(2)
 
             token = AuthTokenModel().get_auth_token(
                 self.request.GET.get('key'))
 
             # verify token is the correct role
             if token is None or token.role != UserApiKeys.ROLE_PASSWORD_RESET:
                 log.debug('Got token with role:%s expected is %s',
                           getattr(token, 'role', 'EMPTY_TOKEN'),
                           UserApiKeys.ROLE_PASSWORD_RESET)
                 h.flash(
                     _('Given reset token is invalid'), category='error')
                 return HTTPFound(self.request.route_path('reset_password'))
 
             try:
                 owner = token.user
                 data = {'email': owner.email, 'token': token.api_key}
                 UserModel().reset_password(data)
                 h.flash(
                     _('Your password reset was successful, '
                       'a new password has been sent to your email'),
                     category='success')
             except Exception as e:
                 log.error(e)
                 return HTTPFound(self.request.route_path('reset_password'))
 
         return HTTPFound(self.request.route_path('login'))

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

@@ -1,258 +1,258 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2020 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import os
 import re
 import sys
 import logging
 import signal
 import tempfile
 from subprocess import Popen, PIPE
 import urllib.parse
 
 from .base import 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()
         username = self.server.user.username
 
         command = [
             self.server.svn_path, '-t',
             '--config-file', self.svn_conf_path,
             '--tunnel-user', username,
             '-r', root]
         log.debug("Final CMD: %s", ' '.join(command))
         return command
 
     def start(self):
         command = self.command()
         self.process = Popen(' '.join(command), stdin=PIPE, shell=True)
 
     def sync(self):
         while self.process.poll() is None:
             next_byte = self.stdin.read(1)
             if not next_byte:
                 break
             self.process.stdin.write(next_byte)
         self.remove_configs()
 
     @property
     def return_code(self):
         return self.process.returncode
 
     def get_first_client_response(self):
         signal.signal(signal.SIGALRM, self.interrupt)
         signal.alarm(self.timeout)
         first_response = self._read_first_client_response()
         signal.alarm(0)
         return (self._parse_first_client_response(first_response)
                 if first_response else None)
 
     def patch_first_client_response(self, response, **kwargs):
         self.create_hooks_env()
         data = response.copy()
         data.update(kwargs)
         data['url'] = self._svn_string(data['url'])
         data['ra_client'] = self._svn_string(data['ra_client'])
         data['client'] = data['client'] or ''
         buffer_ = (
             "( {version} ( {capabilities} ) {url}{ra_client}"
             "( {client}) ) ".format(**data))
         self.process.stdin.write(buffer_)
 
     def fail(self, message):
         print("( failure ( ( 210005 {message} 0: 0 ) ) )".format(
             message=self._svn_string(message)))
         self.remove_configs()
         self.process.kill()
         return 1
 
     def interrupt(self, signum, frame):
         self.fail("Exited by timeout")
 
     def _svn_string(self, str_):
         if not str_:
             return ''
         return '{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
         """
         version_re = r'(?P<version>\d+)'
         capabilities_re = r'\(\s(?P<capabilities>[\w\d\-\ ]+)\s\)'
         url_re = r'\d+\:(?P<url>[\W\w]+)'
         ra_client_re = r'(\d+\:(?P<ra_client>[\W\w]+)\s)'
         client_re = r'(\d+\:(?P<client>[\W\w]+)\s)*'
         regex = re.compile(
             r'^\(\s{version}\s{capabilities}\s{url}\s{ra_client}'
             r'\(\s{client}\)\s\)\s*$'.format(
                 version=version_re, capabilities=capabilities_re,
                 url=url_re, ra_client=ra_client_re, client=client_re))
         matcher = regex.match(buffer_)
 
         return matcher.groupdict() if matcher else None
 
     def _match_repo_name(self, url):
         """
         Given an server url, try to match it against ALL known repository names.
         This handles a tricky SVN case for SSH and subdir commits.
         E.g if our repo name is my-svn-repo, a svn commit on file in a subdir would
         result in the url with this subdir added.
         """
         # case 1 direct match, we don't do any "heavy" lookups
         if url in self.server.user_permissions:
             return url
 
         log.debug('Extracting repository name from subdir path %s', url)
         # case 2 we check all permissions, and match closes possible case...
         # NOTE(dan): In this case we only know that url has a subdir parts, it's safe
         # to assume that it will have the repo name as prefix, we ensure the prefix
         # for similar repositories isn't matched by adding a /
         # e.g subgroup/repo-name/ and subgroup/repo-name-1/ would work correct.
         for repo_name in self.server.user_permissions:
             repo_name_prefix = repo_name + '/'
             if url.startswith(repo_name_prefix):
                 log.debug('Found prefix %s match, returning proper repository name',
                           repo_name_prefix)
                 return repo_name
 
         return
 
     def run(self, extras):
         action = 'pull'
         self.create_svn_config()
         self.start()
 
         first_response = self.get_first_client_response()
         if not first_response:
             return self.fail("Repository name cannot be extracted")
 
-        url_parts = urllib.parse.urlparse.urlparse(first_response['url'])
+        url_parts = urllib.parse.urlparse(first_response['url'])
 
         self.server.repo_name = self._match_repo_name(url_parts.path.strip('/'))
 
         exit_code = self.server._check_permissions(action)
         if exit_code:
             return exit_code
 
         # set the readonly flag to False if we have proper permissions
         if self.server.has_write_perm():
             self.read_only = False
         self.server.update_environment(action=action, extras=extras)
 
         self.patch_first_client_response(first_response)
         self.sync()
         return self.return_code
 
 
 class SubversionServer(VcsServer):
     backend = 'svn'
     repo_user_agent = '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
         # 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.tunnel = SubversionTunnelWrapper(server=self)
 
     def _handle_tunnel(self, extras):
 
         # pre-auth
         action = 'pull'
         # Special case for SVN, we extract repo name at later stage
         # exit_code = self._check_permissions(action)
         # if exit_code:
         #     return exit_code, False
 
         req = self.env['request']
         server_url = req.host_url + req.script_name
         extras['server_url'] = server_url
 
         log.debug('Using %s binaries from path %s', self.backend, self._path)
         exit_code = self.tunnel.run(extras)
 
         return exit_code, action == "push"
 
 

rhodecode/lib/markup_renderer.py

@@ -1,580 +1,580 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2011-2020 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/
 
 
 """
 Renderer for markup languages with ability to parse using rst or markdown
 """
 
 import re
 import os
 import lxml
 import logging
 import urllib.parse
 import bleach
 
 from mako.lookup import TemplateLookup
 from mako.template import Template as MakoTemplate
 
 from docutils.core import publish_parts
 from docutils.parsers.rst import directives
 from docutils import writers
 from docutils.writers import html4css1
 import markdown
 
 from rhodecode.lib.markdown_ext import GithubFlavoredMarkdownExtension
 from rhodecode.lib.utils2 import (safe_unicode, md5_safe, MENTIONS_REGEX)
 
 log = logging.getLogger(__name__)
 
 # default renderer used to generate automated comments
 DEFAULT_COMMENTS_RENDERER = 'rst'
 
 try:
     from lxml.html import fromstring
     from lxml.html import tostring
 except ImportError:
     log.exception('Failed to import lxml')
     fromstring = None
     tostring = None
 
 
 class CustomHTMLTranslator(writers.html4css1.HTMLTranslator):
     """
     Custom HTML Translator used for sandboxing potential
     JS injections in ref links
     """
     def visit_literal_block(self, node):
         self.body.append(self.starttag(node, 'pre', CLASS='codehilite literal-block'))
 
     def visit_reference(self, node):
         if 'refuri' in node.attributes:
             refuri = node['refuri']
             if ':' in refuri:
                 prefix, link = refuri.lstrip().split(':', 1)
                 prefix = prefix or ''
 
                 if prefix.lower() == 'javascript':
                     # we don't allow javascript type of refs...
                     node['refuri'] = 'javascript:alert("SandBoxedJavascript")'
 
         # old style class requires this...
         return html4css1.HTMLTranslator.visit_reference(self, node)
 
 
 class RhodeCodeWriter(writers.html4css1.Writer):
     def __init__(self):
         writers.Writer.__init__(self)
         self.translator_class = CustomHTMLTranslator
 
 
 def relative_links(html_source, server_paths):
     if not html_source:
         return html_source
 
     if not fromstring and tostring:
         return html_source
 
     try:
         doc = lxml.html.fromstring(html_source)
     except Exception:
         return html_source
 
     for el in doc.cssselect('img, video'):
         src = el.attrib.get('src')
         if src:
             el.attrib['src'] = relative_path(src, server_paths['raw'])
 
     for el in doc.cssselect('a:not(.gfm)'):
         src = el.attrib.get('href')
         if src:
             raw_mode = el.attrib['href'].endswith('?raw=1')
             if raw_mode:
                 el.attrib['href'] = relative_path(src, server_paths['raw'])
             else:
                 el.attrib['href'] = relative_path(src, server_paths['standard'])
 
     return lxml.html.tostring(doc)
 
 
 def relative_path(path, request_path, is_repo_file=None):
     """
     relative link support, path is a rel path, and request_path is current
     server path (not absolute)
 
     e.g.
 
     path = '../logo.png'
     request_path= '/repo/files/path/file.md'
     produces: '/repo/files/logo.png'
     """
     # TODO(marcink): unicode/str support ?
     # maybe=> safe_unicode(urllib.quote(safe_str(final_path), '/:'))
 
     def dummy_check(p):
         return True  # assume default is a valid file path
 
     is_repo_file = is_repo_file or dummy_check
     if not path:
         return request_path
 
     path = safe_unicode(path)
     request_path = safe_unicode(request_path)
 
     if path.startswith((u'data:', u'javascript:', u'#', u':')):
         # skip data, anchor, invalid links
         return path
 
-    is_absolute = bool(urllib.parse.urlparse.urlparse(path).netloc)
+    is_absolute = bool(urllib.parse.urlparse(path).netloc)
     if is_absolute:
         return path
 
     if not request_path:
         return path
 
     if path.startswith(u'/'):
         path = path[1:]
 
     if path.startswith(u'./'):
         path = path[2:]
 
     parts = request_path.split('/')
     # compute how deep we need to traverse the request_path
     depth = 0
 
     if is_repo_file(request_path):
         # if request path is a VALID file, we use a relative path with
         # one level up
         depth += 1
 
     while path.startswith(u'../'):
         depth += 1
         path = path[3:]
 
     if depth > 0:
         parts = parts[:-depth]
 
     parts.append(path)
     final_path = u'/'.join(parts).lstrip(u'/')
 
     return u'/' + final_path
 
 
 _cached_markdown_renderer = None
 
 
 def get_markdown_renderer(extensions, output_format):
     global _cached_markdown_renderer
 
     if _cached_markdown_renderer is None:
         _cached_markdown_renderer = markdown.Markdown(
             extensions=extensions,
             enable_attributes=False, output_format=output_format)
     return _cached_markdown_renderer
 
 
 _cached_markdown_renderer_flavored = None
 
 
 def get_markdown_renderer_flavored(extensions, output_format):
     global _cached_markdown_renderer_flavored
 
     if _cached_markdown_renderer_flavored is None:
         _cached_markdown_renderer_flavored = markdown.Markdown(
             extensions=extensions + [GithubFlavoredMarkdownExtension()],
             enable_attributes=False, output_format=output_format)
     return _cached_markdown_renderer_flavored
 
 
 class MarkupRenderer(object):
     RESTRUCTUREDTEXT_DISALLOWED_DIRECTIVES = ['include', 'meta', 'raw']
 
     MARKDOWN_PAT = re.compile(r'\.(md|mkdn?|mdown|markdown)$', re.IGNORECASE)
     RST_PAT = re.compile(r'\.re?st$', re.IGNORECASE)
     JUPYTER_PAT = re.compile(r'\.(ipynb)$', re.IGNORECASE)
     PLAIN_PAT = re.compile(r'^readme$', re.IGNORECASE)
 
     URL_PAT = re.compile(r'(http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]'
                          r'|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+)')
 
     MENTION_PAT = re.compile(MENTIONS_REGEX)
 
     extensions = ['markdown.extensions.codehilite', 'markdown.extensions.extra',
                   'markdown.extensions.def_list', 'markdown.extensions.sane_lists']
 
     output_format = 'html4'
 
     # extension together with weights. Lower is first means we control how
     # extensions are attached to readme names with those.
     PLAIN_EXTS = [
         # prefer no extension
         ('', 0),  # special case that renders READMES names without extension
         ('.text', 2), ('.TEXT', 2),
         ('.txt', 3), ('.TXT', 3)
     ]
 
     RST_EXTS = [
         ('.rst', 1), ('.rest', 1),
         ('.RST', 2), ('.REST', 2)
     ]
 
     MARKDOWN_EXTS = [
         ('.md', 1), ('.MD', 1),
         ('.mkdn', 2), ('.MKDN', 2),
         ('.mdown', 3), ('.MDOWN', 3),
         ('.markdown', 4), ('.MARKDOWN', 4)
     ]
 
     def _detect_renderer(self, source, filename=None):
         """
         runs detection of what renderer should be used for generating html
         from a markup language
 
         filename can be also explicitly a renderer name
 
         :param source:
         :param filename:
         """
 
         if MarkupRenderer.MARKDOWN_PAT.findall(filename):
             detected_renderer = 'markdown'
         elif MarkupRenderer.RST_PAT.findall(filename):
             detected_renderer = 'rst'
         elif MarkupRenderer.JUPYTER_PAT.findall(filename):
             detected_renderer = 'jupyter'
         elif MarkupRenderer.PLAIN_PAT.findall(filename):
             detected_renderer = 'plain'
         else:
             detected_renderer = 'plain'
 
         return getattr(MarkupRenderer, detected_renderer)
 
     @classmethod
     def bleach_clean(cls, text):
         from .bleach_whitelist import markdown_attrs, markdown_tags
         allowed_tags = markdown_tags
         allowed_attrs = markdown_attrs
 
         try:
             return bleach.clean(text, tags=allowed_tags, attributes=allowed_attrs)
         except Exception:
             return 'UNPARSEABLE TEXT'
 
     @classmethod
     def renderer_from_filename(cls, filename, exclude):
         """
         Detect renderer markdown/rst from filename and optionally use exclude
         list to remove some options. This is mostly used in helpers.
         Returns None when no renderer can be detected.
         """
         def _filter(elements):
             if isinstance(exclude, (list, tuple)):
                 return [x for x in elements if x not in exclude]
             return elements
 
         if filename.endswith(
                 tuple(_filter([x[0] for x in cls.MARKDOWN_EXTS if x[0]]))):
             return 'markdown'
         if filename.endswith(tuple(_filter([x[0] for x in cls.RST_EXTS if x[0]]))):
             return 'rst'
 
         return None
 
     def render(self, source, filename=None):
         """
         Renders a given filename using detected renderer
         it detects renderers based on file extension or mimetype.
         At last it will just do a simple html replacing new lines with <br/>
 
         :param file_name:
         :param source:
         """
 
         renderer = self._detect_renderer(source, filename)
         readme_data = renderer(source)
         return readme_data
 
     @classmethod
     def _flavored_markdown(cls, text):
         """
         Github style flavored markdown
 
         :param text:
         """
 
         # Extract pre blocks.
         extractions = {}
 
         def pre_extraction_callback(matchobj):
             digest = md5_safe(matchobj.group(0))
             extractions[digest] = matchobj.group(0)
             return "{gfm-extraction-%s}" % digest
         pattern = re.compile(r'<pre>.*?</pre>', re.MULTILINE | re.DOTALL)
         text = re.sub(pattern, pre_extraction_callback, text)
 
         # Prevent foo_bar_baz from ending up with an italic word in the middle.
         def italic_callback(matchobj):
             s = matchobj.group(0)
             if list(s).count('_') >= 2:
                 return s.replace('_', r'\_')
             return s
         text = re.sub(r'^(?! {4}|\t)\w+_\w+_\w[\w_]*', italic_callback, text)
 
         # Insert pre block extractions.
         def pre_insert_callback(matchobj):
             return '\n\n' + extractions[matchobj.group(1)]
         text = re.sub(r'\{gfm-extraction-([0-9a-f]{32})\}',
                       pre_insert_callback, text)
 
         return text
 
     @classmethod
     def urlify_text(cls, text):
         def url_func(match_obj):
             url_full = match_obj.groups()[0]
             return '<a href="%(url)s">%(url)s</a>' % ({'url': url_full})
 
         return cls.URL_PAT.sub(url_func, text)
 
     @classmethod
     def convert_mentions(cls, text, mode):
         mention_pat = cls.MENTION_PAT
 
         def wrapp(match_obj):
             uname = match_obj.groups()[0]
             hovercard_url = "pyroutes.url('hovercard_username', {'username': '%s'});" % uname
 
             if mode == 'markdown':
                 tmpl = '<strong class="tooltip-hovercard" data-hovercard-alt="{uname}" data-hovercard-url="{hovercard_url}">@{uname}</strong>'
             elif mode == 'rst':
                 tmpl = ' **@{uname}** '
             else:
                 raise ValueError('mode must be rst or markdown')
 
             return tmpl.format(**{'uname': uname,
                                   'hovercard_url': hovercard_url})
 
         return mention_pat.sub(wrapp, text).strip()
 
     @classmethod
     def plain(cls, source, universal_newline=True, leading_newline=True):
         source = safe_unicode(source)
         if universal_newline:
             newline = '\n'
             source = newline.join(source.splitlines())
 
         rendered_source = cls.urlify_text(source)
         source = ''
         if leading_newline:
             source += '<br />'
         source += rendered_source.replace("\n", '<br />')
 
         rendered = cls.bleach_clean(source)
         return rendered
 
     @classmethod
     def markdown(cls, source, safe=True, flavored=True, mentions=False,
                  clean_html=True):
         """
         returns markdown rendered code cleaned by the bleach library
         """
 
         if flavored:
             markdown_renderer = get_markdown_renderer_flavored(
                 cls.extensions, cls.output_format)
         else:
             markdown_renderer = get_markdown_renderer(
                 cls.extensions, cls.output_format)
 
         if mentions:
             mention_hl = cls.convert_mentions(source, mode='markdown')
             # we extracted mentions render with this using Mentions false
             return cls.markdown(mention_hl, safe=safe, flavored=flavored,
                                 mentions=False)
 
         source = safe_unicode(source)
 
         try:
             if flavored:
                 source = cls._flavored_markdown(source)
             rendered = markdown_renderer.convert(source)
         except Exception:
             log.exception('Error when rendering Markdown')
             if safe:
                 log.debug('Fallback to render in plain mode')
                 rendered = cls.plain(source)
             else:
                 raise
 
         if clean_html:
             rendered = cls.bleach_clean(rendered)
         return rendered
 
     @classmethod
     def rst(cls, source, safe=True, mentions=False, clean_html=False):
         if mentions:
             mention_hl = cls.convert_mentions(source, mode='rst')
             # we extracted mentions render with this using Mentions false
             return cls.rst(mention_hl, safe=safe, mentions=False)
 
         source = safe_unicode(source)
         try:
             docutils_settings = dict(
                 [(alias, None) for alias in
                  cls.RESTRUCTUREDTEXT_DISALLOWED_DIRECTIVES])
 
             docutils_settings.update({
                 'input_encoding': 'unicode',
                 'report_level': 4,
                 'syntax_highlight': 'short',
             })
 
             for k, v in docutils_settings.items():
                 directives.register_directive(k, v)
 
             parts = publish_parts(source=source,
                                   writer=RhodeCodeWriter(),
                                   settings_overrides=docutils_settings)
             rendered = parts["fragment"]
             if clean_html:
                 rendered = cls.bleach_clean(rendered)
             return parts['html_title'] + rendered
         except Exception:
             log.exception('Error when rendering RST')
             if safe:
                 log.debug('Fallback to render in plain mode')
                 return cls.plain(source)
             else:
                 raise
 
     @classmethod
     def jupyter(cls, source, safe=True):
         from rhodecode.lib import helpers
 
         from traitlets.config import Config
         import nbformat
         from nbconvert import HTMLExporter
         from nbconvert.preprocessors import Preprocessor
 
         class CustomHTMLExporter(HTMLExporter):
             def _template_file_default(self):
                 return 'basic'
 
         class Sandbox(Preprocessor):
 
             def preprocess(self, nb, resources):
                 sandbox_text = 'SandBoxed(IPython.core.display.Javascript object)'
                 for cell in nb['cells']:
                     if not safe:
                         continue
 
                     if 'outputs' in cell:
                         for cell_output in cell['outputs']:
                             if 'data' in cell_output:
                                 if 'application/javascript' in cell_output['data']:
                                     cell_output['data']['text/plain'] = sandbox_text
                                     cell_output['data'].pop('application/javascript', None)
 
                     if 'source' in cell and cell['cell_type'] == 'markdown':
                         # sanitize similar like in markdown
                         cell['source'] = cls.bleach_clean(cell['source'])
 
                 return nb, resources
 
         def _sanitize_resources(input_resources):
             """
             Skip/sanitize some of the CSS generated and included in jupyter
             so it doesn't messes up UI so much
             """
 
             # TODO(marcink): probably we should replace this with whole custom
             # CSS set that doesn't screw up, but jupyter generated html has some
             # special markers, so it requires Custom HTML exporter template with
             # _default_template_path_default, to achieve that
 
             # strip the reset CSS
             input_resources[0] = input_resources[0][input_resources[0].find('/*! Source'):]
             return input_resources
 
         def as_html(notebook):
             conf = Config()
             conf.CustomHTMLExporter.preprocessors = [Sandbox]
             html_exporter = CustomHTMLExporter(config=conf)
 
             (body, resources) = html_exporter.from_notebook_node(notebook)
             header = '<!-- ## IPYTHON NOTEBOOK RENDERING ## -->'
             js = MakoTemplate(r'''
             <!-- MathJax configuration -->
             <script type="text/x-mathjax-config">
             MathJax.Hub.Config({
                 jax: ["input/TeX","output/HTML-CSS", "output/PreviewHTML"],
                 extensions: ["tex2jax.js","MathMenu.js","MathZoom.js", "fast-preview.js", "AssistiveMML.js", "[Contrib]/a11y/accessibility-menu.js"],
                 TeX: {
                     extensions: ["AMSmath.js","AMSsymbols.js","noErrors.js","noUndefined.js"]
                 },
                 tex2jax: {
                     inlineMath: [ ['$','$'], ["\\(","\\)"] ],
                     displayMath: [ ['$$','$$'], ["\\[","\\]"] ],
                     processEscapes: true,
                     processEnvironments: true
                 },
                 // Center justify equations in code and markdown cells. Elsewhere
                 // we use CSS to left justify single line equations in code cells.
                 displayAlign: 'center',
                 "HTML-CSS": {
                     styles: {'.MathJax_Display': {"margin": 0}},
                     linebreaks: { automatic: true },
                     availableFonts: ["STIX", "TeX"]
                 },
                 showMathMenu: false
             });
             </script>
             <!-- End of MathJax configuration -->
             <script src="${h.asset('js/src/math_jax/MathJax.js')}"></script>
             ''').render(h=helpers)
 
             css = MakoTemplate(r'''
             <link rel="stylesheet" type="text/css" href="${h.asset('css/style-ipython.css', ver=ver)}" media="screen"/>
             ''').render(h=helpers, ver='ver1')
 
             body = '\n'.join([header, css, js, body])
             return body, resources
 
         notebook = nbformat.reads(source, as_version=4)
         (body, resources) = as_html(notebook)
         return body
 
 
 class RstTemplateRenderer(object):
 
     def __init__(self):
         base = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
         rst_template_dirs = [os.path.join(base, 'templates', 'rst_templates')]
         self.template_store = TemplateLookup(
             directories=rst_template_dirs,
             input_encoding='utf-8',
             imports=['from rhodecode.lib import helpers as h'])
 
     def _get_template(self, templatename):
         return self.template_store.get_template(templatename)
 
     def render(self, template_name, **kwargs):
         template = self._get_template(template_name)
         return template.render(**kwargs)

rhodecode/lib/middleware/simplesvn.py

@@ -1,229 +1,229 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 base64
 import logging
 import urllib.request, urllib.parse, urllib.error
 import urllib.parse
 
 import requests
 from pyramid.httpexceptions import HTTPNotAcceptable
 
 from rhodecode.lib import rc_cache
 from rhodecode.lib.middleware import simplevcs
 from rhodecode.lib.utils import is_valid_repo
 from rhodecode.lib.utils2 import str2bool, safe_int, safe_str
 from rhodecode.lib.ext_json import json
 from rhodecode.lib.hooks_daemon import store_txn_id_data
 
 
 log = logging.getLogger(__name__)
 
 
 class SimpleSvnApp(object):
     IGNORED_HEADERS = [
         'connection', 'keep-alive', 'content-encoding',
         'transfer-encoding', 'content-length']
     rc_extras = {}
 
     def __init__(self, config):
         self.config = config
 
     def __call__(self, environ, start_response):
         request_headers = self._get_request_headers(environ)
         data = environ['wsgi.input']
         req_method = environ['REQUEST_METHOD']
         has_content_length = 'CONTENT_LENGTH' in environ
         path_info = self._get_url(
             self.config.get('subversion_http_server_url', ''), environ['PATH_INFO'])
         transfer_encoding = environ.get('HTTP_TRANSFER_ENCODING', '')
         log.debug('Handling: %s method via `%s`', req_method, path_info)
 
         # stream control flag, based on request and content type...
         stream = False
 
         if req_method in ['MKCOL'] or has_content_length:
             data_processed = False
             # read chunk to check if we have txn-with-props
             initial_data = data.read(1024)
             if initial_data.startswith('(create-txn-with-props'):
                 data = initial_data + data.read()
                 # store on-the-fly our rc_extra using svn revision properties
                 # those can be read later on in hooks executed so we have a way
                 # to pass in the data into svn hooks
                 rc_data = base64.urlsafe_b64encode(json.dumps(self.rc_extras))
                 rc_data_len = len(rc_data)
                 # header defines data length, and serialized data
                 skel = ' rc-scm-extras {} {}'.format(rc_data_len, rc_data)
                 data = data[:-2] + skel + '))'
                 data_processed = True
 
             if not data_processed:
                 # NOTE(johbo): Avoid that we end up with sending the request in chunked
                 # transfer encoding (mainly on Gunicorn). If we know the content
                 # length, then we should transfer the payload in one request.
                 data = initial_data + data.read()
 
         if req_method in ['GET', 'PUT'] or transfer_encoding == 'chunked':
             # NOTE(marcink): when getting/uploading files we want to STREAM content
             # back to the client/proxy instead of buffering it here...
             stream = True
 
         stream = stream
         log.debug('Calling SVN PROXY at `%s`, using method:%s. Stream: %s',
                   path_info, req_method, stream)
         try:
             response = requests.request(
                 req_method, path_info,
                 data=data, headers=request_headers, stream=stream)
         except requests.ConnectionError:
             log.exception('ConnectionError occurred for endpoint %s', path_info)
             raise
 
         if response.status_code not in [200, 401]:
             from rhodecode.lib.utils2 import safe_str
             text = '\n{}'.format(safe_str(response.text)) if response.text else ''
             if response.status_code >= 500:
                 log.error('Got SVN response:%s with text:`%s`', response, text)
             else:
                 log.debug('Got SVN response:%s with text:`%s`', response, text)
         else:
             log.debug('got response code: %s', response.status_code)
 
         response_headers = self._get_response_headers(response.headers)
 
         if response.headers.get('SVN-Txn-name'):
             svn_tx_id = response.headers.get('SVN-Txn-name')
             txn_id = rc_cache.utils.compute_key_from_params(
                 self.config['repository'], svn_tx_id)
             port = safe_int(self.rc_extras['hooks_uri'].split(':')[-1])
             store_txn_id_data(txn_id, {'port': port})
 
         start_response(
             '{} {}'.format(response.status_code, response.reason),
             response_headers)
         return response.iter_content(chunk_size=1024)
 
     def _get_url(self, svn_http_server, path):
         svn_http_server_url = (svn_http_server or '').rstrip('/')
-        url_path = urllib.parse.urlparse.urljoin(svn_http_server_url + '/', (path or '').lstrip('/'))
+        url_path = urllib.parse.urljoin(svn_http_server_url + '/', (path or '').lstrip('/'))
         url_path = urllib.parse.quote(url_path, safe="/:=~+!$,;'")
         return url_path
 
     def _get_request_headers(self, environ):
         headers = {}
 
         for key in environ:
             if not key.startswith('HTTP_'):
                 continue
             new_key = key.split('_')
             new_key = [k.capitalize() for k in new_key[1:]]
             new_key = '-'.join(new_key)
             headers[new_key] = environ[key]
 
         if 'CONTENT_TYPE' in environ:
             headers['Content-Type'] = environ['CONTENT_TYPE']
 
         if 'CONTENT_LENGTH' in environ:
             headers['Content-Length'] = environ['CONTENT_LENGTH']
 
         return headers
 
     def _get_response_headers(self, headers):
         headers = [
             (h, headers[h])
             for h in headers
             if h.lower() not in self.IGNORED_HEADERS
         ]
 
         return headers
 
 
 class DisabledSimpleSvnApp(object):
     def __init__(self, config):
         self.config = config
 
     def __call__(self, environ, start_response):
         reason = 'Cannot handle SVN call because: SVN HTTP Proxy is not enabled'
         log.warning(reason)
         return HTTPNotAcceptable(reason)(environ, start_response)
 
 
 class SimpleSvn(simplevcs.SimpleVCS):
 
     SCM = 'svn'
     READ_ONLY_COMMANDS = ('OPTIONS', 'PROPFIND', 'GET', 'REPORT')
     DEFAULT_HTTP_SERVER = 'http://localhost:8090'
 
     def _get_repository_name(self, environ):
         """
         Gets repository name out of PATH_INFO header
 
         :param environ: environ where PATH_INFO is stored
         """
         path = environ['PATH_INFO'].split('!')
         repo_name = path[0].strip('/')
 
         # SVN includes the whole path in it's requests, including
         # subdirectories inside the repo. Therefore we have to search for
         # the repo root directory.
         if not is_valid_repo(
                 repo_name, self.base_path, explicit_scm=self.SCM):
             current_path = ''
             for component in repo_name.split('/'):
                 current_path += component
                 if is_valid_repo(
                         current_path, self.base_path, explicit_scm=self.SCM):
                     return current_path
                 current_path += '/'
 
         return repo_name
 
     def _get_action(self, environ):
         return (
             'pull'
             if environ['REQUEST_METHOD'] in self.READ_ONLY_COMMANDS
             else 'push')
 
     def _should_use_callback_daemon(self, extras, environ, action):
         # only MERGE command triggers hooks, so we don't want to start
         # hooks server too many times. POST however starts the svn transaction
         # so we also need to run the init of callback daemon of POST
         if environ['REQUEST_METHOD'] in ['MERGE', 'POST']:
             return True
         return False
 
     def _create_wsgi_app(self, repo_path, repo_name, config):
         if self._is_svn_enabled():
             return SimpleSvnApp(config)
         # we don't have http proxy enabled return dummy request handler
         return DisabledSimpleSvnApp(config)
 
     def _is_svn_enabled(self):
         conf = self.repo_vcs_config
         return str2bool(conf.get('vcs_svn_proxy', 'http_requests_enabled'))
 
     def _create_config(self, extras, repo_name, scheme='http'):
         conf = self.repo_vcs_config
         server_url = conf.get('vcs_svn_proxy', 'http_server_url')
         server_url = server_url or self.DEFAULT_HTTP_SERVER
 
         extras['subversion_http_server_url'] = server_url
         return extras

rhodecode/lib/middleware/utils/scm_app_http.py

@@ -1,189 +1,189 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2014-2020 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/
 
 """
 Implementation of the scm_app interface using raw HTTP communication.
 """
 
 import base64
 import logging
 import urllib.parse
 import wsgiref.util
 
 import msgpack
 import requests
 import webob.request
 
 import rhodecode
 
 
 log = logging.getLogger(__name__)
 
 
 def create_git_wsgi_app(repo_path, repo_name, config):
     url = _vcs_streaming_url() + 'git/'
     return VcsHttpProxy(url, repo_path, repo_name, config)
 
 
 def create_hg_wsgi_app(repo_path, repo_name, config):
     url = _vcs_streaming_url() + 'hg/'
     return VcsHttpProxy(url, repo_path, repo_name, config)
 
 
 def _vcs_streaming_url():
     template = 'http://{}/stream/'
     return template.format(rhodecode.CONFIG['vcs.server'])
 
 
 # TODO: johbo: Avoid the global.
 session = requests.Session()
 # Requests speedup, avoid reading .netrc and similar
 session.trust_env = False
 
 # prevent urllib3 spawning our logs.
 logging.getLogger("requests.packages.urllib3.connectionpool").setLevel(
     logging.WARNING)
 
 
 class VcsHttpProxy(object):
     """
     A WSGI application which proxies vcs requests.
 
     The goal is to shuffle the data around without touching it. The only
     exception is the extra data from the config object which we send to the
     server as well.
     """
 
     def __init__(self, url, repo_path, repo_name, config):
         """
         :param str url: The URL of the VCSServer to call.
         """
         self._url = url
         self._repo_name = repo_name
         self._repo_path = repo_path
         self._config = config
         self.rc_extras = {}
         log.debug(
             "Creating VcsHttpProxy for repo %s, url %s",
             repo_name, url)
 
     def __call__(self, environ, start_response):
         config = msgpack.packb(self._config)
         request = webob.request.Request(environ)
         request_headers = request.headers
 
         request_headers.update({
             # TODO: johbo: Remove this, rely on URL path only
             'X-RC-Repo-Name': self._repo_name,
             'X-RC-Repo-Path': self._repo_path,
             'X-RC-Path-Info': environ['PATH_INFO'],
 
             'X-RC-Repo-Store': self.rc_extras.get('repo_store'),
             'X-RC-Server-Config-File': self.rc_extras.get('config'),
 
             'X-RC-Auth-User': self.rc_extras.get('username'),
             'X-RC-Auth-User-Id': str(self.rc_extras.get('user_id')),
             'X-RC-Auth-User-Ip': self.rc_extras.get('ip'),
 
             # TODO: johbo: Avoid encoding and put this into payload?
             'X-RC-Repo-Config': base64.b64encode(config),
             'X-RC-Locked-Status-Code': rhodecode.CONFIG.get('lock_ret_code'),
         })
 
         method = environ['REQUEST_METHOD']
 
         # Preserve the query string
         url = self._url
-        url = urllib.parse.urlparse.urljoin(url, self._repo_name)
+        url = urllib.parse.urljoin(url, self._repo_name)
         if environ.get('QUERY_STRING'):
             url += '?' + environ['QUERY_STRING']
 
         log.debug('http-app: preparing request to: %s', url)
         response = session.request(
             method,
             url,
             data=_maybe_stream_request(environ),
             headers=request_headers,
             stream=True)
 
         log.debug('http-app: got vcsserver response: %s', response)
         if response.status_code >= 500:
             log.error('Exception returned by vcsserver at: %s %s, %s',
                       url, response.status_code, response.content)
 
         # Preserve the headers of the response, except hop_by_hop ones
         response_headers = [
             (h, v) for h, v in response.headers.items()
             if not wsgiref.util.is_hop_by_hop(h)
         ]
 
         # Build status argument for start_response callable.
         status = '{status_code} {reason_phrase}'.format(
             status_code=response.status_code,
             reason_phrase=response.reason)
 
         start_response(status, response_headers)
         return _maybe_stream_response(response)
 
 
 def read_in_chunks(stream_obj, block_size=1024, chunks=-1):
     """
     Read Stream in chunks, default chunk size: 1k.
     """
     while chunks:
         data = stream_obj.read(block_size)
         if not data:
             break
         yield data
         chunks -= 1
 
 
 def _is_request_chunked(environ):
     stream = environ.get('HTTP_TRANSFER_ENCODING', '') == 'chunked'
     return stream
 
 
 def _maybe_stream_request(environ):
     path = environ['PATH_INFO']
     stream = _is_request_chunked(environ)
     log.debug('handling request `%s` with stream support: %s', path, stream)
 
     if stream:
         # set stream by 256k
         return read_in_chunks(environ['wsgi.input'], block_size=1024 * 256)
     else:
         return environ['wsgi.input'].read()
 
 
 def _maybe_stream_response(response):
     """
     Try to generate chunks from the response if it is chunked.
     """
     stream = _is_chunked(response)
     log.debug('returning response with stream: %s', stream)
     if stream:
         # read in 256k Chunks
         return response.raw.read_chunked(amt=1024 * 256)
     else:
         return [response.content]
 
 
 def _is_chunked(response):
     return response.headers.get('Transfer-Encoding', '') == 'chunked'

rhodecode/lib/vcs/client_http.py

@@ -1,412 +1,412 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2020 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/
 
 """
 Client for the VCSServer implemented based on HTTP.
 """
 
 import copy
 import logging
 import threading
 import time
 import urllib.request, urllib.error, urllib.parse
 import urllib.parse
 import uuid
 import traceback
 
 import pycurl
 import msgpack
 import requests
 from requests.packages.urllib3.util.retry import Retry
 
 import rhodecode
 from rhodecode.lib import rc_cache
 from rhodecode.lib.rc_cache.utils import compute_key_from_params
 from rhodecode.lib.system_info import get_cert_path
 from rhodecode.lib.vcs import exceptions, CurlSession
 from rhodecode.lib.utils2 import str2bool
 
 log = logging.getLogger(__name__)
 
 
 # TODO: mikhail: Keep it in sync with vcsserver's
 # HTTPApplication.ALLOWED_EXCEPTIONS
 EXCEPTIONS_MAP = {
     'KeyError': KeyError,
     'URLError': urllib.error.URLError,
 }
 
 
 def _remote_call(url, payload, exceptions_map, session):
     try:
         headers = {
             'X-RC-Method': payload.get('method'),
             'X-RC-Repo-Name': payload.get('_repo_name')
         }
         response = session.post(url, data=msgpack.packb(payload), headers=headers)
     except pycurl.error as e:
         msg = '{}. \npycurl traceback: {}'.format(e, traceback.format_exc())
         raise exceptions.HttpVCSCommunicationError(msg)
     except Exception as e:
         message = getattr(e, 'message', '')
         if 'Failed to connect' in message:
             # gevent doesn't return proper pycurl errors
             raise exceptions.HttpVCSCommunicationError(e)
         else:
             raise
 
     if response.status_code >= 400:
         log.error('Call to %s returned non 200 HTTP code: %s',
                   url, response.status_code)
         raise exceptions.HttpVCSCommunicationError(repr(response.content))
 
     try:
         response = msgpack.unpackb(response.content)
     except Exception:
         log.exception('Failed to decode response %r', response.content)
         raise
 
     error = response.get('error')
     if error:
         type_ = error.get('type', 'Exception')
         exc = exceptions_map.get(type_, Exception)
         exc = exc(error.get('message'))
         try:
             exc._vcs_kind = error['_vcs_kind']
         except KeyError:
             pass
 
         try:
             exc._vcs_server_traceback = error['traceback']
             exc._vcs_server_org_exc_name = error['org_exc']
             exc._vcs_server_org_exc_tb = error['org_exc_tb']
         except KeyError:
             pass
 
         raise exc
     return response.get('result')
 
 
 def _streaming_remote_call(url, payload, exceptions_map, session, chunk_size):
     try:
         headers = {
             'X-RC-Method': payload.get('method'),
             'X-RC-Repo-Name': payload.get('_repo_name')
         }
         response = session.post(url, data=msgpack.packb(payload), headers=headers)
     except pycurl.error as e:
         msg = '{}. \npycurl traceback: {}'.format(e, traceback.format_exc())
         raise exceptions.HttpVCSCommunicationError(msg)
     except Exception as e:
         message = getattr(e, 'message', '')
         if 'Failed to connect' in message:
             # gevent doesn't return proper pycurl errors
             raise exceptions.HttpVCSCommunicationError(e)
         else:
             raise
 
     if response.status_code >= 400:
         log.error('Call to %s returned non 200 HTTP code: %s',
                   url, response.status_code)
         raise exceptions.HttpVCSCommunicationError(repr(response.content))
 
     return response.iter_content(chunk_size=chunk_size)
 
 
 class ServiceConnection(object):
     def __init__(self, server_and_port, backend_endpoint, session_factory):
-        self.url = urllib.parse.urlparse.urljoin('http://%s' % server_and_port, backend_endpoint)
+        self.url = urllib.parse.urljoin('http://%s' % server_and_port, backend_endpoint)
         self._session_factory = session_factory
 
     def __getattr__(self, name):
         def f(*args, **kwargs):
             return self._call(name, *args, **kwargs)
         return f
 
     @exceptions.map_vcs_exceptions
     def _call(self, name, *args, **kwargs):
         payload = {
             'id': str(uuid.uuid4()),
             'method': name,
             'params': {'args': args, 'kwargs': kwargs}
         }
         return _remote_call(
             self.url, payload, EXCEPTIONS_MAP, self._session_factory())
 
 
 class RemoteVCSMaker(object):
 
     def __init__(self, server_and_port, backend_endpoint, backend_type, session_factory):
-        self.url = urllib.parse.urlparse.urljoin('http://%s' % server_and_port, backend_endpoint)
-        self.stream_url = urllib.parse.urlparse.urljoin('http://%s' % server_and_port, backend_endpoint+'/stream')
+        self.url = urllib.parse.urljoin('http://%s' % server_and_port, backend_endpoint)
+        self.stream_url = urllib.parse.urljoin('http://%s' % server_and_port, backend_endpoint+'/stream')
 
         self._session_factory = session_factory
         self.backend_type = backend_type
 
     @classmethod
     def init_cache_region(cls, repo_id):
         cache_namespace_uid = 'cache_repo.{}'.format(repo_id)
         region = rc_cache.get_or_create_region('cache_repo', cache_namespace_uid)
         return region, cache_namespace_uid
 
     def __call__(self, path, repo_id, config, with_wire=None):
         log.debug('%s RepoMaker call on %s', self.backend_type.upper(), path)
         return RemoteRepo(path, repo_id, config, self, with_wire=with_wire)
 
     def __getattr__(self, name):
         def remote_attr(*args, **kwargs):
             return self._call(name, *args, **kwargs)
         return remote_attr
 
     @exceptions.map_vcs_exceptions
     def _call(self, func_name, *args, **kwargs):
         payload = {
             'id': str(uuid.uuid4()),
             'method': func_name,
             'backend': self.backend_type,
             'params': {'args': args, 'kwargs': kwargs}
         }
         url = self.url
         return _remote_call(url, payload, EXCEPTIONS_MAP, self._session_factory())
 
 
 class RemoteRepo(object):
     CHUNK_SIZE = 16384
 
     def __init__(self, path, repo_id, config, remote_maker, with_wire=None):
         self.url = remote_maker.url
         self.stream_url = remote_maker.stream_url
         self._session = remote_maker._session_factory()
 
         cache_repo_id = self._repo_id_sanitizer(repo_id)
         _repo_name = self._get_repo_name(config, path)
         self._cache_region, self._cache_namespace = \
             remote_maker.init_cache_region(cache_repo_id)
 
         with_wire = with_wire or {}
 
         repo_state_uid = with_wire.get('repo_state_uid') or 'state'
 
         self._wire = {
             "_repo_name": _repo_name,
             "path": path,  # repo path
             "repo_id": repo_id,
             "cache_repo_id": cache_repo_id,
             "config": config,
             "repo_state_uid": repo_state_uid,
             "context": self._create_vcs_cache_context(path, repo_state_uid)
         }
 
         if with_wire:
             self._wire.update(with_wire)
 
         # NOTE(johbo): Trading complexity for performance. Avoiding the call to
         # log.debug brings a few percent gain even if is is not active.
         if log.isEnabledFor(logging.DEBUG):
             self._call_with_logging = True
 
         self.cert_dir = get_cert_path(rhodecode.CONFIG.get('__file__'))
 
     def _get_repo_name(self, config, path):
         repo_store = config.get('paths', '/')
         return path.split(repo_store)[-1].lstrip('/')
 
     def _repo_id_sanitizer(self, repo_id):
         pathless = repo_id.replace('/', '__').replace('-', '_')
         return ''.join(char if ord(char) < 128 else '_{}_'.format(ord(char)) for char in pathless)
 
     def __getattr__(self, name):
 
         if name.startswith('stream:'):
             def repo_remote_attr(*args, **kwargs):
                 return self._call_stream(name, *args, **kwargs)
         else:
             def repo_remote_attr(*args, **kwargs):
                 return self._call(name, *args, **kwargs)
 
         return repo_remote_attr
 
     def _base_call(self, name, *args, **kwargs):
         # TODO: oliver: This is currently necessary pre-call since the
         # config object is being changed for hooking scenarios
         wire = copy.deepcopy(self._wire)
         wire["config"] = wire["config"].serialize()
         wire["config"].append(('vcs', 'ssl_dir', self.cert_dir))
 
         payload = {
             'id': str(uuid.uuid4()),
             'method': name,
             "_repo_name":  wire['_repo_name'],
             'params': {'wire': wire, 'args': args, 'kwargs': kwargs}
         }
 
         context_uid = wire.get('context')
         return context_uid, payload
 
     def get_local_cache(self, name, args):
         cache_on = False
         cache_key = ''
         local_cache_on = str2bool(rhodecode.CONFIG.get('vcs.methods.cache'))
 
         cache_methods = [
             'branches', 'tags', 'bookmarks',
             'is_large_file', 'is_binary',
             'fctx_size', 'stream:fctx_node_data', 'blob_raw_length',
             'node_history',
             'revision', 'tree_items',
             'ctx_list', 'ctx_branch', 'ctx_description',
             'bulk_request',
             'assert_correct_path'
         ]
 
         if local_cache_on and name in cache_methods:
             cache_on = True
             repo_state_uid = self._wire['repo_state_uid']
             call_args = [a for a in args]
             cache_key = compute_key_from_params(repo_state_uid, name, *call_args)
 
         return cache_on, cache_key
 
     @exceptions.map_vcs_exceptions
     def _call(self, name, *args, **kwargs):
         context_uid, payload = self._base_call(name, *args, **kwargs)
         url = self.url
 
         start = time.time()
         cache_on, cache_key = self.get_local_cache(name, args)
 
         @self._cache_region.conditional_cache_on_arguments(
             namespace=self._cache_namespace, condition=cache_on and cache_key)
         def remote_call(_cache_key):
             if self._call_with_logging:
                 log.debug('Calling %s@%s with args:%.10240r. wire_context: %s cache_on: %s',
                           url, name, args, context_uid, cache_on)
             return _remote_call(url, payload, EXCEPTIONS_MAP, self._session)
 
         result = remote_call(cache_key)
         if self._call_with_logging:
             log.debug('Call %s@%s took: %.4fs. wire_context: %s',
                       url, name, time.time()-start, context_uid)
         return result
 
     @exceptions.map_vcs_exceptions
     def _call_stream(self, name, *args, **kwargs):
         context_uid, payload = self._base_call(name, *args, **kwargs)
         payload['chunk_size'] = self.CHUNK_SIZE
         url = self.stream_url
 
         start = time.time()
         cache_on, cache_key = self.get_local_cache(name, args)
 
         # Cache is a problem because this is a stream
         def streaming_remote_call(_cache_key):
             if self._call_with_logging:
                 log.debug('Calling %s@%s with args:%.10240r. wire_context: %s cache_on: %s',
                           url, name, args, context_uid, cache_on)
             return _streaming_remote_call(url, payload, EXCEPTIONS_MAP, self._session, self.CHUNK_SIZE)
 
         result = streaming_remote_call(cache_key)
         if self._call_with_logging:
             log.debug('Call %s@%s took: %.4fs. wire_context: %s',
                       url, name, time.time()-start, context_uid)
         return result
 
     def __getitem__(self, key):
         return self.revision(key)
 
     def _create_vcs_cache_context(self, *args):
         """
         Creates a unique string which is passed to the VCSServer on every
         remote call. It is used as cache key in the VCSServer.
         """
         hash_key = '-'.join(map(str, args))
         return str(uuid.uuid5(uuid.NAMESPACE_URL, hash_key))
 
     def invalidate_vcs_cache(self):
         """
         This invalidates the context which is sent to the VCSServer on every
         call to a remote method. It forces the VCSServer to create a fresh
         repository instance on the next call to a remote method.
         """
         self._wire['context'] = str(uuid.uuid4())
 
 
 class VcsHttpProxy(object):
 
     CHUNK_SIZE = 16384
 
     def __init__(self, server_and_port, backend_endpoint):
         retries = Retry(total=5, connect=None, read=None, redirect=None)
 
         adapter = requests.adapters.HTTPAdapter(max_retries=retries)
-        self.base_url = urllib.parse.urlparse.urljoin('http://%s' % server_and_port, backend_endpoint)
+        self.base_url = urllib.parse.urljoin('http://%s' % server_and_port, backend_endpoint)
         self.session = requests.Session()
         self.session.mount('http://', adapter)
 
     def handle(self, environment, input_data, *args, **kwargs):
         data = {
             'environment': environment,
             'input_data': input_data,
             'args': args,
             'kwargs': kwargs
         }
         result = self.session.post(
             self.base_url, msgpack.packb(data), stream=True)
         return self._get_result(result)
 
     def _deserialize_and_raise(self, error):
         exception = Exception(error['message'])
         try:
             exception._vcs_kind = error['_vcs_kind']
         except KeyError:
             pass
         raise exception
 
     def _iterate(self, result):
         unpacker = msgpack.Unpacker()
         for line in result.iter_content(chunk_size=self.CHUNK_SIZE):
             unpacker.feed(line)
             for chunk in unpacker:
                 yield chunk
 
     def _get_result(self, result):
         iterator = self._iterate(result)
         error = next(iterator)
         if error:
             self._deserialize_and_raise(error)
 
         status = next(iterator)
         headers = next(iterator)
 
         return iterator, status, headers
 
 
 class ThreadlocalSessionFactory(object):
     """
     Creates one CurlSession per thread on demand.
     """
 
     def __init__(self):
         self._thread_local = threading.local()
 
     def __call__(self):
         if not hasattr(self._thread_local, 'curl_session'):
             self._thread_local.curl_session = CurlSession()
         return self._thread_local.curl_session

rhodecode/tests/lib/middleware/test_simplegit.py

@@ -1,140 +1,140 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 pytest
 import urllib.parse
 import mock
 import simplejson as json
 
 from rhodecode.lib.vcs.backends.base import Config
 from rhodecode.tests.lib.middleware import mock_scm_app
 import rhodecode.lib.middleware.simplegit as simplegit
 
 
 def get_environ(url, request_method):
     """Construct a minimum WSGI environ based on the URL."""
-    parsed_url = urllib.parse.urlparse.urlparse(url)
+    parsed_url = urllib.parse.urlparse(url)
     environ = {
         'PATH_INFO': parsed_url.path,
         'QUERY_STRING': parsed_url.query,
         'REQUEST_METHOD': request_method,
     }
 
     return environ
 
 
 @pytest.mark.parametrize(
     'url, expected_action, request_method',
     [
         ('/foo/bar/info/refs?service=git-upload-pack', 'pull', 'GET'),
         ('/foo/bar/info/refs?service=git-receive-pack', 'push', 'GET'),
         ('/foo/bar/git-upload-pack', 'pull', 'GET'),
         ('/foo/bar/git-receive-pack', 'push', 'GET'),
         # Edge case: missing data for info/refs
         ('/foo/info/refs?service=', 'pull', 'GET'),
         ('/foo/info/refs', 'pull', 'GET'),
         # Edge case: git command comes with service argument
         ('/foo/git-upload-pack?service=git-receive-pack', 'pull', 'GET'),
         ('/foo/git-receive-pack?service=git-upload-pack', 'push', 'GET'),
         # Edge case: repo name conflicts with git commands
         ('/git-receive-pack/git-upload-pack', 'pull', 'GET'),
         ('/git-receive-pack/git-receive-pack', 'push', 'GET'),
         ('/git-upload-pack/git-upload-pack', 'pull', 'GET'),
         ('/git-upload-pack/git-receive-pack', 'push', 'GET'),
         ('/foo/git-receive-pack', 'push', 'GET'),
         # Edge case: not a smart protocol url
         ('/foo/bar', 'pull', 'GET'),
         # GIT LFS cases, batch
         ('/foo/bar/info/lfs/objects/batch', 'push', 'GET'),
         ('/foo/bar/info/lfs/objects/batch', 'pull', 'POST'),
         # GIT LFS oid, dl/upl
         ('/foo/bar/info/lfs/abcdeabcde', 'pull', 'GET'),
         ('/foo/bar/info/lfs/abcdeabcde', 'push', 'PUT'),
         ('/foo/bar/info/lfs/abcdeabcde', 'push', 'POST'),
         # Edge case: repo name conflicts with git commands
         ('/info/lfs/info/lfs/objects/batch', 'push', 'GET'),
         ('/info/lfs/info/lfs/objects/batch', 'pull', 'POST'),
 
     ])
 def test_get_action(url, expected_action, request_method, baseapp, request_stub):
     app = simplegit.SimpleGit(config={'auth_ret_code': '', 'base_path': ''},
                               registry=request_stub.registry)
     assert expected_action == app._get_action(get_environ(url, request_method))
 
 
 @pytest.mark.parametrize(
     'url, expected_repo_name, request_method',
     [
         ('/foo/info/refs?service=git-upload-pack', 'foo', 'GET'),
         ('/foo/bar/info/refs?service=git-receive-pack', 'foo/bar', 'GET'),
         ('/foo/git-upload-pack', 'foo', 'GET'),
         ('/foo/git-receive-pack', 'foo', 'GET'),
         ('/foo/bar/git-upload-pack', 'foo/bar', 'GET'),
         ('/foo/bar/git-receive-pack', 'foo/bar', 'GET'),
 
         # GIT LFS cases, batch
         ('/foo/bar/info/lfs/objects/batch', 'foo/bar', 'GET'),
         ('/example-git/info/lfs/objects/batch', 'example-git', 'POST'),
         # GIT LFS oid, dl/upl
         ('/foo/info/lfs/abcdeabcde', 'foo', 'GET'),
         ('/foo/bar/info/lfs/abcdeabcde', 'foo/bar', 'PUT'),
         ('/my-git-repo/info/lfs/abcdeabcde', 'my-git-repo', 'POST'),
         # Edge case: repo name conflicts with git commands
         ('/info/lfs/info/lfs/objects/batch', 'info/lfs', 'GET'),
         ('/info/lfs/info/lfs/objects/batch', 'info/lfs', 'POST'),
 
     ])
 def test_get_repository_name(url, expected_repo_name, request_method, baseapp, request_stub):
     app = simplegit.SimpleGit(config={'auth_ret_code': '', 'base_path': ''},
                               registry=request_stub.registry)
     assert expected_repo_name == app._get_repository_name(
         get_environ(url, request_method))
 
 
 def test_get_config(user_util, baseapp, request_stub):
     repo = user_util.create_repo(repo_type='git')
     app = simplegit.SimpleGit(config={'auth_ret_code': '', 'base_path': ''},
                               registry=request_stub.registry)
     extras = {'foo': 'FOO', 'bar': 'BAR'}
 
     # We copy the extras as the method below will change the contents.
     git_config = app._create_config(dict(extras), repo_name=repo.repo_name)
 
     expected_config = dict(extras)
     expected_config.update({
         'git_update_server_info': False,
         'git_lfs_enabled': False,
         'git_lfs_store_path': git_config['git_lfs_store_path'],
         'git_lfs_http_scheme': 'http'
     })
 
     assert git_config == expected_config
 
 
 def test_create_wsgi_app_uses_scm_app_from_simplevcs(baseapp, request_stub):
     config = {
         'auth_ret_code': '',
         'base_path': '',
         'vcs.scm_app_implementation':
             'rhodecode.tests.lib.middleware.mock_scm_app',
     }
     app = simplegit.SimpleGit(config=config, registry=request_stub.registry)
     wsgi_app = app._create_wsgi_app('/tmp/test', 'test_repo', {})
     assert wsgi_app is mock_scm_app.mock_git_wsgi

rhodecode/tests/lib/middleware/test_simplehg.py

@@ -1,156 +1,156 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import urllib.parse
 
 import mock
 import pytest
 import simplejson as json
 
 from rhodecode.lib.vcs.backends.base import Config
 from rhodecode.tests.lib.middleware import mock_scm_app
 import rhodecode.lib.middleware.simplehg as simplehg
 
 
 def get_environ(url):
     """Construct a minimum WSGI environ based on the URL."""
-    parsed_url = urllib.parse.urlparse.urlparse(url)
+    parsed_url = urllib.parse.urlparse(url)
     environ = {
         'PATH_INFO': parsed_url.path,
         'QUERY_STRING': parsed_url.query,
     }
 
     return environ
 
 
 @pytest.mark.parametrize(
     'url, expected_action',
     [
         ('/foo/bar?cmd=unbundle&key=tip', 'push'),
         ('/foo/bar?cmd=pushkey&key=tip', 'push'),
         ('/foo/bar?cmd=listkeys&key=tip', 'pull'),
         ('/foo/bar?cmd=changegroup&key=tip', 'pull'),
         ('/foo/bar?cmd=hello', 'pull'),
         ('/foo/bar?cmd=batch', 'push'),
         ('/foo/bar?cmd=putlfile', 'push'),
         # Edge case: unknown argument: assume push
         ('/foo/bar?cmd=unknown&key=tip', 'push'),
         ('/foo/bar?cmd=&key=tip', 'push'),
         # Edge case: not cmd argument
         ('/foo/bar?key=tip', 'push'),
     ])
 def test_get_action(url, expected_action, request_stub):
     app = simplehg.SimpleHg(config={'auth_ret_code': '', 'base_path': ''},
                             registry=request_stub.registry)
     assert expected_action == app._get_action(get_environ(url))
 
 
 @pytest.mark.parametrize(
     'environ, expected_xargs, expected_batch',
     [
         ({},
          [''], ['push']),
 
         ({'HTTP_X_HGARG_1': ''},
          [''], ['push']),
 
         ({'HTTP_X_HGARG_1': 'cmds=listkeys+namespace%3Dphases'},
          ['listkeys namespace=phases'], ['pull']),
 
         ({'HTTP_X_HGARG_1': 'cmds=pushkey+namespace%3Dbookmarks%2Ckey%3Dbm%2Cold%3D%2Cnew%3Dcb9a9f314b8b07ba71012fcdbc544b5a4d82ff5b'},
          ['pushkey namespace=bookmarks,key=bm,old=,new=cb9a9f314b8b07ba71012fcdbc544b5a4d82ff5b'], ['push']),
 
         ({'HTTP_X_HGARG_1': 'namespace=phases'},
          ['namespace=phases'], ['push']),
 
     ])
 def test_xarg_and_batch_commands(environ, expected_xargs, expected_batch):
     app = simplehg.SimpleHg
 
     result = app._get_xarg_headers(environ)
     result_batch = app._get_batch_cmd(environ)
     assert expected_xargs == result
     assert expected_batch == result_batch
 
 
 @pytest.mark.parametrize(
     'url, expected_repo_name',
     [
         ('/foo?cmd=unbundle&key=tip', 'foo'),
         ('/foo/bar?cmd=pushkey&key=tip', 'foo/bar'),
         ('/foo/bar/baz?cmd=listkeys&key=tip', 'foo/bar/baz'),
         # Repos with trailing slashes.
         ('/foo/?cmd=unbundle&key=tip', 'foo'),
         ('/foo/bar/?cmd=pushkey&key=tip', 'foo/bar'),
         ('/foo/bar/baz/?cmd=listkeys&key=tip', 'foo/bar/baz'),
     ])
 def test_get_repository_name(url, expected_repo_name, request_stub):
     app = simplehg.SimpleHg(config={'auth_ret_code': '', 'base_path': ''},
                             registry=request_stub.registry)
     assert expected_repo_name == app._get_repository_name(get_environ(url))
 
 
 def test_get_config(user_util, baseapp, request_stub):
     repo = user_util.create_repo(repo_type='git')
     app = simplehg.SimpleHg(config={'auth_ret_code': '', 'base_path': ''},
                             registry=request_stub.registry)
     extras = [('foo', 'FOO', 'bar', 'BAR')]
 
     hg_config = app._create_config(extras, repo_name=repo.repo_name)
 
     config = simplehg.utils.make_db_config(repo=repo.repo_name)
     config.set('rhodecode', 'RC_SCM_DATA', json.dumps(extras))
     hg_config_org = config
 
     expected_config = [
         ('vcs_svn_tag', 'ff89f8c714d135d865f44b90e5413b88de19a55f', '/tags/*'),
         ('web', 'push_ssl', 'False'),
         ('web', 'allow_push', '*'),
         ('web', 'allow_archive', 'gz zip bz2'),
         ('web', 'baseurl', '/'),
         ('vcs_git_lfs', 'store_location', hg_config_org.get('vcs_git_lfs', 'store_location')),
         ('vcs_svn_branch', '9aac1a38c3b8a0cdc4ae0f960a5f83332bc4fa5e', '/branches/*'),
         ('vcs_svn_branch', 'c7e6a611c87da06529fd0dd733308481d67c71a8', '/trunk'),
         ('largefiles', 'usercache', hg_config_org.get('largefiles', 'usercache')),
         ('hooks', 'preoutgoing.pre_pull', 'python:vcsserver.hooks.pre_pull'),
         ('hooks', 'prechangegroup.pre_push', 'python:vcsserver.hooks.pre_push'),
         ('hooks', 'outgoing.pull_logger', 'python:vcsserver.hooks.log_pull_action'),
         ('hooks', 'pretxnchangegroup.pre_push', 'python:vcsserver.hooks.pre_push'),
         ('hooks', 'changegroup.push_logger', 'python:vcsserver.hooks.log_push_action'),
         ('hooks', 'changegroup.repo_size', 'python:vcsserver.hooks.repo_size'),
         ('phases', 'publish', 'True'),
         ('extensions', 'largefiles', ''),
         ('paths', '/', hg_config_org.get('paths', '/')),
         ('rhodecode', 'RC_SCM_DATA', '[["foo", "FOO", "bar", "BAR"]]')
     ]
     for entry in expected_config:
         assert entry in hg_config
 
 
 def test_create_wsgi_app_uses_scm_app_from_simplevcs(request_stub):
     config = {
         'auth_ret_code': '',
         'base_path': '',
         'vcs.scm_app_implementation':
             'rhodecode.tests.lib.middleware.mock_scm_app',
     }
     app = simplehg.SimpleHg(config=config, registry=request_stub.registry)
     wsgi_app = app._create_wsgi_app('/tmp/test', 'test_repo', {})
     assert wsgi_app is mock_scm_app.mock_hg_wsgi

rhodecode/tests/utils.py

@@ -1,468 +1,467 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 threading
 import time
 import logging
 import os.path
 import subprocess
 import tempfile
 import urllib.request, urllib.error, urllib.parse
 from lxml.html import fromstring, tostring
 from lxml.cssselect import CSSSelector
-import urllib.parse.urlparse
 from urllib.parse import unquote_plus
 import webob
 
 from webtest.app import TestResponse, TestApp
 from webtest.compat import print_stderr
 
 import pytest
 import rc_testdata
 
 from rhodecode.model.db import User, Repository
 from rhodecode.model.meta import Session
 from rhodecode.model.scm import ScmModel
 from rhodecode.lib.vcs.backends.svn.repository import SubversionRepository
 from rhodecode.lib.vcs.backends.base import EmptyCommit
 from rhodecode.tests import login_user_session
 
 log = logging.getLogger(__name__)
 
 
 class CustomTestResponse(TestResponse):
 
     def _save_output(self, out):
         f = tempfile.NamedTemporaryFile(delete=False, prefix='rc-test-', suffix='.html')
         f.write(out)
         return f.name
 
     def mustcontain(self, *strings, **kw):
         """
         Assert that the response contains all of the strings passed
         in as arguments.
 
         Equivalent to::
 
             assert string in res
         """
         print_body = kw.pop('print_body', False)
         if 'no' in kw:
             no = kw['no']
             del kw['no']
             if isinstance(no, str):
                 no = [no]
         else:
             no = []
         if kw:
             raise TypeError(
                 "The only keyword argument allowed is 'no' got %s" % kw)
 
         f = self._save_output(str(self))
 
         for s in strings:
             if not s in self:
                 print_stderr("Actual response (no %r):" % s)
                 print_stderr("body output saved as `%s`" % f)
                 if print_body:
                     print_stderr(str(self))
                 raise IndexError(
                     "Body does not contain string %r, body output saved as %s" % (s, f))
 
         for no_s in no:
             if no_s in self:
                 print_stderr("Actual response (has %r)" % no_s)
                 print_stderr("body output saved as `%s`" % f)
                 if print_body:
                     print_stderr(str(self))
                 raise IndexError(
                     "Body contains bad string %r, body output saved as %s" % (no_s, f))
 
     def assert_response(self):
         return AssertResponse(self)
 
     def get_session_from_response(self):
         """
         This returns the session from a response object.
         """
         from rhodecode.lib.rc_beaker import session_factory_from_settings
         session = session_factory_from_settings(self.test_app._pyramid_settings)
         return session(self.request)
 
 
 class TestRequest(webob.BaseRequest):
 
     # for py.test
     disabled = True
     ResponseClass = CustomTestResponse
 
     def add_response_callback(self, callback):
         pass
 
 
 class CustomTestApp(TestApp):
     """
     Custom app to make mustcontain more Useful, and extract special methods
     """
     RequestClass = TestRequest
     rc_login_data = {}
     rc_current_session = None
 
     def login(self, username=None, password=None):
         from rhodecode.lib import auth
 
         if username and password:
             session = login_user_session(self, username, password)
         else:
             session = login_user_session(self)
 
         self.rc_login_data['csrf_token'] = auth.get_csrf_token(session)
         self.rc_current_session = session
         return session['rhodecode_user']
 
     @property
     def csrf_token(self):
         return self.rc_login_data['csrf_token']
 
     @property
     def _pyramid_registry(self):
         return self.app.config.registry
 
     @property
     def _pyramid_settings(self):
         return self._pyramid_registry.settings
 
 
 def set_anonymous_access(enabled):
     """(Dis)allows anonymous access depending on parameter `enabled`"""
     user = User.get_default_user()
     user.active = enabled
     Session().add(user)
     Session().commit()
     time.sleep(1.5)  # must sleep for cache (1s to expire)
     log.info('anonymous access is now: %s', enabled)
     assert enabled == User.get_default_user().active, (
         'Cannot set anonymous access')
 
 
 def check_xfail_backends(node, backend_alias):
     # Using "xfail_backends" here intentionally, since this marks work
     # which is "to be done" soon.
     skip_marker = node.get_closest_marker('xfail_backends')
     if skip_marker and backend_alias in skip_marker.args:
         msg = "Support for backend %s to be developed." % (backend_alias, )
         msg = skip_marker.kwargs.get('reason', msg)
         pytest.xfail(msg)
 
 
 def check_skip_backends(node, backend_alias):
     # Using "skip_backends" here intentionally, since this marks work which is
     # not supported.
     skip_marker = node.get_closest_marker('skip_backends')
     if skip_marker and backend_alias in skip_marker.args:
         msg = "Feature not supported for backend %s." % (backend_alias, )
         msg = skip_marker.kwargs.get('reason', msg)
         pytest.skip(msg)
 
 
 def extract_git_repo_from_dump(dump_name, repo_name):
     """Create git repo `repo_name` from dump `dump_name`."""
     repos_path = ScmModel().repos_path
     target_path = os.path.join(repos_path, repo_name)
     rc_testdata.extract_git_dump(dump_name, target_path)
     return target_path
 
 
 def extract_hg_repo_from_dump(dump_name, repo_name):
     """Create hg repo `repo_name` from dump `dump_name`."""
     repos_path = ScmModel().repos_path
     target_path = os.path.join(repos_path, repo_name)
     rc_testdata.extract_hg_dump(dump_name, target_path)
     return target_path
 
 
 def extract_svn_repo_from_dump(dump_name, repo_name):
     """Create a svn repo `repo_name` from dump `dump_name`."""
     repos_path = ScmModel().repos_path
     target_path = os.path.join(repos_path, repo_name)
     SubversionRepository(target_path, create=True)
     _load_svn_dump_into_repo(dump_name, target_path)
     return target_path
 
 
 def assert_message_in_log(log_records, message, levelno, module):
     messages = [
         r.message for r in log_records
         if r.module == module and r.levelno == levelno
     ]
     assert message in messages
 
 
 def _load_svn_dump_into_repo(dump_name, repo_path):
     """
     Utility to populate a svn repository with a named dump
 
     Currently the dumps are in rc_testdata. They might later on be
     integrated with the main repository once they stabilize more.
     """
     dump = rc_testdata.load_svn_dump(dump_name)
     load_dump = subprocess.Popen(
         ['svnadmin', 'load', repo_path],
         stdin=subprocess.PIPE, stdout=subprocess.PIPE,
         stderr=subprocess.PIPE)
     out, err = load_dump.communicate(dump)
     if load_dump.returncode != 0:
         log.error("Output of load_dump command: %s", out)
         log.error("Error output of load_dump command: %s", err)
         raise Exception(
             'Failed to load dump "%s" into repository at path "%s".'
             % (dump_name, repo_path))
 
 
 class AssertResponse(object):
     """
     Utility that helps to assert things about a given HTML response.
     """
 
     def __init__(self, response):
         self.response = response
 
     def get_imports(self):
         return fromstring, tostring, CSSSelector
 
     def one_element_exists(self, css_selector):
         self.get_element(css_selector)
 
     def no_element_exists(self, css_selector):
         assert not self._get_elements(css_selector)
 
     def element_equals_to(self, css_selector, expected_content):
         element = self.get_element(css_selector)
         element_text = self._element_to_string(element)
         assert expected_content in element_text
 
     def element_contains(self, css_selector, expected_content):
         element = self.get_element(css_selector)
         assert expected_content in element.text_content()
 
     def element_value_contains(self, css_selector, expected_content):
         element = self.get_element(css_selector)
         assert expected_content in element.value
 
     def contains_one_link(self, link_text, href):
         fromstring, tostring, CSSSelector = self.get_imports()
         doc = fromstring(self.response.body)
         sel = CSSSelector('a[href]')
         elements = [
             e for e in sel(doc) if e.text_content().strip() == link_text]
         assert len(elements) == 1, "Did not find link or found multiple links"
         self._ensure_url_equal(elements[0].attrib.get('href'), href)
 
     def contains_one_anchor(self, anchor_id):
         fromstring, tostring, CSSSelector = self.get_imports()
         doc = fromstring(self.response.body)
         sel = CSSSelector('#' + anchor_id)
         elements = sel(doc)
         assert len(elements) == 1, 'cannot find 1 element {}'.format(anchor_id)
 
     def _ensure_url_equal(self, found, expected):
         assert _Url(found) == _Url(expected)
 
     def get_element(self, css_selector):
         elements = self._get_elements(css_selector)
         assert len(elements) == 1, 'cannot find 1 element {}'.format(css_selector)
         return elements[0]
 
     def get_elements(self, css_selector):
         return self._get_elements(css_selector)
 
     def _get_elements(self, css_selector):
         fromstring, tostring, CSSSelector = self.get_imports()
         doc = fromstring(self.response.body)
         sel = CSSSelector(css_selector)
         elements = sel(doc)
         return elements
 
     def _element_to_string(self, element):
         fromstring, tostring, CSSSelector = self.get_imports()
         return tostring(element)
 
 
 class _Url(object):
     """
     A url object that can be compared with other url orbjects
     without regard to the vagaries of encoding, escaping, and ordering
     of parameters in query strings.
 
     Inspired by
     http://stackoverflow.com/questions/5371992/comparing-two-urls-in-python
     """
 
     def __init__(self, url):
         parts = urllib.parse.urlparse(url)
         _query = frozenset(urllib.parse.parse_qsl(parts.query))
         _path = unquote_plus(parts.path)
         parts = parts._replace(query=_query, path=_path)
         self.parts = parts
 
     def __eq__(self, other):
         return self.parts == other.parts
 
     def __hash__(self):
         return hash(self.parts)
 
 
 def run_test_concurrently(times, raise_catched_exc=True):
     """
     Add this decorator to small pieces of code that you want to test
     concurrently
 
     ex:
 
     @test_concurrently(25)
     def my_test_function():
         ...
     """
     def test_concurrently_decorator(test_func):
         def wrapper(*args, **kwargs):
             exceptions = []
 
             def call_test_func():
                 try:
                     test_func(*args, **kwargs)
                 except Exception as e:
                     exceptions.append(e)
                     if raise_catched_exc:
                         raise
             threads = []
             for i in range(times):
                 threads.append(threading.Thread(target=call_test_func))
             for t in threads:
                 t.start()
             for t in threads:
                 t.join()
             if exceptions:
                 raise Exception(
                     'test_concurrently intercepted %s exceptions: %s' % (
                         len(exceptions), exceptions))
         return wrapper
     return test_concurrently_decorator
 
 
 def wait_for_url(url, timeout=10):
     """
     Wait until URL becomes reachable.
 
     It polls the URL until the timeout is reached or it became reachable.
     If will call to `py.test.fail` in case the URL is not reachable.
     """
     timeout = time.time() + timeout
     last = 0
     wait = 0.1
 
     while timeout > last:
         last = time.time()
         if is_url_reachable(url):
             break
         elif (last + wait) > time.time():
             # Go to sleep because not enough time has passed since last check.
             time.sleep(wait)
     else:
         pytest.fail("Timeout while waiting for URL {}".format(url))
 
 
 def is_url_reachable(url):
     try:
         urllib.request.urlopen(url)
     except urllib.error.URLError:
         log.exception('URL `{}` reach error'.format(url))
         return False
     return True
 
 
 def repo_on_filesystem(repo_name):
     from rhodecode.lib import vcs
     from rhodecode.tests import TESTS_TMP_PATH
     repo = vcs.get_vcs_instance(
         os.path.join(TESTS_TMP_PATH, repo_name), create=False)
     return repo is not None
 
 
 def commit_change(
         repo, filename, content, message, vcs_type, parent=None, newfile=False):
     from rhodecode.tests import TEST_USER_ADMIN_LOGIN
 
     repo = Repository.get_by_repo_name(repo)
     _commit = parent
     if not parent:
         _commit = EmptyCommit(alias=vcs_type)
 
     if newfile:
         nodes = {
             filename: {
                 'content': content
             }
         }
         commit = ScmModel().create_nodes(
             user=TEST_USER_ADMIN_LOGIN, repo=repo,
             message=message,
             nodes=nodes,
             parent_commit=_commit,
             author='{} <[e-mail]>'.format(TEST_USER_ADMIN_LOGIN),
         )
     else:
         commit = ScmModel().commit_change(
             repo=repo.scm_instance(), repo_name=repo.repo_name,
             commit=parent, user=TEST_USER_ADMIN_LOGIN,
             author='{} <[e-mail]>'.format(TEST_USER_ADMIN_LOGIN),
             message=message,
             content=content,
             f_path=filename
         )
     return commit
 
 
 def permission_update_data_generator(csrf_token, default=None, grant=None, revoke=None):
     if not default:
         raise ValueError('Permission for default user must be given')
     form_data = [(
         'csrf_token', csrf_token
     )]
     # add default
     form_data.extend([
         ('u_perm_1', default)
     ])
 
     if grant:
         for cnt, (obj_id, perm, obj_name, obj_type) in enumerate(grant, 1):
             form_data.extend([
                 ('perm_new_member_perm_new{}'.format(cnt), perm),
                 ('perm_new_member_id_new{}'.format(cnt), obj_id),
                 ('perm_new_member_name_new{}'.format(cnt), obj_name),
                 ('perm_new_member_type_new{}'.format(cnt), obj_type),
 
             ])
     if revoke:
         for obj_id, obj_type in revoke:
             form_data.extend([
                 ('perm_del_member_id_{}'.format(obj_id), obj_id),
                 ('perm_del_member_type_{}'.format(obj_id), obj_type),
             ])
     return form_data