Changeset - 8e49b6ceffe1
[Not reviewed]
beta
0 3 0
Marcin Kuzminski - 15 years ago 2011-02-13 02:48:02
marcin@python-works.com
fixes fixes fixes ! optimized queries on journal
added some logging, fixed found bugs due to previous refactorings
3 files changed with 13 insertions and 7 deletions:
0 comments (0 inline, 0 general)
rhodecode/controllers/admin/admin.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
"""
 
    rhodecode.controllers.admin.admin
 
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

	
 
    Controller for Admin panel of Rhodecode
 
    
 
    :created_on: Apr 7, 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; version 2
 
# of the License or (at your opinion) any later version of the license.
 
# 
 
# 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, write to the Free Software
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
 
# MA  02110-1301, USA.
 

	
 
import logging
 

	
 
from pylons import request, tmpl_context as c
 
from sqlalchemy.orm import joinedload
 
from webhelpers.paginate import Page
 

	
 
from rhodecode.lib.auth import LoginRequired, HasPermissionAllDecorator
 
from rhodecode.lib.base import BaseController, render
 
from rhodecode.model.db import UserLog
 
from webhelpers.paginate import Page
 
from rhodecode.lib.auth import LoginRequired, HasPermissionAllDecorator
 

	
 
log = logging.getLogger(__name__)
 

	
 
class AdminController(BaseController):
 

	
 
    @LoginRequired()
 
    def __before__(self):
 
        super(AdminController, self).__before__()
 

	
 
    @HasPermissionAllDecorator('hg.admin')
 
    def index(self):
 

	
 
        users_log = self.sa.query(UserLog).order_by(UserLog.action_date.desc())
 
        users_log = self.sa.query(UserLog)\
 
                .options(joinedload(UserLog.user))\
 
                .options(joinedload(UserLog.repository))\
 
                .order_by(UserLog.action_date.desc())
 

	
 
        p = int(request.params.get('page', 1))
 
        c.users_log = Page(users_log, page=p, items_per_page=10)
 
        c.log_data = render('admin/admin_log.html')
 
        if request.params.get('partial'):
 
            return c.log_data
 
        return render('admin/admin.html')
 

	
rhodecode/lib/auth.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
"""
 
    rhodecode.lib.auth
 
    ~~~~~~~~~~~~~~~~~~
 
    
 
    authentication and permission libraries
 
    
 
    :created_on: Apr 4, 2010
 
    :copyright: (c) 2010 by marcink.
 
    :license: LICENSE_NAME, see LICENSE_FILE 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; version 2
 
# of the License or (at your opinion) any later version of the license.
 
# 
 
# 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, write to the Free Software
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
 
# MA  02110-1301, USA.
 

	
 
import bcrypt
 
import random
 
import logging
 
import traceback
 

	
 
from decorator import decorator
 

	
 
from pylons import config, session, url, request
 
from pylons.controllers.util import abort, redirect
 

	
 
from rhodecode.lib.exceptions import LdapPasswordError, LdapUsernameError
 
from rhodecode.lib.utils import get_repo_slug
 
from rhodecode.lib.auth_ldap import AuthLdap
 

	
 
from rhodecode.model import meta
 
from rhodecode.model.user import UserModel
 
from rhodecode.model.db import User, RepoToPerm, Repository, Permission, \
 
    UserToPerm, UsersGroupToPerm, UsersGroupMember
 

	
 

	
 
log = logging.getLogger(__name__)
 

	
 

	
 
PERM_WEIGHTS = {'repository.none':0,
 
                'repository.read':1,
 
                'repository.write':3,
 
                'repository.admin':3}
 

	
 

	
 
class PasswordGenerator(object):
 
    """This is a simple class for generating password from
 
        different sets of characters
 
        usage:
 
        passwd_gen = PasswordGenerator()
 
        #print 8-letter password containing only big and small letters of alphabet
 
        print passwd_gen.gen_password(8, passwd_gen.ALPHABETS_BIG_SMALL)        
 
    """
 
    ALPHABETS_NUM = r'''1234567890'''#[0]
 
    ALPHABETS_SMALL = r'''qwertyuiopasdfghjklzxcvbnm'''#[1]
 
    ALPHABETS_BIG = r'''QWERTYUIOPASDFGHJKLZXCVBNM'''#[2]
 
    ALPHABETS_SPECIAL = r'''`-=[]\;',./~!@#$%^&*()_+{}|:"<>?'''    #[3]
 
    ALPHABETS_FULL = ALPHABETS_BIG + ALPHABETS_SMALL + ALPHABETS_NUM + ALPHABETS_SPECIAL#[4]
 
    ALPHABETS_ALPHANUM = ALPHABETS_BIG + ALPHABETS_SMALL + ALPHABETS_NUM#[5]
 
    ALPHABETS_BIG_SMALL = ALPHABETS_BIG + ALPHABETS_SMALL
 
    ALPHABETS_ALPHANUM_BIG = ALPHABETS_BIG + ALPHABETS_NUM#[6]
 
    ALPHABETS_ALPHANUM_SMALL = ALPHABETS_SMALL + ALPHABETS_NUM#[7]
 

	
 
    def __init__(self, passwd=''):
 
        self.passwd = passwd
 

	
 
    def gen_password(self, len, type):
 
        self.passwd = ''.join([random.choice(type) for _ in xrange(len)])
 
        return self.passwd
 

	
 

	
 
def get_crypt_password(password):
 
    """Cryptographic function used for password hashing based on pybcrypt
 
    
 
    :param password: password to hash
 
    """
 
    return bcrypt.hashpw(password, bcrypt.gensalt(10))
 

	
 
def check_password(password, hashed):
 
    return bcrypt.hashpw(password, hashed) == hashed
 

	
 
def authfunc(environ, username, password):
 
    """Dummy authentication function used in Mercurial/Git/ and access control,
 
    
 
    :param environ: needed only for using in Basic auth
 
    """
 
    return authenticate(username, password)
 

	
 

	
 
