@@ -479,540 +479,540 @@ class AuthUser(object):
is_external_auth=False):
self.is_authenticated = False
self.is_external_auth = is_external_auth
self.authenticating_api_key = authenticating_api_key
user_model = UserModel()
self._default_user = User.get_default_user(cache=True)
# These attributes will be overridden by fill_data, below, unless the
# requested user cannot be found and the default anonymous user is
# not enabled.
self.user_id = None
self.username = None
self.api_key = None
self.name = ''
self.lastname = ''
self.email = ''
self.admin = False
self.inherit_default_permissions = False
# Look up database user, if necessary.
if user_id is not None:
log.debug('Auth User lookup by USER ID %s', user_id)
dbuser = user_model.get(user_id)
else:
# Note: dbuser is allowed to be None.
log.debug('Auth User lookup by database user %s', dbuser)
is_user_loaded = self._fill_data(dbuser)
# If user cannot be found, try falling back to anonymous.
if not is_user_loaded:
is_user_loaded = self._fill_data(self._default_user)
self.is_default_user = (self.user_id == self._default_user.user_id)
self.is_anonymous = not is_user_loaded or self.is_default_user
if not self.username:
self.username = 'None'
log.debug('Auth User is now %s', self)
def _fill_data(self, dbuser):
"""
Copies database fields from a `db.User` to this `AuthUser`. Does
not copy `api_keys` and `permissions` attributes.
Checks that `dbuser` is `active` (and not None) before copying;
returns True on success.
if dbuser is not None and dbuser.active:
log.debug('filling %s data', dbuser)
for k, v in dbuser.get_dict().iteritems():
assert k not in ['api_keys', 'permissions']
setattr(self, k, v)
return True
return False
@LazyProperty
def permissions(self):
return self.__get_perms(user=self, cache=False)
def has_repository_permission_level(self, repo_name, level, purpose=None):
required_perms = {
'read': ['repository.read', 'repository.write', 'repository.admin'],
'write': ['repository.write', 'repository.admin'],
'admin': ['repository.admin'],
}[level]
actual_perm = self.permissions['repositories'].get(repo_name)
ok = actual_perm in required_perms
log.debug('Checking if user %r can %r repo %r (%s): %s (has %r)',
self.username, level, repo_name, purpose, ok, actual_perm)
return ok
def has_repository_group_permission_level(self, repo_group_name, level, purpose=None):
'read': ['group.read', 'group.write', 'group.admin'],
'write': ['group.write', 'group.admin'],
'admin': ['group.admin'],
actual_perm = self.permissions['repositories_groups'].get(repo_group_name)
log.debug('Checking if user %r can %r repo group %r (%s): %s (has %r)',
self.username, level, repo_group_name, purpose, ok, actual_perm)
def has_user_group_permission_level(self, user_group_name, level, purpose=None):
'read': ['usergroup.read', 'usergroup.write', 'usergroup.admin'],
'write': ['usergroup.write', 'usergroup.admin'],
'admin': ['usergroup.admin'],
actual_perm = self.permissions['user_groups'].get(user_group_name)
log.debug('Checking if user %r can %r user group %r (%s): %s (has %r)',
self.username, level, user_group_name, purpose, ok, actual_perm)
@property
def api_keys(self):
return self._get_api_keys()
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: `AuthUser` instance
: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_by(user_id=self.user_id, is_expired=False):
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']
@staticmethod
def check_ip_allowed(user, ip_addr):
Check if the given IP address (a `str`) is allowed for the given
user (an `AuthUser` or `db.User`).
allowed_ips = AuthUser.get_allowed_ips(user.user_id, cache=True,
inherit_from_default=user.inherit_default_permissions)
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)
log.info('Access for IP:%s forbidden, '
'not in %s' % (ip_addr, allowed_ips))
def __repr__(self):
return "<AuthUser('id:%s[%s] auth:%s')>" \
% (self.user_id, self.username, (self.is_authenticated or self.is_default_user))
def to_cookie(self):
""" Serializes this login session to a cookie `dict`. """
return {
'user_id': self.user_id,
'is_external_auth': self.is_external_auth,
}
def from_cookie(cookie):
Deserializes an `AuthUser` from a cookie `dict`.
au = AuthUser(
user_id=cookie.get('user_id'),
is_external_auth=cookie.get('is_external_auth', False),
)
au.is_authenticated = True
return au
@classmethod
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 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 config instance
log.info('getting information about all available permissions')
all_perms = Session().query(Permission).all()
config['available_permissions'] = [x.permission_name for x in all_perms]
finally:
Session.remove()
#==============================================================================
# CHECK DECORATORS
def _redirect_to_login(message=None):
"""Return an exception that must be raised. It will redirect to the login
page which will redirect back to the current URL after authentication.
The optional message will be shown in a flash message."""
from kallithea.lib import helpers as h
if message:
h.flash(h.literal(message), category='warning')
p = request.path_qs
log.debug('Redirecting to login page, origin: %s', p)
return HTTPFound(location=url('login_home', came_from=p))
# Use as decorator
class LoginRequired(object):
"""Client must be logged in as a valid User, or we'll redirect to the login
page.
If the "default" user is enabled and allow_default_user is true, that is
considered valid too.
Also checks that IP address is allowed, and if using API key instead
of regular cookie authentication, checks that API key access is allowed
(based on `api_access` parameter and the API view whitelist).
def __init__(self, api_access=False, allow_default_user=False):
self.api_access = api_access
self.allow_default_user = allow_default_user
def __call__(self, func):
return decorator(self.__wrapper, func)
def __wrapper(self, func, *fargs, **fkwargs):
controller = fargs[0]
user = request.authuser
loc = "%s:%s" % (controller.__class__.__name__, func.__name__)
log.debug('Checking access for user %s @ %s', user, loc)
if not AuthUser.check_ip_allowed(user, request.ip_addr):
raise _redirect_to_login(_('IP %s not allowed') % request.ip_addr)
# Check if we used an API key to authenticate.
api_key = user.authenticating_api_key
if api_key is not None:
# Check that controller is enabled for API key usage.
if not self.api_access and not allowed_api_access(loc, api_key=api_key):
# controller does not allow API access
log.warning('API access to %s is not allowed', loc)
raise HTTPForbidden()
log.info('user %s authenticated with API key ****%s @ %s',
user, api_key[-4:], loc)
return func(*fargs, **fkwargs)
# CSRF protection: Whenever a request has ambient authority (whether
# through a session cookie or its origin IP address), it must include
# the correct token, unless the HTTP method is GET or HEAD (and thus
# guaranteed to be side effect free. In practice, the only situation
# where we allow side effects without ambient authority is when the
# authority comes from an API key; and that is handled above.
if request.method not in ['GET', 'HEAD']:
token = request.POST.get(secure_form.token_key)
if not token or token != secure_form.authentication_token():
log.error('CSRF check failed')
# regular user authentication
if user.is_authenticated:
log.info('user %s authenticated with regular auth @ %s', user, loc)
elif user.is_default_user:
if self.allow_default_user:
log.info('default user @ %s', loc)
log.info('default user is not accepted here @ %s', loc)
log.warning('user %s NOT authenticated with regular auth @ %s', user, loc)
raise _redirect_to_login()
class NotAnonymous(object):
"""Ensures that client is not logged in as the "default" user, and
redirects to the login page otherwise. Must be used together with
LoginRequired."""
cls = fargs[0]
log.debug('Checking that user %s is not anonymous @%s', user.username, cls)
if user.is_default_user:
raise _redirect_to_login(_('You need to be a registered user to '
'perform this action'))
class _PermsDecorator(object):
"""Base class for controller decorators with multiple permissions"""
def __init__(self, *required_perms):
self.required_perms = required_perms # usually very short - a list is thus fine
log.debug('checking %s permissions %s for %s %s',
self.__class__.__name__, self.required_perms, cls, user)
if self.check_permissions(user):
log.debug('Permission granted for %s %s', cls, user)
log.debug('Permission denied for %s %s', cls, user)
log.info('Permission denied for %s %s', cls, user)
raise _redirect_to_login(_('You need to be signed in to view this page'))
def check_permissions(self, user):
raise NotImplementedError()
class HasPermissionAnyDecorator(_PermsDecorator):
Checks the user has any of the given global permissions.
global_permissions = user.permissions['global'] # usually very short
return any(p in global_permissions for p in self.required_perms)
class _PermDecorator(_PermsDecorator):
"""Base class for controller decorators with a single permission"""
def __init__(self, required_perm):
_PermsDecorator.__init__(self, [required_perm])
self.required_perm = required_perm
class HasRepoPermissionLevelDecorator(_PermDecorator):
Checks the user has at least the specified permission level for the requested repository.
repo_name = get_repo_slug(request)
return user.has_repository_permission_level(repo_name, self.required_perm)
class HasRepoGroupPermissionLevelDecorator(_PermDecorator):
Checks the user has any of given permissions for the requested repository group.
repo_group_name = get_repo_group_slug(request)
return user.has_repository_group_permission_level(repo_group_name, self.required_perm)
class HasUserGroupPermissionLevelDecorator(_PermDecorator):
Checks for access permission for any of given predicates for specific
user group. In order to fulfill the request any of predicates must be meet
user_group_name = get_user_group_slug(request)
return user.has_user_group_permission_level(user_group_name, self.required_perm)
# CHECK FUNCTIONS
class _PermsFunction(object):
"""Base function for other check functions with multiple permissions"""
def __nonzero__(self):
""" Defend against accidentally forgetting to call the object
and instead evaluating it directly in a boolean context,
which could have security implications.
raise AssertionError(self.__class__.__name__ + ' is not a bool and must be called!')
def __call__(self, *a, **b):
class HasPermissionAny(_PermsFunction):
def __call__(self, purpose=None):
global_permissions = request.user.permissions['global'] # usually very short
ok = any(p in global_permissions for p in self.required_perms)
log.debug('Check %s for global %s (%s): %s' %
(request.user.username, self.required_perms, purpose, ok))
class _PermFunction(_PermsFunction):
"""Base function for other check functions with a single permission"""
_PermsFunction.__init__(self, [required_perm])
class HasRepoPermissionLevel(_PermFunction):
def __call__(self, repo_name, purpose=None):
return request.user.has_repository_permission_level(repo_name, self.required_perm, purpose)
class HasRepoGroupPermissionLevel(_PermFunction):
def __call__(self, group_name, purpose=None):
return request.user.has_repository_group_permission_level(group_name, self.required_perm, purpose)
class HasUserGroupPermissionLevel(_PermFunction):
def __call__(self, user_group_name, purpose=None):
return request.user.has_user_group_permission_level(user_group_name, self.required_perm, purpose)
# SPECIAL VERSION TO HANDLE MIDDLEWARE AUTH
class HasPermissionAnyMiddleware(object):
def __init__(self, *perms):
self.required_perms = set(perms)
def __call__(self, user, repo_name, purpose=None):
# repo_name MUST be unicode, since we handle keys in ok
# dict by unicode
repo_name = safe_unicode(repo_name)
user = AuthUser(user.user_id)
ok = user.permissions['repositories'][repo_name] in self.required_perms
except KeyError:
ok = False
log.debug('Middleware check %s for %s for repo %s (%s): %s' % (user.username, self.required_perms, repo_name, purpose, ok))
def check_ip_access(source_ip, allowed_ips=None):
Checks if source_ip is a subnet of any of allowed_ips.
:param source_ip:
:param allowed_ips: list of allowed ips together with mask
from kallithea.lib import ipaddr
log.debug('checking if ip:%s is subnet of %s', source_ip, allowed_ips)
if isinstance(allowed_ips, (tuple, list, set)):
for ip in allowed_ips:
if ipaddr.IPAddress(source_ip) in ipaddr.IPNetwork(ip):
log.debug('IP %s is network %s',
ipaddr.IPAddress(source_ip), ipaddr.IPNetwork(ip))
Status change: