intg: Add more LDAP tests

Add a bunch of LDAP tests.

    * Adding/removing a user/group/membership with rfc2307(bis) schema.
    * Filtering users/groups with rfc2307(bis) schema.
    * The effect of override_homedir option.
    * The effect of fallback_homedir option.
    * The effect of override_shell option.
    * The effect of shell_fallback option.
    * The effect of default_shell option.
    * The effect of vetoed_shells option.

1 files changed, 536 insertions, 0 deletions

diff --git a/src/tests/intg/ldap_test.py b/src/tests/intg/ldap_test.py
index a29f701c5..b20d64843 100644
--- a/src/tests/intg/ldap_test.py
+++ b/src/tests/intg/ldap_test.py
@@ -127,6 +127,23 @@ def format_basic_conf(ldap_conn, bis, enum):
     """).format(**locals())
 
 
+def format_interactive_conf(ldap_conn, bis):
+    """Format an SSSD configuration with all caches refreshing in 4 seconds"""
+    return \
+        format_basic_conf(ldap_conn, bis, True) + \
+        unindent("""
+            [nss]
+            memcache_timeout                    = 4
+            enum_cache_timeout                  = 4
+            entry_negative_timeout              = 4
+
+            [domain/LDAP]
+            ldap_enumeration_refresh_timeout    = 4
+            ldap_purge_cache_timeout            = 1
+            entry_cache_timeout                 = 4
+        """)
+
+
 def create_conf_file(contents):
     """Create sssd.conf with specified contents"""
     conf = open(config.CONF_PATH, "w")
@@ -368,3 +385,522 @@ def test_refresh_after_cleanup_task(ldap_conn, refresh_after_cleanup_task):
     ent.assert_group_by_name(
         "group2",
         dict(mem=ent.contains_only("user1")))
+
+
+@pytest.fixture
+def blank(request, ldap_conn):
+    """Create blank RFC2307 directory fixture with interactive SSSD conf"""
+    create_ldap_cleanup(request, ldap_conn)
+    create_conf_fixture(request, format_interactive_conf(ldap_conn, False))
+    create_sssd_fixture(request)
+
+
+@pytest.fixture
+def blank_bis(request, ldap_conn):
+    """Create blank RFC2307bis directory fixture with interactive SSSD conf"""
+    create_ldap_cleanup(request, ldap_conn)
+    create_conf_fixture(request, format_interactive_conf(ldap_conn, True))
+    create_sssd_fixture(request)
+
+
+@pytest.fixture
+def user_and_group(request, ldap_conn):
+    """
+    Create an RFC2307 directory fixture with interactive SSSD conf,
+    one user and one group
+    """
+    ent_list = ldap_ent.List(ldap_conn.ds_inst.base_dn)
+    ent_list.add_user("user", 1001, 2000)
+    ent_list.add_group("group", 2001)
+    create_ldap_fixture(request, ldap_conn, ent_list)
+    create_conf_fixture(request, format_interactive_conf(ldap_conn, False))
+    create_sssd_fixture(request)
+    return None
+
+
+@pytest.fixture
+def user_and_groups_bis(request, ldap_conn):
+    """
+    Create an RFC2307bis directory fixture with interactive SSSD conf,
+    one user and two groups
+    """
+    ent_list = ldap_ent.List(ldap_conn.ds_inst.base_dn)
+    ent_list.add_user("user", 1001, 2000)
+    ent_list.add_group_bis("group1", 2001)
+    ent_list.add_group_bis("group2", 2002)
+    create_ldap_fixture(request, ldap_conn, ent_list)
+    create_conf_fixture(request, format_interactive_conf(ldap_conn, True))
+    create_sssd_fixture(request)
+    return None
+
+
+def test_add_remove_user(ldap_conn, blank):
+    """Test user addition and removal are reflected by SSSD"""
+    e = ldap_ent.user(ldap_conn.ds_inst.base_dn, "user", 1001, 2000)
+    time.sleep(2)
+    # Add the user
+    ent.assert_passwd(ent.contains_only())
+    ldap_conn.add_s(*e)
+    ent.assert_passwd(ent.contains_only())
+    time.sleep(4)
+    ent.assert_passwd(ent.contains_only(dict(name="user", uid=1001)))
+    # Remove the user
+    ldap_conn.delete_s(e[0])
+    ent.assert_passwd(ent.contains_only(dict(name="user", uid=1001)))
+    time.sleep(4)
+    ent.assert_passwd(ent.contains_only())
+
+
+def test_add_remove_group(ldap_conn, blank):
+    """Test RFC2307 group addition and removal are reflected by SSSD"""
+    e = ldap_ent.group(ldap_conn.ds_inst.base_dn, "group", 2001)
+    time.sleep(2)
+    # Add the group
+    ent.assert_group(ent.contains_only())
+    ldap_conn.add_s(*e)
+    ent.assert_group(ent.contains_only())
+    time.sleep(4)
+    ent.assert_group(ent.contains_only(dict(name="group", gid=2001)))
+    # Remove the group
+    ldap_conn.delete_s(e[0])
+    ent.assert_group(ent.contains_only(dict(name="group", gid=2001)))
+    time.sleep(4)
+    ent.assert_group(ent.contains_only())
+
+
+def test_add_remove_group_bis(ldap_conn, blank_bis):
+    """Test RFC2307bis group addition and removal are reflected by SSSD"""
+    e = ldap_ent.group_bis(ldap_conn.ds_inst.base_dn, "group", 2001)
+    time.sleep(2)
+    # Add the group
+    ent.assert_group(ent.contains_only())
+    ldap_conn.add_s(*e)
+    ent.assert_group(ent.contains_only())
+    time.sleep(4)
+    ent.assert_group(ent.contains_only(dict(name="group", gid=2001)))
+    # Remove the group
+    ldap_conn.delete_s(e[0])
+    ent.assert_group(ent.contains_only(dict(name="group", gid=2001)))
+    time.sleep(4)
+    ent.assert_group(ent.contains_only())
+
+
+def test_add_remove_membership(ldap_conn, user_and_group):
+    """Test user membership addition and removal are reflected by SSSD"""
+    time.sleep(2)
+    # Add user to group
+    ent.assert_group_by_name("group", dict(mem = ent.contains_only()))
+    ldap_conn.modify_s("cn=group,ou=Groups," + ldap_conn.ds_inst.base_dn,
+                       [(ldap.MOD_REPLACE, "memberUid", "user")])
+    ent.assert_group_by_name("group", dict(mem = ent.contains_only()))
+    time.sleep(4)
+    ent.assert_group_by_name("group", dict(mem = ent.contains_only("user")))
+    # Remove user from group
+    ldap_conn.modify_s("cn=group,ou=Groups," + ldap_conn.ds_inst.base_dn,
+                       [(ldap.MOD_DELETE, "memberUid", None)])
+    ent.assert_group_by_name("group", dict(mem = ent.contains_only("user")))
+    time.sleep(4)
+    ent.assert_group_by_name("group", dict(mem = ent.contains_only()))
+
+
+def test_add_remove_membership_bis(ldap_conn, user_and_groups_bis):
+    """
+    Test user and group membership addition and removal are reflected by SSSD,
+    with RFC2307bis schema
+    """
+    time.sleep(2)
+    # Add user to group1
+    ent.assert_group_by_name("group1", dict(mem = ent.contains_only()))
+    ldap_conn.modify_s("cn=group1,ou=Groups," + ldap_conn.ds_inst.base_dn,
+                       [(ldap.MOD_REPLACE, "member",
+                         "uid=user,ou=Users," + ldap_conn.ds_inst.base_dn)])
+    ent.assert_group_by_name("group1", dict(mem = ent.contains_only()))
+    time.sleep(4)
+    ent.assert_group_by_name("group1", dict(mem = ent.contains_only("user")))
+
+    # Add group1 to group2
+    ldap_conn.modify_s("cn=group2,ou=Groups," + ldap_conn.ds_inst.base_dn,
+                       [(ldap.MOD_REPLACE, "member",
+                         "cn=group1,ou=Groups," + ldap_conn.ds_inst.base_dn)])
+    ent.assert_group_by_name("group2", dict(mem = ent.contains_only()))
+    time.sleep(4)
+    ent.assert_group_by_name("group2", dict(mem = ent.contains_only("user")))
+
+    # Remove group1 from group2
+    ldap_conn.modify_s("cn=group2,ou=Groups," + ldap_conn.ds_inst.base_dn,
+                       [(ldap.MOD_DELETE, "member", None)])
+    ent.assert_group_by_name("group2", dict(mem = ent.contains_only("user")))
+    time.sleep(4)
+    ent.assert_group_by_name("group2", dict(mem = ent.contains_only()))
+
+    # Remove user from group1
+    ldap_conn.modify_s("cn=group1,ou=Groups," + ldap_conn.ds_inst.base_dn,
+                       [(ldap.MOD_DELETE, "member", None)])
+    ent.assert_group_by_name("group1", dict(mem = ent.contains_only("user")))
+    time.sleep(4)
+    ent.assert_group_by_name("group1", dict(mem = ent.contains_only()))
+
+
+@pytest.fixture
+def blank(request, ldap_conn):
+    """Create blank RFC2307 directory fixture with interactive SSSD conf"""
+    create_ldap_cleanup(request, ldap_conn)
+    create_conf_fixture(request, format_interactive_conf(ldap_conn, False))
+    create_sssd_fixture(request)
+
+
+@pytest.fixture
+def void_conf(request):
+    create_conf_cleanup(request)
+
+
+@pytest.fixture
+def void_sssd(request):
+    create_sssd_cleanup(request)
+
+
+@pytest.fixture
+def three_users_three_groups(request, ldap_conn):
+    ent_list = ldap_ent.List(ldap_conn.ds_inst.base_dn)
+    ent_list.add_user("user1", 1001, 2001)
+    ent_list.add_user("user2", 1002, 2002)
+    ent_list.add_user("user3", 1003, 2003)
+    ent_list.add_group("group1", 2001, ["user1"])
+    ent_list.add_group("group2", 2002, ["user2"])
+    ent_list.add_group("group3", 2003, ["user3"])
+    create_ldap_fixture(request, ldap_conn, ent_list)
+
+
+def test_filter_users(request, ldap_conn, three_users_three_groups,
+                      void_conf, void_sssd):
+    """Test the effect of the "filter_users" option"""
+    all_users = frozenset([1, 2, 3])
+    for filter_users_in_groups in [False, True]:
+        for filter_users in [frozenset([]),
+                             frozenset([1]),
+                             frozenset([1, 2]),
+                             frozenset([1, 2, 3])]:
+            unfiltered_users = all_users - filter_users
+            filter_users_str = ",".join(map(lambda i: "user" + str(i),
+                                            filter_users))
+
+            conf = \
+                format_basic_conf(ldap_conn, False, True) + \
+                unindent("""
+                    [nss]
+                    filter_users            = {filter_users_str}
+                    filter_users_in_groups  = {filter_users_in_groups}
+                """).format(**locals())
+            create_conf_file(conf)
+            create_sssd_process()
+            ent.assert_passwd(
+                ent.contains_only(
+                    *map(
+                        lambda i: \
+                            dict(name = "user" + str(i), uid = 1000 + i),
+                        unfiltered_users
+                    )
+                )
+            )
+            ent.assert_group(
+                ent.contains_only(
+                    *map(
+                        lambda i: \
+                            dict(
+                                name = "group" + str(i),
+                                gid = 2000 + i,
+                                mem = ent.contains_only() \
+                                        if filter_users_in_groups and \
+                                           i in filter_users else \
+                                            ent.contains_only("user" + str(i))
+                            ),
+                        all_users
+                    )
+                )
+            )
+            cleanup_sssd_process()
+            cleanup_conf_file()
+
+
+def test_filter_groups(request, ldap_conn, three_users_three_groups,
+                       void_conf, void_sssd):
+    """Test the effect of the "filter_groups" option with RFC2307 groups"""
+    all_groups = frozenset([1, 2, 3])
+    for filter_groups in [frozenset([]),
+                          frozenset([1]),
+                          frozenset([1, 2]),
+                          frozenset([1, 2, 3])]:
+        unfiltered_groups = all_groups - filter_groups
+        filter_groups_str = ",".join(map(lambda i: "group" + str(i),
+                                         filter_groups))
+
+        conf = \
+            format_basic_conf(ldap_conn, False, True) + \
+            unindent("""
+                [nss]
+                filter_groups   = {filter_groups_str}
+            """).format(**locals())
+        create_conf_file(conf)
+        create_sssd_process()
+        ent.assert_group(
+            ent.contains_only(
+                *map(
+                    lambda i: \
+                        dict(
+                            name = "group" + str(i),
+                            gid = 2000 + i,
+                            mem = ent.contains_only("user" + str(i))
+                        ),
+                    unfiltered_groups
+                )
+            )
+        )
+        cleanup_sssd_process()
+        cleanup_conf_file()
+
+
+@pytest.fixture
+def three_users_three_groups_bis(request, ldap_conn):
+    ent_list = ldap_ent.List(ldap_conn.ds_inst.base_dn)
+    ent_list.add_user("user1", 1001, 2001)
+    ent_list.add_user("user2", 1002, 2002)
+    ent_list.add_user("user3", 1003, 2003)
+    ent_list.add_group_bis("group1", 2001, ["user1"])
+    ent_list.add_group_bis("group2", 2002, ["user2"], ["group1"])
+    ent_list.add_group_bis("group3", 2003, ["user3"], ["group2"])
+    create_ldap_fixture(request, ldap_conn, ent_list)
+
+
+def test_filter_groups_bis(request, ldap_conn, three_users_three_groups_bis,
+                           void_conf, void_sssd):
+    """Test the effect of the "filter_groups" option with RFC2307bis groups"""
+    all_groups = frozenset([1, 2, 3])
+    for filter_groups in [frozenset([]),
+                          frozenset([1]),
+                          frozenset([1, 2]),
+                          frozenset([1, 2, 3])]:
+        unfiltered_groups = all_groups - filter_groups
+        filter_groups_str = ",".join(map(lambda i: "group" + str(i),
+                                         filter_groups))
+
+        conf = \
+            format_basic_conf(ldap_conn, True, True) + \
+            unindent("""
+                [nss]
+                filter_groups   = {filter_groups_str}
+            """).format(**locals())
+        create_conf_file(conf)
+        create_sssd_process()
+        ent.assert_group(
+            ent.contains_only(
+                *map(
+                    lambda i: \
+                        dict(
+                            name = "group" + str(i),
+                            gid = 2000 + i,
+                            mem = ent.contains_only(
+                                    *map(lambda j: "user" + str(j),
+                                         range(1, i + 1)))
+                        ),
+                    unfiltered_groups
+                )
+            )
+        )
+        cleanup_sssd_process()
+        cleanup_conf_file()
+
+
+@pytest.fixture
+def override_homedir(request, ldap_conn):
+    ent_list = ldap_ent.List(ldap_conn.ds_inst.base_dn)
+    ent_list.add_user("user_with_homedir_A", 1001, 2001,
+                      homeDirectory = "/home/A")
+    ent_list.add_user("user_with_homedir_B", 1002, 2002,
+                      homeDirectory = "/home/B")
+    ent_list.add_user("user_with_empty_homedir", 1003, 2003,
+                      homeDirectory = "")
+    create_ldap_fixture(request, ldap_conn, ent_list)
+    conf = \
+        format_basic_conf(ldap_conn, False, True) + \
+        unindent("""\
+            [nss]
+            override_homedir    = /home/B
+        """).format(**locals())
+    create_conf_fixture(request, conf)
+    create_sssd_fixture(request)
+
+
+def test_override_homedir(override_homedir):
+    """Test the effect of the "override_homedir" option"""
+    ent.assert_passwd(
+        ent.contains_only(
+            dict(name="user_with_homedir_A", uid=1001, dir="/home/B"),
+            dict(name="user_with_homedir_B", uid=1002, dir="/home/B"),
+            dict(name="user_with_empty_homedir", uid=1003, dir="/home/B")
+        )
+    )
+
+
+@pytest.fixture
+def fallback_homedir(request, ldap_conn):
+    ent_list = ldap_ent.List(ldap_conn.ds_inst.base_dn)
+    ent_list.add_user("user_with_homedir_A", 1001, 2001,
+                      homeDirectory = "/home/A")
+    ent_list.add_user("user_with_homedir_B", 1002, 2002,
+                      homeDirectory = "/home/B")
+    ent_list.add_user("user_with_empty_homedir", 1003, 2003,
+                      homeDirectory = "")
+    create_ldap_fixture(request, ldap_conn, ent_list)
+    conf = \
+        format_basic_conf(ldap_conn, False, True) + \
+        unindent("""\
+            [nss]
+            fallback_homedir    = /home/B
+        """).format(**locals())
+    create_conf_fixture(request, conf)
+    create_sssd_fixture(request)
+
+
+def test_fallback_homedir(fallback_homedir):
+    """Test the effect of the "fallback_homedir" option"""
+    ent.assert_passwd(
+        ent.contains_only(
+            dict(name="user_with_homedir_A", uid=1001, dir="/home/A"),
+            dict(name="user_with_homedir_B", uid=1002, dir="/home/B"),
+            dict(name="user_with_empty_homedir", uid=1003, dir="/home/B")
+        )
+    )
+
+
+@pytest.fixture
+def override_shell(request, ldap_conn):
+    ent_list = ldap_ent.List(ldap_conn.ds_inst.base_dn)
+    ent_list.add_user("user_with_shell_A", 1001, 2001,
+                      loginShell = "/bin/A")
+    ent_list.add_user("user_with_shell_B", 1002, 2002,
+                      loginShell = "/bin/B")
+    ent_list.add_user("user_with_empty_shell", 1003, 2003,
+                      loginShell = "")
+    create_ldap_fixture(request, ldap_conn, ent_list)
+    conf = \
+        format_basic_conf(ldap_conn, False, True) + \
+        unindent("""\
+            [nss]
+            override_shell      = /bin/B
+        """).format(**locals())
+    create_conf_fixture(request, conf)
+    create_sssd_fixture(request)
+
+
+def test_override_shell(override_shell):
+    """Test the effect of the "override_shell" option"""
+    ent.assert_passwd(
+        ent.contains_only(
+            dict(name="user_with_shell_A", uid=1001, shell="/bin/B"),
+            dict(name="user_with_shell_B", uid=1002, shell="/bin/B"),
+            dict(name="user_with_empty_shell", uid=1003, shell="/bin/B")
+        )
+    )
+
+
+@pytest.fixture
+def shell_fallback(request, ldap_conn):
+    ent_list = ldap_ent.List(ldap_conn.ds_inst.base_dn)
+    ent_list.add_user("user_with_sh_shell", 1001, 2001,
+                      loginShell = "/bin/sh")
+    ent_list.add_user("user_with_not_installed_shell", 1002, 2002,
+                      loginShell = "/bin/not_installed")
+    ent_list.add_user("user_with_empty_shell", 1003, 2003,
+                      loginShell = "")
+    create_ldap_fixture(request, ldap_conn, ent_list)
+    conf = \
+        format_basic_conf(ldap_conn, False, True) + \
+        unindent("""\
+            [nss]
+            shell_fallback      = /bin/fallback
+            allowed_shells      = /bin/not_installed
+        """).format(**locals())
+    create_conf_fixture(request, conf)
+    create_sssd_fixture(request)
+
+
+def test_shell_fallback(shell_fallback):
+    """Test the effect of the "shell_fallback" option"""
+    ent.assert_passwd(
+        ent.contains_only(
+            dict(name="user_with_sh_shell", uid=1001, shell="/bin/sh"),
+            dict(name="user_with_not_installed_shell", uid=1002,
+                 shell="/bin/fallback"),
+            dict(name="user_with_empty_shell", uid=1003, shell="")
+        )
+    )
+
+
+@pytest.fixture
+def default_shell(request, ldap_conn):
+    ent_list = ldap_ent.List(ldap_conn.ds_inst.base_dn)
+    ent_list.add_user("user_with_sh_shell", 1001, 2001,
+                      loginShell = "/bin/sh")
+    ent_list.add_user("user_with_not_installed_shell", 1002, 2002,
+                      loginShell = "/bin/not_installed")
+    ent_list.add_user("user_with_empty_shell", 1003, 2003,
+                      loginShell = "")
+    create_ldap_fixture(request, ldap_conn, ent_list)
+    conf = \
+        format_basic_conf(ldap_conn, False, True) + \
+        unindent("""\
+            [nss]
+            default_shell       = /bin/default
+            allowed_shells      = /bin/default, /bin/not_installed
+            shell_fallback      = /bin/fallback
+        """).format(**locals())
+    create_conf_fixture(request, conf)
+    create_sssd_fixture(request)
+
+
+def test_default_shell(default_shell):
+    """Test the effect of the "default_shell" option"""
+    ent.assert_passwd(
+        ent.contains_only(
+            dict(name="user_with_sh_shell", uid=1001, shell="/bin/sh"),
+            dict(name="user_with_not_installed_shell", uid=1002,
+                 shell="/bin/fallback"),
+            dict(name="user_with_empty_shell", uid=1003,
+                 shell="/bin/default")
+        )
+    )
+
+
+@pytest.fixture
+def vetoed_shells(request, ldap_conn):
+    ent_list = ldap_ent.List(ldap_conn.ds_inst.base_dn)
+    ent_list.add_user("user_with_sh_shell", 1001, 2001,
+                      loginShell = "/bin/sh")
+    ent_list.add_user("user_with_vetoed_shell", 1002, 2002,
+                      loginShell = "/bin/vetoed")
+    ent_list.add_user("user_with_empty_shell", 1003, 2003,
+                      loginShell = "")
+    create_ldap_fixture(request, ldap_conn, ent_list)
+    conf = \
+        format_basic_conf(ldap_conn, False, True) + \
+        unindent("""\
+            [nss]
+            default_shell       = /bin/default
+            vetoed_shells       = /bin/vetoed
+            shell_fallback      = /bin/fallback
+        """).format(**locals())
+    create_conf_fixture(request, conf)
+    create_sssd_fixture(request)
+
+
+def test_vetoed_shells(vetoed_shells):
+    """Test the effect of the "vetoed_shells" option"""
+    ent.assert_passwd(
+        ent.contains_only(
+            dict(name="user_with_sh_shell", uid=1001, shell="/bin/sh"),
+            dict(name="user_with_vetoed_shell", uid=1002,
+                 shell="/bin/fallback"),
+            dict(name="user_with_empty_shell", uid=1003,
+                 shell="/bin/default")
+        )
+    )