def authenticate(username, password):
 
    """Authentication function used for access control,
 
    firstly checks for db authentication then if ldap is enabled for ldap
 
    authentication, also creates ldap user if not in database
 
    
 
    :param username: username
 
    :param password: password
 
    """
 
    user_model = UserModel()
 
    user = user_model.get_by_username(username, cache=False)
 

	
 
    log.debug('Authenticating user using RhodeCode account')
 
    if user is not None and not user.ldap_dn:
 
        if user.active:
 

	
 
            if user.username == 'default' and user.active:
 
                log.info('user %s authenticated correctly as anonymous user',
 
                         username)
 
                return True
 

	
 
            elif user.username == username and check_password(password, user.password):
 
                log.info('user %s authenticated correctly', username)
 
                return True
 
        else:
 
            log.warning('user %s is disabled', username)
 

	
 
    else:
 
        log.debug('Regular authentication failed')
 
        user_obj = user_model.get_by_username(username, cache=False,
 
                                            case_insensitive=True)
 

	
 
        if user_obj is not None and not user_obj.ldap_dn:
 
            log.debug('this user already exists as non ldap')
 
            return False
 

	
 
        from rhodecode.model.settings import SettingsModel
 
        ldap_settings = SettingsModel().get_ldap_settings()
 

	
 
        #======================================================================
 
        # FALLBACK TO LDAP AUTH IF ENABLE                
 
        #======================================================================
 
        if ldap_settings.get('ldap_active', False):
 
            log.debug("Authenticating user using ldap")
 
            kwargs = {
 
                  'server':ldap_settings.get('ldap_host', ''),
 
                  'base_dn':ldap_settings.get('ldap_base_dn', ''),
 
                  'port':ldap_settings.get('ldap_port'),
 
                  'bind_dn':ldap_settings.get('ldap_dn_user'),
 
                  'bind_pass':ldap_settings.get('ldap_dn_pass'),
 
                  'use_ldaps':ldap_settings.get('ldap_ldaps'),
 
                  'tls_reqcert':ldap_settings.get('ldap_tls_reqcert'),
 
                  'ldap_filter':ldap_settings.get('ldap_filter'),
 
                  'search_scope':ldap_settings.get('ldap_search_scope'),
 
                  'attr_login':ldap_settings.get('ldap_attr_login'),
 
                  'ldap_version':3,
 
                  }
 
            log.debug('Checking for ldap authentication')
 
            try:
 
                aldap = AuthLdap(**kwargs)
 
                (user_dn, ldap_attrs) = aldap.authenticate_ldap(username, password)
 
                log.debug('Got ldap DN response %s', user_dn)
 

	
 
                user_attrs = {
 
                    'name'     : ldap_attrs[ldap_settings.get('ldap_attr_firstname')][0],
 
                    'lastname' : ldap_attrs[ldap_settings.get('ldap_attr_lastname')][0],
 
                    'email'    : ldap_attrs[ldap_settings.get('ldap_attr_email')][0],
 
                    }
 

	
 
                if user_model.create_ldap(username, password, user_dn, user_attrs):
 
                    log.info('created new ldap user %s', username)
 

	
 
                return True
 
            except (LdapUsernameError, LdapPasswordError,):
 
                pass
 
            except (Exception,):
 
                log.error(traceback.format_exc())
 
                pass
 
    return False
 

	
 
class  AuthUser(object):
 
    """A simple object that handles a mercurial username for authentication
 
    """
 

	
 
    def __init__(self):
 
        self.username = 'None'
 
        self.name = ''
 
        self.lastname = ''
 
        self.email = ''
 
        self.user_id = None
 
        self.is_authenticated = False
 
        self.is_admin = False
 
        self.permissions = {}
 

	
 
    def __repr__(self):
 
        return "<AuthUser('id:%s:%s')>" % (self.user_id, self.username)
 

	
 
def set_available_permissions(config):
 
    """This function will propagate pylons globals with all available defined
 
    permission given in db. We don't want to check each time from db for new 
 
    permissions since adding a new permission also requires application restart
 
    ie. to decorate new views with the newly created permission
 
    
 
    :param config: current pylons config instance
 
    
 
    """
 
    log.info('getting information about all available permissions')
 
    try:
 
        sa = meta.Session()
 
        all_perms = sa.query(Permission).all()
 
    except:
 
        pass
 
    finally:
 
        meta.Session.remove()
 

	
 
    config['available_permissions'] = [x.permission_name for x in all_perms]
 

	
 
