diff --git a/src/tests/multihost/alltests/conftest.py b/src/tests/multihost/alltests/conftest.py index 37b17a240f5..eb95df4507c 100644 --- a/src/tests/multihost/alltests/conftest.py +++ b/src/tests/multihost/alltests/conftest.py @@ -664,37 +664,6 @@ def removeds(): request.addfinalizer(removeds) -@pytest.fixture(scope='class') -def multipleds_failover(session_multihost, request): - """ Setup Multiple Directory Servers for failover""" - server_list = [session_multihost.master[0].sys_hostname, - session_multihost.master[1].sys_hostname] - pki_inst = PkiTools() - try: - certdb = pki_inst.createselfsignedcerts(server_list) - except PkiLibException: - pytest.fail("Failed to create CA") - print(certdb) - dsobjlist = [] - for idx in range(2): - host = session_multihost.master[idx] - dsobj = DirSrvWrap(host, - client_obj=session_multihost.client[0], - ssl=True, - ssldb=certdb) - dsobjlist.append(dsobj) - inst_name = 'example' - suffix = 'dc=example,dc=test' - dsobj.create_ds_instance(inst_name, suffix) - - def removeds(): - """ Remove DS Instances """ - for dsinst in dsobjlist: - instname = 'example' - dsinst.remove_ds_instance(instname) - request.addfinalizer(removeds) - - @pytest.fixture(scope='class') # pylint: disable=unused-argument def posix_users_multidomain(session_multihost, multipleds): @@ -990,39 +959,6 @@ def setup_sssd_gssapi(session_multihost, setup_sssd, session_multihost.client[0].service_sssd('restart') -@pytest.fixture(scope='class') -def setup_sssd_failover(session_multihost, request): - """ Configure sssd.conf """ - tools = sssdTools(session_multihost.client[0]) - stop_sssd = 'systemctl stop sssd' - session_multihost.client[0].run_command(stop_sssd) - ds_host1 = session_multihost.master[0].sys_hostname - ds_host2 = session_multihost.master[1].sys_hostname - sssd_params = {'domains': ds_instance_name} - tools.sssd_conf('sssd', sssd_params) - domain_section = 'domain/%s' % ds_instance_name - ldap_uri = 'ldaps://%s, ldaps://%s' % (ds_host1, ds_host2) - domain_params = {'ldap_search_base': ds_suffix, - 'id_provider': 'ldap', - 'auth_provider': 'ldap', - 'ldap_user_home_directory': "/home/%u", - 'ldap_uri': ldap_uri, - 'ldap_tls_cacert': '/etc/openldap/cacerts/cacert.pem', - 'use_fully_qualified_names': 'True', - 'debug_level': '9'} - tools.sssd_conf(domain_section, domain_params) - start_sssd = 'systemctl restart sssd' - session_multihost.client[0].run_command(start_sssd) - - def removesssd(): - """ Remove sssd configuration """ - stop_sssd = 'systemctl stop sssd' - session_multihost.client[0].run_command(stop_sssd) - removeconf = 'rm -f %s' % (SSSD_DEFAULT_CONF) - session_multihost.client[0].run_command(removeconf) - request.addfinalizer(removesssd) - - @pytest.fixture(scope="class") def multihost(session_multihost, request): """ Multihost fixture to be used by tests @@ -1070,40 +1006,6 @@ def create_posix_usersgroups(session_multihost): assert ret == 'Success' -@pytest.fixture(scope='class') -def create_posix_usersgroups_failover(session_multihost): - """ Create posix user and groups """ - for idx in range(2): - ldap_uri = 'ldap://%s' % (session_multihost.master[idx].ip) - ds_rootdn = 'cn=Directory Manager' - ds_rootpw = 'Secret123' - ldap_inst = LdapOperations(ldap_uri, ds_rootdn, ds_rootpw) - for i in range(10): - user_info = {'cn': 'foo%d' % i, - 'uid': 'foo%d' % i, - 'uidNumber': '1458310%d' % i, - 'gidNumber': '14564100'} - ldap_inst.posix_user("ou=People", "dc=example,dc=test", user_info) - - memberdn = 'uid=%s,ou=People,dc=example,dc=test' % ('foo0') - group_info = {'cn': 'ldapusers', - 'gidNumber': '14564100', - 'uniqueMember': memberdn} - try: - ldap_inst.posix_group("ou=Groups", "dc=example,dc=test", - group_info) - except LdapException: - assert False - - group_dn = 'cn=ldapusers,ou=Groups,dc=example,dc=test' - for i in range(1, 10): - user_dn = 'uid=foo%d,ou=People,dc=example,dc=test' % i - add_member = [(ldap.MOD_ADD, 'uniqueMember', - user_dn.encode('utf-8'))] - (ret, _) = ldap_inst.modify_ldap(group_dn, add_member) - assert ret == 'Success' - - @pytest.fixture(scope='class') def create_posix_usersgroups_autoprivategroups(session_multihost): """ Create posix user and groups for autoprivategroup fixture""" diff --git a/src/tests/multihost/alltests/test_failover.py b/src/tests/multihost/alltests/test_failover.py deleted file mode 100644 index 2d5d7774539..00000000000 --- a/src/tests/multihost/alltests/test_failover.py +++ /dev/null @@ -1,106 +0,0 @@ -""" Automation for sssd failover - -:requirement: IDM-SSSD-REQ : Failover -:casecomponent: sssd -:subsystemteam: sst_idm_sssd -:upstream: yes -:status: approved -""" -import pytest -from sssd.testlib.common.utils import sssdTools -from constants import ds_instance_name -from sssd.testlib.common.ssh2_python import check_login_client_bool - - -@pytest.mark.usefixtures('multipleds_failover', - 'create_posix_usersgroups_failover', - 'setup_sssd_failover', ) -@pytest.mark.failover -class TestFailover(object): - """ Bug 1283798 failover automation - :setup: - 1. Configure Directory servers on 2 Hosts (ldap1, ldap2) - with TLS - 2. Configure sssd.conf on client with auth_provider: ldap - 3. specify ldaps in ldap_uri pointing to 2 directory servers - example: ldap_uri: ldaps://ldap1, ldaps://ldap2 - """ - @staticmethod - @pytest.mark.tier2 - def test_0001_getent(multihost): - """ - :title: failover: Verify users can be queried from - second directory server when first directory server is down - :id: 0d145340-e147-4da7-acd0-f1c29891c397 - """ - # query ldap users when both ldaps servers are working - user = 'foo0@%s' % ds_instance_name - getent = 'getent passwd %s' % user - cmd = multihost.client[0].run_command(getent) - assert cmd.returncode == 0 - tools = sssdTools(multihost.client[0]) - # stop first directory server instance_name - stop_ds1 = 'systemctl stop dirsrv@example' - cmd = multihost.master[0].run_command(stop_ds1, raiseonerr=False) - assert cmd.returncode == 0 - # query the new user foo1 - user = 'foo1@%s' % ds_instance_name - getent = 'getent passwd %s' % user - cmd = multihost.client[0].run_command(getent) - assert cmd.returncode == 0 - # clear the cache and query foo1 user again - multihost.client[0].service_sssd('stop') - tools.remove_sss_cache('/var/lib/sss/db') - multihost.client[0].service_sssd('start') - cmd = multihost.client[0].run_command(getent) - assert cmd.returncode == 0 - # start the first directory server - start_ds1 = 'systemctl start dirsrv@example' - cmd = multihost.master[0].run_command(start_ds1, raiseonerr=False) - assert cmd.returncode == 0 - - @staticmethod - @pytest.mark.tier2 - def test_0002_login(multihost): - """ - :title: failover: Verify users can login when the first - ldap server is down - :id: 9c0e0448-3fc2-44c7-96f8-9b8b44fa5cba - """ - user = 'foo2@%s' % ds_instance_name - stop_ds1 = 'systemctl stop dirsrv@example' - cmd = multihost.master[0].run_command(stop_ds1, raiseonerr=False) - assert cmd.returncode == 0 - tools = sssdTools(multihost.client[0]) - multihost.client[0].service_sssd('stop') - tools.remove_sss_cache('/var/lib/sss/db') - multihost.client[0].service_sssd('start') - # login as user - ssh = check_login_client_bool(multihost, user, 'Secret123') - start_ds1 = 'systemctl start dirsrv@example' - cmd = multihost.master[0].run_command(start_ds1, raiseonerr=False) - assert ssh, f'{user} is not able to login.' - assert cmd.returncode == 0 - - @staticmethod - @pytest.mark.tier2 - def test_0003_stopsecondds(multihost): - """ - :title: failover: Stop second ldap server and verify - users are able to login from first ldap server - :id: cf15aea7-a626-4ed2-a205-9180ddfe29b2 - """ - stop_ds2 = 'systemctl stop dirsrv@example' - cmd = multihost.master[1].run_command(stop_ds2, raiseonerr=False) - assert cmd.returncode == 0 - tools = sssdTools(multihost.client[0]) - multihost.client[0].service_sssd('stop') - tools.remove_sss_cache('/var/lib/sss/db') - multihost.client[0].service_sssd('start') - user = 'foo3@%s' % ds_instance_name - # login as user - ssh = check_login_client_bool(multihost, user, 'Secret123') - start_ds1 = 'systemctl start dirsrv@example' - cmd = multihost.master[0].run_command(start_ds1, raiseonerr=False) - assert ssh, f'{user} is not able to login.' - assert cmd.returncode == 0 diff --git a/src/tests/multihost/alltests/test_krb_ldap_connection.py b/src/tests/multihost/alltests/test_krb_ldap_connection.py deleted file mode 100644 index 3880bb5e489..00000000000 --- a/src/tests/multihost/alltests/test_krb_ldap_connection.py +++ /dev/null @@ -1,274 +0,0 @@ -"""Automation for krb ldap connection - -:requirement: krb_ldap_connection -:casecomponent: sssd -:subsystemteam: sst_idm_sssd -:upstream: yes -:status: approved -""" -from __future__ import print_function -import subprocess -import time -import pytest -from sssd.testlib.common.utils import sssdTools -from constants import ds_instance_name - - -@pytest.mark.krbldapconnection -@pytest.mark.tier1_3 -@pytest.mark.usefixtures('setup_sssd', - 'create_posix_usersgroups', - 'krb_connection_timeout') -class TestKrbLdapConnectionTimeout(object): - """ - This is test case class for krb_ldap_connection suite - - Test connection expiration between SSSD and LDAP server. Test for - default connection expire timeout and test for some arbitrary time period, - (in our case, 100 second), invalid timeout period (-100 second), and 0 - second. - - timeouts that we are testing for. 'default' implies default timeout for - sssd which is 900 seconds/15 min. 'timeout_out_of_range' is a timeout - beyond the integer range. Also test the connection timeout between SSSD - and Kerberos server. So, sssd won't restart succesfully. 'krb' implies - connection expires when ticket expires (2 min for our test case). - Connection expires as soon as TGT expires (2 min in our case). - """ - def test_0001_timeoutdefault(self, multihost): - """ - :title: IDM-SSSD-TC: krb_provider: krb_ldap_connection: - Test if connection expires for the default value of ldap connection - timeout that is 900 seconds(15 minutes) after that it should release - the connection - :id: 53ba0b29-f5fc-4daa-8730-04a8aec91829 - """ - domain_params = {'ldap_connection_expire_timeout': None} - sssdTools( - multihost.client[0]).sssd_conf( - 'domain/%s' % - (ds_instance_name), domain_params, 'delete') - multihost.client[0].log.info( - '\n\n\nTesting for default value of ldap_' - 'connection_expire_timeout; i.e. ldap_connection_' - 'expire_timeout = default') - domain_params = { - 'ldap_uri': 'ldap://%s' % (multihost.master[0].sys_hostname)} - sssdTools( - multihost.client[0]).sssd_conf( - 'domain/%s' % - (ds_instance_name), domain_params) - self.run_test(900, multihost) - - def test_0002_timeout100(self, multihost): - """ - :title: IDM-SSSD-TC: krb_provider: krb_ldap_connection: - Test for arbitrary value ldap connection timeout that is 100 - seconds after that it should release the connection - :id: bb8dee0a-8ade-4618-b616-589bfcd46ef3 - """ - multihost.client[0].log.info( - '\n\n\nTesting for ldap_connection_expire_' - 'timeout = 100') - domain_params = { - 'ldap_uri': 'ldap://%s' % (multihost.master[0].sys_hostname)} - sssdTools( - multihost.client[0]).sssd_conf( - 'domain/%s' % - (ds_instance_name), domain_params) - domain_params = {'ldap_connection_expire_timeout': 100} - sssdTools( - multihost.client[0]).sssd_conf( - 'domain/%s' % - (ds_instance_name), domain_params) - - self.run_test(100, multihost) - - def test_0003_timeouttimeoutoutofrange(self, multihost): - """ - :title: IDM-SSSD-TC: krb_provider: krb_ldap_connection: - Test for out of range value of ldap connection timeout that - is value out of range of integer - :id: a3773739-41c7-4379-82d0-721d6993633c - :expectedresults: SSSD sevice must fail to restart - successfully after entering that value in configuration - """ - - multihost.client[0].log.info( - '\n\n\nTesting for the case where timeout value is' - 'out of range (of integer). ldap_connection_expire_' - 'timeout = timeout_out_of_range') - cmd_max_value = "echo $((`getconf INT_MAX`+1))" - cmd = multihost.client[0].run_command(cmd_max_value) - timeout = int(cmd.stdout_text.replace("\n", "")) - domain_params = { - 'ldap_uri': 'ldap://%s' % (multihost.master[0].sys_hostname)} - sssdTools( - multihost.client[0]).sssd_conf( - 'domain/%s' % - (ds_instance_name), domain_params) - domain_params = {'ldap_connection_expire_timeout': timeout} - sssdTools( - multihost.client[0]).sssd_conf( - 'domain/%s' % - (ds_instance_name), domain_params) - tools = sssdTools(multihost.client[0]) - domainname = tools.get_domain_section_name() - tools.delete_sssd_domain_log(domainname) - logfile = '/var/log/sssd/sssd_%s.log' % ds_instance_name - - clear_sssd_cache = True - try: - # stop sssd service - multihost.client[0].service_sssd('stop') - # remove sssd cache - location = '/var/lib/sss/db/' - if not sssdTools(multihost.client[0]).remove_sss_cache(location): - multihost.client[0].log.info('Failed to delete sssd cache') - assert False - cmd_start = 'systemctl start sssd' - multihost.client[0].run_command(cmd_start) - - except subprocess.CalledProcessError: - clear_sssd_cache = False - - if not clear_sssd_cache: - string = "Numerical result out of range" - file_content = multihost.client[0].get_file_contents(logfile) - x = string.encode('utf-8') in file_content - if x is True: - assert True - else: - assert False - return - - def test_0004_timeoutminus100(self, multihost): - """ - :title: IDM-SSSD-TC: krb_provider: krb_ldap_connection: - Test of invalid value of ldap connection timeout that is - -100 in our case. - :id: d68b6d42-30bd-4abb-bbf4-363388da931d - :expectedresults: It shoud instatly release the - connection after establishing - """ - multihost.client[0].log.info( - '\n\n\nTesting for ldap_connection_expire_' - 'timeout = -100') - domain_params = { - 'ldap_uri': 'ldap://%s' % (multihost.master[0].sys_hostname)} - sssdTools( - multihost.client[0]).sssd_conf( - 'domain/%s' % - (ds_instance_name), domain_params) - domain_params = {'ldap_connection_expire_timeout': -100} - sssdTools( - multihost.client[0]).sssd_conf( - 'domain/%s' % - (ds_instance_name), domain_params) - - self.run_test(-100, multihost) - - def test_0005_timeout0(self, multihost): - """ - :title: IDM-SSSD-TC: krb_provider: krb_ldap_connection: - Test for value of ldap connection timeout 0. - :id: 39af02aa-0860-4189-afc9-3ead42fd5fc1 - :expectedresults: It should have to release - the connection instantly after establishing - """ - multihost.client[0].log.info( - '\n\n\nTesting for ldap_connection_expire_' - 'timeout = 0') - domain_params = { - 'ldap_uri': 'ldap://%s' % (multihost.master[0].sys_hostname)} - sssdTools( - multihost.client[0]).sssd_conf( - 'domain/%s' % - (ds_instance_name), domain_params) - domain_params = {'ldap_connection_expire_timeout': 0} - sssdTools( - multihost.client[0]).sssd_conf( - 'domain/%s' % - (ds_instance_name), domain_params) - - self.run_test(0, multihost) - - def run_test(self, timeout, multihost): - """ - Runs the remaining test - :param str timeout:takes the vlalue of timeout for ldap - and string 'krb' in case of kerberos - :param obj multihost: multihost object - - :Steps: - 1. Setup ldap_connection_expire_timeout to a certain timeout. For - Kerberos, this is redundant as connection expires as soon as the - ticked expires. - 2. Lookup a user and get the port number and sleep for the - timeout period. - 3. Lookup another user and get the port number. - 4. Compare the 2 port numbers. - """ - tools = sssdTools(multihost.client[0]) - domainname = tools.get_domain_section_name() - tools.delete_sssd_domain_log(domainname) - logfile = '/var/log/sssd/sssd_%s.log' % ds_instance_name - - sssdTools(multihost.client[0]).clear_sssd_cache() - - if timeout == 'krb': - timeout = 120 - else: - string = "Option ldap_connection_expire_timeout has value %s" % \ - timeout - file_content = multihost.client[0].get_file_contents(logfile) - x = string.encode('utf-8') in file_content - if x is True: - assert True - else: - assert False - lookup_u = 'getent passwd foo1@%s' % ds_instance_name - cmd = multihost.client[0].run_command(lookup_u) - assert cmd.returncode == 0 - - def find_local_port(): - nsreport = multihost.client[0].run_command( - ["ss", "-ant"], log_stdout=False).stdout_text - lines = nsreport.splitlines() - lines1 = [] - - for i in lines: - if i.find('389') != -1 and i.find('ESTAB') != -1: - lines1.append(i) - del lines - - if len(lines1) > 1: - assert False - - lines1 = lines1[0] - port = lines1[ - lines1.find(':') + 1: lines1.find(' ', lines1.find(':')) - ] - return int(port) - - localport1 = find_local_port() - - time.sleep(timeout + 5) if timeout > 0 else time.sleep(5) - - lookup_u = 'getent passwd foo2@%s' % ds_instance_name - cmd = multihost.client[0].run_command(lookup_u) - assert cmd.returncode == 0 - - localport2 = find_local_port() - - assert localport1 != localport2 - if timeout > 0: - string = "Connection is about to expire, releasing it" - file_content = multihost.client[0].get_file_contents(logfile) - x = string.encode('utf-8') in file_content - if x is True: - assert True - else: - assert False - cmd_remove_log = "rm /var/log/sssd/sssd_example1.log" - multihost.client[0].run_command(cmd_remove_log) diff --git a/src/tests/system/tests/test_failover.py b/src/tests/system/tests/test_failover.py index c145b1a6b8a..85b5b004582 100644 --- a/src/tests/system/tests/test_failover.py +++ b/src/tests/system/tests/test_failover.py @@ -9,7 +9,7 @@ import pytest from sssd_test_framework.roles.client import Client from sssd_test_framework.roles.generic import GenericProvider -from sssd_test_framework.topology import KnownTopologyGroup +from sssd_test_framework.topology import KnownTopology, KnownTopologyGroup @pytest.mark.parametrize("value, expected", [(None, 31), (15, 31), (60, 60)]) @@ -61,27 +61,107 @@ def test_failover__reactivation_timeout_is_honored( ), f"'Primary server reactivation timeout set to {expected} seconds' not found in logs!" -@pytest.mark.importance("low") +# We do not authenticate the host on LDAP provider +@pytest.mark.importance("high") +@pytest.mark.ticket(bz=2466974) +@pytest.mark.topology(KnownTopology.IPA) +@pytest.mark.topology(KnownTopology.AD) +@pytest.mark.topology(KnownTopology.Samba) +@pytest.mark.preferred_topology(KnownTopology.IPA) +def test_failover__go_offline_if_kinit_fails(client: Client, provider: GenericProvider): + """ + :title: SSSD goes offline when Kerberos authentication fails + :setup: + 1. Create user + 2. Block outbound port 88 (Kerberos) + 3. Start SSSD + :steps: + 1. Try to resolve user + 2. Check domain status + :expectedresults: + 1. User is not found + 2. SSSD is offline + :customerscenario: False + """ + user = provider.user("testuser").add() + client.firewall.outbound.drop_port((88, "tcp")) + client.firewall.outbound.drop_port((88, "udp")) + client.sssd.start() + + # Make sure SSSD tries to connect + result = client.tools.id(user.name) + assert result is None, f"{user.name} was found, SSSD is not offline!" + + # SSSD was not able to connect. But check that it was actually set to offline internal state. + assert client.sssd.default_domain is not None, "No default domain?" + status = client.sssctl.domain_status(client.sssd.default_domain, online=True) + assert "Offline" in status.stdout, "SSSD is not offline!" + + +@pytest.mark.importance("high") @pytest.mark.topology(KnownTopologyGroup.AnyProvider) -def test_failover__connect_using_ipv4_second_family(client: Client, provider: GenericProvider): +@pytest.mark.preferred_topology(KnownTopology.LDAP) +def test_failover__go_offline_if_ldap_fails(client: Client, provider: GenericProvider): """ - :title: Make sure that we can connect using secondary protocol + :title: SSSD goes offline when LDAP connection fails :setup: 1. Create user - 2. Set family_order to "ipv6_first" - 3. Set IPv6 address in /etc/hosts so it resolves but it - points to non-exesting machine - 4. Start SSSD + 2. Block outbound port 389 (LDAP) + 3. Start SSSD :steps: - 1. Resolve user + 1. Try to resolve user + 2. Check domain status :expectedresults: - 1. SSSD goes online and the user is resolved + 1. User is not found + 2. SSSD is offline :customerscenario: False """ user = provider.user("testuser").add() - client.sssd.domain["lookup_family_order"] = "ipv6_first" - client.fs.append("/etc/hosts", "cafe:cafe::3 %s" % provider.host.hostname) + client.firewall.outbound.drop_port((389, "tcp")) client.sssd.start() + # Make sure SSSD tries to connect result = client.tools.id(user.name) - assert result is not None, f"{user.name} was not found, SSSD did not switch to IPv4 family!" + assert result is None, f"{user.name} was found, SSSD is not offline!" + + # SSSD was not able to connect. But check that it was actually set to offline internal state. + assert client.sssd.default_domain is not None, "No default domain?" + status = client.sssctl.domain_status(client.sssd.default_domain, online=True) + assert "Offline" in status.stdout, "SSSD is not offline!" + + +@pytest.mark.importance("high") +@pytest.mark.ticket(bz=1283798) +@pytest.mark.parametrize("method", ["su", "ssh"]) +@pytest.mark.topology(KnownTopologyGroup.AnyProvider) +@pytest.mark.preferred_topology(KnownTopology.LDAP) +def test_failover__login_via_backup_when_primary_is_unavailable( + client: Client, provider: GenericProvider, method: str +): + """ + :title: User login succeeds via backup server when primary is unavailable + :setup: + 1. Create user "user-1" + 2. Set primary server to an invalid (unreachable) server + 3. Set backup server to the real provider + 4. Start SSSD + :steps: + 1. Login as user-1 + 2. Check that SSSD is connected to the backup server + :expectedresults: + 1. User can login via the backup server + 2. SSSD is connected to the backup server + :customerscenario: True + """ + provider.user("user-1").add(password="Secret123") + client.sssd.set_invalid_primary_server(provider) + client.sssd.enable_responder("ifp") + client.sssd.start() + + assert client.auth.parametrize(method).password( + "user-1", "Secret123" + ), "User login failed, failover to backup server did not work!" + + assert client.sssd.default_domain is not None, "Default domain is not set!" + status = client.sssctl.domain_status(client.sssd.default_domain, active=True) + assert provider.host.hostname in status.stdout, f"SSSD is not connected to backup server {provider.host.hostname}!" diff --git a/src/tests/system/tests/test_ldap.py b/src/tests/system/tests/test_ldap.py index 646dffb0fc2..cb02d399a4d 100644 --- a/src/tests/system/tests/test_ldap.py +++ b/src/tests/system/tests/test_ldap.py @@ -364,3 +364,382 @@ def test_ldap__enumeration_and_group_with_hash_in_name(client: Client, ldap: LDA assert group1.name in result.stdout, f"{group1.name} is not in getent output" assert group2.name in result.stdout, f"{group2.name} is not in getent output" + + +@pytest.mark.ticket(bz=1902280) +@pytest.mark.topology(KnownTopology.LDAP) +def test_ldap__reset_cached_timestamps_to_reflect_changes(client: Client, ldap: LDAP): + """ + :title: SSSCTL cache-expire to also reset cached timestamp + :setup: + 1. Add users and groups to LDAP + 2. Configure and start SSSD + :steps: + 1. Lookup group + 2. Lookup group after clearing the cache with sssctl + :expectedresults: + 1. User is found + 2. User is not found + :customerscenario: True + """ + u = ldap.user("user1").add() + ldap.group("group1", rfc2307bis=True).add().add_member(u) + + client.sssd.domain["ldap_schema"] = "rfc2307bis" + client.sssd.domain["ldap_group_member"] = "member" + + client.sssd.start() + + res = client.tools.getent.group("group1") + assert res is not None, "Group should exist" + assert "user1" in res.members, "User should be in group" + + ldap.group("group1", rfc2307bis=True).remove_member(ldap.user("user1")) + client.sssctl.cache_expire(everything=True) + + res = client.tools.getent.group("group1") + assert res is not None, "Group should still exist" + assert "user1" not in res.members, "User should be removed from group" + + +@pytest.mark.parametrize( + ("ip_addresses", "aliases"), + [ + (["192.168.1.1"], []), + (["192.168.1.1", "192.168.1.2"], ["host1", "host2"]), + (["2001:db8:1::1", "2001:db8:1::2"], ["host1.ldap.test", "host2.ldap.test"]), + ], +) +@pytest.mark.importance("medium") +@pytest.mark.topology(KnownTopology.LDAP) +def test_ldap__resolver_provider_lookup_hosts(client: Client, ldap: LDAP, ip_addresses, aliases): + """ + :title: Resolver provider lookup hosts + :setup: + 1. Create hosts and host aliases + 2. Start SSSD + :steps: + 1. Lookup host and check for IP addresses + 2. Lookup host and check for aliases + :expectedresults: + 1. All IP addresses found + 2. All aliases found + :customerscenario: False + """ + ldap.hosts("host0").add(ip_address=ip_addresses, aliases=aliases) + client.sssd.start() + + result = client.tools.getent.hosts("host0", service="sss") + if result is not None and result.ip is not None: + for ip in ip_addresses: + assert ip in result.ip, f"Host IP addresses {ip} was not found!" + if result.aliases is not None: + for host in aliases: + assert host in result.aliases, f"'Host alias {host} for 'host0' was not found!" + else: + raise AssertionError("Hosts entry not found!") + + for alias in aliases: + result = client.tools.getent.hosts(alias, service="sss") + if result is not None and result.ip is not None: + for ip in ip_addresses: + assert ip in result.ip, f"Alias IP addresses {ip} was not found!" + + +@pytest.mark.parametrize( + "ip_addresses", + [ + ["192.168.1.1"], + ["192.168.1.1", "192.168.1.2"], + ["2001:db8:1::1", "2001:db8:1::2"], + ], +) +@pytest.mark.importance("medium") +@pytest.mark.topology(KnownTopology.LDAP) +def test_ldap__resolver_provider_lookup_hosts_by_ip(client: Client, ldap: LDAP, ip_addresses): + """ + :title: Resolver provider lookup hosts by ip + :setup: + 1. Create hosts and host aliases + 2. Start SSSD + :steps: + 1. Lookup host IP address + :expectedresults: + 1. IP addresses found + :customerscenario: False + """ + ldap.hosts("host0").add(ip_address=ip_addresses, aliases=[]) + client.sssd.start() + + for ip in ip_addresses: + result = client.tools.getent.hosts(ip, service="sss") + if result is not None and result.ip is not None: + assert ip in result.ip, f"Host IP addresses {ip} was not found!" + else: + raise AssertionError("Host entry not found by IP!") + + +@pytest.mark.topology(KnownTopology.LDAP) +@pytest.mark.importance("medium") +def test_ldap__resolver_provider_lookup_hosts_mixed_ip_versions(client: Client, ldap: LDAP): + """ + :title: Resolver provider lookup hosts with mixed ip versions + + ``getent hosts`` has two other commands, ``ahosts`` and ``ahostsv6``. + ``hosts`` is used, when results contains both ipv4 and ipv4 addresses, only ipv6 + will be returned. + + :setup: + 1. Create hosts and host aliases + 2. Start SSSD + :steps: + 1. Lookup host and check for IP addresses + 2. Lookup host and check for aliases + :expectedresults: + 1. All IP addresses found + 2. Only ipv6 aliases are found + :customerscenario: False + """ + ip_addresses = [ + "2001:db8:1::1", + "2001:db8:1::2", + "192.168.1.1", + "192.168.1.2", + ] + ipv6_addresses = ["2001:db8:1::1", "2001:db8:1::2"] + aliases = ["host1.ldap.test", "host2.ldap.test"] + + ldap.hosts("host0").add(ip_address=ip_addresses, aliases=aliases) + client.sssd.start() + + result = client.tools.getent.hosts("host0", service="sss") + if result is not None and result.ip is not None: + for ip in ipv6_addresses: + assert ip in result.ip, f"Host IPv6 address {ip} was not found!" + for alias in aliases: + if result.aliases is not None: + assert alias in result.aliases, f"'Host alias {alias} was not found!" + else: + raise AssertionError("No hosts entry found!") + + +@pytest.mark.parametrize( + ("ip_address", "aliases"), + [ + ("192.168.1.1", []), + ("192.168.2.1", ["net1", "net2"]), + ], +) +@pytest.mark.importance("medium") +@pytest.mark.topology(KnownTopology.LDAP) +def test_ldap__resolver_provider_lookup_networks(client: Client, ldap: LDAP, ip_address, aliases): + """ + :title: Resolver provider lookup networks + :setup: + 1. Create network and network aliases + 2. Start SSSD + :steps: + 1. Lookup IP addresses + 2. Lookup network + 3. Lookup network aliases + :expectedresults: + 1. IP addresses found + 2. Network found + 3. Network aliases found + :customerscenario: False + """ + ldap.networks("net0").add(ip_address=ip_address, aliases=aliases) + client.sssd.start() + + result = client.tools.getent.networks("net0", service="sss") + if result is not None and result.name is not None and result.ip is not None: + assert "net0" == result.name, "Network 'net0' was not found!" + assert all(ip in result.ip for ip in ip_address), f"Network IP addresses {ip_address} was not found!" + else: + raise AssertionError("No networks entry found!") + + result = client.tools.getent.networks(ip_address, service="sss") + if result is not None: + assert "net0" == result.name, "Network 'net0' was not found!" + if result.aliases is not None: + for network in aliases: + assert network in result.aliases, f"Network alias {network} for 'net0' was not found!" + else: + raise AssertionError("No networks entry found by IP!") + + for alias in aliases: + result = client.tools.getent.networks(alias, service="sss") + if result is not None and result.ip is not None: + assert ip_address == result.ip, f"Network IP addresses {ip_address} was not found!" + + +@pytest.mark.parametrize( + ("port", "protocol", "aliases"), + [ + (12345, "tcp", []), + (12345, "tcp", ["service1"]), + ], +) +@pytest.mark.importance("medium") +@pytest.mark.topology(KnownTopology.LDAP) +def test_ldap__resolver_provider_lookup_services(client: Client, ldap: LDAP, port, protocol, aliases): + """ + :title: Resolver provider lookup services + :setup: + 1. Create services + 2. Start SSSD + :steps: + 1. Lookup service + 2. Lookup service by alias + :expectedresults: + 1. Service found and port and protocol matches + 2. Service alias found + :customerscenario: False + """ + ldap.services("service0").add(port=port, protocol=protocol, aliases=aliases) + client.sssd.start() + + result = client.tools.getent.services("service0", service="sss") + if result is not None and result.name is not None and result.port is not None and result.protocol is not None: + assert "service0" == result.name, "Service 'service0' was not found!" + assert port == result.port, f"Service port '{str(port)}' was not found!" + assert protocol in result.protocol, f"Service protocol '{protocol}' was not found!" + for service in aliases: + assert service in aliases, f"Alias service '{service}' was not found!" + else: + raise AssertionError("No service entry found!") + + for alias in aliases: + result = client.tools.getent.services(alias, service="sss") + if result is not None and result.name is not None and result.port is not None and result.protocol is not None: + assert port == result.port, f"Alias service port '{str(port)}' port was not found!" + assert protocol == result.protocol, f"Alias service protocol '{protocol}' protocol was not found!" + + +@pytest.mark.parametrize(("port", "protocol"), [(12345, "tcp"), (12345, "udp")]) +@pytest.mark.importance("medium") +@pytest.mark.topology(KnownTopology.LDAP) +def test_ldap__resolver_provider_lookup_services_by_port(client: Client, ldap: LDAP, port, protocol): + """ + :title: Resolver provider lookup services by port + :setup: + 1. Create services + 2. Start SSSD + :steps: + 1. Lookup service by port + :expectedresults: + 1. Service found and port and protocol matches + :customerscenario: False + """ + ldap.services("service0").add(port=port, protocol=protocol, aliases=[]) + client.sssd.start() + + result = client.tools.getent.services(str(port), service="sss") + if result is not None and result.name is not None and result.port is not None and result.protocol is not None: + assert "service0" == result.name, "Service 'service0' was not found!" + assert port == result.port, f"Service port'{str(port)}' was not found!" + assert protocol in result.protocol, f"Service '{protocol}' was not found!" + else: + raise AssertionError("No service entry found!") + + +@pytest.mark.topology(KnownTopology.LDAP) +@pytest.mark.parametrize("timeout, expect_expire", [(15, True), (0, True), (-100, True)]) +@pytest.mark.importance("medium") +def test_ldap__connection_expire_timeout_releases_connection( + client: Client, ldap: LDAP, timeout: int, expect_expire: bool +): + """ + :title: LDAP connection is released after ldap_connection_expire_timeout + :setup: + 1. Create user "user-1" and "user-2" + 2. Set ldap_connection_expire_timeout to @timeout + 3. Start SSSD + :steps: + 1. Lookup user-1 to establish a connection + 2. Wait for timeout to expire (or a short period for instant-expire values) + 3. Lookup user-2 to trigger a new connection + 4. Verify the connection was recycled by checking domain logs + :expectedresults: + 1. User-1 is found + 2. Timeout period passes + 3. User-2 is found + 4. Log contains "Connection is about to expire, releasing it" for positive timeouts, + or connection is instantly recycled for zero/negative values + :customerscenario: False + """ + ldap.user("user-1").add() + ldap.user("user-2").add() + client.sssd.domain["ldap_connection_expire_timeout"] = str(timeout) + client.sssd.start() + + result = client.tools.id("user-1") + assert result is not None, "user-1 not found!" + + wait = timeout + 5 if timeout > 0 else 5 + time.sleep(wait) + + result = client.tools.id("user-2") + assert result is not None, "user-2 not found!" + + if timeout > 0: + log = client.fs.read(client.sssd.logs.domain()) + assert ( + "Connection is about to expire, releasing it" in log + ), "Connection expire message not found in domain log!" + + +@pytest.mark.topology(KnownTopology.LDAP) +@pytest.mark.importance("medium") +def test_ldap__connection_expire_timeout_out_of_range(client: Client, ldap: LDAP): + """ + :title: SSSD rejects ldap_connection_expire_timeout value out of integer range + :setup: + 1. Create user "user-1" + 2. Set ldap_connection_expire_timeout to a value exceeding INT_MAX + :steps: + 1. Start SSSD + 2. Check domain log for "Numerical result out of range" error + :expectedresults: + 1. SSSD starts but the domain may fail to initialize properly + 2. Log contains the out-of-range error message + :customerscenario: False + """ + ldap.user("user-1").add() + out_of_range = str(2**31) + client.sssd.domain["ldap_connection_expire_timeout"] = out_of_range + client.sssd.start(raise_on_error=False, check_config=False) + + log = client.fs.read(client.sssd.logs.domain()) + assert ( + "Numerical result out of range" in log + ), "Expected 'Numerical result out of range' error not found in domain log!" + + +@pytest.mark.topology(KnownTopology.LDAP) +@pytest.mark.importance("low") +def test_ldap__connection_expire_timeout_default_value_is_logged(client: Client, ldap: LDAP): + """ + :title: Default ldap_connection_expire_timeout value (900) is logged at startup + :setup: + 1. Create user "user-1" + 2. Do not set ldap_connection_expire_timeout (use default) + 3. Start SSSD + :steps: + 1. Lookup user-1 + 2. Check domain log for the default timeout value + :expectedresults: + 1. User is found + 2. Log contains "Option ldap_connection_expire_timeout has value 900" + :customerscenario: False + """ + ldap.user("user-1").add() + client.sssd.start() + + result = client.tools.id("user-1") + assert result is not None, "user-1 not found!" + + log = client.fs.read(client.sssd.logs.domain()) + assert ( + "Option ldap_connection_expire_timeout has value 900" in log + ), "Default ldap_connection_expire_timeout value (900) not found in domain log!"