diff --git a/mig/install/MiGserver-template.conf b/mig/install/MiGserver-template.conf index 650a8ed7d..3aaa9a777 100644 --- a/mig/install/MiGserver-template.conf +++ b/mig/install/MiGserver-template.conf @@ -523,6 +523,8 @@ update_interval = __ACCOUNTING_UPDATE_INTERVAL__ # Web site appearance # Whether to use Python 3 for all Python invocations prefer_python3 = __PREFER_PYTHON3__ +# Which user home dir format to use (default is X509; only experimental UUID) +user_id_format = __USER_ID_FORMAT__ # Dynamic entry page to pick user default with fallback to site landing page autolaunch_page = __AUTOLAUNCH_PAGE__ # Entry page if not explictly provided or overriden by user diff --git a/mig/install/generateconfs.py b/mig/install/generateconfs.py index 3f50354e9..68049e42a 100755 --- a/mig/install/generateconfs.py +++ b/mig/install/generateconfs.py @@ -174,6 +174,7 @@ def main(argv, _generate_confs=generate_confs, _print=print): 'listen_clause', 'serveralias_clause', 'distro', + 'user_id_format', 'autolaunch_page', 'landing_page', 'skin', diff --git a/mig/shared/accountstate.py b/mig/shared/accountstate.py index 8cf30ff51..49ffd9683 100644 --- a/mig/shared/accountstate.py +++ b/mig/shared/accountstate.py @@ -4,7 +4,7 @@ # --- BEGIN_HEADER --- # # accountstate - various user account state helpers -# Copyright (C) 2003-2025 The MiG Project by the Science HPC Center at UCPH +# Copyright (C) 2003-2026 The MiG Project by the Science HPC Center at UCPH # # This file is part of MiG. # @@ -453,17 +453,43 @@ def check_account_accessible(configuration, username, proto, environ=None, # Use client_id_dir to make it work even if already expanded home_dir = os.path.join(configuration.user_home, client_id_dir(client_id)) + # Expand to actual home dir and proceed from there + expanded_home = os.path.realpath(home_dir) if configuration.site_user_id_format == X509_USER_ID_FORMAT: - # Expand to actual home dir - expanded_home = os.path.realpath(home_dir) + client_dir = os.path.basename(expanded_home) elif configuration.site_user_id_format == UUID_USER_ID_FORMAT: - # Only follow first username -> client_id_dir symlink with UUID - expanded_home = os.readlink(home_dir) + # Traverse home_dir links to X509 dir name linking to expanded_home + next_link = home_dir + client_dir = None + expansions, max_expansions = 0, 3 + while os.path.islink(next_link) and expansions < max_expansions: + _logger.debug("compare next link %s to expanded %s" % + (next_link, expanded_home)) + if os.readlink(next_link) == expanded_home: + # Next link is the wanted X509 link to expanded UUID home + _logger.debug("found %s account ID dir: %s -> %s" % + (client_id, next_link, expanded_home)) + client_dir = os.path.basename(next_link) + break + else: + # follow next_link and prepend user_home if relative + next_link = os.readlink(next_link) + max_expansions += 1 + if os.path.basename(next_link) == next_link: + next_link = os.path.join(configuration.user_home, + next_link) + if client_dir is None: + _logger.debug("link traversal for %s ended at %s" % + (client_id, next_link)) + _logger.warning("failed to lookup home and ID for %s" % + username) + return False else: raise ValueError("invalid user ID format requested: %s" % configuration.site_user_id_format) - real_id = os.path.basename(expanded_home) - client_id = client_dir_id(real_id) + + # Now we can extract the desired X509 user ID + client_id = client_dir_id(client_dir) (account_accessible, account_status, _) = check_account_status( configuration, client_id) diff --git a/mig/shared/griddaemons/login.py b/mig/shared/griddaemons/login.py index 4caf98413..0d472f3d8 100644 --- a/mig/shared/griddaemons/login.py +++ b/mig/shared/griddaemons/login.py @@ -4,7 +4,7 @@ # --- BEGIN_HEADER --- # # login - grid daemon login helper functions -# Copyright (C) 2003-2025 The MiG Project by the Science HPC Center at UCPH +# Copyright (C) 2003-2026 The MiG Project by the Science HPC Center at UCPH # # This file is part of MiG. # @@ -47,8 +47,7 @@ https_authkeys, get_authkeys, ssh_authpasswords, davs_authpasswords, \ ftps_authpasswords, https_authpasswords, get_authpasswords, \ ssh_authdigests, davs_authdigests, ftps_authdigests, https_authdigests, \ - generate_password_hash, generate_password_digest, load_user_dict, \ - lookup_client_id + generate_password_hash, generate_password_digest, load_user_dict from mig.shared.validstring import possible_sharelink_id, possible_job_id, \ possible_jupyter_mount_id @@ -636,8 +635,7 @@ def refresh_job_creds(configuration, protocol, username): sessionid = username job_dict = unpickle(link_path, logger) - # We only allow connections from executing jobs that - # has a public key + # We only allow connections from executing jobs with a public key if job_dict is not None and isinstance(job_dict, dict) and \ 'STATUS' in job_dict and \ job_dict['STATUS'] == 'EXECUTING' and \ @@ -852,15 +850,15 @@ def refresh_jupyter_creds(configuration, protocol, username): logger.error("invalid protocol: %s" % protocol) return (conf, active_jupyter_creds) if not possible_jupyter_mount_id(configuration, username): - # logger.debug("ruled out %s as a possible jupyter_mount ID" \ - # % username) + # logger.debug("ruled out %s as a possible jupyter_mount ID" % + # username) return (conf, active_jupyter_creds) logger.info("Getting active jupyter mount creds") link_path = os.path.join(configuration.sessid_to_jupyter_mount_link_home, "%s" % username + ".jupyter_mount") - # logger.debug("jupyter linkpath: %s jupyter path exists: %s" % \ - # (os.path.islink(link_path), os.path.exists(link_path))) + # logger.debug("jupyter linkpath: %s jupyter path exists: %s" % + # (os.path.islink(link_path), os.path.exists(link_path))) jupyter_dict, sessionid = None, None if os.path.islink(link_path) and os.path.exists(link_path): diff --git a/mig/shared/install.py b/mig/shared/install.py index 7184221f0..20f7c381e 100644 --- a/mig/shared/install.py +++ b/mig/shared/install.py @@ -500,6 +500,7 @@ def generate_confs( listen_clause='#Listen', serveralias_clause='#ServerAlias', distro='Debian', + user_id_format='X509', autolaunch_page=None, landing_page=None, skin='migrid-basic', @@ -833,6 +834,7 @@ def _generate_confs_prepare( listen_clause, serveralias_clause, distro, + user_id_format, autolaunch_page, landing_page, skin, @@ -1094,6 +1096,7 @@ def _generate_confs_prepare( user_dict['__LISTEN_CLAUSE__'] = listen_clause user_dict['__SERVERALIAS_CLAUSE__'] = serveralias_clause user_dict['__DISTRO__'] = distro + user_dict['__USER_ID_FORMAT__'] = user_id_format user_dict['__SKIN__'] = skin user_dict['__TITLE__'] = title user_dict['__SHORT_TITLE__'] = short_title diff --git a/mig/shared/useradm.py b/mig/shared/useradm.py index 44549c632..c5302f774 100644 --- a/mig/shared/useradm.py +++ b/mig/shared/useradm.py @@ -284,9 +284,11 @@ def _get_required_user_alias_links(configuration, real_dir, link_dir): return alias_links -def lookup_client_id(configuration, user_id): - """Lookup the X509 format client_id linked to user_id based on the helper - symlink in user_id_alias_dir in configured mig_system_run location. +def lookup_client_id_from_uuid(configuration, user_id): + """Lookup the X509 format client_id linked to UUID user_id based on the + helper symlink in user_id_alias_dir in configured mig_system_run location. + Automatic fallback to reverse lookup based on symlinks in user_home if + helper symlink is missing with write-through to mig_system_run cache. """ _logger = configuration.logger alias_link = os.path.join(configuration.mig_system_run, user_id_alias_dir, @@ -294,7 +296,8 @@ def lookup_client_id(configuration, user_id): _logger.debug("looking for alias link %r" % alias_link) # NOTE: use islink rather than exists here because it is a dead link if so if os.path.islink(alias_link): - target_path = os.path.realpath(alias_link) + # Only follow first chain even if mix of UUID and X509 links overlap + target_path = os.readlink(alias_link) client_dir = os.path.basename(target_path) _logger.debug("found linked alias %r for %r" % (client_dir, user_id)) else: @@ -302,6 +305,9 @@ def lookup_client_id(configuration, user_id): # mig_system_run, which is only scratch space. client_dir = False for name in listdir(configuration.user_home): + # Skip if not a DN-dir name to avoid hit e.g. on short alias links + if '=' not in name or '+' not in name: + continue link_path = os.path.join(configuration.user_home, name) if os.path.islink(link_path): target = os.path.basename(os.path.realpath(link_path)) @@ -312,8 +318,9 @@ def lookup_client_id(configuration, user_id): makedirs_rec(os.path.dirname(alias_link), configuration) make_symlink(client_dir, alias_link, _logger) break + if not client_dir: - _logger.warning("found no alias %r" % user_id) + _logger.error("found no alias %r" % user_id) return user_id client_id = client_dir_id(client_dir) return client_id @@ -1875,7 +1882,8 @@ def get_any_oid_user_dn(configuration, raw_login, (native_dir, raw_login)) if configuration.site_user_id_format == UUID_USER_ID_FORMAT: user_id = native_dir - distinguished_name = lookup_client_id(configuration, user_id) + distinguished_name = lookup_client_id_from_uuid(configuration, + user_id) elif configuration.site_user_id_format == X509_USER_ID_FORMAT: distinguished_name = client_dir_id(native_dir) else: diff --git a/tests/fixture/confs-stdlocal/MiGserver.conf b/tests/fixture/confs-stdlocal/MiGserver.conf index 1e5814010..05b1a89a4 100644 --- a/tests/fixture/confs-stdlocal/MiGserver.conf +++ b/tests/fixture/confs-stdlocal/MiGserver.conf @@ -523,6 +523,8 @@ update_interval = 3600 # Web site appearance # Whether to use Python 3 for all Python invocations prefer_python3 = False +# Which user home dir format to use (default is X509; only experimental UUID) +user_id_format = X509 # Dynamic entry page to pick user default with fallback to site landing page autolaunch_page = /wsgi-bin/autolaunch.py # Entry page if not explictly provided or overriden by user diff --git a/tests/test_mig_shared_accountstate.py b/tests/test_mig_shared_accountstate.py index 56f95378d..cc6eb2dd5 100644 --- a/tests/test_mig_shared_accountstate.py +++ b/tests/test_mig_shared_accountstate.py @@ -27,6 +27,7 @@ """Unit tests for the mig shared accountstate module""" +import base64 import os import time import unittest @@ -49,20 +50,24 @@ ) # Imports required for the unit test wrapping -from mig.shared.base import client_id_dir +from mig.shared.base import client_id_dir, distinguished_name_to_user from mig.shared.defaults import ( AUTH_CERTIFICATE, AUTH_GENERIC, AUTH_OPENID_CONNECT, AUTH_OPENID_V2, DEFAULT_USER_ID_FORMAT, + UUID_USER_ID_FORMAT, + cert_auto_extend_days, + expire_marks_dir, + keyword_auto, cert_auto_extend_days, expire_marks_dir, oid_auto_extend_days, oidc_auto_extend_days, status_marks_dir, ) -from mig.shared.useradm import _ensure_dirs_needed_for_userdb +from mig.shared.useradm import _ensure_dirs_needed_for_userdb, create_user from mig.shared.userdb import load_user_dict, update_user_dict # Imports required for the unit tests themselves @@ -87,9 +92,40 @@ ) TEST_USER_EMAIL = TEST_USER_DN.split("/emailAddress=", 1)[-1] TEST_USER_DIR = client_id_dir(TEST_USER_DN) +# IMPORTANT: TEST_USER_UUID is *generated* as base16 of DN here OTHER_USER_DIR = client_id_dir(OTHER_USER_DN) +def _provision_uuid_test_user(configuration, client_id, client_overrides=None): + """Helper to provision test users when UUID format is used""" + # TODO: merge something like this version into standard _provision_test_user? + # IMPORTANT: we need to use explicit create_user here for UUID format! + user_dict = distinguished_name_to_user(client_id) + # NOTE: generate unique and short id based on id to avoid test collisions + user_dict["unique_id"] = base64.b16encode(client_id.encode("utf8")).decode( + "ascii" + ) + user_dict["short_id"] = base64.b16encode( + user_dict["email"].encode("utf8") + ).decode("ascii") + user_dict["comment"] = "This is the user account comment" + user_dict["locality"] = "" + user_dict["organizational_unit"] = "" + user_dict["password"] = "" + user_dict["password_hash"] = "" + if client_overrides is not None: + user_dict.update(client_overrides) + + create_user( + user_dict, + configuration, + keyword_auto, + default_renew=True, + ask_renew=False, + ) + return user_dict + + # TODO: test gdp usernames, too @@ -1651,6 +1687,480 @@ def test_check_account_accessible_openid_expire_disabled_on_email_alias( self.assertTrue(accessible) +class TestMigSharedAccountstate__check_account_accessible_uuid_user_id( + MigTestCase, UserAssertMixin +): + """Coverage of accountstate check_account_accessible function with UUID user format.""" + + def _provide_configuration(self): + return "testconfig" + + def before_each(self): + """Set up test environment for check_account_accessible tests.""" + configuration = self.configuration + # Force UUID user ID format + configuration.site_user_id_format = UUID_USER_ID_FORMAT + configuration.site_enable_sharelinks = True + configuration.site_enable_jobs = True + configuration.site_enable_jupyter = True + # Ensure necessary directories exist + ensure_dirs_exist(configuration.mig_system_files) + ensure_dirs_exist( + os.path.join(configuration.mig_system_run, expire_marks_dir) + ) + ensure_dirs_exist( + os.path.join(configuration.mig_system_run, status_marks_dir) + ) + # Set up user DB and keep paths for later use + _ensure_dirs_needed_for_userdb(configuration) + self.expected_user_db_home = os.path.normpath( + configuration.user_db_home + ) + self.expected_user_db_file = os.path.join( + self.expected_user_db_home, "MiG-users.db" + ) + # Provision a known test user on UUID format + self.test_user_dict = _provision_uuid_test_user( + configuration, TEST_USER_DN + ) + self.test_user_id = self.test_user_dict["unique_id"] + self.test_user_short_id = self.test_user_dict["short_id"] + + self.test_user_home_path = os.path.join( + configuration.user_home, self.test_user_id + ) + # Verify home dir is functional + self.assertTrue(os.path.isdir(self.test_user_home_path)) + # This simulates an existing short id link to X509 on to uuid dir in user_home + x509_link_in_home = os.path.join( + configuration.user_home, TEST_USER_DIR) + if not os.path.islink(x509_link_in_home): + os.symlink(self.test_user_id, x509_link_in_home) + # Verify link is functional + self.assertTrue(os.path.islink(x509_link_in_home)) + self.assertEqual( + os.path.realpath(x509_link_in_home), self.test_user_home_path + ) + email_link_in_home = os.path.join( + configuration.user_home, TEST_USER_EMAIL + ) + if not os.path.islink(email_link_in_home): + os.symlink(TEST_USER_DIR, email_link_in_home) + # Verify link is functional + self.assertTrue(os.path.islink(email_link_in_home)) + self.assertEqual( + os.path.realpath(email_link_in_home), self.test_user_home_path + ) + short_id_link_in_home = os.path.join( + configuration.user_home, self.test_user_short_id + ) + if not os.path.islink(short_id_link_in_home): + os.symlink(TEST_USER_DIR, short_id_link_in_home) + # Verify link is functional + self.assertTrue(os.path.islink(short_id_link_in_home)) + self.assertEqual( + os.path.realpath(short_id_link_in_home), self.test_user_home_path + ) + + def test_check_account_accessible_active_user_web_access_direct(self): + """Test that an active UUID user is accessible via web (non-IO).""" + configuration = self.configuration + # Ensure the user is active and not expired + user_dict = { + "distinguished_name": TEST_USER_DN, + "status": "active", + "expire": time.time() + 42 * 24 * 3600, # far future + } + update_user_dict( + self.logger, TEST_USER_DN, user_dict, self.expected_user_db_file + ) + update_account_expire_cache(configuration, user_dict) + update_account_status_cache(configuration, user_dict) + + accessible = check_account_accessible( + configuration, TEST_USER_DN, "oidc", io_login=False + ) + self.assertTrue(accessible) + + def test_check_account_accessible_active_user_io_access_direct(self): + """Test that an active UUID user is accessible via SFTP (IO).""" + configuration = self.configuration + # Ensure the user is active and not expired + user_dict = { + "distinguished_name": TEST_USER_DN, + "status": "active", + "expire": time.time() + 42 * 24 * 3600, # far future + } + update_user_dict( + self.logger, TEST_USER_DN, user_dict, self.expected_user_db_file + ) + update_account_expire_cache(configuration, user_dict) + update_account_status_cache(configuration, user_dict) + + accessible = check_account_accessible( + configuration, TEST_USER_DN, "sftp" + ) + self.assertTrue(accessible) + + def test_check_account_accessible_active_user_web_access_on_email_alias( + self, + ): + """Test that an active UUID user is accessible via web (non-IO) using email alias.""" + configuration = self.configuration + # Ensure the user is active and not expired + user_dict = { + "distinguished_name": TEST_USER_DN, + "status": "active", + "expire": time.time() + 42 * 24 * 3600, # far future + } + update_user_dict( + self.logger, TEST_USER_DN, user_dict, self.expected_user_db_file + ) + update_account_expire_cache(configuration, user_dict) + update_account_status_cache(configuration, user_dict) + + accessible = check_account_accessible( + configuration, TEST_USER_EMAIL, "oidc", io_login=False + ) + self.assertTrue(accessible) + + def test_check_account_accessible_active_user_sftp_access_on_email_alias( + self, + ): + """Test that an active UUID user is accessible via SFTP using email alias.""" + configuration = self.configuration + # Ensure the user is active and not expired + user_dict = { + "distinguished_name": TEST_USER_DN, + "status": "active", + "expire": time.time() + 42 * 24 * 3600, # far future + } + update_user_dict( + self.logger, TEST_USER_DN, user_dict, self.expected_user_db_file + ) + update_account_expire_cache(configuration, user_dict) + update_account_status_cache(configuration, user_dict) + + accessible = check_account_accessible( + configuration, TEST_USER_EMAIL, "sftp" + ) + self.assertTrue(accessible) + + def test_check_account_accessible_locked_user_web_blocked(self): + """Test that a locked UUID user is not accessible on web (non-IO).""" + configuration = self.configuration + user_dict = { + "distinguished_name": TEST_USER_DN, + "status": TEST_STATUS_LOCKED, + } + update_user_dict( + self.logger, TEST_USER_DN, user_dict, self.expected_user_db_file + ) + update_account_status_cache(configuration, user_dict) + + accessible = check_account_accessible( + configuration, TEST_USER_DN, "oidc", io_login=False + ) + self.assertFalse(accessible) + + def test_check_account_accessible_locked_user_io_blocked(self): + """Test that a locked UUID user is not accessible on SFTP (IO).""" + configuration = self.configuration + user_dict = { + "distinguished_name": TEST_USER_DN, + "status": TEST_STATUS_LOCKED, + } + update_user_dict( + self.logger, TEST_USER_DN, user_dict, self.expected_user_db_file + ) + update_account_status_cache(configuration, user_dict) + + accessible = check_account_accessible( + configuration, TEST_USER_DN, "sftp" + ) + self.assertFalse(accessible) + + def test_check_account_accessible_expired_user_blocked(self): + """Test that an expired UUID user is inaccessible on web (non-ID) login.""" + configuration = self.configuration + # Set expire to past + expire_ts = 42 + user_dict = { + "distinguished_name": TEST_USER_DN, + "status": "active", + "expire": expire_ts, + } + update_user_dict( + self.logger, TEST_USER_DN, user_dict, self.expected_user_db_file + ) + update_account_expire_cache(configuration, user_dict) + update_account_status_cache(configuration, user_dict) + + accessible = check_account_accessible( + configuration, TEST_USER_DN, "oidc", io_login=False + ) + self.assertFalse(accessible) + + def test_check_account_accessible_expired_user_sftp_access_blocked(self): + """Test that an expired UUID user is inaccessible on sftp if enenforced.""" + configuration = self.configuration + configuration.site_io_account_expire = True + # Set expire to past + expire_ts = 42 + user_dict = { + "distinguished_name": TEST_USER_DN, + "status": "active", + "expire": expire_ts, + } + update_user_dict( + self.logger, TEST_USER_DN, user_dict, self.expected_user_db_file + ) + update_account_expire_cache(configuration, user_dict) + update_account_status_cache(configuration, user_dict) + + accessible = check_account_accessible( + configuration, TEST_USER_DN, "sftp" + ) + self.assertFalse(accessible) + + def test_check_account_accessible_special_login_sharelink_access(self): + """Test that a UUID sharelink is detected as special login and accessible.""" + configuration = self.configuration + # Create a dummy RW sharelink to user home + sharelink_path = os.path.join( + configuration.sharelink_home, "read-write", TEST_RW_SHARE_ID + ) + sharelink_target = os.path.join(configuration.user_home, TEST_USER_DIR) + ensure_dirs_exist(sharelink_target) + ensure_dirs_exist(os.path.dirname(sharelink_path)) + os.symlink(sharelink_target, sharelink_path) + + accessible = check_account_accessible( + configuration, TEST_RW_SHARE_ID, "sftp" + ) + self.assertTrue(accessible) + + def test_check_account_accessible_special_login_job_access(self): + """Test that a UUID job ID is detected as special login and accessible.""" + configuration = self.configuration + configuration.site_enable_jobs = True + # Create a dummy job link to mrsl files entry + job_target_path = os.path.join( + configuration.mrsl_files_dir, TEST_JOB_ID + ) + ensure_dirs_exist(job_target_path) + job_link_path = os.path.join( + configuration.sessid_to_mrsl_link_home, TEST_JOB_ID + ".mRSL" + ) + ensure_dirs_exist(os.path.dirname(job_link_path)) + os.symlink(job_target_path, job_link_path) + + accessible = check_account_accessible( + configuration, TEST_JOB_ID, "sftp" + ) + self.assertTrue(accessible) + + def test_check_account_accessible_special_login_jupyter_access(self): + """Test that a UUID jupyter mount ID is detected as special login and accessible.""" + configuration = self.configuration + configuration.site_enable_jupyter = True + # Create a dummy jupyter‑mount link to user home + jupyter_link_path = os.path.join( + configuration.sessid_to_jupyter_mount_link_home, + TEST_JUPYTER_SESSION_ID, + ) + jupyter_target = os.path.join(configuration.user_home, TEST_USER_DIR) + ensure_dirs_exist(jupyter_target) + ensure_dirs_exist(os.path.dirname(jupyter_link_path)) + os.symlink(jupyter_target, jupyter_link_path) + + accessible = check_account_accessible( + configuration, TEST_JUPYTER_SESSION_ID, "sftp" + ) + self.assertTrue(accessible) + + def test_check_account_accessible_missing_user_web_rejected(self): + """Test that check_account_accessible rejects a missing UUID user on web (non-IO).""" + configuration = self.configuration + accessible = check_account_accessible( + configuration, OTHER_USER_DN, "oidc", io_login=False + ) + self.assertFalse(accessible) + + def test_check_account_accessible_missing_user_io_rejected(self): + """Test that check_account_accessible rejects a missing UUID user on SFTP (IO).""" + configuration = self.configuration + accessible = check_account_accessible( + configuration, OTHER_USER_DN, "sftp" + ) + self.assertFalse(accessible) + + def test_check_account_accessible_io_expire_disabled_on_email_alias(self): + """Test that UUID account remains accessible on SFTP (IO) with email when expire is not enforced.""" + configuration = self.configuration + configuration.site_io_account_expire = False + # Set expire to past + expire_ts = 42 + user_dict = { + "distinguished_name": TEST_USER_DN, + "status": "active", + "expire": expire_ts, + } + update_user_dict( + self.logger, TEST_USER_DN, user_dict, self.expected_user_db_file + ) + update_account_expire_cache(configuration, user_dict) + update_account_status_cache(configuration, user_dict) + + # NOTE: should succeed because IO expire is disabled + accessible = check_account_accessible( + configuration, TEST_USER_EMAIL, "sftp", io_login=True + ) + self.assertTrue(accessible) + + def test_check_account_accessible_io_expire_disabled_on_short_id(self): + """Test that UUID account remains accessible on SFTP (IO) with short id when expire is not enforced.""" + configuration = self.configuration + configuration.site_io_account_expire = False + # Set expire to past + expire_ts = 42 + user_dict = { + "distinguished_name": TEST_USER_DN, + "status": "active", + "expire": expire_ts, + } + update_user_dict( + self.logger, TEST_USER_DN, user_dict, self.expected_user_db_file + ) + update_account_expire_cache(configuration, user_dict) + update_account_status_cache(configuration, user_dict) + + # NOTE: should succeed because IO expire is disabled + accessible = check_account_accessible( + configuration, self.test_user_short_id, "sftp", io_login=True + ) + self.assertTrue(accessible) + + def test_check_account_accessible_io_expire_disabled_direct(self): + """Test that UUID account remains accessible on SFTP (IO) with DN when expire is not enforced.""" + configuration = self.configuration + configuration.site_io_account_expire = False + # Set expire to past + expire_ts = 42 + user_dict = { + "distinguished_name": TEST_USER_DN, + "status": "active", + "expire": expire_ts, + } + update_user_dict( + self.logger, TEST_USER_DN, user_dict, self.expected_user_db_file + ) + update_account_expire_cache(configuration, user_dict) + update_account_status_cache(configuration, user_dict) + + # NOTE: should succeed because IO expire is disabled + accessible = check_account_accessible( + configuration, TEST_USER_DN, "sftp", io_login=True + ) + self.assertTrue(accessible) + + def test_check_account_accessible_openid_expire_disabled_on_email_alias( + self, + ): + """Test that UUID account remains accessible on OpenID (non-IO) with email when expire is not enforced.""" + configuration = self.configuration + configuration.user_openid_enforce_expire = False + # Set expire to past + expire_ts = 42 + user_dict = { + "distinguished_name": TEST_USER_DN, + "status": "active", + "expire": expire_ts, + } + update_user_dict( + self.logger, TEST_USER_DN, user_dict, self.expected_user_db_file + ) + update_account_expire_cache(configuration, user_dict) + update_account_status_cache(configuration, user_dict) + + # NOTE: should succeed because OpenID expire is disabled + accessible = check_account_accessible( + configuration, TEST_USER_EMAIL, "oid", io_login=False + ) + self.assertTrue(accessible) + + def test_check_account_accessible_openid_expire_disabled_on_short_id(self): + """Test that UUID account remains accessible on OpenID (non-IO) with short id when expire is not enforced.""" + configuration = self.configuration + configuration.user_openid_enforce_expire = False + # Set expire to past + expire_ts = 42 + user_dict = { + "distinguished_name": TEST_USER_DN, + "status": "active", + "expire": expire_ts, + } + update_user_dict( + self.logger, TEST_USER_DN, user_dict, self.expected_user_db_file + ) + update_account_expire_cache(configuration, user_dict) + update_account_status_cache(configuration, user_dict) + + # NOTE: should succeed because OpenID expire is disabled + # NOTE: short_id can be tested here because we explicitly link it + accessible = check_account_accessible( + configuration, self.test_user_short_id, "oid", io_login=False + ) + self.assertTrue(accessible) + + def test_check_account_accessible_openid_expire_disabled_direct(self): + """Test that UUID account remains accessible on OpenID (non-IO) with DN when expire is not enforced.""" + configuration = self.configuration + configuration.user_openid_enforce_expire = False + # Set expire to past + expire_ts = 42 + user_dict = { + "distinguished_name": TEST_USER_DN, + "status": "active", + "expire": expire_ts, + } + update_user_dict( + self.logger, TEST_USER_DN, user_dict, self.expected_user_db_file + ) + update_account_expire_cache(configuration, user_dict) + update_account_status_cache(configuration, user_dict) + + # NOTE: should succeed because OpenID expire is disabled + accessible = check_account_accessible( + configuration, TEST_USER_DN, "oid", io_login=False + ) + self.assertTrue(accessible) + + @unittest.skip("TODO: enable login with plain UUID and enable test?") + def test_check_account_accessible_openid_expire_disabled_on_uuid(self): + """Test that UUID account remains accessible on OpenID (non-IO) with UUID when expire is not enforced.""" + configuration = self.configuration + configuration.user_openid_enforce_expire = False + # Set expire to past + expire_ts = 42 + user_dict = { + "distinguished_name": TEST_USER_DN, + "status": "active", + "expire": expire_ts, + } + update_user_dict( + self.logger, TEST_USER_DN, user_dict, self.expected_user_db_file + ) + update_account_expire_cache(configuration, user_dict) + update_account_status_cache(configuration, user_dict) + + # NOTE: should succeed because OpenID expire is disabled + accessible = check_account_accessible( + configuration, self.test_user_id, "oid", io_login=False + ) + self.assertTrue(accessible) + + class TestMigSharedAccountstate__check_update_account_expire( MigTestCase, UserAssertMixin ): diff --git a/tests/test_mig_shared_griddaemons_login.py b/tests/test_mig_shared_griddaemons_login.py index ce3afc057..833cff92a 100644 --- a/tests/test_mig_shared_griddaemons_login.py +++ b/tests/test_mig_shared_griddaemons_login.py @@ -27,6 +27,7 @@ """Unit tests for the griddaemons login helper functions""" +import binascii import os import pickle import socket @@ -35,11 +36,14 @@ import unittest # Imports required for the unit test wrapping -from mig.shared.base import client_id_dir +from mig.shared.base import client_id_dir, distinguished_name_to_user from mig.shared.defaults import ( READ_ONLY_ACCESS, READ_WRITE_ACCESS, + UUID_USER_ID_FORMAT, WRITE_ONLY_ACCESS, + X509_USER_ID_FORMAT, + keyword_auto, ) # Imports of the code under test @@ -62,7 +66,7 @@ ) # More imports required for the unit test wrapping -from mig.shared.useradm import _ensure_dirs_needed_for_userdb +from mig.shared.useradm import _ensure_dirs_needed_for_userdb, create_user # Imports required for the unit tests themselves from tests.support import ( @@ -101,6 +105,36 @@ ) +def _provision_uuid_test_user(configuration, client_id, client_overrides=None): + """Helper to provision test users when UUID format is used""" + # TODO: merge something like this version into standard _provision_test_user? + # IMPORTANT: we need to use explicit create_user here for UUID format! + user_dict = distinguished_name_to_user(client_id) + # NOTE: generate unique and short id based on id to avoid test collisions + user_dict["unique_id"] = binascii.hexlify(client_id.encode("utf8")).decode( + "ascii" + ) + user_dict["short_id"] = binascii.hexlify( + user_dict["email"].encode("utf8") + ).decode("ascii") + user_dict["comment"] = "This is the user account comment" + user_dict["locality"] = "" + user_dict["organizational_unit"] = "" + user_dict["password"] = "" + user_dict["password_hash"] = "" + if client_overrides is not None: + user_dict.update(client_overrides) + + create_user( + user_dict, + configuration, + keyword_auto, + default_renew=True, + ask_renew=False, + ) + return user_dict + + def _prepare_auth_files(home_path, auth_protos=None): """Helper to create helper auth files for on eor more auth_protos. If None is passed ssh, ftps and davs auth files will be made. @@ -554,6 +588,9 @@ def _provide_configuration(self): def before_each(self): """Set up test configuration and reset state before each test""" + # Force X509 user id format + self.configuration.site_user_id_format = X509_USER_ID_FORMAT + # Ensure required directories exist _ensure_dirs_needed_for_userdb(self.configuration) ensure_dirs_exist(self.configuration.sharelink_home) @@ -639,6 +676,119 @@ def test_get_creds_changes_no_changes(self): self.assertEqual(len(changed_paths), 0) +class MigSharedGriddaemonsLogin__get_creds_changes_uuid_user_id(MigTestCase): + """Unit tests for griddaemons login get_creds_changes function with UUID users""" + + def _provide_configuration(self): + """Return a test configuration instance""" + return "testconfig" + + def before_each(self): + """Set up test configuration and reset state before each test""" + # Force UUID user id format + self.configuration.site_user_id_format = UUID_USER_ID_FORMAT + # Ensure required directories exist + ensure_dirs_exist(self.configuration.mig_system_files) + _ensure_dirs_needed_for_userdb(self.configuration) + ensure_dirs_exist(self.configuration.sharelink_home) + + self.configuration.daemon_conf = {} + self.configuration.daemon_conf["time_stamp"] = 0 + self.configuration.daemon_conf["users"] = [] + self.configuration.daemon_conf["allow_publickey"] = True + self.configuration.daemon_conf["allow_password"] = True + # TODO: enable and test unsafe digest auth, too? + # self.configuration.daemon_conf['allow_digest'] = True + self.configuration.daemon_conf["allow_digest"] = False + + user_dict = _provision_uuid_test_user(self.configuration, TEST_USER_DN) + self.test_user_id = user_id = user_dict["unique_id"] + client_dir = client_id_dir(TEST_USER_DN) + self.test_user_home = os.path.join( + self.configuration.user_home, user_id + ) + self.test_user_dir = os.path.basename(self.test_user_home) + self.test_user_home_x509 = os.path.join( + self.configuration.user_home, client_dir + ) + self.test_user_dir_x509 = os.path.basename(self.test_user_home_x509) + # Make sure X509 link are provisioned as well + if not os.path.islink(self.test_user_home_x509): + os.symlink(self.test_user_home, self.test_user_home_x509) + + auth_files = _prepare_auth_files(self.test_user_home, ["ssh"]) + self.auth_keys_path, self.auth_passwords_path = auth_files + self.auth_digests_path = None + + def test_get_creds_changes_detects_new_files(self): + """Verify that new credential files are detected as changes""" + daemon_conf = self.configuration.daemon_conf + + # Create a dummy user with a last_update in the past + past_timestamp = time.time() - 3600 + # NOTE: we use x509 alias over uuid home as in real runs + dummy_user = Login( + configuration=self.configuration, + username=TEST_USER_SHORT_ID, + home=self.test_user_home_x509, + password=TEST_USER_PW_HASH, + digest=None, + public_key=TEST_USER_PUB_KEY, + chroot=True, + access=None, + ip_addr=None, + user_dict=None, + ) + dummy_user.last_update = past_timestamp + daemon_conf["users"].append(dummy_user) + + changed_paths = get_creds_changes( + daemon_conf, + "user", + self.auth_keys_path, + self.auth_passwords_path, + self.auth_digests_path, + ) + + self.assertIn(self.auth_keys_path, changed_paths) + self.assertIn(self.auth_passwords_path, changed_paths) + # self.assertIn(self.auth_digests_path, changed_paths) + + def test_get_creds_changes_no_changes(self): + """Verify that unchanged credential files return an empty list""" + daemon_conf = self.configuration.daemon_conf + + # Set the file modification times to now + current_time = time.time() + + # Create a dummy user with last_update matching the file mtime + # NOTE: we use x509 alias over uuid home as in real runs + dummy_user = Login( + configuration=self.configuration, + username=TEST_USER_SHORT_ID, + home=self.test_user_home_x509, + password=TEST_USER_PW_HASH, + digest=None, + public_key=TEST_USER_PUB_KEY, + chroot=True, + access=None, + ip_addr=None, + user_dict=None, + ) + dummy_user.last_update = current_time + daemon_conf["users"].append(dummy_user) + + changed_paths = get_creds_changes( + daemon_conf, + TEST_USER_SHORT_ID, + self.auth_keys_path, + self.auth_passwords_path, + self.auth_digests_path, + ) + + self.assertEqual(len(changed_paths), 0) + + class MigSharedGriddaemonsLogin__get_job_changes(MigTestCase): """Unit tests for griddaemons login get_job_changes function""" @@ -1713,6 +1863,8 @@ def _provide_configuration(self): def before_each(self): """Set up test configuration and reset state before each test.""" + # Force X509 user id format + self.configuration.site_user_id_format = X509_USER_ID_FORMAT # Ensure required directories exist _ensure_dirs_needed_for_userdb(self.configuration) self.expected_user_db_home = os.path.normpath( @@ -1908,6 +2060,239 @@ def test_refresh_user_creds_invalid_protocol(self): self.assertEqual(len(updated_conf["users"]), 0) +# TODO: merge X509 and UUID versions? + + +class MigSharedGriddaemonsLogin__refresh_user_creds_uuid_user_id( + MigTestCase, UserAssertMixin +): + """Unit tests for the griddaemons login refresh_user_creds helper with UUID users.""" + + def _provide_configuration(self): + """Return a test configuration instance.""" + return "testconfig" + + def before_each(self): + """Set up test configuration and reset state before each test.""" + # Force UUID user id format + self.configuration.site_user_id_format = UUID_USER_ID_FORMAT + # Ensure required directories exist + ensure_dirs_exist(self.configuration.mig_system_files) + _ensure_dirs_needed_for_userdb(self.configuration) + self.expected_user_db_home = os.path.normpath( + self.configuration.user_db_home + ) + self.expected_user_db_file = os.path.join( + self.expected_user_db_home, "MiG-users.db" + ) + # NOTE: we need to set a password_hash for https test to work with + overrides = {} + overrides["password_hash"] = TEST_USER_PW_HASH + + user_dict = _provision_uuid_test_user( + self.configuration, TEST_USER_DN, overrides + ) + self.test_user_id = user_id = user_dict["unique_id"] + client_dir = client_id_dir(TEST_USER_DN) + self.test_user_home = os.path.join( + self.configuration.user_home, user_id + ) + self.test_user_dir = os.path.basename(self.test_user_home) + self.test_user_home_x509 = os.path.join( + self.configuration.user_home, client_dir + ) + self.test_user_dir_x509 = os.path.basename(self.test_user_home_x509) + # Make sure X509 and alias links are provisioned as well + if not os.path.islink(self.test_user_home_x509): + os.symlink(self.test_user_home, self.test_user_home_x509) + alias_link_path = os.path.join( + self.configuration.user_home, TEST_USER_EMAIL + ) + if not os.path.islink(alias_link_path): + os.symlink(self.test_user_home_x509, alias_link_path) + + ALIAS_FIELD = "email" + self.configuration.user_sftp_alias = ALIAS_FIELD + self.configuration.user_ftps_alias = ALIAS_FIELD + self.configuration.user_davs_alias = ALIAS_FIELD + + # Common daemon configuration + self.configuration.daemon_conf = {} + self.configuration.daemon_conf["time_stamp"] = 0 + self.configuration.daemon_conf["users"] = [] + self.configuration.daemon_conf["root_dir"] = ( + self.configuration.user_home + ) + self.configuration.daemon_conf["db_path"] = self.expected_user_db_file + self.configuration.daemon_conf["allow_publickey"] = True + self.configuration.daemon_conf["allow_password"] = True + self.configuration.daemon_conf["allow_digest"] = False + self.configuration.daemon_conf["user_alias"] = ALIAS_FIELD + + def test_refresh_user_creds_ssh_protocol(self): + """Test refreshing user credentials for SSH protocol.""" + username = TEST_USER_EMAIL + _prepare_auth_files(self.test_user_home, ["ssh"]) + + # Call the function under test + updated_conf, changed_users = refresh_user_creds( + configuration=self.configuration, protocol="ssh", username=username + ) + + # The user should be in the changed list + self.assertIn(username, changed_users) + + # Verify that Login objects were added to users + user_logins = [ + obj + for obj in updated_conf["users"] + if obj.username == TEST_USER_DN or obj.username == username + ] + # We expect at least one login (the main username) and possibly aliases + self.assertGreaterEqual(len(user_logins), 1) + + # Check that at least one login has the correct home directory + # TODO: is this X509 dir what we want here or the UUID one? + home_found = any( + login.home == self.test_user_dir_x509 for login in user_logins + ) + self.assertTrue(home_found) + + def test_refresh_user_creds_davs_protocol(self): + """Test refreshing user credentials for DAVS protocol.""" + username = TEST_USER_EMAIL + _prepare_auth_files(self.test_user_home, ["davs"]) + + # Call the function under test + updated_conf, changed_users = refresh_user_creds( + configuration=self.configuration, protocol="davs", username=username + ) + + # The user should be in the changed list + self.assertIn(username, changed_users) + + # Verify that Login objects were added to users + user_logins = [ + obj for obj in updated_conf["users"] if obj.username == username + ] + self.assertEqual(len(user_logins), 1) + # TODO: is this X509 dir what we want here or the UUID one? + self.assertEqual(user_logins[0].home, self.test_user_dir_x509) + + def test_refresh_user_creds_ftps_protocol(self): + """Test refreshing user credentials for FTPS protocol.""" + username = TEST_USER_EMAIL + _prepare_auth_files(self.test_user_home, ["ftps"]) + + # Call the function under test + updated_conf, changed_users = refresh_user_creds( + configuration=self.configuration, protocol="ftps", username=username + ) + + # The user should be in the changed list + self.assertIn(username, changed_users) + + # Verify that Login objects were added to users + user_logins = [ + obj for obj in updated_conf["users"] if obj.username == username + ] + self.assertEqual(len(user_logins), 1) + # TODO: is this X509 dir what we want here or the UUID one? + self.assertEqual(user_logins[0].home, self.test_user_dir_x509) + + def test_refresh_user_creds_https_protocol(self): + """Test refreshing user credentials for HTTPS protocol (uses user DB).""" + username = TEST_USER_EMAIL + + # Call the function under test + updated_conf, changed_users = refresh_user_creds( + configuration=self.configuration, + protocol="https", + username=username, + ) + + # The user alias should be in the changed list + self.assertIn(TEST_USER_EMAIL, changed_users) + + # Verify that Login objects were added to users for the username and its aliases + user_logins = [ + obj + for obj in updated_conf["users"] + if obj.username == TEST_USER_DN or obj.username == username + ] + # We expect at least the main username and possibly aliases + self.assertGreaterEqual(len(user_logins), 1) + + # Check that at least one login has the correct home directory + # TODO: is this X509 dir what we want here or the UUID one? + home_found = any( + login.home == self.test_user_dir_x509 for login in user_logins + ) + self.assertTrue(home_found) + + def test_refresh_user_creds_no_changes(self): + """Test that no changes are reported when credentials are unchanged.""" + username = TEST_USER_EMAIL + _prepare_auth_files(self.test_user_home, ["ssh"]) + + # Pre-populate the users list with a Login object that has + # last_update set to the current time (simulating no changes) + current_time = time.time() + dummy_user = Login( + configuration=self.configuration, + username=username, + home=self.test_user_home, + password=TEST_USER_PW_HASH, + digest=None, + public_key=TEST_USER_PUB_KEY, + chroot=True, + access=None, + ip_addr=None, + user_dict=None, + ) + dummy_user.last_update = current_time + self.configuration.daemon_conf["users"].append(dummy_user) + + # Call the function under test + updated_conf, changed_users = refresh_user_creds( + configuration=self.configuration, protocol="ssh", username=username + ) + + # No changes should be reported + self.assertEqual(len(changed_users), 0) + # The user list should still contain only our dummy user + self.assertEqual(len(updated_conf["users"]), 1) + self.assertEqual(updated_conf["users"][0].username, username) + + def test_refresh_user_creds_missing_user(self): + """Test that a missing user is skipped.""" + username = "nosuchuser" + + # Call the function under test + updated_conf, changed_users = refresh_user_creds( + configuration=self.configuration, protocol="ssh", username=username + ) + + # No changes should be reported because the home directory is missing + self.assertEqual(len(changed_users), 0) + self.assertEqual(len(updated_conf["users"]), 0) + + def test_refresh_user_creds_invalid_protocol(self): + """Test that an invalid protocol returns early without changes.""" + username = TEST_USER_EMAIL + + # Call the function under test with an invalid protocol + updated_conf, changed_users = refresh_user_creds( + configuration=self.configuration, + protocol="invalid", + username=username, + ) + + # No changes should be reported + self.assertEqual(len(changed_users), 0) + self.assertEqual(len(updated_conf["users"]), 0) + + class MigSharedGriddaemonsLogin__update_login_map(MigTestCase): """Unit tests for the update_login_map function from griddaemons/login.py""" @@ -2090,7 +2475,6 @@ def test_update_login_map_jupyter(self): test_session, other_session, ] - # Call the function under test update_login_map( daemon_conf=self.configuration.daemon_conf, diff --git a/tests/test_mig_shared_useradm.py b/tests/test_mig_shared_useradm.py index 5108d4cb2..81a04d9d7 100644 --- a/tests/test_mig_shared_useradm.py +++ b/tests/test_mig_shared_useradm.py @@ -27,6 +27,7 @@ """Unit tests for the migrid module pointed to in the filename""" +import binascii import datetime import io import os @@ -39,6 +40,7 @@ from mig.shared.base import client_id_dir, distinguished_name_to_user from mig.shared.defaults import ( DEFAULT_USER_ID_FORMAT, + UUID_USER_ID_FORMAT, htaccess_filename, keyword_auto, ssh_conf_dir, @@ -62,6 +64,7 @@ delete_user, edit_user, get_any_oid_user_dn, + lookup_client_id_from_uuid, user_account_notify, ) @@ -82,6 +85,7 @@ # TODO: add this client dir in usersupp and import from there instead TEST_USER_DIR = TEST_USER_DN.replace('/', '+').replace(' ', '_') +TEST_USER_UUID = "UniqueUserIdForTestUser" TEST_USER_SHORT_ID = "abc123@some.org" TEST_USER_EMAIL = TEST_USER_DN.split("/emailAddress=", 1)[-1] TEST_USER_EXPIRE = 1776031200 @@ -136,6 +140,34 @@ ) +def _provision_uuid_test_user(configuration, client_id, client_overrides=None): + """Helper to provision test users when UUID format is used""" + # TODO: merge something like this version into standard _provision_test_user? + # IMPORTANT: we need to use explicit create_user here for UUID format! + user_dict = distinguished_name_to_user(client_id) + # NOTE: generate unique and short id based on id to avoid test collisions + user_dict["unique_id"] = binascii.hexlify( + client_id.encode('utf8')).decode('ascii') + user_dict["short_id"] = binascii.hexlify( + user_dict["email"].encode('utf8')).decode('ascii') + user_dict["comment"] = "This is the user account comment" + user_dict["locality"] = "" + user_dict["organizational_unit"] = "" + user_dict["password"] = "" + user_dict["password_hash"] = "" + if client_overrides is not None: + user_dict.update(client_overrides) + + create_user( + user_dict, + configuration, + keyword_auto, + default_renew=True, + ask_renew=False, + ) + return user_dict + + class TestMigSharedUseradm__create_user( MigTestCase, FixtureAssertMixin, PickleAssertMixin ): @@ -148,6 +180,7 @@ class TestMigSharedUseradm__create_user( def before_each(self): configuration = self.configuration + configuration.site_user_id_format = DEFAULT_USER_ID_FORMAT _ensure_dirs_needed_for_userdb(self.configuration) @@ -424,106 +457,18 @@ def test_user_creation_with_id_collission_fails(self): ) -class MigSharedUseradm__assure_current_htaccess(MigTestCase): - """Coverage of useradm behaviours around htaccess.""" - - def before_each(self): - """The create_user call requires quite a few helper dirs""" - os.makedirs(os.path.join(DUMMY_HOME_PATH, DUMMY_USER)) - os.makedirs(os.path.join(DUMMY_SETTINGS_PATH, DUMMY_USER)) - os.makedirs(os.path.join(DUMMY_MRSL_FILES_PATH, DUMMY_USER)) - os.makedirs(os.path.join(DUMMY_RESOURCE_PENDING_PATH, DUMMY_USER)) - os.makedirs(os.path.join(DUMMY_CACHE_PATH, DUMMY_USER)) - cleanpath(DUMMY_HOME_PATH, self) - cleanpath(DUMMY_SETTINGS_PATH, self) - cleanpath(DUMMY_MRSL_FILES_PATH, self) - cleanpath(DUMMY_RESOURCE_PENDING_PATH, self) - cleanpath(DUMMY_CACHE_PATH, self) - - def assertHtaccessRequireUserClause(self, generated, expected): - """Makes sure generated htaccess file contains the expected string""" - if isinstance(generated, basestring): - with io.open(generated) as htaccess_file: - generated = htaccess_file.read() - - generated_lines = generated.split("\n") - if expected not in generated_lines: - raise AssertionError("no such require user line: %s" % expected) - - def test_skips_accounts_without_short_id(self): - user_dict = {} - user_dict.update(DUMMY_USER_DICT) - del user_dict["short_id"] - assure_current_htaccess(DUMMY_CONF, DUMMY_USER, user_dict, False, - False) - - try: - path_kind = self.assertPathExists(DUMMY_REL_HTACCESS_PATH) - # File should not exist here at all - self.assertNotEqual(path_kind, "file") - except OSError: - pass - - def test_creates_missing_htaccess_file(self): - user_dict = {} - user_dict.update(DUMMY_USER_DICT) - assure_current_htaccess(DUMMY_CONF, DUMMY_USER, user_dict, False, - False) - - path_kind = self.assertPathExists(DUMMY_REL_HTACCESS_PATH) - # File should exist here and be valid - self.assertEqual(path_kind, "file") - path_kind = self.assertPathExists(DUMMY_REL_HTACCESS_BACKUP_PATH) - # Backup file should exist here and be empty - self.assertEqual(path_kind, "file") - - self.assertHtaccessRequireUserClause( - DUMMY_HTACCESS_PATH, DUMMY_REQUIRE_USER - ) - - def test_repairs_existing_stale_htaccess_file(self): - user_dict = {} - user_dict.update(DUMMY_USER_DICT) - # Fake stale user ID directly through DN - user_dict["distinguished_name"] = DUMMY_STALE_USER - assure_current_htaccess(DUMMY_CONF, DUMMY_USER, user_dict, False, - False) - - # Verify stale - self.assertHtaccessRequireUserClause( - DUMMY_HTACCESS_PATH, DUMMY_REQUIRE_STALE_USER - ) - - # Reset stale user ID and retry - user_dict = {} - user_dict.update(DUMMY_USER_DICT) - assure_current_htaccess(DUMMY_CONF, DUMMY_USER, user_dict, False, - False) - - path_kind = self.assertPathExists(DUMMY_REL_HTACCESS_PATH) - # File should exist here and be valid - self.assertEqual(path_kind, "file") - path_kind = self.assertPathExists(DUMMY_REL_HTACCESS_BACKUP_PATH) - # Backup file should exist here and be empty - self.assertEqual(path_kind, "file") - - self.assertHtaccessRequireUserClause( - DUMMY_HTACCESS_PATH, DUMMY_REQUIRE_USER - ) - - -class TestMigSharedUseradm__user_account_notify(MigTestCase, UserAssertMixin): - """Coverage of useradm user_account_notify function.""" - - expected_expire = -1 +# TODO: consider merging the tests for UUID and X509 format +class TestMigSharedUseradm__create_user_uuid_user_id( + MigTestCase, FixtureAssertMixin, PickleAssertMixin +): + """Coverage of useradm create_user function with UUID format.""" - def _provide_configuration(self): - """Return configuration to use""" - return "testconfig" + TEST_USER_DN_GDP = "%s/GDP" % (TEST_USER_DN,) + TEST_USER_PASSWORD_HASH = "PBKDF2$sha256$10000$XMZGaar/pU4PvWDr$w0dYjezF6JGtSiYPexyZMt3lM2134uix" def before_each(self): - """Create test environment for useradm tests""" configuration = self.configuration + configuration.site_user_id_format = UUID_USER_ID_FORMAT _ensure_dirs_needed_for_userdb(self.configuration) @@ -534,162 +479,899 @@ def before_each(self): self.expected_user_db_home, "MiG-users.db" ) ensure_dirs_exist(self.configuration.mig_system_files) - self._provision_test_user(self, TEST_USER_DN) - adjusted_datetime = datetime.date.today() + datetime.timedelta(days=5) - self.expected_expire = int(time.mktime(adjusted_datetime.timetuple())) - def test_default_address_and_expire(self): - """Test addresses and expire for test account""" - (_, username, full_name, expire, addresses, errors) = \ - user_account_notify(TEST_USER_DN, {'email': ['AUTO']}, - self.configuration, - self.expected_user_db_file, False, - False) - self.assertEqual(addresses, {'email': [TEST_USER_EMAIL]}) - self.assertEqual(expire, self.expected_expire) - self.assertEqual(errors, []) + def _provide_configuration(self): + return "testconfig" - def test_extra_address_and_expire(self): - """Test addresses and expire for test with extra account""" - (_, username, full_name, expire, addresses, errors) = \ - user_account_notify(TEST_USER_DN, {'email': - ['AUTO', OTHER_USER_EMAIL]}, - self.configuration, - self.expected_user_db_file, False, - False) - self.assertEqual(addresses, {'email': - [TEST_USER_EMAIL, OTHER_USER_EMAIL]}) - self.assertEqual(expire, self.expected_expire) - self.assertEqual(errors, []) + def test_user_db_is_created(self): + user_dict = {} + user_dict["unique_id"] = TEST_USER_UUID + 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" + create_user( + user_dict, self.configuration, keyword_auto, default_renew=True + ) - def test_missing_user_fails(self): - """Test failure for missing user account""" - (_, username, full_name, expire, addresses, errors) = \ - user_account_notify(OTHER_USER_DN, {'email': ['AUTO']}, - self.configuration, - self.expected_user_db_file, False, - False) - self.assertEqual(addresses, {'email': []}) - self.assertEqual(expire, None) - self.assertTrue(errors and 'No such user' in errors[0]) + # presence of user home + path_kind = MigTestCase._absolute_path_kind(self.expected_user_db_home) + self.assertEqual(path_kind, "dir") - def test_missing_user_db_bails_out(self): - """Test failure for missing user db""" - with self.assertLogs(level='ERROR') as log_capture: - (_, username, full_name, expire, addresses, errors) = \ - user_account_notify(OTHER_USER_DN, {'email': ['AUTO']}, - self.configuration, NO_SUCH_USER_DB, - False, False) - self.assertEqual(addresses, []) - self.assertEqual(expire, None) - self.assertTrue(errors and 'Failed to load user DB' in errors[0]) - self.assertTrue(any('Failed to load user DB' in msg for msg in - log_capture.output)) - try: - os.remove("%s.lock" % NO_SUCH_USER_DB) - except Exception: - pass + # presence of user db + path_kind = MigTestCase._absolute_path_kind(self.expected_user_db_file) + self.assertEqual(path_kind, "file") + def test_user_creation_records_a_user(self): + def _adjust_user_dict_for_compare(user_obj): + obj = dict(user_obj) + obj["created"] = 9999999999.9999999 + obj["expire"] = 9999999999.9999999 + obj["unique_id"] = "__UNIQUE_ID__" + return obj -class TestMigSharedUseradm__delete_user( - MigTestCase, FixtureAssertMixin, PickleAssertMixin -): - """Coverage of useradm delete_user function.""" + expected_user_id = TEST_USER_DN + expected_user_password_hash = self.TEST_USER_PASSWORD_HASH - TEST_USER_DN_GDP = "%s/GDP" % (TEST_USER_DN,) - TEST_USER_PASSWORD_HASH = ( - "PBKDF2$sha256$10000$XMZGaar/pU4PvWDr$w0dYjezF6JGtSiYPexyZMt3lM2134uix" - ) + user_dict = {} + user_dict["unique_id"] = TEST_USER_UUID + user_dict["full_name"] = "Test User" + user_dict["organization"] = "Test Org" + user_dict["state"] = "NA" + user_dict["country"] = "DK" + user_dict["email"] = "test@example.com" + user_dict["comment"] = "This is the create comment" + user_dict["locality"] = "" + user_dict["organizational_unit"] = "" + user_dict["password"] = "" + user_dict["password_hash"] = expected_user_password_hash - def before_each(self): - configuration = self.configuration - configuration.site_user_id_format = DEFAULT_USER_ID_FORMAT + create_user( + user_dict, self.configuration, keyword_auto, default_renew=True + ) - _ensure_dirs_needed_for_userdb(self.configuration) + pickled = self.assertPickledFile( + self.expected_user_db_file, + apply_hints=["convert_dict_bytes_to_strings_kv"], + ) + self.assertIn(expected_user_id, pickled) - self.expected_user_db_home = os.path.normpath( - configuration.user_db_home + prepared = self.prepareFixtureAssert( + "MiG-users.db--example", fixture_format="json" ) - self.expected_user_db_file = os.path.join( - self.expected_user_db_home, "MiG-users.db" + + # TODO: remove resetting the handful of keys here + # this is done to allow the comparision to succeed + actual_user_object = _adjust_user_dict_for_compare( + pickled[expected_user_id] + ) + expected_user_object = _adjust_user_dict_for_compare( + prepared.fixture_data[expected_user_id] ) - ensure_dirs_exist(self.configuration.mig_system_files) - def _provide_configuration(self): - return "testconfig" + self.maxDiff = None + self.assertEqual(actual_user_object, expected_user_object) - def test_user_delete_completes(self): - self._provision_test_user(self, TEST_USER_DN) + def test_user_creation_creates_fs_entries(self): + user_dict = {} + user_dict["unique_id"] = TEST_USER_UUID + user_dict["short_id"] = TEST_USER_SHORT_ID + user_dict["full_name"] = "Test User" + user_dict["organization"] = "Test Org" + user_dict["state"] = "NA" + user_dict["country"] = "DK" + user_dict["email"] = "test@example.com" + user_dict["comment"] = "This is the create comment" + user_dict["locality"] = "" + user_dict["organizational_unit"] = "" + user_dict["password"] = "" + user_dict["password_hash"] = self.TEST_USER_PASSWORD_HASH - try: - delete_user( - distinguished_name_to_user(TEST_USER_DN), - self.configuration, - keyword_auto, - force=True, - ) - except Exception: + create_user( + user_dict, self.configuration, keyword_auto, default_renew=True + ) + home_dir = os.path.join(self.configuration.user_home, TEST_USER_UUID) + self.assertTrue(os.path.isdir(home_dir)) + home_link = os.path.join(self.configuration.user_home, + TEST_USER_DIR) + self.assertTrue(os.path.islink(home_link)) + self.assertEqual(os.path.realpath(home_dir), + os.path.realpath(home_link)) + short_link = os.path.join(self.configuration.user_home, + TEST_USER_SHORT_ID) + self.assertTrue(os.path.islink(short_link)) + self.assertEqual(os.path.realpath(home_dir), + os.path.realpath(short_link)) + + settings_dir = os.path.join(self.configuration.user_settings, + TEST_USER_UUID) + self.assertTrue(os.path.isdir(settings_dir)) + settings_link = os.path.join(self.configuration.user_settings, + TEST_USER_DIR) + self.assertTrue(os.path.islink(settings_link)) + self.assertEqual(os.path.realpath(settings_dir), + os.path.realpath(settings_link)) + + ssh_dir = os.path.join(home_dir, ssh_conf_dir) + self.assertTrue(os.path.isdir(ssh_dir)) + davs_dir = os.path.join(home_dir, davs_conf_dir) + self.assertTrue(os.path.isdir(davs_dir)) + ftps_dir = os.path.join(home_dir, ftps_conf_dir) + self.assertTrue(os.path.isdir(ftps_dir)) + htaccess_path = os.path.join(home_dir, htaccess_filename) + self.assertTrue(os.path.isfile(htaccess_path)) + # NOTE: test htaccess contents matches access for UUID and X509 ID + req_pattern = 'require user "%s"' + with open(htaccess_path) as test_fd: + test_contents = test_fd.read() + self.assertIn(req_pattern % TEST_USER_DN, test_contents) + # TODO: add UUID to htaccess and enable next? + # self.assertIn(req_pattern % TEST_USER_UUID, htaccess_contents) + + enc_user_dn = TEST_USER_DN.encode('utf8') + enc_creator = 'CREATOR'.encode('utf8') + welcome_path = os.path.join(home_dir, welcome_filename) + self.assertTrue(os.path.isfile(welcome_path)) + settings_path = os.path.join(settings_dir, settings_filename) + self.assertTrue(os.path.isfile(settings_path)) + pickled = self.assertPickledFile(settings_path) + self.assertIn(enc_user_dn, pickled[enc_creator]) + profile_path = os.path.join(settings_dir, profile_filename) + self.assertTrue(os.path.isfile(profile_path)) + pickled = self.assertPickledFile(profile_path) + self.assertIn(enc_user_dn, pickled[enc_creator]) + widgets_path = os.path.join(settings_dir, widgets_filename) + self.assertTrue(os.path.isfile(widgets_path)) + pickled = self.assertPickledFile(widgets_path) + self.assertIn(enc_user_dn, pickled[enc_creator]) + css_path = os.path.join(home_dir, default_css_filename) + self.assertTrue(os.path.isfile(css_path)) + with open(css_path) as test_fd: + test_contents = test_fd.read() + self.assertIn('No changes - use default', test_contents) + + # NOTE: check permissions on htaccess, .ssh + htaccess_stat = os.stat(htaccess_path) + self.assertEqual(htaccess_stat.st_mode, 0o100444) + ssh_stat = os.stat(ssh_dir) + self.assertEqual(ssh_stat.st_mode, 0o40755) + + def test_user_creation_records_a_user_with_gdp(self): + self.configuration.site_enable_gdp = True + + user_dict = {} + user_dict["unique_id"] = TEST_USER_UUID + user_dict["full_name"] = "Test User" + user_dict["organization"] = "Test Org" + user_dict["state"] = "NA" + user_dict["country"] = "DK" + user_dict["email"] = "test@example.com" + user_dict["comment"] = "This is the create comment" + user_dict["locality"] = "" + user_dict["organizational_unit"] = "" + user_dict["password"] = "" + user_dict["password_hash"] = self.TEST_USER_PASSWORD_HASH + # explicitly setting set a DN suffixed user DN to force GDP + user_dict["distinguished_name"] = self.TEST_USER_DN_GDP + + try: + create_user( + user_dict, self.configuration, keyword_auto, default_renew=True + ) + except Exception: + self.assertFalse(True, "should not be reached") + + def test_user_creation_and_renew_records_a_user(self): + user_dict = {} + user_dict["unique_id"] = TEST_USER_UUID + user_dict["full_name"] = "Test User" + user_dict["organization"] = "Test Org" + user_dict["state"] = "NA" + user_dict["country"] = "DK" + user_dict["email"] = "test@example.com" + user_dict["comment"] = "This is the create comment" + user_dict["locality"] = "" + user_dict["organizational_unit"] = "" + user_dict["password"] = "" + user_dict["password_hash"] = self.TEST_USER_PASSWORD_HASH + + try: + create_user( + user_dict, + self.configuration, + keyword_auto, + default_renew=True, + ask_renew=False, + ) + except Exception: + self.assertFalse(True, "should not be reached") + + try: + create_user( + user_dict, + self.configuration, + keyword_auto, + default_renew=True, + ask_renew=False, + ) + except Exception: + self.assertFalse(True, "should not be reached") + + def test_user_creation_fails_in_renew_when_locked(self): + user_dict = {} + user_dict["unique_id"] = TEST_USER_UUID + user_dict["full_name"] = "Test User" + user_dict["organization"] = "Test Org" + user_dict["state"] = "NA" + user_dict["country"] = "DK" + user_dict["email"] = "test@example.com" + user_dict["comment"] = "This is the create comment" + user_dict["locality"] = "" + user_dict["organizational_unit"] = "" + user_dict["password"] = "" + user_dict["password_hash"] = self.TEST_USER_PASSWORD_HASH + # explicitly setting set a DN suffixed user DN to force GDP + user_dict["distinguished_name"] = self.TEST_USER_DN_GDP + user_dict["status"] = "locked" + + try: + create_user( + user_dict, + self.configuration, + keyword_auto, + default_renew=True, + ask_renew=False, + ) + except Exception: + self.assertFalse(True, "should not be reached") + + def test_user_creation_with_id_collission_fails(self): + user_dict = {} + user_dict["unique_id"] = TEST_USER_UUID + 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 + ) + 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.assertRaises(Exception): + create_user( + user_dict, + self.configuration, + keyword_auto, + default_renew=True, + ask_renew=False, + ) + + +class MigSharedUseradm__assure_current_htaccess(MigTestCase): + """Coverage of useradm behaviours around htaccess.""" + + def before_each(self): + """The create_user call requires quite a few helper dirs""" + os.makedirs(os.path.join(DUMMY_HOME_PATH, DUMMY_USER)) + os.makedirs(os.path.join(DUMMY_SETTINGS_PATH, DUMMY_USER)) + os.makedirs(os.path.join(DUMMY_MRSL_FILES_PATH, DUMMY_USER)) + os.makedirs(os.path.join(DUMMY_RESOURCE_PENDING_PATH, DUMMY_USER)) + os.makedirs(os.path.join(DUMMY_CACHE_PATH, DUMMY_USER)) + cleanpath(DUMMY_HOME_PATH, self) + cleanpath(DUMMY_SETTINGS_PATH, self) + cleanpath(DUMMY_MRSL_FILES_PATH, self) + cleanpath(DUMMY_RESOURCE_PENDING_PATH, self) + cleanpath(DUMMY_CACHE_PATH, self) + + def assertHtaccessRequireUserClause(self, generated, expected): + """Makes sure generated htaccess file contains the expected string""" + if isinstance(generated, basestring): + with io.open(generated) as htaccess_file: + generated = htaccess_file.read() + + generated_lines = generated.split("\n") + if expected not in generated_lines: + raise AssertionError("no such require user line: %s" % expected) + + def test_skips_accounts_without_short_id(self): + user_dict = {} + user_dict.update(DUMMY_USER_DICT) + del user_dict["short_id"] + assure_current_htaccess(DUMMY_CONF, DUMMY_USER, user_dict, False, + False) + + try: + path_kind = self.assertPathExists(DUMMY_REL_HTACCESS_PATH) + # File should not exist here at all + self.assertNotEqual(path_kind, "file") + except OSError: + pass + + def test_creates_missing_htaccess_file(self): + user_dict = {} + user_dict.update(DUMMY_USER_DICT) + assure_current_htaccess(DUMMY_CONF, DUMMY_USER, user_dict, False, + False) + + path_kind = self.assertPathExists(DUMMY_REL_HTACCESS_PATH) + # File should exist here and be valid + self.assertEqual(path_kind, "file") + path_kind = self.assertPathExists(DUMMY_REL_HTACCESS_BACKUP_PATH) + # Backup file should exist here and be empty + self.assertEqual(path_kind, "file") + + self.assertHtaccessRequireUserClause( + DUMMY_HTACCESS_PATH, DUMMY_REQUIRE_USER + ) + + def test_repairs_existing_stale_htaccess_file(self): + user_dict = {} + user_dict.update(DUMMY_USER_DICT) + # Fake stale user ID directly through DN + user_dict["distinguished_name"] = DUMMY_STALE_USER + assure_current_htaccess(DUMMY_CONF, DUMMY_USER, user_dict, False, + False) + + # Verify stale + self.assertHtaccessRequireUserClause( + DUMMY_HTACCESS_PATH, DUMMY_REQUIRE_STALE_USER + ) + + # Reset stale user ID and retry + user_dict = {} + user_dict.update(DUMMY_USER_DICT) + assure_current_htaccess(DUMMY_CONF, DUMMY_USER, user_dict, False, + False) + + path_kind = self.assertPathExists(DUMMY_REL_HTACCESS_PATH) + # File should exist here and be valid + self.assertEqual(path_kind, "file") + path_kind = self.assertPathExists(DUMMY_REL_HTACCESS_BACKUP_PATH) + # Backup file should exist here and be empty + self.assertEqual(path_kind, "file") + + self.assertHtaccessRequireUserClause( + DUMMY_HTACCESS_PATH, DUMMY_REQUIRE_USER + ) + + +class TestMigSharedUseradm__user_account_notify(MigTestCase, UserAssertMixin): + """Coverage of useradm user_account_notify function.""" + + expected_expire = -1 + + def _provide_configuration(self): + """Return configuration to use""" + return "testconfig" + + def before_each(self): + """Create test environment for useradm tests""" + configuration = self.configuration + + _ensure_dirs_needed_for_userdb(self.configuration) + + self.expected_user_db_home = os.path.normpath( + configuration.user_db_home + ) + self.expected_user_db_file = os.path.join( + self.expected_user_db_home, "MiG-users.db" + ) + ensure_dirs_exist(self.configuration.mig_system_files) + self._provision_test_user(self, TEST_USER_DN) + adjusted_datetime = datetime.date.today() + datetime.timedelta(days=5) + self.expected_expire = int(time.mktime(adjusted_datetime.timetuple())) + + def test_default_address_and_expire(self): + """Test addresses and expire for test account""" + (_, username, full_name, expire, addresses, errors) = \ + user_account_notify(TEST_USER_DN, {'email': ['AUTO']}, + self.configuration, + self.expected_user_db_file, False, + False) + self.assertEqual(addresses, {'email': [TEST_USER_EMAIL]}) + self.assertEqual(expire, self.expected_expire) + self.assertEqual(errors, []) + + def test_extra_address_and_expire(self): + """Test addresses and expire for test with extra account""" + (_, username, full_name, expire, addresses, errors) = \ + user_account_notify(TEST_USER_DN, {'email': + ['AUTO', OTHER_USER_EMAIL]}, + self.configuration, + self.expected_user_db_file, False, + False) + self.assertEqual(addresses, {'email': + [TEST_USER_EMAIL, OTHER_USER_EMAIL]}) + self.assertEqual(expire, self.expected_expire) + self.assertEqual(errors, []) + + def test_missing_user_fails(self): + """Test failure for missing user account""" + (_, username, full_name, expire, addresses, errors) = \ + user_account_notify(OTHER_USER_DN, {'email': ['AUTO']}, + self.configuration, + self.expected_user_db_file, False, + False) + self.assertEqual(addresses, {'email': []}) + self.assertEqual(expire, None) + self.assertTrue(errors and 'No such user' in errors[0]) + + def test_missing_user_db_bails_out(self): + """Test failure for missing user db""" + with self.assertLogs(level='ERROR') as log_capture: + (_, username, full_name, expire, addresses, errors) = \ + user_account_notify(OTHER_USER_DN, {'email': ['AUTO']}, + self.configuration, NO_SUCH_USER_DB, + False, False) + self.assertEqual(addresses, []) + self.assertEqual(expire, None) + self.assertTrue(errors and 'Failed to load user DB' in errors[0]) + self.assertTrue(any('Failed to load user DB' in msg for msg in + log_capture.output)) + try: + os.remove("%s.lock" % NO_SUCH_USER_DB) + except Exception: + pass + + +# TODO: consider merging the tests for UUID and X509 format +class TestMigSharedUseradm__delete_user( + MigTestCase, FixtureAssertMixin, PickleAssertMixin +): + """Coverage of useradm delete_user function.""" + + TEST_USER_DN_GDP = "%s/GDP" % (TEST_USER_DN,) + TEST_USER_PASSWORD_HASH = ( + "PBKDF2$sha256$10000$XMZGaar/pU4PvWDr$w0dYjezF6JGtSiYPexyZMt3lM2134uix" + ) + + def before_each(self): + configuration = self.configuration + configuration.site_user_id_format = DEFAULT_USER_ID_FORMAT + + _ensure_dirs_needed_for_userdb(self.configuration) + + self.expected_user_db_home = os.path.normpath( + configuration.user_db_home + ) + self.expected_user_db_file = os.path.join( + self.expected_user_db_home, "MiG-users.db" + ) + ensure_dirs_exist(self.configuration.mig_system_files) + + def _provide_configuration(self): + return "testconfig" + + def test_user_delete_completes(self): + self._provision_test_user(self, TEST_USER_DN) + + try: + delete_user( + distinguished_name_to_user(TEST_USER_DN), + self.configuration, + keyword_auto, + force=True, + ) + except Exception: + self.assertFalse(True, "should not be reached") + + def test_user_deletion_removes_fs_entries(self): + self._provision_test_user(self, TEST_USER_DN) + + try: + delete_user( + distinguished_name_to_user(TEST_USER_DN), + self.configuration, + keyword_auto, + force=True, + ) + except Exception: + self.assertFalse(True, "should not be reached") + + home_dir = os.path.join(self.configuration.user_home, TEST_USER_DIR) + self.assertFalse(os.path.isdir(home_dir)) + self.assertFalse(os.path.exists(home_dir)) + short_link = os.path.join(self.configuration.user_home, + TEST_USER_SHORT_ID) + self.assertFalse(os.path.islink(short_link)) + self.assertFalse(os.path.exists(short_link)) + + settings_dir = os.path.join(self.configuration.user_settings, + TEST_USER_DIR) + self.assertFalse(os.path.isdir(settings_dir)) + self.assertFalse(os.path.exists(settings_dir)) + + ssh_dir = os.path.join(home_dir, ssh_conf_dir) + self.assertFalse(os.path.isdir(ssh_dir)) + self.assertFalse(os.path.exists(ssh_dir)) + davs_dir = os.path.join(home_dir, davs_conf_dir) + self.assertFalse(os.path.isdir(davs_dir)) + self.assertFalse(os.path.exists(davs_dir)) + ftps_dir = os.path.join(home_dir, ftps_conf_dir) + self.assertFalse(os.path.isdir(ftps_dir)) + self.assertFalse(os.path.exists(ftps_dir)) + htaccess_path = os.path.join(home_dir, htaccess_filename) + self.assertFalse(os.path.isfile(htaccess_path)) + self.assertFalse(os.path.exists(htaccess_path)) + welcome_path = os.path.join(home_dir, welcome_filename) + self.assertFalse(os.path.isfile(welcome_path)) + self.assertFalse(os.path.exists(welcome_path)) + settings_path = os.path.join(settings_dir, settings_filename) + self.assertFalse(os.path.isfile(settings_path)) + self.assertFalse(os.path.exists(settings_path)) + profile_path = os.path.join(settings_dir, profile_filename) + self.assertFalse(os.path.isfile(profile_path)) + self.assertFalse(os.path.exists(profile_path)) + widgets_path = os.path.join(settings_dir, widgets_filename) + self.assertFalse(os.path.isfile(widgets_path)) + self.assertFalse(os.path.exists(widgets_path)) + css_path = os.path.join(home_dir, default_css_filename) + self.assertFalse(os.path.isfile(css_path)) + self.assertFalse(os.path.exists(css_path)) + + +class TestMigSharedUseradm__delete_user_uuid_user_id( + MigTestCase, FixtureAssertMixin, PickleAssertMixin +): + """Coverage of useradm delete_user function with UUID format.""" + + TEST_USER_DN_GDP = "%s/GDP" % (TEST_USER_DN,) + TEST_USER_PASSWORD_HASH = "PBKDF2$sha256$10000$XMZGaar/pU4PvWDr$w0dYjezF6JGtSiYPexyZMt3lM2134uix" + + def before_each(self): + configuration = self.configuration + configuration.site_user_id_format = UUID_USER_ID_FORMAT + + _ensure_dirs_needed_for_userdb(self.configuration) + + self.expected_user_db_home = os.path.normpath( + configuration.user_db_home + ) + self.expected_user_db_file = os.path.join( + self.expected_user_db_home, "MiG-users.db" + ) + ensure_dirs_exist(self.configuration.mig_system_files) + + def _provide_configuration(self): + return "testconfig" + + def test_user_delete_completes(self): + user_dict = _provision_uuid_test_user(self.configuration, TEST_USER_DN) + + try: + delete_user( + user_dict, + self.configuration, + keyword_auto, + force=True, + ) + except Exception: + self.assertFalse(True, "should not be reached") + + def test_user_deletion_removes_fs_entries(self): + user_dict = _provision_uuid_test_user(self.configuration, TEST_USER_DN) + + try: + delete_user( + user_dict, + self.configuration, + keyword_auto, + force=True, + ) + except Exception: self.assertFalse(True, "should not be reached") - def test_user_deletion_removes_fs_entries(self): + home_dir = os.path.join(self.configuration.user_home, TEST_USER_UUID) + self.assertFalse(os.path.isdir(home_dir)) + self.assertFalse(os.path.exists(home_dir)) + home_link = os.path.join(self.configuration.user_home, + TEST_USER_DIR) + self.assertFalse(os.path.islink(home_link)) + self.assertFalse(os.path.exists(home_link)) + short_link = os.path.join(self.configuration.user_home, + TEST_USER_SHORT_ID) + self.assertFalse(os.path.islink(short_link)) + self.assertFalse(os.path.exists(short_link)) + + settings_dir = os.path.join(self.configuration.user_settings, + TEST_USER_UUID) + self.assertFalse(os.path.isdir(settings_dir)) + self.assertFalse(os.path.exists(settings_dir)) + settings_link = os.path.join(self.configuration.user_settings, + TEST_USER_DIR) + self.assertFalse(os.path.islink(settings_link)) + self.assertFalse(os.path.exists(settings_link)) + + ssh_dir = os.path.join(home_dir, ssh_conf_dir) + self.assertFalse(os.path.isdir(ssh_dir)) + self.assertFalse(os.path.exists(ssh_dir)) + davs_dir = os.path.join(home_dir, davs_conf_dir) + self.assertFalse(os.path.isdir(davs_dir)) + self.assertFalse(os.path.exists(davs_dir)) + ftps_dir = os.path.join(home_dir, ftps_conf_dir) + self.assertFalse(os.path.isdir(ftps_dir)) + self.assertFalse(os.path.exists(ftps_dir)) + htaccess_path = os.path.join(home_dir, htaccess_filename) + self.assertFalse(os.path.isfile(htaccess_path)) + self.assertFalse(os.path.exists(htaccess_path)) + welcome_path = os.path.join(home_dir, welcome_filename) + self.assertFalse(os.path.isfile(welcome_path)) + self.assertFalse(os.path.exists(welcome_path)) + settings_path = os.path.join(settings_dir, settings_filename) + self.assertFalse(os.path.isfile(settings_path)) + self.assertFalse(os.path.exists(settings_path)) + profile_path = os.path.join(settings_dir, profile_filename) + self.assertFalse(os.path.isfile(profile_path)) + self.assertFalse(os.path.exists(profile_path)) + widgets_path = os.path.join(settings_dir, widgets_filename) + self.assertFalse(os.path.isfile(widgets_path)) + self.assertFalse(os.path.exists(widgets_path)) + css_path = os.path.join(home_dir, default_css_filename) + self.assertFalse(os.path.isfile(css_path)) + self.assertFalse(os.path.exists(css_path)) + + +class TestMigSharedUseradm__edit_user( + MigTestCase, FixtureAssertMixin, PickleAssertMixin, UserAssertMixin +): + """Coverage of useradm edit_user function.""" + + TEST_USER_PASSWORD_HASH = ( + "PBKDF2$sha256$10000$XMZGaar/pU4PvWDr$w0dYjezF6JGtSiYPexyZMt3lM2134uix" + ) + + def before_each(self): + configuration = self.configuration + configuration.site_user_id_format = DEFAULT_USER_ID_FORMAT + + _ensure_dirs_needed_for_userdb(self.configuration) + + self.expected_user_db_home = os.path.normpath( + configuration.user_db_home + ) + self.expected_user_db_file = os.path.join( + self.expected_user_db_home, "MiG-users.db" + ) + ensure_dirs_exist(self.configuration.mig_system_files) + ensure_dirs_exist(self.configuration.resource_home) + + def _provide_configuration(self): + return "testconfig" + + def _flush_test_user(self, client_id): + """Helper to force clean up after provisioned test users""" + try: + delete_user( + {'distinguished_name': client_id}, + self.configuration, + keyword_auto, + force=True, + ) + except Exception: + pass + + def test_edit_user_email(self): + """Test basic user email editing""" + self._provision_test_user(self, TEST_USER_DN) + # Extract original for later verification + pickled = self.assertPickledFile( + self.expected_user_db_file, + apply_hints=["convert_dict_bytes_to_strings_kv"], + ) + user_dict = pickled[TEST_USER_DN] + + # Edit user attribute and verify changed in DB + changes = { + "email": OTHER_USER_EMAIL, + } + removes = [] + edit_user( + TEST_USER_DN, changes, removes, self.configuration, keyword_auto + ) + edited_dn = TEST_USER_DN.replace(TEST_USER_EMAIL, OTHER_USER_EMAIL) + # Verify changes + pickled = self.assertPickledFile( + self.expected_user_db_file, + apply_hints=["convert_dict_bytes_to_strings_kv"], + ) + self.assertIn(edited_dn, pickled) + edited_user = pickled[edited_dn] + for field in user_dict: + if field == "email": + self.assertEqual(edited_user[field], OTHER_USER_EMAIL) + elif field == "distinguished_name": + self.assertEqual(edited_user[field], edited_dn) + else: + self.assertEqual(edited_user[field], user_dict[field]) + + self._flush_test_user(edited_user['distinguished_name']) + + def test_edit_user_remove_attributes(self): + """Test removing user attributes""" + self._provision_test_user(self, TEST_USER_DN) + # Extract original for later verification + pickled = self.assertPickledFile( + self.expected_user_db_file, + apply_hints=["convert_dict_bytes_to_strings_kv"], + ) + user_dict = pickled[TEST_USER_DN] + + changes = {} + removes = ["comment"] + edit_user( + TEST_USER_DN, changes, removes, self.configuration, keyword_auto, + meta_only=True + ) + + pickled = self.assertPickledFile( + self.expected_user_db_file, + apply_hints=["convert_dict_bytes_to_strings_kv"], + ) + edited_user = pickled[TEST_USER_DN] + for field in user_dict: + if field == "comment": + self.assertNotIn(field, edited_user) + else: + self.assertEqual(edited_user[field], user_dict[field]) + + self._flush_test_user(TEST_USER_DN) + + def test_edit_user_meta_only(self): + """Test metadata-only update (no FS changes)""" + self._provision_test_user(self, TEST_USER_DN) + # Extract original for later verification + pickled = self.assertPickledFile( + self.expected_user_db_file, + apply_hints=["convert_dict_bytes_to_strings_kv"], + ) + user_dict = pickled[TEST_USER_DN] + + changes = {"comment": "meta comment"} + removes = [] + edit_user( + TEST_USER_DN, changes, removes, self.configuration, keyword_auto, + meta_only=True + ) + + pickled = self.assertPickledFile( + self.expected_user_db_file, + apply_hints=["convert_dict_bytes_to_strings_kv"], + ) + edited_user = pickled[TEST_USER_DN] + for field in user_dict: + if field == "comment": + self.assertEqual(pickled[TEST_USER_DN][field], "meta comment") + else: + self.assertEqual(edited_user[field], user_dict[field]) + + self._flush_test_user(TEST_USER_DN) + + @unittest.skip("make backend function less noisy (traceback) and enable?") + def test_edit_user_nonexistent(self): + """Test editing nonexistent user""" + changes = {"email": "dummy@example.com"} + removes = [] + with self.assertRaises(Exception): + edit_user( + NO_SUCH_USER_DN, changes, removes, self.configuration, + keyword_auto + ) + + @unittest.skip("make backend function less noisy (traceback) and enable?") + def test_edit_user_nonexistent_force(self): + """Test editing nonexistent user with force=True""" + changes = {"email": "dummy@example.com"} + removes = [] + # NOTE: forced edit logs errors in update_account_X functions + self.logger.forgive_errors() + with self.assertRaises(Exception): + edit_user( + NO_SUCH_USER_DN, changes, removes, self.configuration, + keyword_auto, force=True + ) + + @unittest.skip("make backend function less noisy (traceback) and enable?") + def test_edit_user_dn_collision_fails(self): + """Test that editing a user to a DN that already exists fails""" + # NOTE: we can't use _provision_test_users here as it lacks passwords + test_user_dict = distinguished_name_to_user(TEST_USER_DN) + test_user_dict["comment"] = "This is the create comment" + test_user_dict["password"] = "" + test_user_dict["password_hash"] = self.TEST_USER_PASSWORD_HASH + try: + create_user( + test_user_dict, self.configuration, keyword_auto, default_renew=True + ) + except Exception: + self.assertFalse(True, "should not be reached") + other_user_dict = distinguished_name_to_user(OTHER_USER_DN) + other_user_dict["comment"] = "This is the create comment" + other_user_dict["password"] = "" + other_user_dict["password_hash"] = DUMMY_PASSWORD_HASH + try: + create_user( + other_user_dict, self.configuration, keyword_auto, default_renew=True + ) + except Exception: + self.assertFalse(True, "should not be reached") + + changes = distinguished_name_to_user(OTHER_USER_DN) + removes = [] + with self.assertRaises(Exception): + edit_user( + TEST_USER_DN, changes, removes, self.configuration, + keyword_auto + ) + + self._flush_test_user(TEST_USER_DN) + self._flush_test_user(OTHER_USER_DN) + + def test_edit_user_renames_user(self): + """Test editing fields that change the distinguished name""" self._provision_test_user(self, TEST_USER_DN) + # Extract original for later verification + pickled = self.assertPickledFile( + self.expected_user_db_file, + apply_hints=["convert_dict_bytes_to_strings_kv"], + ) + user_dict = pickled[TEST_USER_DN] + + changes = {"full_name": "Renamed User"} + removes = [] + result = edit_user( + TEST_USER_DN, changes, removes, self.configuration, + keyword_auto + ) + + new_dn = result["distinguished_name"] + new_dir = new_dn.replace("/", "+").replace(" ", "_") - try: - delete_user( - distinguished_name_to_user(TEST_USER_DN), - self.configuration, - keyword_auto, - force=True, - ) - except Exception: - self.assertFalse(True, "should not be reached") + pickled = self.assertPickledFile( + self.expected_user_db_file, + apply_hints=["convert_dict_bytes_to_strings_kv"], + ) + edited_user = pickled[new_dn] + self.assertNotIn(TEST_USER_DN, pickled) + self.assertIn(new_dn, pickled) + for field in user_dict: + if field == "full_name": + self.assertEqual(edited_user[field], "Renamed User") + elif field == "distinguished_name": + self.assertEqual(edited_user[field], new_dn) + else: + self.assertEqual(edited_user[field], user_dict[field]) - home_dir = os.path.join(self.configuration.user_home, TEST_USER_DIR) - self.assertFalse(os.path.isdir(home_dir)) - self.assertFalse(os.path.exists(home_dir)) - short_link = os.path.join(self.configuration.user_home, - TEST_USER_SHORT_ID) - self.assertFalse(os.path.islink(short_link)) - self.assertFalse(os.path.exists(short_link)) + old_home = os.path.join(self.configuration.user_home, TEST_USER_DIR) + new_home = os.path.join(self.configuration.user_home, new_dir) + self.assertFalse(os.path.exists(old_home)) + self.assertTrue(os.path.isdir(new_home)) - settings_dir = os.path.join(self.configuration.user_settings, - TEST_USER_DIR) - self.assertFalse(os.path.isdir(settings_dir)) - self.assertFalse(os.path.exists(settings_dir)) + old_settings = os.path.join( + self.configuration.user_settings, TEST_USER_DIR) + new_settings = os.path.join(self.configuration.user_settings, new_dir) + self.assertFalse(os.path.exists(old_settings)) + self.assertTrue(os.path.isdir(new_settings)) - ssh_dir = os.path.join(home_dir, ssh_conf_dir) - self.assertFalse(os.path.isdir(ssh_dir)) - self.assertFalse(os.path.exists(ssh_dir)) - davs_dir = os.path.join(home_dir, davs_conf_dir) - self.assertFalse(os.path.isdir(davs_dir)) - self.assertFalse(os.path.exists(davs_dir)) - ftps_dir = os.path.join(home_dir, ftps_conf_dir) - self.assertFalse(os.path.isdir(ftps_dir)) - self.assertFalse(os.path.exists(ftps_dir)) - htaccess_path = os.path.join(home_dir, htaccess_filename) - self.assertFalse(os.path.isfile(htaccess_path)) - self.assertFalse(os.path.exists(htaccess_path)) - welcome_path = os.path.join(home_dir, welcome_filename) - self.assertFalse(os.path.isfile(welcome_path)) - self.assertFalse(os.path.exists(welcome_path)) - settings_path = os.path.join(settings_dir, settings_filename) - self.assertFalse(os.path.isfile(settings_path)) - self.assertFalse(os.path.exists(settings_path)) - profile_path = os.path.join(settings_dir, profile_filename) - self.assertFalse(os.path.isfile(profile_path)) - self.assertFalse(os.path.exists(profile_path)) - widgets_path = os.path.join(settings_dir, widgets_filename) - self.assertFalse(os.path.isfile(widgets_path)) - self.assertFalse(os.path.exists(widgets_path)) - css_path = os.path.join(home_dir, default_css_filename) - self.assertFalse(os.path.isfile(css_path)) - self.assertFalse(os.path.exists(css_path)) + self._flush_test_user(new_dn) -class TestMigSharedUseradm__edit_user( +class TestMigSharedUseradm__edit_user_uuid_user_id( MigTestCase, FixtureAssertMixin, PickleAssertMixin, UserAssertMixin ): - """Coverage of useradm edit_user function.""" + """Coverage of useradm edit_user function with UUID format.""" TEST_USER_PASSWORD_HASH = ( "PBKDF2$sha256$10000$XMZGaar/pU4PvWDr$w0dYjezF6JGtSiYPexyZMt3lM2134uix" @@ -697,7 +1379,7 @@ class TestMigSharedUseradm__edit_user( def before_each(self): configuration = self.configuration - configuration.site_user_id_format = DEFAULT_USER_ID_FORMAT + configuration.site_user_id_format = UUID_USER_ID_FORMAT _ensure_dirs_needed_for_userdb(self.configuration) @@ -726,8 +1408,8 @@ def _flush_test_user(self, client_id): pass def test_edit_user_email(self): - """Test basic user email editing""" - self._provision_test_user(self, TEST_USER_DN) + """Test basic user email attribute editing with UUID""" + _provision_uuid_test_user(self.configuration, TEST_USER_DN) # Extract original for later verification pickled = self.assertPickledFile( self.expected_user_db_file, @@ -759,11 +1441,11 @@ def test_edit_user_email(self): else: self.assertEqual(edited_user[field], user_dict[field]) - self._flush_test_user(edited_user['distinguished_name']) + self._flush_test_user(edited_dn) def test_edit_user_remove_attributes(self): - """Test removing user attributes""" - self._provision_test_user(self, TEST_USER_DN) + """Test removing user attributes with UUID""" + _provision_uuid_test_user(self.configuration, TEST_USER_DN) # Extract original for later verification pickled = self.assertPickledFile( self.expected_user_db_file, @@ -792,8 +1474,8 @@ def test_edit_user_remove_attributes(self): self._flush_test_user(TEST_USER_DN) def test_edit_user_meta_only(self): - """Test metadata-only update (no FS changes)""" - self._provision_test_user(self, TEST_USER_DN) + """Test metadata-only update with UUID (no FS changes)""" + _provision_uuid_test_user(self.configuration, TEST_USER_DN) # Extract original for later verification pickled = self.assertPickledFile( self.expected_user_db_file, @@ -815,7 +1497,7 @@ def test_edit_user_meta_only(self): edited_user = pickled[TEST_USER_DN] for field in user_dict: if field == "comment": - self.assertEqual(pickled[TEST_USER_DN][field], "meta comment") + self.assertEqual(edited_user[field], "meta comment") else: self.assertEqual(edited_user[field], user_dict[field]) @@ -834,7 +1516,7 @@ def test_edit_user_nonexistent(self): @unittest.skip("make backend function less noisy (traceback) and enable?") def test_edit_user_nonexistent_force(self): - """Test editing nonexistent user with force=True""" + """Test editing nonexistent user with force=True and UUID""" changes = {"email": "dummy@example.com"} removes = [] # NOTE: forced edit logs errors in update_account_X functions @@ -847,34 +1529,15 @@ def test_edit_user_nonexistent_force(self): @unittest.skip("make backend function less noisy (traceback) and enable?") def test_edit_user_dn_collision_fails(self): - """Test that editing a user to a DN that already exists fails""" - # NOTE: we can't use _provision_test_users here as it lacks passwords - test_user_dict = distinguished_name_to_user(TEST_USER_DN) - test_user_dict["comment"] = "This is the create comment" - test_user_dict["password"] = "" - test_user_dict["password_hash"] = self.TEST_USER_PASSWORD_HASH - try: - create_user( - test_user_dict, self.configuration, keyword_auto, default_renew=True - ) - except Exception: - self.assertFalse(True, "should not be reached") - other_user_dict = distinguished_name_to_user(OTHER_USER_DN) - other_user_dict["comment"] = "This is the create comment" - other_user_dict["password"] = "" - other_user_dict["password_hash"] = DUMMY_PASSWORD_HASH - try: - create_user( - other_user_dict, self.configuration, keyword_auto, default_renew=True - ) - except Exception: - self.assertFalse(True, "should not be reached") + """Test that editing a user to a DN that already exists fails with UUID""" + _provision_uuid_test_user(self.configuration, TEST_USER_DN) + _provision_uuid_test_user(self.configuration, OTHER_USER_DN) - changes = distinguished_name_to_user(OTHER_USER_DN) + changes = distinguished_name_to_user(TEST_USER_DN) removes = [] with self.assertRaises(Exception): edit_user( - TEST_USER_DN, changes, removes, self.configuration, + OTHER_USER_DN, changes, removes, self.configuration, keyword_auto ) @@ -882,8 +1545,8 @@ def test_edit_user_dn_collision_fails(self): self._flush_test_user(OTHER_USER_DN) def test_edit_user_renames_user(self): - """Test editing fields that change the distinguished name""" - self._provision_test_user(self, TEST_USER_DN) + """Test editing fields that change the distinguished name with UUID""" + _provision_uuid_test_user(self.configuration, TEST_USER_DN) # Extract original for later verification pickled = self.assertPickledFile( self.expected_user_db_file, @@ -899,7 +1562,7 @@ def test_edit_user_renames_user(self): ) new_dn = result["distinguished_name"] - new_dir = new_dn.replace("/", "+").replace(" ", "_") + new_link_dir = new_dn.replace("/", "+").replace(" ", "_") pickled = self.assertPickledFile( self.expected_user_db_file, @@ -916,16 +1579,16 @@ def test_edit_user_renames_user(self): else: self.assertEqual(edited_user[field], user_dict[field]) - old_home = os.path.join(self.configuration.user_home, TEST_USER_DIR) - new_home = os.path.join(self.configuration.user_home, new_dir) - self.assertFalse(os.path.exists(old_home)) - self.assertTrue(os.path.isdir(new_home)) - - old_settings = os.path.join( - self.configuration.user_settings, TEST_USER_DIR) - new_settings = os.path.join(self.configuration.user_settings, new_dir) - self.assertFalse(os.path.exists(old_settings)) - self.assertTrue(os.path.isdir(new_settings)) + old_link = os.path.join( + self.configuration.user_home, TEST_USER_DIR) + new_link = os.path.join(self.configuration.user_home, new_link_dir) + real_home = os.path.join( + self.configuration.user_home, user_dict['unique_id']) + self.assertFalse(os.path.exists(old_link)) + self.assertTrue(os.path.islink(new_link)) + self.assertEqual(os.path.realpath(new_link), + os.path.realpath(real_home)) + self.assertTrue(os.path.isdir(real_home)) self._flush_test_user(new_dn) @@ -1007,6 +1670,122 @@ def test_get_required_user_alias_links_without_links(self): self.assertEqual(alias_links, [(False, False)]) +class TestMigSharedUseradm__lookup_client_id_from_uuid(MigTestCase): + """Coverage of useradm lookup_client_id_from_uuid function.""" + + def before_each(self): + """Create test environment for useradm tests""" + configuration = self.configuration + configuration.site_user_id_format = UUID_USER_ID_FORMAT + + _ensure_dirs_needed_for_userdb(self.configuration) + + self.expected_user_db_home = os.path.normpath( + configuration.user_db_home + ) + self.expected_user_db_file = os.path.join( + self.expected_user_db_home, "MiG-users.db" + ) + ensure_dirs_exist(self.configuration.mig_system_files) + + def _provide_configuration(self): + return "testconfig" + + def test_lookup_client_id_from_uuid_via_alias_link(self): + """Test lookup via existing alias link in mig_system_run""" + configuration = self.configuration + client_id = TEST_USER_DN + user_id = TEST_USER_UUID + # Setup real directory for the client and alias link in mig_system_run + real_dir = os.path.join(configuration.user_home, user_id) + ensure_dirs_exist(real_dir) + alias_link = os.path.join( + configuration.mig_system_run, user_id_alias_dir, user_id) + ensure_dirs_exist(os.path.dirname(alias_link)) + client_dir = client_id_dir(client_id) + os.symlink(client_dir, alias_link) + + result = lookup_client_id_from_uuid(configuration, user_id) + self.assertEqual(result, client_id) + + def test_lookup_client_id_from_uuid_via_reverse_lookup(self): + """Test lookup via reverse lookup in user_home (and writeback)""" + configuration = self.configuration + client_id = TEST_USER_DN + user_id = TEST_USER_UUID + # Setup real directory for the client and alias link in home + real_dir = os.path.join(configuration.user_home, user_id) + ensure_dirs_exist(real_dir) + # This simulates an existing X509 alias link to uuid in user_home + client_dir = client_id_dir(client_id) + alias_link_in_home = os.path.join(configuration.user_home, client_dir) + os.symlink(user_id, alias_link_in_home) + + result = lookup_client_id_from_uuid(configuration, user_id) + self.assertEqual(result, client_id) + + # Verify writeback: check if alias link was created in mig_system_run + alias_link_in_run = os.path.join(configuration.mig_system_run, + user_id_alias_dir, user_id) + self.assertTrue(os.path.islink(alias_link_in_run)) + link_target = os.path.basename(os.path.realpath(alias_link_in_run)) + self.assertEqual(link_target, client_dir) + + def test_lookup_client_id_from_uuid_via_alias_link_despite_reverse_link(self): + """Test lookup via existing alias link in mig_system_run even when reverse X509 link exists""" + configuration = self.configuration + client_id = TEST_USER_DN + user_id = TEST_USER_UUID + # Setup real directory for the client and alias link in mig_system_run + real_dir = os.path.join(configuration.user_home, user_id) + ensure_dirs_exist(real_dir) + alias_link = os.path.join( + configuration.mig_system_run, user_id_alias_dir, user_id) + ensure_dirs_exist(os.path.dirname(alias_link)) + client_dir = client_id_dir(client_id) + os.symlink(client_dir, alias_link) + # Make the additional reverse link from X509 to UUID in mig_system_run + reverse_link = os.path.join( + configuration.mig_system_run, user_id_alias_dir, client_dir) + os.symlink(user_id, reverse_link) + + # Assure reverse link doesn't interfere + result = lookup_client_id_from_uuid(configuration, user_id) + self.assertEqual(result, client_id) + + def test_lookup_client_id_from_uuid_fails_with_only_short_id_link(self): + """Test lookup via reverse lookup in user_home fails on short id link""" + configuration = self.configuration + client_id = TEST_USER_DN + user_id = TEST_USER_UUID + # Setup real directory for the client and alias link in home + real_dir = os.path.join(configuration.user_home, user_id) + ensure_dirs_exist(real_dir) + # This simulates an existing short id link to uuid in user_home + short_id_link_in_home = os.path.join(configuration.user_home, + TEST_USER_EMAIL) + os.symlink(user_id, short_id_link_in_home) + + with self.assertLogs(level='ERROR') as log_capture: + result = lookup_client_id_from_uuid(configuration, user_id) + self.assertNotEqual(result, client_id) + self.assertEqual(result, user_id) + self.assertTrue(any('found no alias' in msg + for msg in log_capture.output)) + + def test_lookup_client_id_from_uuid_not_found(self): + """Test lookup when no alias or reverse link exists""" + configuration = self.configuration + client_id = NO_SUCH_USER_DN + # Missing user will cause log error + with self.assertLogs(level='ERROR') as log_capture: + result = lookup_client_id_from_uuid(configuration, client_id) + # Verify: should return the user_id itself if not found + self.assertEqual(result, client_id) + self.assertTrue(any('found no alias' in msg + for msg in log_capture.output)) + + class TestMigSharedUseradm__get_any_oid_user_dn( MigTestCase, FixtureAssertMixin, PickleAssertMixin, UserAssertMixin ): @@ -1092,6 +1871,146 @@ def test_get_any_oid_user_dn_user_check_false(self): ) self.assertEqual(result, raw_login) +# TODO: consider merging the tests for UUID and X509 format + + +class TestMigSharedUseradm__get_any_oid_user_dn_uuid_user_id( + MigTestCase, FixtureAssertMixin, PickleAssertMixin, UserAssertMixin +): + """Unit tests for get_any_oid_user_dn with UUID user ID format.""" + + def before_each(self): + """Prepare a minimal configuration for the tests.""" + configuration = self.configuration + # Use UUID format for the tests – the function works with both UUID and X509. + configuration.site_user_id_format = UUID_USER_ID_FORMAT + _ensure_dirs_needed_for_userdb(self.configuration) + self.expected_user_db_home = os.path.normpath( + configuration.user_db_home + ) + self.expected_user_db_file = os.path.join( + self.expected_user_db_home, "MiG-users.db" + ) + ensure_dirs_exist(self.configuration.mig_system_files) + + def _provide_configuration(self): + return "testconfig" + + def _flush_test_user(self, client_id): + """Helper to force clean up after provisioned test users""" + try: + delete_user( + {'distinguished_name': client_id}, + self.configuration, + keyword_auto, + force=True, + ) + except Exception: + pass + + def test_get_any_oid_user_dn_via_lookup_link(self): + """Return the distinguished name when a valid lookup link exists.""" + client_id = TEST_USER_DN + user_dict = _provision_uuid_test_user(self.configuration, client_id) + user_id = user_dict['unique_id'] + short_id = user_dict['short_id'] + + # Make sure direct lookup link is in place + lookup_link = os.path.join(self.configuration.mig_system_run, + user_id_alias_dir, user_id) + self.assertTrue(os.path.islink(lookup_link)) + + # Call the function – it should resolve the lookup to the client_id. + result = get_any_oid_user_dn(self.configuration, raw_login=short_id, + user_check=True, do_lock=True + ) + self._flush_test_user(TEST_USER_DN) + self.assertEqual(result, client_id) + + def test_get_any_oid_user_dn_via_fallback_id_link(self): + """Return the distinguished name when a valid fallback ID link exists.""" + client_id = TEST_USER_DN + user_dict = _provision_uuid_test_user(self.configuration, client_id) + user_id = user_dict['unique_id'] + short_id = user_dict['short_id'] + client_dir = client_id_dir(client_id) + + # Blow away direct lookup link to force alias lookup + lookup_link = os.path.join(self.configuration.mig_system_run, + user_id_alias_dir, user_id) + if os.path.islink(lookup_link): + os.remove(lookup_link) + self.assertFalse(os.path.islink(lookup_link)) + + # Make sure short alias, DN link and id dir are all in place + alias_link = os.path.join(self.configuration.user_home, short_id) + self.assertTrue(os.path.islink(alias_link)) + self.assertEqual(os.path.basename(os.path.realpath(alias_link)), + user_id) + dn_link = os.path.join(self.configuration.user_home, client_dir) + self.assertTrue(os.path.islink(dn_link)) + self.assertEqual(os.path.basename(os.path.realpath(dn_link)), + user_id) + user_dir = os.path.join(self.configuration.user_home, user_id) + self.assertTrue(os.path.isdir(user_dir)) + + # Call the function – it should resolve the alias to the client_id. + result = get_any_oid_user_dn(self.configuration, raw_login=short_id, + user_check=True, do_lock=True + ) + self._flush_test_user(TEST_USER_DN) + self.assertEqual(result, client_id) + + def test_get_any_oid_user_dn_not_found(self): + """When no alias or reverse link exists, return an empty string.""" + # Missing user will cause log error + with self.assertLogs(level='ERROR') as log_capture: + result = get_any_oid_user_dn(self.configuration, + raw_login="NoSuchUser", + user_check=True, do_lock=True + ) + self.assertEqual(result, "") + self.assertTrue(any('no such openid user' in msg + for msg in log_capture.output)) + + def test_get_any_oid_user_id_direct_dn(self): + """Return the distinguished name when a matching id directory exists.""" + client_id = TEST_USER_DN + user_dict = _provision_uuid_test_user(self.configuration, client_id) + user_id = user_dict['unique_id'] + short_id = user_dict['short_id'] + client_dir = client_id_dir(client_id) + + # Make sure no lookups links get in the way and that id dir is in place + lookup_link = os.path.join(self.configuration.mig_system_run, + user_id_alias_dir, user_id) + os.remove(lookup_link) + self.assertFalse(os.path.islink(lookup_link)) + alias_link = os.path.join(self.configuration.user_home, short_id) + os.remove(alias_link) + self.assertFalse(os.path.islink(alias_link)) + reverse_link = os.path.join(self.configuration.user_home, client_dir) + os.remove(reverse_link) + self.assertFalse(os.path.islink(reverse_link)) + user_dir = os.path.join(self.configuration.user_home, user_id) + self.assertTrue(os.path.isdir(user_dir)) + + # The function should recognise the directory and return the client_id. + result = get_any_oid_user_dn(self.configuration, + raw_login=user_id, + user_check=True, do_lock=True + ) + self._flush_test_user(TEST_USER_DN) + self.assertEqual(result, user_id) + + def test_get_any_oid_user_dn_user_check_false(self): + """When user_check=False the function bypasses the user‑dir lookup.""" + raw_login = TEST_USER_SHORT_ID + result = get_any_oid_user_dn(self.configuration, raw_login=raw_login, + user_check=False, do_lock=True + ) + self.assertEqual(result, raw_login) + if __name__ == "__main__": testmain()