source: main/waeup.kofa/trunk/src/waeup/kofa/authentication.py @ 17917

## $Id: authentication.py 17528 2023-08-01 06:40:01Z henrik $
##
## Copyright (C) 2011 Uli Fouquet & Henrik Bettermann
## This program is free software; you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation; either version 2 of the License, or
## (at your option) any later version.
##
## This program is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
## GNU General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with this program; if not, write to the Free Software
## Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
##
"""Authentication for Kofa.
"""
import grok
import time
import re
from zope.i18n import translate
from zope.event import notify
from zope.component import getUtility, getUtilitiesFor
from zope.component.interfaces import IFactory
from zope.interface import Interface, implementedBy
from zope.schema import getFields
from zope.securitypolicy.interfaces import (
    IPrincipalRoleMap, IPrincipalRoleManager)
from zope.pluggableauth.factories import Principal
from zope.pluggableauth.plugins.session import SessionCredentialsPlugin
from zope.pluggableauth.plugins.httpplugins import (
    HTTPBasicAuthCredentialsPlugin)
from zope.pluggableauth.interfaces import (
    ICredentialsPlugin, IAuthenticatorPlugin, IAuthenticatedPrincipalFactory,
    AuthenticatedPrincipalCreated)
from zope.publisher.interfaces import IRequest
from zope.password.interfaces import IPasswordManager
from zope.securitypolicy.principalrole import principalRoleManager
from waeup.kofa.interfaces import (
    ILocalRoleSetEvent, IUserAccount, IAuthPluginUtility, IPasswordValidator,
    IKofaPrincipal, IKofaPrincipalInfo, IKofaPluggable, IBatchProcessor,
    IGNORE_MARKER, IFailedLoginInfo)
from waeup.kofa.utils.batching import BatchProcessor
from waeup.kofa.permissions import get_all_roles
from waeup.kofa.interfaces import MessageFactory as _


def setup_authentication(pau):
    """Set up plugguble authentication utility.

    Sets up an IAuthenticatorPlugin and
    ICredentialsPlugin (for the authentication mechanism)

    Then looks for any external utilities that want to modify the PAU.
    """
    pau.credentialsPlugins = (
        'No Challenge if Authenticated',
        'xmlrpc-credentials',
        'credentials')
    pau.authenticatorPlugins = ('users',)

    # Give any third-party code and subpackages a chance to modify the PAU
    auth_plugin_utilities = getUtilitiesFor(IAuthPluginUtility)
    for name, util in auth_plugin_utilities:
        util.register(pau)


def get_principal_role_manager():
    """Get a role manager for principals.

    If we are currently 'in a site', return the role manager for the
    portal or the global rolemanager else.
    """
    portal = grok.getSite()
    if portal is not None:
        return IPrincipalRoleManager(portal)
    return principalRoleManager


class KofaSessionCredentialsPlugin(
        grok.GlobalUtility, SessionCredentialsPlugin):
    """Session plugin that picks usernames/passwords from fields in webforms.
    """
    grok.provides(ICredentialsPlugin)
    grok.name('credentials')

    loginpagename = 'login'
    loginfield = 'form.login'
    passwordfield = 'form.password'


class KofaXMLRPCCredentialsPlugin(
        grok.GlobalUtility, HTTPBasicAuthCredentialsPlugin):
    """Plugin that picks useranams/passwords from basic-auth headers.

    As XMLRPC requests send/post their authentication credentials in HTTP
    basic-auth headers, we need a plugin that can handle this.

    This plugin, however, does no challenging. If a user does not provide
    basic-auth infos, we will not ask for some. This is correct as we plan to
    communicate with machines.

    This plugin is planned to be used in "PluggableAuthenitications" registered
    with `University` instances.
    """
    grok.provides(ICredentialsPlugin)
    grok.name('xmlrpc-credentials')

    def challenge(self, request):
        """XMLRPC is for machines. No need to challenge.
        """
        return False

    def logout(self, request):
        """Basic auth does not provide any logout possibility.
        """
        return False


