Changeset - c57d926edd39
[Not reviewed]
default
0 1 0
Mads Kiilerich - 7 years ago 2018-09-01 01:12:13
mads@kiilerich.com
auth: strip RFC4007 zone identifiers from IPv6 addresses before doing access control

If using IPv6, the request IP address might contain a '%' that the ipaddr
module that is used for IP filtering can't handle.

https://tools.ietf.org/html/rfc4007#section-11 specifies how IPv6 addresses can
have zone identifiers like trailing '%13' or '%eth0'. The zone identifier is
used to help distinguish *if* the same address should be available on multiple
interfaces. It *could* potentially have security implications in the odd case
where the same address is different on different interfaces. The IP whitelist
functionality does however not support zone filters, so there is no way users
can expect the zone to be relevant for IP filtering. We can thus safely strip
the zone index and only check for match on the other parts of the address.
1 file changed with 1 insertions and 0 deletions:
0 comments (0 inline, 0 general)
kallithea/lib/auth.py
Show inline comments
 
@@ -915,104 +915,105 @@ class HasUserGroupPermissionLevelDecorat
 
    """
 

	
 
    def check_permissions(self, user):
 
        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 __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 _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):
 
        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)
 

	
 
        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
 
    source_ip = source_ip.split('%', 1)[0]
 
    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)