def fill_perms(user):
 
    """Fills user permission attribute with permissions taken from database
 
    works for permissions given for repositories, and for permissions that
 
    as part of beeing group member
 
    
 
    :param user: user instance to fill his perms
 
    """
 

	
 
    sa = meta.Session()
 
    user.permissions['repositories'] = {}
 
    user.permissions['global'] = set()
 

	
 
    #===========================================================================
 
    # fetch default permissions
 
    #===========================================================================
 
    default_user = UserModel().get_by_username('default', cache=True)
 

	
 
    default_perms = sa.query(RepoToPerm, Repository, Permission)\
 
        .join((Repository, RepoToPerm.repository_id == Repository.repo_id))\
 
        .join((Permission, RepoToPerm.permission_id == Permission.permission_id))\
 
        .filter(RepoToPerm.user == default_user).all()
 

	
 
    if user.is_admin:
 
        #=======================================================================
 
        # #admin have all default rights set to admin        
 
        #=======================================================================
 
        user.permissions['global'].add('hg.admin')
 

	
 
        for perm in default_perms:
 
            p = 'repository.admin'
 
            user.permissions['repositories'][perm.RepoToPerm.repository.repo_name] = p
 

	
 
    else:
 
        #=======================================================================
 
        # set default permissions
 
        #=======================================================================
 

	
 
        #default global
 
        default_global_perms = sa.query(UserToPerm)\
 
            .filter(UserToPerm.user == sa.query(User)\
 
                   .filter(User.username == 'default').one())
 

	
 
        for perm in default_global_perms:
 
            user.permissions['global'].add(perm.permission.permission_name)
 

	
 
        #default for repositories
 
        for perm in default_perms:
 
            if perm.Repository.private and not perm.Repository.user_id == user.user_id:
 
                #disable defaults for private repos,
 
                p = 'repository.none'
 
            elif perm.Repository.user_id == user.user_id:
 
                #set admin if owner
 
                p = 'repository.admin'
 
            else:
 
                p = perm.Permission.permission_name
 

	
 
            user.permissions['repositories'][perm.RepoToPerm.repository.repo_name] = p
 

	
 
        #=======================================================================
 
        # overwrite default with user permissions if any
 
        #=======================================================================
 
        user_perms = sa.query(RepoToPerm, Permission, Repository)\
 
            .join((Repository, RepoToPerm.repository_id == Repository.repo_id))\
 
            .join((Permission, RepoToPerm.permission_id == Permission.permission_id))\
 
            .filter(RepoToPerm.user_id == user.user_id).all()
 

	
 
        for perm in user_perms:
 
            if perm.Repository.user_id == user.user_id:#set admin if owner
 
                p = 'repository.admin'
 
            else:
 
                p = perm.Permission.permission_name
 
            user.permissions['repositories'][perm.RepoToPerm.repository.repo_name] = p
 

	
 

	
 
        #=======================================================================
 
        # check if user is part of groups for this repository and fill in 
 
        # (or replace with higher) permissions
 
        #=======================================================================
 
        user_perms_from_users_groups = sa.query(UsersGroupToPerm, Permission, Repository,)\
 
            .join((Repository, UsersGroupToPerm.repository_id == Repository.repo_id))\
 
            .join((Permission, UsersGroupToPerm.permission_id == Permission.permission_id))\
 
            .join((UsersGroupMember, UsersGroupToPerm.users_group_id == UsersGroupMember.users_group_id))\
 
            .filter(UsersGroupMember.user_id == user.user_id).all()
 

	
 
        for perm in user_perms_from_users_groups:
 
            p = perm.Permission.permission_name
 
            cur_perm = user.permissions['repositories'][perm.UsersGroupToPerm.repository.repo_name]
 
            #overwrite permission only if it's greater than permission given from other sources
 
            if PERM_WEIGHTS[p] > PERM_WEIGHTS[cur_perm]:
 
                user.permissions['repositories'][perm.UsersGroupToPerm.repository.repo_name] = p
 

	
 
    meta.Session.remove()
 
    return user
 

	
 
def get_user(session):
 
    """Gets user from session, and wraps permissions into user
 
    
 
    :param session:
 
    """
 
    user = session.get('rhodecode_user', AuthUser())
 
    #if the user is not logged in we check for anonymous access
 
    #if user is logged and it's a default user check if we still have anonymous
 
    #access enabled
 
    if user.user_id is None or user.username == 'default':
 
        anonymous_user = UserModel().get_by_username('default', cache=True)
 
        if anonymous_user.active is True:
 
            #then we set this user is logged in
 
            user.is_authenticated = True
 
            user.user_id = anonymous_user.user_id
 
        else:
 
            user.is_authenticated = False
 

	
 
    if user.is_authenticated:
 
        user = UserModel().fill_data(user)
 

	
 
    user = fill_perms(user)
 
    session['rhodecode_user'] = user
 
    session.save()
 
    return user
 

	
 
#===============================================================================
 
# CHECK DECORATORS
 
#===============================================================================
 
class LoginRequired(object):
 
    """Must be logged in to execute this function else 
 
    redirect to login page"""
 

	
 
    def __call__(self, func):
 
        return decorator(self.__wrapper, func)
 

	
 
    def __wrapper(self, func, *fargs, **fkwargs):
 
        user = session.get('rhodecode_user', AuthUser())
 
        log.debug('Checking login required for user:%s', user.username)
 
        if user.is_authenticated:
 
            log.debug('user %s is authenticated', user.username)
 
            return func(*fargs, **fkwargs)
 
        else:
 
            log.warn('user %s not authenticated', user.username)
 

	
 
            p = ''
 
            if request.environ.get('SCRIPT_NAME') != '/':
 
                p += request.environ.get('SCRIPT_NAME')
 

	
 
            p += request.environ.get('PATH_INFO')
 
            if request.environ.get('QUERY_STRING'):
 
                p += '?' + request.environ.get('QUERY_STRING')
 

	
 
            log.debug('redirecting to login page with %s', p)
 
            return redirect(url('login_home', came_from=p))
 

	
 
class NotAnonymous(object):
 
    """Must be logged in to execute this function else 
 
    redirect to login page"""
 

	
 
    def __call__(self, func):
 
        return decorator(self.__wrapper, func)
 

	
 
    def __wrapper(self, func, *fargs, **fkwargs):
 
        user = session.get('rhodecode_user', AuthUser())
 
        log.debug('Checking if user is not anonymous')
 

	
 
        anonymous = user.username == 'default'
 

	
 
        if anonymous:
 
            p = ''
 
            if request.environ.get('SCRIPT_NAME') != '/':
 
                p += request.environ.get('SCRIPT_NAME')
 

	
 
            p += request.environ.get('PATH_INFO')
 
            if request.environ.get('QUERY_STRING'):
 
                p += '?' + request.environ.get('QUERY_STRING')
 
            return redirect(url('login_home', came_from=p))
 
        else:
 
            return func(*fargs, **fkwargs)
 

	
 
class PermsDecorator(object):
 
    """Base class for decorators"""
 

	
 
    def __init__(self, *required_perms):
 
        available_perms = config['available_permissions']
 
        for perm in required_perms:
 
            if perm not in available_perms:
 
                raise Exception("'%s' permission is not defined" % perm)
 
        self.required_perms = set(required_perms)
 
        self.user_perms = None
 

	
 
    def __call__(self, func):
 
        return decorator(self.__wrapper, func)
 

	
 

	
 
    def __wrapper(self, func, *fargs, **fkwargs):
 
