app: use simpler way to extract default_user_id, this will be now registered at server

boot.

rhodecode/apps/admin/tests/test_admin_permissions.py

@@ -1,299 +1,299 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 mock
 import pytest
 from rhodecode.model.db import User, UserIpMap
 from rhodecode.model.meta import Session
 from rhodecode.model.permission import PermissionModel
 from rhodecode.model.ssh_key import SshKeyModel
 from rhodecode.tests import (
     TestController, clear_cache_regions, assert_session_flash)
 
 
 def route_path(name, params=None, **kwargs):
     import urllib
     from rhodecode.apps._base import ADMIN_PREFIX
 
     base_url = {
         'edit_user_ips':
             ADMIN_PREFIX + '/users/{user_id}/edit/ips',
         'edit_user_ips_add':
             ADMIN_PREFIX + '/users/{user_id}/edit/ips/new',
         'edit_user_ips_delete':
             ADMIN_PREFIX + '/users/{user_id}/edit/ips/delete',
 
         'admin_permissions_application':
             ADMIN_PREFIX + '/permissions/application',
         'admin_permissions_application_update':
             ADMIN_PREFIX + '/permissions/application/update',
 
         'admin_permissions_global':
             ADMIN_PREFIX + '/permissions/global',
         'admin_permissions_global_update':
             ADMIN_PREFIX + '/permissions/global/update',
 
         'admin_permissions_object':
             ADMIN_PREFIX + '/permissions/object',
         'admin_permissions_object_update':
             ADMIN_PREFIX + '/permissions/object/update',
 
         'admin_permissions_ips':
             ADMIN_PREFIX + '/permissions/ips',
         'admin_permissions_overview':
             ADMIN_PREFIX + '/permissions/overview',
 
         'admin_permissions_ssh_keys':
             ADMIN_PREFIX + '/permissions/ssh_keys',
         'admin_permissions_ssh_keys_data':
             ADMIN_PREFIX + '/permissions/ssh_keys/data',
         'admin_permissions_ssh_keys_update':
             ADMIN_PREFIX + '/permissions/ssh_keys/update'
 
     }[name].format(**kwargs)
 
     if params:
         base_url = '{}?{}'.format(base_url, urllib.urlencode(params))
     return base_url
 
 
 class TestAdminPermissionsController(TestController):
 
     @pytest.fixture(scope='class', autouse=True)
     def prepare(self, request):
         # cleanup and reset to default permissions after
         @request.addfinalizer
         def cleanup():
             PermissionModel().create_default_user_permissions(
                 User.get_default_user(), force=True)
 
     def test_index_application(self):
         self.log_user()
         self.app.get(route_path('admin_permissions_application'))
 
     @pytest.mark.parametrize(
         'anonymous, default_register, default_register_message, default_password_reset,' 
         'default_extern_activate, expect_error, expect_form_error', [
             (True, 'hg.register.none', '', 'hg.password_reset.enabled', 'hg.extern_activate.manual',
              False, False),
             (True, 'hg.register.manual_activate', '', 'hg.password_reset.enabled', 'hg.extern_activate.auto',
              False, False),
             (True, 'hg.register.auto_activate', '', 'hg.password_reset.enabled', 'hg.extern_activate.manual',
              False, False),
             (True, 'hg.register.auto_activate', '', 'hg.password_reset.enabled', 'hg.extern_activate.manual',
              False, False),
             (True, 'hg.register.XXX', '', 'hg.password_reset.enabled', 'hg.extern_activate.manual',
              False, True),
             (True, '', '', 'hg.password_reset.enabled', '', True, False),
         ])
     def test_update_application_permissions(
             self, anonymous, default_register, default_register_message, default_password_reset,
             default_extern_activate, expect_error, expect_form_error):
 
         self.log_user()
 
         # TODO: anonymous access set here to False, breaks some other tests
         params = {
             'csrf_token': self.csrf_token,
             'anonymous': anonymous,
             'default_register': default_register,
             'default_register_message': default_register_message,
             'default_password_reset': default_password_reset,
             'default_extern_activate': default_extern_activate,
         }
         response = self.app.post(route_path('admin_permissions_application_update'),
                                  params=params)
         if expect_form_error:
             assert response.status_int == 200
             response.mustcontain('Value must be one of')
         else:
             if expect_error:
                 msg = 'Error occurred during update of permissions'
             else:
                 msg = 'Application permissions updated successfully'
             assert_session_flash(response, msg)
 
     def test_index_object(self):
         self.log_user()
         self.app.get(route_path('admin_permissions_object'))
 
     @pytest.mark.parametrize(
         'repo, repo_group, user_group, expect_error, expect_form_error', [
             ('repository.none', 'group.none', 'usergroup.none', False, False),
             ('repository.read', 'group.read', 'usergroup.read', False, False),
             ('repository.write', 'group.write', 'usergroup.write',
              False, False),
             ('repository.admin', 'group.admin', 'usergroup.admin',
              False, False),
             ('repository.XXX', 'group.admin', 'usergroup.admin', False, True),
             ('', '', '', True, False),
         ])
     def test_update_object_permissions(self, repo, repo_group, user_group,
                                        expect_error, expect_form_error):
         self.log_user()
 
         params = {
             'csrf_token': self.csrf_token,
             'default_repo_perm': repo,
             'overwrite_default_repo': False,
             'default_group_perm': repo_group,
             'overwrite_default_group': False,
             'default_user_group_perm': user_group,
             'overwrite_default_user_group': False,
         }
         response = self.app.post(route_path('admin_permissions_object_update'),
                                  params=params)
         if expect_form_error:
             assert response.status_int == 200
             response.mustcontain('Value must be one of')
         else:
             if expect_error:
                 msg = 'Error occurred during update of permissions'
             else:
                 msg = 'Object permissions updated successfully'
             assert_session_flash(response, msg)
 
     def test_index_global(self):
         self.log_user()
         self.app.get(route_path('admin_permissions_global'))
 
     @pytest.mark.parametrize(
         'repo_create, repo_create_write, user_group_create, repo_group_create,'
         'fork_create, inherit_default_permissions, expect_error,'
         'expect_form_error', [
             ('hg.create.none', 'hg.create.write_on_repogroup.false',
              'hg.usergroup.create.false', 'hg.repogroup.create.false',
              'hg.fork.none', 'hg.inherit_default_perms.false', False, False),
             ('hg.create.repository', 'hg.create.write_on_repogroup.true',
              'hg.usergroup.create.true', 'hg.repogroup.create.true',
              'hg.fork.repository', 'hg.inherit_default_perms.false',
              False, False),
             ('hg.create.XXX', 'hg.create.write_on_repogroup.true',
              'hg.usergroup.create.true', 'hg.repogroup.create.true',
              'hg.fork.repository', 'hg.inherit_default_perms.false',
              False, True),
             ('', '', '', '', '', '', True, False),
         ])
     def test_update_global_permissions(
             self, repo_create, repo_create_write, user_group_create,
             repo_group_create, fork_create, inherit_default_permissions,
             expect_error, expect_form_error):
         self.log_user()
 
         params = {
             'csrf_token': self.csrf_token,
             'default_repo_create': repo_create,
             'default_repo_create_on_write': repo_create_write,
             'default_user_group_create': user_group_create,
             'default_repo_group_create': repo_group_create,
             'default_fork_create': fork_create,
             'default_inherit_default_permissions': inherit_default_permissions
         }
         response = self.app.post(route_path('admin_permissions_global_update'),
                                  params=params)
         if expect_form_error:
             assert response.status_int == 200
             response.mustcontain('Value must be one of')
         else:
             if expect_error:
                 msg = 'Error occurred during update of permissions'
             else:
                 msg = 'Global permissions updated successfully'
             assert_session_flash(response, msg)
 
     def test_index_ips(self):
         self.log_user()
         response = self.app.get(route_path('admin_permissions_ips'))
         response.mustcontain('All IP addresses are allowed')
 
     def test_add_delete_ips(self):
         clear_cache_regions(['sql_cache_short'])
         self.log_user()
 
         # ADD
-        default_user_id = User.get_default_user().user_id
+        default_user_id = User.get_default_user_id()
         self.app.post(
             route_path('edit_user_ips_add', user_id=default_user_id),
             params={'new_ip': '0.0.0.0/24', 'csrf_token': self.csrf_token})
 
         response = self.app.get(route_path('admin_permissions_ips'))
         response.mustcontain('0.0.0.0/24')
         response.mustcontain('0.0.0.0 - 0.0.0.255')
 
         # DELETE
-        default_user_id = User.get_default_user().user_id
+        default_user_id = User.get_default_user_id()
         del_ip_id = UserIpMap.query().filter(UserIpMap.user_id ==
                                              default_user_id).first().ip_id
 
         response = self.app.post(
             route_path('edit_user_ips_delete', user_id=default_user_id),
             params={'del_ip_id': del_ip_id, 'csrf_token': self.csrf_token})
 
         assert_session_flash(response, 'Removed ip address from user whitelist')
 
         clear_cache_regions(['sql_cache_short'])
         response = self.app.get(route_path('admin_permissions_ips'))
         response.mustcontain('All IP addresses are allowed')
         response.mustcontain(no=['0.0.0.0/24'])
         response.mustcontain(no=['0.0.0.0 - 0.0.0.255'])
 
     def test_index_overview(self):
         self.log_user()
         self.app.get(route_path('admin_permissions_overview'))
 
     def test_ssh_keys(self):
         self.log_user()
         self.app.get(route_path('admin_permissions_ssh_keys'), status=200)
 
     def test_ssh_keys_data(self, user_util, xhr_header):
         self.log_user()
         response = self.app.get(route_path('admin_permissions_ssh_keys_data'),
                                 extra_environ=xhr_header)
         assert response.json == {u'data': [], u'draw': None,
                                  u'recordsFiltered': 0, u'recordsTotal': 0}
 
         dummy_user = user_util.create_user()
         SshKeyModel().create(dummy_user, 'ab:cd:ef', 'KEYKEY', 'test_key')
         Session().commit()
         response = self.app.get(route_path('admin_permissions_ssh_keys_data'),
                                 extra_environ=xhr_header)
         assert response.json['data'][0]['fingerprint'] == 'ab:cd:ef'
 
     def test_ssh_keys_update(self):
         self.log_user()
         response = self.app.post(
             route_path('admin_permissions_ssh_keys_update'),
             dict(csrf_token=self.csrf_token), status=302)
 
         assert_session_flash(
             response, 'Updated SSH keys file')
 
     def test_ssh_keys_update_disabled(self):
         self.log_user()
 
         from rhodecode.apps.admin.views.permissions import AdminPermissionsView
         with mock.patch.object(AdminPermissionsView, 'ssh_enabled',
                                return_value=False):
             response = self.app.post(
                 route_path('admin_permissions_ssh_keys_update'),
                 dict(csrf_token=self.csrf_token), status=302)
 
             assert_session_flash(
                 response, 'SSH key support is disabled in .ini file')
  No newline at end of file

rhodecode/apps/admin/views/permissions.py

@@ -1,519 +1,519 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 re
 import logging
 import formencode
 import formencode.htmlfill
 import datetime
 from pyramid.interfaces import IRoutesMapper
 
 from pyramid.view import view_config
 from pyramid.httpexceptions import HTTPFound
 from pyramid.renderers import render
 from pyramid.response import Response
 
 from rhodecode.apps._base import BaseAppView, DataGridAppView
 from rhodecode.apps.ssh_support import SshKeyFileChangeEvent
 from rhodecode import events
 
 from rhodecode.lib import helpers as h
 from rhodecode.lib.auth import (
     LoginRequired, HasPermissionAllDecorator, CSRFRequired)
 from rhodecode.lib.utils2 import aslist, safe_unicode
 from rhodecode.model.db import (
     or_, coalesce, User, UserIpMap, UserSshKeys)
 from rhodecode.model.forms import (
     ApplicationPermissionsForm, ObjectPermissionsForm, UserPermissionsForm)
 from rhodecode.model.meta import Session
 from rhodecode.model.permission import PermissionModel
 from rhodecode.model.settings import SettingsModel
 
 
 log = logging.getLogger(__name__)
 
 
 class AdminPermissionsView(BaseAppView, DataGridAppView):
     def load_default_context(self):
         c = self._get_local_tmpl_context()
         PermissionModel().set_global_permission_choices(
             c, gettext_translator=self.request.translate)
         return c
 
     @LoginRequired()
     @HasPermissionAllDecorator('hg.admin')
     @view_config(
         route_name='admin_permissions_application', request_method='GET',
         renderer='rhodecode:templates/admin/permissions/permissions.mako')
     def permissions_application(self):
         c = self.load_default_context()
         c.active = 'application'
 
         c.user = User.get_default_user(refresh=True)
 
         app_settings = c.rc_config
 
         defaults = {
             'anonymous': c.user.active,
             'default_register_message': app_settings.get(
                 'rhodecode_register_message')
         }
         defaults.update(c.user.get_default_perms())
 
         data = render('rhodecode:templates/admin/permissions/permissions.mako',
                       self._get_template_context(c), self.request)
         html = formencode.htmlfill.render(
             data,
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
         )
         return Response(html)
 
     @LoginRequired()
     @HasPermissionAllDecorator('hg.admin')
     @CSRFRequired()
     @view_config(
         route_name='admin_permissions_application_update', request_method='POST',
         renderer='rhodecode:templates/admin/permissions/permissions.mako')
     def permissions_application_update(self):
         _ = self.request.translate
         c = self.load_default_context()
         c.active = 'application'
 
         _form = ApplicationPermissionsForm(
             self.request.translate,
             [x[0] for x in c.register_choices],
             [x[0] for x in c.password_reset_choices],
             [x[0] for x in c.extern_activate_choices])()
 
         try:
             form_result = _form.to_python(dict(self.request.POST))
             form_result.update({'perm_user_name': User.DEFAULT_USER})
             PermissionModel().update_application_permissions(form_result)
 
             settings = [
                 ('register_message', 'default_register_message'),
             ]
             for setting, form_key in settings:
                 sett = SettingsModel().create_or_update_setting(
                     setting, form_result[form_key])
                 Session().add(sett)
 
             Session().commit()
             h.flash(_('Application permissions updated successfully'),
                     category='success')
 
         except formencode.Invalid as errors:
             defaults = errors.value
 
             data = render(
                 'rhodecode:templates/admin/permissions/permissions.mako',
                 self._get_template_context(c), self.request)
             html = formencode.htmlfill.render(
                 data,
                 defaults=defaults,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False
             )
             return Response(html)
 
         except Exception:
             log.exception("Exception during update of permissions")
             h.flash(_('Error occurred during update of permissions'),
                     category='error')
 
-        affected_user_ids = [User.get_default_user().user_id]
+        affected_user_ids = [User.get_default_user_id()]
         PermissionModel().trigger_permission_flush(affected_user_ids)
 
         raise HTTPFound(h.route_path('admin_permissions_application'))
 
     @LoginRequired()
     @HasPermissionAllDecorator('hg.admin')
     @view_config(
         route_name='admin_permissions_object', request_method='GET',
         renderer='rhodecode:templates/admin/permissions/permissions.mako')
     def permissions_objects(self):
         c = self.load_default_context()
         c.active = 'objects'
 
         c.user = User.get_default_user(refresh=True)
         defaults = {}
         defaults.update(c.user.get_default_perms())
 
         data = render(
             'rhodecode:templates/admin/permissions/permissions.mako',
             self._get_template_context(c), self.request)
         html = formencode.htmlfill.render(
             data,
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
         )
         return Response(html)
 
     @LoginRequired()
     @HasPermissionAllDecorator('hg.admin')
     @CSRFRequired()
     @view_config(
         route_name='admin_permissions_object_update', request_method='POST',
         renderer='rhodecode:templates/admin/permissions/permissions.mako')
     def permissions_objects_update(self):
         _ = self.request.translate
         c = self.load_default_context()
         c.active = 'objects'
 
         _form = ObjectPermissionsForm(
             self.request.translate,
             [x[0] for x in c.repo_perms_choices],
             [x[0] for x in c.group_perms_choices],
             [x[0] for x in c.user_group_perms_choices],
         )()
 
         try:
             form_result = _form.to_python(dict(self.request.POST))
             form_result.update({'perm_user_name': User.DEFAULT_USER})
             PermissionModel().update_object_permissions(form_result)
 
             Session().commit()
             h.flash(_('Object permissions updated successfully'),
                     category='success')
 
         except formencode.Invalid as errors:
             defaults = errors.value
 
             data = render(
                 'rhodecode:templates/admin/permissions/permissions.mako',
                 self._get_template_context(c), self.request)
             html = formencode.htmlfill.render(
                 data,
                 defaults=defaults,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False
             )
             return Response(html)
         except Exception:
             log.exception("Exception during update of permissions")
             h.flash(_('Error occurred during update of permissions'),
                     category='error')
 
-        affected_user_ids = [User.get_default_user().user_id]
+        affected_user_ids = [User.get_default_user_id()]
         PermissionModel().trigger_permission_flush(affected_user_ids)
 
         raise HTTPFound(h.route_path('admin_permissions_object'))
 
     @LoginRequired()
     @HasPermissionAllDecorator('hg.admin')
     @view_config(
         route_name='admin_permissions_branch', request_method='GET',
         renderer='rhodecode:templates/admin/permissions/permissions.mako')
     def permissions_branch(self):
         c = self.load_default_context()
         c.active = 'branch'
 
         c.user = User.get_default_user(refresh=True)
         defaults = {}
         defaults.update(c.user.get_default_perms())
 
         data = render(
             'rhodecode:templates/admin/permissions/permissions.mako',
             self._get_template_context(c), self.request)
         html = formencode.htmlfill.render(
             data,
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
         )
         return Response(html)
 
     @LoginRequired()
     @HasPermissionAllDecorator('hg.admin')
     @view_config(
         route_name='admin_permissions_global', request_method='GET',
         renderer='rhodecode:templates/admin/permissions/permissions.mako')
     def permissions_global(self):
         c = self.load_default_context()
         c.active = 'global'
 
         c.user = User.get_default_user(refresh=True)
         defaults = {}
         defaults.update(c.user.get_default_perms())
 
         data = render(
             'rhodecode:templates/admin/permissions/permissions.mako',
             self._get_template_context(c), self.request)
         html = formencode.htmlfill.render(
             data,
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
         )
         return Response(html)
 
     @LoginRequired()
     @HasPermissionAllDecorator('hg.admin')
     @CSRFRequired()
     @view_config(
         route_name='admin_permissions_global_update', request_method='POST',
         renderer='rhodecode:templates/admin/permissions/permissions.mako')
     def permissions_global_update(self):
         _ = self.request.translate
         c = self.load_default_context()
         c.active = 'global'
 
         _form = UserPermissionsForm(
             self.request.translate,
             [x[0] for x in c.repo_create_choices],
             [x[0] for x in c.repo_create_on_write_choices],
             [x[0] for x in c.repo_group_create_choices],
             [x[0] for x in c.user_group_create_choices],
             [x[0] for x in c.fork_choices],
             [x[0] for x in c.inherit_default_permission_choices])()
 
         try:
             form_result = _form.to_python(dict(self.request.POST))
             form_result.update({'perm_user_name': User.DEFAULT_USER})
             PermissionModel().update_user_permissions(form_result)
 
             Session().commit()
             h.flash(_('Global permissions updated successfully'),
                     category='success')
 
         except formencode.Invalid as errors:
             defaults = errors.value
 
             data = render(
                 'rhodecode:templates/admin/permissions/permissions.mako',
                 self._get_template_context(c), self.request)
             html = formencode.htmlfill.render(
                 data,
                 defaults=defaults,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False
             )
             return Response(html)
         except Exception:
             log.exception("Exception during update of permissions")
             h.flash(_('Error occurred during update of permissions'),
                     category='error')
 
-        affected_user_ids = [User.get_default_user().user_id]
+        affected_user_ids = [User.get_default_user_id()]
         PermissionModel().trigger_permission_flush(affected_user_ids)
 
         raise HTTPFound(h.route_path('admin_permissions_global'))
 
     @LoginRequired()
     @HasPermissionAllDecorator('hg.admin')
     @view_config(
         route_name='admin_permissions_ips', request_method='GET',
         renderer='rhodecode:templates/admin/permissions/permissions.mako')
     def permissions_ips(self):
         c = self.load_default_context()
         c.active = 'ips'
 
         c.user = User.get_default_user(refresh=True)
         c.user_ip_map = (
             UserIpMap.query().filter(UserIpMap.user == c.user).all())
 
         return self._get_template_context(c)
 
     @LoginRequired()
     @HasPermissionAllDecorator('hg.admin')
     @view_config(
         route_name='admin_permissions_overview', request_method='GET',
         renderer='rhodecode:templates/admin/permissions/permissions.mako')
     def permissions_overview(self):
         c = self.load_default_context()
         c.active = 'perms'
 
         c.user = User.get_default_user(refresh=True)
         c.perm_user = c.user.AuthUser()
         return self._get_template_context(c)
 
     @LoginRequired()
     @HasPermissionAllDecorator('hg.admin')
     @view_config(
         route_name='admin_permissions_auth_token_access', request_method='GET',
         renderer='rhodecode:templates/admin/permissions/permissions.mako')
     def auth_token_access(self):
         from rhodecode import CONFIG
 
         c = self.load_default_context()
         c.active = 'auth_token_access'
 
         c.user = User.get_default_user(refresh=True)
         c.perm_user = c.user.AuthUser()
 
         mapper = self.request.registry.queryUtility(IRoutesMapper)
         c.view_data = []
 
         _argument_prog = re.compile('\{(.*?)\}|:\((.*)\)')
         introspector = self.request.registry.introspector
 
         view_intr = {}
         for view_data in introspector.get_category('views'):
             intr = view_data['introspectable']
 
             if 'route_name' in intr and intr['attr']:
                 view_intr[intr['route_name']] = '{}:{}'.format(
                     str(intr['derived_callable'].func_name), intr['attr']
                 )
 
         c.whitelist_key = 'api_access_controllers_whitelist'
         c.whitelist_file = CONFIG.get('__file__')
         whitelist_views = aslist(
             CONFIG.get(c.whitelist_key), sep=',')
 
         for route_info in mapper.get_routes():
             if not route_info.name.startswith('__'):
                 routepath = route_info.pattern
 
                 def replace(matchobj):
                     if matchobj.group(1):
                         return "{%s}" % matchobj.group(1).split(':')[0]
                     else:
                         return "{%s}" % matchobj.group(2)
 
                 routepath = _argument_prog.sub(replace, routepath)
 
                 if not routepath.startswith('/'):
                     routepath = '/' + routepath
 
                 view_fqn = view_intr.get(route_info.name, 'NOT AVAILABLE')
                 active = view_fqn in whitelist_views
                 c.view_data.append((route_info.name, view_fqn, routepath, active))
 
         c.whitelist_views = whitelist_views
         return self._get_template_context(c)
 
     def ssh_enabled(self):
         return self.request.registry.settings.get(
             'ssh.generate_authorized_keyfile')
 
     @LoginRequired()
     @HasPermissionAllDecorator('hg.admin')
     @view_config(
         route_name='admin_permissions_ssh_keys', request_method='GET',
         renderer='rhodecode:templates/admin/permissions/permissions.mako')
     def ssh_keys(self):
         c = self.load_default_context()
         c.active = 'ssh_keys'
         c.ssh_enabled = self.ssh_enabled()
         return self._get_template_context(c)
 
     @LoginRequired()
     @HasPermissionAllDecorator('hg.admin')
     @view_config(
         route_name='admin_permissions_ssh_keys_data', request_method='GET',
         renderer='json_ext', xhr=True)
     def ssh_keys_data(self):
         _ = self.request.translate
         self.load_default_context()
         column_map = {
             'fingerprint': 'ssh_key_fingerprint',
             'username': User.username
         }
         draw, start, limit = self._extract_chunk(self.request)
         search_q, order_by, order_dir = self._extract_ordering(
             self.request, column_map=column_map)
 
         ssh_keys_data_total_count = UserSshKeys.query()\
             .count()
 
         # json generate
         base_q = UserSshKeys.query().join(UserSshKeys.user)
 
         if search_q:
             like_expression = u'%{}%'.format(safe_unicode(search_q))
             base_q = base_q.filter(or_(
                 User.username.ilike(like_expression),
                 UserSshKeys.ssh_key_fingerprint.ilike(like_expression),
             ))
 
         users_data_total_filtered_count = base_q.count()
 
         sort_col = self._get_order_col(order_by, UserSshKeys)
         if sort_col:
             if order_dir == 'asc':
                 # handle null values properly to order by NULL last
                 if order_by in ['created_on']:
                     sort_col = coalesce(sort_col, datetime.date.max)
                 sort_col = sort_col.asc()
             else:
                 # handle null values properly to order by NULL last
                 if order_by in ['created_on']:
                     sort_col = coalesce(sort_col, datetime.date.min)
                 sort_col = sort_col.desc()
 
         base_q = base_q.order_by(sort_col)
         base_q = base_q.offset(start).limit(limit)
 
         ssh_keys = base_q.all()
 
         ssh_keys_data = []
         for ssh_key in ssh_keys:
             ssh_keys_data.append({
                 "username": h.gravatar_with_user(self.request, ssh_key.user.username),
                 "fingerprint": ssh_key.ssh_key_fingerprint,
                 "description": ssh_key.description,
                 "created_on": h.format_date(ssh_key.created_on),
                 "accessed_on": h.format_date(ssh_key.accessed_on),
                 "action": h.link_to(
                     _('Edit'), h.route_path('edit_user_ssh_keys',
                                             user_id=ssh_key.user.user_id))
             })
 
         data = ({
             'draw': draw,
             'data': ssh_keys_data,
             'recordsTotal': ssh_keys_data_total_count,
             'recordsFiltered': users_data_total_filtered_count,
         })
 
         return data
 
     @LoginRequired()
     @HasPermissionAllDecorator('hg.admin')
     @CSRFRequired()
     @view_config(
         route_name='admin_permissions_ssh_keys_update', request_method='POST',
         renderer='rhodecode:templates/admin/permissions/permissions.mako')
     def ssh_keys_update(self):
         _ = self.request.translate
         self.load_default_context()
 
         ssh_enabled = self.ssh_enabled()
         key_file = self.request.registry.settings.get(
             'ssh.authorized_keys_file_path')
         if ssh_enabled:
             events.trigger(SshKeyFileChangeEvent(), self.request.registry)
             h.flash(_('Updated SSH keys file: {}').format(key_file),
                     category='success')
         else:
             h.flash(_('SSH key support is disabled in .ini file'),
                     category='warning')
 
         raise HTTPFound(h.route_path('admin_permissions_ssh_keys'))

rhodecode/apps/repository/views/repo_settings_advanced.py

@@ -1,325 +1,325 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/
 
 import logging
 
 from pyramid.view import view_config
 from pyramid.httpexceptions import HTTPFound
 from packaging.version import Version
 
 from rhodecode import events
 from rhodecode.apps._base import RepoAppView
 from rhodecode.lib import helpers as h
 from rhodecode.lib import audit_logger
 from rhodecode.lib.auth import (
     LoginRequired, HasRepoPermissionAnyDecorator, CSRFRequired,
     HasRepoPermissionAny)
 from rhodecode.lib.exceptions import AttachedForksError, AttachedPullRequestsError
 from rhodecode.lib.utils2 import safe_int
 from rhodecode.lib.vcs import RepositoryError
 from rhodecode.model.db import Session, UserFollowing, User, Repository
 from rhodecode.model.permission import PermissionModel
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.scm import ScmModel
 
 log = logging.getLogger(__name__)
 
 
 class RepoSettingsView(RepoAppView):
 
     def load_default_context(self):
         c = self._get_local_tmpl_context()
         return c
 
     def _get_users_with_permissions(self):
         user_permissions = {}
         for perm in self.db_repo.permissions():
             user_permissions[perm.user_id] = perm
 
         return user_permissions
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.admin')
     @view_config(
         route_name='edit_repo_advanced', request_method='GET',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
     def edit_advanced(self):
         _ = self.request.translate
         c = self.load_default_context()
         c.active = 'advanced'
 
-        c.default_user_id = User.get_default_user().user_id
+        c.default_user_id = User.get_default_user_id()
         c.in_public_journal = UserFollowing.query() \
             .filter(UserFollowing.user_id == c.default_user_id) \
             .filter(UserFollowing.follows_repository == self.db_repo).scalar()
 
         c.ver_info_dict = self.rhodecode_vcs_repo.get_hooks_info()
         c.hooks_outdated = False
 
         try:
             if Version(c.ver_info_dict['pre_version']) < Version(c.rhodecode_version):
                 c.hooks_outdated = True
         except Exception:
             pass
 
         # update commit cache if GET flag is present
         if self.request.GET.get('update_commit_cache'):
             self.db_repo.update_commit_cache()
             h.flash(_('updated commit cache'), category='success')
 
         return self._get_template_context(c)
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.admin')
     @CSRFRequired()
     @view_config(
         route_name='edit_repo_advanced_archive', request_method='POST',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
     def edit_advanced_archive(self):
         """
         Archives the repository. It will become read-only, and not visible in search
         or other queries. But still visible for super-admins.
         """
 
         _ = self.request.translate
 
         try:
             old_data = self.db_repo.get_api_data()
             RepoModel().archive(self.db_repo)
 
             repo = audit_logger.RepoWrap(repo_id=None, repo_name=self.db_repo.repo_name)
             audit_logger.store_web(
                 'repo.archive', action_data={'old_data': old_data},
                 user=self._rhodecode_user, repo=repo)
 
             ScmModel().mark_for_invalidation(self.db_repo_name, delete=True)
             h.flash(
                 _('Archived repository `%s`') % self.db_repo_name,
                 category='success')
             Session().commit()
         except Exception:
             log.exception("Exception during archiving of repository")
             h.flash(_('An error occurred during archiving of `%s`')
                     % self.db_repo_name, category='error')
             # redirect to advanced for more deletion options
             raise HTTPFound(
                 h.route_path('edit_repo_advanced', repo_name=self.db_repo_name,
                              _anchor='advanced-archive'))
 
         # flush permissions for all users defined in permissions
         affected_user_ids = self._get_users_with_permissions().keys()
         PermissionModel().trigger_permission_flush(affected_user_ids)
 
         raise HTTPFound(h.route_path('home'))
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.admin')
     @CSRFRequired()
     @view_config(
         route_name='edit_repo_advanced_delete', request_method='POST',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
     def edit_advanced_delete(self):
         """
         Deletes the repository, or shows warnings if deletion is not possible
         because of attached forks or other errors.
         """
         _ = self.request.translate
         handle_forks = self.request.POST.get('forks', None)
         if handle_forks == 'detach_forks':
             handle_forks = 'detach'
         elif handle_forks == 'delete_forks':
             handle_forks = 'delete'
 
         try:
             old_data = self.db_repo.get_api_data()
             RepoModel().delete(self.db_repo, forks=handle_forks)
 
             _forks = self.db_repo.forks.count()
             if _forks and handle_forks:
                 if handle_forks == 'detach_forks':
                     h.flash(_('Detached %s forks') % _forks, category='success')
                 elif handle_forks == 'delete_forks':
                     h.flash(_('Deleted %s forks') % _forks, category='success')
 
             repo = audit_logger.RepoWrap(repo_id=None, repo_name=self.db_repo.repo_name)
             audit_logger.store_web(
                 'repo.delete', action_data={'old_data': old_data},
                 user=self._rhodecode_user, repo=repo)
 
             ScmModel().mark_for_invalidation(self.db_repo_name, delete=True)
             h.flash(
                 _('Deleted repository `%s`') % self.db_repo_name,
                 category='success')
             Session().commit()
         except AttachedForksError:
             repo_advanced_url = h.route_path(
                 'edit_repo_advanced', repo_name=self.db_repo_name,
                 _anchor='advanced-delete')
             delete_anchor = h.link_to(_('detach or delete'), repo_advanced_url)
             h.flash(_('Cannot delete `{repo}` it still contains attached forks. '
                       'Try using {delete_or_detach} option.')
                     .format(repo=self.db_repo_name, delete_or_detach=delete_anchor),
                     category='warning')
 
             # redirect to advanced for forks handle action ?
             raise HTTPFound(repo_advanced_url)
 
         except AttachedPullRequestsError:
             repo_advanced_url = h.route_path(
                 'edit_repo_advanced', repo_name=self.db_repo_name,
                 _anchor='advanced-delete')
             attached_prs = len(self.db_repo.pull_requests_source +
                                self.db_repo.pull_requests_target)
             h.flash(
                 _('Cannot delete `{repo}` it still contains {num} attached pull requests. '
                   'Consider archiving the repository instead.').format(
                     repo=self.db_repo_name, num=attached_prs), category='warning')
 
             # redirect to advanced for forks handle action ?
             raise HTTPFound(repo_advanced_url)
 
         except Exception:
             log.exception("Exception during deletion of repository")
             h.flash(_('An error occurred during deletion of `%s`')
                     % self.db_repo_name, category='error')
             # redirect to advanced for more deletion options
             raise HTTPFound(
                 h.route_path('edit_repo_advanced', repo_name=self.db_repo_name,
                              _anchor='advanced-delete'))
 
         raise HTTPFound(h.route_path('home'))
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.admin')
     @CSRFRequired()
     @view_config(
         route_name='edit_repo_advanced_journal', request_method='POST',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
     def edit_advanced_journal(self):
         """
         Set's this repository to be visible in public journal,
         in other words making default user to follow this repo
         """
         _ = self.request.translate
 
         try:
-            user_id = User.get_default_user().user_id
+            user_id = User.get_default_user_id()
             ScmModel().toggle_following_repo(self.db_repo.repo_id, user_id)
             h.flash(_('Updated repository visibility in public journal'),
                     category='success')
             Session().commit()
         except Exception:
             h.flash(_('An error occurred during setting this '
                       'repository in public journal'),
                     category='error')
 
         raise HTTPFound(
             h.route_path('edit_repo_advanced', repo_name=self.db_repo_name))
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.admin')
     @CSRFRequired()
     @view_config(
         route_name='edit_repo_advanced_fork', request_method='POST',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
     def edit_advanced_fork(self):
         """
         Mark given repository as a fork of another
         """
         _ = self.request.translate
 
         new_fork_id = safe_int(self.request.POST.get('id_fork_of'))
 
         # valid repo, re-check permissions
         if new_fork_id:
             repo = Repository.get(new_fork_id)
             # ensure we have at least read access to the repo we mark
             perm_check = HasRepoPermissionAny(
                 'repository.read', 'repository.write', 'repository.admin')
 
             if repo and perm_check(repo_name=repo.repo_name):
                 new_fork_id = repo.repo_id
             else:
                 new_fork_id = None
 
         try:
             repo = ScmModel().mark_as_fork(
                 self.db_repo_name, new_fork_id, self._rhodecode_user.user_id)
             fork = repo.fork.repo_name if repo.fork else _('Nothing')
             Session().commit()
             h.flash(
                 _('Marked repo %s as fork of %s') % (self.db_repo_name, fork),
                 category='success')
         except RepositoryError as e:
             log.exception("Repository Error occurred")
             h.flash(str(e), category='error')
         except Exception:
             log.exception("Exception while editing fork")
             h.flash(_('An error occurred during this operation'),
                     category='error')
 
         raise HTTPFound(
             h.route_path('edit_repo_advanced', repo_name=self.db_repo_name))
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.admin')
     @CSRFRequired()
     @view_config(
         route_name='edit_repo_advanced_locking', request_method='POST',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
     def edit_advanced_locking(self):
         """
         Toggle locking of repository
         """
         _ = self.request.translate
         set_lock = self.request.POST.get('set_lock')
         set_unlock = self.request.POST.get('set_unlock')
 
         try:
             if set_lock:
                 Repository.lock(self.db_repo, self._rhodecode_user.user_id,
                                 lock_reason=Repository.LOCK_WEB)
                 h.flash(_('Locked repository'), category='success')
             elif set_unlock:
                 Repository.unlock(self.db_repo)
                 h.flash(_('Unlocked repository'), category='success')
         except Exception as e:
             log.exception("Exception during unlocking")
             h.flash(_('An error occurred during unlocking'), category='error')
 
         raise HTTPFound(
             h.route_path('edit_repo_advanced', repo_name=self.db_repo_name))
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.admin')
     @view_config(
         route_name='edit_repo_advanced_hooks', request_method='GET',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
     def edit_advanced_install_hooks(self):
         """
         Install Hooks for repository
         """
         _ = self.request.translate
         self.load_default_context()
         self.rhodecode_vcs_repo.install_hooks(force=True)
         h.flash(_('installed updated hooks into this repository'),
                 category='success')
         raise HTTPFound(
             h.route_path('edit_repo_advanced', repo_name=self.db_repo_name))

rhodecode/config/environment.py

@@ -1,86 +1,87 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 os
 import logging
 import rhodecode
 
 from rhodecode.config import utils
 
 from rhodecode.lib.utils import load_rcextensions
 from rhodecode.lib.utils2 import str2bool
 from rhodecode.lib.vcs import connect_vcs
 
 log = logging.getLogger(__name__)
 
 
 def load_pyramid_environment(global_config, settings):
     # Some parts of the code expect a merge of global and app settings.
     settings_merged = global_config.copy()
     settings_merged.update(settings)
 
     # TODO(marcink): probably not required anymore
     # configure channelstream,
     settings_merged['channelstream_config'] = {
         'enabled': str2bool(settings_merged.get('channelstream.enabled', False)),
         'server': settings_merged.get('channelstream.server'),
         'secret': settings_merged.get('channelstream.secret')
     }
 
     # If this is a test run we prepare the test environment like
     # creating a test database, test search index and test repositories.
     # This has to be done before the database connection is initialized.
     if settings['is_test']:
         rhodecode.is_test = True
         rhodecode.disable_error_handler = True
         from rhodecode import authentication
         authentication.plugin_default_auth_ttl = 0
 
         utils.initialize_test_environment(settings_merged)
 
     # Initialize the database connection.
     utils.initialize_database(settings_merged)
 
     load_rcextensions(root_path=settings_merged['here'])
 
     # Limit backends to `vcs.backends` from configuration, and preserve the order
     for alias in rhodecode.BACKENDS.keys():
         if alias not in settings['vcs.backends']:
             del rhodecode.BACKENDS[alias]
 
-    def sorter(item):
-        return settings['vcs.backends'].index(item[0])
-    rhodecode.BACKENDS = rhodecode.OrderedDict(sorted(rhodecode.BACKENDS.items(), key=sorter))
+    _sorted_backend = sorted(rhodecode.BACKENDS.items(),
+                             key=lambda item: settings['vcs.backends'].index(item[0]))
+    rhodecode.BACKENDS = rhodecode.OrderedDict(_sorted_backend)
 
     log.info('Enabled VCS backends: %s', rhodecode.BACKENDS.keys())
 
     # initialize vcs client and optionally run the server if enabled
     vcs_server_uri = settings['vcs.server']
     vcs_server_enabled = settings['vcs.server.enable']
 
     utils.configure_vcs(settings)
 
     # Store the settings to make them available to other modules.
 
     rhodecode.PYRAMID_SETTINGS = settings_merged
     rhodecode.CONFIG = settings_merged
+    rhodecode.CONFIG['default_user_id'] = utils.get_default_user_id()
 
     if vcs_server_enabled:
         connect_vcs(vcs_server_uri, utils.get_vcs_server_protocol(settings))

rhodecode/config/utils.py

@@ -1,93 +1,101 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 os
-import shlex
 import platform
 
 from rhodecode.model import init_model
 
 
 def configure_vcs(config):
     """
     Patch VCS config with some RhodeCode specific stuff
     """
     from rhodecode.lib.vcs import conf
     import rhodecode.lib.vcs.conf.settings
 
     conf.settings.BACKENDS = {
         'hg': 'rhodecode.lib.vcs.backends.hg.MercurialRepository',
         'git': 'rhodecode.lib.vcs.backends.git.GitRepository',
         'svn': 'rhodecode.lib.vcs.backends.svn.SubversionRepository',
     }
 
     conf.settings.HOOKS_PROTOCOL = config['vcs.hooks.protocol']
     conf.settings.HOOKS_HOST = config['vcs.hooks.host']
     conf.settings.HOOKS_DIRECT_CALLS = config['vcs.hooks.direct_calls']
     conf.settings.DEFAULT_ENCODINGS = config['default_encoding']
     conf.settings.ALIASES[:] = config['vcs.backends']
     conf.settings.SVN_COMPATIBLE_VERSION = config['vcs.svn.compatible_version']
 
 
 def initialize_database(config):
     from rhodecode.lib.utils2 import engine_from_config, get_encryption_key
     engine = engine_from_config(config, 'sqlalchemy.db1.')
     init_model(engine, encryption_key=get_encryption_key(config))
 
 
 def initialize_test_environment(settings, test_env=None):
     if test_env is None:
         test_env = not int(os.environ.get('RC_NO_TMP_PATH', 0))
 
     from rhodecode.lib.utils import (
         create_test_directory, create_test_database, create_test_repositories,
         create_test_index)
     from rhodecode.tests import TESTS_TMP_PATH
     from rhodecode.lib.vcs.backends.hg import largefiles_store
     from rhodecode.lib.vcs.backends.git import lfs_store
 
     # test repos
     if test_env:
         create_test_directory(TESTS_TMP_PATH)
         # large object stores
         create_test_directory(largefiles_store(TESTS_TMP_PATH))
         create_test_directory(lfs_store(TESTS_TMP_PATH))
 
         create_test_database(TESTS_TMP_PATH, settings)
         create_test_repositories(TESTS_TMP_PATH, settings)
         create_test_index(TESTS_TMP_PATH, settings)
 
 
 def get_vcs_server_protocol(config):
     return config['vcs.server.protocol']
 
 
 def set_instance_id(config):
     """
     Sets a dynamic generated config['instance_id'] if missing or '*'
     E.g instance_id = *cluster-1 or instance_id = *
     """
 
     config['instance_id'] = config.get('instance_id') or ''
     instance_id = config['instance_id']
     if instance_id.startswith('*') or not instance_id:
         prefix = instance_id.lstrip('*')
         _platform_id = platform.uname()[1] or 'instance'
         config['instance_id'] = '%s%s-%s' % (prefix, _platform_id, os.getpid())
+
+
+def get_default_user_id():
+    from rhodecode.model.db import User, Session
+    user_id = Session()\
+        .query(User.user_id)\
+        .filter(User.username == User.DEFAULT_USER)\
+        .scalar()
+    return user_id

rhodecode/model/permission.py

@@ -1,597 +1,597 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/
 
 """
 permissions model for RhodeCode
 """
 import collections
 import logging
 import traceback
 
 from sqlalchemy.exc import DatabaseError
 
 from rhodecode import events
 from rhodecode.model import BaseModel
 from rhodecode.model.db import (
     User, Permission, UserToPerm, UserRepoToPerm, UserRepoGroupToPerm,
     UserUserGroupToPerm, UserGroup, UserGroupToPerm, UserToRepoBranchPermission)
 from rhodecode.lib.utils2 import str2bool, safe_int
 
 log = logging.getLogger(__name__)
 
 
 class PermissionModel(BaseModel):
     """
     Permissions model for RhodeCode
     """
 
     cls = Permission
     global_perms = {
         'default_repo_create': None,
         # special case for create repos on write access to group
         'default_repo_create_on_write': None,
         'default_repo_group_create': None,
         'default_user_group_create': None,
         'default_fork_create': None,
         'default_inherit_default_permissions': None,
         'default_register': None,
         'default_password_reset': None,
         'default_extern_activate': None,
 
         # object permissions below
         'default_repo_perm': None,
         'default_group_perm': None,
         'default_user_group_perm': None,
 
         # branch
         'default_branch_perm': None,
     }
 
     def set_global_permission_choices(self, c_obj, gettext_translator):
         _ = gettext_translator
 
         c_obj.repo_perms_choices = [
             ('repository.none', _('None'),),
             ('repository.read', _('Read'),),
             ('repository.write', _('Write'),),
             ('repository.admin', _('Admin'),)]
 
         c_obj.group_perms_choices = [
             ('group.none', _('None'),),
             ('group.read', _('Read'),),
             ('group.write', _('Write'),),
             ('group.admin', _('Admin'),)]
 
         c_obj.user_group_perms_choices = [
             ('usergroup.none', _('None'),),
             ('usergroup.read', _('Read'),),
             ('usergroup.write', _('Write'),),
             ('usergroup.admin', _('Admin'),)]
 
         c_obj.branch_perms_choices = [
             ('branch.none', _('Protected/No Access'),),
             ('branch.merge', _('Web merge'),),
             ('branch.push', _('Push'),),
             ('branch.push_force', _('Force Push'),)]
 
         c_obj.register_choices = [
             ('hg.register.none', _('Disabled')),
             ('hg.register.manual_activate', _('Allowed with manual account activation')),
             ('hg.register.auto_activate', _('Allowed with automatic account activation')),]
 
         c_obj.password_reset_choices = [
             ('hg.password_reset.enabled', _('Allow password recovery')),
             ('hg.password_reset.hidden', _('Hide password recovery link')),
             ('hg.password_reset.disabled', _('Disable password recovery')),]
 
         c_obj.extern_activate_choices = [
             ('hg.extern_activate.manual', _('Manual activation of external account')),
             ('hg.extern_activate.auto', _('Automatic activation of external account')),]
 
         c_obj.repo_create_choices = [
             ('hg.create.none', _('Disabled')),
             ('hg.create.repository', _('Enabled'))]
 
         c_obj.repo_create_on_write_choices = [
             ('hg.create.write_on_repogroup.false', _('Disabled')),
             ('hg.create.write_on_repogroup.true', _('Enabled'))]
 
         c_obj.user_group_create_choices = [
             ('hg.usergroup.create.false', _('Disabled')),
             ('hg.usergroup.create.true', _('Enabled'))]
 
         c_obj.repo_group_create_choices = [
             ('hg.repogroup.create.false', _('Disabled')),
             ('hg.repogroup.create.true', _('Enabled'))]
 
         c_obj.fork_choices = [
             ('hg.fork.none', _('Disabled')),
             ('hg.fork.repository', _('Enabled'))]
 
         c_obj.inherit_default_permission_choices = [
             ('hg.inherit_default_perms.false', _('Disabled')),
             ('hg.inherit_default_perms.true', _('Enabled'))]
 
     def get_default_perms(self, object_perms, suffix):
         defaults = {}
         for perm in object_perms:
             # perms
             if perm.permission.permission_name.startswith('repository.'):
                 defaults['default_repo_perm' + suffix] = perm.permission.permission_name
 
             if perm.permission.permission_name.startswith('group.'):
                 defaults['default_group_perm' + suffix] = perm.permission.permission_name
 
             if perm.permission.permission_name.startswith('usergroup.'):
                 defaults['default_user_group_perm' + suffix] = perm.permission.permission_name
 
             # branch
             if perm.permission.permission_name.startswith('branch.'):
                 defaults['default_branch_perm' + suffix] = perm.permission.permission_name
 
             # creation of objects
             if perm.permission.permission_name.startswith('hg.create.write_on_repogroup'):
                 defaults['default_repo_create_on_write' + suffix] = perm.permission.permission_name
 
             elif perm.permission.permission_name.startswith('hg.create.'):
                 defaults['default_repo_create' + suffix] = perm.permission.permission_name
 
             if perm.permission.permission_name.startswith('hg.fork.'):
                 defaults['default_fork_create' + suffix] = perm.permission.permission_name
 
             if perm.permission.permission_name.startswith('hg.inherit_default_perms.'):
                 defaults['default_inherit_default_permissions' + suffix] = perm.permission.permission_name
 
             if perm.permission.permission_name.startswith('hg.repogroup.'):
                 defaults['default_repo_group_create' + suffix] = perm.permission.permission_name
 
             if perm.permission.permission_name.startswith('hg.usergroup.'):
                 defaults['default_user_group_create' + suffix] = perm.permission.permission_name
 
             # registration and external account activation
             if perm.permission.permission_name.startswith('hg.register.'):
                 defaults['default_register' + suffix] = perm.permission.permission_name
 
             if perm.permission.permission_name.startswith('hg.password_reset.'):
                 defaults['default_password_reset' + suffix] = perm.permission.permission_name
 
             if perm.permission.permission_name.startswith('hg.extern_activate.'):
                 defaults['default_extern_activate' + suffix] = perm.permission.permission_name
 
         return defaults
 
     def _make_new_user_perm(self, user, perm_name):
         log.debug('Creating new user permission:%s', perm_name)
         new = UserToPerm()
         new.user = user
         new.permission = Permission.get_by_key(perm_name)
         return new
 
     def _make_new_user_group_perm(self, user_group, perm_name):
         log.debug('Creating new user group permission:%s', perm_name)
         new = UserGroupToPerm()
         new.users_group = user_group
         new.permission = Permission.get_by_key(perm_name)
         return new
 
     def _keep_perm(self, perm_name, keep_fields):
         def get_pat(field_name):
             return {
                 # global perms
                 'default_repo_create': 'hg.create.',
                 # special case for create repos on write access to group
                 'default_repo_create_on_write': 'hg.create.write_on_repogroup.',
                 'default_repo_group_create': 'hg.repogroup.create.',
                 'default_user_group_create': 'hg.usergroup.create.',
                 'default_fork_create': 'hg.fork.',
                 'default_inherit_default_permissions': 'hg.inherit_default_perms.',
 
                 # application perms
                 'default_register': 'hg.register.',
                 'default_password_reset': 'hg.password_reset.',
                 'default_extern_activate': 'hg.extern_activate.',
 
                 # object permissions below
                 'default_repo_perm': 'repository.',
                 'default_group_perm': 'group.',
                 'default_user_group_perm': 'usergroup.',
                 # branch
                 'default_branch_perm': 'branch.',
 
             }[field_name]
         for field in keep_fields:
             pat = get_pat(field)
             if perm_name.startswith(pat):
                 return True
         return False
 
     def _clear_object_perm(self, object_perms, preserve=None):
         preserve = preserve or []
         _deleted = []
         for perm in object_perms:
             perm_name = perm.permission.permission_name
             if not self._keep_perm(perm_name, keep_fields=preserve):
                 _deleted.append(perm_name)
                 self.sa.delete(perm)
         return _deleted
 
     def _clear_user_perms(self, user_id, preserve=None):
         perms = self.sa.query(UserToPerm)\
             .filter(UserToPerm.user_id == user_id)\
             .all()
         return self._clear_object_perm(perms, preserve=preserve)
 
     def _clear_user_group_perms(self, user_group_id, preserve=None):
         perms = self.sa.query(UserGroupToPerm)\
             .filter(UserGroupToPerm.users_group_id == user_group_id)\
             .all()
         return self._clear_object_perm(perms, preserve=preserve)
 
     def _set_new_object_perms(self, obj_type, object, form_result, preserve=None):
         # clear current entries, to make this function idempotent
         # it will fix even if we define more permissions or permissions
         # are somehow missing
         preserve = preserve or []
         _global_perms = self.global_perms.copy()
         if obj_type not in ['user', 'user_group']:
             raise ValueError("obj_type must be on of 'user' or 'user_group'")
         global_perms = len(_global_perms)
         default_user_perms = len(Permission.DEFAULT_USER_PERMISSIONS)
         if global_perms != default_user_perms:
             raise Exception(
                 'Inconsistent permissions definition. Got {} vs {}'.format(
                     global_perms, default_user_perms))
 
         if obj_type == 'user':
             self._clear_user_perms(object.user_id, preserve)
         if obj_type == 'user_group':
             self._clear_user_group_perms(object.users_group_id, preserve)
 
         # now kill the keys that we want to preserve from the form.
         for key in preserve:
             del _global_perms[key]
 
         for k in _global_perms.copy():
             _global_perms[k] = form_result[k]
 
         # at that stage we validate all are passed inside form_result
         for _perm_key, perm_value in _global_perms.items():
             if perm_value is None:
                 raise ValueError('Missing permission for %s' % (_perm_key,))
 
             if obj_type == 'user':
                 p = self._make_new_user_perm(object, perm_value)
                 self.sa.add(p)
             if obj_type == 'user_group':
                 p = self._make_new_user_group_perm(object, perm_value)
                 self.sa.add(p)
 
     def _set_new_user_perms(self, user, form_result, preserve=None):
         return self._set_new_object_perms(
             'user', user, form_result, preserve)
 
     def _set_new_user_group_perms(self, user_group, form_result, preserve=None):
         return self._set_new_object_perms(
             'user_group', user_group, form_result, preserve)
 
     def set_new_user_perms(self, user, form_result):
         # calculate what to preserve from what is given in form_result
         preserve = set(self.global_perms.keys()).difference(set(form_result.keys()))
         return self._set_new_user_perms(user, form_result, preserve)
 
     def set_new_user_group_perms(self, user_group, form_result):
         # calculate what to preserve from what is given in form_result
         preserve = set(self.global_perms.keys()).difference(set(form_result.keys()))
         return self._set_new_user_group_perms(user_group, form_result, preserve)
 
     def create_permissions(self):
         """
         Create permissions for whole system
         """
         for p in Permission.PERMS:
             if not Permission.get_by_key(p[0]):
                 new_perm = Permission()
                 new_perm.permission_name = p[0]
                 new_perm.permission_longname = p[0]  # translation err with p[1]
                 self.sa.add(new_perm)
 
     def _create_default_object_permission(self, obj_type, obj, obj_perms,
                                           force=False):
         if obj_type not in ['user', 'user_group']:
             raise ValueError("obj_type must be on of 'user' or 'user_group'")
 
         def _get_group(perm_name):
             return '.'.join(perm_name.split('.')[:1])
 
         defined_perms_groups = map(
             _get_group, (x.permission.permission_name for x in obj_perms))
         log.debug('GOT ALREADY DEFINED:%s', obj_perms)
 
         if force:
             self._clear_object_perm(obj_perms)
             self.sa.commit()
             defined_perms_groups = []
         # for every default permission that needs to be created, we check if
         # it's group is already defined, if it's not we create default perm
         for perm_name in Permission.DEFAULT_USER_PERMISSIONS:
             gr = _get_group(perm_name)
             if gr not in defined_perms_groups:
                 log.debug('GR:%s not found, creating permission %s',
                           gr, perm_name)
                 if obj_type == 'user':
                     new_perm = self._make_new_user_perm(obj, perm_name)
                     self.sa.add(new_perm)
                 if obj_type == 'user_group':
                     new_perm = self._make_new_user_group_perm(obj, perm_name)
                     self.sa.add(new_perm)
 
     def create_default_user_permissions(self, user, force=False):
         """
         Creates only missing default permissions for user, if force is set it
         resets the default permissions for that user
 
         :param user:
         :param force:
         """
         user = self._get_user(user)
         obj_perms = UserToPerm.query().filter(UserToPerm.user == user).all()
         return self._create_default_object_permission(
             'user', user, obj_perms, force)
 
     def create_default_user_group_permissions(self, user_group, force=False):
         """
         Creates only missing default permissions for user group, if force is
         set it resets the default permissions for that user group
 
         :param user_group:
         :param force:
         """
         user_group = self._get_user_group(user_group)
         obj_perms = UserToPerm.query().filter(UserGroupToPerm.users_group == user_group).all()
         return self._create_default_object_permission(
             'user_group', user_group, obj_perms, force)
 
     def update_application_permissions(self, form_result):
         if 'perm_user_id' in form_result:
             perm_user = User.get(safe_int(form_result['perm_user_id']))
         else:
             # used mostly to do lookup for default user
             perm_user = User.get_by_username(form_result['perm_user_name'])
 
         try:
             # stage 1 set anonymous access
             if perm_user.username == User.DEFAULT_USER:
                 perm_user.active = str2bool(form_result['anonymous'])
                 self.sa.add(perm_user)
 
             # stage 2 reset defaults and set them from form data
             self._set_new_user_perms(perm_user, form_result, preserve=[
                 'default_repo_perm',
                 'default_group_perm',
                 'default_user_group_perm',
                 'default_branch_perm',
 
                 'default_repo_group_create',
                 'default_user_group_create',
                 'default_repo_create_on_write',
                 'default_repo_create',
                 'default_fork_create',
                 'default_inherit_default_permissions',])
 
             self.sa.commit()
         except (DatabaseError,):
             log.error(traceback.format_exc())
             self.sa.rollback()
             raise
 
     def update_user_permissions(self, form_result):
         if 'perm_user_id' in form_result:
             perm_user = User.get(safe_int(form_result['perm_user_id']))
         else:
             # used mostly to do lookup for default user
             perm_user = User.get_by_username(form_result['perm_user_name'])
         try:
             # stage 2 reset defaults and set them from form data
             self._set_new_user_perms(perm_user, form_result, preserve=[
                 'default_repo_perm',
                 'default_group_perm',
                 'default_user_group_perm',
                 'default_branch_perm',
 
                 'default_register',
                 'default_password_reset',
                 'default_extern_activate'])
             self.sa.commit()
         except (DatabaseError,):
             log.error(traceback.format_exc())
             self.sa.rollback()
             raise
 
     def update_user_group_permissions(self, form_result):
         if 'perm_user_group_id' in form_result:
             perm_user_group = UserGroup.get(safe_int(form_result['perm_user_group_id']))
         else:
             # used mostly to do lookup for default user
             perm_user_group = UserGroup.get_by_group_name(form_result['perm_user_group_name'])
         try:
             # stage 2 reset defaults and set them from form data
             self._set_new_user_group_perms(perm_user_group, form_result, preserve=[
                 'default_repo_perm',
                 'default_group_perm',
                 'default_user_group_perm',
                 'default_branch_perm',
 
                 'default_register',
                 'default_password_reset',
                 'default_extern_activate'])
             self.sa.commit()
         except (DatabaseError,):
             log.error(traceback.format_exc())
             self.sa.rollback()
             raise
 
     def update_object_permissions(self, form_result):
         if 'perm_user_id' in form_result:
             perm_user = User.get(safe_int(form_result['perm_user_id']))
         else:
             # used mostly to do lookup for default user
             perm_user = User.get_by_username(form_result['perm_user_name'])
         try:
 
             # stage 2 reset defaults and set them from form data
             self._set_new_user_perms(perm_user, form_result, preserve=[
                 'default_repo_group_create',
                 'default_user_group_create',
                 'default_repo_create_on_write',
                 'default_repo_create',
                 'default_fork_create',
                 'default_inherit_default_permissions',
                 'default_branch_perm',
 
                 'default_register',
                 'default_password_reset',
                 'default_extern_activate'])
 
             # overwrite default repo permissions
             if form_result['overwrite_default_repo']:
                 _def_name = form_result['default_repo_perm'].split('repository.')[-1]
                 _def = Permission.get_by_key('repository.' + _def_name)
                 for r2p in self.sa.query(UserRepoToPerm)\
                                .filter(UserRepoToPerm.user == perm_user)\
                                .all():
                     # don't reset PRIVATE repositories
                     if not r2p.repository.private:
                         r2p.permission = _def
                         self.sa.add(r2p)
 
             # overwrite default repo group permissions
             if form_result['overwrite_default_group']:
                 _def_name = form_result['default_group_perm'].split('group.')[-1]
                 _def = Permission.get_by_key('group.' + _def_name)
                 for g2p in self.sa.query(UserRepoGroupToPerm)\
                                .filter(UserRepoGroupToPerm.user == perm_user)\
                                .all():
                     g2p.permission = _def
                     self.sa.add(g2p)
 
             # overwrite default user group permissions
             if form_result['overwrite_default_user_group']:
                 _def_name = form_result['default_user_group_perm'].split('usergroup.')[-1]
                 # user groups
                 _def = Permission.get_by_key('usergroup.' + _def_name)
                 for g2p in self.sa.query(UserUserGroupToPerm)\
                                .filter(UserUserGroupToPerm.user == perm_user)\
                                .all():
                     g2p.permission = _def
                     self.sa.add(g2p)
 
             # COMMIT
             self.sa.commit()
         except (DatabaseError,):
             log.exception('Failed to set default object permissions')
             self.sa.rollback()
             raise
 
     def update_branch_permissions(self, form_result):
         if 'perm_user_id' in form_result:
             perm_user = User.get(safe_int(form_result['perm_user_id']))
         else:
             # used mostly to do lookup for default user
             perm_user = User.get_by_username(form_result['perm_user_name'])
         try:
 
             # stage 2 reset defaults and set them from form data
             self._set_new_user_perms(perm_user, form_result, preserve=[
                 'default_repo_perm',
                 'default_group_perm',
                 'default_user_group_perm',
 
                 'default_repo_group_create',
                 'default_user_group_create',
                 'default_repo_create_on_write',
                 'default_repo_create',
                 'default_fork_create',
                 'default_inherit_default_permissions',
 
                 'default_register',
                 'default_password_reset',
                 'default_extern_activate'])
 
             # overwrite default branch permissions
             if form_result['overwrite_default_branch']:
                 _def_name = \
                     form_result['default_branch_perm'].split('branch.')[-1]
 
                 _def = Permission.get_by_key('branch.' + _def_name)
 
                 user_perms = UserToRepoBranchPermission.query()\
                     .join(UserToRepoBranchPermission.user_repo_to_perm)\
                     .filter(UserRepoToPerm.user == perm_user).all()
 
                 for g2p in user_perms:
                     g2p.permission = _def
                     self.sa.add(g2p)
 
             # COMMIT
             self.sa.commit()
         except (DatabaseError,):
             log.exception('Failed to set default branch permissions')
             self.sa.rollback()
             raise
 
     def get_users_with_repo_write(self, db_repo):
         write_plus = ['repository.write', 'repository.admin']
-        default_user_id = User.get_default_user().user_id
+        default_user_id = User.get_default_user_id()
         user_write_permissions = collections.OrderedDict()
 
         # write+ and DEFAULT user for inheritance
         for perm in db_repo.permissions():
             if perm.permission in write_plus or perm.user_id == default_user_id:
                 user_write_permissions[perm.user_id] = perm
         return user_write_permissions
 
     def get_user_groups_with_repo_write(self, db_repo):
         write_plus = ['repository.write', 'repository.admin']
         user_group_write_permissions = collections.OrderedDict()
 
         # write+ and DEFAULT user for inheritance
         for p in db_repo.permission_user_groups():
             if p.permission in write_plus:
                 user_group_write_permissions[p.users_group_id] = p
         return user_group_write_permissions
 
     def trigger_permission_flush(self, affected_user_ids):
         events.trigger(events.UserPermissionsChange(affected_user_ids))
 
     def flush_user_permission_caches(self, changes, affected_user_ids=None):
         affected_user_ids = affected_user_ids or []
 
         for change in changes['added'] + changes['updated'] + changes['deleted']:
             if change['type'] == 'user':
                 affected_user_ids.append(change['id'])
             if change['type'] == 'user_group':
                 user_group = UserGroup.get(safe_int(change['id']))
                 if user_group:
                     group_members_ids = [x.user_id for x in user_group.members]
                     affected_user_ids.extend(group_members_ids)
 
         self.trigger_permission_flush(affected_user_ids)
 
         return affected_user_ids

rhodecode/model/validators.py

@@ -1,1116 +1,1116 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/
 
 """
 Set of generic validators
 """
 
 
 import os
 import re
 import logging
 import collections
 
 import formencode
 import ipaddress
 from formencode.validators import (
     UnicodeString, OneOf, Int, Number, Regex, Email, Bool, StringBoolean, Set,
     NotEmpty, IPAddress, CIDR, String, FancyValidator
 )
 
 from sqlalchemy.sql.expression import true
 from sqlalchemy.util import OrderedSet
 from pyramid import compat
 
 from rhodecode.authentication import (
     legacy_plugin_prefix, _import_legacy_plugin)
 from rhodecode.authentication.base import loadplugin
 from rhodecode.apps._base import ADMIN_PREFIX
 from rhodecode.lib.auth import HasRepoGroupPermissionAny, HasPermissionAny
 from rhodecode.lib.utils import repo_name_slug, make_db_config
 from rhodecode.lib.utils2 import safe_int, str2bool, aslist, md5, safe_unicode
 from rhodecode.lib.vcs.backends.git.repository import GitRepository
 from rhodecode.lib.vcs.backends.hg.repository import MercurialRepository
 from rhodecode.lib.vcs.backends.svn.repository import SubversionRepository
 from rhodecode.model.db import (
     RepoGroup, Repository, UserGroup, User, ChangesetStatus, Gist)
 from rhodecode.model.settings import VcsSettingsModel
 
 # silence warnings and pylint
 UnicodeString, OneOf, Int, Number, Regex, Email, Bool, StringBoolean, Set, \
     NotEmpty, IPAddress, CIDR, String, FancyValidator
 
 log = logging.getLogger(__name__)
 
 
 class _Missing(object):
     pass
 
 
 Missing = _Missing()
 
 
 def M(self, key, state, **kwargs):
     """
     returns string from self.message based on given key,
     passed kw params are used to substitute %(named)s params inside
     translated strings
 
     :param msg:
     :param state:
     """
 
     #state._ = staticmethod(_)
     # inject validator into state object
     return self.message(key, state, **kwargs)
 
 
 def UniqueList(localizer, convert=None):
     _ = localizer
 
     class _validator(formencode.FancyValidator):
         """
         Unique List !
         """
         messages = {
             'empty': _(u'Value cannot be an empty list'),
             'missing_value': _(u'Value cannot be an empty list'),
         }
 
         def _to_python(self, value, state):
             ret_val = []
 
             def make_unique(value):
                 seen = []
                 return [c for c in value if not (c in seen or seen.append(c))]
 
             if isinstance(value, list):
                 ret_val = make_unique(value)
             elif isinstance(value, set):
                 ret_val = make_unique(list(value))
             elif isinstance(value, tuple):
                 ret_val = make_unique(list(value))
             elif value is None:
                 ret_val = []
             else:
                 ret_val = [value]
 
             if convert:
                 ret_val = map(convert, ret_val)
             return ret_val
 
         def empty_value(self, value):
             return []
     return _validator
 
 
 def UniqueListFromString(localizer):
     _ = localizer
 
     class _validator(UniqueList(localizer)):
         def _to_python(self, value, state):
             if isinstance(value, compat.string_types):
                 value = aslist(value, ',')
             return super(_validator, self)._to_python(value, state)
     return _validator
 
 
 def ValidSvnPattern(localizer, section, repo_name=None):
     _ = localizer
 
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'pattern_exists': _(u'Pattern already exists'),
         }
 
         def validate_python(self, value, state):
             if not value:
                 return
             model = VcsSettingsModel(repo=repo_name)
             ui_settings = model.get_svn_patterns(section=section)
             for entry in ui_settings:
                 if value == entry.value:
                     msg = M(self, 'pattern_exists', state)
                     raise formencode.Invalid(msg, value, state)
     return _validator
 
 
 def ValidUsername(localizer, edit=False, old_data=None):
     _ = localizer
     old_data = old_data or {}
 
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'username_exists': _(u'Username "%(username)s" already exists'),
             'system_invalid_username':
                 _(u'Username "%(username)s" is forbidden'),
             'invalid_username':
                 _(u'Username may only contain alphanumeric characters '
                     u'underscores, periods or dashes and must begin with '
                     u'alphanumeric character or underscore')
         }
 
         def validate_python(self, value, state):
             if value in ['default', 'new_user']:
                 msg = M(self, 'system_invalid_username', state, username=value)
                 raise formencode.Invalid(msg, value, state)
             # check if user is unique
             old_un = None
             if edit:
                 old_un = User.get(old_data.get('user_id')).username
 
             if old_un != value or not edit:
                 if User.get_by_username(value, case_insensitive=True):
                     msg = M(self, 'username_exists', state, username=value)
                     raise formencode.Invalid(msg, value, state)
 
             if (re.match(r'^[\w]{1}[\w\-\.]{0,254}$', value)
                     is None):
                 msg = M(self, 'invalid_username', state)
                 raise formencode.Invalid(msg, value, state)
     return _validator
 
 
 def ValidRepoUser(localizer, allow_disabled=False):
     _ = localizer
 
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_username': _(u'Username %(username)s is not valid'),
             'disabled_username': _(u'Username %(username)s is disabled')
         }
 
         def validate_python(self, value, state):
             try:
                 user = User.query().filter(User.username == value).one()
             except Exception:
                 msg = M(self, 'invalid_username', state, username=value)
                 raise formencode.Invalid(
                     msg, value, state, error_dict={'username': msg}
                 )
             if user and (not allow_disabled and not user.active):
                 msg = M(self, 'disabled_username', state, username=value)
                 raise formencode.Invalid(
                     msg, value, state, error_dict={'username': msg}
                 )
     return _validator
 
 
 def ValidUserGroup(localizer, edit=False, old_data=None):
     _ = localizer
     old_data = old_data or {}
 
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_group': _(u'Invalid user group name'),
             'group_exist': _(u'User group `%(usergroup)s` already exists'),
             'invalid_usergroup_name':
                 _(u'user group name may only contain alphanumeric '
                   u'characters underscores, periods or dashes and must begin '
                   u'with alphanumeric character')
         }
 
         def validate_python(self, value, state):
             if value in ['default']:
                 msg = M(self, 'invalid_group', state)
                 raise formencode.Invalid(
                     msg, value, state, error_dict={'users_group_name': msg}
                 )
             # check if group is unique
             old_ugname = None
             if edit:
                 old_id = old_data.get('users_group_id')
                 old_ugname = UserGroup.get(old_id).users_group_name
 
             if old_ugname != value or not edit:
                 is_existing_group = UserGroup.get_by_group_name(
                     value, case_insensitive=True)
                 if is_existing_group:
                     msg = M(self, 'group_exist', state, usergroup=value)
                     raise formencode.Invalid(
                         msg, value, state, error_dict={'users_group_name': msg}
                     )
 
             if re.match(r'^[a-zA-Z0-9]{1}[a-zA-Z0-9\-\_\.]+$', value) is None:
                 msg = M(self, 'invalid_usergroup_name', state)
                 raise formencode.Invalid(
                     msg, value, state, error_dict={'users_group_name': msg}
                 )
     return _validator
 
 
 def ValidRepoGroup(localizer, edit=False, old_data=None, can_create_in_root=False):
     _ = localizer
     old_data = old_data or {}
 
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'group_parent_id': _(u'Cannot assign this group as parent'),
             'group_exists': _(u'Group "%(group_name)s" already exists'),
             'repo_exists': _(u'Repository with name "%(group_name)s" '
                              u'already exists'),
             'permission_denied': _(u"no permission to store repository group"
                                    u"in this location"),
             'permission_denied_root': _(
                 u"no permission to store repository group "
                 u"in root location")
         }
 
         def _to_python(self, value, state):
             group_name = repo_name_slug(value.get('group_name', ''))
             group_parent_id = safe_int(value.get('group_parent_id'))
             gr = RepoGroup.get(group_parent_id)
             if gr:
                 parent_group_path = gr.full_path
                 # value needs to be aware of group name in order to check
                 # db key This is an actual just the name to store in the
                 # database
                 group_name_full = (
                     parent_group_path + RepoGroup.url_sep() + group_name)
             else:
                 group_name_full = group_name
 
             value['group_name'] = group_name
             value['group_name_full'] = group_name_full
             value['group_parent_id'] = group_parent_id
             return value
 
         def validate_python(self, value, state):
 
             old_group_name = None
             group_name = value.get('group_name')
             group_name_full = value.get('group_name_full')
             group_parent_id = safe_int(value.get('group_parent_id'))
             if group_parent_id == -1:
                 group_parent_id = None
 
             group_obj = RepoGroup.get(old_data.get('group_id'))
             parent_group_changed = False
             if edit:
                 old_group_name = group_obj.group_name
                 old_group_parent_id = group_obj.group_parent_id
 
                 if group_parent_id != old_group_parent_id:
                     parent_group_changed = True
 
                 # TODO: mikhail: the following if statement is not reached
                 # since group_parent_id's OneOf validation fails before.
                 # Can be removed.
 
                 # check against setting a parent of self
                 parent_of_self = (
                     old_data['group_id'] == group_parent_id
                     if group_parent_id else False
                 )
                 if parent_of_self:
                     msg = M(self, 'group_parent_id', state)
                     raise formencode.Invalid(
                         msg, value, state, error_dict={'group_parent_id': msg}
                     )
 
             # group we're moving current group inside
             child_group = None
             if group_parent_id:
                 child_group = RepoGroup.query().filter(
                     RepoGroup.group_id == group_parent_id).scalar()
 
             # do a special check that we cannot move a group to one of
             # it's children
             if edit and child_group:
                 parents = [x.group_id for x in child_group.parents]
                 move_to_children = old_data['group_id'] in parents
                 if move_to_children:
                     msg = M(self, 'group_parent_id', state)
                     raise formencode.Invalid(
                         msg, value, state, error_dict={'group_parent_id': msg})
 
             # Check if we have permission to store in the parent.
             # Only check if the parent group changed.
             if parent_group_changed:
                 if child_group is None:
                     if not can_create_in_root:
                         msg = M(self, 'permission_denied_root', state)
                         raise formencode.Invalid(
                             msg, value, state,
                             error_dict={'group_parent_id': msg})
                 else:
                     valid = HasRepoGroupPermissionAny('group.admin')
                     forbidden = not valid(
                         child_group.group_name, 'can create group validator')
                     if forbidden:
                         msg = M(self, 'permission_denied', state)
                         raise formencode.Invalid(
                             msg, value, state,
                             error_dict={'group_parent_id': msg})
 
             # if we change the name or it's new group, check for existing names
             # or repositories with the same name
             if old_group_name != group_name_full or not edit:
                 # check group
                 gr = RepoGroup.get_by_group_name(group_name_full)
                 if gr:
                     msg = M(self, 'group_exists', state, group_name=group_name)
                     raise formencode.Invalid(
                         msg, value, state, error_dict={'group_name': msg})
 
                 # check for same repo
                 repo = Repository.get_by_repo_name(group_name_full)
                 if repo:
                     msg = M(self, 'repo_exists', state, group_name=group_name)
                     raise formencode.Invalid(
                         msg, value, state, error_dict={'group_name': msg})
     return _validator
 
 
 def ValidPassword(localizer):
     _ = localizer
 
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_password':
                 _(u'Invalid characters (non-ascii) in password')
         }
 
         def validate_python(self, value, state):
             try:
                 (value or '').decode('ascii')
             except UnicodeError:
                 msg = M(self, 'invalid_password', state)
                 raise formencode.Invalid(msg, value, state,)
     return _validator
 
 
 def ValidPasswordsMatch(
         localizer, passwd='new_password',
         passwd_confirmation='password_confirmation'):
     _ = localizer
 
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'password_mismatch': _(u'Passwords do not match'),
         }
 
         def validate_python(self, value, state):
 
             pass_val = value.get('password') or value.get(passwd)
             if pass_val != value[passwd_confirmation]:
                 msg = M(self, 'password_mismatch', state)
                 raise formencode.Invalid(
                     msg, value, state,
                     error_dict={passwd: msg, passwd_confirmation: msg}
                 )
     return _validator
 
 
 def ValidAuth(localizer):
     _ = localizer
 
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_password': _(u'invalid password'),
             'invalid_username': _(u'invalid user name'),
             'disabled_account': _(u'Your account is disabled')
         }
 
         def validate_python(self, value, state):
             from rhodecode.authentication.base import authenticate, HTTP_TYPE
 
             password = value['password']
             username = value['username']
 
             if not authenticate(username, password, '', HTTP_TYPE,
                                 skip_missing=True):
                 user = User.get_by_username(username)
                 if user and not user.active:
                     log.warning('user %s is disabled', username)
                     msg = M(self, 'disabled_account', state)
                     raise formencode.Invalid(
                         msg, value, state, error_dict={'username': msg}
                     )
                 else:
                     log.warning('user `%s` failed to authenticate', username)
                     msg = M(self, 'invalid_username', state)
                     msg2 = M(self, 'invalid_password', state)
                     raise formencode.Invalid(
                         msg, value, state,
                         error_dict={'username': msg, 'password': msg2}
                     )
     return _validator
 
 
 def ValidRepoName(localizer, edit=False, old_data=None):
     old_data = old_data or {}
     _ = localizer
 
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_repo_name':
                 _(u'Repository name %(repo)s is disallowed'),
             # top level
             'repository_exists': _(u'Repository with name %(repo)s '
                                    u'already exists'),
             'group_exists': _(u'Repository group with name "%(repo)s" '
                               u'already exists'),
             # inside a group
             'repository_in_group_exists': _(u'Repository with name %(repo)s '
                                             u'exists in group "%(group)s"'),
             'group_in_group_exists': _(
                 u'Repository group with name "%(repo)s" '
                 u'exists in group "%(group)s"'),
         }
 
         def _to_python(self, value, state):
             repo_name = repo_name_slug(value.get('repo_name', ''))
             repo_group = value.get('repo_group')
             if repo_group:
                 gr = RepoGroup.get(repo_group)
                 group_path = gr.full_path
                 group_name = gr.group_name
                 # value needs to be aware of group name in order to check
                 # db key This is an actual just the name to store in the
                 # database
                 repo_name_full = group_path + RepoGroup.url_sep() + repo_name
             else:
                 group_name = group_path = ''
                 repo_name_full = repo_name
 
             value['repo_name'] = repo_name
             value['repo_name_full'] = repo_name_full
             value['group_path'] = group_path
             value['group_name'] = group_name
             return value
 
         def validate_python(self, value, state):
 
             repo_name = value.get('repo_name')
             repo_name_full = value.get('repo_name_full')
             group_path = value.get('group_path')
             group_name = value.get('group_name')
 
             if repo_name in [ADMIN_PREFIX, '']:
                 msg = M(self, 'invalid_repo_name', state, repo=repo_name)
                 raise formencode.Invalid(
                     msg, value, state, error_dict={'repo_name': msg})
 
             rename = old_data.get('repo_name') != repo_name_full
             create = not edit
             if rename or create:
 
                 if group_path:
                     if Repository.get_by_repo_name(repo_name_full):
                         msg = M(self, 'repository_in_group_exists', state,
                                 repo=repo_name, group=group_name)
                         raise formencode.Invalid(
                             msg, value, state, error_dict={'repo_name': msg})
                     if RepoGroup.get_by_group_name(repo_name_full):
                         msg = M(self, 'group_in_group_exists', state,
                                 repo=repo_name, group=group_name)
                         raise formencode.Invalid(
                             msg, value, state, error_dict={'repo_name': msg})
                 else:
                     if RepoGroup.get_by_group_name(repo_name_full):
                         msg = M(self, 'group_exists', state, repo=repo_name)
                         raise formencode.Invalid(
                             msg, value, state, error_dict={'repo_name': msg})
 
                     if Repository.get_by_repo_name(repo_name_full):
                         msg = M(
                             self, 'repository_exists', state, repo=repo_name)
                         raise formencode.Invalid(
                             msg, value, state, error_dict={'repo_name': msg})
             return value
     return _validator
 
 
 def ValidForkName(localizer, *args, **kwargs):
     _ = localizer
 
     return ValidRepoName(localizer, *args, **kwargs)
 
 
 def SlugifyName(localizer):
     _ = localizer
     
     class _validator(formencode.validators.FancyValidator):
 
         def _to_python(self, value, state):
             return repo_name_slug(value)
 
         def validate_python(self, value, state):
             pass
     return _validator
 
 
 def CannotHaveGitSuffix(localizer):
     _ = localizer
     
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'has_git_suffix':
                 _(u'Repository name cannot end with .git'),
         }
 
         def _to_python(self, value, state):
             return value
 
         def validate_python(self, value, state):
             if value and value.endswith('.git'):
                 msg = M(
                     self, 'has_git_suffix', state)
                 raise formencode.Invalid(
                     msg, value, state, error_dict={'repo_name': msg})
     return _validator
 
 
 def ValidCloneUri(localizer):
     _ = localizer
     
     class InvalidCloneUrl(Exception):
         allowed_prefixes = ()
 
     def url_handler(repo_type, url):
         config = make_db_config(clear_session=False)
         if repo_type == 'hg':
             allowed_prefixes = ('http', 'svn+http', 'git+http')
 
             if 'http' in url[:4]:
                 # initially check if it's at least the proper URL
                 # or does it pass basic auth
                 MercurialRepository.check_url(url, config)
             elif 'svn+http' in url[:8]:  # svn->hg import
                 SubversionRepository.check_url(url, config)
             elif 'git+http' in url[:8]:  # git->hg import
                 raise NotImplementedError()
             else:
                 exc = InvalidCloneUrl('Clone from URI %s not allowed. '
                                       'Allowed url must start with one of %s'
                                       % (url, ','.join(allowed_prefixes)))
                 exc.allowed_prefixes = allowed_prefixes
                 raise exc
 
         elif repo_type == 'git':
             allowed_prefixes = ('http', 'svn+http', 'hg+http')
             if 'http' in url[:4]:
                 # initially check if it's at least the proper URL
                 # or does it pass basic auth
                 GitRepository.check_url(url, config)
             elif 'svn+http' in url[:8]:  # svn->git import
                 raise NotImplementedError()
             elif 'hg+http' in url[:8]:  # hg->git import
                 raise NotImplementedError()
             else:
                 exc = InvalidCloneUrl('Clone from URI %s not allowed. '
                                       'Allowed url must start with one of %s'
                                       % (url, ','.join(allowed_prefixes)))
                 exc.allowed_prefixes = allowed_prefixes
                 raise exc
 
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'clone_uri': _(u'invalid clone url for %(rtype)s repository'),
             'invalid_clone_uri': _(
                 u'Invalid clone url, provide a valid clone '
                 u'url starting with one of %(allowed_prefixes)s')
         }
 
         def validate_python(self, value, state):
             repo_type = value.get('repo_type')
             url = value.get('clone_uri')
 
             if url:
                 try:
                     url_handler(repo_type, url)
                 except InvalidCloneUrl as e:
                     log.warning(e)
                     msg = M(self, 'invalid_clone_uri', state, rtype=repo_type,
                             allowed_prefixes=','.join(e.allowed_prefixes))
                     raise formencode.Invalid(msg, value, state,
                                              error_dict={'clone_uri': msg})
                 except Exception:
                     log.exception('Url validation failed')
                     msg = M(self, 'clone_uri', state, rtype=repo_type)
                     raise formencode.Invalid(msg, value, state,
                                              error_dict={'clone_uri': msg})
     return _validator
 
 
 def ValidForkType(localizer, old_data=None):
     _ = localizer
     old_data = old_data or {}
 
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_fork_type': _(u'Fork have to be the same type as parent')
         }
 
         def validate_python(self, value, state):
             if old_data['repo_type'] != value:
                 msg = M(self, 'invalid_fork_type', state)
                 raise formencode.Invalid(
                     msg, value, state, error_dict={'repo_type': msg}
                 )
     return _validator
 
 
 def CanWriteGroup(localizer, old_data=None):
     _ = localizer
 
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'permission_denied': _(
                 u"You do not have the permission "
                 u"to create repositories in this group."),
             'permission_denied_root': _(
                 u"You do not have the permission to store repositories in "
                 u"the root location.")
         }
 
         def _to_python(self, value, state):
             # root location
             if value in [-1, "-1"]:
                 return None
             return value
 
         def validate_python(self, value, state):
             gr = RepoGroup.get(value)
             gr_name = gr.group_name if gr else None  # None means ROOT location
             # create repositories with write permission on group is set to true
             create_on_write = HasPermissionAny(
                 'hg.create.write_on_repogroup.true')()
             group_admin = HasRepoGroupPermissionAny('group.admin')(
                 gr_name, 'can write into group validator')
             group_write = HasRepoGroupPermissionAny('group.write')(
                 gr_name, 'can write into group validator')
             forbidden = not (group_admin or (group_write and create_on_write))
             can_create_repos = HasPermissionAny(
                 'hg.admin', 'hg.create.repository')
             gid = (old_data['repo_group'].get('group_id')
                    if (old_data and 'repo_group' in old_data) else None)
             value_changed = gid != safe_int(value)
             new = not old_data
             # do check if we changed the value, there's a case that someone got
             # revoked write permissions to a repository, he still created, we
             # don't need to check permission if he didn't change the value of
             # groups in form box
             if value_changed or new:
                 # parent group need to be existing
                 if gr and forbidden:
                     msg = M(self, 'permission_denied', state)
                     raise formencode.Invalid(
                         msg, value, state, error_dict={'repo_type': msg}
                     )
                 # check if we can write to root location !
                 elif gr is None and not can_create_repos():
                     msg = M(self, 'permission_denied_root', state)
                     raise formencode.Invalid(
                         msg, value, state, error_dict={'repo_type': msg}
                     )
     return _validator
 
 
 def ValidPerms(localizer, type_='repo'):
     _ = localizer
     if type_ == 'repo_group':
         EMPTY_PERM = 'group.none'
     elif type_ == 'repo':
         EMPTY_PERM = 'repository.none'
     elif type_ == 'user_group':
         EMPTY_PERM = 'usergroup.none'
 
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'perm_new_member_name':
                 _(u'This username or user group name is not valid')
         }
 
         def _to_python(self, value, state):
             perm_updates = OrderedSet()
             perm_additions = OrderedSet()
             perm_deletions = OrderedSet()
             # build a list of permission to update/delete and new permission
 
             # Read the perm_new_member/perm_del_member attributes and group
             # them by they IDs
             new_perms_group = collections.defaultdict(dict)
             del_perms_group = collections.defaultdict(dict)
             for k, v in value.copy().iteritems():
                 if k.startswith('perm_del_member'):
                     # delete from org storage so we don't process that later
                     del value[k]
                     # part is `id`, `type`
                     _type, part = k.split('perm_del_member_')
                     args = part.split('_')
                     if len(args) == 2:
                         _key, pos = args
                         del_perms_group[pos][_key] = v
                 if k.startswith('perm_new_member'):
                     # delete from org storage so we don't process that later
                     del value[k]
                     # part is `id`, `type`, `perm`
                     _type, part = k.split('perm_new_member_')
                     args = part.split('_')
                     if len(args) == 2:
                         _key, pos = args
                         new_perms_group[pos][_key] = v
 
             # store the deletes
             for k in sorted(del_perms_group.keys()):
                 perm_dict = del_perms_group[k]
                 del_member = perm_dict.get('id')
                 del_type = perm_dict.get('type')
                 if del_member and del_type:
                     perm_deletions.add(
                         (del_member, None, del_type))
 
             # store additions in order of how they were added in web form
             for k in sorted(new_perms_group.keys()):
                 perm_dict = new_perms_group[k]
                 new_member = perm_dict.get('id')
                 new_type = perm_dict.get('type')
                 new_perm = perm_dict.get('perm')
                 if new_member and new_perm and new_type:
                     perm_additions.add(
                         (new_member, new_perm, new_type))
 
             # get updates of permissions
             # (read the existing radio button states)
-            default_user_id = User.get_default_user().user_id
+            default_user_id = User.get_default_user_id()
 
             for k, update_value in value.iteritems():
                 if k.startswith('u_perm_') or k.startswith('g_perm_'):
                     obj_type = k[0]
                     obj_id = k[7:]
                     update_type = {'u': 'user',
                                    'g': 'user_group'}[obj_type]
 
                     if obj_type == 'u' and safe_int(obj_id) == default_user_id:
                         if str2bool(value.get('repo_private')):
                             # prevent from updating default user permissions
                             # when this repository is marked as private
                             update_value = EMPTY_PERM
 
                     perm_updates.add(
                         (obj_id, update_value, update_type))
 
             value['perm_additions'] = []  # propagated later
             value['perm_updates'] = list(perm_updates)
             value['perm_deletions'] = list(perm_deletions)
 
             updates_map = dict(
                 (x[0], (x[1], x[2])) for x in value['perm_updates'])
             # make sure Additions don't override updates.
             for member_id, perm, member_type in list(perm_additions):
                 if member_id in updates_map:
                     perm = updates_map[member_id][0]
                 value['perm_additions'].append((member_id, perm, member_type))
 
                 # on new entries validate users they exist and they are active !
                 # this leaves feedback to the form
                 try:
                     if member_type == 'user':
                         User.query()\
                             .filter(User.active == true())\
                             .filter(User.user_id == member_id).one()
                     if member_type == 'user_group':
                         UserGroup.query()\
                             .filter(UserGroup.users_group_active == true())\
                             .filter(UserGroup.users_group_id == member_id)\
                             .one()
 
                 except Exception:
                     log.exception('Updated permission failed: org_exc:')
                     msg = M(self, 'perm_new_member_type', state)
                     raise formencode.Invalid(
                         msg, value, state, error_dict={
                             'perm_new_member_name': msg}
                     )
             return value
     return _validator
 
 
 def ValidPath(localizer):
     _ = localizer
 
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_path': _(u'This is not a valid path')
         }
 
         def validate_python(self, value, state):
             if not os.path.isdir(value):
                 msg = M(self, 'invalid_path', state)
                 raise formencode.Invalid(
                     msg, value, state, error_dict={'paths_root_path': msg}
                 )
     return _validator
 
 
 def UniqSystemEmail(localizer, old_data=None):
     _ = localizer
     old_data = old_data or {}
 
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'email_taken': _(u'This e-mail address is already taken')
         }
 
         def _to_python(self, value, state):
             return value.lower()
 
         def validate_python(self, value, state):
             if (old_data.get('email') or '').lower() != value:
                 user = User.get_by_email(value, case_insensitive=True)
                 if user:
                     msg = M(self, 'email_taken', state)
                     raise formencode.Invalid(
                         msg, value, state, error_dict={'email': msg}
                     )
     return _validator
 
 
 def ValidSystemEmail(localizer):
     _ = localizer
 
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'non_existing_email': _(u'e-mail "%(email)s" does not exist.')
         }
 
         def _to_python(self, value, state):
             return value.lower()
 
         def validate_python(self, value, state):
             user = User.get_by_email(value, case_insensitive=True)
             if user is None:
                 msg = M(self, 'non_existing_email', state, email=value)
                 raise formencode.Invalid(
                     msg, value, state, error_dict={'email': msg}
                 )
     return _validator
 
 
 def NotReviewedRevisions(localizer, repo_id):
     _ = localizer
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'rev_already_reviewed':
                 _(u'Revisions %(revs)s are already part of pull request '
                   u'or have set status'),
         }
 
         def validate_python(self, value, state):
             # check revisions if they are not reviewed, or a part of another
             # pull request
             statuses = ChangesetStatus.query()\
                 .filter(ChangesetStatus.revision.in_(value))\
                 .filter(ChangesetStatus.repo_id == repo_id)\
                 .all()
 
             errors = []
             for status in statuses:
                 if status.pull_request_id:
                     errors.append(['pull_req', status.revision[:12]])
                 elif status.status:
                     errors.append(['status', status.revision[:12]])
 
             if errors:
                 revs = ','.join([x[1] for x in errors])
                 msg = M(self, 'rev_already_reviewed', state, revs=revs)
                 raise formencode.Invalid(
                     msg, value, state, error_dict={'revisions': revs})
 
     return _validator
 
 
 def ValidIp(localizer):
     _ = localizer
 
     class _validator(CIDR):
         messages = {
             'badFormat': _(u'Please enter a valid IPv4 or IpV6 address'),
             'illegalBits': _(
                 u'The network size (bits) must be within the range '
                 u'of 0-32 (not %(bits)r)'),
         }
 
         # we ovveride the default to_python() call
         def to_python(self, value, state):
             v = super(_validator, self).to_python(value, state)
             v = safe_unicode(v.strip())
             net = ipaddress.ip_network(address=v, strict=False)
             return str(net)
 
         def validate_python(self, value, state):
             try:
                 addr = safe_unicode(value.strip())
                 # this raises an ValueError if address is not IpV4 or IpV6
                 ipaddress.ip_network(addr, strict=False)
             except ValueError:
                 raise formencode.Invalid(self.message('badFormat', state),
                                          value, state)
     return _validator
 
 
 def FieldKey(localizer):
     _ = localizer
 
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'badFormat': _(
                 u'Key name can only consist of letters, '
                 u'underscore, dash or numbers'),
         }
 
         def validate_python(self, value, state):
             if not re.match('[a-zA-Z0-9_-]+$', value):
                 raise formencode.Invalid(self.message('badFormat', state),
                                          value, state)
     return _validator
 
 
 def ValidAuthPlugins(localizer):
     _ = localizer
 
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'import_duplicate': _(
                 u'Plugins %(loaded)s and %(next_to_load)s '
                 u'both export the same name'),
             'missing_includeme': _(
                 u'The plugin "%(plugin_id)s" is missing an includeme '
                 u'function.'),
             'import_error': _(
                 u'Can not load plugin "%(plugin_id)s"'),
             'no_plugin': _(
                 u'No plugin available with ID "%(plugin_id)s"'),
         }
 
         def _to_python(self, value, state):
             # filter empty values
             return filter(lambda s: s not in [None, ''], value)
 
         def _validate_legacy_plugin_id(self, plugin_id, value, state):
             """
             Validates that the plugin import works. It also checks that the
             plugin has an includeme attribute.
             """
             try:
                 plugin = _import_legacy_plugin(plugin_id)
             except Exception as e:
                 log.exception(
                     'Exception during import of auth legacy plugin "{}"'
                     .format(plugin_id))
                 msg = M(self, 'import_error', state, plugin_id=plugin_id)
                 raise formencode.Invalid(msg, value, state)
 
             if not hasattr(plugin, 'includeme'):
                 msg = M(self, 'missing_includeme', state, plugin_id=plugin_id)
                 raise formencode.Invalid(msg, value, state)
 
             return plugin
 
         def _validate_plugin_id(self, plugin_id, value, state):
             """
             Plugins are already imported during app start up. Therefore this
             validation only retrieves the plugin from the plugin registry and
             if it returns something not None everything is OK.
             """
             plugin = loadplugin(plugin_id)
 
             if plugin is None:
                 msg = M(self, 'no_plugin', state, plugin_id=plugin_id)
                 raise formencode.Invalid(msg, value, state)
 
             return plugin
 
         def validate_python(self, value, state):
             unique_names = {}
             for plugin_id in value:
 
                 # Validate legacy or normal plugin.
                 if plugin_id.startswith(legacy_plugin_prefix):
                     plugin = self._validate_legacy_plugin_id(
                         plugin_id, value, state)
                 else:
                     plugin = self._validate_plugin_id(plugin_id, value, state)
 
                 # Only allow unique plugin names.
                 if plugin.name in unique_names:
                     msg = M(self, 'import_duplicate', state,
                             loaded=unique_names[plugin.name],
                             next_to_load=plugin)
                     raise formencode.Invalid(msg, value, state)
                 unique_names[plugin.name] = plugin
     return _validator
 
 
 def ValidPattern(localizer):
     _ = localizer
 
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'bad_format': _(u'Url must start with http or /'),
         }
 
         def _to_python(self, value, state):
             patterns = []
 
             prefix = 'new_pattern'
             for name, v in value.iteritems():
                 pattern_name = '_'.join((prefix, 'pattern'))
                 if name.startswith(pattern_name):
                     new_item_id = name[len(pattern_name)+1:]
 
                     def _field(name):
                         return '%s_%s_%s' % (prefix, name, new_item_id)
 
                     values = {
                         'issuetracker_pat': value.get(_field('pattern')),
                         'issuetracker_url': value.get(_field('url')),
                         'issuetracker_pref': value.get(_field('prefix')),
                         'issuetracker_desc': value.get(_field('description'))
                     }
                     new_uid = md5(values['issuetracker_pat'])
 
                     has_required_fields = (
                         values['issuetracker_pat']
                         and values['issuetracker_url'])
 
                     if has_required_fields:
                         # validate url that it starts with http or /
                         # otherwise it can lead to JS injections
                         # e.g specifig javascript:<malicios code>
                         if not values['issuetracker_url'].startswith(('http', '/')):
                             raise formencode.Invalid(
                                 self.message('bad_format', state),
                                 value, state)
 
                         settings = [
                             ('_'.join((key, new_uid)), values[key], 'unicode')
                             for key in values]
                         patterns.append(settings)
 
             value['patterns'] = patterns
             delete_patterns = value.get('uid') or []
             if not isinstance(delete_patterns, (list, tuple)):
                 delete_patterns = [delete_patterns]
             value['delete_patterns'] = delete_patterns
             return value
     return _validator

