r640 - in zope-pas/branches/upstream/current: . doc plugins
plugins/tests tests
Fabio Tranchitella
kobold at alioth.debian.org
Fri Feb 9 15:00:25 CET 2007
Author: kobold
Date: 2007-02-09 15:00:25 +0100 (Fri, 09 Feb 2007)
New Revision: 640
Added:
zope-pas/branches/upstream/current/tests/test_utils.py
Modified:
zope-pas/branches/upstream/current/PluggableAuthService.py
zope-pas/branches/upstream/current/doc/CHANGES.txt
zope-pas/branches/upstream/current/plugins/BasePlugin.py
zope-pas/branches/upstream/current/plugins/ZODBUserManager.py
zope-pas/branches/upstream/current/plugins/tests/test_ZODBUserManager.py
zope-pas/branches/upstream/current/tests/conformance.py
zope-pas/branches/upstream/current/tests/test_PluggableAuthService.py
zope-pas/branches/upstream/current/utils.py
zope-pas/branches/upstream/current/version.txt
Log:
[svn-upgrade] Integrating new upstream version, zope-pas (1.4.1)
Modified: zope-pas/branches/upstream/current/PluggableAuthService.py
===================================================================
--- zope-pas/branches/upstream/current/PluggableAuthService.py 2007-02-09 13:59:28 UTC (rev 639)
+++ zope-pas/branches/upstream/current/PluggableAuthService.py 2007-02-09 14:00:25 UTC (rev 640)
@@ -14,7 +14,7 @@
##############################################################################
""" Classes: PluggableAuthService
-$Id: PluggableAuthService.py 69264 2006-07-25 22:59:44Z tseaver $
+$Id: PluggableAuthService.py 71607 2006-12-19 18:38:09Z tseaver $
"""
import logging
@@ -85,6 +85,7 @@
from PropertiedUser import PropertiedUser
from utils import _wwwdir
from utils import createViewName
+from utils import createKeywords
from utils import classImplements
security = ModuleSecurityInfo(
@@ -536,9 +537,6 @@
a user; accumulate a list of the IDs of such users over all
our authentication and extraction plugins.
"""
- result = []
- user_ids = []
-
try:
extractors = plugins.listPlugins( IExtractionPlugin )
except _SWALLOWABLE_PLUGIN_EXCEPTIONS:
@@ -554,6 +552,8 @@
logger.debug('Authenticator plugin listing error', exc_info=True)
authenticators = ()
+ result = []
+
for extractor_id, extractor in extractors:
try:
@@ -568,24 +568,32 @@
try:
credentials[ 'extractor' ] = extractor_id # XXX: in key?
+ # Test if ObjectCacheEntries.aggregateIndex would work
items = credentials.items()
items.sort()
except _SWALLOWABLE_PLUGIN_EXCEPTIONS:
logger.debug( 'Credentials error: %s' % credentials
, exc_info=True
)
- else:
- user_ids = []
+ continue
- if not user_ids:
+ # First try to authenticate against the emergency
+ # user and return immediately if authenticated
+ user_id, name = self._tryEmergencyUserAuthentication(
+ credentials )
- # first try to authenticate against the emergency
- # user, and return immediately if authenticated
- user_id, name = self._tryEmergencyUserAuthentication(
- credentials )
+ if user_id is not None:
+ return [ ( user_id, name ) ]
- if user_id is not None:
- return [ ( user_id, name ) ]
+ # Now see if the user ids can be retrieved from the cache
+ view_name = createViewName('_extractUserIds', credentials.get('login'))
+ keywords = createKeywords(**credentials)
+ user_ids = self.ZCacheable_get( view_name=view_name
+ , keywords=keywords
+ , default=None
+ )
+ if user_ids is None:
+ user_ids = []
for authenticator_id, auth in authenticators:
@@ -607,14 +615,20 @@
if user_id is not None:
user_ids.append( (user_id, info) )
+ if user_ids:
+ self.ZCacheable_set( user_ids
+ , view_name=view_name
+ , keywords=keywords
+ )
+
result.extend( user_ids )
- if not user_ids:
- user_id, name = self._tryEmergencyUserAuthentication(
- DumbHTTPExtractor().extractCredentials( request ) )
+ # Emergency user via HTTP basic auth always wins
+ user_id, name = self._tryEmergencyUserAuthentication(
+ DumbHTTPExtractor().extractCredentials( request ) )
- if user_id is not None:
- result.append( ( user_id, name ) )
+ if user_id is not None:
+ return [ ( user_id, name ) ]
return result
@@ -700,10 +714,8 @@
return self._emergency_user
# See if the user can be retrieved from the cache
- view_name = '_findUser-%s' % user_id
- keywords = { 'user_id' : user_id
- , 'name' : name
- }
+ view_name = createViewName('_findUser', user_id)
+ keywords = createKeywords(user_id=user_id, name=name)
user = self.ZCacheable_get( view_name=view_name
, keywords=keywords
, default=None
@@ -761,8 +773,9 @@
if criteria:
view_name = createViewName('_verifyUser', user_id or login)
+ keywords = createKeywords(**criteria)
cached_info = self.ZCacheable_get( view_name=view_name
- , keywords=criteria
+ , keywords=keywords
, default=None
)
@@ -780,7 +793,7 @@
# Put the computed value into the cache
self.ZCacheable_set( info[0]
, view_name=view_name
- , keywords=criteria
+ , keywords=keywords
)
return info[0]
@@ -928,6 +941,7 @@
for useradder_id, useradder in useradders:
if useradder.doAddUser( login, password ):
+ # XXX: Adds user to cache, but without roles...
user = self.getUser( login )
break
Modified: zope-pas/branches/upstream/current/doc/CHANGES.txt
===================================================================
--- zope-pas/branches/upstream/current/doc/CHANGES.txt 2007-02-09 13:59:28 UTC (rev 639)
+++ zope-pas/branches/upstream/current/doc/CHANGES.txt 2007-02-09 14:00:25 UTC (rev 640)
@@ -1,5 +1,20 @@
PluggableAuthService changelog
+ PluggableAuthService 1.4.1 (2006/12/19)
+
+ Bugs Fixed
+
+ - BasePlugin: The listInterfaces method only considered the old-style
+ __implements__ machinery when determining interfaces provided by
+ a plugin instance.
+
+ - Made sure the emergency user via HTTP basic auth always wins, no matter
+ how borken the plugin landscape.
+
+ - ZODBUserManager: Already encrypted passwords were encrypted again in
+ addUser and updateUserPassword (backported from trunk).
+ (http://www.zope.org/Collectors/Zope/1926)
+
PluggableAuthService 1.4 (2006/08/28)
Bugs Fixed
Modified: zope-pas/branches/upstream/current/plugins/BasePlugin.py
===================================================================
--- zope-pas/branches/upstream/current/plugins/BasePlugin.py 2007-02-09 13:59:28 UTC (rev 639)
+++ zope-pas/branches/upstream/current/plugins/BasePlugin.py 2007-02-09 14:00:25 UTC (rev 640)
@@ -14,7 +14,7 @@
##############################################################################
""" Class: BasePlugin
-$Id: BasePlugin.py 39312 2005-07-06 18:49:05Z urbanape $
+$Id: BasePlugin.py 71608 2006-12-19 20:20:51Z tseaver $
"""
from OFS.SimpleItem import SimpleItem
from OFS.PropertyManager import PropertyManager
@@ -25,7 +25,9 @@
from Products.PageTemplates.PageTemplateFile import PageTemplateFile
-from Products.PluggableAuthService.utils import classImplements, implementedBy
+from Products.PluggableAuthService.utils import classImplements
+from Products.PluggableAuthService.utils import implementedBy
+from Products.PluggableAuthService.utils import providedBy
from Products.PluggableAuthService.permissions import ManageUsers
class BasePlugin(SimpleItem, PropertyManager):
@@ -60,7 +62,7 @@
results = []
- for iface in flattenInterfaces( self.__implements__ ):
+ for iface in flattenInterfaces( providedBy( self ) ):
results.append( iface.__name__ )
return results
Modified: zope-pas/branches/upstream/current/plugins/ZODBUserManager.py
===================================================================
--- zope-pas/branches/upstream/current/plugins/ZODBUserManager.py 2007-02-09 13:59:28 UTC (rev 639)
+++ zope-pas/branches/upstream/current/plugins/ZODBUserManager.py 2007-02-09 14:00:25 UTC (rev 640)
@@ -14,7 +14,7 @@
##############################################################################
""" Classes: ZODBUserManager
-$Id: ZODBUserManager.py 68822 2006-06-24 11:01:17Z jens $
+$Id: ZODBUserManager.py 71605 2006-12-19 18:15:37Z tseaver $
"""
import sha
import copy
@@ -279,7 +279,7 @@
if self._login_to_userid.get( login_name ) is not None:
raise KeyError, 'Duplicate login name: %s' % login_name
- self._user_passwords[ user_id ] = AuthEncoding.pw_encrypt( password )
+ self._user_passwords[ user_id ] = self._pw_encrypt( password)
self._login_to_userid[ login_name ] = user_id
self._userid_to_login[ user_id ] = login_name
@@ -322,9 +322,19 @@
raise KeyError, 'Invalid user ID: %s' % user_id
if password:
- digested = AuthEncoding.pw_encrypt( password )
- self._user_passwords[ user_id ] = digested
+ self._user_passwords[ user_id ] = self._pw_encrypt( password )
+ security.declarePrivate( '_pw_encrypt' )
+ def _pw_encrypt( self, password ):
+ """Returns the AuthEncoding encrypted password
+
+ If 'password' is already encrypted, it is returned
+ as is and not encrypted again.
+ """
+ if AuthEncoding.is_encrypted( password ):
+ return password
+ return AuthEncoding.pw_encrypt( password )
+
#
# ZMI
#
Modified: zope-pas/branches/upstream/current/plugins/tests/test_ZODBUserManager.py
===================================================================
--- zope-pas/branches/upstream/current/plugins/tests/test_ZODBUserManager.py 2007-02-09 13:59:28 UTC (rev 639)
+++ zope-pas/branches/upstream/current/plugins/tests/test_ZODBUserManager.py 2007-02-09 14:00:25 UTC (rev 640)
@@ -426,7 +426,83 @@
info = zum.enumerateUsers(id='special__luser', exact_match=True)
self.assertEqual(len(info), 0)
+ def test_addUser_with_not_yet_encrypted_password(self):
+ # See collector #1869 && #1926
+ from AccessControl.AuthEncoding import is_encrypted
+ USER_ID = 'not_yet_encrypted'
+ PASSWORD = 'password'
+
+ self.failIf(is_encrypted(PASSWORD))
+
+ zum = self._makeOne()
+ zum.addUser(USER_ID, USER_ID, PASSWORD)
+
+ uid_and_info = zum.authenticateCredentials(
+ { 'login': USER_ID
+ , 'password': PASSWORD
+ })
+
+ self.assertEqual(uid_and_info, (USER_ID, USER_ID))
+
+ def test_addUser_with_preencrypted_password(self):
+ # See collector #1869 && #1926
+ from AccessControl.AuthEncoding import pw_encrypt
+
+ USER_ID = 'already_encrypted'
+ PASSWORD = 'password'
+
+ ENCRYPTED = pw_encrypt(PASSWORD)
+
+ zum = self._makeOne()
+ zum.addUser(USER_ID, USER_ID, ENCRYPTED)
+
+ uid_and_info = zum.authenticateCredentials(
+ { 'login': USER_ID
+ , 'password': PASSWORD
+ })
+
+ self.assertEqual(uid_and_info, (USER_ID, USER_ID))
+
+ def test_updateUserPassword_with_not_yet_encrypted_password(self):
+ from AccessControl.AuthEncoding import is_encrypted
+
+ USER_ID = 'not_yet_encrypted'
+ PASSWORD = 'password'
+
+ self.failIf(is_encrypted(PASSWORD))
+
+ zum = self._makeOne()
+ zum.addUser(USER_ID, USER_ID, '')
+ zum.updateUserPassword(USER_ID, PASSWORD)
+
+ uid_and_info = zum.authenticateCredentials(
+ { 'login': USER_ID
+ , 'password': PASSWORD
+ })
+
+ self.assertEqual(uid_and_info, (USER_ID, USER_ID))
+
+ def test_updateUserPassword_with_preencrypted_password(self):
+ from AccessControl.AuthEncoding import pw_encrypt
+
+ USER_ID = 'already_encrypted'
+ PASSWORD = 'password'
+
+ ENCRYPTED = pw_encrypt(PASSWORD)
+
+ zum = self._makeOne()
+ zum.addUser(USER_ID, USER_ID, '')
+ zum.updateUserPassword(USER_ID, ENCRYPTED)
+
+ uid_and_info = zum.authenticateCredentials(
+ { 'login': USER_ID
+ , 'password': PASSWORD
+ })
+
+ self.assertEqual(uid_and_info, (USER_ID, USER_ID))
+
+
if __name__ == "__main__":
unittest.main()
Modified: zope-pas/branches/upstream/current/tests/conformance.py
===================================================================
--- zope-pas/branches/upstream/current/tests/conformance.py 2007-02-09 13:59:28 UTC (rev 639)
+++ zope-pas/branches/upstream/current/tests/conformance.py 2007-02-09 14:00:25 UTC (rev 640)
@@ -14,7 +14,7 @@
##############################################################################
""" Base classes for testing plugin interface conformance.
-$Id: conformance.py 67083 2006-04-18 21:56:22Z jens $
+$Id: conformance.py 71608 2006-12-19 20:20:51Z tseaver $
"""
try:
@@ -22,6 +22,7 @@
except ImportError:
from Interface.Verify import verifyClass
+
class IExtractionPlugin_conformance:
def test_IExtractionPlugin_conformance( self ):
@@ -31,6 +32,15 @@
verifyClass( IExtractionPlugin, self._getTargetClass() )
+ def test_IExtractionPlugin_listInterfaces(self):
+
+ from Products.PluggableAuthService.interfaces.plugins \
+ import IExtractionPlugin
+
+ listed = self._makeOne().listInterfaces()
+ self.failUnless(IExtractionPlugin.__name__ in listed)
+
+
class ILoginPasswordExtractionPlugin_conformance:
def test_ILoginPasswordExtractionPlugin_conformance( self ):
@@ -40,6 +50,15 @@
verifyClass( ILoginPasswordExtractionPlugin, self._getTargetClass() )
+ def test_ILoginPasswordExtractionPlugin_listInterfaces(self):
+
+ from Products.PluggableAuthService.interfaces.plugins \
+ import ILoginPasswordExtractionPlugin
+
+ listed = self._makeOne().listInterfaces()
+ self.failUnless(ILoginPasswordExtractionPlugin.__name__ in listed)
+
+
class ILoginPasswordHostExtractionPlugin_conformance:
def test_ILoginPasswordHostExtractionPlugin_conformance( self ):
@@ -50,6 +69,15 @@
verifyClass( ILoginPasswordHostExtractionPlugin
, self._getTargetClass() )
+ def test_ILoginPasswordHostExtractionPlugin_listInterfaces(self):
+
+ from Products.PluggableAuthService.interfaces.plugins \
+ import ILoginPasswordHostExtractionPlugin
+
+ listed = self._makeOne().listInterfaces()
+ self.failUnless(ILoginPasswordHostExtractionPlugin.__name__ in listed)
+
+
class IChallengePlugin_conformance:
def test_IChallengePlugin_conformance( self ):
@@ -59,6 +87,15 @@
verifyClass( IChallengePlugin, self._getTargetClass() )
+ def test_IChallengePlugin_listInterfaces(self):
+
+ from Products.PluggableAuthService.interfaces.plugins \
+ import IChallengePlugin
+
+ listed = self._makeOne().listInterfaces()
+ self.failUnless(IChallengePlugin.__name__ in listed)
+
+
class ICredentialsUpdatePlugin_conformance:
def test_ICredentialsUpdatePlugin_conformance( self ):
@@ -68,6 +105,15 @@
verifyClass( ICredentialsUpdatePlugin, self._getTargetClass() )
+ def test_ICredentialsUpdatePlugin_listInterfaces(self):
+
+ from Products.PluggableAuthService.interfaces.plugins \
+ import ICredentialsUpdatePlugin
+
+ listed = self._makeOne().listInterfaces()
+ self.failUnless(ICredentialsUpdatePlugin.__name__ in listed)
+
+
class ICredentialsResetPlugin_conformance:
def test_ICredentialsResetPlugin_conformance( self ):
@@ -77,7 +123,15 @@
verifyClass( ICredentialsResetPlugin, self._getTargetClass() )
+ def test_ICredentialsResetPlugin_listInterfaces(self):
+ from Products.PluggableAuthService.interfaces.plugins \
+ import ICredentialsResetPlugin
+
+ listed = self._makeOne().listInterfaces()
+ self.failUnless(ICredentialsResetPlugin.__name__ in listed)
+
+
class IAuthenticationPlugin_conformance:
def test_AuthenticationPlugin_conformance( self ):
@@ -87,7 +141,15 @@
verifyClass( IAuthenticationPlugin, self._getTargetClass() )
+ def test_IAuthenticationPlugin_listInterfaces(self):
+ from Products.PluggableAuthService.interfaces.plugins \
+ import IAuthenticationPlugin
+
+ listed = self._makeOne().listInterfaces()
+ self.failUnless(IAuthenticationPlugin.__name__ in listed)
+
+
class IUserEnumerationPlugin_conformance:
def test_UserEnumerationPlugin_conformance( self ):
@@ -97,7 +159,15 @@
verifyClass( IUserEnumerationPlugin, self._getTargetClass() )
+ def test_IUserEnumerationPlugin_listInterfaces(self):
+ from Products.PluggableAuthService.interfaces.plugins \
+ import IUserEnumerationPlugin
+
+ listed = self._makeOne().listInterfaces()
+ self.failUnless(IUserEnumerationPlugin.__name__ in listed)
+
+
class IUserAdderPlugin_conformance:
def test_UserAdderPlugin_conformance( self ):
@@ -107,7 +177,15 @@
verifyClass( IUserAdderPlugin, self._getTargetClass() )
+ def test_IUserAdderPlugin_listInterfaces(self):
+ from Products.PluggableAuthService.interfaces.plugins \
+ import IUserAdderPlugin
+
+ listed = self._makeOne().listInterfaces()
+ self.failUnless(IUserAdderPlugin.__name__ in listed)
+
+
class IGroupEnumerationPlugin_conformance:
def test_GroupEnumerationPlugin_conformance( self ):
@@ -117,7 +195,15 @@
verifyClass( IGroupEnumerationPlugin, self._getTargetClass() )
+ def test_IGroupEnumerationPlugin_listInterfaces(self):
+ from Products.PluggableAuthService.interfaces.plugins \
+ import IGroupEnumerationPlugin
+
+ listed = self._makeOne().listInterfaces()
+ self.failUnless(IGroupEnumerationPlugin.__name__ in listed)
+
+
class IGroupsPlugin_conformance:
def test_GroupsPlugin_conformance( self ):
@@ -127,7 +213,15 @@
verifyClass( IGroupsPlugin, self._getTargetClass() )
+ def test_IGroupsPlugin_listInterfaces(self):
+ from Products.PluggableAuthService.interfaces.plugins \
+ import IGroupsPlugin
+
+ listed = self._makeOne().listInterfaces()
+ self.failUnless(IGroupsPlugin.__name__ in listed)
+
+
class IRoleEnumerationPlugin_conformance:
def test_RoleEnumerationPlugin_conformance( self ):
@@ -137,7 +231,15 @@
verifyClass( IRoleEnumerationPlugin, self._getTargetClass() )
+ def test_IRoleEnumerationPlugin_listInterfaces(self):
+ from Products.PluggableAuthService.interfaces.plugins \
+ import IRoleEnumerationPlugin
+
+ listed = self._makeOne().listInterfaces()
+ self.failUnless(IRoleEnumerationPlugin.__name__ in listed)
+
+
class IRolesPlugin_conformance:
def test_RolesPlugin_conformance( self ):
@@ -147,6 +249,15 @@
verifyClass( IRolesPlugin, self._getTargetClass() )
+ def test_IRolesPlugin_listInterfaces(self):
+
+ from Products.PluggableAuthService.interfaces.plugins \
+ import IRolesPlugin
+
+ listed = self._makeOne().listInterfaces()
+ self.failUnless(IRolesPlugin.__name__ in listed)
+
+
class IRoleAssignerPlugin_conformance:
def test_RoleAssignerPlugin_conformance( self ):
@@ -156,6 +267,15 @@
verifyClass( IRoleAssignerPlugin, self._getTargetClass() )
+ def test_IRoleAssignerPlugin_listInterfaces(self):
+
+ from Products.PluggableAuthService.interfaces.plugins \
+ import IRoleAssignerPlugin
+
+ listed = self._makeOne().listInterfaces()
+ self.failUnless(IRoleAssignerPlugin.__name__ in listed)
+
+
class IChallengeProtocolChooser_conformance:
def test_ChallengeProtocolChooser_conformance( self ):
@@ -165,6 +285,15 @@
verifyClass( IChallengeProtocolChooser, self._getTargetClass() )
+ def test_IChallengeProtocolChooser_listInterfaces(self):
+
+ from Products.PluggableAuthService.interfaces.plugins \
+ import IChallengeProtocolChooser
+
+ listed = self._makeOne().listInterfaces()
+ self.failUnless(IChallengeProtocolChooser.__name__ in listed)
+
+
class IRequestTypeSniffer_conformance:
def test_RequestTypeSniffer_conformance( self ):
@@ -174,6 +303,15 @@
verifyClass( IRequestTypeSniffer, self._getTargetClass() )
+ def test_IRequestTypeSniffer_listInterfaces(self):
+
+ from Products.PluggableAuthService.interfaces.plugins \
+ import IRequestTypeSniffer
+
+ listed = self._makeOne().listInterfaces()
+ self.failUnless(IRequestTypeSniffer.__name__ in listed)
+
+
class IUserFolder_conformance:
def test_conformance_IUserFolder( self ):
@@ -183,6 +321,7 @@
verifyClass( IUserFolder, self._getTargetClass() )
+
class IBasicUser_conformance:
def test_conformance_IBasicUser( self ):
@@ -192,6 +331,7 @@
verifyClass( IBasicUser, self._getTargetClass() )
+
class IPropertiedUser_conformance:
def test_conformance_IPropertiedUser( self ):
@@ -201,6 +341,7 @@
verifyClass( IPropertiedUser, self._getTargetClass() )
+
class IPropertySheet_conformance:
def test_conformance_IPropertySheet( self ):
@@ -209,3 +350,4 @@
import IPropertySheet
verifyClass( IPropertySheet, self._getTargetClass() )
+
Modified: zope-pas/branches/upstream/current/tests/test_PluggableAuthService.py
===================================================================
--- zope-pas/branches/upstream/current/tests/test_PluggableAuthService.py 2007-02-09 13:59:28 UTC (rev 639)
+++ zope-pas/branches/upstream/current/tests/test_PluggableAuthService.py 2007-02-09 14:00:25 UTC (rev 640)
@@ -572,7 +572,7 @@
self.assertEqual( user_ids[0][0], 'qux' )
self.assertEqual( user_ids[1][0], 'foo' )
- def test__extractUserIds_broken_extractor( self ):
+ def test__extractUserIds_broken_extractor_before_good_extractor( self ):
from Products.PluggableAuthService.interfaces.plugins \
import IExtractionPlugin, IAuthenticationPlugin
@@ -595,8 +595,8 @@
plugins = zcuf._getOb( 'plugins' )
- plugins.activatePlugin( IExtractionPlugin, 'borked' )
- plugins.activatePlugin( IExtractionPlugin, 'login' )
+ plugins.activatePlugin( IExtractionPlugin, 'borked' ) # 1
+ plugins.activatePlugin( IExtractionPlugin, 'login' ) # 2
plugins.activatePlugin( IAuthenticationPlugin, 'login' )
request = self._makeRequest( form={ 'login' : 'foo'
@@ -609,11 +609,48 @@
self.assertEqual( len( user_ids ), 1 )
self.assertEqual( user_ids[0][0], 'foo' )
- def test_authenticate_emergency_user_with_broken_extractor( self ):
+ def test__extractUserIds_broken_extractor_after_good_extractor( self ):
from Products.PluggableAuthService.interfaces.plugins \
import IExtractionPlugin, IAuthenticationPlugin
+ plugins = self._makePlugins()
+ zcuf = self._makeOne( plugins )
+
+ login = DummyPlugin()
+ directlyProvides( login, IExtractionPlugin, IAuthenticationPlugin )
+ login.extractCredentials = _extractLogin
+ login.authenticateCredentials = _authLogin
+
+ zcuf._setObject( 'login', login )
+
+ borked = DummyPlugin()
+ directlyProvides( borked, IExtractionPlugin )
+ borked.extractCredentials = lambda req: 'abc'
+
+ zcuf._setObject( 'borked', borked )
+
+ plugins = zcuf._getOb( 'plugins' )
+
+ plugins.activatePlugin( IExtractionPlugin, 'login' ) # 1
+ plugins.activatePlugin( IExtractionPlugin, 'borked' ) # 2
+ plugins.activatePlugin( IAuthenticationPlugin, 'login' )
+
+ request = self._makeRequest( form={ 'login' : 'foo'
+ , 'password' : 'bar' } )
+
+ user_ids = zcuf._extractUserIds( request=request
+ , plugins=zcuf.plugins
+ )
+
+ self.assertEqual( len( user_ids ), 1 )
+ self.assertEqual( user_ids[0][0], 'foo' )
+
+ def test__extractUserIds_authenticate_emergency_user_with_broken_extractor( self ):
+
+ from Products.PluggableAuthService.interfaces.plugins \
+ import IExtractionPlugin, IAuthenticationPlugin
+
from AccessControl.User import UnrestrictedUser
from Products.PluggableAuthService import PluggableAuthService
@@ -624,21 +661,61 @@
PluggableAuthService.emergency_user = eu
+ try:
+ plugins = self._makePlugins()
+ zcuf = self._makeOne( plugins )
+
+ borked = DummyPlugin()
+ directlyProvides( borked, IExtractionPlugin )
+ borked.extractCredentials = lambda req: 'abc'
+
+ zcuf._setObject( 'borked', borked )
+
+ plugins = zcuf._getOb( 'plugins' )
+
+ plugins.activatePlugin( IExtractionPlugin, 'borked' )
+
+ request = self._makeRequest( form={ 'login' : eu.getUserName()
+ , 'password' : eu._getPassword() } )
+
+ user_ids = zcuf._extractUserIds( request=request
+ , plugins=zcuf.plugins
+ )
+
+ self.assertEqual( len( user_ids ), 1 )
+ self.assertEqual( user_ids[0][0], 'foo' )
+ finally:
+ PluggableAuthService.emergency_user = old_eu
+
+ def test__extractUserIds_broken_authenticator_before_good_authenticator( self ):
+
+ from Products.PluggableAuthService.interfaces.plugins \
+ import IExtractionPlugin, IAuthenticationPlugin
+
plugins = self._makePlugins()
zcuf = self._makeOne( plugins )
+ login = DummyPlugin()
+ directlyProvides( login, IExtractionPlugin, IAuthenticationPlugin )
+ login.extractCredentials = _extractLogin
+ login.authenticateCredentials = _authLogin
+
+ zcuf._setObject( 'login', login )
+
borked = DummyPlugin()
- directlyProvides( borked, IExtractionPlugin )
- borked.extractCredentials = lambda req: 'abc'
+ directlyProvides( borked, IAuthenticationPlugin )
+ borked.authenticateCredentials = lambda creds: creds['nonesuch']
zcuf._setObject( 'borked', borked )
plugins = zcuf._getOb( 'plugins' )
- plugins.activatePlugin( IExtractionPlugin, 'borked' )
+ plugins.activatePlugin( IExtractionPlugin, 'login' )
+ plugins.activatePlugin( IAuthenticationPlugin, 'borked' ) # 1
+ plugins.activatePlugin( IAuthenticationPlugin, 'login' ) # 2
- request = self._makeRequest( form={ 'login' : eu.getUserName()
- , 'password' : eu._getPassword() } )
+ request = self._makeRequest( form={ 'login' : 'foo'
+ , 'password' : 'bar' } )
user_ids = zcuf._extractUserIds( request=request
, plugins=zcuf.plugins
@@ -647,10 +724,8 @@
self.assertEqual( len( user_ids ), 1 )
self.assertEqual( user_ids[0][0], 'foo' )
- PluggableAuthService.emergency_user = old_eu
+ def test__extractUserIds_broken_authenticator_after_good_authenticator( self ):
- def test__extractUserIds_broken_authenticator( self ):
-
from Products.PluggableAuthService.interfaces.plugins \
import IExtractionPlugin, IAuthenticationPlugin
@@ -673,8 +748,8 @@
plugins = zcuf._getOb( 'plugins' )
plugins.activatePlugin( IExtractionPlugin, 'login' )
- plugins.activatePlugin( IAuthenticationPlugin, 'borked' )
- plugins.activatePlugin( IAuthenticationPlugin, 'login' )
+ plugins.activatePlugin( IAuthenticationPlugin, 'login' ) # 1
+ plugins.activatePlugin( IAuthenticationPlugin, 'borked' ) # 2
request = self._makeRequest( form={ 'login' : 'foo'
, 'password' : 'bar' } )
@@ -686,6 +761,101 @@
self.assertEqual( len( user_ids ), 1 )
self.assertEqual( user_ids[0][0], 'foo' )
+ def test__extractUserIds_authenticate_emergency_user_with_broken_authenticator( self ):
+
+ from Products.PluggableAuthService.interfaces.plugins \
+ import IExtractionPlugin, IAuthenticationPlugin
+
+ from AccessControl.User import UnrestrictedUser
+
+ from Products.PluggableAuthService import PluggableAuthService
+
+ old_eu = PluggableAuthService.emergency_user
+
+ eu = UnrestrictedUser( 'foo', 'bar', ( 'manage', ), () )
+
+ PluggableAuthService.emergency_user = eu
+
+ try:
+ plugins = self._makePlugins()
+ zcuf = self._makeOne( plugins )
+
+ login = DummyPlugin()
+ directlyProvides( login, IExtractionPlugin )
+
+ # Make the first attempt at emergency user authentication fail
+ # (but not the extractor itself).
+ login.extractCredentials = lambda req: {'login': '', 'password': ''}
+
+ zcuf._setObject( 'login', login )
+
+ borked = DummyPlugin()
+ directlyProvides( borked, IAuthenticationPlugin )
+ borked.authenticateCredentials = lambda creds: creds['nonesuch']
+
+ zcuf._setObject( 'borked', borked )
+
+ plugins = zcuf._getOb( 'plugins' )
+
+ plugins.activatePlugin( IExtractionPlugin, 'login' )
+ plugins.activatePlugin( IAuthenticationPlugin, 'borked' )
+
+ request = self._makeRequest( form={ 'login' : eu.getUserName()
+ , 'password' : eu._getPassword() } )
+
+ user_ids = zcuf._extractUserIds( request=request
+ , plugins=zcuf.plugins
+ )
+
+ self.assertEqual( len( user_ids ), 1 )
+ self.assertEqual( user_ids[0][0], 'foo' )
+ finally:
+ PluggableAuthService.emergency_user = old_eu
+
+ def test__extractUserIds_emergency_user_always_wins( self ):
+
+ from Products.PluggableAuthService.interfaces.plugins \
+ import IExtractionPlugin, IAuthenticationPlugin
+
+ from AccessControl.User import UnrestrictedUser
+
+ from Products.PluggableAuthService import PluggableAuthService
+
+ old_eu = PluggableAuthService.emergency_user
+
+ eu = UnrestrictedUser( 'foo', 'bar', ( 'manage', ), () )
+
+ PluggableAuthService.emergency_user = eu
+
+ try:
+ plugins = self._makePlugins()
+ zcuf = self._makeOne( plugins )
+
+ login = DummyPlugin()
+ directlyProvides( login, IExtractionPlugin, IAuthenticationPlugin )
+ login.extractCredentials = lambda req: {'login': 'baz', 'password': ''}
+ login.authenticateCredentials = _authLogin
+
+ zcuf._setObject( 'login', login )
+
+ plugins = zcuf._getOb( 'plugins' )
+
+ plugins.activatePlugin( IExtractionPlugin, 'login' )
+ plugins.activatePlugin( IAuthenticationPlugin, 'login' )
+
+ request = self._makeRequest( form={ 'login' : eu.getUserName()
+ , 'password' : eu._getPassword() } )
+
+ # This should authenticate the emergency user and not 'baz'
+ user_ids = zcuf._extractUserIds( request=request
+ , plugins=zcuf.plugins
+ )
+
+ self.assertEqual( len( user_ids ), 1 )
+ self.assertEqual( user_ids[0][0], 'foo' )
+ finally:
+ PluggableAuthService.emergency_user = old_eu
+
def test__getObjectContext_no_steps( self ):
zcuf = self._makeOne()
Added: zope-pas/branches/upstream/current/tests/test_utils.py
===================================================================
--- zope-pas/branches/upstream/current/tests/test_utils.py 2007-02-09 13:59:28 UTC (rev 639)
+++ zope-pas/branches/upstream/current/tests/test_utils.py 2007-02-09 14:00:25 UTC (rev 640)
@@ -0,0 +1,111 @@
+##############################################################################
+#
+# Copyright (c) 2006 Zope Corporation and Contributors. All Rights
+# Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.1 (ZPL). A copy of the ZPL should accompany this
+# distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+
+import unittest
+
+from Products.PluggableAuthService.utils import createViewName
+from Products.PluggableAuthService.utils import createKeywords
+
+
+class UtilityTests(unittest.TestCase):
+
+ def test_createViewName(self):
+ self.assertEqual(createViewName('foo', 'bar'), 'foo-bar')
+
+ def test_createViewName_no_user_handle(self):
+ self.assertEqual(createViewName('foo', None), 'foo')
+
+ def test_createViewName_latin1_umlaut_in_method(self):
+ self.assertEqual(createViewName('f\366o'), 'f\366o')
+
+ def test_createViewName_utf8_umlaut_in_method(self):
+ self.assertEqual(createViewName('f\303\266o'), 'f\303\266o')
+
+ def test_createViewName_unicode_umlaut_in_method(self):
+ self.assertEqual(createViewName(u'f\366o'), 'f\303\266o')
+
+ def test_createViewName_latin1_umlaut_in_handle(self):
+ self.assertEqual(createViewName('foo', 'b\344r'), 'foo-b\344r')
+
+ def test_createViewName_utf8_umlaut_in_handle(self):
+ self.assertEqual(createViewName('foo', 'b\303\244r'), 'foo-b\303\244r')
+
+ def test_createViewName_unicode_umlaut_in_handle(self):
+ self.assertEqual(createViewName('foo', u'b\344r'), 'foo-b\303\244r')
+
+ def test_createKeywords(self):
+ _ITEMS = (('foo', 'bar'),)
+ hashed = _createHashedValue(_ITEMS)
+ self.assertEqual(createKeywords(foo='bar'),
+ {'keywords': hashed})
+
+ def test_createKeywords_multiple(self):
+ _ITEMS = (('foo', 'bar'), ('baz', 'peng'))
+ hashed = _createHashedValue(_ITEMS)
+ self.assertEqual(createKeywords(foo='bar', baz='peng'),
+ {'keywords': hashed})
+
+ def test_createKeywords_latin1_umlaut(self):
+ _ITEMS = (('foo', 'bar'), ('baz', 'M\344dchen'))
+ hashed = _createHashedValue(_ITEMS)
+ self.assertEqual(createKeywords(foo='bar', baz='M\344dchen'),
+ {'keywords': hashed})
+
+ def test_createKeywords_utf8_umlaut(self):
+ _ITEMS = (('foo', 'bar'), ('baz', 'M\303\244dchen'))
+ hashed = _createHashedValue(_ITEMS)
+ self.assertEqual(createKeywords(foo='bar', baz='M\303\244dchen'),
+ {'keywords': hashed})
+
+ def test_createKeywords_unicode_umlaut(self):
+ _ITEMS = (('foo', 'bar'), ('baz', u'M\344dchen'))
+ hashed = _createHashedValue(_ITEMS)
+ self.assertEqual(createKeywords(foo='bar', baz=u'M\344dchen'),
+ {'keywords': hashed})
+
+ def test_createKeywords_utf16_umlaut(self):
+ _ITEMS = (('foo', 'bar'), ('baz', u'M\344dchen'.encode('utf-16')))
+ hashed = _createHashedValue(_ITEMS)
+ self.assertEqual(createKeywords(foo='bar',
+ baz=u'M\344dchen'.encode('utf-16')),
+ {'keywords': hashed})
+
+ def test_createKeywords_unicode_chinese(self):
+ _ITEMS = (('foo', 'bar'), ('baz', u'\u03a4\u03b6'))
+ hashed = _createHashedValue(_ITEMS)
+ self.assertEqual(createKeywords(foo='bar', baz=u'\u03a4\u03b6'),
+ {'keywords': hashed})
+
+def _createHashedValue(items):
+ import sha
+ hasher = sha.new()
+ items = list(items)
+ items.sort()
+ for k, v in items:
+ if isinstance(k, unicode):
+ k = k.encode('utf-8')
+ hasher.update(k)
+ if isinstance(v, unicode):
+ v = v.encode('utf-8')
+ hasher.update(v)
+ return hasher.hexdigest()
+
+def test_suite():
+ return unittest.TestSuite((
+ unittest.makeSuite(UtilityTests),
+ ))
+
+if __name__ == '__main__':
+ unittest.main(defaultTest='test_suite')
Modified: zope-pas/branches/upstream/current/utils.py
===================================================================
--- zope-pas/branches/upstream/current/utils.py 2007-02-09 13:59:28 UTC (rev 639)
+++ zope-pas/branches/upstream/current/utils.py 2007-02-09 14:00:25 UTC (rev 640)
@@ -13,6 +13,7 @@
#
##############################################################################
import os
+import sha
import unittest
from types import TupleType, ListType
@@ -197,13 +198,38 @@
return suite
+
+def makestr(s):
+ """Converts 's' to a non-Unicode string"""
+ if isinstance(s, unicode):
+ s = s.encode('utf-8')
+ return str(s)
+
def createViewName(method_name, user_handle=None):
"""
Centralized place for creating the "View Name" that identifies
- a ZCacheable record in a PASRAMCacheManager
+ a ZCacheable record in a RAMCacheManager
"""
if not user_handle:
- return method_name
+ return makestr(method_name)
else:
- return '%s-%s' % (method_name, user_handle)
+ return '%s-%s' % (makestr(method_name), makestr(user_handle))
+def createKeywords(**kw):
+ """
+ Centralized place for creating the keywords that identify
+ a ZCacheable record in a RAMCacheManager.
+
+ Keywords are hashed so we don't accidentally expose sensitive
+ information.
+ """
+ keywords = sha.new()
+
+ items = kw.items()
+ items.sort()
+ for k, v in items:
+ keywords.update(makestr(k))
+ keywords.update(makestr(v))
+
+ return {'keywords': keywords.hexdigest()}
+
Modified: zope-pas/branches/upstream/current/version.txt
===================================================================
--- zope-pas/branches/upstream/current/version.txt 2007-02-09 13:59:28 UTC (rev 639)
+++ zope-pas/branches/upstream/current/version.txt 2007-02-09 14:00:25 UTC (rev 640)
@@ -1 +1 @@
-PluggableAuthService-1.4
+PluggableAuthService-1.4.1
More information about the pkg-zope-commits
mailing list