From 12ac564ee79b129fe092f48c9ceefd5907763d2a Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Mon, 20 Jul 2026 13:02:31 +0200 Subject: [PATCH] Implement `main_id` conflict checks in `create_user` to match the existing openid `alias` one. Makes sure that user creation with changed ID fields but same `main_id` is logged and refused with a similar conflict message. That is, basically the first iteration of the solution outlined in #604. Adjusts the unit tests to make sure they cover both the openid `alias` and `main_id` cases and extend to verify that the logs in fact contains the expected warnings to rule out false positives for other exceptions. Added a few more log warnings for similar situations we only had interactive output for. Integrates the typo fixes from #605 to reduce merge conflicts. --- mig/shared/useradm.py | 31 ++++++++++++--- tests/test_mig_shared_useradm.py | 67 +++++++++++++++++++++++++++----- 2 files changed, 83 insertions(+), 15 deletions(-) diff --git a/mig/shared/useradm.py b/mig/shared/useradm.py index da74daab5..792a015ee 100644 --- a/mig/shared/useradm.py +++ b/mig/shared/useradm.py @@ -524,28 +524,43 @@ def create_user_in_db(configuration, db_path, client_id, user, now, authorized, # NOTE: careful to skip GDP project users here if configuration.user_openid_providers and \ configuration.user_openid_alias: + user_main_ids = dict([(key, val.get('main_id', None)) + for (key, val) in user_db.items()]) if not configuration.site_enable_gdp: user_aliases = dict([(key, val[configuration.user_openid_alias]) for (key, val) in user_db.items()]) alias = user[configuration.user_openid_alias] + main_id = user.get('main_id', 'UNSET') elif not is_gdp_user(configuration, client_id): user_aliases = dict([(key, val[configuration.user_openid_alias]) for (key, val) in user_db.items() if not is_gdp_user(configuration, key)]) alias = user[configuration.user_openid_alias] + main_id = user.get('main_id', 'UNSET') else: user_aliases = {} alias = None + main_id = 'UNSET' + alias_conflict, main_id_conflict = False, False if alias in user_aliases.values() and \ user_aliases.get(client_id, None) != alias: + alias_conflict = True + if main_id in user_main_ids.values() and \ + user_main_ids.get(client_id, None) != main_id: + main_id_conflict = True + + if alias_conflict or main_id_conflict: if do_lock: unlock_user_db(flock) + _logger.warning('refuse create %r with alias %s / id %s conflict' + % (client_id, alias, main_id)) if verbose: - print('Attempting create user with conflicting alias %s' - % alias) + print('Attempting create user with alias %s / id %s conflict' + % (alias, main_id)) raise Exception( - 'A conflicting user with alias %s already exists' % alias) + 'A conflicting user with alias %s / id %s already exists' % + (alias, main_id)) if client_id not in user_db: _logger.debug('add new user %r in user DB' % client_id) @@ -569,6 +584,8 @@ def create_user_in_db(configuration, db_path, client_id, user, now, authorized, else: if do_lock: unlock_user_db(flock) + _logger.warning('refusing to renew %s account for %r' % + (account_status, client_id)) if verbose: print('Refusing to renew %s account for %r' % (account_status, client_id)) @@ -722,11 +739,13 @@ def create_user_in_db(configuration, db_path, client_id, user, now, authorized, # NOTE: bail out here on edit/renewal as IDs are not unique if do_lock: unlock_user_db(flock) - raise ValueError("unique ID for %s (%r) has a collission!" % + _logger.warning("refuse renew %s (%r) with collision" % + (client_id, user['unique_id'])) + raise ValueError("unique ID for %s (%r) has a collision!" % (client_id, user['unique_id'])) else: - _logger.warning("retry unique ID for %s (%r) - collission" % - (client_id, user['unique_id'])) + _logger.info("regenerate unique ID for %s (%r) - in use" % + (client_id, user['unique_id'])) user['unique_id'] = generate_random_ascii(unique_id_length, unique_id_charset) if not found_unique: diff --git a/tests/test_mig_shared_useradm.py b/tests/test_mig_shared_useradm.py index d3c30145f..728990b60 100644 --- a/tests/test_mig_shared_useradm.py +++ b/tests/test_mig_shared_useradm.py @@ -401,7 +401,11 @@ def test_user_creation_fails_in_renew_when_locked(self): except Exception: self.assertFalse(True, "should not be reached") - def test_user_creation_with_id_collission_fails(self): + def test_user_creation_with_main_id_conflict_fails(self): + # NOTE: we need fake openid setup here for conflict detection code path + self.configuration.user_openid_providers = ["dummyoidprovider.org"] + self.configuration.user_openid_alias = "email" + user_dict = {} user_dict["full_name"] = "Test User" user_dict["organization"] = "Test Org" @@ -411,6 +415,7 @@ def test_user_creation_with_id_collission_fails(self): user_dict["comment"] = "This is the create comment" user_dict["password"] = "password" user_dict["distinguished_name"] = TEST_USER_DN + user_dict["main_id"] = "testmainid1234" try: create_user( @@ -419,17 +424,61 @@ def test_user_creation_with_id_collission_fails(self): except Exception: self.assertFalse(True, "should not be reached") - # NOTE: reset distinguished_name and introduce an ID conflict to test + # NOTE: reset distinguished_name and most ID fields but keep main_id + # to test that main ID conflict still prevents the creation. del user_dict["distinguished_name"] - user_dict["organization"] = "Another Org" - with self.assertRaises(Exception): + del user_dict["unique_id"] + user_dict["full_name"] = "Changed Name" + user_dict["organization"] = "Changed Org" + user_dict["email"] = OTHER_USER_EMAIL + with self.assertLogs(level='WARNING') as log_capture: + with self.assertRaises(Exception): + create_user( + user_dict, + self.configuration, + keyword_auto, + default_renew=True, + ask_renew=False, + ) + self.assertTrue(any('conflict' in msg for msg in + log_capture.output)) + + def test_user_creation_with_openid_alias_conflict_fails(self): + # NOTE: we need fake openid setup here for conflict detection code path + self.configuration.user_openid_providers = ["dummyoidprovider.org"] + self.configuration.user_openid_alias = "email" + + user_dict = {} + user_dict["full_name"] = "Test User" + user_dict["organization"] = "Test Org" + user_dict["state"] = "NA" + user_dict["country"] = "DK" + user_dict["email"] = "user@example.com" + user_dict["comment"] = "This is the create comment" + user_dict["password"] = "password" + user_dict["distinguished_name"] = TEST_USER_DN + + try: create_user( - user_dict, - self.configuration, - keyword_auto, - default_renew=True, - ask_renew=False, + user_dict, self.configuration, keyword_auto, default_renew=True ) + except Exception: + self.assertFalse(True, "should not be reached") + + # NOTE: reset distinguished_name and introduce an ID conflict to test + del user_dict["distinguished_name"] + user_dict["organization"] = "Another Org" + with self.assertLogs(level='WARNING') as log_capture: + with self.assertRaises(Exception): + create_user( + user_dict, + self.configuration, + keyword_auto, + default_renew=True, + ask_renew=False, + ) + self.assertTrue(any('conflict' in msg for msg in + log_capture.output)) class MigSharedUseradm__assure_current_htaccess(MigTestCase):