#        _wrapper.__name__ = func.__name__
 
#        _wrapper.__dict__.update(func.__dict__)
 
#        _wrapper.__doc__ = func.__doc__
 
        self.user = session.get('rhodecode_user', AuthUser())
 
        self.user_perms = self.user.permissions
 
        log.debug('checking %s permissions %s for %s %s',
 
           self.__class__.__name__, self.required_perms, func.__name__,
 
               self.user)
 

	
 
        if self.check_permissions():
 
            log.debug('Permission granted for %s %s', func.__name__, self.user)
 

	
 
            return func(*fargs, **fkwargs)
 

	
 
        else:
 
            log.warning('Permission denied for %s %s', func.__name__, self.user)
 
            #redirect with forbidden ret code
 
            return abort(403)
 

	
 

	
 

	
 
    def check_permissions(self):
 
        """Dummy function for overriding"""
 
        raise Exception('You have to write this function in child class')
 

	
 
class HasPermissionAllDecorator(PermsDecorator):
 
    """Checks for access permission for all given predicates. All of them 
 
    have to be meet in order to fulfill the request
 
    """
 

	
 
    def check_permissions(self):
 
        if self.required_perms.issubset(self.user_perms.get('global')):
 
            return True
 
        return False
 

	
 

	
 
class HasPermissionAnyDecorator(PermsDecorator):
 
    """Checks for access permission for any of given predicates. In order to 
 
    fulfill the request any of predicates must be meet
 
    """
 

	
 
    def check_permissions(self):
 
        if self.required_perms.intersection(self.user_perms.get('global')):
 
            return True
 
        return False
 

	
 
class HasRepoPermissionAllDecorator(PermsDecorator):
 
    """Checks for access permission for all given predicates for specific 
 
    repository. All of them have to be meet in order to fulfill the request
 
    """
 

	
 
    def check_permissions(self):
 
        repo_name = get_repo_slug(request)
 
        try:
 
            user_perms = set([self.user_perms['repositories'][repo_name]])
 
        except KeyError:
 
            return False
 
        if self.required_perms.issubset(user_perms):
 
            return True
 
        return False
 

	
 

	
 
class HasRepoPermissionAnyDecorator(PermsDecorator):
 
    """Checks for access permission for any of given predicates for specific 
 
    repository. In order to fulfill the request any of predicates must be meet
 
    """
 

	
 
    def check_permissions(self):
 
        repo_name = get_repo_slug(request)
 

	
 
        try:
 
            user_perms = set([self.user_perms['repositories'][repo_name]])
 
        except KeyError:
 
            return False
 
        if self.required_perms.intersection(user_perms):
 
            return True
 
        return False
 
#===============================================================================
 
# CHECK FUNCTIONS
 
#===============================================================================
 

	
 
class PermsFunction(object):
 
    """Base function for other check functions"""
 

	
 
    def __init__(self, *perms):
 
        available_perms = config['available_permissions']
 

	
 
        for perm in perms:
 
            if perm not in available_perms:
 
                raise Exception("'%s' permission in not defined" % perm)
 
        self.required_perms = set(perms)
 
        self.user_perms = None
 
        self.granted_for = ''
 
        self.repo_name = None
 

	
 
    def __call__(self, check_Location=''):
 
        user = session.get('rhodecode_user', False)
 
        if not user:
 
            return False
 
        self.user_perms = user.permissions
 
        self.granted_for = user.username
 
        log.debug('checking %s %s %s', self.__class__.__name__,
 
                  self.required_perms, user)
 

	
 
        if self.check_permissions():
 
            log.debug('Permission granted for %s @ %s %s', self.granted_for,
 
                      check_Location, user)
 
            return True
 

	
 
        else:
 
            log.warning('Permission denied for %s @ %s %s', self.granted_for,
 
                        check_Location, user)
 
            return False
 

	
 
    def check_permissions(self):
 
        """Dummy function for overriding"""
 
        raise Exception('You have to write this function in child class')
 

	
 
class HasPermissionAll(PermsFunction):
 
    def check_permissions(self):
 
        if self.required_perms.issubset(self.user_perms.get('global')):
 
            return True
 
        return False
 

	
 
class HasPermissionAny(PermsFunction):
 
    def check_permissions(self):
 
        if self.required_perms.intersection(self.user_perms.get('global')):
 
            return True
 
        return False
 

	
 
class HasRepoPermissionAll(PermsFunction):
 

	
 
    def __call__(self, repo_name=None, check_Location=''):
 
        self.repo_name = repo_name
 
        return super(HasRepoPermissionAll, self).__call__(check_Location)
 

	
 
    def check_permissions(self):
 
        if not self.repo_name:
 
            self.repo_name = get_repo_slug(request)
 

	
 
        try:
 
            self.user_perms = set([self.user_perms['repositories']\
 
                                   [self.repo_name]])
 
        except KeyError:
 
            return False
 
        self.granted_for = self.repo_name
 
        if self.required_perms.issubset(self.user_perms):
 
            return True
 
        return False
 

	
 
class HasRepoPermissionAny(PermsFunction):
 

	
 
    def __call__(self, repo_name=None, check_Location=''):
 
        self.repo_name = repo_name
 
        return super(HasRepoPermissionAny, self).__call__(check_Location)
 

	
 
    def check_permissions(self):
 
        if not self.repo_name:
 
            self.repo_name = get_repo_slug(request)
 

	
 
        try:
 
            self.user_perms = set([self.user_perms['repositories']\
 
                                   [self.repo_name]])
 
        except KeyError:
 
            return False
 
        self.granted_for = self.repo_name
 
        if self.required_perms.intersection(self.user_perms):
 
            return True
 
        return False
 

	
 
#===============================================================================
 
# SPECIAL VERSION TO HANDLE MIDDLEWARE AUTH
 
#===============================================================================
 

	
 
class HasPermissionAnyMiddleware(object):
 
    def __init__(self, *perms):
 
        self.required_perms = set(perms)
 

	
 
    def __call__(self, user, repo_name):
 
        usr = AuthUser()
 
        usr.user_id = user.user_id
 
        usr.username = user.username
 
        usr.is_admin = user.admin
 

	
 
        try:
 
            self.user_perms = set([fill_perms(usr)\
 
                                   .permissions['repositories'][repo_name]])
 
        except:
 
            self.user_perms = set()
 
        self.granted_for = ''
 
        self.username = user.username
 
        self.repo_name = repo_name
 
        return self.check_permissions()
 

	
 
    def check_permissions(self):
 
        log.debug('checking mercurial protocol '
 
                  'permissions for user:%s repository:%s',
 
                  'permissions %s for user:%s repository:%s', self.user_perms,
 
                                                self.username, self.repo_name)
 
        if self.required_perms.intersection(self.user_perms):
 
            log.debug('permission granted')
 
            return True
 
        log.debug('permission denied')
 
        return False
rhodecode/lib/helpers.py
Show inline comments
 
"""Helper functions
 

	
 
Consists of functions to typically be used within templates, but also
 
available to Controllers. This module is available to both as 'h'.
 
"""
 
import random
 
import hashlib
 
import StringIO
 
from pygments.formatters import HtmlFormatter
 
from pygments import highlight as code_highlight
 
from pylons import url
 
from pylons.i18n.translation import _, ungettext
 
from vcs.utils.annotate import annotate_highlight
 
from rhodecode.lib.utils import repo_name_slug
 

	
 
from webhelpers.html import literal, HTML, escape
 
from webhelpers.html.tools import *
 
from webhelpers.html.builder import make_tag
 
from webhelpers.html.tags import auto_discovery_link, checkbox, css_classes, \
 
    end_form, file, form, hidden, image, javascript_link, link_to, link_to_if, \
 
    link_to_unless, ol, required_legend, select, stylesheet_link, submit, text, \
 
    password, textarea, title, ul, xml_declaration, radio
 
from webhelpers.html.tools import auto_link, button_to, highlight, js_obfuscate, \
 
    mail_to, strip_links, strip_tags, tag_re
 
from webhelpers.number import format_byte_size, format_bit_size
 
from webhelpers.pylonslib import Flash as _Flash
 
from webhelpers.pylonslib.secure_form import secure_form
 
from webhelpers.text import chop_at, collapse, convert_accented_entities, \
 
    convert_misc_entities, lchop, plural, rchop, remove_formatting, \
 
    replace_whitespace, urlify, truncate, wrap_paragraphs
 
from webhelpers.date import time_ago_in_words
 

	
 
from webhelpers.html.tags import _set_input_attrs, _set_id_attr, \
 
    convert_boolean_attrs, NotGiven
 

	
 
def _reset(name, value=None, id=NotGiven, type="reset", **attrs):
 
    """Reset button
 
    """
 
    _set_input_attrs(attrs, type, name, value)
 
    _set_id_attr(attrs, id, name)
 
    convert_boolean_attrs(attrs, ["disabled"])
 
    return HTML.input(**attrs)
 

	
 
reset = _reset
 

	
 

	
 
def get_token():
 
    """Return the current authentication token, creating one if one doesn't
 
    already exist.
 
    """
 
    token_key = "_authentication_token"
 
    from pylons import session
 
    if not token_key in session:
 
        try:
 
            token = hashlib.sha1(str(random.getrandbits(128))).hexdigest()
 
        except AttributeError: # Python < 2.4
 
            token = hashlib.sha1(str(random.randrange(2 ** 128))).hexdigest()
 
        session[token_key] = token
 
        if hasattr(session, 'save'):
 
            session.save()
 
    return session[token_key]
 

	
 
class _GetError(object):
 
    """Get error from form_errors, and represent it as span wrapped error
 
    message
 
    
 
    :param field_name: field to fetch errors for
 
    :param form_errors: form errors dict
 
    """
 

	
 
    def __call__(self, field_name, form_errors):
 
        tmpl = """<span class="error_msg">%s</span>"""
 
        if form_errors and form_errors.has_key(field_name):
 
            return literal(tmpl % form_errors.get(field_name))
 

	
 
get_error = _GetError()
 

	
 
