@@ -342,768 +342,769 @@ def _cached_perms_data(user_id, user_is_
Permission.permission_id))\
.join((UserGroupMember, UserGroupRepoToPerm.users_group_id ==
UserGroupMember.users_group_id))\
.filter(UserGroupMember.user_id == uid)\
.all()
multiple_counter = collections.defaultdict(int)
for perm in user_repo_perms_from_users_groups:
r_k = perm.UserGroupRepoToPerm.repository.repo_name
multiple_counter[r_k] += 1
p = perm.Permission.permission_name
cur_perm = permissions[RK][r_k]
if perm.Repository.user_id == uid:
# set admin if owner
p = 'repository.admin'
else:
if multiple_counter[r_k] > 1:
p = _choose_perm(p, cur_perm)
permissions[RK][r_k] = p
# user explicit permissions for repositories, overrides any specified
# by the group permission
user_repo_perms = Permission.get_default_perms(uid)
for perm in user_repo_perms:
r_k = perm.UserRepoToPerm.repository.repo_name
if not explicit:
#======================================================================
# !! PERMISSIONS FOR REPOSITORY GROUPS !!
# check if user is part of user groups for this repository groups and
# fill in his permission from it. _choose_perm decides of which
# permission should be selected based on selected method
# user group for repo groups permissions
user_repo_group_perms_from_users_groups = \
Session().query(UserGroupRepoGroupToPerm, Permission, RepoGroup)\
.join((RepoGroup, UserGroupRepoGroupToPerm.group_id == RepoGroup.group_id))\
.join((Permission, UserGroupRepoGroupToPerm.permission_id
== Permission.permission_id))\
.join((UserGroupMember, UserGroupRepoGroupToPerm.users_group_id
== UserGroupMember.users_group_id))\
for perm in user_repo_group_perms_from_users_groups:
g_k = perm.UserGroupRepoGroupToPerm.group.group_name
multiple_counter[g_k] += 1
cur_perm = permissions[GK][g_k]
if multiple_counter[g_k] > 1:
permissions[GK][g_k] = p
# user explicit permissions for repository groups
user_repo_groups_perms = Permission.get_default_group_perms(uid)
for perm in user_repo_groups_perms:
rg_k = perm.UserRepoGroupToPerm.group.group_name
cur_perm = permissions[GK][rg_k]
permissions[GK][rg_k] = p
# !! PERMISSIONS FOR USER GROUPS !!
# user group for user group permissions
user_group_user_groups_perms = \
Session().query(UserGroupUserGroupToPerm, Permission, UserGroup)\
.join((UserGroup, UserGroupUserGroupToPerm.target_user_group_id
== UserGroup.users_group_id))\
.join((Permission, UserGroupUserGroupToPerm.permission_id
.join((UserGroupMember, UserGroupUserGroupToPerm.user_group_id
for perm in user_group_user_groups_perms:
g_k = perm.UserGroupUserGroupToPerm.target_user_group.users_group_name
cur_perm = permissions[UK][g_k]
permissions[UK][g_k] = p
#user explicit permission for user groups
user_user_groups_perms = Permission.get_default_user_group_perms(uid)
for perm in user_user_groups_perms:
u_k = perm.UserUserGroupToPerm.user_group.users_group_name
cur_perm = permissions[UK][u_k]
permissions[UK][u_k] = p
return permissions
def allowed_api_access(controller_name, whitelist=None, api_key=None):
"""
Check if given controller_name is in whitelist API access
if not whitelist:
from kallithea import CONFIG
whitelist = aslist(CONFIG.get('api_access_controllers_whitelist'),
sep=',')
log.debug('whitelist of API access is: %s' % (whitelist))
api_access_valid = controller_name in whitelist
if api_access_valid:
log.debug('controller:%s is in API whitelist' % (controller_name))
msg = 'controller: %s is *NOT* in API whitelist' % (controller_name)
if api_key:
#if we use API key and don't have access it's a warning
log.warning(msg)
log.debug(msg)
return api_access_valid
class AuthUser(object):
A simple object that handles all attributes of user in Kallithea
It does lookup based on API key,given user, or user present in session
Then it fills all required information for such user. It also checks if
anonymous access is enabled and if so, it returns default user as logged in
def __init__(self, user_id=None, api_key=None, username=None, ip_addr=None):
self.user_id = user_id
self._api_key = api_key
self.api_key = None
self.username = username
self.ip_addr = ip_addr
self.name = ''
self.lastname = ''
self.email = ''
self.is_authenticated = False
self.admin = False
self.inherit_default_permissions = False
self.propagate_data()
self._instance = None
@LazyProperty
def permissions(self):
return self.get_perms(user=self, cache=False)
@property
def api_keys(self):
return self.get_api_keys()
def propagate_data(self):
user_model = UserModel()
self.anonymous_user = User.get_default_user(cache=True)
is_user_loaded = False
# lookup by userid
if self.user_id is not None and self.user_id != self.anonymous_user.user_id:
log.debug('Auth User lookup by USER ID %s' % self.user_id)
is_user_loaded = user_model.fill_data(self, user_id=self.user_id)
# try go get user by API key
elif self._api_key and self._api_key != self.anonymous_user.api_key:
log.debug('Auth User lookup by API key %s' % self._api_key)
is_user_loaded = user_model.fill_data(self, api_key=self._api_key)
# lookup by username
elif self.username:
log.debug('Auth User lookup by USER NAME %s' % self.username)
is_user_loaded = user_model.fill_data(self, username=self.username)
log.debug('No data in %s that could been used to log in' % self)
if not is_user_loaded:
# if we cannot authenticate user try anonymous
if self.anonymous_user.active:
user_model.fill_data(self, user_id=self.anonymous_user.user_id)
# then we set this user is logged in
self.is_authenticated = True
self.user_id = None
self.username = None
if not self.username:
self.username = 'None'
log.debug('Auth User is now %s' % self)
def get_perms(self, user, explicit=True, algo='higherwin', cache=False):
Fills user permission attribute with permissions taken from database
works for permissions given for repositories, and for permissions that
are granted to groups
:param user: instance of User object from database
:param explicit: In case there are permissions both for user and a group
that user is part of, explicit flag will define if user will
explicitly override permissions from group, if it's False it will
make decision based on the algo
:param algo: algorithm to decide what permission should be choose if
it's multiple defined, eg user in two different groups. It also
decides if explicit flag is turned off how to specify the permission
for case when user is in a group + have defined separate permission
user_id = user.user_id
user_is_admin = user.is_admin
user_inherit_default_permissions = user.inherit_default_permissions
log.debug('Getting PERMISSION tree')
compute = conditional_cache('short_term', 'cache_desc',
condition=cache, func=_cached_perms_data)
return compute(user_id, user_is_admin,
user_inherit_default_permissions, explicit, algo)
def get_api_keys(self):
api_keys = [self.api_key]
for api_key in UserApiKeys.query()\
.filter(UserApiKeys.user_id == self.user_id)\
.filter(or_(UserApiKeys.expires == -1,
UserApiKeys.expires >= time.time())).all():
api_keys.append(api_key.api_key)
return api_keys
def is_admin(self):
return self.admin
def repositories_admin(self):
Returns list of repositories you're an admin of
return [x[0] for x in self.permissions['repositories'].iteritems()
if x[1] == 'repository.admin']
def repository_groups_admin(self):
Returns list of repository groups you're an admin of
return [x[0] for x in self.permissions['repositories_groups'].iteritems()
if x[1] == 'group.admin']
def user_groups_admin(self):
Returns list of user groups you're an admin of
return [x[0] for x in self.permissions['user_groups'].iteritems()
if x[1] == 'usergroup.admin']
def ip_allowed(self):
Checks if ip_addr used in constructor is allowed from defined list of
allowed ip_addresses for user
:returns: boolean, True if ip is in allowed ip range
# check IP
inherit = self.inherit_default_permissions
return AuthUser.check_ip_allowed(self.user_id, self.ip_addr,
inherit_from_default=inherit)
@classmethod
def check_ip_allowed(cls, user_id, ip_addr, inherit_from_default):
allowed_ips = AuthUser.get_allowed_ips(user_id, cache=True,
inherit_from_default=inherit_from_default)
if check_ip_access(source_ip=ip_addr, allowed_ips=allowed_ips):
log.debug('IP:%s is in range of %s' % (ip_addr, allowed_ips))
return True
log.info('Access for IP:%s forbidden, '
'not in %s' % (ip_addr, allowed_ips))
return False
def __repr__(self):
return "<AuthUser('id:%s[%s] ip:%s auth:%s')>"\
% (self.user_id, self.username, self.ip_addr, self.is_authenticated)
def set_authenticated(self, authenticated=True):
if self.user_id != self.anonymous_user.user_id:
self.is_authenticated = authenticated
def get_cookie_store(self):
return {'username': self.username,
'user_id': self.user_id,
'is_authenticated': self.is_authenticated}
def from_cookie_store(cls, cookie_store):
Creates AuthUser from a cookie store
:param cls:
:param cookie_store:
user_id = cookie_store.get('user_id')
username = cookie_store.get('username')
api_key = cookie_store.get('api_key')
return AuthUser(user_id, api_key, username)
def get_allowed_ips(cls, user_id, cache=False, inherit_from_default=False):
_set = set()
if inherit_from_default:
default_ips = UserIpMap.query().filter(UserIpMap.user ==
User.get_default_user(cache=True))
if cache:
default_ips = default_ips.options(FromCache("sql_cache_short",
"get_user_ips_default"))
# populate from default user
for ip in default_ips:
try:
_set.add(ip.ip_addr)
except ObjectDeletedError:
# since we use heavy caching sometimes it happens that we get
# deleted objects here, we just skip them
pass
user_ips = UserIpMap.query().filter(UserIpMap.user_id == user_id)
user_ips = user_ips.options(FromCache("sql_cache_short",
"get_user_ips_%s" % user_id))
for ip in user_ips:
return _set or set(['0.0.0.0/0', '::/0'])
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')
sa = meta.Session
all_perms = sa.query(Permission).all()
config['available_permissions'] = [x.permission_name for x in all_perms]
finally:
meta.Session.remove()
#==============================================================================
# CHECK DECORATORS
def redirect_to_login(message=None):
from kallithea.lib import helpers as h
p = url.current()
if message:
h.flash(h.literal(message), category='warning')
log.debug('Redirecting to login page, origin: %s' % p)
return redirect(url('login_home', came_from=p))
class LoginRequired(object):
Must be logged in to execute this function else
redirect to login page
:param api_access: if enabled this checks only for valid auth token
and grants access based on valid token
def __init__(self, api_access=False):
self.api_access = api_access
def __call__(self, func):
return decorator(self.__wrapper, func)
def __wrapper(self, func, *fargs, **fkwargs):
cls = fargs[0]
user = cls.authuser
loc = "%s:%s" % (cls.__class__.__name__, func.__name__)
log.debug('Checking access for user %s @ %s' % (user, loc))
# check if our IP is allowed
if not user.ip_allowed:
return redirect_to_login(_('IP %s not allowed' % (user.ip_addr)))
# check if we used an API key and it's a valid one
api_key = request.GET.get('api_key')
if api_key is not None:
# explicit controller is enabled or API is in our whitelist
if self.api_access or allowed_api_access(loc, api_key=api_key):
if api_key in user.api_keys:
log.info('user %s authenticated with API key ****%s @ %s'
% (user, api_key[-4:], loc))
return func(*fargs, **fkwargs)
log.warning('API key ****%s is NOT valid' % api_key[-4:])
return redirect_to_login(_('Invalid API key'))
# controller does not allow API access
log.warning('API access to %s is not allowed' % loc)
return abort(403)
# CSRF protection - POSTs with session auth must contain correct token
if request.POST and user.is_authenticated:
token = request.POST.get(secure_form.token_key)
if not token or token != secure_form.authentication_token():
log.error('CSRF check failed')
log.debug('Checking if %s is authenticated @ %s' % (user.username, loc))
reason = 'RegularAuth' if user.is_authenticated else 'APIAuth'
if user.is_authenticated:
log.info('user %s authenticating with:%s IS authenticated on func %s '
% (user, reason, loc)
)
log.warning('user %s authenticating with:%s NOT authenticated on func: %s: '
return redirect_to_login()
class NotAnonymous(object):
redirect to login page"""
self.user = cls.authuser
log.debug('Checking if user is not anonymous @%s' % cls)
anonymous = self.user.username == User.DEFAULT_USER
if anonymous:
return redirect_to_login(_('You need to be a registered user to '
'perform this action'))
class PermsDecorator(object):
"""Base class for controller decorators"""
def __init__(self, *required_perms):
self.required_perms = set(required_perms)
self.user_perms = None
self.user_perms = self.user.permissions
log.debug('checking %s permissions %s for %s %s',
self.__class__.__name__, self.required_perms, cls, self.user)
if self.check_permissions():
log.debug('Permission granted for %s %s' % (cls, self.user))
log.debug('Permission denied for %s %s' % (cls, self.user))
return redirect_to_login(_('You need to be signed in to view this page'))
# redirect with forbidden ret code
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
if self.required_perms.issubset(self.user_perms.get('global')):
class HasPermissionAnyDecorator(PermsDecorator):
Checks for access permission for any of given predicates. In order to
fulfill the request any of predicates must be meet
if self.required_perms.intersection(self.user_perms.get('global')):
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
repo_name = get_repo_slug(request)
user_perms = set([self.user_perms['repositories'][repo_name]])
except KeyError:
if self.required_perms.issubset(user_perms):
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
if self.required_perms.intersection(user_perms):
class HasRepoGroupPermissionAllDecorator(PermsDecorator):
repository group. All of them have to be meet in order to fulfill the request
group_name = get_repo_group_slug(request)
user_perms = set([self.user_perms['repositories_groups'][group_name]])
class HasRepoGroupPermissionAnyDecorator(PermsDecorator):
repository group. In order to fulfill the request any of predicates must be meet
class HasUserGroupPermissionAllDecorator(PermsDecorator):
user group. All of them have to be meet in order to fulfill the request
group_name = get_user_group_slug(request)
user_perms = set([self.user_perms['user_groups'][group_name]])
class HasUserGroupPermissionAnyDecorator(PermsDecorator):
user group. In order to fulfill the request any of predicates must be meet
# CHECK FUNCTIONS
class PermsFunction(object):
"""Base function for other check functions"""
def __init__(self, *perms):
self.required_perms = set(perms)
self.repo_name = None
self.group_name = None
def __call__(self, check_location='', user=None):
if not user:
#TODO: remove this someday,put as user as attribute here
user = request.user
# init auth user if not already given
if not isinstance(user, AuthUser):
user = AuthUser(user.user_id)
cls_name = self.__class__.__name__
check_scope = {
'HasPermissionAll': '',
'HasPermissionAny': '',
'HasRepoPermissionAll': 'repo:%s' % self.repo_name,
'HasRepoPermissionAny': 'repo:%s' % self.repo_name,
'HasRepoGroupPermissionAll': 'group:%s' % self.group_name,
'HasRepoGroupPermissionAny': 'group:%s' % self.group_name,
}.get(cls_name, '?')
log.debug('checking cls:%s %s usr:%s %s @ %s', cls_name,
self.required_perms, user, check_scope,
check_location or 'unspecified location')
log.debug('Empty request user')
self.user_perms = user.permissions
log.debug('Permission to %s granted for user: %s @ %s'
% (check_scope, user,
check_location or 'unspecified location'))
log.debug('Permission to %s denied for user: %s @ %s'
class HasPermissionAll(PermsFunction):
class HasPermissionAny(PermsFunction):
class HasRepoPermissionAll(PermsFunction):
def __call__(self, repo_name=None, check_location='', user=None):
self.repo_name = repo_name
return super(HasRepoPermissionAll, self).__call__(check_location, user)
if not self.repo_name:
self.repo_name = get_repo_slug(request)
self._user_perms = set(
[self.user_perms['repositories'][self.repo_name]]
if self.required_perms.issubset(self._user_perms):
class HasRepoPermissionAny(PermsFunction):
return super(HasRepoPermissionAny, self).__call__(check_location, user)
if self.required_perms.intersection(self._user_perms):
class HasRepoGroupPermissionAny(PermsFunction):
def __call__(self, group_name=None, check_location='', user=None):
self.group_name = group_name
return super(HasRepoGroupPermissionAny, self).__call__(check_location, user)
[self.user_perms['repositories_groups'][self.group_name]]
class HasRepoGroupPermissionAll(PermsFunction):
Status change: