From fd53c5285dc7f4d19201dd85fc810285616c0f4e Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Mon, 20 Jul 2026 10:36:59 +0200 Subject: [PATCH] Implement handling of `invalid` request markers in `create_user` as well. That is, make sure that the field does not end up in user DB and that requests detected as invalid during signup/renew are also rejected if handled manually by site admins like they are by the janitor. Add a couple of unit tests to verify `invalid` is honoured for new accounts and for account renewals, while also assuring that valid requests still work and don't result in an (empty list) `invalid` field in the user DB. --- mig/shared/useradm.py | 22 ++++++++++-- tests/test_mig_shared_useradm.py | 61 ++++++++++++++++++++++++++++++++ 2 files changed, 80 insertions(+), 3 deletions(-) diff --git a/mig/shared/useradm.py b/mig/shared/useradm.py index da74daab5..dc4b79ab3 100644 --- a/mig/shared/useradm.py +++ b/mig/shared/useradm.py @@ -476,7 +476,7 @@ def verify_user_peers(configuration, db_path, client_id, user, now, verify_peer, def create_user_in_db(configuration, db_path, client_id, user, now, authorized, - reset_token, reset_auth_type, pw_match, + reset_token, reset_auth_type, pw_match, invalid, accepted_peer_list, force, verbose, ask_renew, default_renew, do_lock, from_edit_user, ask_change_pw, auto_create_db, create_backup): @@ -547,6 +547,17 @@ def create_user_in_db(configuration, db_path, client_id, user, now, authorized, raise Exception( 'A conflicting user with alias %s already exists' % alias) + # Reject all other obviously invalid requests + if invalid: + if do_lock: + unlock_user_db(flock) + _logger.warning("%r requested account with invalid values: %s" + % (client_id, ', '.join(invalid))) + if verbose: + print("User requested account with invalid values") + err = "Cannot create/renew user account with invalid values!" + raise Exception(err) + if client_id not in user_db: _logger.debug('add new user %r in user DB' % client_id) user['created'] = now @@ -1095,6 +1106,11 @@ def create_user(user, conf_path, db_path, force=False, verbose=False, pw_match = user['pw_match'] # Always remove any pw_match fields before DB insert del user['pw_match'] + invalid = False + if 'invalid' in user: + invalid = user['invalid'] + # Always remove any invalid markers before DB insert + del user['invalid'] _logger.info('trying to create or renew user %r' % client_id) if verbose: @@ -1123,8 +1139,8 @@ def create_user(user, conf_path, db_path, force=False, verbose=False, created = create_user_in_db(configuration, db_path, client_id, user, now, authorized, reset_token, reset_auth_type, - pw_match, accepted_peer_list, force, verbose, - ask_renew, default_renew, do_lock, + pw_match, invalid, accepted_peer_list, force, + verbose, ask_renew, default_renew, do_lock, from_edit_user, ask_change_pw, auto_create_db, create_backup) # Mark user updated for all logins diff --git a/tests/test_mig_shared_useradm.py b/tests/test_mig_shared_useradm.py index d3c30145f..a101ce909 100644 --- a/tests/test_mig_shared_useradm.py +++ b/tests/test_mig_shared_useradm.py @@ -431,6 +431,67 @@ def test_user_creation_with_id_collission_fails(self): ask_renew=False, ) + def test_user_creation_with_invalid_marker_fails(self): + 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 + # NOTE: inject invalid marker to prevent create + user_dict["invalid"] = ["invalid request detected"] + + 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('invalid values' in msg for msg in + log_capture.output)) + + def test_user_creation_with_invalid_markers_fails_renewal(self): + 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: + created = create_user( + user_dict, self.configuration, keyword_auto, default_renew=True + ) + except Exception: + self.assertFalse(True, "should not be reached") + + # Double check that valid user was created without 'invalid' field + self.assertTrue(created) + self.assertNotIn('invalid', created) + + # NOTE: inject invalid marker to prevent renew + user_dict["invalid"] = ["invalid request detected"] + 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('invalid values' in msg for msg in + log_capture.output)) + class MigSharedUseradm__assure_current_htaccess(MigTestCase): """Coverage of useradm behaviours around htaccess."""