class _ToolTip(object):
 

	
 
    def __call__(self, tooltip_title, trim_at=50):
 
        """Special function just to wrap our text into nice formatted 
 
        autowrapped text
 
        
 
        :param tooltip_title:
 
        """
 

	
 
        return wrap_paragraphs(escape(tooltip_title), trim_at)\
 
                       .replace('\n', '<br/>')
 

	
 
    def activate(self):
 
        """Adds tooltip mechanism to the given Html all tooltips have to have 
 
        set class `tooltip` and set attribute `tooltip_title`.
 
        Then a tooltip will be generated based on that. All with yui js tooltip
 
        """
 

	
 
        js = '''
 
        YAHOO.util.Event.onDOMReady(function(){
 
            function toolTipsId(){
 
                var ids = [];
 
                var tts = YAHOO.util.Dom.getElementsByClassName('tooltip');
 
                
 
                for (var i = 0; i < tts.length; i++) {
 
                    //if element doesn't not have and id autogenerate one for tooltip
 
                    
 
                    if (!tts[i].id){
 
                        tts[i].id='tt'+i*100;
 
                    }
 
                    ids.push(tts[i].id);
 
                }
 
                return ids        
 
            };
 
            var myToolTips = new YAHOO.widget.Tooltip("tooltip", { 
 
                context: toolTipsId(),
 
                monitorresize:false,
 
                xyoffset :[0,0],
 
                autodismissdelay:300000,
 
                hidedelay:5,
 
                showdelay:20,
 
            });
 
            
 
            // Set the text for the tooltip just before we display it. Lazy method
 
            myToolTips.contextTriggerEvent.subscribe( 
 
                 function(type, args) { 
 

	
 
                        var context = args[0]; 
 
                        
 
                        //positioning of tooltip
 
                        var tt_w = this.element.clientWidth;//tooltip width
 
                        var tt_h = this.element.clientHeight;//tooltip height
 
                        
 
                        var context_w = context.offsetWidth;
 
                        var context_h = context.offsetHeight;
 
                        
 
                        var pos_x = YAHOO.util.Dom.getX(context);
 
                        var pos_y = YAHOO.util.Dom.getY(context);
 

	
 
                        var display_strategy = 'right';
 
                        var xy_pos = [0,0];
 
                        switch (display_strategy){
 
                        
 
                            case 'top':
 
                                var cur_x = (pos_x+context_w/2)-(tt_w/2);
 
                                var cur_y = (pos_y-tt_h-4);
 
                                xy_pos = [cur_x,cur_y];                                
 
                                break;
 
                            case 'bottom':
 
                                var cur_x = (pos_x+context_w/2)-(tt_w/2);
 
                                var cur_y = pos_y+context_h+4;
 
                                xy_pos = [cur_x,cur_y];                                
 
                                break;
 
                            case 'left':
 
                                var cur_x = (pos_x-tt_w-4);
 
                                var cur_y = pos_y-((tt_h/2)-context_h/2);
 
                                xy_pos = [cur_x,cur_y];                                
 
                                break;
 
                            case 'right':
 
                                var cur_x = (pos_x+context_w+4);
 
                                var cur_y = pos_y-((tt_h/2)-context_h/2);
 
                                xy_pos = [cur_x,cur_y];                                
 
                                break;
 
                             default:
 
                                var cur_x = (pos_x+context_w/2)-(tt_w/2);
 
                                var cur_y = pos_y-tt_h-4;
 
                                xy_pos = [cur_x,cur_y];                                
 
                                break;                             
 
                                 
 
                        }
 

	
 
                        this.cfg.setProperty("xy",xy_pos);
 

	
 
                  });
 
                  
 
            //Mouse out 
 
            myToolTips.contextMouseOutEvent.subscribe(
 
                function(type, args) {
 
                    var context = args[0];
 
                    
 
                });
 
        });
 
        '''
 
        return literal(js)
 

	
 
tooltip = _ToolTip()
 

	
 
class _FilesBreadCrumbs(object):
 

	
 
    def __call__(self, repo_name, rev, paths):
 
        if isinstance(paths, str):
 
            paths = paths.decode('utf-8')
 
        url_l = [link_to(repo_name, url('files_home',
 
                                        repo_name=repo_name,
 
                                        revision=rev, f_path=''))]
 
        paths_l = paths.split('/')
 
        for cnt, p in enumerate(paths_l):
 
            if p != '':
 
                url_l.append(link_to(p, url('files_home',
 
                                            repo_name=repo_name,
 
                                            revision=rev,
 
                                            f_path='/'.join(paths_l[:cnt + 1]))))
 

	
 
        return literal('/'.join(url_l))
 

	
 
files_breadcrumbs = _FilesBreadCrumbs()
 

	
 
class CodeHtmlFormatter(HtmlFormatter):
 
    """My code Html Formatter for source codes
 
    """
 

	
 
    def wrap(self, source, outfile):
 
        return self._wrap_div(self._wrap_pre(self._wrap_code(source)))
 

	
 
    def _wrap_code(self, source):
 
        for cnt, it in enumerate(source):
 
            i, t = it
 
            t = '<div id="L%s">%s</div>' % (cnt + 1, t)
 
            yield i, t
 

	
 
    def _wrap_tablelinenos(self, inner):
 
        dummyoutfile = StringIO.StringIO()
 
        lncount = 0
 
        for t, line in inner:
 
            if t:
 
                lncount += 1
 
            dummyoutfile.write(line)
 

	
 
        fl = self.linenostart
 
        mw = len(str(lncount + fl - 1))
 
        sp = self.linenospecial
 
        st = self.linenostep
 
        la = self.lineanchors
 
        aln = self.anchorlinenos
 
        nocls = self.noclasses
 
        if sp:
 
            lines = []
 

	
 
            for i in range(fl, fl + lncount):
 
                if i % st == 0:
 
                    if i % sp == 0:
 
                        if aln:
 
                            lines.append('<a href="#%s%d" class="special">%*d</a>' %
 
                                         (la, i, mw, i))
 
                        else:
 
                            lines.append('<span class="special">%*d</span>' % (mw, i))
 
                    else:
 
                        if aln:
 
                            lines.append('<a href="#%s%d">%*d</a>' % (la, i, mw, i))
 
                        else:
 
                            lines.append('%*d' % (mw, i))
 
                else:
 
                    lines.append('')
 
            ls = '\n'.join(lines)
 
        else:
 
            lines = []
 
            for i in range(fl, fl + lncount):
 
                if i % st == 0:
 
                    if aln:
 
                        lines.append('<a href="#%s%d">%*d</a>' % (la, i, mw, i))
 
                    else:
 
                        lines.append('%*d' % (mw, i))
 
                else:
 
                    lines.append('')
 
            ls = '\n'.join(lines)
 

	
 
        # in case you wonder about the seemingly redundant <div> here: since the
 
        # content in the other cell also is wrapped in a div, some browsers in
 
        # some configurations seem to mess up the formatting...
 
        if nocls:
 
            yield 0, ('<table class="%stable">' % self.cssclass +
 
                      '<tr><td><div class="linenodiv" '
 
                      'style="background-color: #f0f0f0; padding-right: 10px">'
 
                      '<pre style="line-height: 125%">' +
 
                      ls + '</pre></div></td><td class="code">')
 
        else:
 
            yield 0, ('<table class="%stable">' % self.cssclass +
 
                      '<tr><td class="linenos"><div class="linenodiv"><pre>' +
 
                      ls + '</pre></div></td><td class="code">')
 
        yield 0, dummyoutfile.getvalue()
 
        yield 0, '</td></tr></table>'
 

	
 

	
 
def pygmentize(filenode, **kwargs):
 
    """pygmentize function using pygments
 
    
 
    :param filenode:
 
    """
 

	
 
    return literal(code_highlight(filenode.content,
 
                                  filenode.lexer, CodeHtmlFormatter(**kwargs)))
 

	
 
def pygmentize_annotation(filenode, **kwargs):
 
    """pygmentize function for annotation
 
    
 
    :param filenode:
 
    """
 

	
 
    color_dict = {}
 
    def gen_color(n=10000):
 
        """generator for getting n of evenly distributed colors using 
 
        hsv color and golden ratio. It always return same order of colors
 
        
 
        :returns: RGB tuple
 
        """
 
        import colorsys
 
        golden_ratio = 0.618033988749895
 
        h = 0.22717784590367374
 

	
 
        for c in xrange(n):
 
            h += golden_ratio
 
            h %= 1
 
            HSV_tuple = [h, 0.95, 0.95]
 
            RGB_tuple = colorsys.hsv_to_rgb(*HSV_tuple)
 
            yield map(lambda x:str(int(x * 256)), RGB_tuple)
 

	
 
    cgenerator = gen_color()
 

	
 
    def get_color_string(cs):
 
        if color_dict.has_key(cs):
 
            col = color_dict[cs]
 
        else:
 
            col = color_dict[cs] = cgenerator.next()
 
        return "color: rgb(%s)! important;" % (', '.join(col))
 

	
 
    def url_func(changeset):
 
        tooltip_html = "<div style='font-size:0.8em'><b>Author:</b>" + \
 
        " %s<br/><b>Date:</b> %s</b><br/><b>Message:</b> %s<br/></div>"
 

	
 
        tooltip_html = tooltip_html % (changeset.author,
 
                                               changeset.date,
 
                                               tooltip(changeset.message))
 
        lnk_format = '%5s:%s' % ('r%s' % changeset.revision,
 
                                 short_id(changeset.raw_id))
 
        uri = link_to(
 
                lnk_format,
 
                url('changeset_home', repo_name=changeset.repository.name,
 
                    revision=changeset.raw_id),
 
                style=get_color_string(changeset.raw_id),
 
                class_='tooltip',
 
                title=tooltip_html
 
              )
 

	
 
        uri += '\n'
 
        return uri
 
    return literal(annotate_highlight(filenode, url_func, **kwargs))
 

	
 
def get_changeset_safe(repo, rev):
 
    from vcs.backends.base import BaseRepository
 
    from vcs.exceptions import RepositoryError
 
    if not isinstance(repo, BaseRepository):
 
        raise Exception('You must pass an Repository '
 
                        'object as first argument got %s', type(repo))
 

	
 
    try:
 
        cs = repo.get_changeset(rev)
 
    except RepositoryError:
 
        from rhodecode.lib.utils import EmptyChangeset
 
        cs = EmptyChangeset()
 
    return cs
 

	
 

	
 
def is_following_repo(repo_name, user_id):
 
    from rhodecode.model.scm import ScmModel
 
    return ScmModel().is_following_repo(repo_name, user_id)
 

	
 
flash = _Flash()
 

	
 

	
 
#==============================================================================
 
# MERCURIAL FILTERS available via h.
 
#==============================================================================
 
from mercurial import util
 
from mercurial.templatefilters import person as _person
 

	
 
def _age(curdate):
 
    """turns a datetime into an age string."""
 

	
 
    if not curdate:
 
        return ''
 

	
 
    from datetime import timedelta, datetime
 

	
 
    agescales = [("year", 3600 * 24 * 365),
 
                 ("month", 3600 * 24 * 30),
 
                 ("day", 3600 * 24),
 
                 ("hour", 3600),
 
                 ("minute", 60),
 
                 ("second", 1), ]
 

	
 
    age = datetime.now() - curdate
 
    age_seconds = (age.days * agescales[2][1]) + age.seconds
 
    pos = 1
 
    for scale in agescales:
 
        if scale[1] <= age_seconds:
 
            if pos == 6:pos = 5
 
            return time_ago_in_words(curdate, agescales[pos][0]) + ' ' + _('ago')
 
        pos += 1
 

	
 
    return _('just now')
 

	
 
age = lambda  x:_age(x)
 
capitalize = lambda x: x.capitalize()
 
email = util.email
 
email_or_none = lambda x: util.email(x) if util.email(x) != x else None
 
person = lambda x: _person(x)
 
short_id = lambda x: x[:12]
 

	
 

	
 
def bool2icon(value):
 
    """Returns True/False values represented as small html image of true/false
 
    icons
 
    
 
    :param value: bool value
 
    """
 

	
 
    if value is True:
 
        return HTML.tag('img', src="/images/icons/accept.png", alt=_('True'))
 

	
 
    if value is False:
 
        return HTML.tag('img', src="/images/icons/cancel.png", alt=_('False'))
 

	
 
    return value
 

	
 

	
 
