forms.py
347 lines
| 14.5 KiB
| text/x-python
|
PythonLexer
|
r0 | """ this is forms validation classes | ||
http://formencode.org/module-formencode.validators.html | ||||
for list off all availible validators | ||||
we can create our own validators | ||||
The table below outlines the options which can be used in a schema in addition to the validators themselves | ||||
pre_validators [] These validators will be applied before the schema | ||||
chained_validators [] These validators will be applied after the schema | ||||
allow_extra_fields False If True, then it is not an error when keys that aren't associated with a validator are present | ||||
filter_extra_fields False If True, then keys that aren't associated with a validator are removed | ||||
if_key_missing NoDefault If this is given, then any keys that aren't available but are expected will be replaced with this value (and then validated). This does not override a present .if_missing attribute on validators. NoDefault is a special FormEncode class to mean that no default values has been specified and therefore missing keys shouldn't take a default value. | ||||
ignore_key_missing False If True, then missing keys will be missing in the result, if the validator doesn't have .if_missing on it already | ||||
<name> = formencode.validators.<name of validator> | ||||
<name> must equal form name | ||||
list=[1,2,3,4,5] | ||||
r186 | for SELECT use formencode.All(OneOf(list), Int()) | |||
|
r0 | |||
""" | ||||
r242 | from formencode import All | |||
r238 | from formencode.validators import UnicodeString, OneOf, Int, Number, Regex, \ | |||
Email, Bool, StringBoolean | ||||
r186 | from pylons import session | |||
from pylons.i18n.translation import _ | ||||
r416 | from pylons_app.lib.auth import check_password, get_crypt_password | |||
r186 | from pylons_app.model import meta | |||
r442 | from pylons_app.model.user_model import UserModel | |||
r265 | from pylons_app.model.db import User, Repository | |||
r186 | from sqlalchemy.exc import OperationalError | |||
from sqlalchemy.orm.exc import NoResultFound, MultipleResultsFound | ||||
from webhelpers.pylonslib.secure_form import authentication_token | ||||
|
r0 | import formencode | ||
r186 | import logging | |||
r388 | import os | |||
import pylons_app.lib.helpers as h | ||||
r186 | log = logging.getLogger(__name__) | |||
|
r0 | |||
r186 | #this is needed to translate the messages using _() in validators | |||
class State_obj(object): | ||||
_ = staticmethod(_) | ||||
#=============================================================================== | ||||
# VALIDATORS | ||||
#=============================================================================== | ||||
|
r0 | class ValidAuthToken(formencode.validators.FancyValidator): | ||
messages = {'invalid_token':_('Token mismatch')} | ||||
def validate_python(self, value, state): | ||||
if value != authentication_token(): | ||||
r186 | raise formencode.Invalid(self.message('invalid_token', state, | |||
search_number=value), value, state) | ||||
r357 | ||||
def ValidUsername(edit, old_data): | ||||
class _ValidUsername(formencode.validators.FancyValidator): | ||||
def validate_python(self, value, state): | ||||
if value in ['default', 'new_user']: | ||||
raise formencode.Invalid(_('Invalid username'), value, state) | ||||
#check if user is uniq | ||||
sa = meta.Session | ||||
old_un = None | ||||
if edit: | ||||
old_un = sa.query(User).get(old_data.get('user_id')).username | ||||
if old_un != value or not edit: | ||||
if sa.query(User).filter(User.username == value).scalar(): | ||||
raise formencode.Invalid(_('This username already exists') , | ||||
value, state) | ||||
meta.Session.remove() | ||||
return _ValidUsername | ||||
r238 | ||||
class ValidPassword(formencode.validators.FancyValidator): | ||||
def to_python(self, value, state): | ||||
r347 | if value: | |||
return get_crypt_password(value) | ||||
r238 | ||||
r186 | class ValidAuth(formencode.validators.FancyValidator): | |||
messages = { | ||||
'invalid_password':_('invalid password'), | ||||
'invalid_login':_('invalid user name'), | ||||
'disabled_account':_('Your acccount is disabled') | ||||
} | ||||
#error mapping | ||||
e_dict = {'username':messages['invalid_login'], | ||||
'password':messages['invalid_password']} | ||||
r227 | e_dict_disable = {'username':messages['disabled_account']} | |||
r186 | ||||
def validate_python(self, value, state): | ||||
r415 | password = value['password'] | |||
r186 | username = value['username'] | |||
r458 | user = UserModel().get_user_by_name(username) | |||
if user is None: | ||||
r238 | raise formencode.Invalid(self.message('invalid_password', | |||
state=State_obj), value, state, | ||||
error_dict=self.e_dict) | ||||
r186 | if user: | |||
if user.active: | ||||
r474 | if user.username == username and check_password(password, | |||
r442 | user.password): | |||
r186 | return value | |||
else: | ||||
log.warning('user %s not authenticated', username) | ||||
raise formencode.Invalid(self.message('invalid_password', | ||||
state=State_obj), value, state, | ||||
error_dict=self.e_dict) | ||||
else: | ||||
log.warning('user %s is disabled', username) | ||||
raise formencode.Invalid(self.message('disabled_account', | ||||
state=State_obj), | ||||
r227 | value, state, | |||
error_dict=self.e_dict_disable) | ||||
r356 | ||||
r265 | class ValidRepoUser(formencode.validators.FancyValidator): | |||
def to_python(self, value, state): | ||||
try: | ||||
r442 | self.user_db = meta.Session.query(User)\ | |||
r328 | .filter(User.active == True)\ | |||
.filter(User.username == value).one() | ||||
r265 | except Exception: | |||
raise formencode.Invalid(_('This username is not valid'), | ||||
value, state) | ||||
r442 | finally: | |||
meta.Session.remove() | ||||
r265 | return self.user_db.user_id | |||
r356 | def ValidRepoName(edit, old_data): | |||
r265 | class _ValidRepoName(formencode.validators.FancyValidator): | |||
def to_python(self, value, state): | ||||
slug = h.repo_name_slug(value) | ||||
r310 | if slug in ['_admin']: | |||
raise formencode.Invalid(_('This repository name is disallowed'), | ||||
value, state) | ||||
r356 | if old_data.get('repo_name') != value or not edit: | |||
sa = meta.Session | ||||
r367 | if sa.query(Repository).filter(Repository.repo_name == slug).scalar(): | |||
r356 | raise formencode.Invalid(_('This repository already exists') , | |||
value, state) | ||||
meta.Session.remove() | ||||
r265 | return slug | |||
r356 | ||||
r265 | return _ValidRepoName | |||
r296 | ||||
class ValidPerms(formencode.validators.FancyValidator): | ||||
messages = {'perm_new_user_name':_('This username is not valid')} | ||||
def to_python(self, value, state): | ||||
perms_update = [] | ||||
perms_new = [] | ||||
#build a list of permission to update and new permission to create | ||||
for k, v in value.items(): | ||||
if k.startswith('perm_'): | ||||
if k.startswith('perm_new_user'): | ||||
new_perm = value.get('perm_new_user', False) | ||||
new_user = value.get('perm_new_user_name', False) | ||||
if new_user and new_perm: | ||||
if (new_user, new_perm) not in perms_new: | ||||
perms_new.append((new_user, new_perm)) | ||||
else: | ||||
r299 | usr = k[5:] | |||
if usr == 'default': | ||||
if value['private']: | ||||
#set none for default when updating to private repo | ||||
v = 'repository.none' | ||||
perms_update.append((usr, v)) | ||||
r296 | value['perms_updates'] = perms_update | |||
value['perms_new'] = perms_new | ||||
sa = meta.Session | ||||
for k, v in perms_new: | ||||
try: | ||||
r328 | self.user_db = sa.query(User)\ | |||
.filter(User.active == True)\ | ||||
.filter(User.username == k).one() | ||||
r296 | except Exception: | |||
msg = self.message('perm_new_user_name', | ||||
state=State_obj) | ||||
raise formencode.Invalid(msg, value, state, error_dict={'perm_new_user_name':msg}) | ||||
return value | ||||
r320 | ||||
class ValidSettings(formencode.validators.FancyValidator): | ||||
def to_python(self, value, state): | ||||
#settings form can't edit user | ||||
if value.has_key('user'): | ||||
del['value']['user'] | ||||
r388 | return value | |||
class ValidPath(formencode.validators.FancyValidator): | ||||
def to_python(self, value, state): | ||||
isdir = os.path.isdir(value.replace('*', '')) | ||||
if (value.endswith('/*') or value.endswith('/**')) and isdir: | ||||
return value | ||||
elif not isdir: | ||||
msg = _('This is not a valid path') | ||||
else: | ||||
msg = _('You need to specify * or ** at the end of path (ie. /tmp/*)') | ||||
raise formencode.Invalid(msg, value, state, | ||||
error_dict={'paths_root_path':msg}) | ||||
r474 | ||||
r475 | class UniqSystemEmail(formencode.validators.FancyValidator): | |||
def to_python(self, value, state): | ||||
sa = meta.Session | ||||
try: | ||||
user = sa.query(User).filter(User.email == value).scalar() | ||||
if user: | ||||
raise formencode.Invalid(_("That e-mail address is already taken") , | ||||
value, state) | ||||
finally: | ||||
meta.Session.remove() | ||||
return value | ||||
r474 | class ValidSystemEmail(formencode.validators.FancyValidator): | |||
def to_python(self, value, state): | ||||
sa = meta.Session | ||||
try: | ||||
user = sa.query(User).filter(User.email == value).scalar() | ||||
if user is None: | ||||
raise formencode.Invalid(_("That e-mail address doesn't exist.") , | ||||
value, state) | ||||
finally: | ||||
meta.Session.remove() | ||||
return value | ||||
r186 | #=============================================================================== | |||
# FORMS | ||||
#=============================================================================== | ||||
r45 | class LoginForm(formencode.Schema): | |||
allow_extra_fields = True | ||||
filter_extra_fields = True | ||||
username = UnicodeString( | ||||
strip=True, | ||||
min=3, | ||||
not_empty=True, | ||||
messages={ | ||||
'empty':_('Please enter a login'), | ||||
'tooShort':_('Enter a value %(min)i characters long or more')} | ||||
) | ||||
|
r0 | |||
r45 | password = UnicodeString( | |||
strip=True, | ||||
min=3, | ||||
not_empty=True, | ||||
messages={ | ||||
'empty':_('Please enter a password'), | ||||
'tooShort':_('Enter a value %(min)i characters long or more')} | ||||
) | ||||
|
r0 | |||
r186 | #chained validators have access to all data | |||
chained_validators = [ValidAuth] | ||||
r357 | def UserForm(edit=False, old_data={}): | |||
r238 | class _UserForm(formencode.Schema): | |||
allow_extra_fields = True | ||||
filter_extra_fields = True | ||||
r357 | username = All(UnicodeString(strip=True, min=3, not_empty=True), ValidUsername(edit, old_data)) | |||
r238 | if edit: | |||
new_password = All(UnicodeString(strip=True, min=3, not_empty=False), ValidPassword) | ||||
r329 | admin = StringBoolean(if_missing=False) | |||
r238 | else: | |||
r357 | password = All(UnicodeString(strip=True, min=8, not_empty=True), ValidPassword) | |||
r238 | active = StringBoolean(if_missing=False) | |||
name = UnicodeString(strip=True, min=3, not_empty=True) | ||||
lastname = UnicodeString(strip=True, min=3, not_empty=True) | ||||
r475 | email = All(Email(not_empty=True), UniqSystemEmail()) | |||
r238 | ||||
return _UserForm | ||||
r265 | ||||
r363 | RegisterForm = UserForm | |||
r474 | ||||
def PasswordResetForm(): | ||||
class _PasswordResetForm(formencode.Schema): | ||||
allow_extra_fields = True | ||||
filter_extra_fields = True | ||||
email = All(ValidSystemEmail(), Email(not_empty=True)) | ||||
return _PasswordResetForm | ||||
r356 | def RepoForm(edit=False, old_data={}): | |||
r265 | class _RepoForm(formencode.Schema): | |||
allow_extra_fields = True | ||||
r296 | filter_extra_fields = False | |||
r356 | repo_name = All(UnicodeString(strip=True, min=1, not_empty=True), ValidRepoName(edit, old_data)) | |||
r265 | description = UnicodeString(strip=True, min=3, not_empty=True) | |||
private = StringBoolean(if_missing=False) | ||||
if edit: | ||||
user = All(Int(not_empty=True), ValidRepoUser) | ||||
r296 | chained_validators = [ValidPerms] | |||
r265 | return _RepoForm | |||
r320 | ||||
r356 | def RepoSettingsForm(edit=False, old_data={}): | |||
r320 | class _RepoForm(formencode.Schema): | |||
allow_extra_fields = True | ||||
filter_extra_fields = False | ||||
r356 | repo_name = All(UnicodeString(strip=True, min=1, not_empty=True), ValidRepoName(edit, old_data)) | |||
r320 | description = UnicodeString(strip=True, min=3, not_empty=True) | |||
private = StringBoolean(if_missing=False) | ||||
chained_validators = [ValidPerms, ValidSettings] | ||||
return _RepoForm | ||||
r350 | def ApplicationSettingsForm(): | |||
class _ApplicationSettingsForm(formencode.Schema): | ||||
allow_extra_fields = True | ||||
filter_extra_fields = False | ||||
r381 | hg_app_title = UnicodeString(strip=True, min=3, not_empty=True) | |||
hg_app_realm = UnicodeString(strip=True, min=3, not_empty=True) | ||||
r350 | ||||
return _ApplicationSettingsForm | ||||
r388 | def ApplicationUiSettingsForm(): | |||
class _ApplicationUiSettingsForm(formencode.Schema): | ||||
allow_extra_fields = True | ||||
filter_extra_fields = False | ||||
web_push_ssl = OneOf(['true', 'false'], if_missing='false') | ||||
paths_root_path = All(ValidPath(), UnicodeString(strip=True, min=3, not_empty=True)) | ||||
r395 | hooks_changegroup_update = OneOf(['True', 'False'], if_missing=False) | |||
hooks_changegroup_repo_size = OneOf(['True', 'False'], if_missing=False) | ||||
r388 | ||||
return _ApplicationUiSettingsForm | ||||
r320 | ||||
r417 | def DefaultPermissionsForm(perms_choices, register_choices, create_choices): | |||
class _DefaultPermissionsForm(formencode.Schema): | ||||
allow_extra_fields = True | ||||
filter_extra_fields = True | ||||
r418 | overwrite_default = OneOf(['true', 'false'], if_missing='false') | |||
r417 | default_perm = OneOf(perms_choices) | |||
default_register = OneOf(register_choices) | ||||
default_create = OneOf(create_choices) | ||||
return _DefaultPermissionsForm | ||||