class KofaPrincipalInfo(object):
    """An implementation of IKofaPrincipalInfo.

    A Kofa principal info is created with id, login, title, description,
    phone, email, public_name and user_type.
    """
    grok.implements(IKofaPrincipalInfo)

    def __init__(self, id, title, description, email, phone, public_name,
                 user_type):
        self.id = id
        self.title = title
        self.description = description
        self.email = email
        self.phone = phone
        self.public_name = public_name
        self.user_type = user_type
        self.credentialsPlugin = None
        self.authenticatorPlugin = None

    def __eq__(self, obj):
        default = object()
        result = []
        for name in ('id', 'title', 'description', 'email', 'phone',
                     'public_name', 'user_type', 'credentialsPlugin',
                     'authenticatorPlugin'):
            result.append(
                getattr(self, name) == getattr(obj, name, default))
        return False not in result


class KofaPrincipal(Principal):
    """A portal principal.

    Kofa principals provide an extra `email`, `phone`, `public_name`
    and `user_type` attribute extending ordinary principals.
    """

    grok.implements(IKofaPrincipal)

    def __init__(self, id, title=u'', description=u'', email=u'',
                 phone=None, public_name=u'', user_type=u'', prefix=None):
        self.id = id
        if prefix is not None:
            self.id = '%s.%s' % (prefix, self.id)
        self.title = title
        self.description = description
        self.groups = []
        self.email = email
        self.phone = phone
        self.public_name = public_name
        self.user_type = user_type

    def __repr__(self):
        return 'KofaPrincipal(%r)' % self.id


class AuthenticatedKofaPrincipalFactory(grok.MultiAdapter):
    """Creates 'authenticated' Kofa principals.

    Adapts (principal info, request) to a KofaPrincipal instance.

    This adapter is used by the standard PAU to transform
    KofaPrincipalInfos into KofaPrincipal instances.
    """
    grok.adapts(IKofaPrincipalInfo, IRequest)
    grok.implements(IAuthenticatedPrincipalFactory)

    def __init__(self, info, request):
        self.info = info
        self.request = request

    def __call__(self, authentication):
        principal = KofaPrincipal(
            self.info.id,
            self.info.title,
            self.info.description,
            self.info.email,
            self.info.phone,
            self.info.public_name,
            self.info.user_type,
            authentication.prefix,
            )
        notify(
            AuthenticatedPrincipalCreated(
                authentication, principal, self.info, self.request))
        return principal


class FailedLoginInfo(grok.Model):
    grok.implements(IFailedLoginInfo)

    def __init__(self, num=0, last=None):
        self.num = num
        self.last = last
        return

    def as_tuple(self):
        return (self.num, self.last)

    def set_values(self, num=0, last=None):
        self.num, self.last = num, last
        self._p_changed = True
        pass

    def increase(self):
        self.set_values(num=self.num + 1, last=time.time())
        pass

    def reset(self):
        self.set_values(num=0, last=None)
        pass


class Account(grok.Model):
    """Kofa user accounts store infos about a user.

    Beside the usual data and an (encrypted) password, accounts also
    have a persistent attribute `failed_logins` which is an instance
    of `waeup.kofa.authentication.FailedLoginInfo`.

    This attribute can be manipulated directly (set new value,
    increase values, or reset).
    """
    grok.implements(IUserAccount)

    def __init__(self, name, password, title=None, description=None,
                 email=None, phone=None, public_name=None, roles=[]):
        self.name = name
        if title is None:
            title = name
        self.title = title
        self.description = description
        self.email = email
        self.phone = phone
        self.public_name = public_name
        self.suspended = False
        self.setPassword(password)
        self.setSiteRolesForPrincipal(roles)

        # We don't want to share this dict with other accounts
        self._local_roles = dict()
        self.failed_logins = FailedLoginInfo()

    def setPassword(self, password):
        passwordmanager = getUtility(IPasswordManager, 'SSHA')
        self.password = passwordmanager.encodePassword(password)

    def checkPassword(self, password):
        try:
            blocker = grok.getSite()['configuration'].maintmode_enabled_by
            if blocker and blocker != self.name and self.name != 'admin':
                return False
        except (TypeError, KeyError):  # in unit tests
            pass
        if not isinstance(password, basestring):
            return False
        if not self.password:
            # unset/empty passwords do never match
            return False
        # Do not accept password if password is insecure.
        validator = getUtility(IPasswordValidator)
        if validator.validate_secure_password(password, password):
            return False
        if self.suspended:
            return False
        passwordmanager = getUtility(IPasswordManager, 'SSHA')
        return passwordmanager.checkPassword(self.password, password)

    def getSiteRolesForPrincipal(self):
        prm = get_principal_role_manager()
        roles = [x[0] for x in prm.getRolesForPrincipal(self.name)
                 if x[0].startswith('waeup.')]
        return roles

    def setSiteRolesForPrincipal(self, roles):
        prm = get_principal_role_manager()
        old_roles = self.getSiteRolesForPrincipal()
        if sorted(old_roles) == sorted(roles):
            return
        for role in old_roles:
            # Remove old roles, not to be set now...
            if role.startswith('waeup.') and role not in roles:
                prm.unsetRoleForPrincipal(role, self.name)
        for role in roles:
            # Convert role to ASCII string to be in line with the
            # event handler
            prm.assignRoleToPrincipal(str(role), self.name)
        return

    roles = property(getSiteRolesForPrincipal, setSiteRolesForPrincipal)

    def getLocalRoles(self):
        return self._local_roles

    def notifyLocalRoleChanged(self, obj, role_id, granted=True):
        objects = self._local_roles.get(role_id, [])
        if granted and obj not in objects:
            objects.append(obj)
        if not granted and obj in objects:
            objects.remove(obj)
        self._local_roles[role_id] = objects
        if len(objects) == 0:
            del self._local_roles[role_id]
        self._p_changed = True
        return


