auth.py
609 lines
| 22.6 KiB
| text/x-python
|
PythonLexer
r895 | # -*- 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. | ||||
""" | ||||
r547 | # 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. | ||||
r895 | import bcrypt | |||
import random | ||||
import logging | ||||
import traceback | ||||
from decorator import decorator | ||||
r547 | from pylons import config, session, url, request | |||
from pylons.controllers.util import abort, redirect | ||||
r895 | ||||
from rhodecode.lib.exceptions import LdapPasswordError, LdapUsernameError | ||||
r547 | from rhodecode.lib.utils import get_repo_slug | |||
r713 | from rhodecode.lib.auth_ldap import AuthLdap | |||
r895 | ||||
r547 | from rhodecode.model import meta | |||
r673 | from rhodecode.model.user import UserModel | |||
r547 | from rhodecode.model.db import User, RepoToPerm, Repository, Permission, \ | |||
r1016 | UserToPerm, UsersGroupToPerm, UsersGroupMember | |||
r895 | ||||
r547 | ||||
r673 | log = logging.getLogger(__name__) | |||
r547 | ||||
r1016 | ||||
PERM_WEIGHTS = {'repository.none':0, | ||||
'repository.read':1, | ||||
'repository.write':3, | ||||
'repository.admin':3} | ||||
r547 | 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] | ||||
r673 | ||||
r547 | 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 | ||||
r673 | ||||
r547 | def get_crypt_password(password): | |||
r1016 | """Cryptographic function used for password hashing based on pybcrypt | |||
r604 | :param password: password to hash | |||
r673 | """ | |||
r547 | return bcrypt.hashpw(password, bcrypt.gensalt(10)) | |||
def check_password(password, hashed): | ||||
return bcrypt.hashpw(password, hashed) == hashed | ||||
def authfunc(environ, username, password): | ||||
r1016 | """Dummy authentication function used in Mercurial/Git/ and access control, | |||
r761 | ||||
:param environ: needed only for using in Basic auth | ||||
""" | ||||
return authenticate(username, password) | ||||
def authenticate(username, password): | ||||
r1016 | """Authentication function used for access control, | |||
r699 | firstly checks for db authentication then if ldap is enabled for ldap | |||
r705 | authentication, also creates ldap user if not in database | |||
r699 | :param username: username | |||
:param password: password | ||||
""" | ||||
r705 | user_model = UserModel() | |||
user = user_model.get_by_username(username, cache=False) | ||||
r699 | ||||
r761 | log.debug('Authenticating user using RhodeCode account') | |||
|
r991 | if user is not None and not user.ldap_dn: | ||
r547 | if user.active: | |||
r674 | ||||
if user.username == 'default' and user.active: | ||||
r761 | log.info('user %s authenticated correctly as anonymous user', | |||
username) | ||||
r674 | return True | |||
elif user.username == username and check_password(password, user.password): | ||||
r547 | log.info('user %s authenticated correctly', username) | |||
return True | ||||
else: | ||||
r761 | log.warning('user %s is disabled', username) | |||
r705 | ||||
else: | ||||
r761 | log.debug('Regular authentication failed') | |||
r742 | user_obj = user_model.get_by_username(username, cache=False, | |||
case_insensitive=True) | ||||
r761 | ||||
|
r991 | if user_obj is not None and not user_obj.ldap_dn: | ||
r749 | log.debug('this user already exists as non ldap') | |||
r748 | return False | |||
r705 | from rhodecode.model.settings import SettingsModel | |||
ldap_settings = SettingsModel().get_ldap_settings() | ||||
#====================================================================== | ||||
r1016 | # FALLBACK TO LDAP AUTH IF ENABLE | |||
r705 | #====================================================================== | |||
if ldap_settings.get('ldap_active', False): | ||||
r761 | log.debug("Authenticating user using ldap") | |||
r705 | 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'), | ||||
|
r991 | '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'), | ||||
r705 | 'ldap_version':3, | |||
} | ||||
log.debug('Checking for ldap authentication') | ||||
try: | ||||
aldap = AuthLdap(**kwargs) | ||||
|
r991 | (user_dn, ldap_attrs) = aldap.authenticate_ldap(username, password) | ||
log.debug('Got ldap DN response %s', user_dn) | ||||
r705 | ||||
|
r991 | 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): | ||||
r1016 | log.info('created new ldap user %s', username) | |||
r705 | ||||
r761 | return True | |||
except (LdapUsernameError, LdapPasswordError,): | ||||
pass | ||||
except (Exception,): | ||||
r705 | log.error(traceback.format_exc()) | |||
r761 | pass | |||
r547 | return False | |||
class AuthUser(object): | ||||
r1016 | """A simple object that handles a mercurial username for authentication | |||
r547 | """ | |||
r1016 | ||||
r547 | 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 = {} | ||||
r673 | def __repr__(self): | |||
return "<AuthUser('id:%s:%s')>" % (self.user_id, self.username) | ||||
r547 | ||||
def set_available_permissions(config): | ||||
r895 | """This function will propagate pylons globals with all available defined | |||
r1016 | permission given in db. We don't want to check each time from db for new | |||
r547 | permissions since adding a new permission also requires application restart | |||
ie. to decorate new views with the newly created permission | ||||
r895 | ||||
:param config: current pylons config instance | ||||
r547 | """ | |||
log.info('getting information about all available permissions') | ||||
try: | ||||
r629 | sa = meta.Session() | |||
r547 | all_perms = sa.query(Permission).all() | |||
r629 | except: | |||
pass | ||||
r547 | finally: | |||
meta.Session.remove() | ||||
r673 | ||||
r547 | config['available_permissions'] = [x.permission_name for x in all_perms] | |||
def fill_perms(user): | ||||
r895 | """Fills user permission attribute with permissions taken from database | |||
r1016 | works for permissions given for repositories, and for permissions that | |||
as part of beeing group member | ||||
r895 | ||||
r1016 | :param user: user instance to fill his perms | |||
r547 | """ | |||
r673 | ||||
r629 | sa = meta.Session() | |||
r547 | user.permissions['repositories'] = {} | |||
user.permissions['global'] = set() | ||||
r673 | ||||
r547 | #=========================================================================== | |||
# fetch default permissions | ||||
#=========================================================================== | ||||
r692 | default_user = UserModel().get_by_username('default', cache=True) | |||
r673 | ||||
r547 | default_perms = sa.query(RepoToPerm, Repository, Permission)\ | |||
.join((Repository, RepoToPerm.repository_id == Repository.repo_id))\ | ||||
.join((Permission, RepoToPerm.permission_id == Permission.permission_id))\ | ||||
r609 | .filter(RepoToPerm.user == default_user).all() | |||
r673 | ||||
r547 | if user.is_admin: | |||
#======================================================================= | ||||
# #admin have all default rights set to admin | ||||
#======================================================================= | ||||
user.permissions['global'].add('hg.admin') | ||||
r673 | ||||
r547 | for perm in default_perms: | |||
p = 'repository.admin' | ||||
user.permissions['repositories'][perm.RepoToPerm.repository.repo_name] = p | ||||
r673 | ||||
r547 | else: | |||
#======================================================================= | ||||
# set default permissions | ||||
#======================================================================= | ||||
r673 | ||||
r547 | #default global | |||
default_global_perms = sa.query(UserToPerm)\ | ||||
r741 | .filter(UserToPerm.user == sa.query(User)\ | |||
.filter(User.username == 'default').one()) | ||||
r673 | ||||
r547 | for perm in default_global_perms: | |||
user.permissions['global'].add(perm.permission.permission_name) | ||||
r673 | ||||
r1016 | #default for repositories | |||
r547 | 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 | ||||
r673 | ||||
r547 | user.permissions['repositories'][perm.RepoToPerm.repository.repo_name] = p | |||
r673 | ||||
r547 | #======================================================================= | |||
r1016 | # overwrite default with user permissions if any | |||
r547 | #======================================================================= | |||
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() | ||||
r673 | ||||
r547 | 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 | ||||
r1016 | ||||
#======================================================================= | ||||
# 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 | ||||
r673 | meta.Session.remove() | |||
r547 | return user | |||
r673 | ||||
r547 | def get_user(session): | |||
r1016 | """Gets user from session, and wraps permissions into user | |||
r604 | :param session: | |||
r547 | """ | |||
r548 | user = session.get('rhodecode_user', AuthUser()) | |||
r673 | #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 | ||||
r686 | user.user_id = anonymous_user.user_id | |||
r673 | else: | |||
user.is_authenticated = False | ||||
r547 | if user.is_authenticated: | |||
r673 | user = UserModel().fill_data(user) | |||
r547 | user = fill_perms(user) | |||
r548 | session['rhodecode_user'] = user | |||
r547 | session.save() | |||
return user | ||||
r673 | ||||
r547 | #=============================================================================== | |||
# CHECK DECORATORS | ||||
#=============================================================================== | ||||
class LoginRequired(object): | ||||
r779 | """Must be logged in to execute this function else | |||
redirect to login page""" | ||||
r673 | ||||
r547 | def __call__(self, func): | |||
return decorator(self.__wrapper, func) | ||||
r673 | ||||
r547 | def __wrapper(self, func, *fargs, **fkwargs): | |||
r548 | user = session.get('rhodecode_user', AuthUser()) | |||
r547 | 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) | ||||
r673 | ||||
r547 | p = '' | |||
if request.environ.get('SCRIPT_NAME') != '/': | ||||
p += request.environ.get('SCRIPT_NAME') | ||||
r673 | ||||
r547 | p += request.environ.get('PATH_INFO') | |||
if request.environ.get('QUERY_STRING'): | ||||
p += '?' + request.environ.get('QUERY_STRING') | ||||
r673 | ||||
log.debug('redirecting to login page with %s', p) | ||||
r547 | return redirect(url('login_home', came_from=p)) | |||
r779 | 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) | ||||
r547 | class PermsDecorator(object): | |||
"""Base class for decorators""" | ||||
r673 | ||||
r547 | 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 | ||||
r673 | ||||
r547 | def __call__(self, func): | |||
return decorator(self.__wrapper, func) | ||||
r673 | ||||
r547 | def __wrapper(self, func, *fargs, **fkwargs): | |||
# _wrapper.__name__ = func.__name__ | ||||
# _wrapper.__dict__.update(func.__dict__) | ||||
# _wrapper.__doc__ = func.__doc__ | ||||
r673 | 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) | ||||
r547 | ||||
if self.check_permissions(): | ||||
r673 | log.debug('Permission granted for %s %s', func.__name__, self.user) | |||
r547 | return func(*fargs, **fkwargs) | |||
r673 | ||||
r547 | else: | |||
r673 | log.warning('Permission denied for %s %s', func.__name__, self.user) | |||
r547 | #redirect with forbidden ret code | |||
return abort(403) | ||||
r673 | ||||
r547 | 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 | ||||
""" | ||||
r673 | ||||
r547 | def check_permissions(self): | |||
if self.required_perms.issubset(self.user_perms.get('global')): | ||||
return True | ||||
return False | ||||
r673 | ||||
r547 | ||||
class HasPermissionAnyDecorator(PermsDecorator): | ||||
"""Checks for access permission for any of given predicates. In order to | ||||
fulfill the request any of predicates must be meet | ||||
""" | ||||
r673 | ||||
r547 | 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 | ||||
""" | ||||
r673 | ||||
r547 | 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 | ||||
r673 | ||||
r547 | ||||
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 | ||||
""" | ||||
r673 | ||||
r547 | def check_permissions(self): | |||
repo_name = get_repo_slug(request) | ||||
r673 | ||||
r547 | 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""" | ||||
r673 | ||||
r547 | def __init__(self, *perms): | |||
available_perms = config['available_permissions'] | ||||
r673 | ||||
r547 | 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 | ||||
r673 | ||||
r547 | def __call__(self, check_Location=''): | |||
r548 | user = session.get('rhodecode_user', False) | |||
r547 | if not user: | |||
return False | ||||
self.user_perms = user.permissions | ||||
r673 | self.granted_for = user.username | |||
log.debug('checking %s %s %s', self.__class__.__name__, | ||||
self.required_perms, user) | ||||
r547 | if self.check_permissions(): | |||
r673 | log.debug('Permission granted for %s @ %s %s', self.granted_for, | |||
check_Location, user) | ||||
r547 | return True | |||
r673 | ||||
r547 | else: | |||
r673 | log.warning('Permission denied for %s @ %s %s', self.granted_for, | |||
check_Location, user) | ||||
return False | ||||
r547 | def check_permissions(self): | |||
"""Dummy function for overriding""" | ||||
raise Exception('You have to write this function in child class') | ||||
r673 | ||||
r547 | 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): | ||||
r673 | ||||
r547 | def __call__(self, repo_name=None, check_Location=''): | |||
self.repo_name = repo_name | ||||
return super(HasRepoPermissionAll, self).__call__(check_Location) | ||||
r673 | ||||
r547 | 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 | ||||
r673 | self.granted_for = self.repo_name | |||
r547 | if self.required_perms.issubset(self.user_perms): | |||
return True | ||||
return False | ||||
r673 | ||||
r547 | class HasRepoPermissionAny(PermsFunction): | |||
r673 | ||||
r547 | def __call__(self, repo_name=None, check_Location=''): | |||
self.repo_name = repo_name | ||||
return super(HasRepoPermissionAny, self).__call__(check_Location) | ||||
r673 | ||||
r547 | 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) | ||||
r673 | ||||
r547 | def __call__(self, user, repo_name): | |||
usr = AuthUser() | ||||
usr.user_id = user.user_id | ||||
usr.username = user.username | ||||
usr.is_admin = user.admin | ||||
r673 | ||||
r547 | try: | |||
self.user_perms = set([fill_perms(usr)\ | ||||
.permissions['repositories'][repo_name]]) | ||||
except: | ||||
self.user_perms = set() | ||||
self.granted_for = '' | ||||
self.username = user.username | ||||
r673 | self.repo_name = repo_name | |||
r547 | return self.check_permissions() | |||
r673 | ||||
r547 | def check_permissions(self): | |||
log.debug('checking mercurial protocol ' | ||||
r1040 | 'permissions %s for user:%s repository:%s', self.user_perms, | |||
r547 | 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 | ||||