>From ef38ade73f23becfb3006ff68dd3a06c4ab62549 Mon Sep 17 00:00:00 2001
From: Nikolai Kondrashov <Nikolai.Kondrashov@redhat.com>
Date: Tue, 29 Sep 2015 21:18:18 +0300
Subject: [PATCH] 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.
---
 src/tests/intg/ldap_test.py | 539 +++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 537 insertions(+), 2 deletions(-)

diff --git a/src/tests/intg/ldap_test.py b/src/tests/intg/ldap_test.py
index 6d61726..40f9009 100644
--- a/src/tests/intg/ldap_test.py
+++ b/src/tests/intg/ldap_test.py
@@ -127,6 +127,22 @@ def format_basic_conf(ldap_conn, schema, enum):
     """).format(**locals())
 
 
+def format_interactive_conf(ldap_conn, schema):
+    """Format an SSSD configuration with all caches refreshing in 4 seconds"""
+    return \
+        format_basic_conf(ldap_conn, schema, enum=True) + \
+        unindent("""
+            [nss]
+            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")
@@ -177,8 +193,10 @@ def cleanup_sssd_process():
     subprocess.call(["sss_cache", "-E"])
     for path in os.listdir(config.DB_PATH):
         os.unlink(config.DB_PATH + "/" + path)
-    for path in os.listdir(config.MCACHE_PATH):
-        os.unlink(config.MCACHE_PATH + "/" + path)
+    # FIXME: Uncomment when removed cache invalidation is fixed:
+    # https://fedorahosted.org/sssd/ticket/2726
+    # for path in os.listdir(config.MCACHE_PATH):
+    #    os.unlink(config.MCACHE_PATH + "/" + path)
 
 
 def create_sssd_cleanup(request):
@@ -389,3 +407,520 @@ 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_rfc2307(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, SCHEMA_RFC2307))
+    create_sssd_fixture(request)
+
+
+@pytest.fixture
+def blank_rfc2307_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, SCHEMA_RFC2307_BIS))
+    create_sssd_fixture(request)
+
+
+@pytest.fixture
+def user_and_group_rfc2307(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, SCHEMA_RFC2307))
+    create_sssd_fixture(request)
+    return None
+
+
+@pytest.fixture
+def user_and_groups_rfc2307_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, SCHEMA_RFC2307_BIS))
+    create_sssd_fixture(request)
+    return None
+
+
+def test_add_remove_user(ldap_conn, blank_rfc2307):
+    """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_rfc2307(ldap_conn, blank_rfc2307):
+    """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_rfc2307_bis(ldap_conn, blank_rfc2307_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_rfc2307(ldap_conn, user_and_group_rfc2307):
+    """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_rfc2307_bis(ldap_conn,
+                                           user_and_groups_rfc2307_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 void_conf(request):
+    create_conf_cleanup(request)
+
+
+@pytest.fixture
+def void_sssd(request):
+    create_sssd_cleanup(request)
+
+
+@pytest.fixture
+def three_users_three_groups_rfc2307(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_rfc2307,
+                      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, SCHEMA_RFC2307, enum=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_rfc2307(request, ldap_conn,
+                               three_users_three_groups_rfc2307,
+                               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, SCHEMA_RFC2307, enum=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_rfc2307_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_rfc2307_bis(request, ldap_conn,
+                                   three_users_three_groups_rfc2307_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, SCHEMA_RFC2307_BIS, enum=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, SCHEMA_RFC2307, enum=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, SCHEMA_RFC2307, enum=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, SCHEMA_RFC2307, enum=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, SCHEMA_RFC2307, enum=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, SCHEMA_RFC2307, enum=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, SCHEMA_RFC2307, enum=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")
+        )
+    )
-- 
2.6.1

