##// END OF EJS Templates
caches: use context based cache for repository....
caches: use context based cache for repository. - due to gevent sharing connections problem we have to make the cache key per each green thread.

File last commit:

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