Skip to content

erp5
erp5
Commit f8cd3922 authored by Jérome Perrin's avatar Jérome Perrin
Browse files
Options

base: introduce Person.initUserId 

Person.setUserId does expensive checks to ensure that user ids are
uniques, but the default id generator already guarantees unicity, so
when default id generator is used we don't need Person.setUserId unicity
checks.

Now when generating user ids, we only consider user id conflict with existing
users, because it's not so expensive and might still happen, for example if
user ids have been migrated from person references when erp5_users PAS plugin
was used.
person.setUserId still performs the expensive checks to prevent duplications
against other transactions using person.setUserId, but not against other
transactions using person.initUserId
parent 05fe5e19
Hide whitespace changes
Inline Side-by-side
Showing with 171 additions and 111 deletions
bt5/erp5_base/DocumentTemplateItem/portal_components/document.erp5.Person.py
View file @ f8cd3922
......@@ -138,7 +138,7 @@ class Person(Node, LoginAccountProviderMixin, EncryptedPasswordMixin, ERP5UserMi
self.hasMiddleName() or \
self._baseHasTitle()
def __checkUserIdAvailability(self, pas_plugin_class, user_id=None, login=None):
def __checkUserIdAvailability(self, pas_plugin_class, user_id=None, login=None, check_concurrent_execution=True):
# Encode reference to hex to prevent uppercase/lowercase conflict in
# activity table (when calling countMessageWithTag)
if user_id:
......@@ -163,12 +163,15 @@ class Person(Node, LoginAccountProviderMixin, EncryptedPasswordMixin, ERP5UserMi
exact_match=True,
)
):
raise UserExistsError(user_id)
raise UserExistsError(user_id or login)
else:
# PAS is used, without expected enumeration plugin: property has no
# effect on user enumeration, skip checks.
# XXX: what if desired plugin becomes active later ?
return
if not check_concurrent_execution:
return
# Check that there is no reindexation related to reference indexation
if self.getPortalObject().portal_activities.countMessageWithTag(tag):
raise UserExistsError(user_id)
......@@ -221,6 +224,53 @@ class Person(Node, LoginAccountProviderMixin, EncryptedPasswordMixin, ERP5UserMi
)
self._baseSetUserId(value)
security.declareProtected(Permissions.ModifyPortalContent, 'initUserId')
def initUserId(self):
"""Initialize user id.
ERP5 guarantees unicity of user id when setUserId is called, but this
comes at the expense of performance, because two transactions are not
allowed to change any user id at a time.
This implementation uses an id generator which already guarantees the
unicity of generated values, so when using this method we can trust
ourselves and don't need setUserId to check unicity of the user id
we are generating with other concurrent generations.
We ignore the risk that another concurrent transaction might be modifying
a user with a conflicting user id (using another method than initUserId)
and only check we are not generating an user id that would already be
used before, in case some user ids were already set by other methods than
initUserId - the most probable case being migration of persons created
before introduction of user id and ERP5 Logins.
If user id are really important in a project(which is very unlikely), this
method can be customized in a type based method named Person_initUserId
"""
method = self.getTypeBasedMethod('initUserId')
if method is not None:
return method()
if not self.hasUserId():
portal = self.getPortalObject()
user_id = 'P%i' % portal.portal_ids.generateNewId(
id_group='user_id',
id_generator='non_continuous_integer_increasing',
)
self.__checkUserIdAvailability(
pas_plugin_class=ERP5LoginUserManager,
user_id=user_id,
check_concurrent_execution=False
)
# until migration from ERP5UserManager -> ERP5UserManager is completed
# we want to make sure we are not generating a user id that was used
# as a reference of a not yet migrated person, otherwise we'll have a
# duplicate when this person will be migrated.
self.__checkUserIdAvailability(
pas_plugin_class=ERP5UserManager,
login=user_id,
check_concurrent_execution=False
)
self._baseSetUserId(user_id)
self.reindexObject()
# Time management
security.declareProtected(Permissions.AccessContentsInformation,
'getAvailableTime')
......
bt5/erp5_base/SkinTemplateItem/portal_skins/erp5_base/Person_afterClone.py
View file @ f8cd3922
......@@ -11,6 +11,6 @@ context.manage_delObjects(ids=[
)
])
context.Person_initUserId()
context.initUserId()
if not context.REQUEST.get('is_business_template_installation', 0):
context.setReference(None)
bt5/erp5_base/SkinTemplateItem/portal_skins/erp5_base/Person_init.py
View file @ f8cd3922
# this script can be overridden, but always call it when doing so (use skinSuper).
context.Person_initUserId()
context.initUserId()
bt5/erp5_base/SkinTemplateItem/portal_skins/erp5_base/Person_initUserId.py deleted 100644 → 0
View file @ 05fe5e19
if not context.hasUserId():
context.setUserId(
'P%i' % (
context.getPortalObject().portal_ids.generateNewId(
id_group='user_id',
id_generator='non_continuous_integer_increasing',
),
),
)
bt5/erp5_base/SkinTemplateItem/portal_skins/erp5_base/Person_initUserId.xml deleted 100644 → 0
View file @ 05fe5e19
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<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>*args, **kw</string> </value>
</item>
<item>
<key> <string>_proxy_roles</string> </key>
<value>
<tuple>
<string>Manager</string>
<string>Owner</string>
</tuple>
</value>
</item>
<item>
<key> <string>guard</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>Person_initUserId</string> </value>
</item>
</dictionary>
</pickle>
</record>
<record id="2" aka="AAAAAAAAAAI=">
<pickle>
<global name="Guard" module="Products.DCWorkflow.Guard"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>roles</string> </key>
<value>
<tuple>
<string>Owner</string>
</tuple>
</value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
bt5/erp5_core_test/TestTemplateItem/portal_components/test.erp5.testPerson.py
View file @ f8cd3922
......@@ -28,6 +28,7 @@
##############################################################################
import unittest
import mock
from AccessControl.SecurityManagement import newSecurityManager
from AccessControl import Unauthorized
......@@ -225,6 +226,18 @@ class TestPerson(ERP5TypeTestCase):
p.setFirstName('bob')
self.assertTrue(p.hasTitle())
def testCreatingPersonDoesNotModifyPersonModule(self):
# creating a first entry in the module modifies the module, so make sure we have at least one.
self.portal.person_module.newContent()
self.tic()
self.assertFalse(self.portal.person_module._p_changed)
# use an id generator which does not modify the module
with mock.patch.object(self.portal.person_module.__class__, 'generateNewId', return_value=self.id()):
person = self.portal.person_module.newContent(portal_type='Person')
self.assertTrue(person.getUserId())
self.assertFalse(self.portal.person_module._p_changed)
def testSetPasswordSecurity(self):
p = self._makeOne(id='person')
p.manage_permission(Permissions.SetOwnPassword, [], 0)
......
product/ERP5Security/tests/testERP5Security.py
View file @ f8cd3922
......@@ -30,6 +30,7 @@
"""Tests ERP5 User Management.
"""
import mock
import itertools
import transaction
import unittest
......@@ -489,6 +490,28 @@ class DuplicatePrevention(UserManagementTestCase):
def test_duplicateLoginReferenceInAnotherTransaction(self):
self._duplicateLoginReference(True)
def test_AutoGenerateExistingUserId(self):
person = self.portal.person_module.newContent(portal_type='Person')
# Create a state where the next generated user_id is already taken.
# This test assumes that generated user id are "P%i" % i where i is inscreasing
self.assertEqual(person.getUserId()[0], 'P')
latest_user_id = int(person.getUserId()[1:])
person.setUserId('P%i' % (latest_user_id + 1))
self.tic()
# When generating an user id we check this user id already exists
# and fail if we generate an already existant user id
self.assertRaisesRegexp(
ValidationFailed,
'user id P%i already exists' % (latest_user_id + 1),
self.portal.person_module.newContent,
portal_type='Person')
# This error is not permanent, the id is skipped and next generation should succeed
self.assertEqual(
self.portal.person_module.newContent(portal_type='Person').getUserId(),
'P%i' % (latest_user_id + 2))
class TestPreferences(UserManagementTestCase):
......@@ -734,28 +757,41 @@ class TestPASAPI(UserManagementTestCase):
class TestMigration(UserManagementTestCase):
"""Tests migration to from login on the person to ERP5 Login documents.
"""
def test_PersonLoginMigration(self):
def _enableERP5UsersPlugin(self):
"""enable the legacy erp5_users plugin"""
if 'erp5_users' not in self.portal.acl_users:
self.portal.acl_users.manage_addProduct['ERP5Security'].addERP5UserManager('erp5_users')
self.portal.acl_users.erp5_users.manage_activateInterfaces([
'IAuthenticationPlugin',
'IUserEnumerationPlugin',
])
def _createERP5UserPerson(self, username, password):
pers = self.portal.person_module.newContent(
portal_type='Person',
reference='the_user',
reference=username,
user_id=None,
)
pers.newContent(
portal_type='Assignment',
).open()
pers.setPassword('secret')
pers.setPassword(password)
self.assertEqual(len(pers.objectValues(portal_type='ERP5 Login')), 0)
self.tic()
return pers
def test_PersonLoginMigration(self):
# migration creates ERP5 Login for person and disable erp5_users plugin at the end.
self._enableERP5UsersPlugin()
pers = self._createERP5UserPerson('the_user', 'secret')
self._assertUserExists('the_user', 'secret')
self.portal.portal_templates.fixConsistency(filter={'constraint_type': 'post_upgrade'})
self.portal.portal_caches.clearAllCache()
self.tic()
# during migration, old users can still login
def stop_condition(message_list):
self._assertUserExists('the_user', 'secret')
return False
self.tic(stop_condition=stop_condition)
self._assertUserExists('the_user', 'secret')
self.assertEqual(pers.getPassword(), None)
self.assertEqual(pers.Person_getUserId(), 'the_user')
......@@ -771,6 +807,7 @@ class TestMigration(UserManagementTestCase):
[x[0] for x in self.portal.acl_users.plugins.listPlugins(IUserEnumerationPlugin)])
def test_ERP5LoginUserManagerMigration(self):
# migration creates and activate erp5_login_users plugin
acl_users= self.portal.acl_users
acl_users.manage_delObjects(ids=['erp5_login_users'])
portal_templates = self.portal.portal_templates
......@@ -784,6 +821,69 @@ class TestMigration(UserManagementTestCase):
self.assertTrue('erp5_login_users' in \
[x[0] for x in self.portal.acl_users.plugins.listPlugins(IUserEnumerationPlugin)])
def test_DuplicateUserIdPreventionDuringMigration(self):
self._enableERP5UsersPlugin()
pers = self._createERP5UserPerson('old_user_id', 'secret')
self.assertRaisesRegexp(
ValidationFailed,
'user id old_user_id already exists',
self.portal.person_module.newContent,
portal_type='Person',
reference='old_user_id')
self.portal.portal_templates.fixConsistency(filter={'constraint_type': 'post_upgrade'})
def stop_condition(message_list):
if [m for m in message_list if m.method_id != 'immediateReindexObject']:
self.assertRaisesRegexp(
ValidationFailed,
'user id old_user_id already exists',
self.portal.person_module.newContent,
portal_type='Person',
reference='old_user_id')
return False
self.tic(stop_condition=stop_condition)
self.portal.person_module.newContent(portal_type='Person', reference='old_user_id')
self.assertRaisesRegexp(
ValidationFailed,
'user id old_user_id already exists',
self.portal.person_module.newContent,
portal_type='Person',
user_id='old_user_id')
def test_DuplicateUserIdFromInitUserIdPreventionDuringMigration(self):
self._enableERP5UsersPlugin()
self._createERP5UserPerson('P1234', 'secret')
# mock user id generator to generate a value that was used as reference on that existing person.
def generateNewId(id_group, id_generator, *args, **kw):
if id_group == 'user_id' and id_generator == 'non_continuous_integer_increasing':
return 1234 # this will be 'P%i' and become P1234
return mock.DEFAULT
with mock.patch.object(self.portal.portal_ids.__class__, 'generateNewId', side_effect=generateNewId):
self.assertRaisesRegexp(
ValidationFailed,
'user id P1234 already exists',
self.portal.person_module.newContent,
portal_type='Person',)
self.portal.portal_templates.fixConsistency(filter={'constraint_type': 'post_upgrade'})
def stop_condition(message_list):
if [m for m in message_list if m.method_id != 'immediateReindexObject']:
self.assertRaisesRegexp(
ValidationFailed,
'user id P1234 already exists',
self.portal.person_module.newContent,
portal_type='Person',)
return False
self.tic(stop_condition=stop_condition)
self.assertRaisesRegexp(
ValidationFailed,
'user id P1234 already exists',
self.portal.person_module.newContent,
portal_type='Person',)
class TestUserManagementExternalAuthentication(TestUserManagement):
def getTitle(self):
......
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