kallithea Changeset - 2aa7f454204e

Changeset - 2aa7f454204e

Parent rev.

Child rev.

[Not reviewed]

beta

0 2 0

Marcin Kuzminski - 14 years ago 2011-12-06 00:18:27
marcin@python-works.com

fixes #298, ldap email addresses created by rhodecode automatically during first login didn't get converted to lower case, which lead to lookup failures and than wrong checks for uniqueness. Fixed that by putting a setter on db model column that will enforce converting to lowercase.

2 files changed with 17 insertions and 6 deletions:

rhodecode/model/db.py

rhodecode/model/forms.py

0 comments (0 inline, 0 general)

rhodecode/model/db.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
     rhodecode.model.db
     ~~~~~~~~~~~~~~~~~~
     Database Models for RhodeCode
     :created_on: Apr 08, 2010
     :author: marcink
     :copyright: (C) 2009-2011 Marcin Kuzminski <marcin@python-works.com>
     :license: GPLv3, see COPYING for more details.
 """
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # 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 General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 import os
 import logging
 import datetime
 import traceback
 from sqlalchemy import *
 from sqlalchemy.ext.hybrid import hybrid_property
 from sqlalchemy.orm import relationship, joinedload, class_mapper, validates
 from beaker.cache import cache_region, region_invalidate
 from vcs import get_backend
 from vcs.utils.helpers import get_scm
 from vcs.exceptions import VCSError
 from vcs.utils.lazy import LazyProperty
 from rhodecode.lib import str2bool, safe_str, get_changeset_safe, safe_unicode
 from rhodecode.lib.exceptions import UsersGroupsAssignedException
 from rhodecode.lib.compat import json
 from rhodecode.lib.caching_query import FromCache
 from rhodecode.model.meta import Base, Session
 log = logging.getLogger(__name__)
 #==============================================================================
 # BASE CLASSES
 #==============================================================================
 class ModelSerializer(json.JSONEncoder):
     """
     Simple Serializer for JSON,
     usage::
         to make object customized for serialization implement a __json__
         method that will return a dict for serialization into json
     example::
         class Task(object):
             def __init__(self, name, value):
                 self.name = name
                 self.value = value
             def __json__(self):
                 return dict(name=self.name,
                             value=self.value)
     """
     def default(self, obj):
         if hasattr(obj, '__json__'):
             return obj.__json__()
         else:
             return json.JSONEncoder.default(self, obj)
 class BaseModel(object):
     """Base Model for all classess
     """
     @classmethod
     def _get_keys(cls):
         """return column names for this model """
         return class_mapper(cls).c.keys()
     def get_dict(self):
         """return dict with keys and values corresponding
         to this model data """
         d = {}
         for k in self._get_keys():
             d[k] = getattr(self, k)
         return d
     def get_appstruct(self):
         """return list with keys and values tupples corresponding
         to this model data """
         l = []
         for k in self._get_keys():
             l.append((k, getattr(self, k),))
         return l
     def populate_obj(self, populate_dict):
         """populate model with data from given populate_dict"""
         for k in self._get_keys():
             if k in populate_dict:
                 setattr(self, k, populate_dict[k])
     @classmethod
     def query(cls):
         return Session.query(cls)
     @classmethod
     def get(cls, id_):
         if id_:
             return cls.query().get(id_)
     @classmethod
     def getAll(cls):
         return cls.query().all()
     @classmethod
     def delete(cls, id_):
         obj = cls.query().get(id_)
         Session.delete(obj)
 class RhodeCodeSetting(Base, BaseModel):
     __tablename__ = 'rhodecode_settings'
     __table_args__ = (UniqueConstraint('app_settings_name'), {'extend_existing':True})
     app_settings_id = Column("app_settings_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
     app_settings_name = Column("app_settings_name", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     _app_settings_value = Column("app_settings_value", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     def __init__(self, k='', v=''):
         self.app_settings_name = k
         self.app_settings_value = v
     @validates('_app_settings_value')
     def validate_settings_value(self, key, val):
         assert type(val) == unicode
         return val
     @hybrid_property
     def app_settings_value(self):
         v = self._app_settings_value
         if v == 'ldap_active':
             v = str2bool(v)
         return v
     @app_settings_value.setter
     def app_settings_value(self, val):
         """
         Setter that will always make sure we use unicode in app_settings_value
         :param val:
         """
         self._app_settings_value = safe_unicode(val)
     def __repr__(self):
         return "<%s('%s:%s')>" % (self.__class__.__name__,
                                   self.app_settings_name, self.app_settings_value)
     @classmethod
     def get_by_name(cls, ldap_key):
         return cls.query()\
             .filter(cls.app_settings_name == ldap_key).scalar()
     @classmethod
     def get_app_settings(cls, cache=False):
         ret = cls.query()
         if cache:
             ret = ret.options(FromCache("sql_cache_short", "get_hg_settings"))
         if not ret:
             raise Exception('Could not get application settings !')
         settings = {}
         for each in ret:
             settings['rhodecode_' + each.app_settings_name] = \
                 each.app_settings_value
         return settings
     @classmethod
     def get_ldap_settings(cls, cache=False):
         ret = cls.query()\
                 .filter(cls.app_settings_name.startswith('ldap_')).all()
         fd = {}
         for row in ret:
             fd.update({row.app_settings_name:row.app_settings_value})
         return fd
 class RhodeCodeUi(Base, BaseModel):
     __tablename__ = 'rhodecode_ui'
     __table_args__ = (UniqueConstraint('ui_key'), {'extend_existing':True})
     HOOK_UPDATE = 'changegroup.update'
     HOOK_REPO_SIZE = 'changegroup.repo_size'
     HOOK_PUSH = 'pretxnchangegroup.push_logger'
     HOOK_PULL = 'preoutgoing.pull_logger'
     ui_id = Column("ui_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
     ui_section = Column("ui_section", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     ui_key = Column("ui_key", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     ui_value = Column("ui_value", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     ui_active = Column("ui_active", Boolean(), nullable=True, unique=None, default=True)
     @classmethod
     def get_by_key(cls, key):
         return cls.query().filter(cls.ui_key == key)
     @classmethod
     def get_builtin_hooks(cls):
         q = cls.query()
         q = q.filter(cls.ui_key.in_([cls.HOOK_UPDATE,
                                     cls.HOOK_REPO_SIZE,
                                     cls.HOOK_PUSH, cls.HOOK_PULL]))
         return q.all()
     @classmethod
     def get_custom_hooks(cls):
         q = cls.query()
         q = q.filter(~cls.ui_key.in_([cls.HOOK_UPDATE,
                                     cls.HOOK_REPO_SIZE,
                                     cls.HOOK_PUSH, cls.HOOK_PULL]))
         q = q.filter(cls.ui_section == 'hooks')
         return q.all()
     @classmethod
     def create_or_update_hook(cls, key, val):
         new_ui = cls.get_by_key(key).scalar() or cls()
         new_ui.ui_section = 'hooks'
         new_ui.ui_active = True
         new_ui.ui_key = key
         new_ui.ui_value = val
         Session.add(new_ui)
 class User(Base, BaseModel):
     __tablename__ = 'users'
     __table_args__ = (UniqueConstraint('username'), UniqueConstraint('email'), {'extend_existing':True})
     user_id = Column("user_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
     username = Column("username", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     password = Column("password", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     active = Column("active", Boolean(), nullable=True, unique=None, default=None)
     admin = Column("admin", Boolean(), nullable=True, unique=None, default=False)
     name = Column("name", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     lastname = Column("lastname", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
-    email = Column("email", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
+    _email = Column("email", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     last_login = Column("last_login", DateTime(timezone=False), nullable=True, unique=None, default=None)
     ldap_dn = Column("ldap_dn", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     api_key = Column("api_key", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     user_log = relationship('UserLog', cascade='all')
     user_perms = relationship('UserToPerm', primaryjoin="User.user_id==UserToPerm.user_id", cascade='all')
     repositories = relationship('Repository')
     user_followers = relationship('UserFollowing', primaryjoin='UserFollowing.follows_user_id==User.user_id', cascade='all')
     repo_to_perm = relationship('UserRepoToPerm', primaryjoin='UserRepoToPerm.user_id==User.user_id', cascade='all')
     group_member = relationship('UsersGroupMember', cascade='all')
     notifications = relationship('UserNotification',)
     @hybrid_property
     def email(self):
         return self._email
     @email.setter
     def email(self, val):
         self._email = val.lower() if val else None
     @property
     def full_name(self):
         return '%s %s' % (self.name, self.lastname)
     @property
     def full_contact(self):
         return '%s %s <%s>' % (self.name, self.lastname, self.email)
     @property
     def short_contact(self):
         return '%s %s' % (self.name, self.lastname)
     @property
     def is_admin(self):
         return self.admin
     def __repr__(self):
         return "<%s('id:%s:%s')>" % (self.__class__.__name__,
                                      self.user_id, self.username)
     @classmethod
     def get_by_username(cls, username, case_insensitive=False, cache=False):
         if case_insensitive:
             q = cls.query().filter(cls.username.ilike(username))
         else:
             q = cls.query().filter(cls.username == username)
         if cache:
             q = q.options(FromCache("sql_cache_short",
                                     "get_user_%s" % username))
         return q.scalar()
     @classmethod
     def get_by_api_key(cls, api_key, cache=False):
         q = cls.query().filter(cls.api_key == api_key)
         if cache:
             q = q.options(FromCache("sql_cache_short",
                                     "get_api_key_%s" % api_key))
         return q.scalar()
     @classmethod
     def get_by_email(cls, email, cache=False):
         q = cls.query().filter(cls.email == email)
     def get_by_email(cls, email, case_insensitive=False, cache=False):
         if case_insensitive:
             q = cls.query().filter(cls.email.ilike(email))
         else:
             q = cls.query().filter(cls.email == email)
         if cache:
             q = q.options(FromCache("sql_cache_short",
                                     "get_api_key_%s" % email))
         return q.scalar()
     def update_lastlogin(self):
         """Update user lastlogin"""
         self.last_login = datetime.datetime.now()
         Session.add(self)
         log.debug('updated user %s lastlogin', self.username)
 class UserLog(Base, BaseModel):
     __tablename__ = 'user_logs'
     __table_args__ = {'extend_existing':True}
     user_log_id = Column("user_log_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
     user_id = Column("user_id", Integer(), ForeignKey('users.user_id'), nullable=False, unique=None, default=None)
     repository_id = Column("repository_id", Integer(), ForeignKey('repositories.repo_id'), nullable=True)
     repository_name = Column("repository_name", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     user_ip = Column("user_ip", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     action = Column("action", UnicodeText(length=1200000, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     action_date = Column("action_date", DateTime(timezone=False), nullable=True, unique=None, default=None)
     @property
     def action_as_day(self):
         return datetime.date(*self.action_date.timetuple()[:3])
     user = relationship('User')
     repository = relationship('Repository',cascade='')
 class UsersGroup(Base, BaseModel):
     __tablename__ = 'users_groups'
     __table_args__ = {'extend_existing':True}
     users_group_id = Column("users_group_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
     users_group_name = Column("users_group_name", String(length=255, convert_unicode=False, assert_unicode=None), nullable=False, unique=True, default=None)
     users_group_active = Column("users_group_active", Boolean(), nullable=True, unique=None, default=None)
     members = relationship('UsersGroupMember', cascade="all, delete, delete-orphan", lazy="joined")
     def __repr__(self):
         return '<userGroup(%s)>' % (self.users_group_name)
     @classmethod
     def get_by_group_name(cls, group_name, cache=False,
                           case_insensitive=False):
         if case_insensitive:
             q = cls.query().filter(cls.users_group_name.ilike(group_name))
         else:
             q = cls.query().filter(cls.users_group_name == group_name)
         if cache:
             q = q.options(FromCache("sql_cache_short",
                                     "get_user_%s" % group_name))
         return q.scalar()
     @classmethod
     def get(cls, users_group_id, cache=False):
         users_group = cls.query()
         if cache:
             users_group = users_group.options(FromCache("sql_cache_short",
                                     "get_users_group_%s" % users_group_id))
         return users_group.get(users_group_id)
 class UsersGroupMember(Base, BaseModel):
     __tablename__ = 'users_groups_members'
     __table_args__ = {'extend_existing':True}
     users_group_member_id = Column("users_group_member_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
     users_group_id = Column("users_group_id", Integer(), ForeignKey('users_groups.users_group_id'), nullable=False, unique=None, default=None)
     user_id = Column("user_id", Integer(), ForeignKey('users.user_id'), nullable=False, unique=None, default=None)
     user = relationship('User', lazy='joined')
     users_group = relationship('UsersGroup')
     def __init__(self, gr_id='', u_id=''):
         self.users_group_id = gr_id
         self.user_id = u_id
     @staticmethod
     def add_user_to_group(group, user):
         ugm = UsersGroupMember()
         ugm.users_group = group
         ugm.user = user
         Session.add(ugm)
         Session.commit()
         return ugm
 class Repository(Base, BaseModel):
     __tablename__ = 'repositories'
     __table_args__ = (UniqueConstraint('repo_name'), {'extend_existing':True},)
     repo_id = Column("repo_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
     repo_name = Column("repo_name", String(length=255, convert_unicode=False, assert_unicode=None), nullable=False, unique=True, default=None)
     clone_uri = Column("clone_uri", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=False, default=None)
     repo_type = Column("repo_type", String(length=255, convert_unicode=False, assert_unicode=None), nullable=False, unique=False, default='hg')
     user_id = Column("user_id", Integer(), ForeignKey('users.user_id'), nullable=False, unique=False, default=None)
     private = Column("private", Boolean(), nullable=True, unique=None, default=None)
     enable_statistics = Column("statistics", Boolean(), nullable=True, unique=None, default=True)
     enable_downloads = Column("downloads", Boolean(), nullable=True, unique=None, default=True)
     description = Column("description", String(length=10000, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     created_on = Column('created_on', DateTime(timezone=False), nullable=True, unique=None, default=datetime.datetime.now)
     fork_id = Column("fork_id", Integer(), ForeignKey('repositories.repo_id'), nullable=True, unique=False, default=None)
     group_id = Column("group_id", Integer(), ForeignKey('groups.group_id'), nullable=True, unique=False, default=None)
     user = relationship('User')
     fork = relationship('Repository', remote_side=repo_id)
     group = relationship('RepoGroup')
     repo_to_perm = relationship('UserRepoToPerm', cascade='all', order_by='UserRepoToPerm.repo_to_perm_id')
     users_group_to_perm = relationship('UsersGroupRepoToPerm', cascade='all')
     stats = relationship('Statistics', cascade='all', uselist=False)
     followers = relationship('UserFollowing', primaryjoin='UserFollowing.follows_repo_id==Repository.repo_id', cascade='all')
     logs = relationship('UserLog')
     def __repr__(self):
         return "<%s('%s:%s')>" % (self.__class__.__name__,
                                   self.repo_id, self.repo_name)
     @classmethod
     def url_sep(cls):
         return '/'
     @classmethod
     def get_by_repo_name(cls, repo_name):
         q = Session.query(cls).filter(cls.repo_name == repo_name)
         q = q.options(joinedload(Repository.fork))\
                 .options(joinedload(Repository.user))\
                 .options(joinedload(Repository.group))
         return q.scalar()
     @classmethod
     def get_repo_forks(cls, repo_id):
         return cls.query().filter(Repository.fork_id == repo_id)
     @classmethod
     def base_path(cls):
         """
         Returns base path when all repos are stored
         :param cls:
         """
         q = Session.query(RhodeCodeUi)\
             .filter(RhodeCodeUi.ui_key == cls.url_sep())
         q = q.options(FromCache("sql_cache_short", "repository_repo_path"))
         return q.one().ui_value
     @property
     def just_name(self):
         return self.repo_name.split(Repository.url_sep())[-1]
     @property
     def groups_with_parents(self):
         groups = []
         if self.group is None:
             return groups
         cur_gr = self.group
         groups.insert(0, cur_gr)
         while 1:
             gr = getattr(cur_gr, 'parent_group', None)
             cur_gr = cur_gr.parent_group
             if gr is None:
                 break
             groups.insert(0, gr)
         return groups
     @property
     def groups_and_repo(self):
         return self.groups_with_parents, self.just_name
     @LazyProperty
     def repo_path(self):
         """
         Returns base full path for that repository means where it actually
         exists on a filesystem
         """
         q = Session.query(RhodeCodeUi).filter(RhodeCodeUi.ui_key ==
                                               Repository.url_sep())
         q = q.options(FromCache("sql_cache_short", "repository_repo_path"))
         return q.one().ui_value
     @property
     def repo_full_path(self):
         p = [self.repo_path]
         # we need to split the name by / since this is how we store the
         # names in the database, but that eventually needs to be converted
         # into a valid system path
         p += self.repo_name.split(Repository.url_sep())
         return os.path.join(*p)
     def get_new_name(self, repo_name):
         """
         returns new full repository name based on assigned group and new new
         :param group_name:
         """
         path_prefix = self.group.full_path_splitted if self.group else []
         return Repository.url_sep().join(path_prefix + [repo_name])
     @property
     def _ui(self):
         """
         Creates an db based ui object for this repository
         """
         from mercurial import ui
         from mercurial import config
         baseui = ui.ui()
         #clean the baseui object
         baseui._ocfg = config.config()
         baseui._ucfg = config.config()
         baseui._tcfg = config.config()
         ret = RhodeCodeUi.query()\
             .options(FromCache("sql_cache_short", "repository_repo_ui")).all()
         hg_ui = ret
         for ui_ in hg_ui:
             if ui_.ui_active:
                 log.debug('settings ui from db[%s]%s:%s', ui_.ui_section,
                           ui_.ui_key, ui_.ui_value)
                 baseui.setconfig(ui_.ui_section, ui_.ui_key, ui_.ui_value)
         return baseui
     @classmethod
     def is_valid(cls, repo_name):
         """
         returns True if given repo name is a valid filesystem repository
         @param cls:
         @param repo_name:
         """
         from rhodecode.lib.utils import is_valid_repo
         return is_valid_repo(repo_name, cls.base_path())
     #==========================================================================
     # SCM PROPERTIES
     #==========================================================================
     def get_changeset(self, rev):
         return get_changeset_safe(self.scm_instance, rev)
     @property
     def tip(self):
         return self.get_changeset('tip')
     @property
     def author(self):
         return self.tip.author
     @property
     def last_change(self):
         return self.scm_instance.last_change
     #==========================================================================
     # SCM CACHE INSTANCE
     #==========================================================================
     @property
     def invalidate(self):
         return CacheInvalidation.invalidate(self.repo_name)
     def set_invalidate(self):
         """
         set a cache for invalidation for this instance
         """
         CacheInvalidation.set_invalidate(self.repo_name)
     @LazyProperty
     def scm_instance(self):
         return self.__get_instance()
     @property
     def scm_instance_cached(self):
         @cache_region('long_term')
         def _c(repo_name):
             return self.__get_instance()
         rn = self.repo_name
         log.debug('Getting cached instance of repo')
         inv = self.invalidate
         if inv is not None:
             region_invalidate(_c, None, rn)
             # update our cache
             CacheInvalidation.set_valid(inv.cache_key)
         return _c(rn)
     def __get_instance(self):
         repo_full_path = self.repo_full_path
         try:
             alias = get_scm(repo_full_path)[0]
             log.debug('Creating instance of %s repository', alias)
             backend = get_backend(alias)
         except VCSError:
             log.error(traceback.format_exc())
             log.error('Perhaps this repository is in db and not in '
                       'filesystem run rescan repositories with '
                       '"destroy old data " option from admin panel')
             return
         if alias == 'hg':
             repo = backend(safe_str(repo_full_path), create=False,
                            baseui=self._ui)
             # skip hidden web repository
             if repo._get_hidden():
                 return
         else:
             repo = backend(repo_full_path, create=False)
         return repo
 class RepoGroup(Base, BaseModel):
     __tablename__ = 'groups'
     __table_args__ = (UniqueConstraint('group_name', 'group_parent_id'),
                       CheckConstraint('group_id != group_parent_id'), {'extend_existing':True},)
     __mapper_args__ = {'order_by':'group_name'}
     group_id = Column("group_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
     group_name = Column("group_name", String(length=255, convert_unicode=False, assert_unicode=None), nullable=False, unique=True, default=None)
     group_parent_id = Column("group_parent_id", Integer(), ForeignKey('groups.group_id'), nullable=True, unique=None, default=None)
     group_description = Column("group_description", String(length=10000, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     parent_group = relationship('RepoGroup', remote_side=group_id)
     def __init__(self, group_name='', parent_group=None):
         self.group_name = group_name
         self.parent_group = parent_group
     def __repr__(self):
         return "<%s('%s:%s')>" % (self.__class__.__name__, self.group_id,
                                   self.group_name)
     @classmethod
     def groups_choices(cls):
         from webhelpers.html import literal as _literal
         repo_groups = [('', '')]
         sep = ' &raquo; '
         _name = lambda k: _literal(sep.join(k))
         repo_groups.extend([(x.group_id, _name(x.full_path_splitted))
                               for x in cls.query().all()])
         repo_groups = sorted(repo_groups, key=lambda t: t[1].split(sep)[0])
         return repo_groups
     @classmethod
     def url_sep(cls):
         return '/'
     @classmethod
     def get_by_group_name(cls, group_name, cache=False, case_insensitive=False):
         if case_insensitive:
             gr = cls.query()\
                 .filter(cls.group_name.ilike(group_name))
         else:
             gr = cls.query()\
                 .filter(cls.group_name == group_name)
         if cache:
             gr = gr.options(FromCache("sql_cache_short",
                                           "get_group_%s" % group_name))
         return gr.scalar()
     @property
     def parents(self):
         parents_recursion_limit = 5
         groups = []
         if self.parent_group is None:
             return groups
         cur_gr = self.parent_group
         groups.insert(0, cur_gr)
         cnt = 0
         while 1:
             cnt += 1

rhodecode/model/forms.py

➞

Show inline comments

@@ @@ -15,668 +15,668 @@ ignore_key_missing False If Tru @@
 <name> = formencode.validators.<name of validator>
 <name> must equal form name
 list=[1,2,3,4,5]
 for SELECT use formencode.All(OneOf(list), Int())
 """
 import os
 import re
 import logging
 import traceback
 import formencode
 from formencode import All
 from formencode.validators import UnicodeString, OneOf, Int, Number, Regex, \
     Email, Bool, StringBoolean, Set
 from pylons.i18n.translation import _
 from webhelpers.pylonslib.secure_form import authentication_token
 from rhodecode.config.routing import ADMIN_PREFIX
 from rhodecode.lib.utils import repo_name_slug
 from rhodecode.lib.auth import authenticate, get_crypt_password
 from rhodecode.lib.exceptions import LdapImportError
 from rhodecode.model.db import User, UsersGroup, RepoGroup, Repository
 from rhodecode import BACKENDS
 log = logging.getLogger(__name__)
 #this is needed to translate the messages using _() in validators
 class State_obj(object):
     _ = staticmethod(_)
 #==============================================================================
 # VALIDATORS
 #==============================================================================
 class ValidAuthToken(formencode.validators.FancyValidator):
     messages = {'invalid_token':_('Token mismatch')}
     def validate_python(self, value, state):
         if value != authentication_token():
             raise formencode.Invalid(self.message('invalid_token', state,
                                             search_number=value), value, state)
 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 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):
                     raise formencode.Invalid(_('This username already '
                                                'exists') , value, state)
             if re.match(r'^[a-zA-Z0-9]{1}[a-zA-Z0-9\-\_\.]+$', value) is None:
                 raise formencode.Invalid(_('Username may only contain '
                                            'alphanumeric characters '
                                            'underscores, periods or dashes '
                                            'and must begin with alphanumeric '
                                            'character'), value, state)
     return _ValidUsername
 def ValidUsersGroup(edit, old_data):
     class _ValidUsersGroup(formencode.validators.FancyValidator):
         def validate_python(self, value, state):
             if value in ['default']:
                 raise formencode.Invalid(_('Invalid group name'), value, state)
             #check if group is unique
             old_ugname = None
             if edit:
                 old_ugname = UsersGroup.get(
                             old_data.get('users_group_id')).users_group_name
             if old_ugname != value or not edit:
                 if UsersGroup.get_by_group_name(value, cache=False,
                                                case_insensitive=True):
                     raise formencode.Invalid(_('This users group '
                                                'already exists') , value,
                                              state)
             if re.match(r'^[a-zA-Z0-9]{1}[a-zA-Z0-9\-\_\.]+$', value) is None:
                 raise formencode.Invalid(_('RepoGroup name may only contain '
                                            'alphanumeric characters '
                                            'underscores, periods or dashes '
                                            'and must begin with alphanumeric '
                                            'character'), value, state)
     return _ValidUsersGroup
 def ValidReposGroup(edit, old_data):
     class _ValidReposGroup(formencode.validators.FancyValidator):
         def validate_python(self, value, state):
             # TODO WRITE VALIDATIONS
             group_name = value.get('group_name')
             group_parent_id = value.get('group_parent_id')
             # slugify repo group just in case :)
             slug = repo_name_slug(group_name)
             # check for parent of self
             parent_of_self = lambda:(old_data['group_id'] == int(group_parent_id)
                                      if group_parent_id else False)
             if edit and parent_of_self():
                     e_dict = {'group_parent_id':_('Cannot assign this group '
                                                   'as parent')}
                     raise formencode.Invalid('', value, state,
                                              error_dict=e_dict)
             old_gname = None
             if edit:
                 old_gname = RepoGroup.get(old_data.get('group_id')).group_name
             if old_gname != group_name or not edit:
                 # check filesystem
                 gr = RepoGroup.query().filter(RepoGroup.group_name == slug)\
                     .filter(RepoGroup.group_parent_id == group_parent_id).scalar()
                 if gr:
                     e_dict = {'group_name':_('This group already exists')}
                     raise formencode.Invalid('', value, state,
                                              error_dict=e_dict)
     return _ValidReposGroup
 class ValidPassword(formencode.validators.FancyValidator):
     def to_python(self, value, state):
         if value:
             if value.get('password'):
                 try:
                     value['password'] = get_crypt_password(value['password'])
                 except UnicodeEncodeError:
                     e_dict = {'password':_('Invalid characters in password')}
                     raise formencode.Invalid('', value, state, error_dict=e_dict)
             if value.get('password_confirmation'):
                 try:
                     value['password_confirmation'] = \
                         get_crypt_password(value['password_confirmation'])
                 except UnicodeEncodeError:
                     e_dict = {'password_confirmation':_('Invalid characters in password')}
                     raise formencode.Invalid('', value, state, error_dict=e_dict)
             if value.get('new_password'):
                 try:
                     value['new_password'] = \
                         get_crypt_password(value['new_password'])
                 except UnicodeEncodeError:
                     e_dict = {'new_password':_('Invalid characters in password')}
                     raise formencode.Invalid('', value, state, error_dict=e_dict)
             return value
 class ValidPasswordsMatch(formencode.validators.FancyValidator):
     def validate_python(self, value, state):
         pass_val = value.get('password') or value.get('new_password')
         if pass_val != value['password_confirmation']:
             e_dict = {'password_confirmation':
                    _('Passwords do not match')}
             raise formencode.Invalid('', value, state, error_dict=e_dict)
 class ValidAuth(formencode.validators.FancyValidator):
     messages = {
         'invalid_password':_('invalid password'),
         'invalid_login':_('invalid user name'),
         'disabled_account':_('Your account is disabled')
+    }
     # error mapping
     e_dict = {'username':messages['invalid_login'],
               'password':messages['invalid_password']}
     e_dict_disable = {'username':messages['disabled_account']}
     def validate_python(self, value, state):
         password = value['password']
         username = value['username']
         user = User.get_by_username(username)
         if authenticate(username, password):
             return value
         else:
             if user and user.active is False:
                 log.warning('user %s is disabled', username)
                 raise formencode.Invalid(self.message('disabled_account',
                                          state=State_obj),
                                          value, state,
                                          error_dict=self.e_dict_disable)
             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)
 class ValidRepoUser(formencode.validators.FancyValidator):
     def to_python(self, value, state):
         try:
             User.query().filter(User.active == True)\
                 .filter(User.username == value).one()
         except Exception:
             raise formencode.Invalid(_('This username is not valid'),
                                      value, state)
         return value
 def ValidRepoName(edit, old_data):
     class _ValidRepoName(formencode.validators.FancyValidator):
         def to_python(self, value, state):
             repo_name = value.get('repo_name')
             slug = repo_name_slug(repo_name)
             if slug in [ADMIN_PREFIX, '']:
                 e_dict = {'repo_name': _('This repository name is disallowed')}
                 raise formencode.Invalid('', value, state, error_dict=e_dict)
             if value.get('repo_group'):
                 gr = RepoGroup.get(value.get('repo_group'))
                 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
                 repo_name_full = group_path + RepoGroup.url_sep() + repo_name
             else:
                 group_path = ''
                 repo_name_full = repo_name
             value['repo_name_full'] = repo_name_full
             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):
                         e_dict = {'repo_name':_('This repository already '
                                                 'exists in a group "%s"') %
                                   gr.group_name}
                         raise formencode.Invalid('', value, state,
                                                  error_dict=e_dict)
                 elif RepoGroup.get_by_group_name(repo_name_full):
                         e_dict = {'repo_name':_('There is a group with this'
                                                 ' name already "%s"') %
                                   repo_name_full}
                         raise formencode.Invalid('', value, state,
                                                  error_dict=e_dict)
                 elif Repository.get_by_repo_name(repo_name_full):
                         e_dict = {'repo_name':_('This repository '
                                                 'already exists')}
                         raise formencode.Invalid('', value, state,
                                                  error_dict=e_dict)
             return value
     return _ValidRepoName
 def ValidForkName(*args, **kwargs):
     return ValidRepoName(*args, **kwargs)
 def SlugifyName():
     class _SlugifyName(formencode.validators.FancyValidator):
         def to_python(self, value, state):
             return repo_name_slug(value)
     return _SlugifyName
 def ValidCloneUri():
     from mercurial.httprepo import httprepository, httpsrepository
     from rhodecode.lib.utils import make_ui
     class _ValidCloneUri(formencode.validators.FancyValidator):
         def to_python(self, value, state):
             if not value:
                 pass
             elif value.startswith('https'):
                 try:
                     httpsrepository(make_ui('db'), value).capabilities
                 except Exception, e:
                     log.error(traceback.format_exc())
                     raise formencode.Invalid(_('invalid clone url'), value,
                                              state)
             elif value.startswith('http'):
                 try:
                     httprepository(make_ui('db'), value).capabilities
                 except Exception, e:
                     log.error(traceback.format_exc())
                     raise formencode.Invalid(_('invalid clone url'), value,
                                              state)
             else:
                 raise formencode.Invalid(_('Invalid clone url, provide a '
                                            'valid clone http\s url'), value,
                                          state)
             return value
     return _ValidCloneUri
 def ValidForkType(old_data):
     class _ValidForkType(formencode.validators.FancyValidator):
         def to_python(self, value, state):
             if old_data['repo_type'] != value:
                 raise formencode.Invalid(_('Fork have to be the same '
                                            'type as original'), value, state)
             return value
     return _ValidForkType
 class ValidPerms(formencode.validators.FancyValidator):
     messages = {'perm_new_member_name':_('This username or users group name'
                                          ' 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():
             #means new added member to permissions
             if k.startswith('perm_new_member'):
                 new_perm = value.get('perm_new_member', False)
                 new_member = value.get('perm_new_member_name', False)
                 new_type = value.get('perm_new_member_type')
                 if new_member and new_perm:
                     if (new_member, new_perm, new_type) not in perms_new:
                         perms_new.append((new_member, new_perm, new_type))
             elif k.startswith('u_perm_') or k.startswith('g_perm_'):
                 member = k[7:]
                 t = {'u':'user',
                      'g':'users_group'}[k[0]]
                 if member == 'default':
                     if value['private']:
                         #set none for default when updating to private repo
                         v = 'repository.none'
                 perms_update.append((member, v, t))
         value['perms_updates'] = perms_update
         value['perms_new'] = perms_new
         #update permissions
         for k, v, t in perms_new:
             try:
                 if t is 'user':
                     self.user_db = User.query()\
                         .filter(User.active == True)\
                         .filter(User.username == k).one()
                 if t is 'users_group':
                     self.user_db = UsersGroup.query()\
                         .filter(UsersGroup.users_group_active == True)\
                         .filter(UsersGroup.users_group_name == k).one()
             except Exception:
                 msg = self.message('perm_new_member_name',
                                      state=State_obj)
                 raise formencode.Invalid(msg, value, state,
                                          error_dict={'perm_new_member_name':msg})
         return value
 class ValidSettings(formencode.validators.FancyValidator):
     def to_python(self, value, state):
         #settings  form can't edit user
+        # settings  form can't edit user
         if value.has_key('user'):
             del['value']['user']
         return value
 class ValidPath(formencode.validators.FancyValidator):
     def to_python(self, value, state):
         if not os.path.isdir(value):
             msg = _('This is not a valid path')
             raise formencode.Invalid(msg, value, state,
                                      error_dict={'paths_root_path':msg})
         return value
 def UniqSystemEmail(old_data):
     class _UniqSystemEmail(formencode.validators.FancyValidator):
         def to_python(self, value, state):
             value = value.lower()
             if old_data.get('email') != value:
-                user = User.query().filter(User.email == value).scalar()
+                user = User.get_by_email(value, case_insensitive=True)
                 if user:
                     raise formencode.Invalid(
                                     _("This e-mail address is already taken"),
                                     value, state)
             return value
     return _UniqSystemEmail
 class ValidSystemEmail(formencode.validators.FancyValidator):
     def to_python(self, value, state):
         value = value.lower()
-        user = User.query().filter(User.email == value).scalar()
+        user = User.get_by_email(value, case_insensitive=True)
         if  user is None:
             raise formencode.Invalid(_("This e-mail address doesn't exist.") ,
                                      value, state)
         return value
 class LdapLibValidator(formencode.validators.FancyValidator):
     def to_python(self, value, state):
         try:
             import ldap
         except ImportError:
             raise LdapImportError
         return value
 class AttrLoginValidator(formencode.validators.FancyValidator):
     def to_python(self, value, state):
         if not value or not isinstance(value, (str, unicode)):
             raise formencode.Invalid(_("The LDAP Login attribute of the CN "
                                        "must be specified - this is the name "
                                        "of the attribute that is equivalent "
                                        "to 'username'"),
                                      value, state)
         return value
 #===============================================================================
 # FORMS
 #===============================================================================
 class LoginForm(formencode.Schema):
     allow_extra_fields = True
     filter_extra_fields = True
     username = UnicodeString(
                              strip=True,
                              min=1,
                              not_empty=True,
                              messages={
                                 'empty':_('Please enter a login'),
                                 'tooShort':_('Enter a value %(min)i characters long or more')}
+                            )
     password = UnicodeString(
                             strip=True,
                             min=3,
                             not_empty=True,
                             messages={
                                 'empty':_('Please enter a password'),
                                 'tooShort':_('Enter %(min)i characters or more')}
+                                )
     chained_validators = [ValidAuth]
 def UserForm(edit=False, old_data={}):
     class _UserForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         username = All(UnicodeString(strip=True, min=1, not_empty=True),
                        ValidUsername(edit, old_data))
         if edit:
             new_password = All(UnicodeString(strip=True, min=6, not_empty=False))
             password_confirmation = All(UnicodeString(strip=True, min=6, not_empty=False))
             admin = StringBoolean(if_missing=False)
         else:
             password = All(UnicodeString(strip=True, min=6, not_empty=True))
             password_confirmation = All(UnicodeString(strip=True, min=6, not_empty=False))
         active = StringBoolean(if_missing=False)
         name = UnicodeString(strip=True, min=1, not_empty=True)
         lastname = UnicodeString(strip=True, min=1, not_empty=True)
         email = All(Email(not_empty=True), UniqSystemEmail(old_data))
         chained_validators = [ValidPasswordsMatch, ValidPassword]
     return _UserForm
 def UsersGroupForm(edit=False, old_data={}, available_members=[]):
     class _UsersGroupForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         users_group_name = All(UnicodeString(strip=True, min=1, not_empty=True),
                        ValidUsersGroup(edit, old_data))
         users_group_active = StringBoolean(if_missing=False)
         if edit:
             users_group_members = OneOf(available_members, hideList=False,
                                         testValueList=True,
                                         if_missing=None, not_empty=False)
     return _UsersGroupForm
 def ReposGroupForm(edit=False, old_data={}, available_groups=[]):
     class _ReposGroupForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         group_name = All(UnicodeString(strip=True, min=1, not_empty=True),
                                SlugifyName())
         group_description = UnicodeString(strip=True, min=1,
                                                 not_empty=True)
         group_parent_id = OneOf(available_groups, hideList=False,
                                         testValueList=True,
                                         if_missing=None, not_empty=False)
         chained_validators = [ValidReposGroup(edit, old_data)]
     return _ReposGroupForm
 def RegisterForm(edit=False, old_data={}):
     class _RegisterForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         username = All(ValidUsername(edit, old_data),
                        UnicodeString(strip=True, min=1, not_empty=True))
         password = All(UnicodeString(strip=True, min=6, not_empty=True))
         password_confirmation = All(UnicodeString(strip=True, min=6, not_empty=True))
         active = StringBoolean(if_missing=False)
         name = UnicodeString(strip=True, min=1, not_empty=True)
         lastname = UnicodeString(strip=True, min=1, not_empty=True)
         email = All(Email(not_empty=True), UniqSystemEmail(old_data))
         chained_validators = [ValidPasswordsMatch, ValidPassword]
     return _RegisterForm
 def PasswordResetForm():
     class _PasswordResetForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         email = All(ValidSystemEmail(), Email(not_empty=True))
     return _PasswordResetForm
 def RepoForm(edit=False, old_data={}, supported_backends=BACKENDS.keys(),
              repo_groups=[]):
     class _RepoForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         repo_name = All(UnicodeString(strip=True, min=1, not_empty=True),
                         SlugifyName())
         clone_uri = All(UnicodeString(strip=True, min=1, not_empty=False),
                         ValidCloneUri()())
         repo_group = OneOf(repo_groups, hideList=True)
         repo_type = OneOf(supported_backends)
         description = UnicodeString(strip=True, min=1, not_empty=True)
         private = StringBoolean(if_missing=False)
         enable_statistics = StringBoolean(if_missing=False)
         enable_downloads = StringBoolean(if_missing=False)
         if edit:
             #this is repo owner
             user = All(UnicodeString(not_empty=True), ValidRepoUser)
         chained_validators = [ValidRepoName(edit, old_data), ValidPerms]
     return _RepoForm
 def RepoForkForm(edit=False, old_data={}, supported_backends=BACKENDS.keys(),
                  repo_groups=[]):
     class _RepoForkForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         repo_name = All(UnicodeString(strip=True, min=1, not_empty=True),
                         SlugifyName())
         repo_group = OneOf(repo_groups, hideList=True)
         repo_type = All(ValidForkType(old_data), OneOf(supported_backends))
         description = UnicodeString(strip=True, min=1, not_empty=True)
         private = StringBoolean(if_missing=False)
         copy_permissions = StringBoolean(if_missing=False)
         update_after_clone = StringBoolean(if_missing=False)
         fork_parent_id = UnicodeString()
         chained_validators = [ValidForkName(edit, old_data)]
     return _RepoForkForm
 def RepoSettingsForm(edit=False, old_data={}, supported_backends=BACKENDS.keys(),
                      repo_groups=[]):
     class _RepoForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         repo_name = All(UnicodeString(strip=True, min=1, not_empty=True),
                         SlugifyName())
         description = UnicodeString(strip=True, min=1, not_empty=True)
         repo_group = OneOf(repo_groups, hideList=True)
         private = StringBoolean(if_missing=False)
         chained_validators = [ValidRepoName(edit, old_data), ValidPerms,
                               ValidSettings]
     return _RepoForm
 def ApplicationSettingsForm():
     class _ApplicationSettingsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         rhodecode_title = UnicodeString(strip=True, min=1, not_empty=True)
         rhodecode_realm = UnicodeString(strip=True, min=1, not_empty=True)
         rhodecode_ga_code = UnicodeString(strip=True, min=1, not_empty=False)
     return _ApplicationSettingsForm
 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=1, not_empty=True))
         hooks_changegroup_update = OneOf(['True', 'False'], if_missing=False)
         hooks_changegroup_repo_size = OneOf(['True', 'False'], if_missing=False)
         hooks_pretxnchangegroup_push_logger = OneOf(['True', 'False'], if_missing=False)
         hooks_preoutgoing_pull_logger = OneOf(['True', 'False'], if_missing=False)
     return _ApplicationUiSettingsForm
 def DefaultPermissionsForm(perms_choices, register_choices, create_choices):
     class _DefaultPermissionsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         overwrite_default = StringBoolean(if_missing=False)
         anonymous = OneOf(['True', 'False'], if_missing=False)
         default_perm = OneOf(perms_choices)
         default_register = OneOf(register_choices)
         default_create = OneOf(create_choices)
     return _DefaultPermissionsForm
 def LdapSettingsForm(tls_reqcert_choices, search_scope_choices, tls_kind_choices):
     class _LdapSettingsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         pre_validators = [LdapLibValidator]
         ldap_active = StringBoolean(if_missing=False)
         ldap_host = UnicodeString(strip=True,)
         ldap_port = Number(strip=True,)
         ldap_tls_kind = OneOf(tls_kind_choices)
         ldap_tls_reqcert = OneOf(tls_reqcert_choices)
         ldap_dn_user = UnicodeString(strip=True,)
         ldap_dn_pass = UnicodeString(strip=True,)
         ldap_base_dn = UnicodeString(strip=True,)
         ldap_filter = UnicodeString(strip=True,)
         ldap_search_scope = OneOf(search_scope_choices)
         ldap_attr_login = All(AttrLoginValidator, UnicodeString(strip=True,))
         ldap_attr_firstname = UnicodeString(strip=True,)
         ldap_attr_lastname = UnicodeString(strip=True,)
         ldap_attr_email = UnicodeString(strip=True,)
     return _LdapSettingsForm

0 comments (0 inline, 0 general)