Show More
validators.py
1115 lines
| 42.4 KiB
| text/x-python
|
PythonLexer
r1 | # -*- coding: utf-8 -*- | |||
r2487 | # Copyright (C) 2010-2018 RhodeCode GmbH | |||
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 | ||||
""" | ||||
r2351 | ||||
r1 | import os | |||
import re | ||||
r2351 | import logging | |||
import collections | ||||
r1 | ||||
import formencode | ||||
import ipaddress | ||||
from formencode.validators import ( | ||||
UnicodeString, OneOf, Int, Number, Regex, Email, Bool, StringBoolean, Set, | ||||
NotEmpty, IPAddress, CIDR, String, FancyValidator | ||||
) | ||||
r2351 | ||||
r1 | from sqlalchemy.sql.expression import true | |||
from sqlalchemy.util import OrderedSet | ||||
r136 | from rhodecode.authentication import ( | |||
legacy_plugin_prefix, _import_legacy_plugin) | ||||
from rhodecode.authentication.base import loadplugin | ||||
r2309 | from rhodecode.apps._base import ADMIN_PREFIX | |||
r1 | from rhodecode.lib.auth import HasRepoGroupPermissionAny, HasPermissionAny | |||
from rhodecode.lib.utils import repo_name_slug, make_db_config | ||||
r1907 | from rhodecode.lib.utils2 import safe_int, str2bool, aslist, md5, safe_unicode | |||
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 | ||||
r2351 | ||||
r1 | Missing = _Missing() | |||
r2351 | def M(self, key, state, **kwargs): | |||
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: | ||||
""" | ||||
r2351 | ||||
#state._ = staticmethod(_) | ||||
r1 | # inject validator into state object | |||
return self.message(key, state, **kwargs) | ||||
r2351 | def UniqueList(localizer, convert=None): | |||
_ = localizer | ||||
class _validator(formencode.FancyValidator): | ||||
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 [] | ||||
r2351 | return _validator | |||
r1 | ||||
r2351 | def UniqueListFromString(localizer): | |||
_ = localizer | ||||
class _validator(UniqueList(localizer)): | ||||
r1 | def _to_python(self, value, state): | |||
if isinstance(value, basestring): | ||||
value = aslist(value, ',') | ||||
r2351 | return super(_validator, self)._to_python(value, state) | |||
return _validator | ||||
r1 | ||||
r2351 | def ValidSvnPattern(localizer, section, repo_name=None): | |||
_ = localizer | ||||
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 | ||||
r2351 | def ValidUsername(localizer, edit=False, old_data=None): | |||
_ = localizer | ||||
old_data = old_data or {} | ||||
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 | ||||
r2351 | def ValidRepoUser(localizer, allow_disabled=False): | |||
_ = localizer | ||||
r1 | ||||
class _validator(formencode.validators.FancyValidator): | ||||
messages = { | ||||
r224 | 'invalid_username': _(u'Username %(username)s is not valid'), | |||
'disabled_username': _(u'Username %(username)s is disabled') | ||||
r1 | } | |||
def validate_python(self, value, state): | ||||
try: | ||||
r224 | user = User.query().filter(User.username == value).one() | |||
r1 | except Exception: | |||
msg = M(self, 'invalid_username', state, username=value) | ||||
raise formencode.Invalid( | ||||
msg, value, state, error_dict={'username': msg} | ||||
) | ||||
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} | ||||
) | ||||
r1 | return _validator | |||
r2351 | def ValidUserGroup(localizer, edit=False, old_data=None): | |||
_ = localizer | ||||
old_data = old_data or {} | ||||
r1 | class _validator(formencode.validators.FancyValidator): | |||
messages = { | ||||
'invalid_group': _(u'Invalid user group name'), | ||||
r2068 | 'group_exist': _(u'User group `%(usergroup)s` already exists'), | |||
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 | ||||
r2351 | def ValidRepoGroup(localizer, edit=False, old_data=None, can_create_in_root=False): | |||
_ = localizer | ||||
old_data = old_data or {} | ||||
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 | ||||
r2351 | def ValidPassword(localizer): | |||
_ = localizer | ||||
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 | ||||
r2351 | def ValidPasswordsMatch( | |||
localizer, passwd='new_password', | ||||
passwd_confirmation='password_confirmation'): | ||||
_ = localizer | ||||
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 | ||||
r2351 | def ValidAuth(localizer): | |||
_ = localizer | ||||
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'] | ||||
r76 | if not authenticate(username, password, '', HTTP_TYPE, | |||
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: | ||||
r76 | log.warning('user `%s` failed to authenticate', username) | |||
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 | ||||
r2351 | def ValidRepoName(localizer, edit=False, old_data=None): | |||
old_data = old_data or {} | ||||
_ = localizer | ||||
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 | ||||
r2351 | def ValidForkName(localizer, *args, **kwargs): | |||
_ = localizer | ||||
return ValidRepoName(localizer, *args, **kwargs) | ||||
r1 | ||||
r2351 | def SlugifyName(localizer): | |||
_ = localizer | ||||
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 | ||||
r2351 | def CannotHaveGitSuffix(localizer): | |||
_ = localizer | ||||
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 | ||||
r2351 | def ValidCloneUri(localizer): | |||
_ = localizer | ||||
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) | ||||
r2351 | msg = M(self, 'invalid_clone_uri', state, rtype=repo_type, | |||
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') | ||||
r2351 | msg = M(self, 'clone_uri', state, rtype=repo_type) | |||
r1 | raise formencode.Invalid(msg, value, state, | |||
error_dict={'clone_uri': msg}) | ||||
return _validator | ||||
r2351 | def ValidForkType(localizer, old_data=None): | |||
_ = localizer | ||||
old_data = old_data or {} | ||||
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 | ||||
r2351 | def CanWriteGroup(localizer, old_data=None): | |||
_ = localizer | ||||
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 | ||||
r2351 | def ValidPerms(localizer, type_='repo'): | |||
_ = localizer | ||||
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 | ||||
r2351 | new_perms_group = collections.defaultdict(dict) | |||
del_perms_group = collections.defaultdict(dict) | ||||
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: | ||||
r1734 | perm_deletions.add( | |||
(del_member, None, del_type)) | ||||
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: | ||||
r1734 | perm_additions.add( | |||
(new_member, new_perm, new_type)) | ||||
r1 | ||||
# get updates of permissions | ||||
# (read the existing radio button states) | ||||
r1734 | default_user_id = User.get_default_user().user_id | |||
r2505 | ||||
r1 | for k, update_value in value.iteritems(): | |||
if k.startswith('u_perm_') or k.startswith('g_perm_'): | ||||
r2505 | obj_type = k[0] | |||
obj_id = k[7:] | ||||
r1 | update_type = {'u': 'user', | |||
r2505 | 'g': 'users_group'}[obj_type] | |||
r1734 | ||||
r2505 | if obj_type == 'u' and safe_int(obj_id) == default_user_id: | |||
r1 | if str2bool(value.get('repo_private')): | |||
r1734 | # prevent from updating default user permissions | |||
# when this repository is marked as private | ||||
r1 | update_value = EMPTY_PERM | |||
r1734 | perm_updates.add( | |||
r2505 | (obj_id, update_value, update_type)) | |||
r1734 | ||||
value['perm_additions'] = [] # propagated later | ||||
r1 | value['perm_updates'] = list(perm_updates) | |||
value['perm_deletions'] = list(perm_deletions) | ||||
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 | ||||
r1 | try: | |||
if member_type == 'user': | ||||
r1734 | User.query()\ | |||
r1 | .filter(User.active == true())\ | |||
.filter(User.user_id == member_id).one() | ||||
if member_type == 'users_group': | ||||
r1734 | UserGroup.query()\ | |||
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 | ||||
r2351 | def ValidPath(localizer): | |||
_ = localizer | ||||
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 | ||||
r2351 | def UniqSystemEmail(localizer, old_data=None): | |||
_ = localizer | ||||
old_data = old_data or {} | ||||
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 | ||||
r2351 | def ValidSystemEmail(localizer): | |||
_ = localizer | ||||
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 | ||||
r2351 | def NotReviewedRevisions(localizer, repo_id): | |||
_ = localizer | ||||
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 | ||||
r2351 | def ValidIp(localizer): | |||
_ = localizer | ||||
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) | ||||
r1907 | v = safe_unicode(v.strip()) | |||
r1 | net = ipaddress.ip_network(address=v, strict=False) | |||
return str(net) | ||||
def validate_python(self, value, state): | ||||
try: | ||||
r1907 | addr = safe_unicode(value.strip()) | |||
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 | ||||
r2351 | def FieldKey(localizer): | |||
_ = localizer | ||||
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 | ||||
r2351 | def ValidAuthPlugins(localizer): | |||
_ = localizer | ||||
r1 | class _validator(formencode.validators.FancyValidator): | |||
messages = { | ||||
'import_duplicate': _( | ||||
u'Plugins %(loaded)s and %(next_to_load)s ' | ||||
u'both export the same name'), | ||||
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"'), | ||||
r1 | } | |||
def _to_python(self, value, state): | ||||
# filter empty values | ||||
return filter(lambda s: s not in [None, ''], value) | ||||
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. | ||||
""" | ||||
r133 | try: | |||
r136 | plugin = _import_legacy_plugin(plugin_id) | |||
except Exception as e: | ||||
log.exception( | ||||
'Exception during import of auth legacy plugin "{}"' | ||||
.format(plugin_id)) | ||||
r2351 | msg = M(self, 'import_error', state, plugin_id=plugin_id) | |||
r136 | raise formencode.Invalid(msg, value, state) | |||
if not hasattr(plugin, 'includeme'): | ||||
r2351 | msg = M(self, 'missing_includeme', state, plugin_id=plugin_id) | |||
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: | ||||
r2351 | msg = M(self, 'no_plugin', state, plugin_id=plugin_id) | |||
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 | ||||
r1 | return _validator | |||
r2351 | def ValidPattern(localizer): | |||
_ = localizer | ||||
r1 | ||||
r2351 | class _validator(formencode.validators.FancyValidator): | |||
r2334 | messages = { | |||
'bad_format': _(u'Url must start with http or /'), | ||||
} | ||||
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: | ||||
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) | ||||
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 | ||||
r2351 | return _validator | |||