Approach for preveting annomyous access to CKAN site

Question

I have a CKAN site running with the ckanext-ldap extension configured, but I only wan't authenticated users to be able access the site.

This is my solution so far, but I'm not totally satisfied:

import ckan.plugins as plugins
import ckan.plugins.toolkit as toolkit

def site_read(context, data_dict):
    # List of allowed paths, when not logged in
    allowed_anon_paths = ['/user/login', '/ldap_login_handler']

    # Prevent "site read" if the user is not logged in and the
    # request path is not in the list of allowed anonymous paths
    if not context.get('user') and toolkit.request.path not in allowed_anon_paths:
        return {'success': False}

    return {'success': True}

class Disable_Anon_AccessPlugin(plugins.SingletonPlugin):
    plugins.implements(plugins.IAuthFunctions)

    def get_auth_functions(self):
        return {'site_read': site_read}

It prevents anonymous users from accessing any pages (other than login related), but it provides a 403 Forbidden error, on all pages until logged in.

(also API requests fails with 500 error, unless logged in or providing an API key, but I can live with that)

I can't figure a way to redirect, to the login page, if not logged in and/or making the "remember me" feature work.

Adding something like: toolkit.redirect_to('/user/login') instead of return {'success': False} does not have an effect.

I also looked into the IRoutes interface, but I cannot figure out how to get the current logged in user (or checking if a user is logged in)

Answer 1

I've seen the following approach used in CKAN to prevent non-logged in access to the site.

It uses the IMiddleware plugin interface:

class AuthMiddleware(object):
    def __init__(self, app, app_conf):
        self.app = app
    def __call__(self, environ, start_response):
        # if logged in via browser cookies or API key, all pages accessible
        if 'repoze.who.identity' in environ or self._get_user_for_apikey(environ) or not is_iar():
            return self.app(environ,start_response)
        else:
            # otherwise only login/reset and front pages are accessible
            if (environ['PATH_INFO'] == '/' or environ['PATH_INFO'] == '/user/login' or environ['PATH_INFO'] == '/user/_logout'
                                or '/user/reset' in environ['PATH_INFO'] or environ['PATH_INFO'] == '/user/logged_out'
                                or environ['PATH_INFO'] == '/user/logged_in' or environ['PATH_INFO'] == '/user/logged_out_redirect'):
                return self.app(environ,start_response)
            else:
                # http://rufuspollock.org/2006/09/28/wsgi-middleware/
                environ['wsgiorg.routing_args'] = '',{'action': 'login', 'controller': 'user'}
                return self.app(environ,start_response)

    def _get_user_for_apikey(self, environ):
        # Adapted from https://github.com/ckan/ckan/blob/625b51cdb0f1697add59c7e3faf723a48c8e04fd/ckan/lib/base.py#L396
        apikey_header_name = config.get(base.APIKEY_HEADER_NAME_KEY,
                                        base.APIKEY_HEADER_NAME_DEFAULT)
        apikey = environ.get(apikey_header_name, '')
        if not apikey:
            # For misunderstanding old documentation (now fixed).
            apikey = environ.get('HTTP_AUTHORIZATION', '')
        if not apikey:
            apikey = environ.get('Authorization', '')
            # Forget HTTP Auth credentials (they have spaces).
            if ' ' in apikey:
                apikey = ''
        if not apikey:
            return None
        apikey = unicode(apikey)
        # check if API key is valid by comparing against keys of registered users
        query = model.Session.query(model.User)
        user = query.filter_by(apikey=apikey).first()
        return user

Which is then added to your plugin class, e.g.

class Disable_Anon_AccessPlugin(plugins.SingletonPlugin):
    plugins.implements(plugins.IMiddleware, inherit=True)

    def make_middleware(self, app, config):
        return AuthMiddleware(app, config)