class UserAuthenticatorPlugin(grok.GlobalUtility):
    grok.implements(IAuthenticatorPlugin)
    grok.provides(IAuthenticatorPlugin)
    grok.name('users')

    def authenticateCredentials(self, credentials):
        if not isinstance(credentials, dict):
            return None
        if not ('login' in credentials and 'password' in credentials):
            return None
        account = self.getAccount(credentials['login'])
        if account is None:
            return None
        # The following shows how 'time penalties' could be enforced
        # on failed logins. First three failed logins are 'for
        # free'. After that the user has to wait for 1, 2, 4, 8, 16,
        # 32, ... seconds before a login can succeed.
        # There are, however, some problems to discuss, before we
        # really use this in all authenticators.

        #num, last = account.failed_logins.as_tuple()
        #if (num > 2) and (time.time() < (last + 2**(num-3))):
        #    # tried login while account still blocked due to previous
        #    # login errors.
        #    return None
        if not account.checkPassword(credentials['password']):
            #account.failed_logins.increase()
            return None
        return KofaPrincipalInfo(
            id=account.name,
            title=account.title,
            description=account.description,
            email=account.email,
            phone=account.phone,
            public_name=account.public_name,
            user_type=u'user')

    def principalInfo(self, id):
        account = self.getAccount(id)
        if account is None:
            return None
        return KofaPrincipalInfo(
            id=account.name,
            title=account.title,
            description=account.description,
            email=account.email,
            phone=account.phone,
            public_name=account.public_name,
            user_type=u'user')

    def getAccount(self, login):
        # ... look up the account object and return it ...
        userscontainer = self.getUsersContainer()
        if userscontainer is None:
            return
        return userscontainer.get(login, None)

    def addAccount(self, account):
        userscontainer = self.getUsersContainer()
        if userscontainer is None:
            return
        # XXX: complain if name already exists...
        userscontainer.addAccount(account)

    def addUser(self, name, password, title=None, description=None):
        userscontainer = self.getUsersContainer()
        if userscontainer is None:
            return
        userscontainer.addUser(name, password, title, description)

    def getUsersContainer(self):
        site = grok.getSite()
        return site['users']


class PasswordValidator(grok.GlobalUtility):

    grok.implements(IPasswordValidator)

    def validate_password(self, pw, pw_repeat):
        errors = []
        if len(pw) < 6:
            errors.append(translate(_(
                'Password must have at least 6 characters.')))
        if pw != pw_repeat:
            errors.append(translate(_('Passwords do not match.')))
        return errors

    def validate_secure_password(self, pw, pw_repeat):
        """
        ^(?=.*[A-Z])(?=.*[a-z])(?=.*[0-9]).{8,}$

        ^              Start anchor
        (?=.*[A-Z])    Ensure password has one uppercase letters.
        (?=.*[0-9])    Ensure password has one digit.
        (?=.*[a-z])    Ensure password has one lowercase letter.
        .{8,}          Ensure password is of length 8.
        $              End anchor
        """
        check_pw = re.compile(r"^(?=.*[A-Z])(?=.*[a-z])(?=.*[0-9]).{8,}$").match
        errors = []
        if not check_pw(pw):
            errors.append(translate(_(
                'Passwords must be at least 8 characters long, '
                'must contain at least one uppercase letter, '
                'one lowercase letter and one digit.')))
        if pw != pw_repeat:
            errors.append(translate(_('Passwords do not match.')))
        return errors


