Skip to content

erp5
erp5
Commit 354c857e authored by Vincent Pelletier's avatar Vincent Pelletier
Browse files
Options

Optimise security group generation performance 

Improvements compared to the previous implementation:
- avoid looking up the user document again, when the PAS plugin already did
  that job
- make it possible to call a single script when multiple sources of groups
  are based on the same documents, avoiding iterating unnecessarily on
  those same documents multiple times
- avoid repeating the same membership value (ex: when a user has multiple
  assignments with a common membership subset)
- avoid resolving the same relation more than once
- do not go from document value to relative URL only to go from relative
  URL back to document value at the next step
- move security group id extraction to unrestricted python, as the security
  overhead was taking a large amount of time
In a security setup with 8 scripts (all Assignment-based), 6 base
categories, and 4 Assignments (all valid), this implementation is 10 times
faster at producing the same group id set as the previous one.
parent 244fb9fa
Hide whitespace changes
Inline Side-by-side
Showing with 534 additions and 331 deletions
bt5/erp5_administration/SkinTemplateItem/portal_skins/erp5_administration/Base_viewSecurity.py
View file @ 354c857e
from AccessControl import getSecurityManager
from zExceptions import Unauthorized
from pprint import pformat
u = getSecurityManager().getUser()
......@@ -37,14 +36,4 @@ except AttributeError:
print()
print('Local roles on document:\n', pformat(context.get_local_roles()))
print('''
----------------
Security mapping
----------------''')
if u.getId() is not None:
try:
print(context.Base_viewSecurityMappingAsUser(u.getId()))
except Unauthorized:
print("user doesn't have permission to security mapping in this context")
return printed
bt5/erp5_administration/SkinTemplateItem/portal_skins/erp5_administration/Base_viewSecurityMappingAsUser.py deleted 100644 → 0
View file @ 244fb9fa
group_id_list_generator = getattr(context, 'ERP5Type_asSecurityGroupId')
security_category_dict = {}
# XXX This is a duplicate of logic present deep inside ERP5GroupManager.getGroupsForPrincipal()
# Please refactor into an accessible method so this code can be removed
def getDefaultSecurityCategoryMapping():
return ((
'ERP5Type_getSecurityCategoryFromAssignment',
context.getPortalObject().getPortalAssignmentBaseCategoryList()
),)
getSecurityCategoryMapping = getattr(context, 'ERP5Type_getSecurityCategoryMapping', getDefaultSecurityCategoryMapping)
# XXX end of code duplication
for method_id, base_category_list in getSecurityCategoryMapping():
try:
security_category_dict.setdefault(tuple(base_category_list), []).extend(
getattr(context, method_id)(base_category_list, login, context, ''))
except Exception: # XXX: it is not possible to log message with traceback from python script
print('It was not possible to invoke method %s with base_category_list %s'%(method_id, base_category_list))
for base_category_list, category_value_list in security_category_dict.items():
print('base_category_list: %s' % (base_category_list,))
for category_dict in category_value_list:
print('-> category_dict: %s' % category_dict)
print('--> %s' % group_id_list_generator(category_order=base_category_list,
**category_dict))
return printed
bt5/erp5_administration/SkinTemplateItem/portal_skins/erp5_administration/Base_viewSecurityMappingAsUser.xml deleted 100644 → 0
View file @ 244fb9fa
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="_reconstructor" module="copy_reg"/>
</klass>
<tuple>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
<global name="object" module="__builtin__"/>
<none/>
</tuple>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>login</string> </value>
</item>
<item>
<key> <string>_proxy_roles</string> </key>
<value>
<tuple>
<string>Manager</string>
<string>Member</string>
</tuple>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>Base_viewSecurityMappingAsUser</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
bt5/erp5_base/DocumentTemplateItem/portal_components/document.erp5.RoleDefinition.py
View file @ 354c857e
......@@ -29,8 +29,10 @@ import zope.interface
from AccessControl import ClassSecurityInfo, Unauthorized
from Products.ERP5Type import Permissions, PropertySheet, interfaces
from Products.ERP5Type.XMLObject import XMLObject
from Products.ERP5Type.ERP5Type \
import ERP5TYPE_SECURITY_GROUP_ID_GENERATION_SCRIPT
from Products.ERP5Type.ERP5Type import (
ERP5TYPE_SECURITY_GROUP_ID_GENERATION_SCRIPT,
ERP5TYPE_SECURITY_GROUP_ID_GENERATION_SCRIPT_V2,
)
@zope.interface.implementer(interfaces.ILocalRoleGenerator)
class RoleDefinition(XMLObject):
......@@ -60,8 +62,21 @@ class RoleDefinition(XMLObject):
security.declarePrivate("getLocalRolesFor")
def getLocalRolesFor(self, ob, user_name=None):
group_id_generator = getattr(ob,
ERP5TYPE_SECURITY_GROUP_ID_GENERATION_SCRIPT)
ERP5TYPE_SECURITY_GROUP_ID_GENERATION_SCRIPT, None)
if group_id_generator is None:
group_id_list = getattr(
ob,
ERP5TYPE_SECURITY_GROUP_ID_GENERATION_SCRIPT_V2,
)(
category_dict={
'agent': [(x, False) for x in self.getAgentValueList()],
},
)
else: # BBB
group_id_list = group_id_generator(
category_order=('agent',),
agent=self.getAgentList(),
)
role_list = self.getRoleName(),
return {group_id: role_list
for group_id in group_id_generator(category_order=('agent',),
agent=self.getAgentList())}
for group_id in group_id_list}
product/ERP5/ERP5Site.py
View file @ 354c857e
......@@ -1614,27 +1614,12 @@ class ERP5Site(ResponseHeaderGenerator, FolderMixIn, PortalObjectBase, CacheCook
def getPortalSecurityCategoryMapping(self):
"""
Returns a list of pairs composed of a script id and a list of base
category ids to use for computing security groups.
This is used during indexation, so involved scripts must not rely on
catalog at any point in their execution.
Example:
(
('script_1', ['base_category_1', 'base_category_2', ...]),
('script_2', ['base_category_1', 'base_category_3', ...])
)
DEPRECATED: implement ERP5User_getUserSecurityCategoryValueList instead.
"""
return getattr(
self,
'ERP5Type_getSecurityCategoryMapping',
lambda: ( # BBB
(
'ERP5Type_getSecurityCategoryFromAssignment',
self.getPortalAssignmentBaseCategoryList(),
),
),
lambda: (),
)()
security.declareProtected(Permissions.AccessContentsInformation,
......
product/ERP5/bootstrap/erp5_core/ExtensionTemplateItem/portal_components/extension.erp5.StandardSecurity.py
View file @ 354c857e
......@@ -24,8 +24,177 @@
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
from collections import defaultdict
from itertools import product
import six
from DateTime import DateTime
_STOP_RECURSION_PORTAL_TYPE_SET = ('Base Category', 'ERP5 Site')
def getSecurityCategoryValueFromAssignment(self, rule_dict):
"""
This function returns a list of dictionaries which represent
the security groups the user represented by self is member of,
based on the applicable Assignment objects it contains.
rule_dict (dict)
Keys: tuples listings base category names set on the Assignment
and which represent security groups assigned to the user.
Values: tuples describing which combinations of the base categories
whose parent categories the user is also member of.
Example call to illustrate argument and return value structures:
getSecurityCategoryValueFromAssignment(
rule_dict={
('function', ): ((), ('function', )),
('group', ): ((), ),
('function', 'group'): ((), ('function', )),
}
)
[
{
'function': (
(<Category function/accountant/chief>, False),
(<Category function/accountant/chief>, True),
(<Category function/accountant>, True),
),
}
{
'group': ((<Category group/nexedi>, False), ),
},
{
'function': (
(<Category function/accountant/chief>, False),
(<Category function/accountant/chief>, True),
(<Category function/accountant>, True),
),
'group': ((<Category group/nexedi>, False), ),
},
]
"""
base_category_set = set(sum((tuple(x) for x in rule_dict), ()))
recursive_base_category_set = set(sum((sum((tuple(y) for y in x), ()) for x in six.itervalues(rule_dict)), ()))
category_value_set_dict = defaultdict(set)
parent_category_value_dict = {}
assignment_membership_dict_list = []
now = DateTime()
for assignment_value in self.objectValues(portal_type='Assignment'):
if assignment_value.getValidationState() == 'open' and (
not assignment_value.hasStartDate() or assignment_value.getStartDate() <= now
) and (
not assignment_value.hasStopDate() or assignment_value.getStopDate() >= now
):
assignment_membership_dict = {}
for base_category in base_category_set:
category_value_list = assignment_value.getAcquiredValueList(base_category)
if category_value_list:
assignment_membership_dict[base_category] = tuple(set(category_value_list))
category_value_set_dict[base_category].update(category_value_list)
if base_category in recursive_base_category_set:
for category_value in category_value_list:
while True:
parent_category_value = category_value.getParentValue()
if (
parent_category_value in parent_category_value_dict or
parent_category_value.getPortalType() in _STOP_RECURSION_PORTAL_TYPE_SET
):
break
parent_category_value_dict[category_value] = parent_category_value
category_value = parent_category_value
if assignment_membership_dict:
assignment_membership_dict_list.append(assignment_membership_dict)
result = []
for base_category_list, recursion_list in six.iteritems(rule_dict):
result_entry = set()
for assignment_membership_dict in assignment_membership_dict_list:
assignment_category_list = []
for base_category in base_category_list:
category_set = set()
for category_value in assignment_membership_dict.get(base_category, ()):
for recursion_base_category_set in recursion_list:
if base_category in recursion_base_category_set:
while True:
category_set.add((category_value, True))
try:
category_value = parent_category_value_dict[category_value]
except KeyError:
break
else:
category_set.add((category_value, False))
assignment_category_list.append((base_category, tuple(category_set)))
if assignment_category_list:
result_entry.add(tuple(assignment_category_list))
for result_item in result_entry:
result.append(dict(result_item))
return result
def asSecurityGroupIdSet(category_dict, key_sort=sorted):
"""
The script takes the following parameters:
category_dict (dict)
keys: base category names.
values: list of categories composing the security groups the document is member of.
key_sort ((dict) -> key vector)
Function receiving the value of category_dict and returning the list of keys to
use to construct security group names, in the order in which these group names
will be used.
May return keys which are not part of category_dict.
Defaults to a lexicographic sort of all keys.
The External Method pointing at this implementation should be overridden (and
called using skinSuper) in order to provide this argument with a custom value.
Example call:
context.ERP5Type_asSecurityGroupIdSet(
category_dict={
'site': [(<Category france/lille>, False)],
'group': [(<Category nexedi>, False)],
'function': [(<Category accounting/accountant>, True)],
}
)
This will generate a string like 'ACT*_NXD_LIL' where "LIL", "NXD" and "ACT" are
the codification of respecively "france/lille", "nexedi" and "accounting/accountant"
categories.
If the category points to a document portal type (ex. trade condition, project, etc.),
and if no codification property is defined for this type of object,
the security ID group is generated by considering the object reference or
the object ID.
ERP5Type_asSecurityGroupIdSet can also return a list of users whenever a
category points to a Person instance. This is useful to implement user based local
role assignments instead of abstract security based local roles.
"""
list_of_list = []
user_list = []
associative_list = []
for base_category_id in key_sort(category_dict):
try:
category_list = category_dict[base_category_id]
except KeyError:
continue
for category_value, is_child_category in category_list:
if category_value.getPortalType() == 'Person':
user_name = category_value.Person_getUserId()
if user_name is not None:
user_list.append(user_name)
associative_list = []
elif not user_list:
associative_list.append(
(
category_value.getProperty('codification') or
category_value.getProperty('reference') or
category_value.getId()
) + ('*' if is_child_category else ''),
)
if associative_list:
list_of_list.append(associative_list)
associative_list = []
if user_list:
return user_list
return ['_'.join(x) for x in product(*list_of_list) if x]
def getSecurityCategoryFromAssignment(
self,
base_category_list,
......@@ -35,6 +204,8 @@ def getSecurityCategoryFromAssignment(
child_category_list=None
):
"""
DEPRECATED: use getSecurityCategoryValueFromAssignment for better performance.
This script returns a list of dictionaries which represent
the security groups which a person is member of. It extracts
the categories from the current user assignment.
......
product/ERP5/bootstrap/erp5_core/SkinTemplateItem/portal_skins/erp5_core/ERP5Type_asSecurityGroupId.py deleted 100644 → 0
View file @ 244fb9fa
"""
This script is used to convert a list of categories into an security
identifier (security ID). It is invoked by two classes in ERP5:
- ERP5Type.py to convert security definitions made of
multiple categories into security ID strings
- ERP5GroupManager.py to convert an assignment definition
into a single security ID string. It should be noted here
that ERP5GroupManager.py also tries to invoke ERP5Type_asSecurityGroupIdList
(DEPRECATED) in order associate a user to multiple security groups.
In this case ERP5Type_asSecurityGroupId is not invoked.
The script takes the following parameters:
category_order - list of base_categories we want to use to generate the group id
kw - keys should be base categories, values should be value
of corresponding relative urls (obtained by getBaseCategory())
Example call:
context.ERP5TypeSecurity_asGroupId(category_order=('site', 'group', 'function'),
site='france/lille', group='nexedi', function='accounting/accountant')
This will generate a string like 'LIL_NXD_ACT' where "LIL", "NXD" and "ACT" are
the codification of respecively "france/lille", "nexedi" and "accounting/accountant" categories
If the category points to a document portal type (ex. trade condition, project, etc.),
and if no codification property is defined for this type of object,
the security ID group is generated by considering the object reference or
the object ID.
ERP5Type_asSecurityGroupId can also return a list of users whenever a category points
to a Person instance. This is useful to implement user based local role assignments
instead of abstract security based local roles.
"""
portal = context.getPortalObject()
getCategoryValue = portal.portal_categories.getCategoryValue
# sort the category list lexicographically
# this prevents us to choose the exact order we want,
# but also prevents some human mistake to break everything by creating site_function instead of function_site
if category_order not in (None, ''):
category_order = list(category_order)
category_order.sort()
else:
category_order = []
# Prepare a cartesian product
from Products.ERP5Type.Utils import cartesianProduct
list_of_list = []
user_list = []
for base_category in category_order:
# It is acceptable for a category not to be defined
try:
category_list = kw[base_category]
except KeyError:
continue
associative_list = []
if isinstance(category_list, str):
category_list = [category_list]
for category in category_list:
if category[-1] == '*':
category = category[:-1]
is_child_category = 1
else:
is_child_category = 0
category_path = '%s/%s' % (base_category, category)
category_object = getCategoryValue(category_path)
if category_object is None:
raise RuntimeError("Security definition error (category %r not found)" % (category_path,))
portal_type = category_object.getPortalType()
if portal_type == 'Person':
# We define a person here
user_name = category_object.Person_getUserId()
if user_name is not None:
user_list.append(user_name)
else:
category_code = (category_object.getProperty('codification') or
category_object.getProperty('reference') or
category_object.getId())
if is_child_category:
category_code += '*'
associative_list.append(category_code)
# Prevent making a cartesian product with an empty set
if associative_list:
list_of_list.append(associative_list)
# Return a list of users if any was defined
if user_list:
return user_list
# Compute the cartesian product and return the codes
# return filter(lambda x: x, map(lambda x: '_'.join(x), cartesianProduct(list_of_list)))
return ['_'.join(x) for x in cartesianProduct(list_of_list) if x]
product/ERP5/bootstrap/erp5_core/SkinTemplateItem/portal_skins/erp5_core/ERP5Type_asSecurityGroupIdSet.xml 0 → 100644
View file @ 354c857e
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="ExternalMethod" module="Products.ExternalMethod.ExternalMethod"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_function</string> </key>
<value> <string>asSecurityGroupIdSet</string> </value>
</item>
<item>
<key> <string>_module</string> </key>
<value> <string>StandardSecurity</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>ERP5Type_asSecurityGroupIdSet</string> </value>
</item>
<item>
<key> <string>title</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
product/ERP5/bootstrap/erp5_core/SkinTemplateItem/portal_skins/erp5_core/ERP5User_getSecurityCategoryValueFromAssignment.xml 0 → 100644
View file @ 354c857e
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="ExternalMethod" module="Products.ExternalMethod.ExternalMethod"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_function</string> </key>
<value> <string>getSecurityCategoryValueFromAssignment</string> </value>
</item>
<item>
<key> <string>_module</string> </key>
<value> <string>StandardSecurity</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>ERP5User_getSecurityCategoryValueFromAssignment</string> </value>
</item>
<item>
<key> <string>title</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
product/ERP5/bootstrap/erp5_core/SkinTemplateItem/portal_skins/erp5_core/ERP5User_getUserSecurityCategoryValueList.py 0 → 100644
View file @ 354c857e
"""
Override this script to customise user security group generation.
It is called on the context of the ERP5 document which represents the user.
This is called when a user object is being prepared by PAS, typically at the
end of traversal but also when calling getUser and getUserById API, in order
to list the security groups the user is member of.
When called by PAS, this script is called in a super-user security context.
"""
return context.ERP5User_getSecurityCategoryValueFromAssignment(
rule_dict={
tuple(context.getPortalObject().getPortalAssignmentBaseCategoryList()): ((), )
},
)
product/ERP5/bootstrap/erp5_core/SkinTemplateItem/portal_skins/erp5_core/ERP5Type_asSecurityGroupId.xml → product/ERP5/bootstrap/erp5_core/SkinTemplateItem/portal_skins/erp5_core/ERP5User_getUserSecurityCategoryValueList.xml
View file @ 354c857e
......@@ -50,11 +50,11 @@
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>category_order, **kw</string> </value>
<value> <string></string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>ERP5Type_asSecurityGroupId</string> </value>
<value> <string>ERP5User_getUserSecurityCategoryValueList</string> </value>
</item>
</dictionary>
</pickle>
......
product/ERP5Security/ERP5GroupManager.py
View file @ 354c857e
......@@ -25,8 +25,10 @@ from Products.PluggableAuthService.plugins.BasePlugin import BasePlugin
from Products.PluggableAuthService.utils import classImplements
from Products.PluggableAuthService.interfaces.plugins import IGroupsPlugin
from Products.ERP5Type.Cache import CachingMethod
from Products.ERP5Type.ERP5Type \
import ERP5TYPE_SECURITY_GROUP_ID_GENERATION_SCRIPT
from Products.ERP5Type.ERP5Type import (
ERP5TYPE_SECURITY_GROUP_ID_GENERATION_SCRIPT,
ERP5TYPE_SECURITY_GROUP_ID_GENERATION_SCRIPT_V2,
)
from Products.ERP5Type.UnrestrictedMethod import UnrestrictedMethod
from Products.ZSQLCatalog.SQLCatalog import SimpleQuery
from ZODB.POSException import ConflictError
......@@ -115,60 +117,136 @@ class ERP5GroupManager(BasePlugin):
user_value = getattr(principal, 'getUserValue', lambda: None)()
if user_value is None:
return ()
security_category_dict = defaultdict(list)
for method_name, base_category_list in self.getPortalSecurityCategoryMapping():
base_category_list = tuple(base_category_list)
security_category_list = security_category_dict[base_category_list]
try:
# The called script may want to distinguish if it is called
# from here or from _updateLocalRolesOnSecurityGroups.
# Currently, passing portal_type='' (instead of 'Person')
# is the only way to make the difference.
security_category_list.extend(
getattr(self, method_name)(
base_category_list,
user_id,
user_value,
'',
)
)
except ConflictError:
raise
except Exception:
LOG(
'ERP5GroupManager',
WARNING,
'could not get security categories from %s' % (method_name, ),
error=True,
)
# Get group names from category values
# XXX try ERP5Type_asSecurityGroupIdList first for compatibility
generator_name = 'ERP5Type_asSecurityGroupIdList'
group_id_list_generator = getattr(self, generator_name, None)
if group_id_list_generator is None:
generator_name = ERP5TYPE_SECURITY_GROUP_ID_GENERATION_SCRIPT
group_id_list_generator = getattr(self, generator_name, None)
security_group_list = []
for base_category_list, category_value_list in six.iteritems(security_category_dict):
for category_dict in category_value_list:
category_mapping = self.getPortalSecurityCategoryMapping()
if category_mapping: # BBB
has_relative_urls = True
security_category_dict = defaultdict(list)
for method_name, base_category_list in category_mapping:
base_category_list = tuple(base_category_list)
security_category_list = security_category_dict[base_category_list]
try:
group_id_list = group_id_list_generator(
category_order=base_category_list,
**category_dict
# The called script may want to distinguish if it is called
# from here or from _updateLocalRolesOnSecurityGroups.
# Currently, passing portal_type='' (instead of 'Person')
# is the only way to make the difference.
security_category_list.extend(
getattr(self, method_name)(
base_category_list,
user_id,
user_value,
'',
)
)
if isinstance(group_id_list, str):
group_id_list = [group_id_list]
security_group_list.extend(group_id_list)
except ConflictError:
raise
except Exception:
LOG(
'ERP5GroupManager',
WARNING,
'could not get security groups from %s' % (generator_name, ),
'could not get security categories from %s' % (method_name, ),
error=True,
)
return tuple(security_group_list)
else:
has_relative_urls = False
try:
getUserSecurityCategoryValueList = user_value.ERP5User_getUserSecurityCategoryValueList
except AttributeError: # BBB
security_category_value_dict_list = []
else:
security_category_value_dict_list = getUserSecurityCategoryValueList()
security_group_set = set()
# XXX try ERP5Type_asSecurityGroupIdList first for compatibility
generator_name = 'ERP5Type_asSecurityGroupIdList'
group_id_list_generator = getattr(self, generator_name, None)
if group_id_list_generator is None:
generator_name = ERP5TYPE_SECURITY_GROUP_ID_GENERATION_SCRIPT
group_id_list_generator = getattr(self, generator_name, None)
if group_id_list_generator is None:
if has_relative_urls:
# Convert security_category_dict to security_category_value_dict_list
# Differences with direct security_category_value_dict_list production:
# - incomplete deduplication
# - extra intermediate sorting
getCategoryValue = self.portal_categories.getCategoryValue
security_category_value_dict_list = (
dict(x)
for x in {
tuple(sorted(
(
(
base_category,
tuple(
(
getCategoryValue(base_category + '/' + relative_url.rstrip('*')),
relative_url.endswith('*'),
)
for relative_url in (
(relative_url_list, )
if isinstance(relative_url_list, str) else
sorted(relative_url_list)
)
)
)
for (
base_category,
relative_url_list,
) in six.iteritems(security_dict)
),
# Avoid comparing persistent objects, for performance purposes:
# these are stored by path.
key=lambda x: x[0]
))
for security_category_list in six.itervalues(security_category_dict)
for security_dict in security_category_list
}
)
for security_category_value_dict in security_category_value_dict_list:
security_group_set.update(
getattr(self, ERP5TYPE_SECURITY_GROUP_ID_GENERATION_SCRIPT_V2)(
category_dict=security_category_value_dict,
),
)
else: # BBB
if not has_relative_urls:
# Convert security_category_value_dict_list to security_category_dict
# Differences with direct security_category_dict generation:
# - the order of items in the tuples used as keys is random
# - repetitions will be missing (which is arguably an improvement)
security_category_dict = {
tuple(security_category_value_dict): [
{
base_category: [
category_value.getRelativeUrl() + ('*' if parent else '')
for category_value, parent in category_value_list
]
for base_category, category_value_list in six.iteritems(
security_category_value_dict,
)
}
]
for security_category_value_dict in security_category_value_dict_list
}
# Get group names from category values
for base_category_list, category_value_list in six.iteritems(security_category_dict):
for category_dict in category_value_list:
try:
group_id_list = group_id_list_generator(
category_order=base_category_list,
**category_dict
)
if isinstance(group_id_list, str):
group_id_list = [group_id_list]
security_group_set.update(group_id_list)
except ConflictError:
raise
except Exception:
LOG(
'ERP5GroupManager',
WARNING,
'could not get security groups from %s' % (generator_name, ),
error=True,
)
return tuple(security_group_set)
if not NO_CACHE_MODE and getattr(_CACHE_ENABLED_LOCAL, 'value', True):
_getGroupsForPrincipal = CachingMethod(_getGroupsForPrincipal,
......
product/ERP5Security/tests/testERP5Security.py
View file @ 354c857e
......@@ -189,23 +189,22 @@ class UserManagementTestCase(ERP5TypeTestCase):
return dummy_document
class RoleManagementTestCase(UserManagementTestCase):
class RoleManagementTestCaseBase(UserManagementTestCase):
"""Test case with required configuration to test role definitions.
"""
def afterSetUp(self):
"""Initialize requirements of security configuration.
"""
super(RoleManagementTestCase, self).afterSetUp()
super(RoleManagementTestCaseBase, self).afterSetUp()
# create a security configuration script
skin_folder = self.portal.portal_skins.custom
if 'ERP5Type_getSecurityCategoryMapping' not in skin_folder.objectIds():
if self._security_configuration_script_id not in skin_folder.objectIds():
createZODBPythonScript(
skin_folder, 'ERP5Type_getSecurityCategoryMapping', '',
"""return ((
'ERP5Type_getSecurityCategoryFromAssignment',
context.getPortalObject().getPortalAssignmentBaseCategoryList()
),)
""")
skin_folder,
self._security_configuration_script_id,
'',
self._security_configuration_script_body,
)
# configure group, site, function categories
category_tool = self.getCategoryTool()
for bc in ['group', 'site', 'function']:
......@@ -241,6 +240,29 @@ class RoleManagementTestCase(UserManagementTestCase):
self.tic()
class RoleManagementTestCaseOld(RoleManagementTestCaseBase):
"""
RoleManagementTestCaseBase variant using the deprecated security declaration API.
"""
_security_configuration_script_id = 'ERP5Type_getSecurityCategoryMapping'
_security_configuration_script_body = """return ((
'ERP5Type_getSecurityCategoryFromAssignment',
context.getPortalObject().getPortalAssignmentBaseCategoryList()
),)"""
class RoleManagementTestCase(RoleManagementTestCaseBase):
"""
RoleManagementTestCaseBase variant using the current security declaration API.
"""
_security_configuration_script_id = 'ERP5User_getUserSecurityCategoryValueList'
_security_configuration_script_body = """return context.ERP5User_getSecurityCategoryValueFromAssignment(
rule_dict={
tuple(context.getPortalObject().getPortalAssignmentBaseCategoryList()): ((), )
},
)"""
class TestUserManagement(UserManagementTestCase):
"""Tests User Management in ERP5Security.
"""
......@@ -1184,7 +1206,7 @@ class TestUserManagementExternalAuthentication(TestUserManagement):
self.assertIn(login, response.getBody())
class TestLocalRoleManagement(RoleManagementTestCase):
class _TestLocalRoleManagementMixIn(object):
"""Tests Local Role Management with ERP5Security.
"""
......@@ -1194,24 +1216,29 @@ class TestLocalRoleManagement(RoleManagementTestCase):
def afterSetUp(self):
"""Called after setup completed.
"""
super(TestLocalRoleManagement, self).afterSetUp()
super(_TestLocalRoleManagementMixIn, self).afterSetUp()
# any member can add organisations
self.portal.organisation_module.manage_permission(
'Add portal content', roles=['Member', 'Manager'], acquire=1)
self.username = 'usérn@me'
# create a user and open an assignement
pers = self.getPersonModule().newContent(portal_type='Person',
user_id=self.username)
assignment = pers.newContent( portal_type='Assignment',
group='subcat',
site='subcat',
function='subcat' )
assignment.open()
pers.newContent(portal_type='ERP5 Login',
reference=self.username,
password=self.username).validate()
user_list = self.portal.acl_users.getUserById(self.username)
if not user_list:
# create a user and open an assignement
pers = self.getPersonModule().newContent(portal_type='Person',
user_id=self.username)
assignment = pers.newContent( portal_type='Assignment',
group='subcat',
site='subcat',
function='subcat' )
assignment.open()
pers.newContent(portal_type='ERP5 Login',
reference=self.username,
password=self.username).validate()
else:
user, = user_list
pers = user.getUserValue()
self.person = pers
self.tic()
......@@ -1229,6 +1256,15 @@ class TestLocalRoleManagement(RoleManagementTestCase):
def _makeOne(self):
return self.getOrganisationModule().newContent(portal_type='Organisation')
def _createOrGetObject(self, container, content_id, new_content_kw):
try:
return container[content_id]
except KeyError:
return container.newContent(
id=content_id,
**new_content_kw
)
def getBusinessTemplateList(self):
"""List of BT to install. """
return ('erp5_base', 'erp5_web', 'erp5_ingestion', 'erp5_dms', 'erp5_administration')
......@@ -1251,11 +1287,6 @@ class TestLocalRoleManagement(RoleManagementTestCase):
def testSimpleLocalRole(self):
"""Test simple case of setting a role.
"""
def viewSecurity():
return self.publish(
self.portal.absolute_url_path() + '/Base_viewSecurity',
basic='%s:%s' % (self.username, self.username),
)
self._getTypeInfo().newContent(portal_type='Role Information',
role_name='Assignor',
description='desc.',
......@@ -1269,35 +1300,37 @@ class TestLocalRoleManagement(RoleManagementTestCase):
self.assertIn('Assignor', user.getRolesInContext(obj))
self.assertNotIn('Assignee', user.getRolesInContext(obj))
person_value = self.person
user_id = person_value.getUserId()
getUserById = self.portal.acl_users.getUserById
def assertRoleItemsEqual(expected_role_set):
self.assertItemsEqual(getUserById(user_id).getGroups(), expected_role_set)
# check if assignment change is effective immediately
assertRoleItemsEqual(['F1_G1_S1'])
self.login()
res = viewSecurity()
self.assertEqual([x for x in res.body.splitlines() if x.startswith('-->')],
["--> ['F1_G1_S1']"], res.body)
assignment = self.person.newContent( portal_type='Assignment',
group='subcat',
site='subcat',
function='another_subcat' )
assignment.open()
res = viewSecurity()
self.assertEqual([x for x in res.body.splitlines() if x.startswith('-->')],
["--> ['F1_G1_S1']", "--> ['F2_G1_S1']"], res.body)
assertRoleItemsEqual(['F1_G1_S1', 'F2_G1_S1'])
assignment.setGroup('another_subcat')
res = viewSecurity()
self.assertEqual([x for x in res.body.splitlines() if x.startswith('-->')],
["--> ['F1_G1_S1']", "--> ['F2_G2_S1']"], res.body)
assertRoleItemsEqual(['F1_G1_S1', 'F2_G1_S1'])
self.abort()
def testLocalRolesGroupId(self):
"""Assigning a role with local roles group id.
"""
self.portal.portal_categories.local_role_group.newContent(
portal_type='Category',
reference = 'Alternate',
id = 'Alternate')
self._getTypeInfo().newContent(portal_type='Role Information',
role_name='Assignor',
local_role_group_value=self.portal.portal_categories.local_role_group.Alternate,
local_role_group_value=self._createOrGetObject(
container=self.portal.portal_categories.local_role_group,
content_id='Alternate',
new_content_kw={
'portal_type': 'Category',
'reference': 'Alternate',
},
),
role_category=self.defined_category)
self.loginAsUser(self.username)
......@@ -1433,11 +1466,19 @@ class TestLocalRoleManagement(RoleManagementTestCase):
self.assertEqual(response.getStatus(), 401)
class TestKeyAuthentication(RoleManagementTestCase):
class TestLocalRoleManagementOld(_TestLocalRoleManagementMixIn, RoleManagementTestCaseOld):
pass
class TestLocalRoleManagement(_TestLocalRoleManagementMixIn, RoleManagementTestCase):
pass
class _TestKeyAuthenticationMixIn(object):
def getBusinessTemplateList(self):
"""This test also uses web and dms
"""
return super(TestKeyAuthentication, self).getBusinessTemplateList() + (
return super(_TestKeyAuthenticationMixIn, self).getBusinessTemplateList() + (
'erp5_core_proxy_field_legacy', # for erp5_web
'erp5_base', 'erp5_web', 'erp5_ingestion', 'erp5_dms', 'erp5_administration')
......@@ -1449,14 +1490,16 @@ class TestKeyAuthentication(RoleManagementTestCase):
# add key authentication PAS plugin
portal = self.portal
uf = portal.acl_users
uf.manage_addProduct['ERP5Security'].addERP5KeyAuthPlugin(
try:
erp5_auth_key_plugin = getattr(uf, "erp5_auth_key")
except AttributeError:
uf.manage_addProduct['ERP5Security'].addERP5KeyAuthPlugin(
id="erp5_auth_key", \
title="ERP5 Auth key",\
encryption_key='fdgfhkfjhltylutyu',
cookie_name='__key',\
default_cookie_name='__ac')
erp5_auth_key_plugin = getattr(uf, "erp5_auth_key")
erp5_auth_key_plugin = getattr(uf, "erp5_auth_key")
erp5_auth_key_plugin.manage_activateInterfaces(
interfaces=['IExtractionPlugin',
'IAuthenticationPlugin',
......@@ -1546,6 +1589,14 @@ class TestKeyAuthentication(RoleManagementTestCase):
self.assertEqual(response.getStatus(), 200)
class TestKeyAuthenticationOld(_TestKeyAuthenticationMixIn, RoleManagementTestCaseOld):
pass
class TestKeyAuthentication(_TestKeyAuthenticationMixIn, RoleManagementTestCase):
pass
class TestOwnerRole(UserManagementTestCase):
def _createZodbUser(self, login, role_list=None):
if role_list is None:
......
product/ERP5Type/Core/RoleInformation.py
View file @ 354c857e
......@@ -28,6 +28,7 @@
""" Information about customizable roles.
"""
from collections import defaultdict
from six import string_types as basestring
import zope.interface
from AccessControl import ClassSecurityInfo
......@@ -37,12 +38,28 @@ from Products.ERP5Type.Globals import InitializeClass
from Products.CMFCore.Expression import Expression
from Products.ERP5Type import interfaces, Permissions, PropertySheet
from Products.ERP5Type.ERP5Type \
import ERP5TYPE_SECURITY_GROUP_ID_GENERATION_SCRIPT
from Products.ERP5Type.ERP5Type import (
ERP5TYPE_SECURITY_GROUP_ID_GENERATION_SCRIPT,
ERP5TYPE_SECURITY_GROUP_ID_GENERATION_SCRIPT_V2,
)
from Products.ERP5Type.Permissions import AccessContentsInformation
from Products.ERP5Type.XMLObject import XMLObject
import six
def _toSecurityGroupIdGenerationScriptV2(
getCategoryValue,
category_definition_dict,
):
result = {}
for base_category, relative_url_list in six.iteritems(category_definition_dict):
result[base_category] = result_value_list = []
if isinstance(relative_url_list, str):
relative_url_list = (relative_url_list, )
for relative_url in relative_url_list:
category_value = getCategoryValue(base_category + '/' + relative_url.rstrip('*'))
assert category_value is not None, (base_category, relative_url, category_definition_dict)
result_value_list.append((category_value, relative_url.endswith('*')))
return result
@zope.interface.implementer(interfaces.ILocalRoleGenerator)
class RoleInformation(XMLObject):
......@@ -143,7 +160,7 @@ class RoleInformation(XMLObject):
# defined categories)
category_result = [{}]
group_id_role_dict = {}
group_id_role_dict = defaultdict(set)
role_list = self.getRoleNameList()
if isinstance(category_result, dict):
......@@ -153,26 +170,44 @@ class RoleInformation(XMLObject):
for role, group_id_list in six.iteritems(category_result):
if role in role_list:
for group_id in group_id_list:
group_id_role_dict.setdefault(group_id, set()).add(role)
group_id_role_dict[group_id].add(role)
else:
group_id_generator = getattr(ob,
ERP5TYPE_SECURITY_GROUP_ID_GENERATION_SCRIPT)
ERP5TYPE_SECURITY_GROUP_ID_GENERATION_SCRIPT, None)
# Prepare definition dict once only
category_definition_dict = {}
category_definition_dict = defaultdict(list)
for c in self.getRoleCategoryList():
bc, value = c.split('/', 1)
category_definition_dict.setdefault(bc, []).append(value)
category_definition_dict[bc].append(value)
if group_id_generator is None:
group_id_generator = getattr(ob,
ERP5TYPE_SECURITY_GROUP_ID_GENERATION_SCRIPT_V2)
getCategoryValue = self.getPortalObject().portal_categories.getCategoryValue
category_definition_dict = _toSecurityGroupIdGenerationScriptV2(
getCategoryValue=getCategoryValue,
category_definition_dict=category_definition_dict,
)
category_result = [
_toSecurityGroupIdGenerationScriptV2(
getCategoryValue=getCategoryValue,
category_definition_dict=category_dict,
)
for category_dict in category_result
]
else: # BBB
for category_dict in category_result:
category_dict.setdefault('category_order', category_order_list)
group_id_generator_ = group_id_generator
group_id_generator = lambda category_dict: group_id_generator_(**category_dict)
# category_result is a list of dicts that represents the resolved
# categories we create a category_value_dict from each of these
# dicts aggregated with category_order and statically defined
# categories
for category_dict in category_result:
category_value_dict = {'category_order':category_order_list}
category_value_dict.update(category_dict)
category_value_dict = category_dict.copy()
category_value_dict.update(category_definition_dict)
group_id_list = group_id_generator(**category_value_dict)
group_id_list = group_id_generator(category_dict=category_value_dict)
if group_id_list:
if isinstance(group_id_list, str):
# Single group is defined (this is usually for group membership)
......
product/ERP5Type/ERP5Type.py
View file @ 354c857e
......@@ -45,6 +45,7 @@ from Products.ERP5Type.dynamic.accessor_holder import getPropertySheetValueList,
import six
ERP5TYPE_SECURITY_GROUP_ID_GENERATION_SCRIPT = 'ERP5Type_asSecurityGroupId'
ERP5TYPE_SECURITY_GROUP_ID_GENERATION_SCRIPT_V2 = 'ERP5Type_asSecurityGroupIdSet'
from .TranslationProviderBase import TranslationProviderBase
from Products.ERP5Type.Accessor.Constant import PropertyGetter as ConstantGetter
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7