@@ -671,348 +671,348 @@ class AuthUser(object):
@staticmethod
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)
else:
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))
return 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))
return True
return False
Status change: