[Zope-CVS] CVS: Products/PluggableAuthService/Extensions -
upgrade.py:1.1
Zachery Bir
zbir at urbanape.com
Mon Aug 30 11:23:26 EDT 2004
Update of /cvs-repository/Products/PluggableAuthService/Extensions
In directory cvs.zope.org:/tmp/cvs-serv17892/Extensions
Added Files:
upgrade.py
Log Message:
Some new files/modules:
- CHANGES.txt - changelog
- Extensions/upgrade.py - an example External Method for replacing
the root acl_users folder with a PluggableAuthService instance.
- plugins/tests/helpers.py - some commonly used dummy classes
=== Added File Products/PluggableAuthService/Extensions/upgrade.py ===
##############################################################################
#
# Copyright (c) 2004 Zope Corporation. All Rights Reserved.
#
# This software is subject to the provisions of the Zope Visible Source
# License, Version 1.0 (ZVSL). A copy of the ZVSL 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.
#
##############################################################################
""" External method for upgrading existing AccessControl.User.UserFolder
NOTA BENE: Use at your own risk. This external method will replace a
stock User Folder (AccessControl.User.UserFolder) with a
PluggableAuthService consisting of the following:
- ZODBUserManager with a record for each existing User
(AccessControl.User.User)
- ZODBRoleManger with a record for each existing role present
in the __ac_roles__ attribute of the container (minus Anonymous
and Authenticated)
Each migrated user will be assigned the global roles they have in the
previous acl_users record.
$Id: upgrade.py,v 1.1 2004/08/30 15:23:25 urbanape Exp $
"""
from zLOG import LOG, INFO
def _write( response, tool, message, level=INFO ):
LOG( tool, level, message )
if response is not None:
response.write( message )
def _replaceUserFolder(self, RESPONSE=None):
"""replaces the old acl_users folder with a PluggableAuthService,
preserving users and passwords, if possible
"""
from Acquisition import aq_base
from Products.PluggableAuthService.PluggableAuthService \
import PluggableAuthService, _PLUGIN_TYPE_INFO
from Products.PluginRegistry.PluginRegistry import PluginRegistry
from Products.PluggableAuthService.plugins.ZODBUserManager \
import ZODBUserManager
from Products.PluggableAuthService.plugins.ZODBRoleManager \
import ZODBRoleManager
from Products.PluggableAuthService.interfaces.plugins \
import IAuthenticationPlugin, IUserEnumerationPlugin
from Products.PluggableAuthService.interfaces.plugins \
import IRolesPlugin, IRoleEnumerationPlugin, IRoleAssignerPlugin
if getattr( aq_base(self), '__allow_groups__', None ):
if self.__allow_groups__.__class__ is PluggableAuthService:
_write( RESPONSE
, 'replaceUserFolder'
, 'Already replaced this user folder\n' )
return
old_acl = self.__allow_groups__
new_acl = PluggableAuthService()
preg = PluginRegistry( _PLUGIN_TYPE_INFO )
preg._setId( 'plugins' )
new_acl._setObject( 'plugins', preg )
self._setObject( 'new_acl_users', new_acl )
new_acl = getattr( self, 'new_acl_users' )
user_folder = ZODBUserManager( 'users' )
new_acl._setObject( 'users', user_folder )
role_manager = ZODBRoleManager( 'roles' )
new_acl._setObject( 'roles', role_manager )
plugins = getattr( new_acl, 'plugins' )
plugins.activatePlugin( IAuthenticationPlugin, 'users' )
plugins.activatePlugin( IUserEnumerationPlugin, 'users' )
plugins.activatePlugin( IRolesPlugin, 'roles' )
plugins.activatePlugin( IRoleEnumerationPlugin, 'roles' )
plugins.activatePlugin( IRoleAssignerPlugin, 'roles' )
for user_name in old_acl.getUserNames():
old_user = old_acl.getUser( user_name )
_write( RESPONSE
, 'replaceRootUserFolder'
, 'Translating user %s\n' % user_name )
_migrate_user( new_acl.users, user_name, old_user._getPassword() )
new_user = new_acl.getUser( user_name )
for role_id in old_user.getRoles():
if role_id not in ['Authenticated', 'Anonymous']:
new_acl.roles.assignRoleToPrincipal( role_id,
new_user.getId() )
self._delObject( 'acl_users' )
self._setObject( 'acl_users', aq_base( new_acl ) )
self._delObject( 'new_acl_users' )
self.__allow_groups__ = aq_base( new_acl )
_write( RESPONSE
, 'replaceRootUserFolder'
, 'Replaced root acl_users with PluggableAuthService\n' )
get_transaction().commit()
def _migrate_user( new_user_folder, login, password ):
from AccessControl import AuthEncoding
if AuthEncoding.is_encrypted( password ):
new_user_folder._user_passwords[ login ] = password
new_user_folder._login_to_userid[ login ] = login
new_user_folder._userid_to_login[ login ] = login
else:
new_user_folder.addUser( login, login, password )
def _upgradeLocalRoleAssignments(self, RESPONSE=None):
""" upgrades the __ac_local_roles__ attributes on objects to account
for a move to using the PluggableAuthService.
"""
from Acquisition import aq_base
seen = {}
def descend(user_folder, obj):
path = obj.getPhysicalPath()
if path not in seen:
# get __ac_local_roles__, break it apart and refashion it
# with new spellings.
seen[path] = 1
if getattr( aq_base( obj ), '__ac_local_roles__', None ):
if not callable(obj.__ac_local_roles__):
new_map = {}
map = obj.__ac_local_roles__
for key in map.keys():
new_principals = user_folder.searchPrincipals(id=key)
if not new_principals:
_write(
RESPONSE
, 'upgradeLocalRoleAssignmentsFromRoot'
, ' Ignoring map for unknown principal %s\n'
% key )
new_map[key] = map[key]
continue
npid = new_principals[0]['id']
new_map[npid] = map[key]
_write( RESPONSE
, 'upgradeLocalRoleAssignmentsFromRoot'
, ' Translated %s to %s\n' % ( key, npid ) )
_write( RESPONSE
, 'upgradeLocalRoleAssignmentsFromRoot'
, ' Assigned roles %s to %s\n' % ( map[key]
, npid ) )
obj.__ac_local_roles__ = new_map
_write( RESPONSE
, 'upgradeLocalRoleAssignmentsFromRoot'
, ( 'Local Roles map changed for (%s)\n'
% '/'.join(path) ) )
if (len(seen) % 100 ) == 0:
get_transaction().commit()
_write( RESPONSE
, 'upgradeLocalRoleAssignmentsFromRoot'
, " -- committed at object # %d\n" % len( seen ) )
if getattr(aq_base(obj), 'isPrincipiaFolderish', 0):
for o in obj.objectValues():
descend(user_folder, o)
if getattr( self, '_upgraded_acl_users', None ):
_write( RESPONSE
, '_upgradeLocalRoleAssignments'
, 'Local role assignments have already been updated.\n' )
return
descend(self.acl_users, self)
get_transaction().commit()
# External Method to use
def replace_acl_users(self, RESPONSE=None):
_replaceUserFolder(self, RESPONSE)
_upgradeLocalRoleAssignments(self, RESPONSE)
self._upgraded_acl_users = 1
_write( RESPONSE
, 'replace_acl_users'
, 'Root acl_users has been replaced,'
' and local role assignments updated.\n' )
More information about the Zope-CVS
mailing list