source: main/waeup.cas/trunk/waeup/cas/authenticators.py @ 10478

"""Components that do the real authentication stuff.
"""
import re
from pkg_resources import iter_entry_points
try:
    import xmlrpclib                     # Python 2.x
except ImportError:                      # pragma: no cover
    import xmlrpc.client as xmlrpclib    # Python 3.x


def get_all_authenticators():
    """Get all authenticators registered via entry points.

    To create an own authenticator, in the ``setup.py`` of your
    package add a block like this::

        entry_points='''
           [waeup.cas.authenticators]
           myname = my.pkg.my.module:MyAuthenticator
           '''

    where ``my.pkg.my.module`` must be a module importable at runtime
    and ``MyAuthenticator`` must be some Authenticator class defined
    in this module.

    ``myname`` can be chosen as you like, but must not clash with
    other ``waeup.cas.authenticators``. The name should contain ASCII
    letters and numbers only, and start with a letter.
    """
    return dict(
        [(x.name, x.load())
         for x in iter_entry_points(group='waeup.cas.authenticators')])


def get_authenticator(local_conf):
    """Get an `Authenticator` configured by ``local_conf``.

    `local_conf` must be a dictionary of settings.

    An authenticator is identified by its entry-point name which must
    be given as value of the ``auth`` key.

    It is created passing all keys/values where key starts with
    ``auth_``.

    Returns a dict of remaining keys/values in `local_conf` with the
    value of ``auth`` key replaced by an authenticator instance.
    """
    gen_opts, auth_opts = filter_auth_opts(local_conf)
    if 'auth' in local_conf:
        auth_dict = get_all_authenticators()
        factory = auth_dict.get(local_conf['auth'])  # get authenticator
        auth = factory(**auth_opts)
        gen_opts.update(auth=auth)
    return gen_opts


def filter_auth_opts(local_conf):
    """Filter out auth-related opts in `local_conf`, a dict.

    Returns two dicts ``(<GENERAL_OPTS>, <AUTHENTICATOR_OPTS>)``
    containing the general options and authenticator related options.

    This is designed to work with typical paste.deploy config files
    like this::

      [foo]
      foo = bar
      auth = foo
      auth_opt1 = bar

    All settings not starting with `auth_` are put into the general
    opts dict, while other are put into the authenticator opts dict.
    """
    auth_opts = {}
    general_opts = {}
    for name, value in local_conf.items():
        if name.startswith('auth_'):
            auth_opts[name] = value
        else:
            general_opts[name] = value
    return general_opts, auth_opts


class Authenticator(object):

    #: The name of an authenticator
    name = 'basic'

    def __init__(self, **kw):
        pass


class DummyAuthenticator(Authenticator):
    """A dummy authenticator for tests.

    This authenticator does no real authentication. It simply approves
    credentials ``('bird', 'bebop')`` and denies all other.
    """

    name = 'dummy'

    def check_credentials(self, username='', password=''):
        """If username is ``'bird'`` and password ``'bebop'`` check
        will succeed.

        Returns a tuple ``(<STATUS>, <REASON>)`` with ``<STATUS>``
        being a boolean and ``<REASON>`` being a string giving the
        reason why login failed (if so) or an empty string.
        """
        reason = ''
        result = username == 'bird' and password == 'bebop'
        if not result:
            reason = 'Invalid username or password.'
        return result, reason


#: Regular expression matching a starting university marker like
#: the string 'MA-' in 'MA-M121212' or 'MA-' in 'MA-APP-13123'
RE_SCHOOL_MARKER = re.compile('^[^\-]+-')


class KofaAuthenticator(Authenticator):
    """Authenticate against a running Kofa instance.
    """

    name = 'kofa1'

    def __init__(self, auth_backends="{}"):
        try:
            self.backends = eval(auth_backends)
        except:
            raise ValueError('auth_backends must be a '
                             'valid Python expression.')
        self._check_options()

    def _check_options(self):
        if not isinstance(self.backends, dict):
            raise ValueError('Backends must be configured as dicts.')
        for key, val in self.backends.items():
            if not isinstance(val, dict):
                raise ValueError(
                    'Backend %s: config must be a dict' % key)
            if not 'url' in val:
                raise ValueError(
                    'Backend %s: config must contain an `url` key.' % key)
            if not 'marker' in val:
                self.backends[key]['marker'] = '.+'
            try:
                re.compile(self.backends[key]['marker'])
            except:
                raise ValueError(
                    'Backend %s: marker must be a valid regular expr.:' % (
                        key,))

    def check_credentials(self, username='', password=''):
        """Do the real check.
        """
        for backend_name, backend in self.backends.items():
            if not re.match(backend['marker'], username):
                continue
            # remove school marker
            username = RE_SCHOOL_MARKER.sub('', username)
            proxy = xmlrpclib.ServerProxy(
                backend['url'], allow_none=True)
            valid = proxy.check_credentials(username, password)
            if valid is not None:
                return (True, '')
        return (False, 'Invalid username or password.')