+++ /dev/null
-# -*- coding: utf-8 -*-
-## GroupUserFolder
-## Copyright (C)2006 Ingeniweb
-
-## 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; see the file COPYING. If not, write to the
-## Free Software Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
-"""
-GroupUserFolder product
-"""
-__version__ = "$Revision: $"
-# $Source: $
-# $Id: GroupUserFolder.py 40118 2007-04-01 15:13:44Z alecm $
-__docformat__ = 'restructuredtext'
-
-
-# fakes a method from a DTML file
-from Globals import MessageDialog, DTMLFile
-
-from AccessControl import ClassSecurityInfo
-from AccessControl import Permissions
-from AccessControl import getSecurityManager
-from AccessControl import Unauthorized
-from Globals import InitializeClass
-from Acquisition import aq_base, aq_inner, aq_parent
-from Acquisition import Implicit
-from Globals import Persistent
-from AccessControl.Role import RoleManager
-from OFS.SimpleItem import Item
-from OFS.PropertyManager import PropertyManager
-import OFS
-from OFS import ObjectManager, SimpleItem
-from DateTime import DateTime
-from App import ImageFile
-from Products.PageTemplates import PageTemplateFile
-import AccessControl.Role, webdav.Collection
-import Products
-import os
-import string
-import sys
-import time
-import math
-import random
-from global_symbols import *
-import AccessControl.User
-import GRUFFolder
-import GRUFUser
-from Products.PageTemplates import PageTemplateFile
-import class_utility
-from Products.GroupUserFolder import postonly
-
-from interfaces.IUserFolder import IUserFolder
-
-## Developers notes
-##
-## The REQUEST.GRUF_PROBLEM variable is defined whenever GRUF encounters
-## a problem than can be showed in the management screens. It's always
-## logged as LOG_WARNING level anyway.
-
-_marker = []
-
-def unique(sequence, _list = 0):
- """Make a sequence a list of unique items"""
- uniquedict = {}
- for v in sequence:
- uniquedict[v] = 1
- if _list:
- return list(uniquedict.keys())
- return tuple(uniquedict.keys())
-
-
-def manage_addGroupUserFolder(self, dtself=None, REQUEST=None, **ignored):
- """ Factory method that creates a UserFolder"""
- f=GroupUserFolder()
- self=self.this()
- try: self._setObject('acl_users', f)
- except: return MessageDialog(
- title ='Item Exists',
- message='This object already contains a User Folder',
- action ='%s/manage_main' % REQUEST['URL1'])
- self.__allow_groups__=f
- self.acl_users._post_init()
-
- self.acl_users.Users.manage_addUserFolder()
- self.acl_users.Groups.manage_addUserFolder()
-
- if REQUEST is not None:
- REQUEST['RESPONSE'].redirect(self.absolute_url()+'/manage_main')
-
-
-
-
-class GroupUserFolder(OFS.ObjectManager.ObjectManager,
- AccessControl.User.BasicUserFolder,
- ):
- """
- GroupUserFolder => User folder with groups management
- """
-
- # #
- # ZOPE INFORMATION #
- # #
-
- meta_type='Group User Folder'
- id ='acl_users'
- title ='Group-aware User Folder'
-
- __implements__ = (IUserFolder, )
- def __creatable_by_emergency_user__(self): return 1
-
- isAnObjectManager = 1
- isPrincipiaFolderish = 1
- isAUserFolder = 1
-
-## _haveLDAPUF = 0
-
- security = ClassSecurityInfo()
-
- manage_options=(
- (
- {'label':'Overview', 'action':'manage_overview'},
- {'label':'Sources', 'action':'manage_GRUFSources'},
- {'label':'LDAP Wizard', 'action':'manage_wizard'},
- {'label':'Groups', 'action':'manage_groups'},
- {'label':'Users', 'action':'manage_users'},
- {'label':'Audit', 'action':'manage_audit'},
- ) + \
- OFS.ObjectManager.ObjectManager.manage_options + \
- RoleManager.manage_options + \
- Item.manage_options )
-
- manage_main = OFS.ObjectManager.ObjectManager.manage_main
-## manage_overview = DTMLFile('dtml/GRUF_overview', globals())
- manage_overview = PageTemplateFile.PageTemplateFile('dtml/GRUF_overview', globals())
- manage_audit = PageTemplateFile.PageTemplateFile('dtml/GRUF_audit', globals())
- manage_wizard = PageTemplateFile.PageTemplateFile('dtml/GRUF_wizard', globals())
- manage_groups = PageTemplateFile.PageTemplateFile('dtml/GRUF_groups', globals())
- manage_users = PageTemplateFile.PageTemplateFile('dtml/GRUF_users', globals())
- manage_newusers = PageTemplateFile.PageTemplateFile('dtml/GRUF_newusers', globals())
- manage_GRUFSources = PageTemplateFile.PageTemplateFile('dtml/GRUF_contents', globals())
- manage_user = PageTemplateFile.PageTemplateFile('dtml/GRUF_user', globals())
-
- __ac_permissions__=(
- ('Manage users',
- ('manage_users',
- 'user_names', 'setDomainAuthenticationMode',
- )
- ),
- )
-
-
- # Color constants, only useful within GRUF management screens
- user_color = "#006600"
- group_color = "#000099"
- role_color = "#660000"
-
- # User and group images
- img_user = ImageFile.ImageFile('www/GRUFUsers.gif', globals())
- img_group = ImageFile.ImageFile('www/GRUFGroups.gif', globals())
-
-
-
- # #
- # OFFICIAL INTERFACE #
- # #
-
- security.declarePublic("hasUsers")
- def hasUsers(self, ):
- """
- From Zope 2.7's User.py:
- This is not a formal API method: it is used only to provide
- a way for the quickstart page to determine if the default user
- folder contains any users to provide instructions on how to
- add a user for newbies. Using getUserNames or getUsers would have
- posed a denial of service risk.
- In GRUF, this method always return 1."""
- return 1
-
- security.declareProtected(Permissions.manage_users, "user_names")
- def user_names(self,):
- """
- user_names() => return user IDS and not user NAMES !!!
- Due to a Zope inconsistency, the Role.get_valid_userids return user names
- and not user ids - which is bad. As GRUF distinguishes names and ids, this
- will cause it to break, especially in the listLocalRoles form. So we change
- user_names() behaviour so that it will return ids and not names.
- """
- return self.getUserIds()
-
-
- security.declareProtected(Permissions.manage_users, "getUserNames")
- def getUserNames(self, __include_groups__ = 1, __include_users__ = 1, __groups_prefixed__ = 0):
- """
- Return a list of all possible user atom names in the system.
- Groups will be returned WITHOUT their prefix by this method.
- So, there might be a collision between a user name and a group name.
- [NOTA: This method is time-expensive !]
- """
- if __include_users__:
- LogCallStack(LOG_DEBUG, "This call can be VERY expensive!")
- names = []
- ldap_sources = []
-
- # Fetch users in user sources
- if __include_users__:
- for src in self.listUserSources():
- names.extend(src.getUserNames())
-
- # Append groups if possible
- if __include_groups__:
- # Regular groups
- if "acl_users" in self._getOb('Groups').objectIds():
- names.extend(self.Groups.listGroups(prefixed = __groups_prefixed__))
-
- # LDAP groups
- for ldapuf in ldap_sources:
- if ldapuf._local_groups:
- continue
- for g in ldapuf.getGroups(attr = LDAP_GROUP_RDN):
- if __groups_prefixed__:
- names.append("%s%s" % (GROUP_PREFIX, g))
- else:
- names.append(g)
- # Return a list of unique names
- return unique(names, _list = 1)
-
- security.declareProtected(Permissions.manage_users, "getUserIds")
- def getUserIds(self,):
- """
- Return a list of all possible user atom ids in the system.
- WARNING: Please see the id Vs. name consideration at the
- top of this document. So, groups will be returned
- WITH their prefix by this method
- [NOTA: This method is time-expensive !]
- """
- return self.getUserNames(__groups_prefixed__ = 1)
-
- security.declareProtected(Permissions.manage_users, "getUsers")
- def getUsers(self, __include_groups__ = 1, __include_users__ = 1):
- """Return a list of user and group objects.
- In case of some UF implementations, the returned object may only be a subset
- of all possible users.
- In other words, you CANNOT assert that len(getUsers()) equals len(getUserNames()).
- With cache-support UserFolders, such as LDAPUserFolder, the getUser() method will
- return only cached user objects instead of fetching all possible users.
- """
- Log(LOG_DEBUG, "getUsers")
- ret = []
- names_set = {}
-
- # avoid too many lookups for 'has_key' in loops
- isUserProcessed = names_set.has_key
-
- # Fetch groups first (then the user must be
- # prefixed by 'group_' prefix)
- if __include_groups__:
- # Fetch regular groups
- for u in self._getOb('Groups').acl_users.getUsers():
- if not u:
- continue # Ignore empty users
-
- name = u.getId()
- if isUserProcessed(name):
- continue # Prevent double users inclusion
-
- # Append group
- names_set[name] = True
- ret.append(
- GRUFUser.GRUFGroup(u, self, isGroup = 1, source_id = "Groups").__of__(self)
- )
-
- # Fetch users then
- if __include_users__:
- for src in self.listUserSources():
- for u in src.getUsers():
- if not u:
- continue # Ignore empty users
-
- name = u.getId()
- if isUserProcessed(name):
- continue # Prevent double users inclusion
-
- # Append user
- names_set[name] = True
- ret.append(
- GRUFUser.GRUFUser(u, self, source_id = src.getUserSourceId(), isGroup = 0).__of__(self)
- )
-
- return tuple(ret)
-
- security.declareProtected(Permissions.manage_users, "getUser")
- def getUser(self, name, __include_users__ = 1, __include_groups__ = 1, __force_group_id__ = 0):
- """
- Return the named user object or None.
- User have precedence over group.
- If name is None, getUser() will return None.
- """
- # Basic check
- if name is None:
- return None
-
- # Prevent infinite recursion when instanciating a GRUF
- # without having sub-acl_users set
- if not "acl_users" in self._getOb('Groups').objectIds():
- return None
-
- # Fetch groups first (then the user must be prefixed by 'group_' prefix)
- if __include_groups__ and name.startswith(GROUP_PREFIX):
- id = name[GROUP_PREFIX_LEN:]
-
- # Fetch regular groups
- u = self._getOb('Groups')._getGroup(id)
- if u:
- ret = GRUFUser.GRUFGroup(
- u, self, isGroup = 1, source_id = "Groups"
- ).__of__(self)
- return ret # XXX This violates precedence
-
- # Fetch users then
- if __include_users__:
- for src in self.listUserSources():
- u = src.getUser(name)
- if u:
- ret = GRUFUser.GRUFUser(u, self, source_id = src.getUserSourceId(), isGroup = 0).__of__(self)
- return ret
-
- # Then desperatly try to fetch groups (without beeing prefixed by 'group_' prefix)
- if __include_groups__ and (not __force_group_id__):
- u = self._getOb('Groups')._getGroup(name)
- if u:
- ret = GRUFUser.GRUFGroup(u, self, isGroup = 1, source_id = "Groups").__of__(self)
- return ret
-
- return None
-
-
- security.declareProtected(Permissions.manage_users, "getUserById")
- def getUserById(self, id, default=_marker):
- """Return the user atom corresponding to the given id. Can return groups.
- """
- ret = self.getUser(id, __force_group_id__ = 1)
- if not ret:
- if default is _marker:
- return None
- ret = default
- return ret
-
-
- security.declareProtected(Permissions.manage_users, "getUserByName")
- def getUserByName(self, name, default=_marker):
- """Same as getUser() but works with a name instead of an id.
- [NOTA: Theorically, the id is a handle, while the name is the actual login name.
- But difference between a user id and a user name is unsignificant in
- all current User Folder implementations... except for GROUPS.]
- """
- # Try to fetch a user first
- usr = self.getUser(name)
-
- # If not found, try to fetch a group by appending the prefix
- if not usr:
- name = "%s%s" % (GROUP_PREFIX, name)
- usr = self.getUserById(name, default)
-
- return usr
-
- security.declareProtected(Permissions.manage_users, "getPureUserNames")
- def getPureUserNames(self, ):
- """Fetch the list of actual users from GRUFUsers.
- """
- return self.getUserNames(__include_groups__ = 0)
-
-
- security.declareProtected(Permissions.manage_users, "getPureUserIds")
- def getPureUserIds(self,):
- """Same as getUserIds() but without groups
- """
- return self.getUserNames(__include_groups__ = 0)
-
- security.declareProtected(Permissions.manage_users, "getPureUsers")
- def getPureUsers(self):
- """Return a list of pure user objects.
- """
- return self.getUsers(__include_groups__ = 0)
-
- security.declareProtected(Permissions.manage_users, "getPureUser")
- def getPureUser(self, id, ):
- """Return the named user object or None"""
- # Performance tricks
- if not id:
- return None
-
- # Fetch it
- return self.getUser(id, __include_groups__ = 0)
-
-
- security.declareProtected(Permissions.manage_users, "getGroupNames")
- def getGroupNames(self, ):
- """Same as getUserNames() but without pure users.
- """
- return self.getUserNames(__include_users__ = 0, __groups_prefixed__ = 0)
-
- security.declareProtected(Permissions.manage_users, "getGroupIds")
- def getGroupIds(self, ):
- """Same as getUserNames() but without pure users.
- """
- return self.getUserNames(__include_users__ = 0, __groups_prefixed__ = 1)
-
- security.declareProtected(Permissions.manage_users, "getGroups")
- def getGroups(self):
- """Same as getUsers() but without pure users.
- """
- return self.getUsers(__include_users__ = 0)
-
- security.declareProtected(Permissions.manage_users, "getGroup")
- def getGroup(self, name, prefixed = 1):
- """Return the named user object or None"""
- # Performance tricks
- if not name:
- return None
-
- # Unprefix group name
- if not name.startswith(GROUP_PREFIX):
- name = "%s%s" % (GROUP_PREFIX, name, )
-
- # Fetch it
- return self.getUser(name, __include_users__ = 0)
-
- security.declareProtected(Permissions.manage_users, "getGroupById")
- def getGroupById(self, id, default = _marker):
- """Same as getUserById(id) but forces returning a group.
- """
- ret = self.getUser(id, __include_users__ = 0, __force_group_id__ = 1)
- if not ret:
- if default is _marker:
- return None
- ret = default
- return ret
-
- security.declareProtected(Permissions.manage_users, "getGroupByName")
- def getGroupByName(self, name, default = _marker):
- """Same as getUserByName(name) but forces returning a group.
- """
- ret = self.getUser(name, __include_users__ = 0, __force_group_id__ = 0)
- if not ret:
- if default is _marker:
- return None
- ret = default
- return ret
-
-
-
- # #
- # REGULAR MUTATORS #
- # #
-
- security.declareProtected(Permissions.manage_users, "userFolderAddUser")
- def userFolderAddUser(self, name, password, roles, domains, groups = (),
- REQUEST=None, **kw):
- """API method for creating a new user object. Note that not all
- user folder implementations support dynamic creation of user
- objects.
- """
- return self._doAddUser(name, password, roles, domains, groups, **kw)
- userFolderAddUser = postonly(userFolderAddUser)
-
- security.declareProtected(Permissions.manage_users, "userFolderEditUser")
- def userFolderEditUser(self, name, password, roles, domains, groups = None,
- REQUEST=None, **kw):
- """API method for changing user object attributes. Note that not
- all user folder implementations support changing of user object
- attributes.
- Arguments ARE required.
- """
- return self._doChangeUser(name, password, roles, domains, groups, **kw)
- userFolderEditUser = postonly(userFolderEditUser)
-
- security.declareProtected(Permissions.manage_users, "userFolderUpdateUser")
- def userFolderUpdateUser(self, name, password = None, roles = None,
- domains = None, groups = None, REQUEST=None, **kw):
- """API method for changing user object attributes. Note that not
- all user folder implementations support changing of user object
- attributes.
- Arguments are optional"""
- return self._updateUser(name, password, roles, domains, groups, **kw)
- userFolderUpdateUser = postonly(userFolderUpdateUser)
-
- security.declareProtected(Permissions.manage_users, "userFolderDelUsers")
- def userFolderDelUsers(self, names, REQUEST=None):
- """API method for deleting one or more user atom objects. Note that not
- all user folder implementations support deletion of user objects."""
- return self._doDelUsers(names)
- userFolderDelUsers = postonly(userFolderDelUsers)
-
- security.declareProtected(Permissions.manage_users, "userFolderAddGroup")
- def userFolderAddGroup(self, name, roles, groups = (), REQUEST=None, **kw):
- """API method for creating a new group.
- """
- while name.startswith(GROUP_PREFIX):
- name = name[GROUP_PREFIX_LEN:]
- return self._doAddGroup(name, roles, groups, **kw)
- userFolderAddGroup = postonly(userFolderAddGroup)
-
- security.declareProtected(Permissions.manage_users, "userFolderEditGroup")
- def userFolderEditGroup(self, name, roles, groups = None, REQUEST=None,
- **kw):
- """API method for changing group object attributes.
- """
- return self._doChangeGroup(name, roles = roles, groups = groups, **kw)
- userFolderEditGroup = postonly(userFolderEditGroup)
-
- security.declareProtected(Permissions.manage_users, "userFolderUpdateGroup")
- def userFolderUpdateGroup(self, name, roles = None, groups = None,
- REQUEST=None, **kw):
- """API method for changing group object attributes.
- """
- return self._updateGroup(name, roles = roles, groups = groups, **kw)
- userFolderUpdateGroup = postonly(userFolderUpdateGroup)
-
- security.declareProtected(Permissions.manage_users, "userFolderDelGroups")
- def userFolderDelGroups(self, names, REQUEST=None):
- """API method for deleting one or more group objects.
- Implem. note : All ids must be prefixed with 'group_',
- so this method ends up beeing only a filter of non-prefixed ids
- before calling userFolderDelUsers().
- """
- return self._doDelGroups(names)
- userFolderDelUsers = postonly(userFolderDelUsers)
-
-
-
- # #
- # SEARCH METHODS #
- # #
-
-
- security.declareProtected(Permissions.manage_users, "searchUsersByAttribute")
- def searchUsersByAttribute(self, attribute, search_term):
- """Return user ids whose 'attribute' match the specified search_term.
- If search_term is an empty string, behaviour depends on the underlying user folder:
- it may return all users, return only cached users (for LDAPUF) or return no users.
- This will return all users whose name contains search_term (whaterver its case).
- THIS METHOD MAY BE VERY EXPENSIVE ON USER FOLDER KINDS WHICH DO NOT PROVIDE A
- SEARCHING METHOD (ie. every UF kind except LDAPUF).
- 'attribute' can be 'id' or 'name' for all UF kinds, or anything else for LDAPUF.
- """
- ret = []
- for src in self.listUserSources():
- # Use source-specific search methods if available
- if hasattr(src.aq_base, "findUser"):
- # LDAPUF
- Log(LOG_DEBUG, "We use LDAPUF to find users")
- id_attr = src._uid_attr
- if attribute == 'name':
- attr = src._login_attr
- elif attribute == 'id':
- attr = src._uid_attr
- else:
- attr = attribute
- Log(LOG_DEBUG, "we use findUser", attr, search_term, )
- users = src.findUser(attr, search_term, exact_match = True)
- ret.extend(
- [ u[id_attr] for u in users ],
- )
- else:
- # Other types of user folder
- search_term = search_term.lower()
-
- # Find the proper method according to the attribute type
- if attribute == "name":
- method = "getName"
- elif attribute == "id":
- method = "getId"
- else:
- raise NotImplementedError, "Attribute searching is only supported for LDAPUserFolder by now."
-
- # Actually search
- src_id = src.getUserSourceId()
- for u in src.getUsers():
- if not u:
- continue
- u = GRUFUser.GRUFUser(u, self, source_id=src_id,
- isGroup=0).__of__(self)
- s = getattr(u, method)().lower()
- if string.find(s, search_term) != -1:
- ret.append(u.getId())
- Log(LOG_DEBUG, "We've found them:", ret)
- return ret
-
- security.declareProtected(Permissions.manage_users, "searchUsersByName")
- def searchUsersByName(self, search_term):
- """Return user ids whose name match the specified search_term.
- If search_term is an empty string, behaviour depends on the underlying user folder:
- it may return all users, return only cached users (for LDAPUF) or return no users.
- This will return all users whose name contains search_term (whaterver its case).
- THIS METHOD MAY BE VERY EXPENSIVE ON USER FOLDER KINDS WHICH DO NOT PROVIDE A
- SEARCHING METHOD (ie. every UF kind except LDAPUF)
- """
- return self.searchUsersByAttribute("name", search_term)
-
- security.declareProtected(Permissions.manage_users, "searchUsersById")
- def searchUsersById(self, search_term):
- """Return user ids whose id match the specified search_term.
- If search_term is an empty string, behaviour depends on the underlying user folder:
- it may return all users, return only cached users (for LDAPUF) or return no users.
- This will return all users whose name contains search_term (whaterver its case).
- THIS METHOD MAY BE VERY EXPENSIVE ON USER FOLDER KINDS WHICH DO NOT PROVIDE A
- SEARCHING METHOD (ie. every UF kind except LDAPUF)
- """
- return self.searchUsersByAttribute("id", search_term)
-
-
- security.declareProtected(Permissions.manage_users, "searchGroupsByAttribute")
- def searchGroupsByAttribute(self, attribute, search_term):
- """Return group ids whose 'attribute' match the specified search_term.
- If search_term is an empty string, behaviour depends on the underlying group folder:
- it may return all groups, return only cached groups (for LDAPUF) or return no groups.
- This will return all groups whose name contains search_term (whaterver its case).
- THIS METHOD MAY BE VERY EXPENSIVE ON GROUP FOLDER KINDS WHICH DO NOT PROVIDE A
- SEARCHING METHOD (ie. every UF kind except LDAPUF).
- 'attribute' can be 'id' or 'name' for all UF kinds, or anything else for LDAPUF.
- """
- ret = []
- src = self.Groups
-
- # Use source-specific search methods if available
- if hasattr(src.aq_base, "findGroup"):
- # LDAPUF
- id_attr = src._uid_attr
- if attribute == 'name':
- attr = src._login_attr
- elif attribute == 'id':
- attr = src._uid_attr
- else:
- attr = attribute
- groups = src.findGroup(attr, search_term)
- ret.extend(
- [ u[id_attr] for u in groups ],
- )
- else:
- # Other types of group folder
- search_term = search_term.lower()
-
- # Find the proper method according to the attribute type
- if attribute == "name":
- method = "getName"
- elif attribute == "id":
- method = "getId"
- else:
- raise NotImplementedError, "Attribute searching is only supported for LDAPGroupFolder by now."
-
- # Actually search
- for u in self.getGroups():
- s = getattr(u, method)().lower()
- if string.find(s, search_term) != -1:
- ret.append(u.getId())
- return ret
-
- security.declareProtected(Permissions.manage_users, "searchGroupsByName")
- def searchGroupsByName(self, search_term):
- """Return group ids whose name match the specified search_term.
- If search_term is an empty string, behaviour depends on the underlying group folder:
- it may return all groups, return only cached groups (for LDAPUF) or return no groups.
- This will return all groups whose name contains search_term (whaterver its case).
- THIS METHOD MAY BE VERY EXPENSIVE ON GROUP FOLDER KINDS WHICH DO NOT PROVIDE A
- SEARCHING METHOD (ie. every UF kind except LDAPUF)
- """
- return self.searchGroupsByAttribute("name", search_term)
-
- security.declareProtected(Permissions.manage_users, "searchGroupsById")
- def searchGroupsById(self, search_term):
- """Return group ids whose id match the specified search_term.
- If search_term is an empty string, behaviour depends on the underlying group folder:
- it may return all groups, return only cached groups (for LDAPUF) or return no groups.
- This will return all groups whose name contains search_term (whaterver its case).
- THIS METHOD MAY BE VERY EXPENSIVE ON GROUP FOLDER KINDS WHICH DO NOT PROVIDE A
- SEARCHING METHOD (ie. every UF kind except LDAPUF)
- """
- return self.searchGroupsByAttribute("id", search_term)
-
- # #
- # SECURITY MANAGEMENT METHODS #
- # #
-
- security.declareProtected(Permissions.manage_users, "setRolesOnUsers")
- def setRolesOnUsers(self, roles, userids, REQUEST = None):
- """Set a common set of roles for a bunch of user atoms.
- """
- for usr in userids:
- self.userSetRoles(usr, roles)
- setRolesOnUsers = postonly(setRolesOnUsers)
-
-## def setUsersOfRole(self, usernames, role):
-## """Sets the users of a role.
-## XXX THIS METHOD SEEMS TO BE SEAMLESS.
-## """
-## raise NotImplementedError, "Not implemented."
-
- security.declareProtected(Permissions.manage_users, "getUsersOfRole")
- def getUsersOfRole(self, role, object = None):
- """Gets the user (and group) ids having the specified role...
- ...on the specified Zope object if it's not None
- ...on their own information if the object is None.
- NOTA: THIS METHOD IS VERY EXPENSIVE.
- XXX PERFORMANCES HAVE TO BE IMPROVED
- """
- ret = []
- for id in self.getUserIds():
- if role in self.getRolesOfUser(id):
- ret.append(id)
- return tuple(ret)
-
- security.declarePublic("getRolesOfUser")
- def getRolesOfUser(self, userid):
- """Alias for user.getRoles()
- """
- return self.getUserById(userid).getRoles()
-
- security.declareProtected(Permissions.manage_users, "userFolderAddRole")
- def userFolderAddRole(self, role, REQUEST=None):
- """Add a new role. The role will be appended, in fact, in GRUF's surrounding folder.
- """
- if role in self.aq_parent.valid_roles():
- raise ValueError, "Role '%s' already exist" % (role, )
-
- return self.aq_parent._addRole(role)
- userFolderAddRole = postonly(userFolderAddRole)
-
- security.declareProtected(Permissions.manage_users, "userFolderDelRoles")
- def userFolderDelRoles(self, roles, REQUEST=None):
- """Delete roles.
- The removed roles will be removed from the UserFolder's users and groups as well,
- so this method can be very time consuming with a large number of users.
- """
- # Check that roles exist
- ud_roles = self.aq_parent.userdefined_roles()
- for r in roles:
- if not r in ud_roles:
- raise ValueError, "Role '%s' is not defined on acl_users' parent folder" % (r, )
-
- # Remove role on all users
- for r in roles:
- for u in self.getUsersOfRole(r, ):
- self.userRemoveRole(u, r, )
-
- # Actually remove role
- return self.aq_parent._delRoles(roles, None)
- userFolderDelRoles = postonly(userFolderDelRoles)
-
- security.declarePublic("userFolderGetRoles")
- def userFolderGetRoles(self, ):
- """
- userFolderGetRoles(self,) => tuple of strings
- List the roles defined at the top of GRUF's folder.
- This includes both user-defined roles and default roles.
- """
- return tuple(self.aq_parent.valid_roles())
-
-
- # Groups support
-
- security.declareProtected(Permissions.manage_users, "setMembers")
- def setMembers(self, groupid, userids, REQUEST=None):
- """Set the members of the group
- """
- self.getGroup(groupid).setMembers(userids)
- setMembers = postonly(setMembers)
-
- security.declareProtected(Permissions.manage_users, "addMember")
- def addMember(self, groupid, userid, REQUEST=None):
- """Add a member to a group
- """
- return self.getGroup(groupid).addMember(userid)
- addMember = postonly(addMember)
-
- security.declareProtected(Permissions.manage_users, "removeMember")
- def removeMember(self, groupid, userid, REQUEST=None):
- """Remove a member from a group.
- """
- return self.getGroup(groupid).removeMember(userid)
- removeMember = postonly(removeMember)
-
- security.declareProtected(Permissions.manage_users, "getMemberIds")
- def getMemberIds(self, groupid):
- """Return the list of member ids (groups and users) in this group
- """
- m = self.getGroup(groupid)
- if not m:
- raise ValueError, "Invalid group: '%s'" % groupid
- return self.getGroup(groupid).getMemberIds()
-
- security.declareProtected(Permissions.manage_users, "getUserMemberIds")
- def getUserMemberIds(self, groupid):
- """Return the list of member ids (groups and users) in this group
- """
- return self.getGroup(groupid).getUserMemberIds()
-
- security.declareProtected(Permissions.manage_users, "getGroupMemberIds")
- def getGroupMemberIds(self, groupid):
- """Return the list of member ids (groups and users) in this group
- XXX THIS MAY BE VERY EXPENSIVE !
- """
- return self.getGroup(groupid).getGroupMemberIds()
-
- security.declareProtected(Permissions.manage_users, "hasMember")
- def hasMember(self, groupid, id):
- """Return true if the specified atom id is in the group.
- This is the contrary of IUserAtom.isInGroup(groupid).
- THIS CAN BE VERY EXPENSIVE
- """
- return self.getGroup(groupid).hasMember(id)
-
-
- # User mutation
-
-## def setUserId(id, newId):
-## """Change id of a user atom.
-## """
-
-## def setUserName(id, newName):
-## """Change the name of a user atom.
-## """
-
- security.declareProtected(Permissions.manage_users, "userSetRoles")
- def userSetRoles(self, id, roles, REQUEST=None):
- """Change the roles of a user atom.
- """
- self._updateUser(id, roles = roles)
- userSetRoles = postonly(userSetRoles)
-
- security.declareProtected(Permissions.manage_users, "userAddRole")
- def userAddRole(self, id, role, REQUEST=None):
- """Append a role for a user atom
- """
- roles = list(self.getUser(id).getRoles())
- if not role in roles:
- roles.append(role)
- self._updateUser(id, roles = roles)
- userAddRole = postonly(userAddRole)
-
- security.declareProtected(Permissions.manage_users, "userRemoveRole")
- def userRemoveRole(self, id, role, REQUEST=None):
- """Remove the role of a user atom. Will NOT complain if role doesn't exist
- """
- roles = list(self.getRolesOfUser(id))
- if role in roles:
- roles.remove(role)
- self._updateUser(id, roles = roles)
- userRemoveRole = postonly(userRemoveRole)
-
- security.declareProtected(Permissions.manage_users, "userSetPassword")
- def userSetPassword(self, id, newPassword, REQUEST=None):
- """Set the password of a user
- """
- u = self.getPureUser(id)
- if not u:
- raise ValueError, "Invalid pure user id: '%s'" % (id,)
- self._updateUser(u.getId(), password = newPassword, )
- userSetPassword = postonly(userSetPassword)
-
- security.declareProtected(Permissions.manage_users, "userGetDomains")
- def userGetDomains(self, id):
- """get domains for a user
- """
- usr = self.getPureUser(id)
- return tuple(usr.getDomains())
-
- security.declareProtected(Permissions.manage_users, "userSetDomains")
- def userSetDomains(self, id, domains, REQUEST=None):
- """Set domains for a user
- """
- usr = self.getPureUser(id)
- self._updateUser(usr.getId(), domains = domains, )
- userSetDomains = postonly(userSetDomains)
-
- security.declareProtected(Permissions.manage_users, "userAddDomain")
- def userAddDomain(self, id, domain, REQUEST=None):
- """Append a domain to a user
- """
- usr = self.getPureUser(id)
- domains = list(usr.getDomains())
- if not domain in domains:
- roles.append(domain)
- self._updateUser(usr.getId(), domains = domains, )
- userAddDomain = postonly(userAddDomain)
-
- security.declareProtected(Permissions.manage_users, "userRemoveDomain")
- def userRemoveDomain(self, id, domain, REQUEST=None):
- """Remove a domain from a user
- """
- usr = self.getPureUser(id)
- domains = list(usr.getDomains())
- if not domain in domains:
- raise ValueError, "User '%s' doesn't have domain '%s'" % (id, domain, )
- while domain in domains:
- roles.remove(domain)
- self._updateUser(usr.getId(), domains = domains)
- userRemoveDomain = postonly(userRemoveDomain)
-
- security.declareProtected(Permissions.manage_users, "userSetGroups")
- def userSetGroups(self, id, groupnames, REQUEST=None):
- """Set the groups of a user
- """
- self._updateUser(id, groups = groupnames)
- userSetGroups = postonly(userSetGroups)
-
- security.declareProtected(Permissions.manage_users, "userAddGroup")
- def userAddGroup(self, id, groupname, REQUEST=None):
- """add a group to a user atom
- """
- groups = list(self.getUserById(id).getGroups())
- if not groupname in groups:
- groups.append(groupname)
- self._updateUser(id, groups = groups)
- userAddGroup = postonly(userAddGroup)
-
-
- security.declareProtected(Permissions.manage_users, "userRemoveGroup")
- def userRemoveGroup(self, id, groupname, REQUEST=None):
- """remove a group from a user atom.
- """
- groups = list(self.getUserById(id).getGroupNames())
- if groupname.startswith(GROUP_PREFIX):
- groupname = groupname[GROUP_PREFIX_LEN:]
- if groupname in groups:
- groups.remove(groupname)
- self._updateUser(id, groups = groups)
- userRemoveGroup = postonly(userRemoveGroup)
-
-
- # #
- # VARIOUS OPERATIONS #
- # #
-
- def __init__(self):
- """
- __init__(self) -> initialization method
- We define it to prevend calling ancestor's __init__ methods.
- """
- pass
-
-
- security.declarePrivate('_post_init')
- def _post_init(self):
- """
- _post_init(self) => meant to be called when the
- object is in the Zope tree
- """
- uf = GRUFFolder.GRUFUsers()
- gf = GRUFFolder.GRUFGroups()
- self._setObject('Users', uf)
- self._setObject('Groups', gf)
- self.id = "acl_users"
-
- def manage_beforeDelete(self, item, container):
- """
- Special overloading for __allow_groups__ attribute
- """
- if item is self:
- try:
- del container.__allow_groups__
- except:
- pass
-
- def manage_afterAdd(self, item, container):
- """Same
- """
- if item is self:
- container.__allow_groups__ = aq_base(self)
-
- # #
- # VARIOUS UTILITIES #
- # #
- # These methods shouldn't be used directly for most applications, #
- # but they might be useful for some special processing. #
- # #
-
- security.declarePublic('getGroupPrefix')
- def getGroupPrefix(self):
- """ group prefix """
- return GROUP_PREFIX
-
- security.declarePrivate('getGRUFPhysicalRoot')
- def getGRUFPhysicalRoot(self,):
- # $$$ trick meant to be used within
- # fake_getPhysicalRoot (see __init__)
- return self.getPhysicalRoot()
-
- security.declareProtected(Permissions.view, 'getGRUFId')
- def getGRUFId(self,):
- """
- Alias to self.getId()
- """
- return self.getId()
-
- security.declareProtected(Permissions.manage_users, "getUnwrappedUser")
- def getUnwrappedUser(self, name):
- """
- getUnwrappedUser(self, name) => user object or None
-
- This method is used to get a User object directly from the User's
- folder acl_users, without wrapping it with group information.
-
- This is useful for UserFolders that define additional User classes,
- when you want to call specific methods on these user objects.
-
- For example, LDAPUserFolder defines a 'getProperty' method that's
- not inherited from the standard User object. You can, then, use
- the getUnwrappedUser() to get the matching user and call this
- method.
- """
- src_id = self.getUser(name).getUserSourceId()
- return self.getUserSource(src_id).getUser(name)
-
- security.declareProtected(Permissions.manage_users, "getUnwrappedGroup")
- def getUnwrappedGroup(self, name):
- """
- getUnwrappedGroup(self, name) => user object or None
-
- Same as getUnwrappedUser but for groups.
- """
- return self.Groups.acl_users.getUser(name)
-
- # #
- # AUTHENTICATION INTERFACE #
- # #
-
- security.declarePrivate("authenticate")
- def authenticate(self, name, password, request):
- """
- Pass the request along to the underlying user-related UserFolder
- object
- THIS METHOD RETURNS A USER OBJECT OR NONE, as specified in the code
- in AccessControl/User.py.
- We also check for inituser in there.
- """
- # Emergency user checking stuff
- emergency = self._emergency_user
- if emergency and name == emergency.getUserName():
- if emergency.authenticate(password, request):
- return emergency
- else:
- return None
-
- # Usual GRUF authentication
- for src in self.listUserSources():
- # XXX We can imagine putting a try/except here to "ignore"
- # UF errors such as SQL or LDAP shutdown
- u = src.authenticate(name, password, request)
- if u:
- return GRUFUser.GRUFUser(u, self, isGroup = 0, source_id = src.getUserSourceId()).__of__(self)
-
- # No acl_users in the Users folder or no user authenticated
- # => we refuse authentication
- return None
-
-
-
-
- # #
- # GRUF'S GUTS :-) #
- # #
-
- security.declarePrivate("_doAddUser")
- def _doAddUser(self, name, password, roles, domains, groups = (), **kw):
- """
- Create a new user. This should be implemented by subclasses to
- do the actual adding of a user. The 'password' will be the
- original input password, unencrypted. The implementation of this
- method is responsible for performing any needed encryption.
- """
- prefix = GROUP_PREFIX
-
- # Prepare groups
- roles = list(roles)
- gruf_groups = self.getGroupIds()
- for group in groups:
- if not group.startswith(prefix):
- group = "%s%s" % (prefix, group, )
- if not group in gruf_groups:
- raise ValueError, "Invalid group: '%s'" % (group, )
- roles.append(group)
-
- # Reset the users overview batch
- self._v_batch_users = []
-
- # Really add users
- return self.getDefaultUserSource()._doAddUser(
- name,
- password,
- roles,
- domains,
- **kw)
-
- security.declarePrivate("_doChangeUser")
- def _doChangeUser(self, name, password, roles, domains, groups = None, **kw):
- """
- Modify an existing user. This should be implemented by subclasses
- to make the actual changes to a user. The 'password' will be the
- original input password, unencrypted. The implementation of this
- method is responsible for performing any needed encryption.
-
- A None password should not change it (well, we hope so)
- """
- # Get actual user name and id
- usr = self.getUser(name)
- if usr is None:
- raise ValueError, "Invalid user: '%s'" % (name,)
- id = usr.getRealId()
-
- # Don't lose existing groups
- if groups is None:
- groups = usr.getGroups()
-
- roles = list(roles)
- groups = list(groups)
-
- # Change groups affectation
- cur_groups = self.getGroups()
- given_roles = tuple(usr.getRoles()) + tuple(roles)
- for group in groups:
- if not group.startswith(GROUP_PREFIX, ):
- group = "%s%s" % (GROUP_PREFIX, group, )
- if not group in cur_groups and not group in given_roles:
- roles.append(group)
-
- # Reset the users overview batch
- self._v_batch_users = []
-
- # Change the user itself
- src = usr.getUserSourceId()
- Log(LOG_NOTICE, name, "Source:", src)
- ret = self.getUserSource(src)._doChangeUser(
- id, password, roles, domains, **kw)
-
- # Invalidate user cache if necessary
- usr.clearCachedGroupsAndRoles()
- authenticated = getSecurityManager().getUser()
- if id == authenticated.getId() and hasattr(authenticated, 'clearCachedGroupsAndRoles'):
- authenticated.clearCachedGroupsAndRoles(self.getUserSource(src).getUser(id))
-
- return ret
-
- security.declarePrivate("_updateUser")
- def _updateUser(self, id, password = None, roles = None, domains = None, groups = None):
- """
- _updateUser(self, id, password = None, roles = None, domains = None, groups = None)
-
- This one should work for users AND groups.
-
- Front-end to _doChangeUser, but with a better default value support.
- We guarantee that None values will let the underlying UF keep the original ones.
- This is not true for the password: some buggy UF implementation may not
- handle None password correctly :-(
- """
- # Get the former values if necessary. Username must be valid !
- usr = self.getUser(id)
- if roles is None:
- # Remove invalid roles and group names
- roles = usr._original_roles
- roles = filter(lambda x: not x.startswith(GROUP_PREFIX), roles)
- roles = filter(lambda x: x not in ('Anonymous', 'Authenticated', 'Shared', ''), roles)
- else:
- # Check if roles are valid
- roles = filter(lambda x: x not in ('Anonymous', 'Authenticated', 'Shared', ''), roles)
- vr = self.userFolderGetRoles()
- for r in roles:
- if not r in vr:
- raise ValueError, "Invalid or inexistant role: '%s'." % (r, )
- if domains is None:
- domains = usr._original_domains
- if groups is None:
- groups = usr.getGroups(no_recurse = 1)
- else:
- # Check if given groups are valid
- glist = self.getGroupNames()
- glist.extend(map(lambda x: "%s%s" % (GROUP_PREFIX, x), glist))
- for g in groups:
- if not g in glist:
- raise ValueError, "Invalid group: '%s'" % (g, )
-
- # Reset the users overview batch
- self._v_batch_users = []
-
- # Change the user
- return self._doChangeUser(id, password, roles, domains, groups)
-
- security.declarePrivate("_doDelUsers")
- def _doDelUsers(self, names):
- """
- Delete one or more users. This should be implemented by subclasses
- to do the actual deleting of users.
- This won't delete groups !
- """
- # Collect information about user sources
- sources = {}
- for name in names:
- usr = self.getUser(name, __include_groups__ = 0)
- if not usr:
- continue # Ignore invalid user names
- src = usr.getUserSourceId()
- if not sources.has_key(src):
- sources[src] = []
- sources[src].append(name)
- for src, names in sources.items():
- self.getUserSource(src)._doDelUsers(names)
-
- # Reset the users overview batch
- self._v_batch_users = []
-
-
- # #
- # Groups interface #
- # #
-
- security.declarePrivate("_doAddGroup")
- def _doAddGroup(self, name, roles, groups = (), **kw):
- """
- Create a new group. Password will be randomly created, and domain will be None.
- Supports nested groups.
- """
- # Prepare initial data
- domains = ()
- password = ""
- if roles is None:
- roles = []
- if groups is None:
- groups = []
-
- for x in range(0, 10): # Password will be 10 chars long
- password = "%s%s" % (password, random.choice(string.lowercase), )
-
- # Compute roles
- roles = list(roles)
- prefix = GROUP_PREFIX
- gruf_groups = self.getGroupIds()
- for group in groups:
- if not group.startswith(prefix):
- group = "%s%s" % (prefix, group, )
- if group == "%s%s" % (prefix, name, ):
- raise ValueError, "Infinite recursion for group '%s'." % (group, )
- if not group in gruf_groups:
- raise ValueError, "Invalid group: '%s' (defined groups are %s)" % (group, gruf_groups)
- roles.append(group)
-
- # Reset the users overview batch
- self._v_batch_users = []
-
- # Actual creation
- return self.Groups.acl_users._doAddUser(
- name, password, roles, domains, **kw
- )
-
- security.declarePrivate("_doChangeGroup")
- def _doChangeGroup(self, name, roles, groups = None, **kw):
- """Modify an existing group."""
- # Remove prefix if given
- if name.startswith(self.getGroupPrefix()):
- name = name[GROUP_PREFIX_LEN:]
-
- # Check if group exists
- grp = self.getGroup(name, prefixed = 0)
- if grp is None:
- raise ValueError, "Invalid group: '%s'" % (name,)
-
- # Don't lose existing groups
- if groups is None:
- groups = grp.getGroups()
-
- roles = list(roles or [])
- groups = list(groups or [])
-
- # Change groups affectation
- cur_groups = self.getGroups()
- given_roles = tuple(grp.getRoles()) + tuple(roles)
- for group in groups:
- if not group.startswith(GROUP_PREFIX, ):
- group = "%s%s" % (GROUP_PREFIX, group, )
- if group == "%s%s" % (GROUP_PREFIX, grp.id):
- raise ValueError, "Cannot affect group '%s' to itself!" % (name, ) # Prevent direct inclusion of self
- new_grp = self.getGroup(group)
- if not new_grp:
- raise ValueError, "Invalid or inexistant group: '%s'" % (group, )
- if "%s%s" % (GROUP_PREFIX, grp.id) in new_grp.getGroups():
- raise ValueError, "Cannot affect %s to group '%s' as it would lead to circular references." % (group, name, ) # Prevent indirect inclusion of self
- if not group in cur_groups and not group in given_roles:
- roles.append(group)
-
- # Reset the users overview batch
- self._v_batch_users = []
-
- # Perform the change
- domains = ""
- password = ""
- for x in range(0, 10): # Password will be 10 chars long
- password = "%s%s" % (password, random.choice(string.lowercase), )
- return self.Groups.acl_users._doChangeUser(name, password,
- roles, domains, **kw)
-
- security.declarePrivate("_updateGroup")
- def _updateGroup(self, name, roles = None, groups = None):
- """
- _updateGroup(self, name, roles = None, groups = None)
-
- Front-end to _doChangeUser, but with a better default value support.
- We guarantee that None values will let the underlying UF keep the original ones.
- This is not true for the password: some buggy UF implementation may not
- handle None password correctly but we do not care for Groups.
-
- group name can be prefixed or not
- """
- # Remove prefix if given
- if name.startswith(self.getGroupPrefix()):
- name = name[GROUP_PREFIX_LEN:]
-
- # Get the former values if necessary. Username must be valid !
- usr = self.getGroup(name, prefixed = 0)
- if roles is None:
- # Remove invalid roles and group names
- roles = usr._original_roles
- roles = filter(lambda x: not x.startswith(GROUP_PREFIX), roles)
- roles = filter(lambda x: x not in ('Anonymous', 'Authenticated', 'Shared'), roles)
- if groups is None:
- groups = usr.getGroups(no_recurse = 1)
-
- # Reset the users overview batch
- self._v_batch_users = []
-
- # Change the user
- return self._doChangeGroup(name, roles, groups)
-
-
- security.declarePrivate("_doDelGroup")
- def _doDelGroup(self, name):
- """Delete one user."""
- # Remove prefix if given
- if name.startswith(self.getGroupPrefix()):
- name = name[GROUP_PREFIX_LEN:]
-
- # Reset the users overview batch
- self._v_batch_users = []
-
- # Delete it
- return self.Groups.acl_users._doDelUsers([name])
-
- security.declarePrivate("_doDelGroups")
- def _doDelGroups(self, names):
- """Delete one or more users."""
- for group in names:
- if not self.getGroupByName(group, None):
- continue # Ignore invalid groups
- self._doDelGroup(group)
-
-
-
-
- # #
- # Pretty Management form methods #
- # #
-
-
- security.declarePublic('getGRUFVersion')
- def getGRUFVersion(self,):
- """
- getGRUFVersion(self,) => Return human-readable GRUF version as a string.
- """
- rev_date = "$Date: 2007-04-01 17:13:44 +0200 (dim, 01 avr 2007) $"[7:-2]
- return "%s / Revised %s" % (version__, rev_date)
-
-
- reset_entry = "__None__" # Special entry used for reset
-
- security.declareProtected(Permissions.manage_users, "changeUser")
- def changeUser(self, user, groups = [], roles = [], REQUEST = {}, ):
- """
- changeUser(self, user, groups = [], roles = [], REQUEST = {}, ) => used in ZMI
- """
- obj = self.getUser(user)
- if obj.isGroup():
- self._updateGroup(name = user, groups = groups, roles = roles, )
- else:
- self._updateUser(id = user, groups = groups, roles = roles, )
-
-
- if REQUEST.has_key('RESPONSE'):
- return REQUEST.RESPONSE.redirect(self.absolute_url() + "/" + obj.getId() + "/manage_workspace?FORCE_USER=1")
- changeUser = postonly(changeUser)
-
- security.declareProtected(Permissions.manage_users, "deleteUser")
- def deleteUser(self, user, REQUEST = {}, ):
- """
- deleteUser(self, user, REQUEST = {}, ) => used in ZMI
- """
- pass
- deleteUser = postonly(deleteUser)
-
- security.declareProtected(Permissions.manage_users, "changeOrCreateUsers")
- def changeOrCreateUsers(self, users = [], groups = [], roles = [], new_users = [], default_password = '', REQUEST = {}, ):
- """
- changeOrCreateUsers => affect roles & groups to users and/or create new users
-
- All parameters are strings or lists (NOT tuples !).
- NO CHECKING IS DONE. This is an utility method, it's not part of the official API.
- """
- # Manage roles / groups deletion
- del_roles = 0
- del_groups = 0
- if self.reset_entry in roles:
- roles.remove(self.reset_entry)
- del_roles = 1
- if self.reset_entry in groups:
- groups.remove(self.reset_entry)
- del_groups = 1
- if not roles and not del_roles:
- roles = None # None instead of [] to avoid deletion
- add_roles = []
- else:
- add_roles = roles
- if not groups and not del_groups:
- groups = None
- add_groups = []
- else:
- add_groups = groups
-
- # Passwords management
- passwords_list = []
-
- # Create brand new users
- for new in new_users:
- # Strip name
- name = string.strip(new)
- if not name:
- continue
-
- # Avoid erasing former users
- if name in map(lambda x: x.getId(), self.getUsers()):
- continue
-
- # Use default password or generate a random one
- if default_password:
- password = default_password
- else:
- password = ""
- for x in range(0, 8): # Password will be 8 chars long
- password = "%s%s" % (password, random.choice("ABCDEFGHJKMNPQRSTUVWXYZabcdefghijkmnpqrstuvwxyz23456789"), )
- self._doAddUser(name, password, add_roles, (), add_groups, )
-
- # Store the newly created password
- passwords_list.append({'name':name, 'password':password})
-
- # Update existing users
- for user in users:
- self._updateUser(id = user, groups = groups, roles = roles, )
-
- # Web request
- if REQUEST.has_key('RESPONSE'):
- # Redirect if no users have been created
- if not passwords_list:
- return REQUEST.RESPONSE.redirect(self.absolute_url() + "/manage_users")
-
- # Show passwords form
- else:
- REQUEST.set('USER_PASSWORDS', passwords_list)
- return self.manage_newusers(None, self)
-
- # Simply return the list of created passwords
- return passwords_list
- changeOrCreateUsers = postonly(changeOrCreateUsers)
-
- security.declareProtected(Permissions.manage_users, "deleteUsers")
- def deleteUsers(self, users = [], REQUEST = {}):
- """
- deleteUsers => explicit
-
- All parameters are strings. NO CHECKING IS DONE. This is an utility method !
- """
- # Delete them
- self._doDelUsers(users, )
-
- # Redirect
- if REQUEST.has_key('RESPONSE'):
- return REQUEST.RESPONSE.redirect(self.absolute_url() + "/manage_users")
- deleteUsers = postonly(deleteUsers)
-
- security.declareProtected(Permissions.manage_users, "changeOrCreateGroups")
- def changeOrCreateGroups(self, groups = [], roles = [], nested_groups = [], new_groups = [], REQUEST = {}, ):
- """
- changeOrCreateGroups => affect roles to groups and/or create new groups
-
- All parameters are strings. NO CHECKING IS DONE. This is an utility method !
- """
- # Manage roles / groups deletion
- del_roles = 0
- del_groups = 0
- if self.reset_entry in roles:
- roles.remove(self.reset_entry)
- del_roles = 1
- if self.reset_entry in nested_groups:
- nested_groups.remove(self.reset_entry)
- del_groups = 1
- if not roles and not del_roles:
- roles = None # None instead of [] to avoid deletion
- add_roles = []
- else:
- add_roles = roles
- if not nested_groups and not del_groups:
- nested_groups = None
- add_groups = []
- else:
- add_groups = nested_groups
-
- # Create brand new groups
- for new in new_groups:
- name = string.strip(new)
- if not name:
- continue
- self._doAddGroup(name, roles, groups = add_groups)
-
- # Update existing groups
- for group in groups:
- self._updateGroup(group, roles = roles, groups = nested_groups)
-
- # Redirect
- if REQUEST.has_key('RESPONSE'):
- return REQUEST.RESPONSE.redirect(self.absolute_url() + "/manage_groups")
- changeOrCreateGroups = postonly(changeOrCreateGroups)
-
- security.declareProtected(Permissions.manage_users, "deleteGroups")
- def deleteGroups(self, groups = [], REQUEST = {}):
- """
- deleteGroups => explicit
-
- All parameters are strings. NO CHECKING IS DONE. This is an utility method !
- """
- # Delete groups
- for group in groups:
- self._doDelGroup(group, )
-
- # Redirect
- if REQUEST.has_key('RESPONSE'):
- return REQUEST.RESPONSE.redirect(self.absolute_url() + "/manage_groups")
- deleteGroups = postonly(deleteGroups)
-
- # #
- # Local Roles Acquisition Blocking #
- # Those two methods perform their own security check. #
- # #
-
- security.declarePublic("acquireLocalRoles")
- def acquireLocalRoles(self, folder, status, REQUEST=None):
- """
- Enable or disable local role acquisition on the specified folder.
- If status is true, it will enable, else it will disable.
- Note that the user _must_ have the change_permissions permission on the
- folder to allow changes on it.
- If you want to use this code from a product, please use _acquireLocalRoles()
- instead: this private method won't check security on the destination folder.
- It's usually a bad idea to use _acquireLocalRoles() directly in your product,
- but, well, after all, you do what you want ! :^)
- """
- # Perform security check on destination folder
- if not getSecurityManager().checkPermission(Permissions.change_permissions, folder):
- raise Unauthorized(name = "acquireLocalRoles")
-
- return self._acquireLocalRoles(folder, status)
- acquireLocalRoles = postonly(acquireLocalRoles)
-
- def _acquireLocalRoles(self, folder, status):
- """Same as _acquireLocalRoles() but won't perform security check on the folder.
- """
- # Set the variable (or unset it if it's defined)
- if not status:
- folder.__ac_local_roles_block__ = 1
- else:
- if getattr(folder, '__ac_local_roles_block__', None):
- folder.__ac_local_roles_block__ = None
-
-
- security.declarePublic("isLocalRoleAcquired")
- def isLocalRoleAcquired(self, folder):
- """Return true if the specified folder allows local role acquisition.
- """
- if getattr(folder, '__ac_local_roles_block__', None):
- return 0
- return 1
-
-
- # #
- # Security audit and info methods #
- # #
-
-
- # This method normally has NOT to be public ! It is because of a CMF inconsistancy.
- # folder_localrole_form is accessible to users who have the manage_properties permissions
- # (according to portal_types/Folder/Actions information). This is silly !
- # folder_localrole_form should be, in CMF, accessible only to those who have the
- # manage_users permissions instead of manage_properties permissions.
- # This is yet another one CMF bug we have to care about.
- # To deal with that in Plone2.1, we check for a particular permission on the destination
- # object _inside_ the method.
- security.declarePublic("getLocalRolesForDisplay")
- def getLocalRolesForDisplay(self, object):
- """This is used for plone's local roles display
- This method returns a tuple (massagedUsername, roles, userType, actualUserName).
- This method is protected by the 'Manage properties' permission. We may
- change that if it's too permissive..."""
- # Perform security check on destination object
- if not getSecurityManager().checkPermission(Permissions.manage_properties, object):
- raise Unauthorized(name = "getLocalRolesForDisplay")
-
- return self._getLocalRolesForDisplay(object)
-
- def _getLocalRolesForDisplay(self, object):
- """This is used for plone's local roles display
- This method returns a tuple (massagedUsername, roles, userType, actualUserName)"""
- result = []
- local_roles = object.get_local_roles()
- prefix = self.getGroupPrefix()
- for one_user in local_roles:
- massagedUsername = username = one_user[0]
- roles = one_user[1]
- userType = 'user'
- if prefix:
- if self.getGroupById(username) is not None:
- massagedUsername = username[len(prefix):]
- userType = 'group'
- else:
- userType = 'unknown'
- result.append((massagedUsername, roles, userType, username))
- return tuple(result)
-
-
- security.declarePublic("getAllLocalRoles")
- def getAllLocalRoles(self, object):
- """getAllLocalRoles(self, object): return a dictionnary {useratom_id: roles} of local
- roles defined AND herited at a certain point. This will handle lr-blocking
- as well.
- """
- # Perform security check on destination object
- if not getSecurityManager().checkPermission(Permissions.change_permissions, object):
- raise Unauthorized(name = "getAllLocalRoles")
-
- return self._getAllLocalRoles(object)
-
-
- def _getAllLocalRoles(self, object):
- """getAllLocalRoles(self, object): return a dictionnary {useratom_id: roles} of local
- roles defined AND herited at a certain point. This will handle lr-blocking
- as well.
- """
- # Modified from AccessControl.User.getRolesInContext().
- merged = {}
- object = getattr(object, 'aq_inner', object)
- while 1:
- if hasattr(object, '__ac_local_roles__'):
- dict = object.__ac_local_roles__ or {}
- if callable(dict): dict = dict()
- for k, v in dict.items():
- if not merged.has_key(k):
- merged[k] = {}
- for role in v:
- merged[k][role] = 1
- if not self.isLocalRoleAcquired(object):
- break
- if hasattr(object, 'aq_parent'):
- object=object.aq_parent
- object=getattr(object, 'aq_inner', object)
- continue
- if hasattr(object, 'im_self'):
- object=object.im_self
- object=getattr(object, 'aq_inner', object)
- continue
- break
- for key, value in merged.items():
- merged[key] = value.keys()
- return merged
-
-
-
- # Plone-specific security matrix computing method.
- security.declarePublic("getPloneSecurityMatrix")
- def getPloneSecurityMatrix(self, object):
- """getPloneSecurityMatrix(self, object): return a list of dicts of the current object
- and all its parents. The list is sorted with portal object first.
- Each dict has the following structure:
- {
- depth: (0 for portal root, 1 for 1st-level folders and so on),
- id:
- title:
- icon:
- absolute_url:
- security_permission: true if current user can change security on this object
- state: (workflow state)
- acquired_local_roles: 0 if local role blocking is enabled for this folder
- roles: {
- 'role1': {
- 'all_local_roles': [r1, r2, r3, ] (all defined local roles, including parent ones)
- 'defined_local_roles': [r3, ] (local-defined only local roles)
- 'permissions': ['Access contents information', 'Modify portal content', ] (only a subset)
- 'same_permissions': true if same permissions as the parent
- 'same_all_local_roles': true if all_local_roles is the same as the parent
- 'same_defined_local_roles': true if defined_local_roles is the same as the parent
- },
- 'role2': {...},
- },
- }
- """
- # Perform security check on destination object
- if not getSecurityManager().checkPermission(Permissions.access_contents_information, object):
- raise Unauthorized(name = "getPloneSecurityMatrix")
-
- # Basic inits
- mt = self.portal_membership
-
- # Fetch all possible roles in the portal
- all_roles = ['Anonymous'] + mt.getPortalRoles()
-
- # Fetch parent folders list until the portal
- all_objects = []
- cur_object = object
- while 1:
- if not getSecurityManager().checkPermission(Permissions.access_contents_information, cur_object):
- raise Unauthorized(name = "getPloneSecurityMatrix")
- all_objects.append(cur_object)
- if cur_object.meta_type == "Plone Site":
- break
- cur_object = object.aq_parent
- all_objects.reverse()
-
- # Scan those folders to get all the required information about them
- ret = []
- previous = None
- count = 0
- for obj in all_objects:
- # Basic information
- current = {
- "depth": count,
- "id": obj.getId(),
- "title": obj.Title(),
- "icon": obj.getIcon(),
- "absolute_url": obj.absolute_url(),
- "security_permission": getSecurityManager().checkPermission(Permissions.change_permissions, obj),
- "acquired_local_roles": self.isLocalRoleAcquired(obj),
- "roles": {},
- "state": "XXX TODO XXX", # XXX TODO
- }
- count += 1
-
- # Workflow state
- # XXX TODO
-
- # Roles
- all_local_roles = {}
- local_roles = self._getAllLocalRoles(obj)
- for user, roles in self._getAllLocalRoles(obj).items():
- for role in roles:
- if not all_local_roles.has_key(role):
- all_local_roles[role] = {}
- all_local_roles[role][user] = 1
- defined_local_roles = {}
- if hasattr(obj.aq_base, 'get_local_roles'):
- for user, roles in obj.get_local_roles():
- for role in roles:
- if not defined_local_roles.has_key(role):
- defined_local_roles[role] = {}
- defined_local_roles[role][user] = 1
-
- for role in all_roles:
- all = all_local_roles.get(role, {}).keys()
- defined = defined_local_roles.get(role, {}).keys()
- all.sort()
- defined.sort()
- same_all_local_roles = 0
- same_defined_local_roles = 0
- if previous:
- if previous['roles'][role]['all_local_roles'] == all:
- same_all_local_roles = 1
- if previous['roles'][role]['defined_local_roles'] == defined:
- same_defined_local_roles = 1
-
- current['roles'][role] = {
- "all_local_roles": all,
- "defined_local_roles": defined,
- "same_all_local_roles": same_all_local_roles,
- "same_defined_local_roles": same_defined_local_roles,
- "permissions": [], # XXX TODO
- }
-
- ret.append(current)
- previous = current
-
- return ret
-
-
- security.declareProtected(Permissions.manage_users, "computeSecuritySettings")
- def computeSecuritySettings(self, folders, actors, permissions, cache = {}):
- """
- computeSecuritySettings(self, folders, actors, permissions, cache = {}) => return a structure that is suitable for security audit Page Template.
-
- - folders is the structure returned by getSiteTree()
- - actors is the structure returned by listUsersAndRoles()
- - permissions is ((id: permission), (id: permission), ...)
- - cache is passed along requests to make computing faster
- """
- # Scan folders and actors to get the relevant information
- usr_cache = {}
- for id, depth, path in folders:
- folder = self.unrestrictedTraverse(path)
- for kind, actor, display, handle, html in actors:
- if kind in ("user", "group"):
- # Init structure
- if not cache.has_key(path):
- cache[path] = {(kind, actor): {}}
- elif not cache[path].has_key((kind, actor)):
- cache[path][(kind, actor)] = {}
- else:
- cache[path][(kind, actor)] = {}
-
- # Split kind into groups and get individual role information
- perm_keys = []
- usr = usr_cache.get(actor)
- if not usr:
- usr = self.getUser(actor)
- usr_cache[actor] = usr
- roles = usr.getRolesInContext(folder,)
- for role in roles:
- for perm_key in self.computeSetting(path, folder, role, permissions, cache).keys():
- cache[path][(kind, actor)][perm_key] = 1
-
- else:
- # Get role information
- self.computeSetting(path, folder, actor, permissions, cache)
-
- # Return the computed cache
- return cache
-
-
- security.declareProtected(Permissions.manage_users, "computeSetting")
- def computeSetting(self, path, folder, actor, permissions, cache):
- """
- computeSetting(......) => used by computeSecuritySettings to populate the cache for ROLES
- """
- # Avoid doing things twice
- kind = "role"
- if cache.get(path, {}).get((kind, actor), None) is not None:
- return cache[path][(kind, actor)]
-
- # Initilize cache structure
- if not cache.has_key(path):
- cache[path] = {(kind, actor): {}}
- elif not cache[path].has_key((kind, actor)):
- cache[path][(kind, actor)] = {}
-
- # Analyze permission settings
- ps = folder.permission_settings()
- for perm_key, permission in permissions:
- # Check acquisition of permission setting.
- can = 0
- acquired = 0
- for p in ps:
- if p['name'] == permission:
- acquired = not not p['acquire']
-
- # If acquired, call the parent recursively
- if acquired:
- parent = folder.aq_parent.getPhysicalPath()
- perms = self.computeSetting(parent, self.unrestrictedTraverse(parent), actor, permissions, cache)
- can = perms.get(perm_key, None)
-
- # Else, check permission here
- else:
- for p in folder.rolesOfPermission(permission):
- if p['name'] == "Anonymous":
- # If anonymous is allowed, then everyone is allowed
- if p['selected']:
- can = 1
- break
- if p['name'] == actor:
- if p['selected']:
- can = 1
- break
-
- # Extend the data structure according to 'can' setting
- if can:
- cache[path][(kind, actor)][perm_key] = 1
-
- return cache[path][(kind, actor)]
-
-
- security.declarePrivate('_getNextHandle')
- def _getNextHandle(self, index):
- """
- _getNextHandle(self, index) => utility function to
- get an unique handle for each legend item.
- """
- return "%02d" % index
-
-
- security.declareProtected(Permissions.manage_users, "listUsersAndRoles")
- def listUsersAndRoles(self,):
- """
- listUsersAndRoles(self,) => list of tuples
-
- This method is used by the Security Audit page.
- XXX HAS TO BE OPTIMIZED
- """
- request = self.REQUEST
- display_roles = request.get('display_roles', 0)
- display_groups = request.get('display_groups', 0)
- display_users = request.get('display_users', 0)
-
- role_index = 0
- user_index = 0
- group_index = 0
- ret = []
-
- # Collect roles
- if display_roles:
- for r in self.aq_parent.valid_roles():
- handle = "R%02d" % role_index
- role_index += 1
- ret.append(('role', r, r, handle, r))
-
- # Collect users
- if display_users:
- for u in map(lambda x: x.getId(), self.getPureUsers()):
- obj = self.getUser(u)
- html = obj.asHTML()
- handle = "U%02d" % user_index
- user_index += 1
- ret.append(('user', u, u, handle, html))
-
- if display_groups:
- for u in self.getGroupNames():
- obj = self.getUser(u)
- handle = "G%02d" % group_index
- html = obj.asHTML()
- group_index += 1
- ret.append(('group', u, obj.getUserNameWithoutGroupPrefix(), handle, html))
-
- # Return list
- return ret
-
- security.declareProtected(Permissions.manage_users, "getSiteTree")
- def getSiteTree(self, obj=None, depth=0):
- """
- getSiteTree(self, obj=None, depth=0) => special structure
-
- This is used by the security audit page
- """
- ret = []
- if not obj:
- if depth==0:
- obj = self.aq_parent
- else:
- return ret
-
- ret.append([obj.getId(), depth, string.join(obj.getPhysicalPath(), '/')])
- for sub in obj.objectValues():
- try:
- # Ignore user folders
- if sub.getId() in ('acl_users', ):
- continue
-
- # Ignore portal_* stuff
- if sub.getId()[:len('portal_')] == 'portal_':
- continue
-
- if sub.isPrincipiaFolderish:
- ret.extend(self.getSiteTree(sub, depth + 1))
-
- except:
- # We ignore exceptions
- pass
-
- return ret
-
- security.declareProtected(Permissions.manage_users, "listAuditPermissions")
- def listAuditPermissions(self,):
- """
- listAuditPermissions(self,) => return a list of eligible permissions
- """
- ps = self.permission_settings()
- return map(lambda p: p['name'], ps)
-
- security.declareProtected(Permissions.manage_users, "getDefaultPermissions")
- def getDefaultPermissions(self,):
- """
- getDefaultPermissions(self,) => return default R & W permissions for security audit.
- """
- # If there's a Plone site in the above folder, use plonish permissions
- hasPlone = 0
- p = self.aq_parent
- if p.meta_type == "CMF Site":
- hasPlone = 1
- else:
- for obj in p.objectValues():
- if obj.meta_type == "CMF Site":
- hasPlone = 1
- break
-
- if hasPlone:
- return {'R': 'View',
- 'W': 'Modify portal content',
- }
- else:
- return {'R': 'View',
- 'W': 'Change Images and Files',
- }
-
-
- # #
- # Users/Groups tree view #
- # (ZMI only) #
- # #
-
-
- security.declarePrivate('getTreeInfo')
- def getTreeInfo(self, usr, dict = {}):
- "utility method"
- # Prevend infinite recursions
- name = usr.getUserName()
- if dict.has_key(name):
- return
- dict[name] = {}
-
- # Properties
- noprefix = usr.getUserNameWithoutGroupPrefix()
- is_group = usr.isGroup()
- if usr.isGroup():
- icon = string.join(self.getPhysicalPath(), '/') + '/img_group'
-## icon = self.absolute_url() + '/img_group'
- else:
- icon = ' img_user'
-## icon = self.absolute_url() + '/img_user'
-
- # Subobjects
- belongs_to = []
- for grp in usr.getGroups(no_recurse = 1):
- belongs_to.append(grp)
- self.getTreeInfo(self.getGroup(grp))
-
- # Append (and return) structure
- dict[name] = {
- "name": noprefix,
- "is_group": is_group,
- "icon": icon,
- "belongs_to": belongs_to,
- }
- return dict
-
-
- security.declarePrivate("tpValues")
- def tpValues(self):
- # Avoid returning HUUUUUUGE lists
- # Use the cache at first
- if self._v_no_tree and self._v_cache_no_tree > time.time():
- return [] # Do not use the tree
-
- # XXX - I DISABLE THE TREE BY NOW (Pb. with icon URL)
- return []
-
- # Then, use a simple computation to determine opportunity to use the tree or not
- ngroups = len(self.getGroupNames())
- if ngroups > MAX_TREE_USERS_AND_GROUPS:
- self._v_no_tree = 1
- self._v_cache_no_tree = time.time() + TREE_CACHE_TIME
- return []
- nusers = len(self.getUsers())
- if ngroups + nusers > MAX_TREE_USERS_AND_GROUPS:
- meth_list = self.getGroups
- else:
- meth_list = self.getUsers
- self._v_no_tree = 0
-
- # Get top-level user and groups list
- tree_dict = {}
- top_level_names = []
- top_level = []
- for usr in meth_list():
- self.getTreeInfo(usr, tree_dict)
- if not usr.getGroups(no_recurse = 1):
- top_level_names.append(usr.getUserName())
- for id in top_level_names:
- top_level.append(treeWrapper(id, tree_dict))
-
- # Return this top-level list
- top_level.sort(lambda x, y: cmp(x.sortId(), y.sortId()))
- return top_level
-
-
- def tpId(self,):
- return self.getId()
-
-
- # #
- # Direct traversal to user or group info #
- # #
-
- def manage_workspace(self, REQUEST):
- """
- manage_workspace(self, REQUEST) => Overrided to allow direct user or group traversal
- via the left tree view.
- """
- path = string.split(REQUEST.PATH_INFO, '/')[:-1]
- userid = path[-1]
-
- # Use individual usr/grp management screen (only if name is passed along the mgt URL)
- if userid != "acl_users":
- usr = self.getUserById(userid)
- if usr:
- REQUEST.set('username', userid)
- REQUEST.set('MANAGE_TABS_NO_BANNER', '1') # Prevent use of the manage banner
- return self.restrictedTraverse('manage_user')()
-
- # Default management screen
- return self.restrictedTraverse('manage_overview')()
-
-
- # Tree caching information
- _v_no_tree = 0
- _v_cache_no_tree = 0
- _v_cache_tree = (0, [])
-
-
- def __bobo_traverse__(self, request, name):
- """
- Looks for the name of a user or a group.
- This applies only if users list is not huge.
- """
- # Check if it's an attribute
- if hasattr(self.aq_base, name, ):
- return getattr(self, name)
-
- # It's not an attribute, maybe it's a user/group
- # (this feature is used for the tree)
- if name.startswith('_'):
- pass # Do not fetch users
- elif name.startswith('manage_'):
- pass # Do not fetch users
- elif name in INVALID_USER_NAMES:
- pass # Do not fetch users
- else:
- # Only try to get users is fetch_user is true.
- # This is only for performance reasons.
- # The following code block represent what we want to minimize
- if self._v_cache_tree[0] < time.time():
- un = map(lambda x: x.getId(), self.getUsers()) # This is the cost we want to avoid
- self._v_cache_tree = (time.time() + TREE_CACHE_TIME, un, )
- else:
- un = self._v_cache_tree[1]
-
- # Get the user if we can
- if name in un:
- self._v_no_tree = 0
- return self
-
- # Force getting the user if we must
- if request.get("FORCE_USER"):
- self._v_no_tree = 0
- return self
-
- # This will raise if it's not possible to acquire 'name'
- return getattr(self, name, )
-
-
-
- # #
- # USERS / GROUPS BATCHING (ZMI SCREENS) #
- # #
-
- _v_batch_users = []
-
- security.declareProtected(Permissions.view_management_screens, "listUsersBatches")
- def listUsersBatches(self,):
- """
- listUsersBatches(self,) => return a list of (start, end) tuples.
- Return None if batching is not necessary
- """
- # Time-consuming stuff !
- un = map(lambda x: x.getId(), self.getPureUsers())
- if len(un) <= MAX_USERS_PER_PAGE:
- return None
- un.sort()
-
- # Split this list into small groups if necessary
- ret = []
- idx = 0
- l_un = len(un)
- nbatches = int(math.ceil(l_un / float(MAX_USERS_PER_PAGE)))
- for idx in range(0, nbatches):
- first = idx * MAX_USERS_PER_PAGE
- last = first + MAX_USERS_PER_PAGE - 1
- if last >= l_un:
- last = l_un - 1
- # Append a tuple (not dict) to avoid too much memory consumption
- ret.append((first, last, un[first], un[last]))
-
- # Cache & return it
- self._v_batch_users = un
- return ret
-
- security.declareProtected(Permissions.view_management_screens, "listUsersBatchTable")
- def listUsersBatchTable(self,):
- """
- listUsersBatchTable(self,) => Same a mgt screens but divided into sublists to
- present them into 5 columns.
- XXX have to merge this w/getUsersBatch to make it in one single pass
- """
- # Iterate
- ret = []
- idx = 0
- current = []
- for rec in (self.listUsersBatches() or []):
- if not idx % 5:
- if current:
- ret.append(current)
- current = []
- current.append(rec)
- idx += 1
-
- if current:
- ret.append(current)
-
- return ret
-
- security.declareProtected(Permissions.view_management_screens, "getUsersBatch")
- def getUsersBatch(self, start):
- """
- getUsersBatch(self, start) => user list
- """
- # Rebuild the list if necessary
- if not self._v_batch_users:
- un = map(lambda x: x.getId(), self.getPureUsers())
- self._v_batch_users = un
-
- # Return the batch
- end = start + MAX_USERS_PER_PAGE
- ids = self._v_batch_users[start:end]
- ret = []
- for id in ids:
- usr = self.getUser(id)
- if usr: # Prevent adding invalid users
- ret.append(usr)
- return ret
-
-
- # #
- # Multiple sources management #
- # #
-
- # Arrows
- img_up_arrow = ImageFile.ImageFile('www/up_arrow.gif', globals())
- img_down_arrow = ImageFile.ImageFile('www/down_arrow.gif', globals())
- img_up_arrow_grey = ImageFile.ImageFile('www/up_arrow_grey.gif', globals())
- img_down_arrow_grey = ImageFile.ImageFile('www/down_arrow_grey.gif', globals())
-
- security.declareProtected(Permissions.manage_users, "toggleSource")
- def toggleSource(self, src_id, REQUEST = {}):
- """
- toggleSource(self, src_id, REQUEST = {}) => toggle enabled/disabled source
- """
- # Find the source
- ids = self.objectIds('GRUFUsers')
- if not src_id in ids:
- raise ValueError, "Invalid source: '%s' (%s)" % (src_id, ids)
- src = getattr(self, src_id)
- if src.enabled:
- src.disableSource()
- else:
- src.enableSource()
-
- # Redirect where we want to
- if REQUEST.has_key('RESPONSE'):
- return REQUEST.RESPONSE.redirect(self.absolute_url() + '/manage_GRUFSources')
-
-
- security.declareProtected(Permissions.manage_users, "listUserSources")
- def listUserSources(self, ):
- """
- listUserSources(self, ) => Return a list of userfolder objects
- Only return VALID (ie containing an acl_users) user sources if all is None
- XXX HAS TO BE OPTIMIZED VERY MUCH!
- We add a check in debug mode to ensure that invalid sources won't be added
- to the list.
- This method return only _enabled_ user sources.
- """
- ret = []
- dret = {}
- if DEBUG_MODE:
- for src in self.objectValues(['GRUFUsers']):
- if not src.enabled:
- continue
- if 'acl_users' in src.objectIds():
- if getattr(aq_base(src.acl_users), 'authenticate', None): # Additional check in debug mode
- dret[src.id] = src.acl_users # we cannot use restrictedTraverse here because
- # of infinite recursion issues.
- else:
- for src in self.objectValues(['GRUFUsers']):
- if not src.enabled:
- continue
- if not 'acl_users' in src.objectIds():
- continue
- dret[src.id] = src.acl_users
- ret = dret.items()
- ret.sort()
- return [ src[1] for src in ret ]
-
- security.declareProtected(Permissions.manage_users, "listUserSourceFolders")
- def listUserSourceFolders(self, ):
- """
- listUserSources(self, ) => Return a list of GRUFUsers objects
- """
- ret = []
- for src in self.objectValues(['GRUFUsers']):
- ret.append(src)
- ret.sort(lambda x,y: cmp(x.id, y.id))
- return ret
-
- security.declarePrivate("getUserSource")
- def getUserSource(self, id):
- """
- getUserSource(self, id) => GRUFUsers.acl_users object.
- Raises if no acl_users available
- """
- return getattr(self, id).acl_users
-
- security.declarePrivate("getUserSourceFolder")
- def getUserSourceFolder(self, id):
- """
- getUserSourceFolder(self, id) => GRUFUsers object
- """
- return getattr(self, id)
-
- security.declareProtected(Permissions.manage_users, "addUserSource")
- def addUserSource(self, factory_uri, REQUEST = {}, *args, **kw):
- """
- addUserSource(self, factory_uri, REQUEST = {}, *args, **kw) => redirect
- Adds the specified user folder
- """
- # Get the initial Users id
- ids = self.objectIds('GRUFUsers')
- if ids:
- ids.sort()
- if ids == ['Users',]:
- last = 0
- else:
- last = int(ids[-1][-2:])
- next_id = "Users%02d" % (last + 1, )
- else:
- next_id = "Users"
-
- # Add the GRUFFolder object
- uf = GRUFFolder.GRUFUsers(id = next_id)
- self._setObject(next_id, uf)
-
-## # If we use ldap, tag it
-## if string.find(factory_uri.lower(), "ldap") > -1:
-## self._haveLDAPUF += 1
-
- # Add its underlying UserFolder
- # If we're called TTW, uses a redirect else tries to call the UF factory directly
- if REQUEST.has_key('RESPONSE'):
- return REQUEST.RESPONSE.redirect("%s/%s/%s" % (self.absolute_url(), next_id, factory_uri))
- return getattr(self, next_id).unrestrictedTraverse(factory_uri)(*args, **kw)
- addUserSource = postonly(addUserSource)
-
- security.declareProtected(Permissions.manage_users, "deleteUserSource")
- def deleteUserSource(self, id = None, REQUEST = {}):
- """
- deleteUserSource(self, id = None, REQUEST = {}) => Delete the specified user source
- """
- # Check the source id
- if type(id) != type('s'):
- raise ValueError, "You must choose a valid source to delete and confirm it."
-
- # Delete it
- self.manage_delObjects([id,])
- if REQUEST.has_key('RESPONSE'):
- return REQUEST.RESPONSE.redirect(self.absolute_url() + '/manage_GRUFSources')
- deleteUserSource = postonly(deleteUserSource)
-
- security.declareProtected(Permissions.manage_users, "getDefaultUserSource")
- def getDefaultUserSource(self,):
- """
- getDefaultUserSource(self,) => acl_users object
- Return default user source for user writing.
- XXX By now, the FIRST source is the default one. This may change in the future.
- """
- lst = self.listUserSources()
- if not lst:
- raise RuntimeError, "No valid User Source to add users in."
- return lst[0]
-
-
- security.declareProtected(Permissions.manage_users, "listAvailableUserSources")
- def listAvailableUserSources(self, filter_permissions = 1, filter_classes = 1):
- """
- listAvailableUserSources(self, filter_permissions = 1, filter_classes = 1) => tuples (name, factory_uri)
- List UserFolder replacement candidates.
-
- - if filter_classes is true, return only ones which have a base UserFolder class
- - if filter_permissions, return only types the user has rights to add
- """
- ret = []
-
- # Fetch candidate types
- user = getSecurityManager().getUser()
- meta_types = []
- if callable(self.all_meta_types):
- all=self.all_meta_types()
- else:
- all=self.all_meta_types
- for meta_type in all:
- if filter_permissions and meta_type.has_key('permission'):
- if user.has_permission(meta_type['permission'],self):
- meta_types.append(meta_type)
- else:
- meta_types.append(meta_type)
-
- # Keep only, if needed, BasicUserFolder-derived classes
- for t in meta_types:
- if t['name'] == self.meta_type:
- continue # Do not keep GRUF ! ;-)
-
- if filter_classes:
- try:
- if t.get('instance', None) and t['instance'].isAUserFolder:
- ret.append((t['name'], t['action']))
- continue
- if t.get('instance', None) and class_utility.isBaseClass(AccessControl.User.BasicUserFolder, t['instance']):
- ret.append((t['name'], t['action']))
- continue
- except AttributeError:
- pass # We ignore 'invalid' instances (ie. that wouldn't define a __base__ attribute)
- else:
- ret.append((t['name'], t['action']))
-
- return tuple(ret)
-
- security.declareProtected(Permissions.manage_users, "moveUserSourceUp")
- def moveUserSourceUp(self, id, REQUEST = {}):
- """
- moveUserSourceUp(self, id, REQUEST = {}) => used in management screens
- try to get ids as consistant as possible
- """
- # List and sort sources and preliminary checks
- ids = self.objectIds('GRUFUsers')
- ids.sort()
- if not ids or not id in ids:
- raise ValueError, "Invalid User Source: '%s'" % (id,)
-
- # Find indexes to swap
- src_index = ids.index(id)
- if src_index == 0:
- raise ValueError, "Cannot move '%s' User Source up." % (id, )
- dest_index = src_index - 1
-
- # Find numbers to swap, fix them if they have more than 1 as offset
- if ids[dest_index] == 'Users':
- dest_num = 0
- else:
- dest_num = int(ids[dest_index][-2:])
- src_num = dest_num + 1
-
- # Get ids
- src_id = id
- if dest_num == 0:
- dest_id = "Users"
- else:
- dest_id = "Users%02d" % (dest_num,)
- tmp_id = "%s_" % (dest_id, )
-
- # Perform the swap
- self._renameUserSource(src_id, tmp_id)
- self._renameUserSource(dest_id, src_id)
- self._renameUserSource(tmp_id, dest_id)
-
- # Return back to the forms
- if REQUEST.has_key('RESPONSE'):
- return REQUEST.RESPONSE.redirect(self.absolute_url() + '/manage_GRUFSources')
- moveUserSourceUp = postonly(moveUserSourceUp)
-
- security.declareProtected(Permissions.manage_users, "moveUserSourceDown")
- def moveUserSourceDown(self, id, REQUEST = {}):
- """
- moveUserSourceDown(self, id, REQUEST = {}) => used in management screens
- try to get ids as consistant as possible
- """
- # List and sort sources and preliminary checks
- ids = self.objectIds('GRUFUsers')
- ids.sort()
- if not ids or not id in ids:
- raise ValueError, "Invalid User Source: '%s'" % (id,)
-
- # Find indexes to swap
- src_index = ids.index(id)
- if src_index == len(ids) - 1:
- raise ValueError, "Cannot move '%s' User Source up." % (id, )
- dest_index = src_index + 1
-
- # Find numbers to swap, fix them if they have more than 1 as offset
- if id == 'Users':
- dest_num = 1
- else:
- dest_num = int(ids[dest_index][-2:])
- src_num = dest_num - 1
-
- # Get ids
- src_id = id
- if dest_num == 0:
- dest_id = "Users"
- else:
- dest_id = "Users%02d" % (dest_num,)
- tmp_id = "%s_" % (dest_id, )
-
- # Perform the swap
- self._renameUserSource(src_id, tmp_id)
- self._renameUserSource(dest_id, src_id)
- self._renameUserSource(tmp_id, dest_id)
-
- # Return back to the forms
- if REQUEST.has_key('RESPONSE'):
- return REQUEST.RESPONSE.redirect(self.absolute_url() + '/manage_GRUFSources')
- moveUserSourceDown = postonly(moveUserSourceDown)
-
-
- security.declarePrivate('_renameUserSource')
- def _renameUserSource(self, id, new_id, ):
- """
- Rename a particular sub-object.
- Taken fro CopySupport.manage_renameObject() code, modified to disable verifications.
- """
- try: self._checkId(new_id)
- except: raise CopyError, MessageDialog(
- title='Invalid Id',
- message=sys.exc_info()[1],
- action ='manage_main')
- ob=self._getOb(id)
-## if not ob.cb_isMoveable():
-## raise "Copy Error", eNotSupported % id
-## self._verifyObjectPaste(ob) # This is what we disable
- try: ob._notifyOfCopyTo(self, op=1)
- except: raise CopyError, MessageDialog(
- title='Rename Error',
- message=sys.exc_info()[1],
- action ='manage_main')
- self._delObject(id)
- ob = aq_base(ob)
- ob._setId(new_id)
-
- # Note - because a rename always keeps the same context, we
- # can just leave the ownership info unchanged.
- self._setObject(new_id, ob, set_owner=0)
-
-
- security.declareProtected(Permissions.manage_users, "replaceUserSource")
- def replaceUserSource(self, id = None, new_factory = None, REQUEST = {}, *args, **kw):
- """
- replaceUserSource(self, id = None, new_factory = None, REQUEST = {}, *args, **kw) => perform user source replacement
-
- If new_factory is None, find it inside REQUEST (useful for ZMI screens)
- """
- # Check the source id
- if type(id) != type('s'):
- raise ValueError, "You must choose a valid source to replace and confirm it."
-
- # Retreive factory if not explicitly passed
- if not new_factory:
- for record in REQUEST.get("source_rec", []):
- if record['id'] == id:
- new_factory = record['new_factory']
- break
- if not new_factory:
- raise ValueError, "You must select a new User Folder type."
-
- # Delete the former one
- us = getattr(self, id)
- if "acl_users" in us.objectIds():
- us.manage_delObjects(['acl_users'])
-
- ## If we use ldap, tag it
- #if string.find(new_factory.lower(), "ldap") > -1:
- # self._haveLDAPUF += 1
-
- # Re-create the underlying UserFolder
- # If we're called TTW, uses a redirect else tries to call the UF factory directly
- if REQUEST.has_key('RESPONSE'):
- return REQUEST.RESPONSE.redirect("%s/%s/%s" % (self.absolute_url(), id, new_factory))
- return us.unrestrictedTraverse(new_factory)(*args, **kw) # XXX minor security pb ?
- replaceUserSource = postonly(replaceUserSource)
-
-
- security.declareProtected(Permissions.manage_users, "hasLDAPUserFolderSource")
- def hasLDAPUserFolderSource(self, ):
- """
- hasLDAPUserFolderSource(self,) => boolean
- Return true if a LUF source is instanciated.
- """
- for src in self.listUserSources():
- if src.meta_type == "LDAPUserFolder":
- return 1
- return None
-
-
- security.declareProtected(Permissions.manage_users, "updateLDAPUserFolderMapping")
- def updateLDAPUserFolderMapping(self, REQUEST = None):
- """
- updateLDAPUserFolderMapping(self, REQUEST = None) => None
-
- Update the first LUF source in the process so that LDAP-group-to-Zope-role mapping
- is done.
- This is done by calling the appropriate method in LUF and affecting all 'group_' roles
- to the matching LDAP groups.
- """
- # Fetch all groups
- groups = self.getGroupIds()
-
- # Scan sources
- for src in self.listUserSources():
- if not src.meta_type == "LDAPUserFolder":
- continue
-
- # Delete all former group mappings
- deletes = []
- for (grp, role) in src.getGroupMappings():
- if role.startswith('group_'):
- deletes.append(grp)
- src.manage_deleteGroupMappings(deletes)
-
- # Append all group mappings if it can be done
- ldap_groups = src.getGroups(attr = "cn")
- for grp in groups:
- if src._local_groups:
- grp_name = grp
- else:
- grp_name = grp[len('group_'):]
- Log(LOG_DEBUG, "cheching", grp_name, "in", ldap_groups, )
- if not grp_name in ldap_groups:
- continue
- Log(LOG_DEBUG, "Map", grp, "to", grp_name)
- src.manage_addGroupMapping(
- grp_name,
- grp,
- )
-
- # Return
- if REQUEST:
- return REQUEST.RESPONSE.redirect(
- self.absolute_url() + "/manage_wizard",
- )
- updateLDAPUserFolderMapping = postonly(updateLDAPUserFolderMapping)
-
-
- # #
- # The Wizard Section #
- # #
-
- def listLDAPUserFolderMapping(self,):
- """
- listLDAPUserFolderMapping(self,) => utility method
- """
- ret = []
- gruf_done = []
- ldap_done = []
-
- # Scan sources
- for src in self.listUserSources():
- if not src.meta_type == "LDAPUserFolder":
- continue
-
- # Get all GRUF & LDAP groups
- if src._local_groups:
- gruf_ids = self.getGroupIds()
- else:
- gruf_ids = self.getGroupIds()
- ldap_mapping = src.getGroupMappings()
- ldap_groups = src.getGroups(attr = "cn")
- for grp,role in ldap_mapping:
- if role in gruf_ids:
- ret.append((role, grp))
- gruf_done.append(role)
- ldap_done.append(grp)
- if not src._local_groups:
- ldap_done.append(role)
- for grp in ldap_groups:
- if not grp in ldap_done:
- ret.append((None, grp))
- for grp in gruf_ids:
- if not grp in gruf_done:
- ret.append((grp, None))
- Log(LOG_DEBUG, "return", ret)
- return ret
-
-
- security.declareProtected(Permissions.manage_users, "getInvalidMappings")
- def getInvalidMappings(self,):
- """
- return true if LUF mapping looks good
- """
- wrong = []
- grufs = []
- for gruf, ldap in self.listLDAPUserFolderMapping():
- if gruf and ldap:
- continue
- if not gruf:
- continue
- if gruf.startswith('group_'):
- gruf = gruf[len('group_'):]
- grufs.append(gruf)
- for gruf, ldap in self.listLDAPUserFolderMapping():
- if gruf and ldap:
- continue
- if not ldap:
- continue
- if ldap.startswith('group_'):
- ldap = ldap[len('group_'):]
- if ldap in grufs:
- wrong.append(ldap)
-
- return wrong
-
- security.declareProtected(Permissions.manage_users, "getLUFSource")
- def getLUFSource(self,):
- """
- getLUFSource(self,) => Helper to get a pointer to the LUF src.
- Return None if not available
- """
- for src in self.listUserSources():
- if src.meta_type == "LDAPUserFolder":
- return src
-
- security.declareProtected(Permissions.manage_users, "areLUFGroupsLocal")
- def areLUFGroupsLocal(self,):
- """return true if luf groups are stored locally"""
- return hasattr(self.getLUFSource(), '_local_groups')
-
-
- security.declareProtected(Permissions.manage_users, "haveLDAPGroupFolder")
- def haveLDAPGroupFolder(self,):
- """return true if LDAPGroupFolder is the groups source
- """
- return not not self.Groups.acl_users.meta_type == 'LDAPGroupFolder'
-
- security.declarePrivate('searchGroups')
- def searchGroups(self, **kw):
- names = self.getUserNames(__include_users__ = 0, __groups_prefixed__ = 1)
- return [{'id' : gn} for gn in names]
-
-
-
-class treeWrapper:
- """
- treeWrapper: Wrapper around user/group objects for the tree
- """
- def __init__(self, id, tree, parents = []):
- """
- __init__(self, id, tree, parents = []) => wraps the user object for dtml-tree
- """
- # Prepare self-contained information
- self._id = id
- self.name = tree[id]['name']
- self.icon = tree[id]['icon']
- self.is_group = tree[id]['is_group']
- parents.append(id)
- self.path = parents
-
- # Prepare subobjects information
- subobjects = []
- for grp_id in tree.keys():
- if id in tree[grp_id]['belongs_to']:
- subobjects.append(treeWrapper(grp_id, tree, parents))
- subobjects.sort(lambda x, y: cmp(x.sortId(), y.sortId()))
- self.subobjects = subobjects
-
- def id(self,):
- return self.name
-
- def sortId(self,):
- if self.is_group:
- return "__%s" % (self._id,)
- else:
- return self._id
-
- def tpValues(self,):
- """
- Return 'subobjects'
- """
- return self.subobjects
-
- def tpId(self,):
- return self._id
-
- def tpURL(self,):
- return self.tpId()
-
-InitializeClass(GroupUserFolder)
projects / GroupUserFolder.git / blobdiff
