forms.py
242 lines
| 10.6 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 _ | ||||
from pylons_app.lib.auth import get_crypt_password | ||||
r265 | import pylons_app.lib.helpers as h | |||
r186 | from pylons_app.model import meta | |||
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 | ||||
r242 | import datetime | |||
|
r0 | import formencode | ||
r186 | import logging | |||
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) | ||||
r238 | class ValidUsername(formencode.validators.FancyValidator): | |||
|
r0 | |||
r238 | def validate_python(self, value, state): | |||
r296 | if value in ['default', 'new_user']: | |||
raise formencode.Invalid(_('Invalid username'), value, state) | ||||
r238 | ||||
class ValidPassword(formencode.validators.FancyValidator): | ||||
def to_python(self, value, state): | ||||
return get_crypt_password(value) | ||||
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): | ||||
sa = meta.Session | ||||
crypted_passwd = get_crypt_password(value['password']) | ||||
username = value['username'] | ||||
try: | ||||
r234 | user = sa.query(User).filter(User.username == username).one() | |||
r186 | except (NoResultFound, MultipleResultsFound, OperationalError) as e: | |||
log.error(e) | ||||
user = 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: | ||||
if user.username == username and user.password == crypted_passwd: | ||||
from pylons_app.lib.auth import AuthUser | ||||
auth_user = AuthUser() | ||||
auth_user.username = username | ||||
auth_user.is_authenticated = True | ||||
auth_user.is_admin = user.admin | ||||
r265 | auth_user.user_id = user.user_id | |||
r186 | session['hg_app_user'] = auth_user | |||
session.save() | ||||
r201 | log.info('user %s is now authenticated', username) | |||
r242 | ||||
try: | ||||
user.last_login = datetime.datetime.now() | ||||
sa.add(user) | ||||
sa.commit() | ||||
except (OperationalError) as e: | ||||
log.error(e) | ||||
sa.rollback() | ||||
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) | ||||
r186 | ||||
r265 | class ValidRepoUser(formencode.validators.FancyValidator): | |||
def to_python(self, value, state): | ||||
sa = meta.Session | ||||
try: | ||||
self.user_db = sa.query(User).filter(User.username == value).one() | ||||
except Exception: | ||||
raise formencode.Invalid(_('This username is not valid'), | ||||
value, state) | ||||
return self.user_db.user_id | ||||
def ValidRepoName(edit=False): | ||||
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) | ||||
r265 | sa = meta.Session | |||
if sa.query(Repository).get(slug) and not edit: | ||||
raise formencode.Invalid(_('This repository already exists'), | ||||
value, state) | ||||
return slug | ||||
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: | ||||
self.user_db = sa.query(User).filter(User.username == k).one() | ||||
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 | ||||
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] | ||||
r238 | def UserForm(edit=False): | |||
class _UserForm(formencode.Schema): | ||||
allow_extra_fields = True | ||||
filter_extra_fields = True | ||||
username = All(UnicodeString(strip=True, min=3, not_empty=True), ValidUsername) | ||||
if edit: | ||||
new_password = All(UnicodeString(strip=True, min=3, not_empty=False), ValidPassword) | ||||
else: | ||||
password = All(UnicodeString(strip=True, min=3, not_empty=False), ValidPassword) | ||||
active = StringBoolean(if_missing=False) | ||||
name = UnicodeString(strip=True, min=3, not_empty=True) | ||||
lastname = UnicodeString(strip=True, min=3, not_empty=True) | ||||
email = Email(not_empty=True) | ||||
return _UserForm | ||||
r265 | ||||
def RepoForm(edit=False): | ||||
class _RepoForm(formencode.Schema): | ||||
allow_extra_fields = True | ||||
r296 | filter_extra_fields = False | |||
r265 | repo_name = All(UnicodeString(strip=True, min=1, not_empty=True), ValidRepoName(edit)) | |||
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 | |||