Logo Search packages:      
Sourcecode: zope-ldapmultiplugins version File versions  Download package

LDAPPluginBase.py

##############################################################################
#
# LDAPPluginBase    Base class for LDAP-based PAS-Plugins
#
##############################################################################

__doc__     = """ LDAPPluginBase module """
__version__ = '$Revision: 1378 $'[11:-2]

# General Python imports
import copy
import logging
import os
from urllib import quote_plus

# Zope imports
from Acquisition import aq_base
from OFS.Folder import Folder
from OFS.Cache import Cacheable
from Globals import InitializeClass
from AccessControl import ClassSecurityInfo
from AccessControl.SecurityManagement import getSecurityManager

from Products.PluggableAuthService.plugins.BasePlugin import BasePlugin
from Products.PluggableAuthService.interfaces.plugins import \
     IAuthenticationPlugin, IRolesPlugin, ICredentialsResetPlugin, \
     IPropertiesPlugin
from Products.PluggableAuthService.utils import classImplements


logger = logging.getLogger('event.LDAPMultiPlugin')


00034 class LDAPPluginBase(Folder, BasePlugin, Cacheable):
    """ Base class for LDAP-based PAS plugins """
    security = ClassSecurityInfo()

    manage_options = ( BasePlugin.manage_options[:1]
                     + Folder.manage_options 
                     + Cacheable.manage_options
                     )

    _properties = BasePlugin._properties + Folder._properties


    # default 'id' attribute for groups
    groupid_attr = 'cn'

00049     def __init__(self, id, title=''):
        """ Initialize a new instance """
        self.id = id
        self.title = title


    security.declarePrivate('_getLDAPUserFolder')
00056     def _getLDAPUserFolder(self):
        """ Safely retrieve a LDAPUserFolder to work with """
        embedded_luf = getattr(aq_base(self), 'acl_users', None)

        return embedded_luf


    security.declarePrivate('authenticateCredentials')
00064     def authenticateCredentials(self, credentials):
        """ Fulfill AuthenticationPlugin requirements """
        acl = self._getLDAPUserFolder()
        login = credentials.get('login')
        password = credentials.get('password')

        if not acl or not login or not password:
            return None, None

        user = acl.getUser(login, pwd=password)

        if user is None:
            return None, None

        return (user.getId(), user.getUserName())


    security.declarePrivate('resetCredentials')
00082     def resetCredentials(self, request, response):
        """ Fulfill CredentialsResetPlugin requirements """
        user = getSecurityManager().getUser()
        acl = self._getLDAPUserFolder()

        if user:
            acl._expireUser(user)


    security.declarePrivate('getPropertiesForUser')
00092     def getPropertiesForUser(self, user, request=None):
        """ Fullfill PropertiesPlugin requirements """
        acl = self._getLDAPUserFolder()

        if acl is None:
            return {}

        unmangled_userid = self._demangle(user.getId())
        if unmangled_userid is None:
            return {}

        ldap_user = acl.getUserById(unmangled_userid)

        if ldap_user is None:
            return {}

        # XXX Direct attribute access. Waaa!
        properties = copy.deepcopy(ldap_user._properties)

        # Need to clean up: The propertysheet mechanism will
        # blow up if "None" is encountered
        for key, val in properties.items():
            if val is None:
                properties[key] = ''

        return properties


    security.declarePrivate('getRolesForPrincipal')
00121     def getRolesForPrincipal(self, user, request=None):
        """ Fullfill RolesPlugin requirements """
        acl = self._getLDAPUserFolder()

        if acl is None:
            return ()

        unmangled_userid = self._demangle(user.getId())
        if unmangled_userid is None:
            return ()

        ldap_user = acl.getUserById(unmangled_userid)
        if ldap_user is None:
            return ()  

        groups = self.getGroupsForPrincipal(user, request)
        roles = list(acl._mapRoles(groups))
        roles.extend(acl._roles)

        return tuple(roles)


    security.declarePrivate('_demangle')
    def _demangle(self, princid):
        # User must start with our prefix (which is likely to be blank anyway)
        if not princid.startswith(self.prefix):
            return None
        return princid[len(self.prefix):]

    # Helper methods for simple group caching
    security.declarePrivate('_getGroupInfoCacheKey')
00152     def _getGroupInfoCacheKey(self, gid):
        """_getGroupInfoCacheKey(id) -> (view_name, keywords)

        given a group id, return view_name and keywords to be used when 
        querying and storing into the group cache
        """
        view_name = self.getId() + '__GroupInfoCache'
        keywords = { 'id' : gid }
        return view_name, keywords

    security.declarePrivate('_setGroupInfoCache')
00163     def _setGroupInfoCache(self, info):
        """Cache a group info"""
        gid = info['id']
        view_name, keywords = self._getGroupInfoCacheKey(gid)
        self.ZCacheable_set(info, view_name=view_name, keywords=keywords)

    security.declarePrivate('_getGroupInfoCache')
00170     def _getGroupInfoCache(self, gid, default=None):
        """Retrieve a group info from cache, given its group id.

        Returns None or the passed-in default if the cache
        has no group with such id
        """
        view_name, keywords = self._getGroupInfoCacheKey(gid)
        result = self.ZCacheable_get( view_name=view_name
                                    , keywords=keywords
                                    , default=default
                                    )
        return result


classImplements( LDAPPluginBase
               , IAuthenticationPlugin
               , ICredentialsResetPlugin
               , IPropertiesPlugin
               , IRolesPlugin
               )

InitializeClass(LDAPPluginBase)

Generated by  Doxygen 1.6.0   Back to index