# 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)

        if cache:

            user_ips = user_ips.options(FromCache("sql_cache_short",

                                                  "get_user_ips_%s" % user_id))

        for ip in user_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

        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')

    try:

        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, **request.GET))

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):

        controller = fargs[0]

        user = controller.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, controller.ip_addr):

            return redirect_to_login(_('IP %s not allowed') % controller.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)

                else:

                    log.warning('API key ****%s is NOT valid', api_key[-4:])

                    return redirect_to_login(_('Invalid API key'))

            else:

                # controller does not allow API access

                log.warning('API access to %s is not allowed', loc)

                return abort(403)

        # 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')

                return abort(403)

        # regular user authentication

        if user.is_authenticated:

            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)

            return redirect_to_login()

class NotAnonymous(object):

    """

    Must be logged in to execute this function else

    redirect to login page"""

    def __call__(self, func):

        return decorator(self.__wrapper, func)

    def __wrapper(self, func, *fargs, **fkwargs):

        cls = fargs[0]

        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'))

        else:

            return func(*fargs, **fkwargs)

class PermsDecorator(object):

    """Base class for controller decorators"""

    def __init__(self, *required_perms):

        self.required_perms = set(required_perms)

        self.user_perms = None

    def __call__(self, func):

        return decorator(self.__wrapper, func)

    def __wrapper(self, func, *fargs, **fkwargs):

        cls = fargs[0]

        self.user = cls.authuser

        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)

            return func(*fargs, **fkwargs)

        else:

            log.debug('Permission denied for %s %s', cls, self.user)

            anonymous = self.user.username == User.DEFAULT_USER

            if anonymous:

                return redirect_to_login(_('You need to be signed in to view this page'))

            else:

                # redirect with forbidden ret code

                return abort(403)

    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

    """

    def check_permissions(self):

        if self.required_perms.issubset(self.user_perms.get('global')):

            return True

        return False

class HasPermissionAnyDecorator(PermsDecorator):

    """

    Checks for access permission for any of given predicates. In order to

    fulfill the request any of predicates must be meet

    """

    def check_permissions(self):

        if self.required_perms.intersection(self.user_perms.get('global')):

            return True

        return False

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

    """

from kallithea.tests import *

from kallithea.model.db import Setting

class TestDefaultsController(TestController):

    def test_index(self):

        self.log_user()

        response = self.app.get(url('defaults'))

        response.mustcontain('default_repo_private')

        response.mustcontain('default_repo_enable_statistics')

        response.mustcontain('default_repo_enable_downloads')

        response.mustcontain('default_repo_enable_locking')

    def test_index_as_xml(self):

        response = self.app.get(url('formatted_defaults', format='xml'))

    def test_create(self):

        response = self.app.post(url('defaults'),

            {'_authentication_token': self.authentication_token()})

    def test_new(self):

        response = self.app.get(url('new_default'))

    def test_new_as_xml(self):

        response = self.app.get(url('formatted_new_default', format='xml'))

    def test_update_params_true_hg(self):

        self.log_user()

        params = {

            'default_repo_enable_locking': True,

            'default_repo_enable_downloads': True,

            'default_repo_enable_statistics': True,

            'default_repo_private': True,

            'default_repo_type': 'hg',

            '_authentication_token': self.authentication_token(),

        }

        response = self.app.put(url('default', id='default'), params=params)

        self.checkSessionFlash(response, 'Default settings updated successfully')

        params.pop('_authentication_token')

        defs = Setting.get_default_repo_settings()

        self.assertEqual(params, defs)

    def test_update_params_false_git(self):

        self.log_user()

        params = {

            'default_repo_enable_locking': False,

            'default_repo_enable_downloads': False,

            'default_repo_enable_statistics': False,

            'default_repo_private': False,

            'default_repo_type': 'git',

            '_authentication_token': self.authentication_token(),

        }

        response = self.app.put(url('default', id='default'), params=params)

        self.checkSessionFlash(response, 'Default settings updated successfully')

        params.pop('_authentication_token')

        defs = Setting.get_default_repo_settings()

        self.assertEqual(params, defs)

    def test_update_browser_fakeout(self):

        response = self.app.post(url('default', id=1), params=dict(_method='put', _authentication_token=self.authentication_token()))

    def test_delete(self):

        # Not possible due to CSRF protection.

    def test_delete_browser_fakeout(self):

        response = self.app.post(url('default', id=1), params=dict(_method='delete', _authentication_token=self.authentication_token()))

    def test_show(self):

        response = self.app.get(url('default', id=1))

    def test_show_as_xml(self):

        response = self.app.get(url('formatted_default', id=1, format='xml'))

    def test_edit(self):

        response = self.app.get(url('edit_default', id=1))

    def test_edit_as_xml(self):

        response = self.app.get(url('formatted_edit_default', id=1, format='xml'))