kallithea Changeset - 2ce710e81e61

Changeset - 2ce710e81e61

Parent rev.

Child rev.

[Not reviewed]

default

0 11 0

Mads Kiilerich (mads) - 6 years ago 2020-04-11 20:18:29
mads@kiilerich.com

Grafted from: 7c0220adaae7

permissions: drop hg.create.write_on_repogroup "Repository creation with group write access" setting

Simplify permissions system and get rid of some confusing tech debt.

Before, the global 'write_on_repogroup' setting controlled what write
permission on a repo group meant.

With this change, users can create repositories in a repo group if and only if
they have write access. Write access to a repo group will now mean the
permission to create repositories in it.

Write access to repo groups must be granted explicitly. There should not be any
other reason to grant write access than to allow users to create repos. There
is thus no upgrade concerns for this change.

An admin that doesn't want users to create repos in a repogroup should just not
give them write access.

These global settings might still exist in the database, but is ignored and no
longer used and do no harm.

11 files changed with 14 insertions and 50 deletions:

kallithea/controllers/admin/permissions.py

kallithea/controllers/admin/repos.py

kallithea/controllers/forks.py

kallithea/lib/auth.py

kallithea/model/db.py

kallithea/model/forms.py

kallithea/model/permission.py

kallithea/model/validators.py

kallithea/templates/admin/permissions/permissions_globals.html

kallithea/templates/index_base.html

kallithea/tests/models/test_permissions.py

0 comments (0 inline, 0 general)

kallithea/controllers/admin/permissions.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # 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 General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.admin.permissions
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 permissions controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 27, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode
 from formencode import htmlfill
 from tg import request
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound
 from kallithea.config.routing import url
 from kallithea.lib import helpers as h
 from kallithea.lib.auth import AuthUser, HasPermissionAnyDecorator, LoginRequired
 from kallithea.lib.base import BaseController, render
 from kallithea.model.db import User, UserIpMap
 from kallithea.model.forms import DefaultPermissionsForm
 from kallithea.model.meta import Session
 from kallithea.model.permission import PermissionModel
 log = logging.getLogger(__name__)
 class PermissionsController(BaseController):
     """REST Controller styled on the Atom Publishing Protocol"""
     # To properly map this controller, ensure your config/routing.py
     # file has a resource setup:
     #     map.resource('permission', 'permissions')
     @LoginRequired()
     @HasPermissionAnyDecorator('hg.admin')
     def _before(self, *args, **kwargs):
         super(PermissionsController, self)._before(*args, **kwargs)
     def __load_data(self):
         c.repo_perms_choices = [('repository.none', _('None'),),
                                    ('repository.read', _('Read'),),
                                    ('repository.write', _('Write'),),
                                    ('repository.admin', _('Admin'),)]
         c.group_perms_choices = [('group.none', _('None'),),
                                  ('group.read', _('Read'),),
                                  ('group.write', _('Write'),),
                                  ('group.admin', _('Admin'),)]
         c.user_group_perms_choices = [('usergroup.none', _('None'),),
                                       ('usergroup.read', _('Read'),),
                                       ('usergroup.write', _('Write'),),
                                       ('usergroup.admin', _('Admin'),)]
         c.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.extern_activate_choices = [
             ('hg.extern_activate.manual', _('Manual activation of external account')),
             ('hg.extern_activate.auto', _('Automatic activation of external account')),
+        ]
         c.repo_create_choices = [('hg.create.none', _('Disabled')),
                                  ('hg.create.repository', _('Enabled'))]
         c.repo_create_on_write_choices = [
             ('hg.create.write_on_repogroup.true', _('Enabled')),
             ('hg.create.write_on_repogroup.false', _('Disabled')),
+        ]
         c.user_group_create_choices = [('hg.usergroup.create.false', _('Disabled')),
                                        ('hg.usergroup.create.true', _('Enabled'))]
         c.fork_choices = [('hg.fork.none', _('Disabled')),
                           ('hg.fork.repository', _('Enabled'))]
     def permission_globals(self):
         c.active = 'globals'
         self.__load_data()
         if request.POST:
             _form = DefaultPermissionsForm(
                 [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],
                 [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.register_choices],
                 [x[0] for x in c.extern_activate_choices])()
             try:
                 form_result = _form.to_python(dict(request.POST))
                 form_result.update({'perm_user_name': 'default'})
                 PermissionModel().update(form_result)
                 Session().commit()
                 h.flash(_('Global permissions updated successfully'),
                         category='success')
             except formencode.Invalid as errors:
                 defaults = errors.value
                 return htmlfill.render(
                     render('admin/permissions/permissions.html'),
                     defaults=defaults,
                     errors=errors.error_dict or {},
                     prefix_error=False,
                     encoding="UTF-8",
                     force_defaults=False)
             except Exception:
                 log.error(traceback.format_exc())
                 h.flash(_('Error occurred during update of permissions'),
                         category='error')
             raise HTTPFound(location=url('admin_permissions'))
         c.user = User.get_default_user()
         defaults = {'anonymous': c.user.active}
         for p in c.user.user_perms:
             if p.permission.permission_name.startswith('repository.'):
                 defaults['default_repo_perm'] = p.permission.permission_name
             if p.permission.permission_name.startswith('group.'):
                 defaults['default_group_perm'] = p.permission.permission_name
             if p.permission.permission_name.startswith('usergroup.'):
                 defaults['default_user_group_perm'] = p.permission.permission_name
             if p.permission.permission_name.startswith('hg.create.write_on_repogroup.'):
                 defaults['create_on_write'] = p.permission.permission_name
             elif p.permission.permission_name.startswith('hg.create.'):
                 defaults['default_repo_create'] = p.permission.permission_name
             if p.permission.permission_name.startswith('hg.usergroup.'):
                 defaults['default_user_group_create'] = p.permission.permission_name
             if p.permission.permission_name.startswith('hg.register.'):
                 defaults['default_register'] = p.permission.permission_name
             if p.permission.permission_name.startswith('hg.extern_activate.'):
                 defaults['default_extern_activate'] = p.permission.permission_name
             if p.permission.permission_name.startswith('hg.fork.'):
                 defaults['default_fork'] = p.permission.permission_name
         return htmlfill.render(
             render('admin/permissions/permissions.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     def permission_ips(self):
         c.active = 'ips'
         c.user = User.get_default_user()
         c.user_ip_map = UserIpMap.query() \
                         .filter(UserIpMap.user == c.user).all()
         return render('admin/permissions/permissions.html')
     def permission_perms(self):
         c.active = 'perms'
         c.user = User.get_default_user()
         c.perm_user = AuthUser(dbuser=c.user)
         return render('admin/permissions/permissions.html')

kallithea/controllers/admin/repos.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # 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 General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.admin.repos
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Repositories controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 7, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import celery.result
 import formencode
 from formencode import htmlfill
 from tg import request
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPForbidden, HTTPFound, HTTPInternalServerError, HTTPNotFound
 import kallithea
 from kallithea.config.routing import url
 from kallithea.lib import helpers as h
-from kallithea.lib.auth import HasPermissionAny, HasRepoPermissionLevelDecorator, LoginRequired, NotAnonymous
 from kallithea.lib.auth import HasRepoPermissionLevelDecorator, LoginRequired, NotAnonymous
 from kallithea.lib.base import BaseRepoController, jsonify, render
 from kallithea.lib.exceptions import AttachedForksError
 from kallithea.lib.utils import action_logger
 from kallithea.lib.utils2 import safe_int
 from kallithea.lib.vcs import RepositoryError
 from kallithea.model.db import RepoGroup, Repository, RepositoryField, Setting, UserFollowing
 from kallithea.model.forms import RepoFieldForm, RepoForm, RepoPermsForm
 from kallithea.model.meta import Session
 from kallithea.model.repo import RepoModel
 from kallithea.model.scm import AvailableRepoGroupChoices, RepoList, ScmModel
 log = logging.getLogger(__name__)
 class ReposController(BaseRepoController):
     """
     REST Controller styled on the Atom Publishing Protocol"""
     # To properly map this controller, ensure your config/routing.py
     # file has a resource setup:
     #     map.resource('repo', 'repos')
     @LoginRequired(allow_default_user=True)
     def _before(self, *args, **kwargs):
         super(ReposController, self)._before(*args, **kwargs)
     def _load_repo(self):
         repo_obj = c.db_repo
         if repo_obj is None:
             h.not_mapped_error(c.repo_name)
             raise HTTPFound(location=url('repos'))
         return repo_obj
     def __load_defaults(self, repo=None):
         if HasPermissionAny('hg.create.write_on_repogroup.true')():
             repo_group_perm_level = 'write'
         else:
             repo_group_perm_level = 'admin'
         extras = [] if repo is None else [repo.group]
-        c.repo_groups = AvailableRepoGroupChoices(repo_group_perm_level, extras)
+        c.repo_groups = AvailableRepoGroupChoices('write', extras)
         c.landing_revs_choices, c.landing_revs = ScmModel().get_repo_landing_revs(repo)
     def __load_data(self):
         """
         Load defaults settings for edit, and update
         """
         c.repo_info = self._load_repo()
         self.__load_defaults(c.repo_info)
         defaults = RepoModel()._get_defaults(c.repo_name)
         defaults['clone_uri'] = c.repo_info.clone_uri_hidden # don't show password
         defaults['permanent_url'] = c.repo_info.clone_url(clone_uri_tmpl=c.clone_uri_tmpl, with_id=True)
         return defaults
     def index(self, format='html'):
         repos_list = RepoList(Repository.query(sorted=True).all(), perm_level='admin')
         # the repo list will be filtered to only show repos where the user has read permissions
         repos_data = RepoModel().get_repos_as_dict(repos_list, admin=True)
         # data used to render the grid
         c.data = repos_data
         return render('admin/repos/repos.html')
     @NotAnonymous()
     def create(self):
         self.__load_defaults()
         try:
             # CanWriteGroup validators checks permissions of this POST
             form_result = RepoForm(repo_groups=c.repo_groups,
                                    landing_revs=c.landing_revs_choices)() \
                             .to_python(dict(request.POST))
         except formencode.Invalid as errors:
             log.info(errors)
             return htmlfill.render(
                 render('admin/repos/repo_add.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 force_defaults=False,
                 encoding="UTF-8")
         try:
             # create is done sometimes async on celery, db transaction
             # management is handled there.
             task = RepoModel().create(form_result, request.authuser.user_id)
             task_id = task.task_id
         except Exception:
             log.error(traceback.format_exc())
             msg = (_('Error creating repository %s')
                    % form_result.get('repo_name'))
             h.flash(msg, category='error')
             raise HTTPFound(location=url('home'))
         raise HTTPFound(location=h.url('repo_creating_home',
                               repo_name=form_result['repo_name_full'],
                               task_id=task_id))
     @NotAnonymous()
     def create_repository(self):
         self.__load_defaults()
         if not c.repo_groups:
             raise HTTPForbidden
         parent_group = request.GET.get('parent_group')
         ## apply the defaults from defaults page
         defaults = Setting.get_default_repo_settings(strip_prefix=True)
         if parent_group:
             prg = RepoGroup.get(parent_group)
             if prg is None or not any(rgc[0] == prg.group_id
                                       for rgc in c.repo_groups):
                 raise HTTPForbidden
             defaults.update({'repo_group': parent_group})
         return htmlfill.render(
             render('admin/repos/repo_add.html'),
             defaults=defaults,
             errors={},
             prefix_error=False,
             encoding="UTF-8",
             force_defaults=False)
     @LoginRequired()
     def repo_creating(self, repo_name):
         c.repo = repo_name
         c.task_id = request.GET.get('task_id')
         if not c.repo:
             raise HTTPNotFound()
         return render('admin/repos/repo_creating.html')
     @LoginRequired()
     @jsonify
     def repo_check(self, repo_name):
         c.repo = repo_name
         task_id = request.GET.get('task_id')
         if task_id and task_id not in ['None']:
             if kallithea.CELERY_APP:
                 task_result = celery.result.AsyncResult(task_id, app=kallithea.CELERY_APP)
                 if task_result.failed():
                     raise HTTPInternalServerError(task_result.traceback)
         repo = Repository.get_by_repo_name(repo_name)
         if repo and repo.repo_state == Repository.STATE_CREATED:
             if repo.clone_uri:
                 h.flash(_('Created repository %s from %s')
                         % (repo.repo_name, repo.clone_uri_hidden), category='success')
             else:
                 repo_url = h.link_to(repo.repo_name,
                                      h.url('summary_home',
                                            repo_name=repo.repo_name))
                 fork = repo.fork
                 if fork is not None:
                     fork_name = fork.repo_name
                     h.flash(h.HTML(_('Forked repository %s as %s'))
                             % (fork_name, repo_url), category='success')
                 else:
                     h.flash(h.HTML(_('Created repository %s')) % repo_url,
                             category='success')
             return {'result': True}
         return {'result': False}
     @HasRepoPermissionLevelDecorator('admin')
     def update(self, repo_name):
         c.repo_info = self._load_repo()
         self.__load_defaults(c.repo_info)
         c.active = 'settings'
         c.repo_fields = RepositoryField.query() \
             .filter(RepositoryField.repository == c.repo_info).all()
         repo_model = RepoModel()
         changed_name = repo_name
         repo = Repository.get_by_repo_name(repo_name)
         old_data = {
             'repo_name': repo_name,
             'repo_group': repo.group.get_dict() if repo.group else {},
             'repo_type': repo.repo_type,
+        }
         _form = RepoForm(edit=True, old_data=old_data,
                          repo_groups=c.repo_groups,
                          landing_revs=c.landing_revs_choices)()
         try:
             form_result = _form.to_python(dict(request.POST))
             repo = repo_model.update(repo_name, **form_result)
             ScmModel().mark_for_invalidation(repo_name)
             h.flash(_('Repository %s updated successfully') % repo_name,
                     category='success')
             changed_name = repo.repo_name
             action_logger(request.authuser, 'admin_updated_repo',
                 changed_name, request.ip_addr)
             Session().commit()
         except formencode.Invalid as errors:
             log.info(errors)
             defaults = self.__load_data()
             defaults.update(errors.value)
             return htmlfill.render(
                 render('admin/repos/repo_edit.html'),
                 defaults=defaults,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('Error occurred during update of repository %s')
                     % repo_name, category='error')
         raise HTTPFound(location=url('edit_repo', repo_name=changed_name))
     @HasRepoPermissionLevelDecorator('admin')
     def delete(self, repo_name):
         repo_model = RepoModel()
         repo = repo_model.get_by_repo_name(repo_name)
         if not repo:
             h.not_mapped_error(repo_name)
             raise HTTPFound(location=url('repos'))
         try:
             _forks = repo.forks.count()
             handle_forks = None
             if _forks and request.POST.get('forks'):
                 do = request.POST['forks']
                 if do == 'detach_forks':
                     handle_forks = 'detach'
                     h.flash(_('Detached %s forks') % _forks, category='success')
                 elif do == 'delete_forks':
                     handle_forks = 'delete'
                     h.flash(_('Deleted %s forks') % _forks, category='success')
             repo_model.delete(repo, forks=handle_forks)
             action_logger(request.authuser, 'admin_deleted_repo',
                 repo_name, request.ip_addr)
             ScmModel().mark_for_invalidation(repo_name)
             h.flash(_('Deleted repository %s') % repo_name, category='success')
             Session().commit()
         except AttachedForksError:
             h.flash(_('Cannot delete repository %s which still has forks')
                         % repo_name, category='warning')
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during deletion of %s') % repo_name,
                     category='error')
         if repo.group:
             raise HTTPFound(location=url('repos_group_home', group_name=repo.group.group_name))
         raise HTTPFound(location=url('repos'))
     @HasRepoPermissionLevelDecorator('admin')
     def edit(self, repo_name):
         defaults = self.__load_data()
         c.repo_fields = RepositoryField.query() \
             .filter(RepositoryField.repository == c.repo_info).all()
         c.active = 'settings'
         return htmlfill.render(
             render('admin/repos/repo_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasRepoPermissionLevelDecorator('admin')
     def edit_permissions(self, repo_name):
         c.repo_info = self._load_repo()
         c.active = 'permissions'
         defaults = RepoModel()._get_defaults(repo_name)
         return htmlfill.render(
             render('admin/repos/repo_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasRepoPermissionLevelDecorator('admin')
     def edit_permissions_update(self, repo_name):
         form = RepoPermsForm()().to_python(request.POST)
         RepoModel()._update_permissions(repo_name, form['perms_new'],
                                         form['perms_updates'])
         # TODO: implement this
         #action_logger(request.authuser, 'admin_changed_repo_permissions',
         #              repo_name, request.ip_addr)
         Session().commit()
         h.flash(_('Repository permissions updated'), category='success')
         raise HTTPFound(location=url('edit_repo_perms', repo_name=repo_name))
     @HasRepoPermissionLevelDecorator('admin')
     def edit_permissions_revoke(self, repo_name):
         try:
             obj_type = request.POST.get('obj_type')
             obj_id = None
             if obj_type == 'user':
                 obj_id = safe_int(request.POST.get('user_id'))
             elif obj_type == 'user_group':
                 obj_id = safe_int(request.POST.get('user_group_id'))
             else:
                 assert False
             if obj_type == 'user':
                 RepoModel().revoke_user_permission(repo=repo_name, user=obj_id)
             elif obj_type == 'user_group':
                 RepoModel().revoke_user_group_permission(
                     repo=repo_name, group_name=obj_id
+                )
             else:
                 assert False
             # TODO: implement this
             #action_logger(request.authuser, 'admin_revoked_repo_permissions',
             #              repo_name, request.ip_addr)
             Session().commit()
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during revoking of permission'),
                     category='error')
             raise HTTPInternalServerError()
         return []
     @HasRepoPermissionLevelDecorator('admin')
     def edit_fields(self, repo_name):
         c.repo_info = self._load_repo()
         c.repo_fields = RepositoryField.query() \
             .filter(RepositoryField.repository == c.repo_info).all()
         c.active = 'fields'
         if request.POST:
             raise HTTPFound(location=url('repo_edit_fields'))
         return render('admin/repos/repo_edit.html')
     @HasRepoPermissionLevelDecorator('admin')
     def create_repo_field(self, repo_name):
         try:
             form_result = RepoFieldForm()().to_python(dict(request.POST))
             new_field = RepositoryField()
             new_field.repository = Repository.get_by_repo_name(repo_name)
             new_field.field_key = form_result['new_field_key']
             new_field.field_type = form_result['new_field_type']  # python type
             new_field.field_value = form_result['new_field_value']  # set initial blank value
             new_field.field_desc = form_result['new_field_desc']
             new_field.field_label = form_result['new_field_label']
             Session().add(new_field)
             Session().commit()
         except formencode.Invalid as e:
             h.flash(_('Field validation error: %s') % e.msg, category='error')
         except Exception as e:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during creation of field: %r') % e, category='error')
         raise HTTPFound(location=url('edit_repo_fields', repo_name=repo_name))
     @HasRepoPermissionLevelDecorator('admin')
     def delete_repo_field(self, repo_name, field_id):
         field = RepositoryField.get_or_404(field_id)
         try:
             Session().delete(field)
             Session().commit()
         except Exception as e:
             log.error(traceback.format_exc())
             msg = _('An error occurred during removal of field')
             h.flash(msg, category='error')
         raise HTTPFound(location=url('edit_repo_fields', repo_name=repo_name))
     @HasRepoPermissionLevelDecorator('admin')
     def edit_advanced(self, repo_name):
         c.repo_info = self._load_repo()
         c.default_user_id = kallithea.DEFAULT_USER_ID
         c.in_public_journal = UserFollowing.query() \
             .filter(UserFollowing.user_id == c.default_user_id) \
             .filter(UserFollowing.follows_repository == c.repo_info).scalar()
         _repos = Repository.query(sorted=True).all()
         read_access_repos = RepoList(_repos, perm_level='read')
         c.repos_list = [(None, _('-- Not a fork --'))]
         c.repos_list += [(x.repo_id, x.repo_name)
                          for x in read_access_repos
                          if x.repo_id != c.repo_info.repo_id
                          and x.repo_type == c.repo_info.repo_type]
         defaults = {
             'id_fork_of': c.repo_info.fork_id if c.repo_info.fork_id else ''
+        }
         c.active = 'advanced'
         if request.POST:
             raise HTTPFound(location=url('repo_edit_advanced'))
         return htmlfill.render(
             render('admin/repos/repo_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasRepoPermissionLevelDecorator('admin')
     def edit_advanced_journal(self, repo_name):
         """
         Sets this repository to be visible in public journal,
         in other words asking default user to follow this repo
         :param repo_name:
         """
         try:
             repo_id = Repository.get_by_repo_name(repo_name).repo_id
             user_id = kallithea.DEFAULT_USER_ID
             self.scm_model.toggle_following_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(location=url('edit_repo_advanced', repo_name=repo_name))
     @HasRepoPermissionLevelDecorator('admin')
     def edit_advanced_fork(self, repo_name):
         """
         Mark given repository as a fork of another
         :param repo_name:
         """
         try:
             fork_id = request.POST.get('id_fork_of')
             repo = ScmModel().mark_as_fork(repo_name, fork_id,
                                            request.authuser.username)
             fork = repo.fork.repo_name if repo.fork else _('Nothing')
             Session().commit()
             h.flash(_('Marked repository %s as fork of %s') % (repo_name, fork),
                     category='success')

kallithea/controllers/forks.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # 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 General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.forks
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~
 forks controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 23, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode
 from formencode import htmlfill
 from tg import request
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound
 import kallithea
 import kallithea.lib.helpers as h
 from kallithea.config.routing import url
-from kallithea.lib.auth import HasPermissionAny, HasPermissionAnyDecorator, HasRepoPermissionLevel, HasRepoPermissionLevelDecorator, LoginRequired
 from kallithea.lib.auth import HasPermissionAnyDecorator, HasRepoPermissionLevel, HasRepoPermissionLevelDecorator, LoginRequired
 from kallithea.lib.base import BaseRepoController, render
 from kallithea.lib.page import Page
 from kallithea.lib.utils2 import safe_int
 from kallithea.model.db import Repository, Ui, UserFollowing
 from kallithea.model.forms import RepoForkForm
 from kallithea.model.repo import RepoModel
 from kallithea.model.scm import AvailableRepoGroupChoices, ScmModel
 log = logging.getLogger(__name__)
 class ForksController(BaseRepoController):
     def __load_defaults(self):
         if HasPermissionAny('hg.create.write_on_repogroup.true')():
             repo_group_perm_level = 'write'
         else:
             repo_group_perm_level = 'admin'
         c.repo_groups = AvailableRepoGroupChoices(repo_group_perm_level)
         c.repo_groups = AvailableRepoGroupChoices('write')
         c.landing_revs_choices, c.landing_revs = ScmModel().get_repo_landing_revs()
         c.can_update = Ui.get_by_key('hooks', Ui.HOOK_UPDATE).ui_active
     def __load_data(self):
         """
         Load defaults settings for edit, and update
         """
         self.__load_defaults()
         c.repo_info = c.db_repo
         repo = c.db_repo.scm_instance
         if c.repo_info is None:
             h.not_mapped_error(c.repo_name)
             raise HTTPFound(location=url('repos'))
         c.default_user_id = kallithea.DEFAULT_USER_ID
         c.in_public_journal = UserFollowing.query() \
             .filter(UserFollowing.user_id == c.default_user_id) \
             .filter(UserFollowing.follows_repository == c.repo_info).scalar()
         if c.repo_info.stats:
             last_rev = c.repo_info.stats.stat_on_revision + 1
         else:
             last_rev = 0
         c.stats_revision = last_rev
         c.repo_last_rev = repo.count() if repo.revisions else 0
         if last_rev == 0 or c.repo_last_rev == 0:
             c.stats_percentage = 0
         else:
             c.stats_percentage = '%.2f' % ((float((last_rev)) /
                                             c.repo_last_rev) * 100)
         defaults = RepoModel()._get_defaults(c.repo_name)
         # alter the description to indicate a fork
         defaults['description'] = ('fork of repository: %s \n%s'
                                    % (defaults['repo_name'],
                                       defaults['description']))
         # add suffix to fork
         defaults['repo_name'] = '%s-fork' % defaults['repo_name']
         return defaults
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def forks(self, repo_name):
         p = safe_int(request.GET.get('page'), 1)
         repo_id = c.db_repo.repo_id
         d = []
         for r in Repository.get_repo_forks(repo_id):
             if not HasRepoPermissionLevel('read')(r.repo_name, 'get forks check'):
                 continue
             d.append(r)
         c.forks_pager = Page(d, page=p, items_per_page=20)
         if request.environ.get('HTTP_X_PARTIAL_XHR'):
             return render('/forks/forks_data.html')
         return render('/forks/forks.html')
     @LoginRequired()
     @HasPermissionAnyDecorator('hg.admin', 'hg.fork.repository')
     @HasRepoPermissionLevelDecorator('read')
     def fork(self, repo_name):
         c.repo_info = Repository.get_by_repo_name(repo_name)
         if not c.repo_info:
             h.not_mapped_error(repo_name)
             raise HTTPFound(location=url('home'))
         defaults = self.__load_data()
         return htmlfill.render(
             render('forks/fork.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @LoginRequired()
     @HasPermissionAnyDecorator('hg.admin', 'hg.fork.repository')
     @HasRepoPermissionLevelDecorator('read')
     def fork_create(self, repo_name):
         self.__load_defaults()
         c.repo_info = Repository.get_by_repo_name(repo_name)
         _form = RepoForkForm(old_data={'repo_type': c.repo_info.repo_type},
                              repo_groups=c.repo_groups,
                              landing_revs=c.landing_revs_choices)()
         form_result = {}
         task_id = None
         try:
             form_result = _form.to_python(dict(request.POST))
             # an approximation that is better than nothing
             if not Ui.get_by_key('hooks', Ui.HOOK_UPDATE).ui_active:
                 form_result['update_after_clone'] = False
             # create fork is done sometimes async on celery, db transaction
             # management is handled there.
             task = RepoModel().create_fork(form_result, request.authuser.user_id)
             task_id = task.task_id
         except formencode.Invalid as errors:
             return htmlfill.render(
                 render('forks/fork.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during repository forking %s') %
                     repo_name, category='error')
         raise HTTPFound(location=h.url('repo_creating_home',
                               repo_name=form_result['repo_name_full'],
                               task_id=task_id))

kallithea/lib/auth.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # 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 General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.lib.auth
 ~~~~~~~~~~~~~~~~~~
 authentication and permission libraries
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 4, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import hashlib
 import itertools
 import logging
 import os
 import string
 import bcrypt
 import ipaddr
 from decorator import decorator
 from sqlalchemy.orm import joinedload
 from sqlalchemy.orm.exc import ObjectDeletedError
 from tg import request
 from tg.i18n import ugettext as _
 from webob.exc import HTTPForbidden, HTTPFound
 import kallithea
 from kallithea.config.routing import url
 from kallithea.lib.utils import get_repo_group_slug, get_repo_slug, get_user_group_slug
 from kallithea.lib.utils2 import ascii_bytes, ascii_str, safe_bytes
 from kallithea.lib.vcs.utils.lazy import LazyProperty
 from kallithea.model.db import (Permission, UserApiKeys, UserGroup, UserGroupMember, UserGroupRepoGroupToPerm, UserGroupRepoToPerm, UserGroupToPerm,
                                 UserGroupUserGroupToPerm, UserIpMap, UserToPerm)
 from kallithea.model.meta import Session
 from kallithea.model.user import UserModel
 log = logging.getLogger(__name__)
 class PasswordGenerator(object):
     """
     This is a simple class for generating password from different sets of
     characters
     usage::
         passwd_gen = PasswordGenerator()
         #print 8-letter password containing only big and small letters
             of alphabet
         passwd_gen.gen_password(8, passwd_gen.ALPHABETS_BIG_SMALL)
     """
     ALPHABETS_NUM = r'''1234567890'''
     ALPHABETS_SMALL = r'''qwertyuiopasdfghjklzxcvbnm'''
     ALPHABETS_BIG = r'''QWERTYUIOPASDFGHJKLZXCVBNM'''
     ALPHABETS_SPECIAL = r'''`-=[]\;',./~!@#$%^&*()_+{}|:"<>?'''
     ALPHABETS_FULL = ALPHABETS_BIG + ALPHABETS_SMALL \
         + ALPHABETS_NUM + ALPHABETS_SPECIAL
     ALPHABETS_ALPHANUM = ALPHABETS_BIG + ALPHABETS_SMALL + ALPHABETS_NUM
     ALPHABETS_BIG_SMALL = ALPHABETS_BIG + ALPHABETS_SMALL
     ALPHABETS_ALPHANUM_BIG = ALPHABETS_BIG + ALPHABETS_NUM
     ALPHABETS_ALPHANUM_SMALL = ALPHABETS_SMALL + ALPHABETS_NUM
     def gen_password(self, length, alphabet=ALPHABETS_FULL):
         assert len(alphabet) <= 256, alphabet
         l = []
         while len(l) < length:
             i = ord(os.urandom(1))
             if i < len(alphabet):
                 l.append(alphabet[i])
         return ''.join(l)
 def get_crypt_password(password):
     """
     Cryptographic function used for bcrypt password hashing.
     :param password: password to hash
     """
     return ascii_str(bcrypt.hashpw(safe_bytes(password), bcrypt.gensalt(10)))
 def check_password(password, hashed):
     """
     Checks password match the hashed value using bcrypt.
     Remains backwards compatible and accept plain sha256 hashes which used to
     be used on Windows.
     :param password: password
     :param hashed: password in hashed form
     """
     # sha256 hashes will always be 64 hex chars
     # bcrypt hashes will always contain $ (and be shorter)
     if len(hashed) == 64 and all(x in string.hexdigits for x in hashed):
         return hashlib.sha256(password).hexdigest() == hashed
     try:
         return bcrypt.checkpw(safe_bytes(password), ascii_bytes(hashed))
     except ValueError as e:
         # bcrypt will throw ValueError 'Invalid hashed_password salt' on all password errors
         log.error('error from bcrypt checking password: %s', e)
         return False
     log.error('check_password failed - no method found for hash length %s', len(hashed))
     return False
 def _cached_perms_data(user_id, user_is_admin):
     RK = 'repositories'
     GK = 'repositories_groups'
     UK = 'user_groups'
     GLOBAL = 'global'
     PERM_WEIGHTS = Permission.PERM_WEIGHTS
     permissions = {RK: {}, GK: {}, UK: {}, GLOBAL: set()}
     def bump_permission(kind, key, new_perm):
         """Add a new permission for kind and key.
         Assuming the permissions are comparable, set the new permission if it
         has higher weight, else drop it and keep the old permission.
         """
         cur_perm = permissions[kind][key]
         new_perm_val = PERM_WEIGHTS[new_perm]
         cur_perm_val = PERM_WEIGHTS[cur_perm]
         if new_perm_val > cur_perm_val:
             permissions[kind][key] = new_perm
     #======================================================================
     # fetch default permissions
     #======================================================================
     default_repo_perms = Permission.get_default_perms(kallithea.DEFAULT_USER_ID)
     default_repo_groups_perms = Permission.get_default_group_perms(kallithea.DEFAULT_USER_ID)
     default_user_group_perms = Permission.get_default_user_group_perms(kallithea.DEFAULT_USER_ID)
     if user_is_admin:
         #==================================================================
         # admin users have all rights;
         # based on default permissions, just set everything to admin
         #==================================================================
         permissions[GLOBAL].add('hg.admin')
         permissions[GLOBAL].add('hg.create.write_on_repogroup.true')
         # repositories
         for perm in default_repo_perms:
             r_k = perm.repository.repo_name
             p = 'repository.admin'
             permissions[RK][r_k] = p
         # repository groups
         for perm in default_repo_groups_perms:
             rg_k = perm.group.group_name
             p = 'group.admin'
             permissions[GK][rg_k] = p
         # user groups
         for perm in default_user_group_perms:
             u_k = perm.user_group.users_group_name
             p = 'usergroup.admin'
             permissions[UK][u_k] = p
         return permissions
     #==================================================================
     # SET DEFAULTS GLOBAL, REPOS, REPOSITORY GROUPS
     #==================================================================
     # default global permissions taken from the default user
     default_global_perms = UserToPerm.query() \
         .filter(UserToPerm.user_id == kallithea.DEFAULT_USER_ID) \
         .options(joinedload(UserToPerm.permission))
     for perm in default_global_perms:
         permissions[GLOBAL].add(perm.permission.permission_name)
     # defaults for repositories, taken from default user
     for perm in default_repo_perms:
         r_k = perm.repository.repo_name
         if perm.repository.owner_id == user_id:
             p = 'repository.admin'
         elif perm.repository.private:
             p = 'repository.none'
         else:
             p = perm.permission.permission_name
         permissions[RK][r_k] = p
     # defaults for repository groups taken from default user permission
     # on given group
     for perm in default_repo_groups_perms:
         rg_k = perm.group.group_name
         p = perm.permission.permission_name
         permissions[GK][rg_k] = p
     # defaults for user groups taken from default user permission
     # on given user group
     for perm in default_user_group_perms:
         u_k = perm.user_group.users_group_name
         p = perm.permission.permission_name
         permissions[UK][u_k] = p
     #======================================================================
     # !! Augment GLOBALS with user permissions if any found !!
     #======================================================================
     # USER GROUPS comes first
     # user group global permissions
     user_perms_from_users_groups = Session().query(UserGroupToPerm) \
         .options(joinedload(UserGroupToPerm.permission)) \
         .join((UserGroupMember, UserGroupToPerm.users_group_id ==
                UserGroupMember.users_group_id)) \
         .filter(UserGroupMember.user_id == user_id) \
         .join((UserGroup, UserGroupMember.users_group_id ==
                UserGroup.users_group_id)) \
         .filter(UserGroup.users_group_active == True) \
         .order_by(UserGroupToPerm.users_group_id) \
         .all()
     # need to group here by groups since user can be in more than
     # one group
     _grouped = [[x, list(y)] for x, y in
                 itertools.groupby(user_perms_from_users_groups,
                                   lambda x:x.users_group)]
     for gr, perms in _grouped:
         for perm in perms:
             permissions[GLOBAL].add(perm.permission.permission_name)
     # user specific global permissions
     user_perms = Session().query(UserToPerm) \
             .options(joinedload(UserToPerm.permission)) \
             .filter(UserToPerm.user_id == user_id).all()
     for perm in user_perms:
         permissions[GLOBAL].add(perm.permission.permission_name)
     # for each kind of global permissions, only keep the one with heighest weight
     kind_max_perm = {}
-    for perm in sorted(permissions[GLOBAL], key=lambda n: PERM_WEIGHTS[n]):
+    for perm in sorted(permissions[GLOBAL], key=lambda n: PERM_WEIGHTS.get(n, -1)):
         kind = perm.rsplit('.', 1)[0]
         kind_max_perm[kind] = perm
     permissions[GLOBAL] = set(kind_max_perm.values())
     ## END GLOBAL PERMISSIONS
     #======================================================================
     # !! PERMISSIONS FOR REPOSITORIES !!
     #======================================================================
     #======================================================================
     # check if user is part of user groups for this repository and
     # fill in his permission from it.
     #======================================================================
     # user group for repositories permissions
     user_repo_perms_from_users_groups = \
      Session().query(UserGroupRepoToPerm) \
         .join((UserGroup, UserGroupRepoToPerm.users_group_id ==
                UserGroup.users_group_id)) \
         .filter(UserGroup.users_group_active == True) \
         .join((UserGroupMember, UserGroupRepoToPerm.users_group_id ==
                UserGroupMember.users_group_id)) \
         .filter(UserGroupMember.user_id == user_id) \
         .options(joinedload(UserGroupRepoToPerm.repository)) \
         .options(joinedload(UserGroupRepoToPerm.permission)) \
         .all()
     for perm in user_repo_perms_from_users_groups:
         bump_permission(RK,
             perm.repository.repo_name,
             perm.permission.permission_name)
     # user permissions for repositories
     user_repo_perms = Permission.get_default_perms(user_id)
     for perm in user_repo_perms:
         bump_permission(RK,
             perm.repository.repo_name,
             perm.permission.permission_name)
     #======================================================================
     # !! PERMISSIONS FOR REPOSITORY GROUPS !!
     #======================================================================
     #======================================================================
     # check if user is part of user groups for this repository groups and
     # fill in his permission from it.
     #======================================================================
     # user group for repo groups permissions
     user_repo_group_perms_from_users_groups = \
      Session().query(UserGroupRepoGroupToPerm) \
      .join((UserGroup, UserGroupRepoGroupToPerm.users_group_id ==
             UserGroup.users_group_id)) \
      .filter(UserGroup.users_group_active == True) \
      .join((UserGroupMember, UserGroupRepoGroupToPerm.users_group_id
             == UserGroupMember.users_group_id)) \
      .filter(UserGroupMember.user_id == user_id) \
      .options(joinedload(UserGroupRepoGroupToPerm.permission)) \
      .all()
     for perm in user_repo_group_perms_from_users_groups:
         bump_permission(GK,
             perm.group.group_name,
             perm.permission.permission_name)
     # user explicit permissions for repository groups
     user_repo_groups_perms = Permission.get_default_group_perms(user_id)
     for perm in user_repo_groups_perms:
         bump_permission(GK,
             perm.group.group_name,
             perm.permission.permission_name)
     #======================================================================
     # !! PERMISSIONS FOR USER GROUPS !!
     #======================================================================
     # user group for user group permissions
     user_group_user_groups_perms = \
      Session().query(UserGroupUserGroupToPerm) \
      .join((UserGroup, UserGroupUserGroupToPerm.target_user_group_id
             == UserGroup.users_group_id)) \
      .join((UserGroupMember, UserGroupUserGroupToPerm.user_group_id
             == UserGroupMember.users_group_id)) \
      .filter(UserGroupMember.user_id == user_id) \
      .join((UserGroup, UserGroupMember.users_group_id ==
             UserGroup.users_group_id), aliased=True, from_joinpoint=True) \
      .filter(UserGroup.users_group_active == True) \
      .options(joinedload(UserGroupUserGroupToPerm.permission)) \
      .all()
     for perm in user_group_user_groups_perms:
         bump_permission(UK,
             perm.target_user_group.users_group_name,
             perm.permission.permission_name)
     # user explicit permission for user groups
     user_user_groups_perms = Permission.get_default_user_group_perms(user_id)
     for perm in user_user_groups_perms:
         bump_permission(UK,
             perm.user_group.users_group_name,
             perm.permission.permission_name)
     return permissions
 class AuthUser(object):
     """
     Represents a Kallithea user, including various authentication and
     authorization information. Typically used to store the current user,
     but is also used as a generic user information data structure in
     parts of the code, e.g. user management.
     Constructed from a database `User` object, a user ID or cookie dict,
     it looks up the user (if needed) and copies all attributes to itself,
     adding various non-persistent data. If lookup fails but anonymous
     access to Kallithea is enabled, the default user is loaded instead.
     `AuthUser` does not by itself authenticate users. It's up to other parts of
     the code to check e.g. if a supplied password is correct, and if so, trust
     the AuthUser object as an authenticated user.
     However, `AuthUser` does refuse to load a user that is not `active`.
     Note that Kallithea distinguishes between the default user (an actual
     user in the database with username "default") and "no user" (no actual
     User object, AuthUser filled with blank values and username "None").
     If the default user is active, that will always be used instead of
     "no user". On the other hand, if the default user is disabled (and
     there is no login information), we instead get "no user"; this should
     only happen on the login page (as all other requests are redirected).
     `is_default_user` specifically checks if the AuthUser is the user named
     "default". Use `is_anonymous` to check for both "default" and "no user".
     """
     @classmethod
     def make(cls, dbuser=None, is_external_auth=False, ip_addr=None):
         """Create an AuthUser to be authenticated ... or return None if user for some reason can't be authenticated.
         Checks that a non-None dbuser is provided, is active, and that the IP address is ok.
         """
         assert ip_addr is not None
         if dbuser is None:
             log.info('No db user for authentication')
             return None
         if not dbuser.active:
             log.info('Db user %s not active', dbuser.username)
             return None
         allowed_ips = AuthUser.get_allowed_ips(dbuser.user_id)
         if not check_ip_access(source_ip=ip_addr, allowed_ips=allowed_ips):
             log.info('Access for %s from %s forbidden - not in %s', dbuser.username, ip_addr, allowed_ips)
             return None
         return cls(dbuser=dbuser, is_external_auth=is_external_auth)
     def __init__(self, user_id=None, dbuser=None, is_external_auth=False):
         self.is_external_auth = is_external_auth # container auth - don't show logout option
         # These attributes will be overridden below if the requested user is
         # found or anonymous access (using the default user) is enabled.
         self.user_id = None
         self.username = None
         self.api_key = None
         self.name = ''
         self.lastname = ''
         self.email = ''
         self.admin = False
         # Look up database user, if necessary.
         if user_id is not None:
             assert dbuser is None
             log.debug('Auth User lookup by USER ID %s', user_id)
             dbuser = UserModel().get(user_id)
             assert dbuser is not None
         else:
             assert dbuser is not None
             log.debug('Auth User lookup by database user %s', dbuser)
         log.debug('filling %s data', dbuser)
         self.is_anonymous = dbuser.is_default_user
         if dbuser.is_default_user and not dbuser.active:
             self.username = 'None'
             self.is_default_user = False
         else:
             # copy non-confidential database fields from a `db.User` to this `AuthUser`.
             for k, v in dbuser.get_dict().items():
                 assert k not in ['api_keys', 'permissions']
                 setattr(self, k, v)
             self.is_default_user = dbuser.is_default_user
         log.debug('Auth User is now %s', self)
     @LazyProperty
     def permissions(self):
         """
         Fills user permission attribute with permissions taken from database
         works for permissions given for repositories, and for permissions that
         are granted to groups
         :param user: `AuthUser` instance
         """
         log.debug('Getting PERMISSION tree for %s', self)
         return _cached_perms_data(self.user_id, self.is_admin)
     def has_repository_permission_level(self, repo_name, level, purpose=None):
         required_perms = {
             'read': ['repository.read', 'repository.write', 'repository.admin'],
             'write': ['repository.write', 'repository.admin'],
             'admin': ['repository.admin'],
         }[level]
         actual_perm = self.permissions['repositories'].get(repo_name)
         ok = actual_perm in required_perms
         log.debug('Checking if user %r can %r repo %r (%s): %s (has %r)',
             self.username, level, repo_name, purpose, ok, actual_perm)
         return ok
     def has_repository_group_permission_level(self, repo_group_name, level, purpose=None):
         required_perms = {
             'read': ['group.read', 'group.write', 'group.admin'],
             'write': ['group.write', 'group.admin'],
             'admin': ['group.admin'],
         }[level]
         actual_perm = self.permissions['repositories_groups'].get(repo_group_name)
         ok = actual_perm in required_perms
         log.debug('Checking if user %r can %r repo group %r (%s): %s (has %r)',
             self.username, level, repo_group_name, purpose, ok, actual_perm)
         return ok
     def has_user_group_permission_level(self, user_group_name, level, purpose=None):
         required_perms = {
             'read': ['usergroup.read', 'usergroup.write', 'usergroup.admin'],
             'write': ['usergroup.write', 'usergroup.admin'],
             'admin': ['usergroup.admin'],
         }[level]
         actual_perm = self.permissions['user_groups'].get(user_group_name)
         ok = actual_perm in required_perms
         log.debug('Checking if user %r can %r user group %r (%s): %s (has %r)',
             self.username, level, user_group_name, purpose, ok, actual_perm)
         return ok
     @property
     def api_keys(self):
         return self._get_api_keys()
     def _get_api_keys(self):
         api_keys = [self.api_key]
         for api_key in UserApiKeys.query() \
                 .filter_by(user_id=self.user_id, is_expired=False):
             api_keys.append(api_key.api_key)
         return api_keys
     @property
     def is_admin(self):
         return self.admin
     @property
     def repositories_admin(self):
         """
         Returns list of repositories you're an admin of
         """
         return [x[0] for x in self.permissions['repositories'].items()
                 if x[1] == 'repository.admin']
     @property
     def repository_groups_admin(self):
         """
         Returns list of repository groups you're an admin of
         """
         return [x[0] for x in self.permissions['repositories_groups'].items()
                 if x[1] == 'group.admin']
     @property
     def user_groups_admin(self):
         """
         Returns list of user groups you're an admin of
         """
         return [x[0] for x in self.permissions['user_groups'].items()
                 if x[1] == 'usergroup.admin']
     def __repr__(self):
         return "<%s %s: %r>" % (self.__class__.__name__, self.user_id, self.username)
     def to_cookie(self):
         """ Serializes this login session to a cookie `dict`. """
         return {
             'user_id': self.user_id,
             'is_external_auth': self.is_external_auth,
+        }
     @staticmethod
     def from_cookie(cookie, ip_addr):
         """
         Deserializes an `AuthUser` from a cookie `dict` ... or return None.
         """
         return AuthUser.make(
             dbuser=UserModel().get(cookie.get('user_id')),
             is_external_auth=cookie.get('is_external_auth', False),
             ip_addr=ip_addr,
+        )
     @classmethod
     def get_allowed_ips(cls, user_id):
         _set = set()
         default_ips = UserIpMap.query().filter(UserIpMap.user_id == kallithea.DEFAULT_USER_ID)
         for ip in default_ips:
             try:
                 _set.add(ip.ip_addr)
             except ObjectDeletedError:
                 # since we use heavy caching sometimes it happens that we get
                 # deleted objects here, we just skip them
                 pass
         user_ips = UserIpMap.query().filter(UserIpMap.user_id == user_id)
         for ip in user_ips:
             try:
                 _set.add(ip.ip_addr)
             except ObjectDeletedError:
                 # since we use heavy caching sometimes it happens that we get
                 # deleted objects here, we just skip them
                 pass
         return _set or set(['0.0.0.0/0', '::/0'])
 #==============================================================================
 # CHECK DECORATORS
 #==============================================================================
 def _redirect_to_login(message=None):
     """Return an exception that must be raised. It will redirect to the login
     page which will redirect back to the current URL after authentication.
     The optional message will be shown in a flash message."""
     from kallithea.lib import helpers as h
     if message:
         h.flash(message, category='warning')
     p = request.path_qs
     log.debug('Redirecting to login page, origin: %s', p)
     return HTTPFound(location=url('login_home', came_from=p))
 # Use as decorator
 class LoginRequired(object):
     """Client must be logged in as a valid User, or we'll redirect to the login
     page.
     If the "default" user is enabled and allow_default_user is true, that is
     considered valid too.
     Also checks that IP address is allowed.
     """
     def __init__(self, allow_default_user=False):
         self.allow_default_user = allow_default_user
     def __call__(self, func):
         return decorator(self.__wrapper, func)
     def __wrapper(self, func, *fargs, **fkwargs):
         controller = fargs[0]
         user = request.authuser
         loc = "%s:%s" % (controller.__class__.__name__, func.__name__)
         log.debug('Checking access for user %s @ %s', user, loc)
         # regular user authentication
         if user.is_default_user:
             if self.allow_default_user:
                 log.info('default user @ %s', loc)
                 return func(*fargs, **fkwargs)
             log.info('default user is not accepted here @ %s', loc)
         elif user.is_anonymous: # default user is disabled and no proper authentication
             log.warning('user is anonymous and NOT authenticated with regular auth @ %s', loc)
         else: # regular authentication
             log.info('user %s authenticated with regular auth @ %s', user, loc)
             return func(*fargs, **fkwargs)
         raise _redirect_to_login()
 # Use as decorator
 class NotAnonymous(object):
     """Ensures that client is not logged in as the "default" user, and
     redirects to the login page otherwise. Must be used together with
     LoginRequired."""
     def __call__(self, func):
         return decorator(self.__wrapper, func)
     def __wrapper(self, func, *fargs, **fkwargs):
         cls = fargs[0]
         user = request.authuser

kallithea/model/db.py

➞

Show inline comments

@@ @@ -1181,825 +1181,818 @@ class Repository(Base, BaseDbModel): @@
         if clone_uri:
             import urlobject
             url_obj = urlobject.URLObject(self.clone_uri)
             if url_obj.password:
                 clone_uri = url_obj.with_password('*****')
         return clone_uri
     def clone_url(self, clone_uri_tmpl, with_id=False, username=None):
         if '{repo}' not in clone_uri_tmpl and '_{repoid}' not in clone_uri_tmpl:
             log.error("Configured clone_uri_tmpl %r has no '{repo}' or '_{repoid}' and cannot toggle to use repo id URLs", clone_uri_tmpl)
         elif with_id:
             clone_uri_tmpl = clone_uri_tmpl.replace('{repo}', '_{repoid}')
         else:
             clone_uri_tmpl = clone_uri_tmpl.replace('_{repoid}', '{repo}')
         import kallithea.lib.helpers as h
         prefix_url = h.canonical_url('home')
         return get_clone_url(clone_uri_tmpl=clone_uri_tmpl,
                              prefix_url=prefix_url,
                              repo_name=self.repo_name,
                              repo_id=self.repo_id,
                              username=username)
     def set_state(self, state):
         self.repo_state = state
     #==========================================================================
     # SCM PROPERTIES
     #==========================================================================
     def get_changeset(self, rev=None):
         return get_changeset_safe(self.scm_instance, rev)
     def get_landing_changeset(self):
         """
         Returns landing changeset, or if that doesn't exist returns the tip
         """
         _rev_type, _rev = self.landing_rev
         cs = self.get_changeset(_rev)
         if isinstance(cs, EmptyChangeset):
             return self.get_changeset()
         return cs
     def update_changeset_cache(self, cs_cache=None):
         """
         Update cache of last changeset for repository, keys should be::
             short_id
             raw_id
             revision
             message
             date
             author
         :param cs_cache:
         """
         from kallithea.lib.vcs.backends.base import BaseChangeset
         if cs_cache is None:
             cs_cache = EmptyChangeset()
             # use no-cache version here
             scm_repo = self.scm_instance_no_cache()
             if scm_repo:
                 cs_cache = scm_repo.get_changeset()
         if isinstance(cs_cache, BaseChangeset):
             cs_cache = cs_cache.__json__()
         if (not self.changeset_cache or cs_cache['raw_id'] != self.changeset_cache['raw_id']):
             _default = datetime.datetime.fromtimestamp(0)
             last_change = cs_cache.get('date') or _default
             log.debug('updated repo %s with new cs cache %s',
                       self.repo_name, cs_cache)
             self.updated_on = last_change
             self.changeset_cache = cs_cache
             Session().commit()
         else:
             log.debug('changeset_cache for %s already up to date with %s',
                       self.repo_name, cs_cache['raw_id'])
     @property
     def tip(self):
         return self.get_changeset('tip')
     @property
     def author(self):
         return self.tip.author
     @property
     def last_change(self):
         return self.scm_instance.last_change
     def get_comments(self, revisions=None):
         """
         Returns comments for this repository grouped by revisions
         :param revisions: filter query by revisions only
         """
         cmts = ChangesetComment.query() \
             .filter(ChangesetComment.repo == self)
         if revisions is not None:
             if not revisions:
                 return {} # don't use sql 'in' on empty set
             cmts = cmts.filter(ChangesetComment.revision.in_(revisions))
         grouped = collections.defaultdict(list)
         for cmt in cmts.all():
             grouped[cmt.revision].append(cmt)
         return grouped
     def statuses(self, revisions):
         """
         Returns statuses for this repository.
         PRs without any votes do _not_ show up as unreviewed.
         :param revisions: list of revisions to get statuses for
         """
         if not revisions:
             return {}
         statuses = ChangesetStatus.query() \
             .filter(ChangesetStatus.repo == self) \
             .filter(ChangesetStatus.version == 0) \
             .filter(ChangesetStatus.revision.in_(revisions))
         grouped = {}
         for stat in statuses.all():
             pr_id = pr_nice_id = pr_repo = None
             if stat.pull_request:
                 pr_id = stat.pull_request.pull_request_id
                 pr_nice_id = PullRequest.make_nice_id(pr_id)
                 pr_repo = stat.pull_request.other_repo.repo_name
             grouped[stat.revision] = [str(stat.status), stat.status_lbl,
                                       pr_id, pr_repo, pr_nice_id,
                                       stat.author]
         return grouped
     def _repo_size(self):
         from kallithea.lib import helpers as h
         log.debug('calculating repository size...')
         return h.format_byte_size(self.scm_instance.size)
     #==========================================================================
     # SCM CACHE INSTANCE
     #==========================================================================
     def set_invalidate(self):
         """
         Flush SA session caches of instances of on disk repo.
         """
         try:
             del self._scm_instance
         except AttributeError:
             pass
     _scm_instance = None  # caching inside lifetime of SA session
     @property
     def scm_instance(self):
         if self._scm_instance is None:
             return self.scm_instance_no_cache()  # will populate self._scm_instance
         return self._scm_instance
     def scm_instance_no_cache(self):
         repo_full_path = self.repo_full_path
         alias = get_scm(repo_full_path)[0]
         log.debug('Creating instance of %s repository from %s',
                   alias, self.repo_full_path)
         backend = get_backend(alias)
         if alias == 'hg':
             self._scm_instance = backend(repo_full_path, create=False, baseui=self._ui)
         else:
             self._scm_instance = backend(repo_full_path, create=False)
         return self._scm_instance
     def __json__(self):
         return dict(
             repo_id=self.repo_id,
             repo_name=self.repo_name,
             landing_rev=self.landing_rev,
+        )
 class RepoGroup(Base, BaseDbModel):
     __tablename__ = 'groups'
     __table_args__ = (
         _table_args_default_dict,
+    )
     SEP = ' &raquo; '
     group_id = Column(Integer(), primary_key=True)
     group_name = Column(Unicode(255), nullable=False, unique=True) # full path
     parent_group_id = Column('group_parent_id', Integer(), ForeignKey('groups.group_id'), nullable=True)
     group_description = Column(Unicode(10000), nullable=False)
     owner_id = Column('user_id', Integer(), ForeignKey('users.user_id'), nullable=False)
     created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     repo_group_to_perm = relationship('UserRepoGroupToPerm', cascade='all', order_by='UserRepoGroupToPerm.group_to_perm_id')
     users_group_to_perm = relationship('UserGroupRepoGroupToPerm', cascade='all')
     parent_group = relationship('RepoGroup', remote_side=group_id)
     owner = relationship('User')
     @classmethod
     def query(cls, sorted=False):
         """Add RepoGroup-specific helpers for common query constructs.
         sorted: if True, apply the default ordering (name, case insensitive).
         """
         q = super(RepoGroup, cls).query()
         if sorted:
             q = q.order_by(sqlalchemy.func.lower(RepoGroup.group_name))
         return q
     def __init__(self, group_name='', parent_group=None):
         self.group_name = group_name
         self.parent_group = parent_group
     def __repr__(self):
         return "<%s %s: %s>" % (self.__class__.__name__,
                                 self.group_id, self.group_name)
     @classmethod
     def _generate_choice(cls, repo_group):
         """Return tuple with group_id and name as html literal"""
         from webhelpers2.html import literal
         if repo_group is None:
             return (-1, '-- %s --' % _('top level'))
         return repo_group.group_id, literal(cls.SEP.join(repo_group.full_path_splitted))
     @classmethod
     def groups_choices(cls, groups):
         """Return tuples with group_id and name as html literal."""
         return sorted((cls._generate_choice(g) for g in groups),
                       key=lambda c: c[1].split(cls.SEP))
     @classmethod
     def guess_instance(cls, value):
         return super(RepoGroup, cls).guess_instance(value, RepoGroup.get_by_group_name)
     @classmethod
     def get_by_group_name(cls, group_name, case_insensitive=False):
         group_name = group_name.rstrip('/')
         if case_insensitive:
             gr = cls.query() \
                 .filter(sqlalchemy.func.lower(cls.group_name) == sqlalchemy.func.lower(group_name))
         else:
             gr = cls.query() \
                 .filter(cls.group_name == group_name)
         return gr.scalar()
     @property
     def parents(self):
         groups = []
         group = self.parent_group
         while group is not None:
             groups.append(group)
             group = group.parent_group
             assert group not in groups, group # avoid recursion on bad db content
         groups.reverse()
         return groups
     @property
     def children(self):
         return RepoGroup.query().filter(RepoGroup.parent_group == self)
     @property
     def name(self):
         return self.group_name.split(URL_SEP)[-1]
     @property
     def full_path(self):
         return self.group_name
     @property
     def full_path_splitted(self):
         return self.group_name.split(URL_SEP)
     @property
     def repositories(self):
         return Repository.query(sorted=True).filter_by(group=self)
     @property
     def repositories_recursive_count(self):
         cnt = self.repositories.count()
         def children_count(group):
             cnt = 0
             for child in group.children:
                 cnt += child.repositories.count()
                 cnt += children_count(child)
             return cnt
         return cnt + children_count(self)
     def _recursive_objects(self, include_repos=True):
         all_ = []
         def _get_members(root_gr):
             if include_repos:
                 for r in root_gr.repositories:
                     all_.append(r)
             childs = root_gr.children.all()
             if childs:
                 for gr in childs:
                     all_.append(gr)
                     _get_members(gr)
         _get_members(self)
         return [self] + all_
     def recursive_groups_and_repos(self):
         """
         Recursive return all groups, with repositories in those groups
         """
         return self._recursive_objects()
     def recursive_groups(self):
         """
         Returns all children groups for this group including children of children
         """
         return self._recursive_objects(include_repos=False)
     def get_new_name(self, group_name):
         """
         returns new full group name based on parent and new name
         :param group_name:
         """
         path_prefix = (self.parent_group.full_path_splitted if
                        self.parent_group else [])
         return URL_SEP.join(path_prefix + [group_name])
     def get_api_data(self):
         """
         Common function for generating api data
         """
         group = self
         data = dict(
             group_id=group.group_id,
             group_name=group.group_name,
             group_description=group.group_description,
             parent_group=group.parent_group.group_name if group.parent_group else None,
             repositories=[x.repo_name for x in group.repositories],
             owner=group.owner.username
+        )
         return data
 class Permission(Base, BaseDbModel):
     __tablename__ = 'permissions'
     __table_args__ = (
         Index('p_perm_name_idx', 'permission_name'),
         _table_args_default_dict,
+    )
     PERMS = (
         ('hg.admin', _('Kallithea Administrator')),
         ('repository.none', _('Default user has no access to new repositories')),
         ('repository.read', _('Default user has read access to new repositories')),
         ('repository.write', _('Default user has write access to new repositories')),
         ('repository.admin', _('Default user has admin access to new repositories')),
         ('group.none', _('Default user has no access to new repository groups')),
         ('group.read', _('Default user has read access to new repository groups')),
         ('group.write', _('Default user has write access to new repository groups')),
         ('group.admin', _('Default user has admin access to new repository groups')),
         ('usergroup.none', _('Default user has no access to new user groups')),
         ('usergroup.read', _('Default user has read access to new user groups')),
         ('usergroup.write', _('Default user has write access to new user groups')),
         ('usergroup.admin', _('Default user has admin access to new user groups')),
         ('hg.usergroup.create.false', _('Only admins can create user groups')),
         ('hg.usergroup.create.true', _('Non-admins can create user groups')),
         ('hg.create.none', _('Only admins can create top level repositories')),
         ('hg.create.repository', _('Non-admins can create top level repositories')),
         ('hg.create.write_on_repogroup.true', _('Repository creation enabled with write permission to a repository group')),
         ('hg.create.write_on_repogroup.false', _('Repository creation disabled with write permission to a repository group')),
         ('hg.fork.none', _('Only admins can fork repositories')),
         ('hg.fork.repository', _('Non-admins can fork repositories')),
         ('hg.register.none', _('Registration disabled')),
         ('hg.register.manual_activate', _('User registration with manual account activation')),
         ('hg.register.auto_activate', _('User registration with automatic account activation')),
         ('hg.extern_activate.manual', _('Manual activation of external account')),
         ('hg.extern_activate.auto', _('Automatic activation of external account')),
+    )
     # definition of system default permissions for DEFAULT user
     DEFAULT_USER_PERMISSIONS = (
         'repository.read',
         'group.read',
         'usergroup.read',
         'hg.create.repository',
         'hg.create.write_on_repogroup.true',
         'hg.fork.repository',
         'hg.register.manual_activate',
         'hg.extern_activate.auto',
+    )
     # defines which permissions are more important higher the more important
     # Weight defines which permissions are more important.
     # The higher number the more important.
     PERM_WEIGHTS = {
         'repository.none': 0,
         'repository.read': 1,
         'repository.write': 3,
         'repository.admin': 4,
         'group.none': 0,
         'group.read': 1,
         'group.write': 3,
         'group.admin': 4,
         'usergroup.none': 0,
         'usergroup.read': 1,
         'usergroup.write': 3,
         'usergroup.admin': 4,
         'hg.usergroup.create.false': 0,
         'hg.usergroup.create.true': 1,
         'hg.fork.none': 0,
         'hg.fork.repository': 1,
         'hg.create.none': 0,
         'hg.create.repository': 1,
         'hg.create.write_on_repogroup.false': 0,
         'hg.create.write_on_repogroup.true': 1,
         'hg.register.none': 0,
         'hg.register.manual_activate': 1,
         'hg.register.auto_activate': 2,
         'hg.extern_activate.manual': 0,
         'hg.extern_activate.auto': 1,
+    }
     permission_id = Column(Integer(), primary_key=True)
     permission_name = Column(String(255), nullable=False)
     def __repr__(self):
         return "<%s %s: %r>" % (
             self.__class__.__name__, self.permission_id, self.permission_name
+        )
     @classmethod
     def guess_instance(cls, value):
         return super(Permission, cls).guess_instance(value, Permission.get_by_key)
     @classmethod
     def get_by_key(cls, key):
         return cls.query().filter(cls.permission_name == key).scalar()
     @classmethod
     def get_default_perms(cls, default_user_id):
         q = Session().query(UserRepoToPerm) \
          .options(joinedload(UserRepoToPerm.repository)) \
          .options(joinedload(UserRepoToPerm.permission)) \
          .filter(UserRepoToPerm.user_id == default_user_id)
         return q.all()
     @classmethod
     def get_default_group_perms(cls, default_user_id):
         q = Session().query(UserRepoGroupToPerm) \
          .options(joinedload(UserRepoGroupToPerm.group)) \
          .options(joinedload(UserRepoGroupToPerm.permission)) \
          .filter(UserRepoGroupToPerm.user_id == default_user_id)
         return q.all()
     @classmethod
     def get_default_user_group_perms(cls, default_user_id):
         q = Session().query(UserUserGroupToPerm) \
          .options(joinedload(UserUserGroupToPerm.user_group)) \
          .options(joinedload(UserUserGroupToPerm.permission)) \
          .filter(UserUserGroupToPerm.user_id == default_user_id)
         return q.all()
 class UserRepoToPerm(Base, BaseDbModel):
     __tablename__ = 'repo_to_perm'
     __table_args__ = (
         UniqueConstraint('user_id', 'repository_id', 'permission_id'),
         _table_args_default_dict,
+    )
     repo_to_perm_id = Column(Integer(), primary_key=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     permission_id = Column(Integer(), ForeignKey('permissions.permission_id'), nullable=False)
     repository_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=False)
     user = relationship('User')
     repository = relationship('Repository')
     permission = relationship('Permission')
     @classmethod
     def create(cls, user, repository, permission):
         n = cls()
         n.user = user
         n.repository = repository
         n.permission = permission
         Session().add(n)
         return n
     def __repr__(self):
         return '<%s %s at %s: %s>' % (
             self.__class__.__name__, self.user, self.repository, self.permission)
 class UserUserGroupToPerm(Base, BaseDbModel):
     __tablename__ = 'user_user_group_to_perm'
     __table_args__ = (
         UniqueConstraint('user_id', 'user_group_id', 'permission_id'),
         _table_args_default_dict,
+    )
     user_user_group_to_perm_id = Column(Integer(), primary_key=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     permission_id = Column(Integer(), ForeignKey('permissions.permission_id'), nullable=False)
     user_group_id = Column(Integer(), ForeignKey('users_groups.users_group_id'), nullable=False)
     user = relationship('User')
     user_group = relationship('UserGroup')
     permission = relationship('Permission')
     @classmethod
     def create(cls, user, user_group, permission):
         n = cls()
         n.user = user
         n.user_group = user_group
         n.permission = permission
         Session().add(n)
         return n
     def __repr__(self):
         return '<%s %s at %s: %s>' % (
             self.__class__.__name__, self.user, self.user_group, self.permission)
 class UserToPerm(Base, BaseDbModel):
     __tablename__ = 'user_to_perm'
     __table_args__ = (
         UniqueConstraint('user_id', 'permission_id'),
         _table_args_default_dict,
+    )
     user_to_perm_id = Column(Integer(), primary_key=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     permission_id = Column(Integer(), ForeignKey('permissions.permission_id'), nullable=False)
     user = relationship('User')
     permission = relationship('Permission')
     def __repr__(self):
         return '<%s %s: %s>' % (
             self.__class__.__name__, self.user, self.permission)
 class UserGroupRepoToPerm(Base, BaseDbModel):
     __tablename__ = 'users_group_repo_to_perm'
     __table_args__ = (
         UniqueConstraint('repository_id', 'users_group_id', 'permission_id'),
         _table_args_default_dict,
+    )
     users_group_to_perm_id = Column(Integer(), primary_key=True)
     users_group_id = Column(Integer(), ForeignKey('users_groups.users_group_id'), nullable=False)
     permission_id = Column(Integer(), ForeignKey('permissions.permission_id'), nullable=False)
     repository_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=False)
     users_group = relationship('UserGroup')
     permission = relationship('Permission')
     repository = relationship('Repository')
     @classmethod
     def create(cls, users_group, repository, permission):
         n = cls()
         n.users_group = users_group
         n.repository = repository
         n.permission = permission
         Session().add(n)
         return n
     def __repr__(self):
         return '<%s %s at %s: %s>' % (
             self.__class__.__name__, self.users_group, self.repository, self.permission)
 class UserGroupUserGroupToPerm(Base, BaseDbModel):
     __tablename__ = 'user_group_user_group_to_perm'
     __table_args__ = (
         UniqueConstraint('target_user_group_id', 'user_group_id', 'permission_id'),
         _table_args_default_dict,
+    )
     user_group_user_group_to_perm_id = Column(Integer(), primary_key=True)
     target_user_group_id = Column(Integer(), ForeignKey('users_groups.users_group_id'), nullable=False)
     permission_id = Column(Integer(), ForeignKey('permissions.permission_id'), nullable=False)
     user_group_id = Column(Integer(), ForeignKey('users_groups.users_group_id'), nullable=False)
     target_user_group = relationship('UserGroup', primaryjoin='UserGroupUserGroupToPerm.target_user_group_id==UserGroup.users_group_id')
     user_group = relationship('UserGroup', primaryjoin='UserGroupUserGroupToPerm.user_group_id==UserGroup.users_group_id')
     permission = relationship('Permission')
     @classmethod
     def create(cls, target_user_group, user_group, permission):
         n = cls()
         n.target_user_group = target_user_group
         n.user_group = user_group
         n.permission = permission
         Session().add(n)
         return n
     def __repr__(self):
         return '<%s %s at %s: %s>' % (
             self.__class__.__name__, self.user_group, self.target_user_group, self.permission)
 class UserGroupToPerm(Base, BaseDbModel):
     __tablename__ = 'users_group_to_perm'
     __table_args__ = (
         UniqueConstraint('users_group_id', 'permission_id',),
         _table_args_default_dict,
+    )
     users_group_to_perm_id = Column(Integer(), primary_key=True)
     users_group_id = Column(Integer(), ForeignKey('users_groups.users_group_id'), nullable=False)
     permission_id = Column(Integer(), ForeignKey('permissions.permission_id'), nullable=False)
     users_group = relationship('UserGroup')
     permission = relationship('Permission')
 class UserRepoGroupToPerm(Base, BaseDbModel):
     __tablename__ = 'user_repo_group_to_perm'
     __table_args__ = (
         UniqueConstraint('user_id', 'group_id', 'permission_id'),
         _table_args_default_dict,
+    )
     group_to_perm_id = Column(Integer(), primary_key=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     group_id = Column(Integer(), ForeignKey('groups.group_id'), nullable=False)
     permission_id = Column(Integer(), ForeignKey('permissions.permission_id'), nullable=False)
     user = relationship('User')
     group = relationship('RepoGroup')
     permission = relationship('Permission')
     @classmethod
     def create(cls, user, repository_group, permission):
         n = cls()
         n.user = user
         n.group = repository_group
         n.permission = permission
         Session().add(n)
         return n
 class UserGroupRepoGroupToPerm(Base, BaseDbModel):
     __tablename__ = 'users_group_repo_group_to_perm'
     __table_args__ = (
         UniqueConstraint('users_group_id', 'group_id'),
         _table_args_default_dict,
+    )
     users_group_repo_group_to_perm_id = Column(Integer(), primary_key=True)
     users_group_id = Column(Integer(), ForeignKey('users_groups.users_group_id'), nullable=False)
     group_id = Column(Integer(), ForeignKey('groups.group_id'), nullable=False)
     permission_id = Column(Integer(), ForeignKey('permissions.permission_id'), nullable=False)
     users_group = relationship('UserGroup')
     permission = relationship('Permission')
     group = relationship('RepoGroup')
     @classmethod
     def create(cls, user_group, repository_group, permission):
         n = cls()
         n.users_group = user_group
         n.group = repository_group
         n.permission = permission
         Session().add(n)
         return n
 class Statistics(Base, BaseDbModel):
     __tablename__ = 'statistics'
     __table_args__ = (
          _table_args_default_dict,
+    )
     stat_id = Column(Integer(), primary_key=True)
     repository_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=False, unique=True)
     stat_on_revision = Column(Integer(), nullable=False)
     commit_activity = Column(LargeBinary(1000000), nullable=False) # JSON data
     commit_activity_combined = Column(LargeBinary(), nullable=False) # JSON data
     languages = Column(LargeBinary(1000000), nullable=False) # JSON data
     repository = relationship('Repository', single_parent=True)
 class UserFollowing(Base, BaseDbModel):
     __tablename__ = 'user_followings'
     __table_args__ = (
         UniqueConstraint('user_id', 'follows_repository_id', name='uq_user_followings_user_repo'),
         UniqueConstraint('user_id', 'follows_user_id', name='uq_user_followings_user_user'),
         _table_args_default_dict,
+    )
     user_following_id = Column(Integer(), primary_key=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     follows_repository_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=True)
     follows_user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=True)
     follows_from = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     user = relationship('User', primaryjoin='User.user_id==UserFollowing.user_id')
     follows_user = relationship('User', primaryjoin='User.user_id==UserFollowing.follows_user_id')
     follows_repository = relationship('Repository', order_by=lambda: sqlalchemy.func.lower(Repository.repo_name))
     @classmethod
     def get_repo_followers(cls, repo_id):
         return cls.query().filter(cls.follows_repository_id == repo_id)
 class ChangesetComment(Base, BaseDbModel):
     __tablename__ = 'changeset_comments'
     __table_args__ = (
         Index('cc_revision_idx', 'revision'),
         Index('cc_pull_request_id_idx', 'pull_request_id'),
         _table_args_default_dict,
+    )
     comment_id = Column(Integer(), primary_key=True)
     repo_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=False)
     revision = Column(String(40), nullable=True)
     pull_request_id = Column(Integer(), ForeignKey('pull_requests.pull_request_id'), nullable=True)
     line_no = Column(Unicode(10), nullable=True)
     f_path = Column(Unicode(1000), nullable=True)
     author_id = Column('user_id', Integer(), ForeignKey('users.user_id'), nullable=False)
     text = Column(UnicodeText(), nullable=False)
     created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     modified_at = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     author = relationship('User')
     repo = relationship('Repository')
     # status_change is frequently used directly in templates - make it a lazy
     # join to avoid fetching each related ChangesetStatus on demand.
     # There will only be one ChangesetStatus referencing each comment so the join will not explode.
     status_change = relationship('ChangesetStatus',
                                  cascade="all, delete-orphan", lazy='joined')
     pull_request = relationship('PullRequest')
     def url(self):
         anchor = "comment-%s" % self.comment_id
         import kallithea.lib.helpers as h
         if self.revision:
             return h.url('changeset_home', repo_name=self.repo.repo_name, revision=self.revision, anchor=anchor)
         elif self.pull_request_id is not None:
             return self.pull_request.url(anchor=anchor)
     def __json__(self):
         return dict(
             comment_id=self.comment_id,
             username=self.author.username,
             text=self.text,
+        )
     def deletable(self):
         return self.created_on > datetime.datetime.now() - datetime.timedelta(minutes=5)
 class ChangesetStatus(Base, BaseDbModel):
     __tablename__ = 'changeset_statuses'
     __table_args__ = (
         Index('cs_revision_idx', 'revision'),
         Index('cs_version_idx', 'version'),
         Index('cs_pull_request_id_idx', 'pull_request_id'),
         Index('cs_changeset_comment_id_idx', 'changeset_comment_id'),
         Index('cs_pull_request_id_user_id_version_idx', 'pull_request_id', 'user_id', 'version'),
         Index('cs_repo_id_pull_request_id_idx', 'repo_id', 'pull_request_id'),
         UniqueConstraint('repo_id', 'revision', 'version'),
         _table_args_default_dict,
+    )
     STATUS_NOT_REVIEWED = DEFAULT = 'not_reviewed'
     STATUS_APPROVED = 'approved'
     STATUS_REJECTED = 'rejected' # is shown as "Not approved" - TODO: change database content / scheme
     STATUS_UNDER_REVIEW = 'under_review'
     STATUSES = [
         (STATUS_NOT_REVIEWED, _("Not reviewed")),  # (no icon) and default
         (STATUS_UNDER_REVIEW, _("Under review")),
         (STATUS_REJECTED, _("Not approved")),
         (STATUS_APPROVED, _("Approved")),
+    ]
     STATUSES_DICT = dict(STATUSES)
     changeset_status_id = Column(Integer(), primary_key=True)
     repo_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=False)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     revision = Column(String(40), nullable=True)
     status = Column(String(128), nullable=False, default=DEFAULT)
     comment_id = Column('changeset_comment_id', Integer(), ForeignKey('changeset_comments.comment_id'), nullable=False)
     modified_at = Column(DateTime(), nullable=False, default=datetime.datetime.now)
     version = Column(Integer(), nullable=False, default=0)
     pull_request_id = Column(Integer(), ForeignKey('pull_requests.pull_request_id'), nullable=True)
     author = relationship('User')
     repo = relationship('Repository')
     comment = relationship('ChangesetComment')

kallithea/model/forms.py

➞

Show inline comments

@@ @@ -15,553 +15,552 @@ @@
 these are form validation classes
 http://formencode.org/module-formencode.validators.html
 for list of all available validators
 we can create our own validators
 The table below outlines the options which can be used in a schema in addition to the validators themselves
 pre_validators          []     These validators will be applied before the schema
 chained_validators      []     These validators will be applied after the schema
 allow_extra_fields      False     If True, then it is not an error when keys that aren't associated with a validator are present
 filter_extra_fields     False     If True, then keys that aren't associated with a validator are removed
 if_key_missing          NoDefault If this is given, then any keys that aren't available but are expected will be replaced with this value (and then validated). This does not override a present .if_missing attribute on validators. NoDefault is a special FormEncode class to mean that no default values has been specified and therefore missing keys shouldn't take a default value.
 ignore_key_missing      False     If True, then missing keys will be missing in the result, if the validator doesn't have .if_missing on it already
 <name> = formencode.validators.<name of validator>
 <name> must equal form name
 list=[1,2,3,4,5]
 for SELECT use formencode.All(OneOf(list), Int())
 """
 import logging
 import formencode
 from formencode import All
 from tg.i18n import ugettext as _
 from kallithea import BACKENDS
 from kallithea.model import validators as v
 log = logging.getLogger(__name__)
 def LoginForm():
     class _LoginForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         username = v.UnicodeString(
             strip=True,
             min=1,
             not_empty=True,
             messages={
                'empty': _('Please enter a login'),
                'tooShort': _('Enter a value %(min)i characters long or more')}
+        )
         password = v.UnicodeString(
             strip=False,
             min=3,
             not_empty=True,
             messages={
                 'empty': _('Please enter a password'),
                 'tooShort': _('Enter %(min)i characters or more')}
+        )
         remember = v.StringBoolean(if_missing=False)
         chained_validators = [v.ValidAuth()]
     return _LoginForm
 def PasswordChangeForm(username):
     class _PasswordChangeForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         current_password = v.ValidOldPassword(username)(not_empty=True)
         new_password = All(v.ValidPassword(), v.UnicodeString(strip=False, min=6))
         new_password_confirmation = All(v.ValidPassword(), v.UnicodeString(strip=False, min=6))
         chained_validators = [v.ValidPasswordsMatch('new_password',
                                                     'new_password_confirmation')]
     return _PasswordChangeForm
 def UserForm(edit=False, old_data=None):
     old_data = old_data or {}
     class _UserForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         username = All(v.UnicodeString(strip=True, min=1, not_empty=True),
                        v.ValidUsername(edit, old_data))
         if edit:
             new_password = All(
                 v.ValidPassword(),
                 v.UnicodeString(strip=False, min=6, not_empty=False)
+            )
             password_confirmation = All(
                 v.ValidPassword(),
                 v.UnicodeString(strip=False, min=6, not_empty=False),
+            )
             admin = v.StringBoolean(if_missing=False)
             chained_validators = [v.ValidPasswordsMatch('new_password',
                                                         'password_confirmation')]
         else:
             password = All(
                 v.ValidPassword(),
                 v.UnicodeString(strip=False, min=6, not_empty=True)
+            )
             password_confirmation = All(
                 v.ValidPassword(),
                 v.UnicodeString(strip=False, min=6, not_empty=False)
+            )
             chained_validators = [v.ValidPasswordsMatch('password',
                                                         'password_confirmation')]
         active = v.StringBoolean(if_missing=False)
         firstname = v.UnicodeString(strip=True, min=1, not_empty=False)
         lastname = v.UnicodeString(strip=True, min=1, not_empty=False)
         email = All(v.Email(not_empty=True), v.UniqSystemEmail(old_data))
         extern_name = v.UnicodeString(strip=True, if_missing=None)
         extern_type = v.UnicodeString(strip=True, if_missing=None)
     return _UserForm
 def UserGroupForm(edit=False, old_data=None, available_members=None):
     old_data = old_data or {}
     available_members = available_members or []
     class _UserGroupForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         users_group_name = All(
             v.UnicodeString(strip=True, min=1, not_empty=True),
             v.ValidUserGroup(edit, old_data)
+        )
         user_group_description = v.UnicodeString(strip=True, min=1,
                                                  not_empty=False)
         users_group_active = v.StringBoolean(if_missing=False)
         if edit:
             users_group_members = v.OneOf(
                 available_members, hideList=False, testValueList=True,
                 if_missing=None, not_empty=False
+            )
     return _UserGroupForm
 def RepoGroupForm(edit=False, old_data=None, repo_groups=None,
                    can_create_in_root=False):
     old_data = old_data or {}
     repo_groups = repo_groups or []
     repo_group_ids = [rg[0] for rg in repo_groups]
     class _RepoGroupForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         group_name = All(v.UnicodeString(strip=True, min=1, not_empty=True),
                          v.SlugifyName(),
                          v.ValidRegex(msg=_('Name must not contain only digits'))(r'(?!^\d+$)^.+$'))
         group_description = v.UnicodeString(strip=True, min=1,
                                             not_empty=False)
         group_copy_permissions = v.StringBoolean(if_missing=False)
         if edit:
             # FIXME: do a special check that we cannot move a group to one of
             # its children
             pass
         parent_group_id = All(v.CanCreateGroup(can_create_in_root),
                               v.OneOf(repo_group_ids, hideList=False,
                                       testValueList=True,
                                       if_missing=None, not_empty=True),
                               v.Int(min=-1, not_empty=True))
         chained_validators = [v.ValidRepoGroup(edit, old_data)]
     return _RepoGroupForm
 def RegisterForm(edit=False, old_data=None):
     class _RegisterForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         username = All(
             v.ValidUsername(edit, old_data),
             v.UnicodeString(strip=True, min=1, not_empty=True)
+        )
         password = All(
             v.ValidPassword(),
             v.UnicodeString(strip=False, min=6, not_empty=True)
+        )
         password_confirmation = All(
             v.ValidPassword(),
             v.UnicodeString(strip=False, min=6, not_empty=True)
+        )
         active = v.StringBoolean(if_missing=False)
         firstname = v.UnicodeString(strip=True, min=1, not_empty=False)
         lastname = v.UnicodeString(strip=True, min=1, not_empty=False)
         email = All(v.Email(not_empty=True), v.UniqSystemEmail(old_data))
         chained_validators = [v.ValidPasswordsMatch('password',
                                                     'password_confirmation')]
     return _RegisterForm
 def PasswordResetRequestForm():
     class _PasswordResetRequestForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         email = v.Email(not_empty=True)
     return _PasswordResetRequestForm
 def PasswordResetConfirmationForm():
     class _PasswordResetConfirmationForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         email = v.UnicodeString(strip=True, not_empty=True)
         timestamp = v.Number(strip=True, not_empty=True)
         token = v.UnicodeString(strip=True, not_empty=True)
         password = All(v.ValidPassword(), v.UnicodeString(strip=False, min=6))
         password_confirm = All(v.ValidPassword(), v.UnicodeString(strip=False, min=6))
         chained_validators = [v.ValidPasswordsMatch('password',
                                                     'password_confirm')]
     return _PasswordResetConfirmationForm
 def RepoForm(edit=False, old_data=None, supported_backends=BACKENDS,
              repo_groups=None, landing_revs=None):
     old_data = old_data or {}
     repo_groups = repo_groups or []
     landing_revs = landing_revs or []
     repo_group_ids = [rg[0] for rg in repo_groups]
     class _RepoForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         repo_name = All(v.UnicodeString(strip=True, min=1, not_empty=True),
                         v.SlugifyName())
         repo_group = All(v.CanWriteGroup(old_data),
                          v.OneOf(repo_group_ids, hideList=True),
                          v.Int(min=-1, not_empty=True))
         repo_type = v.OneOf(supported_backends, required=False,
                             if_missing=old_data.get('repo_type'))
         repo_description = v.UnicodeString(strip=True, min=1, not_empty=False)
         repo_private = v.StringBoolean(if_missing=False)
         repo_landing_rev = v.OneOf(landing_revs, hideList=True)
         repo_copy_permissions = v.StringBoolean(if_missing=False)
         clone_uri = All(v.UnicodeString(strip=True, min=1, not_empty=False))
         repo_enable_statistics = v.StringBoolean(if_missing=False)
         repo_enable_downloads = v.StringBoolean(if_missing=False)
         if edit:
             owner = All(v.UnicodeString(not_empty=True), v.ValidRepoUser())
             # Not a real field - just for reference for validation:
             # clone_uri_hidden = v.UnicodeString(if_missing='')
         chained_validators = [v.ValidCloneUri(),
                               v.ValidRepoName(edit, old_data)]
     return _RepoForm
 def RepoPermsForm():
     class _RepoPermsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         chained_validators = [v.ValidPerms(type_='repo')]
     return _RepoPermsForm
 def RepoGroupPermsForm(valid_recursive_choices):
     class _RepoGroupPermsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         recursive = v.OneOf(valid_recursive_choices)
         chained_validators = [v.ValidPerms(type_='repo_group')]
     return _RepoGroupPermsForm
 def UserGroupPermsForm():
     class _UserPermsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         chained_validators = [v.ValidPerms(type_='user_group')]
     return _UserPermsForm
 def RepoFieldForm():
     class _RepoFieldForm(formencode.Schema):
         filter_extra_fields = True
         allow_extra_fields = True
         new_field_key = All(v.FieldKey(),
                             v.UnicodeString(strip=True, min=3, not_empty=True))
         new_field_value = v.UnicodeString(not_empty=False, if_missing='')
         new_field_type = v.OneOf(['str', 'unicode', 'list', 'tuple'],
                                  if_missing='str')
         new_field_label = v.UnicodeString(not_empty=False)
         new_field_desc = v.UnicodeString(not_empty=False)
     return _RepoFieldForm
 def RepoForkForm(edit=False, old_data=None, supported_backends=BACKENDS,
                  repo_groups=None, landing_revs=None):
     old_data = old_data or {}
     repo_groups = repo_groups or []
     landing_revs = landing_revs or []
     repo_group_ids = [rg[0] for rg in repo_groups]
     class _RepoForkForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         repo_name = All(v.UnicodeString(strip=True, min=1, not_empty=True),
                         v.SlugifyName())
         repo_group = All(v.CanWriteGroup(),
                          v.OneOf(repo_group_ids, hideList=True),
                          v.Int(min=-1, not_empty=True))
         repo_type = All(v.ValidForkType(old_data), v.OneOf(supported_backends))
         description = v.UnicodeString(strip=True, min=1, not_empty=True)
         private = v.StringBoolean(if_missing=False)
         copy_permissions = v.StringBoolean(if_missing=False)
         update_after_clone = v.StringBoolean(if_missing=False)
         fork_parent_id = v.UnicodeString()
         chained_validators = [v.ValidForkName(edit, old_data)]
         landing_rev = v.OneOf(landing_revs, hideList=True)
     return _RepoForkForm
 def ApplicationSettingsForm():
     class _ApplicationSettingsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         title = v.UnicodeString(strip=True, not_empty=False)
         realm = v.UnicodeString(strip=True, min=1, not_empty=True)
         ga_code = v.UnicodeString(strip=True, min=1, not_empty=False)
         captcha_public_key = v.UnicodeString(strip=True, min=1, not_empty=False)
         captcha_private_key = v.UnicodeString(strip=True, min=1, not_empty=False)
     return _ApplicationSettingsForm
 def ApplicationVisualisationForm():
     class _ApplicationVisualisationForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         show_public_icon = v.StringBoolean(if_missing=False)
         show_private_icon = v.StringBoolean(if_missing=False)
         stylify_metalabels = v.StringBoolean(if_missing=False)
         repository_fields = v.StringBoolean(if_missing=False)
         lightweight_journal = v.StringBoolean(if_missing=False)
         dashboard_items = v.Int(min=5, not_empty=True)
         admin_grid_items = v.Int(min=5, not_empty=True)
         show_version = v.StringBoolean(if_missing=False)
         use_gravatar = v.StringBoolean(if_missing=False)
         gravatar_url = v.UnicodeString(min=3)
         clone_uri_tmpl = v.UnicodeString(min=3)
         clone_ssh_tmpl = v.UnicodeString()
     return _ApplicationVisualisationForm
 def ApplicationUiSettingsForm():
     class _ApplicationUiSettingsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         paths_root_path = All(
             v.ValidPath(),
             v.UnicodeString(strip=True, min=1, not_empty=True)
+        )
         hooks_changegroup_update = v.StringBoolean(if_missing=False)
         hooks_changegroup_repo_size = v.StringBoolean(if_missing=False)
         extensions_largefiles = v.StringBoolean(if_missing=False)
         extensions_hgsubversion = v.StringBoolean(if_missing=False)
         extensions_hggit = v.StringBoolean(if_missing=False)
     return _ApplicationUiSettingsForm
 def DefaultPermissionsForm(repo_perms_choices, group_perms_choices,
                            user_group_perms_choices, create_choices,
-                           create_on_write_choices, repo_group_create_choices,
                            repo_group_create_choices,
                            user_group_create_choices, fork_choices,
                            register_choices, extern_activate_choices):
     class _DefaultPermissionsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         overwrite_default_repo = v.StringBoolean(if_missing=False)
         overwrite_default_group = v.StringBoolean(if_missing=False)
         overwrite_default_user_group = v.StringBoolean(if_missing=False)
         anonymous = v.StringBoolean(if_missing=False)
         default_repo_perm = v.OneOf(repo_perms_choices)
         default_group_perm = v.OneOf(group_perms_choices)
         default_user_group_perm = v.OneOf(user_group_perms_choices)
         default_repo_create = v.OneOf(create_choices)
         create_on_write = v.OneOf(create_on_write_choices)
         default_user_group_create = v.OneOf(user_group_create_choices)
         default_fork = v.OneOf(fork_choices)
         default_register = v.OneOf(register_choices)
         default_extern_activate = v.OneOf(extern_activate_choices)
     return _DefaultPermissionsForm
 def CustomDefaultPermissionsForm():
     class _CustomDefaultPermissionsForm(formencode.Schema):
         filter_extra_fields = True
         allow_extra_fields = True
         create_repo_perm = v.StringBoolean(if_missing=False)
         create_user_group_perm = v.StringBoolean(if_missing=False)
         #create_repo_group_perm Impl. later
         fork_repo_perm = v.StringBoolean(if_missing=False)
     return _CustomDefaultPermissionsForm
 def DefaultsForm(edit=False, old_data=None, supported_backends=BACKENDS):
     class _DefaultsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         default_repo_type = v.OneOf(supported_backends)
         default_repo_private = v.StringBoolean(if_missing=False)
         default_repo_enable_statistics = v.StringBoolean(if_missing=False)
         default_repo_enable_downloads = v.StringBoolean(if_missing=False)
     return _DefaultsForm
 def AuthSettingsForm(current_active_modules):
     class _AuthSettingsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         auth_plugins = All(v.ValidAuthPlugins(),
                            v.UniqueListFromString()(not_empty=True))
         def __init__(self, *args, **kwargs):
             # The auth plugins tell us what form validators they use
             if current_active_modules:
                 import kallithea.lib.auth_modules
                 from kallithea.lib.auth_modules import LazyFormencode
                 for module in current_active_modules:
                     plugin = kallithea.lib.auth_modules.loadplugin(module)
                     plugin_name = plugin.name
                     for sv in plugin.plugin_settings():
                         newk = "auth_%s_%s" % (plugin_name, sv["name"])
                         # can be a LazyFormencode object from plugin settings
                         validator = sv["validator"]
                         if isinstance(validator, LazyFormencode):
                             validator = validator()
                         # init all lazy validators from formencode.All
                         if isinstance(validator, All):
                             init_validators = []
                             for validator in validator.validators:
                                 if isinstance(validator, LazyFormencode):
                                     validator = validator()
                                 init_validators.append(validator)
                             validator.validators = init_validators
                         self.add_field(newk, validator)
             formencode.Schema.__init__(self, *args, **kwargs)
     return _AuthSettingsForm
 def LdapSettingsForm(tls_reqcert_choices, search_scope_choices,
                      tls_kind_choices):
     class _LdapSettingsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         #pre_validators = [LdapLibValidator]
         ldap_active = v.StringBoolean(if_missing=False)
         ldap_host = v.UnicodeString(strip=True,)
         ldap_port = v.Number(strip=True,)
         ldap_tls_kind = v.OneOf(tls_kind_choices)
         ldap_tls_reqcert = v.OneOf(tls_reqcert_choices)
         ldap_dn_user = v.UnicodeString(strip=True,)
         ldap_dn_pass = v.UnicodeString(strip=True,)
         ldap_base_dn = v.UnicodeString(strip=True,)
         ldap_filter = v.UnicodeString(strip=True,)
         ldap_search_scope = v.OneOf(search_scope_choices)
         ldap_attr_login = v.AttrLoginValidator()(not_empty=True)
         ldap_attr_firstname = v.UnicodeString(strip=True,)
         ldap_attr_lastname = v.UnicodeString(strip=True,)
         ldap_attr_email = v.UnicodeString(strip=True,)
     return _LdapSettingsForm
 def UserExtraEmailForm():
     class _UserExtraEmailForm(formencode.Schema):
         email = All(v.UniqSystemEmail(), v.Email(not_empty=True))
     return _UserExtraEmailForm
 def UserExtraIpForm():
     class _UserExtraIpForm(formencode.Schema):
         ip = v.ValidIp()(not_empty=True)
     return _UserExtraIpForm
 def PullRequestForm(repo_id):
     class _PullRequestForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         org_repo = v.UnicodeString(strip=True, required=True)
         org_ref = v.UnicodeString(strip=True, required=True)
         other_repo = v.UnicodeString(strip=True, required=True)
         other_ref = v.UnicodeString(strip=True, required=True)
         pullrequest_title = v.UnicodeString(strip=True, required=True)
         pullrequest_desc = v.UnicodeString(strip=True, required=False)
     return _PullRequestForm
 def PullRequestPostForm():
     class _PullRequestPostForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         pullrequest_title = v.UnicodeString(strip=True, required=True)
         pullrequest_desc = v.UnicodeString(strip=True, required=False)
         org_review_members = v.Set()
         review_members = v.Set()
         updaterev = v.UnicodeString(strip=True, required=False, if_missing=None)
         owner = All(v.UnicodeString(strip=True, required=True),
                     v.ValidRepoUser())
     return _PullRequestPostForm
 def GistForm(lifetime_options):
     class _GistForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         filename = All(v.BasePath()(),
                        v.UnicodeString(strip=True, required=False))
         description = v.UnicodeString(required=False, if_missing='')
         lifetime = v.OneOf(lifetime_options)
         mimetype = v.UnicodeString(required=False, if_missing=None)
         content = v.UnicodeString(required=True, not_empty=True)
         public = v.UnicodeString(required=False, if_missing='')
         private = v.UnicodeString(required=False, if_missing='')
     return _GistForm

kallithea/model/permission.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # 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 General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.permission
 ~~~~~~~~~~~~~~~~~~~~~~~~~~
 permissions model for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Aug 20, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 from sqlalchemy.exc import DatabaseError
 from kallithea.lib.utils2 import asbool
 from kallithea.model.db import Permission, Session, User, UserRepoGroupToPerm, UserRepoToPerm, UserToPerm, UserUserGroupToPerm
 log = logging.getLogger(__name__)
 class PermissionModel(object):
     """
     Permissions model for Kallithea
     """
     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]
                 Session().add(new_perm)
     def create_default_permissions(self, user, force=False):
         """
         Create missing default permissions for user. If force is set, the default
         permissions for the user are reset, otherwise only missing permissions are
         created.
         :param user:
         """
         user = User.guess_instance(user)
         def _make_perm(perm):
             new_perm = UserToPerm()
             new_perm.user = user
             new_perm.permission = Permission.get_by_key(perm)
             return new_perm
         def _get_group(perm_name):
             return '.'.join(perm_name.split('.')[:1])
         perms = UserToPerm.query().filter(UserToPerm.user == user).all()
         defined_perms_groups = set(_get_group(x.permission.permission_name) for x in perms)
         log.debug('GOT ALREADY DEFINED:%s', perms)
         if force:
             for perm in perms:
                 Session().delete(perm)
             Session().commit()
             defined_perms_groups = []
         # For every default permission that needs to be created, we check if
         # its group is already defined. If it's not, we create default permission.
         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)
                 new_perm = _make_perm(perm_name)
                 Session().add(new_perm)
     def update(self, form_result):
         perm_user = User.get_by_username(username=form_result['perm_user_name'])
         try:
             # stage 1 set anonymous access
             if perm_user.is_default_user:
                 perm_user.active = asbool(form_result['anonymous'])
             # stage 2 reset defaults and set them from form data
             def _make_new(usr, perm_name):
                 log.debug('Creating new permission:%s', perm_name)
                 new = UserToPerm()
                 new.user = usr
                 new.permission = Permission.get_by_key(perm_name)
                 return new
             # clear current entries, to make this function idempotent
             # it will fix even if we define more permissions or permissions
             # are somehow missing
             u2p = UserToPerm.query() \
                 .filter(UserToPerm.user == perm_user) \
                 .all()
             for p in u2p:
                 Session().delete(p)
             # create fresh set of permissions
             for def_perm_key in ['default_repo_perm',
                                  'default_group_perm',
                                  'default_user_group_perm',
                                  'default_repo_create',
                                  'create_on_write', # special case for create repos on write access to group
                                  'default_user_group_create',
                                  'default_fork',
                                  'default_register',
                                  'default_extern_activate']:
                 p = _make_new(perm_user, form_result[def_perm_key])
                 Session().add(p)
             # stage 3 update all default permissions for repos if checked
             if form_result['overwrite_default_repo']:
                 _def_name = form_result['default_repo_perm'].split('repository.')[-1]
                 _def = Permission.get_by_key('repository.' + _def_name)
                 # repos
                 for r2p in UserRepoToPerm.query() \
                                .filter(UserRepoToPerm.user == perm_user) \
                                .all():
                     # don't reset PRIVATE repositories
                     if not r2p.repository.private:
                         r2p.permission = _def
             if form_result['overwrite_default_group']:
                 _def_name = form_result['default_group_perm'].split('group.')[-1]
                 # groups
                 _def = Permission.get_by_key('group.' + _def_name)
                 for g2p in UserRepoGroupToPerm.query() \
                                .filter(UserRepoGroupToPerm.user == perm_user) \
                                .all():
                     g2p.permission = _def
             if form_result['overwrite_default_user_group']:
                 _def_name = form_result['default_user_group_perm'].split('usergroup.')[-1]
                 # groups
                 _def = Permission.get_by_key('usergroup.' + _def_name)
                 for g2p in UserUserGroupToPerm.query() \
                                .filter(UserUserGroupToPerm.user == perm_user) \
                                .all():
                     g2p.permission = _def
             Session().commit()
         except (DatabaseError,):
             log.error(traceback.format_exc())
             Session().rollback()
             raise

kallithea/model/validators.py

➞

Show inline comments

@@ @@ -75,727 +75,726 @@ def ValidUsername(edit=False, old_data=N @@
                 _('Username "%(username)s" cannot be used'),
             'invalid_username':
                 _('Username may only contain alphanumeric characters '
                   'underscores, periods or dashes and must begin with an '
                   'alphanumeric character or underscore')
+        }
         def _validate_python(self, value, state):
             if value in ['default', 'new_user']:
                 msg = self.message('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 = self.message('username_exists', state, username=value)
                     raise formencode.Invalid(msg, value, state)
             if re.match(r'^[a-zA-Z0-9\_]{1}[a-zA-Z0-9\-\_\.]*$', value) is None:
                 msg = self.message('invalid_username', state)
                 raise formencode.Invalid(msg, value, state)
     return _validator
 def ValidRegex(msg=None):
     class _validator(formencode.validators.Regex):
         messages = dict(invalid=msg or _('The input is not valid'))
     return _validator
 def ValidRepoUser():
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_username': _('Username %(username)s is not valid')
+        }
         def _validate_python(self, value, state):
             try:
                 User.query().filter(User.active == True) \
                     .filter(User.username == value).one()
             except sqlalchemy.exc.InvalidRequestError: # NoResultFound/MultipleResultsFound
                 msg = self.message('invalid_username', state, username=value)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(username=msg)
+                )
     return _validator
 def ValidUserGroup(edit=False, old_data=None):
     old_data = old_data or {}
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_group': _('Invalid user group name'),
             'group_exist': _('User group "%(usergroup)s" already exists'),
             'invalid_usergroup_name':
                 _('user group name may only contain alphanumeric '
                   'characters underscores, periods or dashes and must begin '
                   'with alphanumeric character')
+        }
         def _validate_python(self, value, state):
             if value in ['default']:
                 msg = self.message('invalid_group', state)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=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 = self.message('group_exist', state, usergroup=value)
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(users_group_name=msg)
+                    )
             if re.match(r'^[a-zA-Z0-9]{1}[a-zA-Z0-9\-\_\.]+$', value) is None:
                 msg = self.message('invalid_usergroup_name', state)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(users_group_name=msg)
+                )
     return _validator
 def ValidRepoGroup(edit=False, old_data=None):
     old_data = old_data or {}
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'parent_group_id': _('Cannot assign this group as parent'),
             'group_exists': _('Group "%(group_name)s" already exists'),
             'repo_exists':
                 _('Repository with name "%(group_name)s" already exists')
+        }
         def _validate_python(self, value, state):
             # TODO WRITE VALIDATIONS
             group_name = value.get('group_name')
             parent_group_id = value.get('parent_group_id')
             # slugify repo group just in case :)
             slug = repo_name_slug(group_name)
             # check for parent of self
             if edit and parent_group_id and old_data['group_id'] == parent_group_id:
                 msg = self.message('parent_group_id', state)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(parent_group_id=msg)
+                )
             old_gname = None
             if edit:
                 old_gname = RepoGroup.get(old_data.get('group_id')).group_name
             if old_gname != group_name or not edit:
                 # check group
                 gr = RepoGroup.query() \
                       .filter(func.lower(RepoGroup.group_name) == func.lower(slug)) \
                       .filter(RepoGroup.parent_group_id == parent_group_id) \
                       .scalar()
                 if gr is not None:
                     msg = self.message('group_exists', state, group_name=slug)
                     raise formencode.Invalid(msg, value, state,
                             error_dict=dict(group_name=msg)
+                    )
                 # check for same repo
                 repo = Repository.query() \
                       .filter(func.lower(Repository.repo_name) == func.lower(slug)) \
                       .scalar()
                 if repo is not None:
                     msg = self.message('repo_exists', state, group_name=slug)
                     raise formencode.Invalid(msg, value, state,
                             error_dict=dict(group_name=msg)
+                    )
     return _validator
 def ValidPassword():
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_password':
                 _('Invalid characters (non-ascii) in password')
+        }
         def _validate_python(self, value, state):
             try:
                 (value or '').encode('ascii')
             except UnicodeError:
                 msg = self.message('invalid_password', state)
                 raise formencode.Invalid(msg, value, state,)
     return _validator
 def ValidOldPassword(username):
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_password': _('Invalid old password')
+        }
         def _validate_python(self, value, state):
             from kallithea.lib import auth_modules
             if auth_modules.authenticate(username, value, '') is None:
                 msg = self.message('invalid_password', state)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(current_password=msg)
+                )
     return _validator
 def ValidPasswordsMatch(password_field, password_confirmation_field):
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'password_mismatch': _('Passwords do not match'),
+        }
         def _validate_python(self, value, state):
             if value.get(password_field) != value[password_confirmation_field]:
                 msg = self.message('password_mismatch', state)
                 raise formencode.Invalid(msg, value, state,
                      error_dict={password_field: msg, password_confirmation_field: msg}
+                )
     return _validator
 def ValidAuth():
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_auth': _('Invalid username or password'),
+        }
         def _validate_python(self, value, state):
             from kallithea.lib import auth_modules
             password = value['password']
             username = value['username']
             # authenticate returns unused dict but has called
             # plugin._authenticate which has create_or_update'ed the username user in db
             if auth_modules.authenticate(username, password) is None:
                 user = User.get_by_username_or_email(username)
                 if user and not user.active:
                     log.warning('user %s is disabled', username)
                     msg = self.message('invalid_auth', state)
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(username=' ', password=msg)
+                    )
                 else:
                     log.warning('user %s failed to authenticate', username)
                     msg = self.message('invalid_auth', state)
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(username=' ', password=msg)
+                    )
     return _validator
 def ValidRepoName(edit=False, old_data=None):
     old_data = old_data or {}
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_repo_name':
                 _('Repository name %(repo)s is not allowed'),
             'repository_exists':
                 _('Repository named %(repo)s already exists'),
             'repository_in_group_exists': _('Repository "%(repo)s" already '
                                             'exists in group "%(group)s"'),
             'same_group_exists': _('Repository group with name "%(repo)s" '
                                    'already exists')
+        }
         def _convert_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 + db.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 = self.message('invalid_repo_name', state, repo=repo_name)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(repo_name=msg)
+                )
             rename = old_data.get('repo_name') != repo_name_full
             create = not edit
             if rename or create:
                 repo = Repository.get_by_repo_name(repo_name_full, case_insensitive=True)
                 repo_group = RepoGroup.get_by_group_name(repo_name_full, case_insensitive=True)
                 if group_path != '':
                     if repo is not None:
                         msg = self.message('repository_in_group_exists', state,
                                 repo=repo.repo_name, group=group_name)
                         raise formencode.Invalid(msg, value, state,
                             error_dict=dict(repo_name=msg)
+                        )
                 elif repo_group is not None:
                     msg = self.message('same_group_exists', state,
                             repo=repo_name)
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(repo_name=msg)
+                    )
                 elif repo is not None:
                     msg = self.message('repository_exists', state,
                             repo=repo.repo_name)
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(repo_name=msg)
+                    )
             return value
     return _validator
 def ValidForkName(*args, **kwargs):
     return ValidRepoName(*args, **kwargs)
 def SlugifyName():
     class _validator(formencode.validators.FancyValidator):
         def _convert_to_python(self, value, state):
             return repo_name_slug(value)
         def _validate_python(self, value, state):
             pass
     return _validator
 def ValidCloneUri():
     from kallithea.lib.utils import make_ui
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'clone_uri': _('Invalid repository URL'),
             'invalid_clone_uri': _('Invalid repository URL. It must be a '
                                    'valid http, https, ssh, svn+http or svn+https URL'),
+        }
         def _validate_python(self, value, state):
             repo_type = value.get('repo_type')
             url = value.get('clone_uri')
             if url and url != value.get('clone_uri_hidden'):
                 try:
                     is_valid_repo_uri(repo_type, url, make_ui())
                 except InvalidCloneUriException as e:
                     log.warning('validation of clone URL %r failed: %s', url, e)
                     msg = self.message('clone_uri', state)
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(clone_uri=msg)
+                    )
     return _validator
 def ValidForkType(old_data=None):
     old_data = old_data or {}
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_fork_type': _('Fork has to be the same type as parent')
+        }
         def _validate_python(self, value, state):
             if old_data['repo_type'] != value:
                 msg = self.message('invalid_fork_type', state)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(repo_type=msg)
+                )
     return _validator
 def CanWriteGroup(old_data=None):
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'permission_denied': _("You don't have permissions "
                                    "to create repository in this group"),
             'permission_denied_root': _("no permission to create repository "
                                         "in root location")
+        }
         def _convert_to_python(self, value, state):
             # root location
             if value == -1:
                 return None
             return value
         def _validate_python(self, value, state):
             gr = RepoGroup.get(value)
             gr_name = gr.group_name if gr is not None 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 = HasRepoGroupPermissionLevel('admin')(gr_name,
                                             'can write into group validator')
             group_write = HasRepoGroupPermissionLevel('write')(gr_name,
                                             'can write into group validator')
-            forbidden = not (group_admin or (group_write and create_on_write))
             forbidden = not (group_admin or group_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 != 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 = self.message('permission_denied', state)
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(repo_type=msg)
+                    )
                 ## check if we can write to root location !
                 elif gr is None and not can_create_repos():
                     msg = self.message('permission_denied_root', state)
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(repo_type=msg)
+                    )
     return _validator
 def CanCreateGroup(can_create_in_root=False):
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'permission_denied': _("You don't have permissions "
                                    "to create a group in this location")
+        }
         def to_python(self, value, state):
             # root location
             if value == -1:
                 return None
             return value
         def _validate_python(self, value, state):
             gr = RepoGroup.get(value)
             gr_name = gr.group_name if gr is not None else None # None means ROOT location
             if can_create_in_root and gr is None:
                 # we can create in root, we're fine no validations required
                 return
             forbidden_in_root = gr is None and not can_create_in_root
             forbidden = not HasRepoGroupPermissionLevel('admin')(gr_name, 'can create group validator')
             if forbidden_in_root or forbidden:
                 msg = self.message('permission_denied', state)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(parent_group_id=msg)
+                )
     return _validator
 def ValidPerms(type_='repo'):
     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':
                 _('This username or user group name is not valid')
+        }
         def to_python(self, value, state):
             perms_update = OrderedSet()
             perms_new = OrderedSet()
             # build a list of permission to update and new permission to create
             # CLEAN OUT ORG VALUE FROM NEW MEMBERS, and group them using
             new_perms_group = defaultdict(dict)
             for k, v in value.copy().items():
                 if k.startswith('perm_new_member'):
                     del value[k]
                     _type, part = k.split('perm_new_member_')
                     args = part.split('_')
                     if len(args) == 1:
                         new_perms_group[args[0]]['perm'] = v
                     elif len(args) == 2:
                         _key, pos = args
                         new_perms_group[pos][_key] = v
             # fill new permissions in order of how they were added
             for k in sorted(new_perms_group, key=lambda k: int(k)):
                 perm_dict = new_perms_group[k]
                 new_member = perm_dict.get('name')
                 new_perm = perm_dict.get('perm')
                 new_type = perm_dict.get('type')
                 if new_member and new_perm and new_type:
                     perms_new.add((new_member, new_perm, new_type))
             for k, v in value.items():
                 if k.startswith('u_perm_') or k.startswith('g_perm_'):
                     member_name = k[7:]
                     t = {'u': 'user',
                          'g': 'users_group'
                     }[k[0]]
                     if member_name == User.DEFAULT_USER_NAME:
                         if asbool(value.get('repo_private')):
                             # set none for default when updating to
                             # private repo protects against form manipulation
                             v = EMPTY_PERM
                     perms_update.add((member_name, v, t))
             value['perms_updates'] = list(perms_update)
             value['perms_new'] = list(perms_new)
             # update permissions
             for k, v, t in perms_new:
                 try:
                     if t == 'user':
                         _user_db = User.query() \
                             .filter(User.active == True) \
                             .filter(User.username == k).one()
                     if t == 'users_group':
                         _user_db = UserGroup.query() \
                             .filter(UserGroup.users_group_active == True) \
                             .filter(UserGroup.users_group_name == k).one()
                 except Exception as e:
                     log.warning('Error validating %s permission %s', t, k)
                     msg = self.message('perm_new_member_type', state)
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(perm_new_member_name=msg)
+                    )
             return value
     return _validator
 def ValidSettings():
     class _validator(formencode.validators.FancyValidator):
         def _convert_to_python(self, value, state):
             # settings  form for users that are not admin
             # can't edit certain parameters, it's extra backup if they mangle
             # with forms
             forbidden_params = [
                 'user', 'repo_type',
                 'repo_enable_downloads', 'repo_enable_statistics'
+            ]
             for param in forbidden_params:
                 if param in value:
                     del value[param]
             return value
         def _validate_python(self, value, state):
             pass
     return _validator
 def ValidPath():
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_path': _('This is not a valid path')
+        }
         def _validate_python(self, value, state):
             if not os.path.isdir(value):
                 msg = self.message('invalid_path', state)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(paths_root_path=msg)
+                )
     return _validator
 def UniqSystemEmail(old_data=None):
     old_data = old_data or {}
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'email_taken': _('This email address is already in use')
+        }
         def _convert_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)
                 if user is not None:
                     msg = self.message('email_taken', state)
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(email=msg)
+                    )
     return _validator
 def ValidSystemEmail():
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'non_existing_email': _('Email address "%(email)s" not found')
+        }
         def _convert_to_python(self, value, state):
             return value.lower()
         def _validate_python(self, value, state):
             user = User.get_by_email(value)
             if user is None:
                 msg = self.message('non_existing_email', state, email=value)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(email=msg)
+                )
     return _validator
 def LdapLibValidator():
     class _validator(formencode.validators.FancyValidator):
         messages = {
+        }
         def _validate_python(self, value, state):
             try:
                 import ldap
                 ldap  # pyflakes silence !
             except ImportError:
                 raise LdapImportError()
     return _validator
 def AttrLoginValidator():
     class _validator(formencode.validators.UnicodeString):
         messages = {
             'invalid_cn':
                   _('The LDAP Login attribute of the CN must be specified - '
                     'this is the name of the attribute that is equivalent '
                     'to "username"')
+        }
         messages['empty'] = messages['invalid_cn']
     return _validator
 def ValidIp():
     class _validator(CIDR):
         messages = dict(
             badFormat=_('Please enter a valid IPv4 or IPv6 address'),
             illegalBits=_('The network size (bits) must be within the range'
                 ' of 0-32 (not %(bits)r)')
+        )
         def to_python(self, value, state):
             v = super(_validator, self).to_python(value, state)
             v = v.strip()
             net = ipaddr.IPNetwork(address=v)
             if isinstance(net, ipaddr.IPv4Network):
                 # if IPv4 doesn't end with a mask, add /32
                 if '/' not in value:
                     v += '/32'
             if isinstance(net, ipaddr.IPv6Network):
                 # if IPv6 doesn't end with a mask, add /128
                 if '/' not in value:
                     v += '/128'
             return v
         def _validate_python(self, value, state):
             try:
                 addr = value.strip()
                 # this raises an ValueError if address is not IPv4 or IPv6
                 ipaddr.IPNetwork(address=addr)
             except ValueError:
                 raise formencode.Invalid(self.message('badFormat', state),
                                          value, state)
     return _validator
 def FieldKey():
     class _validator(formencode.validators.FancyValidator):
         messages = dict(
             badFormat=_('Key name can only consist of letters, '
                         '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 BasePath():
     class _validator(formencode.validators.FancyValidator):
         messages = dict(
             badPath=_('Filename cannot be inside a directory')
+        )
         def _convert_to_python(self, value, state):
             return value
         def _validate_python(self, value, state):
             if value != os.path.basename(value):
                 raise formencode.Invalid(self.message('badPath', state),
                                          value, state)
     return _validator
 def ValidAuthPlugins():
     class _validator(formencode.validators.FancyValidator):
         messages = dict(
             import_duplicate=_('Plugins %(loaded)s and %(next_to_load)s both export the same name')
+        )
         def _convert_to_python(self, value, state):
             # filter empty values
             return [s for s in value if s not in [None, '']]
         def _validate_python(self, value, state):
             from kallithea.lib import auth_modules
             module_list = value
             unique_names = {}
             try:
                 for module in module_list:
                     plugin = auth_modules.loadplugin(module)
                     plugin_name = plugin.name
                     if plugin_name in unique_names:
                         msg = self.message('import_duplicate', state,
                                 loaded=unique_names[plugin_name],
                                 next_to_load=plugin_name)
                         raise formencode.Invalid(msg, value, state)
                     unique_names[plugin_name] = plugin
             except (ImportError, AttributeError, TypeError) as e:
                 raise formencode.Invalid(str(e), value, state)
     return _validator

kallithea/templates/admin/permissions/permissions_globals.html

➞

Show inline comments

 ${h.form(url('admin_permissions'), method='post')}
     <div class="form">
             <div class="form-group">
                 <label class="control-label" for="anonymous">${_('Anonymous access')}:</label>
                 <div class="form-inline">
                     <label>
                         ${h.checkbox('anonymous',True)}
                         <span>${_('Allow anonymous access')}</span>
                     </label>
                     <span class="help-block">${h.HTML(_('Allow access to Kallithea without needing to log in. Anonymous users use %s user permissions.')) % (h.link_to('*default*',h.url('admin_permissions_perms')))}</span>
                 </div>
             </div>
             <div class="form-group">
                 <label class="control-label" for="default_repo_perm">${_('Repository')}:</label>
                 <div class="form-inline">
                     ${h.select('default_repo_perm','',c.repo_perms_choices,class_='form-control')}
                     <label>
                         ${h.checkbox('overwrite_default_repo','true')}
                         <span data-toggle="tooltip" title="${_('All default permissions on each repository will be reset to chosen permission, note that all custom default permission on repositories will be lost')}">
                             ${_('Apply to all existing repositories')}
                         </span>
                     </label>
                     <span class="help-block">${_('Permissions for the Default user on new repositories.')}</span>
                 </div>
             </div>
             <div class="form-group">
                 <label class="control-label" for="default_group_perm">${_('Repository group')}:</label>
                 <div class="form-inline">
                     ${h.select('default_group_perm','',c.group_perms_choices,class_='form-control')}
                     <label>
                         ${h.checkbox('overwrite_default_group','true')}
                         <span data-toggle="tooltip" title="${_('All default permissions on each repository group will be reset to chosen permission, note that all custom default permission on repository groups will be lost')}">
                             ${_('Apply to all existing repository groups')}
                         </span>
                     </label>
                     <span class="help-block">${_('Permissions for the Default user on new repository groups.')}</span>
                 </div>
             </div>
             <div class="form-group">
                 <label class="control-label" for="default_user_group_perm">${_('User group')}:</label>
                 <div class="form-inline">
                     ${h.select('default_user_group_perm','',c.user_group_perms_choices,class_='form-control')}
                     <label>
                         ${h.checkbox('overwrite_default_user_group','true')}
                         <span data-toggle="tooltip" title="${_('All default permissions on each user group will be reset to chosen permission, note that all custom default permission on user groups will be lost')}">
                             ${_('Apply to all existing user groups')}
                         </span>
                     </label>
                     <span class="help-block">${_('Permissions for the Default user on new user groups.')}</span>
                 </div>
             </div>
             <div class="form-group">
                 <label class="control-label" for="default_repo_create">${_('Top level repository creation')}:</label>
                 <div>
                     ${h.select('default_repo_create','',c.repo_create_choices,class_='form-control')}
                     <span class="help-block">${_('Enable this to allow non-admins to create repositories at the top level.')}</span>
                     <span class="help-block">${_('Note: This will also give all users API access to create repositories everywhere. That might change in future versions.')}</span>
                 </div>
             </div>
             <div class="form-group">
                 <label class="control-label" for="create_on_write">${_('Repository creation with group write access')}:</label>
                 <div>
                     ${h.select('create_on_write','',c.repo_create_on_write_choices,class_='form-control')}
                     <span class="help-block">${_('With this, write permission to a repository group allows creating repositories inside that group. Without this, group write permissions mean nothing.')}</span>
                 </div>
             </div>
             <div class="form-group">
                 <label class="control-label" for="default_user_group_create">${_('User group creation')}:</label>
                 <div>
                     ${h.select('default_user_group_create','',c.user_group_create_choices,class_='form-control')}
                     <span class="help-block">${_('Enable this to allow non-admins to create user groups.')}</span>
                 </div>
             </div>
             <div class="form-group">
                 <label class="control-label" for="default_fork">${_('Repository forking')}:</label>
                 <div>
                     ${h.select('default_fork','',c.fork_choices,class_='form-control')}
                     <span class="help-block">${_('Enable this to allow non-admins to fork repositories.')}</span>
                 </div>
             </div>
             <div class="form-group">
                 <label class="control-label" for="default_register">${_('Registration')}:</label>
                 <div>
                     ${h.select('default_register','',c.register_choices,class_='form-control')}
                 </div>
             </div>
             <div class="form-group">
                 <label class="control-label" for="default_extern_activate">${_('External auth account activation')}:</label>
                 <div>
                     ${h.select('default_extern_activate','',c.extern_activate_choices,class_='form-control')}
                 </div>
             </div>
             <div class="form-group">
                 <div class="buttons">
                     ${h.submit('save',_('Save'),class_="btn btn-default")}
                     ${h.reset('reset',_('Reset'),class_="btn btn-default")}
                 </div>
             </div>
     </div>
 ${h.end_form()}

kallithea/templates/index_base.html

➞

Show inline comments

 <%page args="parent,group_name=''" />
     <div class="panel panel-primary">
         <div class="panel-heading clearfix">
             <div class="pull-left panel-title">
                 %if c.group is not None:
                     %for group in c.group.parents:
                         ${h.link_to(group.name, url('repos_group_home', group_name=group.group_name))}
                         &raquo;
                     %endfor
                     ${c.group.name}
                 %endif
             </div>
             %if request.authuser.username != 'default':
               <div class="pull-right">
                 <%
                     gr_name = c.group.group_name if c.group else None
                     # create repositories with write permission on group is set to true
                     create_on_write = h.HasPermissionAny('hg.create.write_on_repogroup.true')()
                     group_admin = h.HasRepoGroupPermissionLevel('admin')(gr_name, 'can write into group index page')
                     group_write = h.HasRepoGroupPermissionLevel('write')(gr_name, 'can write into group index page')
                 %>
-                %if h.HasPermissionAny('hg.admin','hg.create.repository')() or (group_admin or (group_write and create_on_write)):
                 %if h.HasPermissionAny('hg.admin','hg.create.repository')() or group_admin or group_write:
                   %if c.group:
                         <a href="${h.url('new_repo',parent_group=c.group.group_id)}" class="btn btn-default btn-xs"><i class="icon-plus"></i>${_('Add Repository')}</a>
                         %if h.HasPermissionAny('hg.admin')() or h.HasRepoGroupPermissionLevel('admin')(c.group.group_name):
                             <a href="${h.url('new_repos_group', parent_group=c.group.group_id)}" class="btn btn-default btn-xs"><i class="icon-plus"></i>${_('Add Repository Group')}</a>
                         %endif
                   %else:
                     <a href="${h.url('new_repo')}" class="btn btn-default btn-xs"><i class="icon-plus"></i>${_('Add Repository')}</a>
                     %if h.HasPermissionAny('hg.admin')():
                         <a href="${h.url('new_repos_group')}" class="btn btn-default btn-xs"><i class="icon-plus"></i>${_('Add Repository Group')}</a>
                     %endif
                   %endif
                 %endif
                 %if c.group and h.HasRepoGroupPermissionLevel('admin')(c.group.group_name):
                     <a href="${h.url('edit_repo_group',group_name=c.group.group_name)}" title="${_('You have admin right to this group, and can edit it')}" class="btn btn-default btn-xs"><i class="icon-pencil"></i>${_('Edit Repository Group')}</a>
                 %endif
               </div>
             %endif
         </div>
         <div class="panel-body">
             <table class="table" id="repos_list_wrap" width="100%"></table>
         </div>
     </div>
       <script>'use strict';
         var data = ${h.js(c.data)};
         $("#repos_list_wrap").DataTable({
                 data: data.records,
                 columns: [
                     {data: "raw_name", visible: false, searchable: false},
                     {title: ${h.jshtml(_('Repository'))}, data: "name", orderData: [0,], render: {
                         filter: function(data, type, row) {
                             return row.just_name;
+                        }
                     }},
                     {data: "following", defaultContent: '', sortable: false},
                     {data: "desc", title: ${h.jshtml(_('Description'))}, searchable: false},
                     {data: "last_change_iso", defaultContent: '', visible: false, searchable: false},
                     {data: "last_change", defaultContent: '', title: ${h.jshtml(_('Last Change'))}, orderData: [4,], searchable: false},
                     {data: "last_rev_raw", defaultContent: '', visible: false, searchable: false},
                     {data: "last_changeset", defaultContent: '', title: ${h.jshtml(_('Tip'))}, orderData: [6,], searchable: false},
                     {data: "owner", defaultContent: '', title: ${h.jshtml(_('Owner'))}, searchable: false},
                     {data: "atom", defaultContent: '', sortable: false}
                 ],
                 order: [[1, "asc"]],
                 dom: '<"dataTables_left"f><"dataTables_right"ip>t',
                 pageLength: 100
             });
       </script>

kallithea/tests/models/test_permissions.py

➞

Show inline comments

 from kallithea.lib.auth import AuthUser
 from kallithea.model import db
 from kallithea.model.db import Permission, User, UserGroupRepoGroupToPerm, UserToPerm
 from kallithea.model.meta import Session
 from kallithea.model.permission import PermissionModel
 from kallithea.model.repo import RepoModel
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.model.user import UserModel
 from kallithea.model.user_group import UserGroupModel
 from kallithea.tests import base
 from kallithea.tests.fixture import Fixture
 fixture = Fixture()
 class TestPermissions(base.TestController):
     @classmethod
     def setup_class(cls):
         # recreate default user to get a clean start
         PermissionModel().create_default_permissions(user=User.DEFAULT_USER_NAME,
                                                      force=True)
         Session().commit()
     def setup_method(self, method):
         self.u1 = UserModel().create_or_update(
             username='u1', password='qweqwe',
             email='u1@example.com', firstname='u1', lastname='u1'
+        )
         self.u2 = UserModel().create_or_update(
             username='u2', password='qweqwe',
             email='u2@example.com', firstname='u2', lastname='u2'
+        )
         self.u3 = UserModel().create_or_update(
             username='u3', password='qweqwe',
             email='u3@example.com', firstname='u3', lastname='u3'
+        )
         self.anon = User.get_default_user()
         self.a1 = UserModel().create_or_update(
             username='a1', password='qweqwe',
             email='a1@example.com', firstname='a1', lastname='a1', admin=True
+        )
         Session().commit()
     def teardown_method(self, method):
         if hasattr(self, 'test_repo'):
             RepoModel().delete(repo=self.test_repo)
         UserModel().delete(self.u1)
         UserModel().delete(self.u2)
         UserModel().delete(self.u3)
         UserModel().delete(self.a1)
         Session().commit() # commit early to avoid SQLAlchemy warning from double cascade delete to users_groups_members
         if hasattr(self, 'g1'):
             RepoGroupModel().delete(self.g1.group_id)
         if hasattr(self, 'g2'):
             RepoGroupModel().delete(self.g2.group_id)
         if hasattr(self, 'ug2'):
             UserGroupModel().delete(self.ug2, force=True)
         if hasattr(self, 'ug1'):
             UserGroupModel().delete(self.ug1, force=True)
         Session().commit()
     def test_default_perms_set(self):
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['repositories'][base.HG_REPO] == 'repository.read'
         new_perm = 'repository.write'
         RepoModel().grant_user_permission(repo=base.HG_REPO, user=self.u1,
                                           perm=new_perm)
         Session().commit()
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['repositories'][base.HG_REPO] == new_perm
     def test_default_admin_perms_set(self):
         a1_auth = AuthUser(user_id=self.a1.user_id)
         assert a1_auth.permissions['repositories'][base.HG_REPO] == 'repository.admin'
         new_perm = 'repository.write'
         RepoModel().grant_user_permission(repo=base.HG_REPO, user=self.a1,
                                           perm=new_perm)
         Session().commit()
         # cannot really downgrade admins permissions !? they still gets set as
         # admin !
         u1_auth = AuthUser(user_id=self.a1.user_id)
         assert u1_auth.permissions['repositories'][base.HG_REPO] == 'repository.admin'
     def test_default_group_perms(self):
         self.g1 = fixture.create_repo_group('test1', skip_if_exists=True)
         self.g2 = fixture.create_repo_group('test2', skip_if_exists=True)
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['repositories'][base.HG_REPO] == 'repository.read'
         assert u1_auth.permissions['repositories_groups'].get('test1') == 'group.read'
         assert u1_auth.permissions['repositories_groups'].get('test2') == 'group.read'
         assert u1_auth.permissions['global'] == set(Permission.DEFAULT_USER_PERMISSIONS)
     def test_default_admin_group_perms(self):
         self.g1 = fixture.create_repo_group('test1', skip_if_exists=True)
         self.g2 = fixture.create_repo_group('test2', skip_if_exists=True)
         a1_auth = AuthUser(user_id=self.a1.user_id)
         assert a1_auth.permissions['repositories'][base.HG_REPO] == 'repository.admin'
         assert a1_auth.permissions['repositories_groups'].get('test1') == 'group.admin'
         assert a1_auth.permissions['repositories_groups'].get('test2') == 'group.admin'
     def test_propagated_permission_from_users_group_by_explicit_perms_exist(self):
         # make group
         self.ug1 = fixture.create_user_group('G1')
         UserGroupModel().add_user_to_group(self.ug1, self.u1)
         # set user permission none
         RepoModel().grant_user_permission(repo=base.HG_REPO, user=self.u1, perm='repository.none')
         Session().commit()
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['repositories'][base.HG_REPO] == 'repository.read' # inherit from default user
         # grant perm for group this should override permission from user
         RepoModel().grant_user_group_permission(repo=base.HG_REPO,
                                                  group_name=self.ug1,
                                                  perm='repository.write')
         # verify that user group permissions win
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['repositories'][base.HG_REPO] == 'repository.write'
     def test_propagated_permission_from_users_group(self):
         # make group
         self.ug1 = fixture.create_user_group('G1')
         UserGroupModel().add_user_to_group(self.ug1, self.u3)
         # grant perm for group this should override default permission from user
         new_perm_gr = 'repository.write'
         RepoModel().grant_user_group_permission(repo=base.HG_REPO,
                                                  group_name=self.ug1,
                                                  perm=new_perm_gr)
         # check perms
         u3_auth = AuthUser(user_id=self.u3.user_id)
         assert u3_auth.permissions['repositories'][base.HG_REPO] == new_perm_gr
     def test_propagated_permission_from_users_group_lower_weight(self):
         # make group
         self.ug1 = fixture.create_user_group('G1')
         # add user to group
         UserGroupModel().add_user_to_group(self.ug1, self.u1)
         # set permission to lower
         new_perm_h = 'repository.write'
         RepoModel().grant_user_permission(repo=base.HG_REPO, user=self.u1,
                                           perm=new_perm_h)
         Session().commit()
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['repositories'][base.HG_REPO] == new_perm_h
         # grant perm for group this should NOT override permission from user
         # since it's lower than granted
         new_perm_l = 'repository.read'
         RepoModel().grant_user_group_permission(repo=base.HG_REPO,
                                                  group_name=self.ug1,
                                                  perm=new_perm_l)
         # check perms
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['repositories'][base.HG_REPO] == new_perm_h
     def test_repo_in_group_permissions(self):
         self.g1 = fixture.create_repo_group('group1', skip_if_exists=True)
         self.g2 = fixture.create_repo_group('group2', skip_if_exists=True)
         # both perms should be read !
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['repositories_groups'].get('group1') == 'group.read'
         assert u1_auth.permissions['repositories_groups'].get('group2') == 'group.read'
         a1_auth = AuthUser(user_id=self.anon.user_id)
         assert a1_auth.permissions['repositories_groups'].get('group1') == 'group.read'
         assert a1_auth.permissions['repositories_groups'].get('group2') == 'group.read'
         # Change perms to none for both groups
         RepoGroupModel().grant_user_permission(repo_group=self.g1,
                                                user=self.anon,
                                                perm='group.none')
         RepoGroupModel().grant_user_permission(repo_group=self.g2,
                                                user=self.anon,
                                                perm='group.none')
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['repositories_groups'].get('group1') == 'group.none'
         assert u1_auth.permissions['repositories_groups'].get('group2') == 'group.none'
         a1_auth = AuthUser(user_id=self.anon.user_id)
         assert a1_auth.permissions['repositories_groups'].get('group1') == 'group.none'
         assert a1_auth.permissions['repositories_groups'].get('group2') == 'group.none'
         # add repo to group
         name = db.URL_SEP.join([self.g1.group_name, 'test_perm'])
         self.test_repo = fixture.create_repo(name=name,
                                              repo_type='hg',
                                              repo_group=self.g1,
                                              cur_user=self.u1,)
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['repositories_groups'].get('group1') == 'group.none'
         assert u1_auth.permissions['repositories_groups'].get('group2') == 'group.none'
         a1_auth = AuthUser(user_id=self.anon.user_id)
         assert a1_auth.permissions['repositories_groups'].get('group1') == 'group.none'
         assert a1_auth.permissions['repositories_groups'].get('group2') == 'group.none'
         # grant permission for u2 !
         RepoGroupModel().grant_user_permission(repo_group=self.g1, user=self.u2,
                                                perm='group.read')
         RepoGroupModel().grant_user_permission(repo_group=self.g2, user=self.u2,
                                                perm='group.read')
         Session().commit()
         assert self.u1 != self.u2
         # u1 and anon should have not change perms while u2 should !
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['repositories_groups'].get('group1') == 'group.none'
         assert u1_auth.permissions['repositories_groups'].get('group2') == 'group.none'
         u2_auth = AuthUser(user_id=self.u2.user_id)
         assert u2_auth.permissions['repositories_groups'].get('group1') == 'group.read'
         assert u2_auth.permissions['repositories_groups'].get('group2') == 'group.read'
         a1_auth = AuthUser(user_id=self.anon.user_id)
         assert a1_auth.permissions['repositories_groups'].get('group1') == 'group.none'
         assert a1_auth.permissions['repositories_groups'].get('group2') == 'group.none'
     def test_repo_group_user_as_user_group_member(self):
         # create Group1
         self.g1 = fixture.create_repo_group('group1', skip_if_exists=True)
         a1_auth = AuthUser(user_id=self.anon.user_id)
         assert a1_auth.permissions['repositories_groups'].get('group1') == 'group.read'
         # set default permission to none
         RepoGroupModel().grant_user_permission(repo_group=self.g1,
                                                user=self.anon,
                                                perm='group.none')
         # make group
         self.ug1 = fixture.create_user_group('G1')
         # add user to group
         UserGroupModel().add_user_to_group(self.ug1, self.u1)
         Session().commit()
         # check if user is in the group
         members = [x.user_id for x in UserGroupModel().get(self.ug1.users_group_id).members]
         assert members == [self.u1.user_id]
         # add some user to that group
         # check his permissions
         a1_auth = AuthUser(user_id=self.anon.user_id)
         assert a1_auth.permissions['repositories_groups'].get('group1') == 'group.none'
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['repositories_groups'].get('group1') == 'group.none'
         # grant ug1 read permissions for
         RepoGroupModel().grant_user_group_permission(repo_group=self.g1,
                                                       group_name=self.ug1,
                                                       perm='group.read')
         Session().commit()
         # check if the
         obj = Session().query(UserGroupRepoGroupToPerm) \
             .filter(UserGroupRepoGroupToPerm.group == self.g1) \
             .filter(UserGroupRepoGroupToPerm.users_group == self.ug1) \
             .scalar()
         assert obj.permission.permission_name == 'group.read'
         a1_auth = AuthUser(user_id=self.anon.user_id)
         assert a1_auth.permissions['repositories_groups'].get('group1') == 'group.none'
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['repositories_groups'].get('group1') == 'group.read'
     def test_inherit_nice_permissions_from_default_user(self):
         user_model = UserModel()
         # enable fork and create on default user
         usr = 'default'
         user_model.revoke_perm(usr, 'hg.create.none')
         user_model.grant_perm(usr, 'hg.create.repository')
         user_model.revoke_perm(usr, 'hg.fork.none')
         user_model.grant_perm(usr, 'hg.fork.repository')
         Session().commit()
         u1_auth = AuthUser(user_id=self.u1.user_id)
         # this user will have inherited permissions from default user
         assert u1_auth.permissions['global'] == set(['hg.create.repository', 'hg.fork.repository',
                               'hg.register.manual_activate',
                               'hg.extern_activate.auto',
                               'repository.read', 'group.read',
-                              'usergroup.read', 'hg.create.write_on_repogroup.true'])
                               'usergroup.read'])
     def test_inherit_sad_permissions_from_default_user(self):
         user_model = UserModel()
         # disable fork and create on default user
         usr = 'default'
         user_model.revoke_perm(usr, 'hg.create.repository')
         user_model.grant_perm(usr, 'hg.create.none')
         user_model.revoke_perm(usr, 'hg.fork.repository')
         user_model.grant_perm(usr, 'hg.fork.none')
         Session().commit()
         u1_auth = AuthUser(user_id=self.u1.user_id)
         # this user will have inherited permissions from default user
         assert u1_auth.permissions['global'] == set(['hg.create.none', 'hg.fork.none',
                               'hg.register.manual_activate',
                               'hg.extern_activate.auto',
                               'repository.read', 'group.read',
-                              'usergroup.read', 'hg.create.write_on_repogroup.true'])
                               'usergroup.read'])
     def test_inherit_more_permissions_from_default_user(self):
         user_model = UserModel()
         # enable fork and create on default user
         usr = 'default'
         user_model.revoke_perm(usr, 'hg.create.none')
         user_model.grant_perm(usr, 'hg.create.repository')
         user_model.revoke_perm(usr, 'hg.fork.none')
         user_model.grant_perm(usr, 'hg.fork.repository')
         # disable global perms on specific user
         user_model.revoke_perm(self.u1, 'hg.create.repository')
         user_model.grant_perm(self.u1, 'hg.create.none')
         user_model.revoke_perm(self.u1, 'hg.fork.repository')
         user_model.grant_perm(self.u1, 'hg.fork.none')
         Session().commit()
         u1_auth = AuthUser(user_id=self.u1.user_id)
         # this user will have inherited more permissions from default user
         assert u1_auth.permissions['global'] == set([
                               'hg.create.repository',
                               'hg.fork.repository',
                               'hg.register.manual_activate',
                               'hg.extern_activate.auto',
                               'repository.read', 'group.read',
-                              'usergroup.read', 'hg.create.write_on_repogroup.true'])
                               'usergroup.read'])
     def test_inherit_less_permissions_from_default_user(self):
         user_model = UserModel()
         # disable fork and create on default user
         usr = 'default'
         user_model.revoke_perm(usr, 'hg.create.repository')
         user_model.grant_perm(usr, 'hg.create.none')
         user_model.revoke_perm(usr, 'hg.fork.repository')
         user_model.grant_perm(usr, 'hg.fork.none')
         # enable global perms on specific user
         user_model.revoke_perm(self.u1, 'hg.create.none')
         user_model.grant_perm(self.u1, 'hg.create.repository')
         user_model.revoke_perm(self.u1, 'hg.fork.none')
         user_model.grant_perm(self.u1, 'hg.fork.repository')
         Session().commit()
         u1_auth = AuthUser(user_id=self.u1.user_id)
         # this user will have inherited less permissions from default user
         assert u1_auth.permissions['global'] == set([
                               'hg.create.repository',
                               'hg.fork.repository',
                               'hg.register.manual_activate',
                               'hg.extern_activate.auto',
                               'repository.read', 'group.read',
-                              'usergroup.read', 'hg.create.write_on_repogroup.true'])
                               'usergroup.read'])
     def test_inactive_user_group_does_not_affect_global_permissions(self):
         # Add user to inactive user group, set specific permissions on user
         # group and and verify it really is inactive.
         self.ug1 = fixture.create_user_group('G1')
         user_group_model = UserGroupModel()
         user_group_model.add_user_to_group(self.ug1, self.u1)
         user_group_model.update(self.ug1, {'users_group_active': False})
         # enable fork and create on user group
         user_group_model.revoke_perm(self.ug1, perm='hg.create.none')
         user_group_model.grant_perm(self.ug1, perm='hg.create.repository')
         user_group_model.revoke_perm(self.ug1, perm='hg.fork.none')
         user_group_model.grant_perm(self.ug1, perm='hg.fork.repository')
         user_model = UserModel()
         # disable fork and create on default user
         usr = 'default'
         user_model.revoke_perm(usr, 'hg.create.repository')
         user_model.grant_perm(usr, 'hg.create.none')
         user_model.revoke_perm(usr, 'hg.fork.repository')
         user_model.grant_perm(usr, 'hg.fork.none')
         Session().commit()
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['global'] == set(['hg.create.none', 'hg.fork.none',
                               'hg.register.manual_activate',
                               'hg.extern_activate.auto',
                               'repository.read', 'group.read',
                               'usergroup.read',
-                              'hg.create.write_on_repogroup.true'])
                               ])
     def test_inactive_user_group_does_not_affect_global_permissions_inverse(self):
         # Add user to inactive user group, set specific permissions on user
         # group and and verify it really is inactive.
         self.ug1 = fixture.create_user_group('G1')
         user_group_model = UserGroupModel()
         user_group_model.add_user_to_group(self.ug1, self.u1)
         user_group_model.update(self.ug1, {'users_group_active': False})
         # disable fork and create on user group
         user_group_model.revoke_perm(self.ug1, perm='hg.create.repository')
         user_group_model.grant_perm(self.ug1, perm='hg.create.none')
         user_group_model.revoke_perm(self.ug1, perm='hg.fork.repository')
         user_group_model.grant_perm(self.ug1, perm='hg.fork.none')
         user_model = UserModel()
         # enable fork and create on default user
         usr = 'default'
         user_model.revoke_perm(usr, 'hg.create.none')
         user_model.grant_perm(usr, 'hg.create.repository')
         user_model.revoke_perm(usr, 'hg.fork.none')
         user_model.grant_perm(usr, 'hg.fork.repository')
         Session().commit()
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['global'] == set(['hg.create.repository', 'hg.fork.repository',
                               'hg.register.manual_activate',
                               'hg.extern_activate.auto',
                               'repository.read', 'group.read',
                               'usergroup.read',
-                              'hg.create.write_on_repogroup.true'])
                               ])
     def test_inactive_user_group_does_not_affect_repo_permissions(self):
         self.ug1 = fixture.create_user_group('G1')
         user_group_model = UserGroupModel()
         user_group_model.add_user_to_group(self.ug1, self.u1)
         user_group_model.update(self.ug1, {'users_group_active': False})
         # note: make u2 repo owner rather than u1, because the owner always has
         # admin permissions
         self.test_repo = fixture.create_repo(name='myownrepo',
                                              repo_type='hg',
                                              cur_user=self.u2)
         # enable admin access for user group on repo
         RepoModel().grant_user_group_permission(self.test_repo,
                                                 group_name=self.ug1,
                                                 perm='repository.admin')
         # enable only write access for default user on repo
         RepoModel().grant_user_permission(self.test_repo,
                                           user='default',
                                           perm='repository.write')
         Session().commit()
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['repositories']['myownrepo'] == 'repository.write'
     def test_inactive_user_group_does_not_affect_repo_permissions_inverse(self):
         self.ug1 = fixture.create_user_group('G1')
         user_group_model = UserGroupModel()
         user_group_model.add_user_to_group(self.ug1, self.u1)
         user_group_model.update(self.ug1, {'users_group_active': False})
         # note: make u2 repo owner rather than u1, because the owner always has
         # admin permissions
         self.test_repo = fixture.create_repo(name='myownrepo',
                                              repo_type='hg',
                                              cur_user=self.u2)
         # enable only write access for user group on repo
         RepoModel().grant_user_group_permission(self.test_repo,
                                                 group_name=self.ug1,
                                                 perm='repository.write')
         # enable admin access for default user on repo
         RepoModel().grant_user_permission(self.test_repo,
                                           user='default',
                                           perm='repository.admin')
         Session().commit()
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['repositories']['myownrepo'] == 'repository.admin'
     def test_inactive_user_group_does_not_affect_repo_group_permissions(self):
         self.ug1 = fixture.create_user_group('G1')
         user_group_model = UserGroupModel()
         user_group_model.add_user_to_group(self.ug1, self.u1)
         user_group_model.update(self.ug1, {'users_group_active': False})
         self.g1 = fixture.create_repo_group('group1', skip_if_exists=True)
         # enable admin access for user group on repo group
         RepoGroupModel().grant_user_group_permission(self.g1,
                                                      group_name=self.ug1,
                                                      perm='group.admin')
         # enable only write access for default user on repo group
         RepoGroupModel().grant_user_permission(self.g1,
                                                user='default',
                                                perm='group.write')
         Session().commit()
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['repositories_groups'].get('group1') == 'group.write'
     def test_inactive_user_group_does_not_affect_repo_group_permissions_inverse(self):
         self.ug1 = fixture.create_user_group('G1')
         user_group_model = UserGroupModel()
         user_group_model.add_user_to_group(self.ug1, self.u1)
         user_group_model.update(self.ug1, {'users_group_active': False})
         self.g1 = fixture.create_repo_group('group1', skip_if_exists=True)
         # enable only write access for user group on repo group
         RepoGroupModel().grant_user_group_permission(self.g1,
                                                      group_name=self.ug1,
                                                      perm='group.write')
         # enable admin access for default user on repo group
         RepoGroupModel().grant_user_permission(self.g1,
                                                user='default',
                                                perm='group.admin')
         Session().commit()
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['repositories_groups'].get('group1') == 'group.admin'
     def test_inactive_user_group_does_not_affect_user_group_permissions(self):
         self.ug1 = fixture.create_user_group('G1')
         user_group_model = UserGroupModel()
         user_group_model.add_user_to_group(self.ug1, self.u1)
         user_group_model.update(self.ug1, {'users_group_active': False})
         self.ug2 = fixture.create_user_group('G2')
         # enable admin access for user group on user group
         UserGroupModel().grant_user_group_permission(self.ug2,
                                                      user_group=self.ug1,
                                                      perm='usergroup.admin')
         # enable only write access for default user on user group
         UserGroupModel().grant_user_permission(self.ug2,
                                                user='default',
                                                perm='usergroup.write')
         Session().commit()
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['user_groups']['G1'] == 'usergroup.read'
         assert u1_auth.permissions['user_groups']['G2'] == 'usergroup.write'
     def test_inactive_user_group_does_not_affect_user_group_permissions_inverse(self):
         self.ug1 = fixture.create_user_group('G1')
         user_group_model = UserGroupModel()
         user_group_model.add_user_to_group(self.ug1, self.u1)
         user_group_model.update(self.ug1, {'users_group_active': False})
         self.ug2 = fixture.create_user_group('G2')
         # enable only write access for user group on user group
         UserGroupModel().grant_user_group_permission(self.ug2,
                                                      user_group=self.ug1,
                                                      perm='usergroup.write')
         # enable admin access for default user on user group
         UserGroupModel().grant_user_permission(self.ug2,
                                                user='default',
                                                perm='usergroup.admin')
         Session().commit()
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['user_groups']['G1'] == 'usergroup.read'
         assert u1_auth.permissions['user_groups']['G2'] == 'usergroup.admin'
     def test_owner_permissions_doesnot_get_overwritten_by_group(self):
         # create repo as USER,
         self.test_repo = fixture.create_repo(name='myownrepo',
                                              repo_type='hg',
                                              cur_user=self.u1)
         # he has permissions of admin as owner
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['repositories']['myownrepo'] == 'repository.admin'
         # set his permission as user group, he should still be admin
         self.ug1 = fixture.create_user_group('G1')
         UserGroupModel().add_user_to_group(self.ug1, self.u1)
         RepoModel().grant_user_group_permission(self.test_repo,
                                                  group_name=self.ug1,
                                                  perm='repository.none')
         Session().commit()
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['repositories']['myownrepo'] == 'repository.admin'
     def test_owner_permissions_doesnot_get_overwritten_by_others(self):
         # create repo as USER,
         self.test_repo = fixture.create_repo(name='myownrepo',
                                              repo_type='hg',
                                              cur_user=self.u1)
         # he has permissions of admin as owner
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['repositories']['myownrepo'] == 'repository.admin'
         # set his permission as user, he should still be admin
         RepoModel().grant_user_permission(self.test_repo, user=self.u1,
                                           perm='repository.none')
         Session().commit()
         u1_auth = AuthUser(user_id=self.u1.user_id)
         assert u1_auth.permissions['repositories']['myownrepo'] == 'repository.admin'
     def _test_def_perm_equal(self, user, change_factor=0):
         perms = UserToPerm.query() \
                 .filter(UserToPerm.user == user) \
                 .all()
         assert len(perms) == len(Permission.DEFAULT_USER_PERMISSIONS,)+change_factor, perms
     def test_set_default_permissions(self):
         PermissionModel().create_default_permissions(user=self.u1)
         self._test_def_perm_equal(user=self.u1)
     def test_set_default_permissions_after_one_is_missing(self):
         PermissionModel().create_default_permissions(user=self.u1)
         self._test_def_perm_equal(user=self.u1)
         # now we delete one, it should be re-created after another call
         perms = UserToPerm.query() \
                 .filter(UserToPerm.user == self.u1) \
                 .all()
         Session().delete(perms[0])
         Session().commit()
         self._test_def_perm_equal(user=self.u1, change_factor=-1)
         # create missing one !
         PermissionModel().create_default_permissions(user=self.u1)
         self._test_def_perm_equal(user=self.u1)
     @base.parametrize('perm,modify_to', [
         ('repository.read', 'repository.none'),
         ('group.read', 'group.none'),
         ('usergroup.read', 'usergroup.none'),
         ('hg.create.repository', 'hg.create.none'),
         ('hg.fork.repository', 'hg.fork.none'),
         ('hg.register.manual_activate', 'hg.register.auto_activate',)
     ])
     def test_set_default_permissions_after_modification(self, perm, modify_to):
         PermissionModel().create_default_permissions(user=self.u1)
         self._test_def_perm_equal(user=self.u1)
         old = Permission.get_by_key(perm)
         new = Permission.get_by_key(modify_to)
         assert old is not None
         assert new is not None
         # now modify permissions
         p = UserToPerm.query() \
                 .filter(UserToPerm.user == self.u1) \
                 .filter(UserToPerm.permission == old) \
                 .one()
         p.permission = new
         Session().commit()
         PermissionModel().create_default_permissions(user=self.u1)
         self._test_def_perm_equal(user=self.u1)

0 comments (0 inline, 0 general)