def action_parser(user_log):
 
    """This helper will map the specified string action into translated
 
    fancy names with icons and links
 
    
 
    :param user_log: user log instance
 
    """
 

	
 
    action = user_log.action
 
    action_params = ' '
 

	
 
    x = action.split(':')
 

	
 
    if len(x) > 1:
 
        action, action_params = x
 

	
 
    def get_cs_links():
 
        revs_limit = 5 #display this amount always
 
        revs_top_limit = 50 #show upto this amount of changesets hidden
 
        revs = action_params.split(',')
 
        repo_name = user_log.repository.repo_name
 
        from rhodecode.model.scm import ScmModel
 

	
 
        message = lambda rev: get_changeset_safe(ScmModel().get(repo_name),
 
                                                 rev).message
 
        repo, dbrepo = ScmModel().get(repo_name, retval='repo')
 
        message = lambda rev: get_changeset_safe(repo, rev).message
 

	
 
        cs_links = " " + ', '.join ([link_to(rev,
 
                url('changeset_home',
 
                repo_name=repo_name,
 
                revision=rev), title=tooltip(message(rev)),
 
                class_='tooltip') for rev in revs[:revs_limit] ])
 

	
 
        compare_view = (' <div class="compare_view tooltip" title="%s">'
 
                        '<a href="%s">%s</a> '
 
                        '</div>' % (_('Show all combined changesets %s->%s' \
 
                                      % (revs[0], revs[-1])),
 
                                    url('changeset_home', repo_name=repo_name,
 
                                        revision='%s...%s' % (revs[0], revs[-1])
 
                                    ),
 
                                    _('compare view'))
 
                        )
 

	
 
        if len(revs) > revs_limit:
 
            uniq_id = revs[0]
 
            html_tmpl = ('<span> %s '
 
            '<a class="show_more" id="_%s" href="#more">%s</a> '
 
            '%s</span>')
 
            cs_links += html_tmpl % (_('and'), uniq_id, _('%s more') \
 
                                        % (len(revs) - revs_limit),
 
                                        _('revisions'))
 

	
 
            html_tmpl = '<span id="%s" style="display:none"> %s </span>'
 
            cs_links += html_tmpl % (uniq_id, ', '.join([link_to(rev,
 
                url('changeset_home',
 
                repo_name=repo_name, revision=rev),
 
                title=message(rev), class_='tooltip')
 
                for rev in revs[revs_limit:revs_top_limit]]))
 
        if len(revs) > 1:
 
            cs_links += compare_view
 
        return cs_links
 

	
 
    def get_fork_name():
 
        from rhodecode.model.scm import ScmModel
 
        repo_name = action_params
 
        repo, dbrepo = ScmModel().get(repo_name)
 
        if repo is None:
 
            return repo_name
 
        return link_to(action_params, url('summary_home',
 
                                          repo_name=repo.name,),
 
                                          title=dbrepo.description)
 

	
 
    map = {'user_deleted_repo':(_('User [deleted] repository'), None),
 
           'user_created_repo':(_('User [created] repository'), None),
 
           'user_forked_repo':(_('User [forked] repository as:'), get_fork_name),
 
           'user_updated_repo':(_('User [updated] repository'), None),
 
           'admin_deleted_repo':(_('Admin [delete] repository'), None),
 
           'admin_created_repo':(_('Admin [created] repository'), None),
 
           'admin_forked_repo':(_('Admin [forked] repository'), None),
 
           'admin_updated_repo':(_('Admin [updated] repository'), None),
 
           'push':(_('[Pushed]'), get_cs_links),
 
           'pull':(_('[Pulled]'), None),
 
           'started_following_repo':(_('User [started following] repository'), None),
 
           'stopped_following_repo':(_('User [stopped following] repository'), None),
 
            }
 

	
 
    action_str = map.get(action, action)
 
    action = action_str[0].replace('[', '<span class="journal_highlight">')\
 
                   .replace(']', '</span>')
 
    if action_str[1] is not None:
 
        action = action + " " + action_str[1]()
 

	
 
    return literal(action)
 

	
 
def action_parser_icon(user_log):
 
    action = user_log.action
 
    action_params = None
 
    x = action.split(':')
 

	
 
    if len(x) > 1:
 
        action, action_params = x
 

	
 
    tmpl = """<img src="/images/icons/%s" alt="%s"/>"""
 
    map = {'user_deleted_repo':'database_delete.png',
 
           'user_created_repo':'database_add.png',
 
           'user_forked_repo':'arrow_divide.png',
 
           'user_updated_repo':'database_edit.png',
 
           'admin_deleted_repo':'database_delete.png',
 
           'admin_created_repo':'database_add.png',
 
           'admin_forked_repo':'arrow_divide.png',
 
           'admin_updated_repo':'database_edit.png',
 
           'push':'script_add.png',
 
           'pull':'down_16.png',
 
           'started_following_repo':'heart_add.png',
 
           'stopped_following_repo':'heart_delete.png',
 
            }
 
    return literal(tmpl % (map.get(action, action), action))
 

	
 

	
 
#==============================================================================
 
# PERMS
 
#==============================================================================
 
from rhodecode.lib.auth import HasPermissionAny, HasPermissionAll, \
 
HasRepoPermissionAny, HasRepoPermissionAll
 

	
 
#==============================================================================
 
# GRAVATAR URL
 
#==============================================================================
 
import hashlib
 
import urllib
 
from pylons import request
 

	
 
def gravatar_url(email_address, size=30):
 
    ssl_enabled = 'https' == request.environ.get('wsgi.url_scheme')
 
    default = 'identicon'
 
    baseurl_nossl = "http://www.gravatar.com/avatar/"
 
    baseurl_ssl = "https://secure.gravatar.com/avatar/"
 
    baseurl = baseurl_ssl if ssl_enabled else baseurl_nossl
 

	
 

	
 
    # construct the url
 
    gravatar_url = baseurl + hashlib.md5(email_address.lower()).hexdigest() + "?"
 
    gravatar_url += urllib.urlencode({'d':default, 's':str(size)})
 

	
 
    return gravatar_url
 

	
 
def safe_unicode(str):
 
    """safe unicode function. In case of UnicodeDecode error we try to return
 
    unicode with errors replace, if this failes we return unicode with 
 
    string_escape decoding """
 

	
 
    try:
 
        u_str = unicode(str)
 
    except UnicodeDecodeError:
 
        try:
 
            u_str = unicode(str, 'utf-8', 'replace')
 
        except UnicodeDecodeError:
 
            #incase we have a decode error just represent as byte string
 
            u_str = unicode(str(str).encode('string_escape'))
 

	
 
    return u_str
 

	
 
def changed_tooltip(nodes):
 
    if nodes:
 
        pref = ': <br/> '
 
        suf = ''
 
        if len(nodes) > 30:
 
            suf = '<br/>' + _(' and %s more') % (len(nodes) - 30)
 
        return literal(pref + '<br/> '.join([x.path for x in nodes[:30]]) + suf)
 
    else:
 
        return ': ' + _('No Files')
0 comments (0 inline, 0 general)