rhodecode/tests/lib/test_auth.py

@@ -1,632 +1,632 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 os
 from hashlib import sha1
 
 import pytest
 from mock import patch
 
 from rhodecode.lib import auth
 from rhodecode.lib.utils2 import md5
 from rhodecode.model.auth_token import AuthTokenModel
 from rhodecode.model.db import Session, User
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.user import UserModel
 from rhodecode.model.user_group import UserGroupModel
 
 
 def test_perm_origin_dict():
     pod = auth.PermOriginDict()
     pod['thing'] = 'read', 'default', 1
     assert pod['thing'] == 'read'
 
     assert pod.perm_origin_stack == {
         'thing': [('read', 'default', 1)]}
 
     pod['thing'] = 'write', 'admin', 1
     assert pod['thing'] == 'write'
 
     assert pod.perm_origin_stack == {
         'thing': [('read', 'default', 1), ('write', 'admin', 1)]}
 
     pod['other'] = 'write', 'default', 8
 
     assert pod.perm_origin_stack == {
         'other': [('write', 'default', 8)],
         'thing': [('read', 'default', 1), ('write', 'admin', 1)]}
 
     pod['other'] = 'none', 'override', 8
 
     assert pod.perm_origin_stack == {
         'other': [('write', 'default', 8), ('none', 'override', 8)],
         'thing': [('read', 'default', 1), ('write', 'admin', 1)]}
 
     with pytest.raises(ValueError):
         pod['thing'] = 'read'
 
 
 def test_cached_perms_data(user_regular, backend_random):
     permissions = get_permissions(user_regular)
     repo_name = backend_random.repo.repo_name
     expected_global_permissions = {
         'repository.read', 'group.read', 'usergroup.read'}
     assert expected_global_permissions.issubset(permissions['global'])
     assert permissions['repositories'][repo_name] == 'repository.read'
 
 
 def test_cached_perms_data_with_admin_user(user_regular, backend_random):
     permissions = get_permissions(user_regular, user_is_admin=True)
     repo_name = backend_random.repo.repo_name
     assert 'hg.admin' in permissions['global']
     assert permissions['repositories'][repo_name] == 'repository.admin'
 
 
 def test_cached_perms_data_with_admin_user_extended_calculation(user_regular, backend_random):
     permissions = get_permissions(user_regular, user_is_admin=True,
                                   calculate_super_admin=True)
     repo_name = backend_random.repo.repo_name
     assert 'hg.admin' in permissions['global']
     assert permissions['repositories'][repo_name] == 'repository.admin'
 
 
 def test_cached_perms_data_user_group_global_permissions(user_util):
     user, user_group = user_util.create_user_with_group()
     user_group.inherit_default_permissions = False
 
     granted_permission = 'repository.write'
     UserGroupModel().grant_perm(user_group, granted_permission)
     Session().commit()
 
     permissions = get_permissions(user)
     assert granted_permission in permissions['global']
 
 
 @pytest.mark.xfail(reason="Not implemented, see TODO note")
 def test_cached_perms_data_user_group_global_permissions_(user_util):
     user, user_group = user_util.create_user_with_group()
 
     granted_permission = 'repository.write'
     UserGroupModel().grant_perm(user_group, granted_permission)
     Session().commit()
 
     permissions = get_permissions(user)
     assert granted_permission in permissions['global']
 
 
 def test_cached_perms_data_user_global_permissions(user_util):
     user = user_util.create_user()
     UserModel().grant_perm(user, 'repository.none')
     Session().commit()
 
     permissions = get_permissions(user, user_inherit_default_permissions=True)
     assert 'repository.read' in permissions['global']
 
 
 def test_cached_perms_data_repository_permissions_on_private_repository(
         backend_random, user_util):
     user, user_group = user_util.create_user_with_group()
 
     repo = backend_random.create_repo()
     repo.private = True
 
     granted_permission = 'repository.write'
     RepoModel().grant_user_group_permission(
         repo, user_group.users_group_name, granted_permission)
     Session().commit()
 
     permissions = get_permissions(user)
     assert permissions['repositories'][repo.repo_name] == granted_permission
 
 
 def test_cached_perms_data_repository_permissions_for_owner(
         backend_random, user_util):
     user = user_util.create_user()
 
     repo = backend_random.create_repo()
     repo.user_id = user.user_id
 
     permissions = get_permissions(user)
     assert permissions['repositories'][repo.repo_name] == 'repository.admin'
 
     # TODO: johbo: Make cleanup in UserUtility smarter, then remove this hack
-    repo.user_id = User.get_default_user().user_id
+    repo.user_id = User.get_default_user_id()
 
 
 def test_cached_perms_data_repository_permissions_not_inheriting_defaults(
         backend_random, user_util):
     user = user_util.create_user()
     repo = backend_random.create_repo()
 
     # Don't inherit default object permissions
     UserModel().grant_perm(user, 'hg.inherit_default_perms.false')
     Session().commit()
 
     permissions = get_permissions(user)
     assert permissions['repositories'][repo.repo_name] == 'repository.none'
 
 
 def test_cached_perms_data_default_permissions_on_repository_group(user_util):
     # Have a repository group with default permissions set
     repo_group = user_util.create_repo_group()
     default_user = User.get_default_user()
     user_util.grant_user_permission_to_repo_group(
         repo_group, default_user, 'repository.write')
     user = user_util.create_user()
 
     permissions = get_permissions(user)
     assert permissions['repositories_groups'][repo_group.group_name] == \
         'repository.write'
 
 
 def test_cached_perms_data_default_permissions_on_repository_group_owner(
         user_util):
     # Have a repository group
     repo_group = user_util.create_repo_group()
     default_user = User.get_default_user()
 
     # Add a permission for the default user to hit the code path
     user_util.grant_user_permission_to_repo_group(
         repo_group, default_user, 'repository.write')
 
     # Have an owner of the group
     user = user_util.create_user()
     repo_group.user_id = user.user_id
 
     permissions = get_permissions(user)
     assert permissions['repositories_groups'][repo_group.group_name] == \
         'group.admin'
 
 
 def test_cached_perms_data_default_permissions_on_repository_group_no_inherit(
         user_util):
     # Have a repository group
     repo_group = user_util.create_repo_group()
     default_user = User.get_default_user()
 
     # Add a permission for the default user to hit the code path
     user_util.grant_user_permission_to_repo_group(
         repo_group, default_user, 'repository.write')
 
     # Don't inherit default object permissions
     user = user_util.create_user()
     UserModel().grant_perm(user, 'hg.inherit_default_perms.false')
     Session().commit()
 
     permissions = get_permissions(user)
     assert permissions['repositories_groups'][repo_group.group_name] == \
         'group.none'
 
 
 def test_cached_perms_data_repository_permissions_from_user_group(
         user_util, backend_random):
     user, user_group = user_util.create_user_with_group()
 
     # Needs a second user group to make sure that we select the right
     # permissions.
     user_group2 = user_util.create_user_group()
     UserGroupModel().add_user_to_group(user_group2, user)
 
     repo = backend_random.create_repo()
 
     RepoModel().grant_user_group_permission(
         repo, user_group.users_group_name, 'repository.read')
     RepoModel().grant_user_group_permission(
         repo, user_group2.users_group_name, 'repository.write')
     Session().commit()
 
     permissions = get_permissions(user)
     assert permissions['repositories'][repo.repo_name] == 'repository.write'
 
 
 def test_cached_perms_data_repository_permissions_from_user_group_owner(
         user_util, backend_random):
     user, user_group = user_util.create_user_with_group()
 
     repo = backend_random.create_repo()
     repo.user_id = user.user_id
 
     RepoModel().grant_user_group_permission(
         repo, user_group.users_group_name, 'repository.write')
     Session().commit()
 
     permissions = get_permissions(user)
     assert permissions['repositories'][repo.repo_name] == 'repository.admin'
 
 
 def test_cached_perms_data_user_repository_permissions(
         user_util, backend_random):
     user = user_util.create_user()
     repo = backend_random.create_repo()
     granted_permission = 'repository.write'
     RepoModel().grant_user_permission(repo, user, granted_permission)
     Session().commit()
 
     permissions = get_permissions(user)
     assert permissions['repositories'][repo.repo_name] == granted_permission
 
 
 def test_cached_perms_data_user_repository_permissions_explicit(
         user_util, backend_random):
     user = user_util.create_user()
     repo = backend_random.create_repo()
     granted_permission = 'repository.none'
     RepoModel().grant_user_permission(repo, user, granted_permission)
     Session().commit()
 
     permissions = get_permissions(user, explicit=True)
     assert permissions['repositories'][repo.repo_name] == granted_permission
 
 
 def test_cached_perms_data_user_repository_permissions_owner(
         user_util, backend_random):
     user = user_util.create_user()
     repo = backend_random.create_repo()
     repo.user_id = user.user_id
     RepoModel().grant_user_permission(repo, user, 'repository.write')
     Session().commit()
 
     permissions = get_permissions(user)
     assert permissions['repositories'][repo.repo_name] == 'repository.admin'
 
 
 def test_cached_perms_data_repository_groups_permissions_inherited(
         user_util, backend_random):
     user, user_group = user_util.create_user_with_group()
 
     # Needs a second group to hit the last condition
     user_group2 = user_util.create_user_group()
     UserGroupModel().add_user_to_group(user_group2, user)
 
     repo_group = user_util.create_repo_group()
 
     user_util.grant_user_group_permission_to_repo_group(
         repo_group, user_group, 'group.read')
     user_util.grant_user_group_permission_to_repo_group(
         repo_group, user_group2, 'group.write')
 
     permissions = get_permissions(user)
     assert permissions['repositories_groups'][repo_group.group_name] == \
         'group.write'
 
 
 def test_cached_perms_data_repository_groups_permissions_inherited_owner(
         user_util, backend_random):
     user, user_group = user_util.create_user_with_group()
     repo_group = user_util.create_repo_group()
     repo_group.user_id = user.user_id
 
     granted_permission = 'group.write'
     user_util.grant_user_group_permission_to_repo_group(
         repo_group, user_group, granted_permission)
 
     permissions = get_permissions(user)
     assert permissions['repositories_groups'][repo_group.group_name] == \
         'group.admin'
 
 
 def test_cached_perms_data_repository_groups_permissions(
         user_util, backend_random):
     user = user_util.create_user()
 
     repo_group = user_util.create_repo_group()
 
     granted_permission = 'group.write'
     user_util.grant_user_permission_to_repo_group(
         repo_group, user, granted_permission)
 
     permissions = get_permissions(user)
     assert permissions['repositories_groups'][repo_group.group_name] == \
         'group.write'
 
 
 def test_cached_perms_data_repository_groups_permissions_explicit(
         user_util, backend_random):
     user = user_util.create_user()
 
     repo_group = user_util.create_repo_group()
 
     granted_permission = 'group.none'
     user_util.grant_user_permission_to_repo_group(
         repo_group, user, granted_permission)
 
     permissions = get_permissions(user, explicit=True)
     assert permissions['repositories_groups'][repo_group.group_name] == \
         'group.none'
 
 
 def test_cached_perms_data_repository_groups_permissions_owner(
         user_util, backend_random):
     user = user_util.create_user()
 
     repo_group = user_util.create_repo_group()
     repo_group.user_id = user.user_id
 
     granted_permission = 'group.write'
     user_util.grant_user_permission_to_repo_group(
         repo_group, user, granted_permission)
 
     permissions = get_permissions(user)
     assert permissions['repositories_groups'][repo_group.group_name] == \
         'group.admin'
 
 
 def test_cached_perms_data_user_group_permissions_inherited(
         user_util, backend_random):
     user, user_group = user_util.create_user_with_group()
     user_group2 = user_util.create_user_group()
     UserGroupModel().add_user_to_group(user_group2, user)
 
     target_user_group = user_util.create_user_group()
 
     user_util.grant_user_group_permission_to_user_group(
         target_user_group, user_group, 'usergroup.read')
     user_util.grant_user_group_permission_to_user_group(
         target_user_group, user_group2, 'usergroup.write')
 
     permissions = get_permissions(user)
     assert permissions['user_groups'][target_user_group.users_group_name] == \
         'usergroup.write'
 
 
 def test_cached_perms_data_user_group_permissions(
         user_util, backend_random):
     user = user_util.create_user()
     user_group = user_util.create_user_group()
     UserGroupModel().grant_user_permission(user_group, user, 'usergroup.write')
     Session().commit()
 
     permissions = get_permissions(user)
     assert permissions['user_groups'][user_group.users_group_name] == \
         'usergroup.write'
 
 
 def test_cached_perms_data_user_group_permissions_explicit(
         user_util, backend_random):
     user = user_util.create_user()
     user_group = user_util.create_user_group()
     UserGroupModel().grant_user_permission(user_group, user, 'usergroup.none')
     Session().commit()
 
     permissions = get_permissions(user, explicit=True)
     assert permissions['user_groups'][user_group.users_group_name] == \
         'usergroup.none'
 
 
 def test_cached_perms_data_user_group_permissions_not_inheriting_defaults(
         user_util, backend_random):
     user = user_util.create_user()
     user_group = user_util.create_user_group()
 
     # Don't inherit default object permissions
     UserModel().grant_perm(user, 'hg.inherit_default_perms.false')
     Session().commit()
 
     permissions = get_permissions(user)
     assert permissions['user_groups'][user_group.users_group_name] == \
         'usergroup.none'
 
 
 def test_permission_calculator_admin_permissions(
         user_util, backend_random):
     user = user_util.create_user()
     user_group = user_util.create_user_group()
     repo = backend_random.repo
     repo_group = user_util.create_repo_group()
 
     calculator = auth.PermissionCalculator(
         user.user_id, {}, False, False, True, 'higherwin')
     permissions = calculator._calculate_admin_permissions()
 
     assert permissions['repositories_groups'][repo_group.group_name] == \
         'group.admin'
     assert permissions['user_groups'][user_group.users_group_name] == \
         'usergroup.admin'
     assert permissions['repositories'][repo.repo_name] == 'repository.admin'
     assert 'hg.admin' in permissions['global']
 
 
 def test_permission_calculator_repository_permissions_robustness_from_group(
         user_util, backend_random):
     user, user_group = user_util.create_user_with_group()
 
     RepoModel().grant_user_group_permission(
         backend_random.repo, user_group.users_group_name, 'repository.write')
 
     calculator = auth.PermissionCalculator(
         user.user_id, {}, False, False, False, 'higherwin')
     calculator._calculate_repository_permissions()
 
 
 def test_permission_calculator_repository_permissions_robustness_from_user(
         user_util, backend_random):
     user = user_util.create_user()
 
     RepoModel().grant_user_permission(
         backend_random.repo, user, 'repository.write')
     Session().commit()
 
     calculator = auth.PermissionCalculator(
         user.user_id, {}, False, False, False, 'higherwin')
     calculator._calculate_repository_permissions()
 
 
 def test_permission_calculator_repo_group_permissions_robustness_from_group(
         user_util, backend_random):
     user, user_group = user_util.create_user_with_group()
     repo_group = user_util.create_repo_group()
 
     user_util.grant_user_group_permission_to_repo_group(
         repo_group, user_group, 'group.write')
 
     calculator = auth.PermissionCalculator(
         user.user_id, {}, False, False, False, 'higherwin')
     calculator._calculate_repository_group_permissions()
 
 
 def test_permission_calculator_repo_group_permissions_robustness_from_user(
         user_util, backend_random):
     user = user_util.create_user()
     repo_group = user_util.create_repo_group()
 
     user_util.grant_user_permission_to_repo_group(
         repo_group, user, 'group.write')
 
     calculator = auth.PermissionCalculator(
         user.user_id, {}, False, False, False, 'higherwin')
     calculator._calculate_repository_group_permissions()
 
 
 def test_permission_calculator_user_group_permissions_robustness_from_group(
         user_util, backend_random):
     user, user_group = user_util.create_user_with_group()
     target_user_group = user_util.create_user_group()
 
     user_util.grant_user_group_permission_to_user_group(
         target_user_group, user_group, 'usergroup.write')
 
     calculator = auth.PermissionCalculator(
         user.user_id, {}, False, False, False, 'higherwin')
     calculator._calculate_user_group_permissions()
 
 
 def test_permission_calculator_user_group_permissions_robustness_from_user(
         user_util, backend_random):
     user = user_util.create_user()
     target_user_group = user_util.create_user_group()
 
     user_util.grant_user_permission_to_user_group(
         target_user_group, user, 'usergroup.write')
 
     calculator = auth.PermissionCalculator(
         user.user_id, {}, False, False, False, 'higherwin')
     calculator._calculate_user_group_permissions()
 
 
 @pytest.mark.parametrize("algo, new_permission, old_permission, expected", [
     ('higherwin', 'repository.none', 'repository.none', 'repository.none'),
     ('higherwin', 'repository.read', 'repository.none', 'repository.read'),
     ('lowerwin', 'repository.write', 'repository.write', 'repository.write'),
     ('lowerwin', 'repository.read', 'repository.write', 'repository.read'),
 ])
 def test_permission_calculator_choose_permission(
         user_regular, algo, new_permission, old_permission, expected):
     calculator = auth.PermissionCalculator(
         user_regular.user_id, {}, False, False, False, algo)
     result = calculator._choose_permission(new_permission, old_permission)
     assert result == expected
 
 
 def test_permission_calculator_choose_permission_raises_on_wrong_algo(
         user_regular):
     calculator = auth.PermissionCalculator(
         user_regular.user_id, {}, False, False, False, 'invalid')
     result = calculator._choose_permission(
         'repository.read', 'repository.read')
     # TODO: johbo: This documents the existing behavior. Think of an
     # improvement.
     assert result is None
 
 
 def test_auth_user_get_cookie_store_for_normal_user(user_util):
     user = user_util.create_user()
     auth_user = auth.AuthUser(user_id=user.user_id)
     expected_data = {
         'username': user.username,
         'user_id': user.user_id,
         'password': md5(user.password),
         'is_authenticated': False
     }
     assert auth_user.get_cookie_store() == expected_data
 
 
 def test_auth_user_get_cookie_store_for_default_user():
     default_user = User.get_default_user()
     auth_user = auth.AuthUser()
     expected_data = {
         'username': User.DEFAULT_USER,
         'user_id': default_user.user_id,
         'password': md5(default_user.password),
         'is_authenticated': True
     }
     assert auth_user.get_cookie_store() == expected_data
 
 
 def get_permissions(user, **kwargs):
     """
     Utility filling in useful defaults into the call to `_cached_perms_data`.
 
     Fill in `**kwargs` if specific values are needed for a test.
     """
     call_args = {
         'user_id': user.user_id,
         'scope': {},
         'user_is_admin': False,
         'user_inherit_default_permissions': False,
         'explicit': False,
         'algo': 'higherwin',
         'calculate_super_admin': False,
     }
     call_args.update(kwargs)
     permissions = auth._cached_perms_data(**call_args)
     return permissions
 
 
 class TestGenerateAuthToken(object):
     def test_salt_is_used_when_specified(self):
         salt = 'abcde'
         user_name = 'test_user'
         result = auth.generate_auth_token(user_name, salt)
         expected_result = sha1(user_name + salt).hexdigest()
         assert result == expected_result
 
     def test_salt_is_geneated_when_not_specified(self):
         user_name = 'test_user'
         random_salt = os.urandom(16)
         with patch.object(auth, 'os') as os_mock:
             os_mock.urandom.return_value = random_salt
             result = auth.generate_auth_token(user_name)
         expected_result = sha1(user_name + random_salt).hexdigest()
         assert result == expected_result
 
 
 @pytest.mark.parametrize("test_token, test_roles, auth_result, expected_tokens", [
     ('', None, False,
      []),
     ('wrongtoken', None, False,
      []),
     ('abracadabra_vcs', [AuthTokenModel.cls.ROLE_API], False,
      [('abracadabra_api', AuthTokenModel.cls.ROLE_API, -1)]),
     ('abracadabra_api', [AuthTokenModel.cls.ROLE_API], True,
      [('abracadabra_api', AuthTokenModel.cls.ROLE_API, -1)]),
     ('abracadabra_api', [AuthTokenModel.cls.ROLE_API], True,
      [('abracadabra_api', AuthTokenModel.cls.ROLE_API, -1),
       ('abracadabra_http', AuthTokenModel.cls.ROLE_HTTP, -1)]),
 ])
 def test_auth_by_token(test_token, test_roles, auth_result, expected_tokens,
                        user_util):
     user = user_util.create_user()
     user_id = user.user_id
     for token, role, expires in expected_tokens:
         new_token = AuthTokenModel().create(user_id, u'test-token', expires, role)
         new_token.api_key = token  # inject known name for testing...
 
     assert auth_result == user.authenticate_by_token(
         test_token, roles=test_roles)

rhodecode/tests/models/test_user_permissions_on_repo_groups.py

@@ -1,325 +1,325 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 functools
 
 import pytest
 
 from rhodecode.model.db import RepoGroup, User
 from rhodecode.model.meta import Session
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.tests.models.common import (
     _create_project_tree, check_tree_perms, _get_perms, _check_expected_count,
     expected_count, _destroy_project_tree)
 
 
 test_u1_id = None
 _get_repo_perms = None
 _get_group_perms = None
 
 
 @pytest.fixture(autouse=True)
 def setup_read_permission():
     permissions_setup_func()
 
 
 def permissions_setup_func(group_name='g0', perm='group.read', recursive='all',
                            user_id=None):
     """
     Resets all permissions to perm attribute
     """
     if not user_id:
         user_id = test_u1_id
         # called by the @with_setup decorator also reset the default user stuff
         permissions_setup_func(group_name, perm, recursive,
-                               user_id=User.get_default_user().user_id)
+                               user_id=User.get_default_user_id())
 
     # TODO: DRY, compare test_user_group:permissions_setup_func
     repo_group = RepoGroup.get_by_group_name(group_name=group_name)
     if not repo_group:
         raise Exception('Cannot get group %s' % group_name)
 
     perm_updates = [[user_id, perm, 'user']]
     RepoGroupModel().update_permissions(repo_group,
                                         perm_updates=perm_updates,
                                         recursive=recursive, check_perms=False)
     Session().commit()
 
 
 @pytest.fixture(scope='module', autouse=True)
 def prepare(request, baseapp):
     global test_u1_id, _get_repo_perms, _get_group_perms
     test_u1 = _create_project_tree()
     Session().commit()
     test_u1_id = test_u1.user_id
     _get_repo_perms = functools.partial(_get_perms, key='repositories',
                                         test_u1_id=test_u1_id)
     _get_group_perms = functools.partial(_get_perms, key='repositories_groups',
                                          test_u1_id=test_u1_id)
 
     @request.addfinalizer
     def cleanup():
         _destroy_project_tree(test_u1_id)
 
 
 def test_user_permissions_on_group_without_recursive_mode():
     # set permission to g0 non-recursive mode
     recursive = 'none'
     group = 'g0'
     permissions_setup_func(group, 'group.write', recursive=recursive)
 
     items = [x for x in _get_repo_perms(group, recursive)]
     expected = 0
     assert len(items) == expected, ' %s != %s' % (len(items), expected)
     for name, perm in items:
         check_tree_perms(name, perm, group, 'repository.read')
 
     items = [x for x in _get_group_perms(group, recursive)]
     expected = 1
     assert len(items) == expected, ' %s != %s' % (len(items), expected)
     for name, perm in items:
         check_tree_perms(name, perm, group, 'group.write')
 
 
 def test_user_permissions_on_group_without_recursive_mode_subgroup():
     # set permission to g0 non-recursive mode
     recursive = 'none'
     group = 'g0/g0_1'
     permissions_setup_func(group, 'group.write', recursive=recursive)
 
     items = [x for x in _get_repo_perms(group, recursive)]
     expected = 0
     assert len(items) == expected, ' %s != %s' % (len(items), expected)
     for name, perm in items:
         check_tree_perms(name, perm, group, 'repository.read')
 
     items = [x for x in _get_group_perms(group, recursive)]
     expected = 1
     assert len(items) == expected, ' %s != %s' % (len(items), expected)
     for name, perm in items:
         check_tree_perms(name, perm, group, 'group.write')
 
 
 def test_user_permissions_on_group_with_recursive_mode():
 
     # set permission to g0 recursive mode, all children including
     # other repos and groups should have this permission now set !
     recursive = 'all'
     group = 'g0'
     permissions_setup_func(group, 'group.write', recursive=recursive)
 
     repo_items = [x for x in _get_repo_perms(group, recursive)]
     items = [x for x in _get_group_perms(group, recursive)]
     _check_expected_count(items, repo_items, expected_count(group, True))
 
     for name, perm in repo_items:
         check_tree_perms(name, perm, group, 'repository.write')
 
     for name, perm in items:
         check_tree_perms(name, perm, group, 'group.write')
 
 
 def test_user_permissions_on_group_with_recursive_mode_for_default_user():
 
     # set permission to g0 recursive mode, all children including
     # other repos and groups should have this permission now set !
     recursive = 'all'
     group = 'g0'
-    default_user_id = User.get_default_user().user_id
+    default_user_id = User.get_default_user_id()
     permissions_setup_func(group, 'group.write', recursive=recursive,
                            user_id=default_user_id)
 
     # change default to get perms for default user
     _get_repo_perms = functools.partial(_get_perms, key='repositories',
                                         test_u1_id=default_user_id)
     _get_group_perms = functools.partial(_get_perms, key='repositories_groups',
                                          test_u1_id=default_user_id)
 
     repo_items = [x for x in _get_repo_perms(group, recursive)]
     items = [x for x in _get_group_perms(group, recursive)]
     _check_expected_count(items, repo_items, expected_count(group, True))
 
     for name, perm in repo_items:
         check_tree_perms(name, perm, group, 'repository.write')
 
     for name, perm in items:
         check_tree_perms(name, perm, group, 'group.write')
 
 
 def test_user_permissions_on_group_with_recursive_mode_inner_group():
     # set permission to g0_3 group to none
     recursive = 'all'
     group = 'g0/g0_3'
     permissions_setup_func(group, 'group.none', recursive=recursive)
 
     repo_items = [x for x in _get_repo_perms(group, recursive)]
     items = [x for x in _get_group_perms(group, recursive)]
     _check_expected_count(items, repo_items, expected_count(group, True))
 
     for name, perm in repo_items:
         check_tree_perms(name, perm, group, 'repository.none')
 
     for name, perm in items:
         check_tree_perms(name, perm, group, 'group.none')
 
 
 def test_user_permissions_on_group_with_recursive_mode_deepest():
     # set permission to g0_3 group to none
     recursive = 'all'
     group = 'g0/g0_1/g0_1_1'
     permissions_setup_func(group, 'group.write', recursive=recursive)
 
     repo_items = [x for x in _get_repo_perms(group, recursive)]
     items = [x for x in _get_group_perms(group, recursive)]
     _check_expected_count(items, repo_items, expected_count(group, True))
 
     for name, perm in repo_items:
         check_tree_perms(name, perm, group, 'repository.write')
 
     for name, perm in items:
         check_tree_perms(name, perm, group, 'group.write')
 
 
 def test_user_permissions_on_group_with_recursive_mode_only_with_repos():
     # set permission to g0_3 group to none
     recursive = 'all'
     group = 'g0/g0_2'
     permissions_setup_func(group, 'group.admin', recursive=recursive)
 
     repo_items = [x for x in _get_repo_perms(group, recursive)]
     items = [x for x in _get_group_perms(group, recursive)]
     _check_expected_count(items, repo_items, expected_count(group, True))
 
     for name, perm in repo_items:
         check_tree_perms(name, perm, group, 'repository.admin')
 
     for name, perm in items:
         check_tree_perms(name, perm, group, 'group.admin')
 
 
 def test_user_permissions_on_group_with_recursive_repo_mode_for_default_user():
     # set permission to g0/g0_1 recursive repos only mode, all children
     # including other repos should have this permission now set, inner groups
     # are excluded!
     recursive = 'repos'
     group = 'g0/g0_1'
     perm = 'group.none'
-    default_user_id = User.get_default_user().user_id
+    default_user_id = User.get_default_user_id()
 
     # TODO: workaround due to different setup calls, adept to py.test style
     permissions_setup_func()
     permissions_setup_func(group, perm, recursive=recursive,
                            user_id=default_user_id)
 
     # change default to get perms for default user
     _get_repo_perms = functools.partial(_get_perms, key='repositories',
                                         test_u1_id=default_user_id)
     _get_group_perms = functools.partial(_get_perms, key='repositories_groups',
                                          test_u1_id=default_user_id)
 
     repo_items = [x for x in _get_repo_perms(group, recursive)]
     items = [x for x in _get_group_perms(group, recursive)]
     _check_expected_count(items, repo_items, expected_count(group, True))
 
     for name, perm in repo_items:
         check_tree_perms(name, perm, group, 'repository.none')
 
     for name, perm in items:
         # permission is set with repos only mode, but we also change the
         # permission on the group we trigger the apply to children from, thus
         # we need to change its permission check
         old_perm = 'group.read'
         if name == group:
             old_perm = perm
         check_tree_perms(name, perm, group, old_perm)
 
 
 def test_user_permissions_on_group_with_recursive_repo_mode_inner_group():
     # set permission to g0_3 group to none, with recursive repos only
     recursive = 'repos'
     group = 'g0/g0_3'
     perm = 'group.none'
     permissions_setup_func(group, perm, recursive=recursive)
 
     repo_items = [x for x in _get_repo_perms(group, recursive)]
     items = [x for x in _get_group_perms(group, recursive)]
     _check_expected_count(items, repo_items, expected_count(group, True))
 
     for name, perm in repo_items:
         check_tree_perms(name, perm, group, 'repository.none')
 
     for name, perm in items:
         # permission is set with repos only mode, but we also change the
         # permission on the group we trigger the apply to children from, thus
         # we need to change its permission check
         old_perm = 'group.read'
         if name == group:
             old_perm = perm
         check_tree_perms(name, perm, group, old_perm)
 
 
 def test_user_permissions_on_group_with_rec_group_mode_for_default_user():
     # set permission to g0/g0_1 with recursive groups only mode, all children
     # sincluding other groups should have this permission now set. repositories
     # should remain intact as we use groups only mode !
     recursive = 'groups'
     group = 'g0/g0_1'
-    default_user_id = User.get_default_user().user_id
+    default_user_id = User.get_default_user_id()
 
     # TODO: workaround due to different setup calls, adept to py.test style
     permissions_setup_func()
     permissions_setup_func(group, 'group.write', recursive=recursive,
                            user_id=default_user_id)
 
     # change default to get perms for default user
     _get_repo_perms = functools.partial(_get_perms, key='repositories',
                                         test_u1_id=default_user_id)
     _get_group_perms = functools.partial(_get_perms, key='repositories_groups',
                                          test_u1_id=default_user_id)
 
     repo_items = [x for x in _get_repo_perms(group, recursive)]
     items = [x for x in _get_group_perms(group, recursive)]
     _check_expected_count(items, repo_items, expected_count(group, True))
 
     for name, perm in repo_items:
         check_tree_perms(name, perm, group, 'repository.read')
 
     for name, perm in items:
         check_tree_perms(name, perm, group, 'group.write')
 
 
 def test_user_permissions_on_group_with_recursive_group_mode_inner_group():
     # set permission to g0_3 group to none, with recursive mode for groups only
     recursive = 'groups'
     group = 'g0/g0_3'
 
     # TODO: workaround due to different setup calls, adept to py.test style
     permissions_setup_func()
     permissions_setup_func(group, 'group.none', recursive=recursive)
 
     repo_items = [x for x in _get_repo_perms(group, recursive)]
     items = [x for x in _get_group_perms(group, recursive)]
     _check_expected_count(items, repo_items, expected_count(group, True))
 
     for name, perm in repo_items:
         check_tree_perms(name, perm, group, 'repository.read')
 
     for name, perm in items:
         check_tree_perms(name, perm, group, 'group.none')