class LocalRoleSetEvent(object):

    grok.implements(ILocalRoleSetEvent)

    def __init__(self, object, role_id, principal_id, granted=True):
        self.object = object
        self.role_id = role_id
        self.principal_id = principal_id
        self.granted = granted


@grok.subscribe(IUserAccount, grok.IObjectRemovedEvent)
def handle_account_removed(account, event):
    """When an account is removed, local and global roles might
    have to be deleted.
    """
    local_roles = account.getLocalRoles()
    principal = account.name

    for role_id, object_list in local_roles.items():
        for object in object_list:
            try:
                role_manager = IPrincipalRoleManager(object)
            except TypeError:
                # No Account object, no role manager, no local roles to remove
                continue
            role_manager.unsetRoleForPrincipal(role_id, principal)
    role_manager = IPrincipalRoleManager(grok.getSite())
    roles = account.getSiteRolesForPrincipal()
    for role_id in roles:
        role_manager.unsetRoleForPrincipal(role_id, principal)
    return


@grok.subscribe(IUserAccount, grok.IObjectAddedEvent)
def handle_account_added(account, event):
    """When an account is added, the local owner role and the global
    AcademicsOfficer role must be set.
    """
    # We set the local Owner role
    role_manager_account = IPrincipalRoleManager(account)
    role_manager_account.assignRoleToPrincipal(
        'waeup.local.Owner', account.name)
    # We set the global AcademicsOfficer role
    site = grok.getSite()
    role_manager_site = IPrincipalRoleManager(site)
    role_manager_site.assignRoleToPrincipal(
        'waeup.AcademicsOfficer', account.name)
    # Finally we have to notify the user account that the local role
    # of the same object has changed
    notify(LocalRoleSetEvent(
        account, 'waeup.local.Owner', account.name, granted=True))
    return


@grok.subscribe(Interface, ILocalRoleSetEvent)
def handle_local_role_changed(obj, event):
    site = grok.getSite()
    if site is None:
        return
    users = site.get('users', None)
    if users is None:
        return
    if event.principal_id not in users.keys():
        return
    user = users[event.principal_id]
    user.notifyLocalRoleChanged(event.object, event.role_id, event.granted)
    return


@grok.subscribe(Interface, grok.IObjectRemovedEvent)
def handle_local_roles_on_obj_removed(obj, event):
    try:
        role_map = IPrincipalRoleMap(obj)
    except TypeError:
        # no map, no roles to remove
        return
    for local_role, user_name, setting in role_map.getPrincipalsAndRoles():
        notify(LocalRoleSetEvent(
            obj, local_role, user_name, granted=False))
    return


class UserAccountFactory(grok.GlobalUtility):
    """A factory for user accounts.

    This factory is only needed for imports.
    """
    grok.implements(IFactory)
    grok.name(u'waeup.UserAccount')
    title = u"Create a user.",
    description = u"This factory instantiates new user account instances."

    def __call__(self, *args, **kw):
        return Account(name=None, password='')

    def getInterfaces(self):
        return implementedBy(Account)


