kallithea Changeset - a32ca3200ca7

Changeset - a32ca3200ca7

Parent rev.

Child rev.

[Not reviewed]

default

0 1 0

Mads Kiilerich - 9 years ago 2017-05-13 03:53:49
mads@kiilerich.com

auth: refactor to make it explicit in the function profile when they only takes one permission

1 file changed with 30 insertions and 20 deletions:

kallithea/lib/auth.py

0 comments (0 inline, 0 general)

kallithea/lib/auth.py

➞

Show inline comments

@@ @@ -736,261 +736,271 @@ def set_available_permissions(config): @@
 #==============================================================================
 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 (but the "default" user,
     if enabled, is considered valid), or we'll redirect to the login page.
     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):
         self.api_access = api_access
     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')
                 raise HTTPForbidden()
         # regular user authentication
         if user.is_authenticated or user.is_default_user:
             log.info('user %s authenticated with regular auth @ %s', user, loc)
             return func(*fargs, **fkwargs)
         else:
             log.warning('user %s NOT authenticated with regular auth @ %s', user, loc)
             raise _redirect_to_login()
 # Use as decorator
 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."""
     def __call__(self, func):
         return decorator(self.__wrapper, func)
     def __wrapper(self, func, *fargs, **fkwargs):
         cls = fargs[0]
         user = request.authuser
         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'))
         else:
             return func(*fargs, **fkwargs)
 class _PermsDecorator(object):
     """Base class for controller decorators"""
+    """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
     def __call__(self, func):
         return decorator(self.__wrapper, func)
     def __wrapper(self, func, *fargs, **fkwargs):
         cls = fargs[0]
         user = request.authuser
         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)
             return func(*fargs, **fkwargs)
         else:
             log.debug('Permission denied for %s %s', cls, user)
             if user.is_default_user:
                 raise _redirect_to_login(_('You need to be signed in to view this page'))
             else:
                 raise HTTPForbidden()
     def check_permissions(self, user):
         raise NotImplementedError()
 class HasPermissionAnyDecorator(_PermsDecorator):
     """
     Checks the user has any of the given global permissions.
     """
     def check_permissions(self, user):
         global_permissions = user.permissions['global'] # usually very short
         return any(p in global_permissions for p in self.required_perms)
 class HasRepoPermissionLevelDecorator(_PermsDecorator):
 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.
     """
     def check_permissions(self, user):
         repo_name = get_repo_slug(request)
         (level,) = self.required_perms
         return user.has_repository_permission_level(repo_name, level)
         return user.has_repository_permission_level(repo_name, self.required_perm)
-class HasRepoGroupPermissionLevelDecorator(_PermsDecorator):
+class HasRepoGroupPermissionLevelDecorator(_PermDecorator):
     """
     Checks the user has any of given permissions for the requested repository group.
     """
     def check_permissions(self, user):
         repo_group_name = get_repo_group_slug(request)
         (level,) = self.required_perms
         return user.has_repository_group_permission_level(repo_group_name, level)
         return user.has_repository_group_permission_level(repo_group_name, self.required_perm)
-class HasUserGroupPermissionLevelDecorator(_PermsDecorator):
+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
     """
     def check_permissions(self, user):
         user_group_name = get_user_group_slug(request)
         (level,) = self.required_perms
         return user.has_user_group_permission_level(user_group_name, level)
         return user.has_user_group_permission_level(user_group_name, self.required_perm)
 #==============================================================================
 # CHECK FUNCTIONS
 #==============================================================================
 class _PermsFunction(object):
     """Base function for other check functions"""
+    """Base function for other check functions with multiple permissions"""
     def __init__(self, *required_perms):
         self.required_perms = required_perms # usually very short - a list is thus fine
     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):
         raise NotImplementedError()
 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 HasRepoPermissionLevel(_PermsFunction):
 class _PermFunction(_PermsFunction):
     """Base function for other check functions with a single permission"""
     def __init__(self, required_perm):
         _PermsFunction.__init__(self, [required_perm])
         self.required_perm = required_perm
 class HasRepoPermissionLevel(_PermFunction):
     def __call__(self, repo_name, purpose=None):
         (level,) = self.required_perms
         return request.user.has_repository_permission_level(repo_name, level, purpose)
         return request.user.has_repository_permission_level(repo_name, self.required_perm, purpose)
-class HasRepoGroupPermissionLevel(_PermsFunction):
+class HasRepoGroupPermissionLevel(_PermFunction):
     def __call__(self, group_name, purpose=None):
         (level,) = self.required_perms
         return request.user.has_repository_group_permission_level(group_name, level, purpose)
         return request.user.has_repository_group_permission_level(group_name, self.required_perm, purpose)
-class HasUserGroupPermissionLevel(_PermsFunction):
+class HasUserGroupPermissionLevel(_PermFunction):
     def __call__(self, user_group_name, purpose=None):
         (level,) = self.required_perms
         return request.user.has_user_group_permission_level(user_group_name, level, purpose)
         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)
         try:
             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))
         return 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

0 comments (0 inline, 0 general)