##// END OF EJS Templates
validators: fix url_validator tests and make it flag controllable.
validators: fix url_validator tests and make it flag controllable.

File last commit:

r2988:390bbe68 default
r3072:fe39713b default
Show More
permission.py
558 lines | 22.7 KiB | text/x-python | PythonLexer
project: added all source files and assets
r1 # -*- coding: utf-8 -*-
release: update copyright year to 2018
r2487 # Copyright (C) 2010-2018 RhodeCode GmbH
project: added all source files and assets
r1 #
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
"""
permissions model for RhodeCode
"""
import logging
import traceback
from sqlalchemy.exc import DatabaseError
from rhodecode.model import BaseModel
from rhodecode.model.db import (
User, Permission, UserToPerm, UserRepoToPerm, UserRepoGroupToPerm,
branch permissions: added logic to define in UI branch permissions....
r2975 UserUserGroupToPerm, UserGroup, UserGroupToPerm, UserToRepoBranchPermission)
project: added all source files and assets
r1 from rhodecode.lib.utils2 import str2bool, safe_int
log = logging.getLogger(__name__)
class PermissionModel(BaseModel):
"""
Permissions model for RhodeCode
"""
cls = Permission
global_perms = {
'default_repo_create': None,
# special case for create repos on write access to group
'default_repo_create_on_write': None,
'default_repo_group_create': None,
'default_user_group_create': None,
'default_fork_create': None,
'default_inherit_default_permissions': None,
'default_register': None,
settings: fix #3944 add password reset permission
r1034 'default_password_reset': None,
project: added all source files and assets
r1 'default_extern_activate': None,
# object permissions below
'default_repo_perm': None,
'default_group_perm': None,
'default_user_group_perm': None,
branch permissions: added logic to define in UI branch permissions....
r2975
# branch
'default_branch_perm': None,
project: added all source files and assets
r1 }
i18n: expose custom name to catch the translation strings.
r1098 def set_global_permission_choices(self, c_obj, gettext_translator):
global-permissions: ported controller to pyramid view....
r1941 _ = gettext_translator
i18n: expose custom name to catch the translation strings.
r1098
project: added all source files and assets
r1 c_obj.repo_perms_choices = [
global-permissions: ported controller to pyramid view....
r1941 ('repository.none', _('None'),),
('repository.read', _('Read'),),
('repository.write', _('Write'),),
('repository.admin', _('Admin'),)]
project: added all source files and assets
r1
c_obj.group_perms_choices = [
global-permissions: ported controller to pyramid view....
r1941 ('group.none', _('None'),),
('group.read', _('Read'),),
('group.write', _('Write'),),
('group.admin', _('Admin'),)]
project: added all source files and assets
r1
c_obj.user_group_perms_choices = [
global-permissions: ported controller to pyramid view....
r1941 ('usergroup.none', _('None'),),
('usergroup.read', _('Read'),),
('usergroup.write', _('Write'),),
('usergroup.admin', _('Admin'),)]
project: added all source files and assets
r1
branch permissions: added logic to define in UI branch permissions....
r2975 c_obj.branch_perms_choices = [
('branch.none', _('Protected/No Access'),),
('branch.merge', _('Web merge'),),
('branch.push', _('Push'),),
('branch.push_force', _('Force Push'),)]
project: added all source files and assets
r1 c_obj.register_choices = [
global-permissions: ported controller to pyramid view....
r1941 ('hg.register.none', _('Disabled')),
('hg.register.manual_activate', _('Allowed with manual account activation')),
('hg.register.auto_activate', _('Allowed with automatic account activation')),]
project: added all source files and assets
r1
settings: fix #3944 add password reset permission
r1034 c_obj.password_reset_choices = [
global-permissions: ported controller to pyramid view....
r1941 ('hg.password_reset.enabled', _('Allow password recovery')),
('hg.password_reset.hidden', _('Hide password recovery link')),
('hg.password_reset.disabled', _('Disable password recovery')),]
settings: fix #3944 add password reset permission
r1034
project: added all source files and assets
r1 c_obj.extern_activate_choices = [
global-permissions: ported controller to pyramid view....
r1941 ('hg.extern_activate.manual', _('Manual activation of external account')),
('hg.extern_activate.auto', _('Automatic activation of external account')),]
project: added all source files and assets
r1
c_obj.repo_create_choices = [
global-permissions: ported controller to pyramid view....
r1941 ('hg.create.none', _('Disabled')),
('hg.create.repository', _('Enabled'))]
project: added all source files and assets
r1
c_obj.repo_create_on_write_choices = [
global-permissions: ported controller to pyramid view....
r1941 ('hg.create.write_on_repogroup.false', _('Disabled')),
('hg.create.write_on_repogroup.true', _('Enabled'))]
project: added all source files and assets
r1
c_obj.user_group_create_choices = [
global-permissions: ported controller to pyramid view....
r1941 ('hg.usergroup.create.false', _('Disabled')),
('hg.usergroup.create.true', _('Enabled'))]
project: added all source files and assets
r1
c_obj.repo_group_create_choices = [
global-permissions: ported controller to pyramid view....
r1941 ('hg.repogroup.create.false', _('Disabled')),
('hg.repogroup.create.true', _('Enabled'))]
project: added all source files and assets
r1
c_obj.fork_choices = [
global-permissions: ported controller to pyramid view....
r1941 ('hg.fork.none', _('Disabled')),
('hg.fork.repository', _('Enabled'))]
project: added all source files and assets
r1
c_obj.inherit_default_permission_choices = [
global-permissions: ported controller to pyramid view....
r1941 ('hg.inherit_default_perms.false', _('Disabled')),
('hg.inherit_default_perms.true', _('Enabled'))]
project: added all source files and assets
r1
def get_default_perms(self, object_perms, suffix):
defaults = {}
for perm in object_perms:
# perms
if perm.permission.permission_name.startswith('repository.'):
defaults['default_repo_perm' + suffix] = perm.permission.permission_name
if perm.permission.permission_name.startswith('group.'):
defaults['default_group_perm' + suffix] = perm.permission.permission_name
if perm.permission.permission_name.startswith('usergroup.'):
defaults['default_user_group_perm' + suffix] = perm.permission.permission_name
branch permissions: added logic to define in UI branch permissions....
r2975 # branch
if perm.permission.permission_name.startswith('branch.'):
defaults['default_branch_perm' + suffix] = perm.permission.permission_name
project: added all source files and assets
r1 # creation of objects
if perm.permission.permission_name.startswith('hg.create.write_on_repogroup'):
defaults['default_repo_create_on_write' + suffix] = perm.permission.permission_name
elif perm.permission.permission_name.startswith('hg.create.'):
defaults['default_repo_create' + suffix] = perm.permission.permission_name
if perm.permission.permission_name.startswith('hg.fork.'):
defaults['default_fork_create' + suffix] = perm.permission.permission_name
if perm.permission.permission_name.startswith('hg.inherit_default_perms.'):
defaults['default_inherit_default_permissions' + suffix] = perm.permission.permission_name
if perm.permission.permission_name.startswith('hg.repogroup.'):
defaults['default_repo_group_create' + suffix] = perm.permission.permission_name
if perm.permission.permission_name.startswith('hg.usergroup.'):
defaults['default_user_group_create' + suffix] = perm.permission.permission_name
# registration and external account activation
if perm.permission.permission_name.startswith('hg.register.'):
defaults['default_register' + suffix] = perm.permission.permission_name
settings: fix #3944 add password reset permission
r1034 if perm.permission.permission_name.startswith('hg.password_reset.'):
defaults['default_password_reset' + suffix] = perm.permission.permission_name
project: added all source files and assets
r1 if perm.permission.permission_name.startswith('hg.extern_activate.'):
defaults['default_extern_activate' + suffix] = perm.permission.permission_name
return defaults
def _make_new_user_perm(self, user, perm_name):
log.debug('Creating new user permission:%s', perm_name)
new = UserToPerm()
new.user = user
new.permission = Permission.get_by_key(perm_name)
return new
def _make_new_user_group_perm(self, user_group, perm_name):
log.debug('Creating new user group permission:%s', perm_name)
new = UserGroupToPerm()
new.users_group = user_group
new.permission = Permission.get_by_key(perm_name)
return new
def _keep_perm(self, perm_name, keep_fields):
def get_pat(field_name):
return {
# global perms
'default_repo_create': 'hg.create.',
# special case for create repos on write access to group
'default_repo_create_on_write': 'hg.create.write_on_repogroup.',
'default_repo_group_create': 'hg.repogroup.create.',
'default_user_group_create': 'hg.usergroup.create.',
'default_fork_create': 'hg.fork.',
'default_inherit_default_permissions': 'hg.inherit_default_perms.',
# application perms
'default_register': 'hg.register.',
settings: fix #3944 add password reset permission
r1034 'default_password_reset': 'hg.password_reset.',
project: added all source files and assets
r1 'default_extern_activate': 'hg.extern_activate.',
# object permissions below
'default_repo_perm': 'repository.',
'default_group_perm': 'group.',
'default_user_group_perm': 'usergroup.',
branch permissions: added logic to define in UI branch permissions....
r2975 # branch
'default_branch_perm': 'branch.',
project: added all source files and assets
r1 }[field_name]
for field in keep_fields:
pat = get_pat(field)
if perm_name.startswith(pat):
return True
return False
def _clear_object_perm(self, object_perms, preserve=None):
preserve = preserve or []
_deleted = []
for perm in object_perms:
perm_name = perm.permission.permission_name
if not self._keep_perm(perm_name, keep_fields=preserve):
_deleted.append(perm_name)
self.sa.delete(perm)
return _deleted
def _clear_user_perms(self, user_id, preserve=None):
perms = self.sa.query(UserToPerm)\
.filter(UserToPerm.user_id == user_id)\
.all()
return self._clear_object_perm(perms, preserve=preserve)
def _clear_user_group_perms(self, user_group_id, preserve=None):
perms = self.sa.query(UserGroupToPerm)\
.filter(UserGroupToPerm.users_group_id == user_group_id)\
.all()
return self._clear_object_perm(perms, preserve=preserve)
def _set_new_object_perms(self, obj_type, object, form_result, preserve=None):
# clear current entries, to make this function idempotent
# it will fix even if we define more permissions or permissions
# are somehow missing
preserve = preserve or []
_global_perms = self.global_perms.copy()
if obj_type not in ['user', 'user_group']:
raise ValueError("obj_type must be on of 'user' or 'user_group'")
branch permissions: added logic to define in UI branch permissions....
r2975 global_perms = len(_global_perms)
default_user_perms = len(Permission.DEFAULT_USER_PERMISSIONS)
if global_perms != default_user_perms:
raise Exception(
'Inconsistent permissions definition. Got {} vs {}'.format(
global_perms, default_user_perms))
project: added all source files and assets
r1
if obj_type == 'user':
self._clear_user_perms(object.user_id, preserve)
if obj_type == 'user_group':
self._clear_user_group_perms(object.users_group_id, preserve)
# now kill the keys that we want to preserve from the form.
for key in preserve:
del _global_perms[key]
for k in _global_perms.copy():
_global_perms[k] = form_result[k]
# at that stage we validate all are passed inside form_result
for _perm_key, perm_value in _global_perms.items():
if perm_value is None:
raise ValueError('Missing permission for %s' % (_perm_key,))
if obj_type == 'user':
p = self._make_new_user_perm(object, perm_value)
self.sa.add(p)
if obj_type == 'user_group':
p = self._make_new_user_group_perm(object, perm_value)
self.sa.add(p)
def _set_new_user_perms(self, user, form_result, preserve=None):
return self._set_new_object_perms(
'user', user, form_result, preserve)
def _set_new_user_group_perms(self, user_group, form_result, preserve=None):
return self._set_new_object_perms(
'user_group', user_group, form_result, preserve)
def set_new_user_perms(self, user, form_result):
# calculate what to preserve from what is given in form_result
preserve = set(self.global_perms.keys()).difference(set(form_result.keys()))
return self._set_new_user_perms(user, form_result, preserve)
def set_new_user_group_perms(self, user_group, form_result):
# calculate what to preserve from what is given in form_result
preserve = set(self.global_perms.keys()).difference(set(form_result.keys()))
return self._set_new_user_group_perms(user_group, form_result, preserve)
def create_permissions(self):
"""
Create permissions for whole system
"""
for p in Permission.PERMS:
if not Permission.get_by_key(p[0]):
new_perm = Permission()
new_perm.permission_name = p[0]
new_perm.permission_longname = p[0] # translation err with p[1]
self.sa.add(new_perm)
def _create_default_object_permission(self, obj_type, obj, obj_perms,
force=False):
if obj_type not in ['user', 'user_group']:
raise ValueError("obj_type must be on of 'user' or 'user_group'")
def _get_group(perm_name):
return '.'.join(perm_name.split('.')[:1])
defined_perms_groups = map(
_get_group, (x.permission.permission_name for x in obj_perms))
log.debug('GOT ALREADY DEFINED:%s', obj_perms)
if force:
self._clear_object_perm(obj_perms)
self.sa.commit()
defined_perms_groups = []
# for every default permission that needs to be created, we check if
# it's group is already defined, if it's not we create default perm
for perm_name in Permission.DEFAULT_USER_PERMISSIONS:
gr = _get_group(perm_name)
if gr not in defined_perms_groups:
log.debug('GR:%s not found, creating permission %s',
gr, perm_name)
if obj_type == 'user':
new_perm = self._make_new_user_perm(obj, perm_name)
self.sa.add(new_perm)
if obj_type == 'user_group':
new_perm = self._make_new_user_group_perm(obj, perm_name)
self.sa.add(new_perm)
def create_default_user_permissions(self, user, force=False):
"""
Creates only missing default permissions for user, if force is set it
resets the default permissions for that user
:param user:
:param force:
"""
user = self._get_user(user)
obj_perms = UserToPerm.query().filter(UserToPerm.user == user).all()
return self._create_default_object_permission(
'user', user, obj_perms, force)
def create_default_user_group_permissions(self, user_group, force=False):
"""
branch permissions: added logic to define in UI branch permissions....
r2975 Creates only missing default permissions for user group, if force is
set it resets the default permissions for that user group
project: added all source files and assets
r1
:param user_group:
:param force:
"""
user_group = self._get_user_group(user_group)
obj_perms = UserToPerm.query().filter(UserGroupToPerm.users_group == user_group).all()
return self._create_default_object_permission(
'user_group', user_group, obj_perms, force)
def update_application_permissions(self, form_result):
if 'perm_user_id' in form_result:
perm_user = User.get(safe_int(form_result['perm_user_id']))
else:
# used mostly to do lookup for default user
perm_user = User.get_by_username(form_result['perm_user_name'])
try:
# stage 1 set anonymous access
if perm_user.username == User.DEFAULT_USER:
perm_user.active = str2bool(form_result['anonymous'])
self.sa.add(perm_user)
# stage 2 reset defaults and set them from form data
self._set_new_user_perms(perm_user, form_result, preserve=[
'default_repo_perm',
'default_group_perm',
'default_user_group_perm',
branch permissions: added logic to define in UI branch permissions....
r2975 'default_branch_perm',
project: added all source files and assets
r1
'default_repo_group_create',
'default_user_group_create',
'default_repo_create_on_write',
'default_repo_create',
'default_fork_create',
'default_inherit_default_permissions',])
self.sa.commit()
except (DatabaseError,):
log.error(traceback.format_exc())
self.sa.rollback()
raise
def update_user_permissions(self, form_result):
if 'perm_user_id' in form_result:
perm_user = User.get(safe_int(form_result['perm_user_id']))
else:
# used mostly to do lookup for default user
perm_user = User.get_by_username(form_result['perm_user_name'])
try:
# stage 2 reset defaults and set them from form data
self._set_new_user_perms(perm_user, form_result, preserve=[
'default_repo_perm',
'default_group_perm',
'default_user_group_perm',
branch permissions: added logic to define in UI branch permissions....
r2975 'default_branch_perm',
project: added all source files and assets
r1
'default_register',
settings: fix #3944 add password reset permission
r1034 'default_password_reset',
project: added all source files and assets
r1 'default_extern_activate'])
self.sa.commit()
except (DatabaseError,):
log.error(traceback.format_exc())
self.sa.rollback()
raise
def update_user_group_permissions(self, form_result):
if 'perm_user_group_id' in form_result:
perm_user_group = UserGroup.get(safe_int(form_result['perm_user_group_id']))
else:
# used mostly to do lookup for default user
perm_user_group = UserGroup.get_by_group_name(form_result['perm_user_group_name'])
try:
# stage 2 reset defaults and set them from form data
self._set_new_user_group_perms(perm_user_group, form_result, preserve=[
'default_repo_perm',
'default_group_perm',
'default_user_group_perm',
branch permissions: added logic to define in UI branch permissions....
r2975 'default_branch_perm',
project: added all source files and assets
r1
'default_register',
settings: fix #3944 add password reset permission
r1034 'default_password_reset',
project: added all source files and assets
r1 'default_extern_activate'])
self.sa.commit()
except (DatabaseError,):
log.error(traceback.format_exc())
self.sa.rollback()
raise
def update_object_permissions(self, form_result):
if 'perm_user_id' in form_result:
perm_user = User.get(safe_int(form_result['perm_user_id']))
else:
# used mostly to do lookup for default user
perm_user = User.get_by_username(form_result['perm_user_name'])
try:
# stage 2 reset defaults and set them from form data
self._set_new_user_perms(perm_user, form_result, preserve=[
'default_repo_group_create',
'default_user_group_create',
'default_repo_create_on_write',
'default_repo_create',
'default_fork_create',
'default_inherit_default_permissions',
branch permissions: added logic to define in UI branch permissions....
r2975 'default_branch_perm',
project: added all source files and assets
r1
'default_register',
settings: fix #3944 add password reset permission
r1034 'default_password_reset',
project: added all source files and assets
r1 'default_extern_activate'])
# overwrite default repo permissions
if form_result['overwrite_default_repo']:
_def_name = form_result['default_repo_perm'].split('repository.')[-1]
_def = Permission.get_by_key('repository.' + _def_name)
for r2p in self.sa.query(UserRepoToPerm)\
.filter(UserRepoToPerm.user == perm_user)\
.all():
# don't reset PRIVATE repositories
if not r2p.repository.private:
r2p.permission = _def
self.sa.add(r2p)
# overwrite default repo group permissions
if form_result['overwrite_default_group']:
_def_name = form_result['default_group_perm'].split('group.')[-1]
_def = Permission.get_by_key('group.' + _def_name)
for g2p in self.sa.query(UserRepoGroupToPerm)\
.filter(UserRepoGroupToPerm.user == perm_user)\
.all():
g2p.permission = _def
self.sa.add(g2p)
# overwrite default user group permissions
if form_result['overwrite_default_user_group']:
_def_name = form_result['default_user_group_perm'].split('usergroup.')[-1]
# user groups
_def = Permission.get_by_key('usergroup.' + _def_name)
for g2p in self.sa.query(UserUserGroupToPerm)\
.filter(UserUserGroupToPerm.user == perm_user)\
.all():
g2p.permission = _def
self.sa.add(g2p)
branch permissions: added logic to define in UI branch permissions....
r2975
# COMMIT
project: added all source files and assets
r1 self.sa.commit()
except (DatabaseError,):
log.exception('Failed to set default object permissions')
self.sa.rollback()
raise
branch permissions: added logic to define in UI branch permissions....
r2975
def update_branch_permissions(self, form_result):
if 'perm_user_id' in form_result:
perm_user = User.get(safe_int(form_result['perm_user_id']))
else:
# used mostly to do lookup for default user
perm_user = User.get_by_username(form_result['perm_user_name'])
try:
# stage 2 reset defaults and set them from form data
self._set_new_user_perms(perm_user, form_result, preserve=[
'default_repo_perm',
'default_group_perm',
'default_user_group_perm',
'default_repo_group_create',
'default_user_group_create',
'default_repo_create_on_write',
'default_repo_create',
'default_fork_create',
'default_inherit_default_permissions',
'default_register',
'default_password_reset',
'default_extern_activate'])
# overwrite default branch permissions
if form_result['overwrite_default_branch']:
_def_name = \
form_result['default_branch_perm'].split('branch.')[-1]
_def = Permission.get_by_key('branch.' + _def_name)
branch-permissions: fixed override of default permissions for branches from admin page
r2988 user_perms = UserToRepoBranchPermission.query()\
.join(UserToRepoBranchPermission.user_repo_to_perm)\
.filter(UserRepoToPerm.user == perm_user).all()
for g2p in user_perms:
branch permissions: added logic to define in UI branch permissions....
r2975 g2p.permission = _def
self.sa.add(g2p)
# COMMIT
self.sa.commit()
except (DatabaseError,):
log.exception('Failed to set default branch permissions')
self.sa.rollback()
raise