class UserProcessor(BatchProcessor):
    """The User Processor processes user accounts, i.e. `Account` objects in
    the ``users`` container.

    The `roles` columns must contain Python list
    expressions like ``['waeup.PortalManager', 'waeup.ImportManager']``.

    The processor does not import local roles. These can be imported
    by means of batch processors in the academic section.
    """
    grok.implements(IBatchProcessor)
    grok.provides(IBatchProcessor)
    grok.context(Interface)
    util_name = 'userprocessor'
    grok.name(util_name)

    name = u'User Processor'
    iface = IUserAccount

    location_fields = ['name', ]
    factory_name = 'waeup.UserAccount'

    mode = None

    def parentsExist(self, row, site):
        return 'users' in site.keys()

    def entryExists(self, row, site):
        return row['name'] in site['users'].keys()

    def getParent(self, row, site):
        return site['users']

    def getEntry(self, row, site):
        if not self.entryExists(row, site):
            return None
        parent = self.getParent(row, site)
        return parent.get(row['name'])

    def addEntry(self, obj, row, site):
        parent = self.getParent(row, site)
        parent.addAccount(obj)
        return

    def delEntry(self, row, site):
        user = self.getEntry(row, site)
        if user is not None:
            parent = self.getParent(row, site)
            grok.getSite().logger.info(
                '%s - %s - User removed' % (self.name, row['name']))
            del parent[user.name]
        pass

    def updateEntry(self, obj, row, site, filename):
        """Update obj to the values given in row.
        """
        changed = []
        for key, value in row.items():
            if key == 'roles':
                # We cannot simply set the roles attribute here because
                # we can't assure that the name attribute is set before
                # the roles attribute is set.
                continue
            # Skip fields to be ignored.
            if value == IGNORE_MARKER:
                continue
            if not hasattr(obj, key):
                continue
            setattr(obj, key, value)
            changed.append('%s=%s' % (key, value))
        roles = row.get('roles', IGNORE_MARKER)
        if roles not in ('', IGNORE_MARKER):
            evalvalue = eval(roles)
            if isinstance(evalvalue, list):
                setattr(obj, 'roles', evalvalue)
                changed.append('roles=%s' % roles)
        # Log actions...
        items_changed = ', '.join(changed)
        grok.getSite().logger.info(
            '%s - %s - %s - updated: %s' % (
                self.name, filename, row['name'], items_changed))
        return

    def checkConversion(self, row, mode='ignore'):
        """Validates all values in row.
        """
        errs, inv_errs, conv_dict = super(
            UserProcessor, self).checkConversion(row, mode=mode)
        # We need to check if roles exist.
        roles = row.get('roles', IGNORE_MARKER)
        all_roles = [i[0] for i in get_all_roles()]
        if roles not in ('', IGNORE_MARKER):
            evalvalue = eval(roles)
            for role in evalvalue:
                if role not in all_roles:
                    errs.append(('roles', 'invalid role'))
        return errs, inv_errs, conv_dict


class UsersPlugin(grok.GlobalUtility):
    """A plugin that updates users.
    """
    grok.implements(IKofaPluggable)
    grok.name('users')

    deprecated_attributes = []

    def setup(self, site, name, logger):
        return

    def update(self, site, name, logger):
        users = site['users']
        items = getFields(IUserAccount).items()
        for user in users.values():
            # Add new attributes
            for i in items:
                if not hasattr(user, i[0]):
                    setattr(user, i[0], i[1].missing_value)
                    logger.info(
                        'UsersPlugin: %s attribute %s added.' % (
                            user.name, i[0]))
            if not hasattr(user, 'failed_logins'):
                # add attribute `failed_logins`...
                user.failed_logins = FailedLoginInfo()
                logger.info(
                    'UsersPlugin: attribute failed_logins added.')
            # Remove deprecated attributes
            for i in self.deprecated_attributes:
                try:
                    delattr(user, i)
                    logger.info(
                        'UsersPlugin: %s attribute %s deleted.' % (
                            user.name, i))
                except AttributeError:
                    pass
        return


class UpdatePAUPlugin(grok.GlobalUtility):
    """A plugin that updates a local PAU.

    We insert an 'xmlrpc-credentials' PAU-plugin into a sites PAU if it is not
    present already. There must be 'credentials' plugin registered already.

    XXX: This Plugin fixes a shortcoming of waeup.kofa 1.5. Sites created or
         updated afterwards do not need this plugin and it should be removed.
    """
    grok.implements(IKofaPluggable)
    grok.name('site-pluggable-auth')

    def setup(self, site, name, logger):
        return

    def update(self, site, name, logger):
        pau = site.getSiteManager()['PluggableAuthentication']
        if 'xmlrpc-credentials' in pau.credentialsPlugins:
            return
        plugins = list(pau.credentialsPlugins)
        plugins.insert(plugins.index('credentials'), 'xmlrpc-credentials')
        pau.credentialsPlugins = tuple(plugins)