From 857713c5a9c8e0b62c06dd92e69c09eeb34b2e99 Mon Sep 17 00:00:00 2001

From: Anuja More <amore@redhat.com>

Date: Mon, 23 May 2022 12:26:34 +0530

Subject: [PATCH] Add end to end integration tests for external IdP

Added tests for HBAC and SUDO rule and other

test scenarios.

Related : https://pagure.io/freeipa/issue/8805

Related: https://pagure.io/freeipa/issue/8803

Related: https://pagure.io/freeipa/issue/8804

Signed-off-by: Anuja More <amore@redhat.com>

Reviewed-By: Florence Blanc-Renaud <flo@redhat.com>

Reviewed-By: Florence Blanc-Renaud <flo@redhat.com>

---

 ipatests/test_integration/test_idp.py | 260 ++++++++++++++++++++++----

 1 file changed, 226 insertions(+), 34 deletions(-)

diff --git a/ipatests/test_integration/test_idp.py b/ipatests/test_integration/test_idp.py

index 8f9e92e6a..2ffe6a208 100644

--- a/ipatests/test_integration/test_idp.py

+++ b/ipatests/test_integration/test_idp.py

@@ -1,6 +1,8 @@

 from __future__ import absolute_import

 import time

+import pytest

+import re

 import textwrap

 from ipaplatform.paths import paths

@@ -22,12 +24,12 @@ driver.get(verification_uri)

 try:

     element = WebDriverWait(driver, 90).until(

         EC.presence_of_element_located((By.ID, "username")))

-    driver.find_element_by_id("username").send_keys("testuser1")

-    driver.find_element_by_id("password").send_keys("{passwd}")

-    driver.find_element_by_id("kc-login").click()

+    driver.find_element(By.ID, "username").send_keys("testuser1")

+    driver.find_element(By.ID, "password").send_keys("{passwd}")

+    driver.find_element(By.ID, "kc-login").click()

     element = WebDriverWait(driver, 90).until(

         EC.presence_of_element_located((By.ID, "kc-login")))

-    driver.find_element_by_id("kc-login").click()

+    driver.find_element(By.ID, "kc-login").click()

     assert "Device Login Successful" in driver.page_source

 finally:

     now = datetime.now().strftime("%M-%S")

@@ -39,18 +41,12 @@ finally:

 def add_user_code(host, verification_uri):

     contents = user_code_script.format(uri=verification_uri,

                                        passwd=host.config.admin_password)

-    host.put_file_contents("/tmp/add_user_code.py", contents)

-    tasks.run_repeatedly(

-        host, ['python3', '/tmp/add_user_code.py'])

-

-

-def get_verification_uri(host, since, keycloak_server_name):

-    command = textwrap.dedent("""

-    journalctl -u ipa-otpd\\* --since="%s" | grep "user_code:" | awk '{ print substr($7,2,9) }'""" % since)  # noqa: E501

-    user_code = host.run_command(command).stdout_text.rstrip("\r\n")

-    uri = ("https://{0}:8443/auth/realms/master/device?user_code={1}".format(

-        keycloak_server_name, user_code))

-    return uri

+    try:

+        host.put_file_contents("/tmp/add_user_code.py", contents)

+        tasks.run_repeatedly(

+            host, ['python3', '/tmp/add_user_code.py'])

+    finally:

+        host.run_command(["rm", "-f", "/tmp/add_user_code.py"])

 def kinit_idp(host, user, keycloak_server):

@@ -58,11 +54,14 @@ def kinit_idp(host, user, keycloak_server):

     tasks.kdestroy_all(host)

     # create armor for FAST

     host.run_command(["kinit", "-n", "-c", ARMOR])

-    since = time.strftime('%Y-%m-%d %H:%M:%S')

     cmd = ["kinit", "-T", ARMOR, user]

+

     with host.spawn_expect(cmd, default_timeout=100) as e:

-        e.expect('Authenticate at .+: ')

-        uri = get_verification_uri(host, since, keycloak_server.hostname)

+        e.expect('Authenticate at (.+) and press ENTER.:')

+        prompt = e.get_last_output()

+        uri = re.search(r'Authenticate at (.*?) and press ENTER.:', prompt

+                        ).group(1)

+        time.sleep(15)

         if uri:

             add_user_code(keycloak_server, uri)

         e.sendline('\n')

@@ -74,21 +73,27 @@ def kinit_idp(host, user, keycloak_server):

 class TestIDPKeycloak(IntegrationTest):

-    num_replicas = 1

+    num_replicas = 2

     topology = 'line'

     @classmethod

     def install(cls, mh):

-        tasks.install_master(cls.master, setup_dns=True)

-        tasks.install_client(cls.master, cls.replicas[0])

-        content = cls.master.get_file_contents(paths.IPA_DEFAULT_CONF,

-                                               encoding='utf-8')

-        new_content = content + "\noidc_child_debug_level = 10"

-        cls.master.put_file_contents(paths.IPA_DEFAULT_CONF, new_content)

+        cls.client = cls.replicas[0]

+        cls.replica = cls.replicas[1]

+        tasks.install_master(cls.master)

+        tasks.install_client(cls.master, cls.replicas[0],

+                             extra_args=["--mkhomedir"])

+        tasks.install_replica(cls.master, cls.replicas[1])

+        for host in [cls.master, cls.replicas[0], cls.replicas[1]]:

+            content = host.get_file_contents(paths.IPA_DEFAULT_CONF,

+                                             encoding='utf-8')

+            new_content = content + "\noidc_child_debug_level = 10"

+            host.put_file_contents(paths.IPA_DEFAULT_CONF, new_content)

         with tasks.remote_sssd_config(cls.master) as sssd_config:

             sssd_config.edit_domain(

                 cls.master.domain, 'krb5_auth_timeout', 1100)

         tasks.clear_sssd_cache(cls.master)

+        tasks.clear_sssd_cache(cls.replicas[0])

         tasks.kinit_admin(cls.master)

         cls.master.run_command(["ipa", "config-mod", "--user-auth-type=idp",

                                 "--user-auth-type=password"])

@@ -97,20 +102,207 @@ class TestIDPKeycloak(IntegrationTest):

         cls.replicas[0].run_command(xvfb)

     def test_auth_keycloak_idp(self):

-        keycloak_srv = self.replicas[0]

-        create_quarkus.setup_keycloakserver(keycloak_srv)

+        """

+        Test case to check that OAuth 2.0 Device

+        Authorization Grant is working as

+        expected for user configured with external idp.

+        """

+        create_quarkus.setup_keycloakserver(self.client)

         time.sleep(60)

-        create_quarkus.setup_keycloak_client(keycloak_srv)

+        create_quarkus.setup_keycloak_client(self.client)

         tasks.kinit_admin(self.master)

-        cmd = ["ipa", "idp-add", "keycloak", "--provider=keycloak",

+        cmd = ["ipa", "idp-add", "keycloakidp", "--provider=keycloak",

                "--client-id=ipa_oidc_client", "--org=master",

-               "--base-url={0}:8443/auth".format(keycloak_srv.hostname)]

+               "--base-url={0}:8443/auth".format(self.client.hostname)]

         self.master.run_command(cmd, stdin_text="{0}\n{0}".format(

-            keycloak_srv.config.admin_password))

+            self.client.config.admin_password))

         tasks.user_add(self.master, 'keycloakuser',

                        extra_args=["--user-auth-type=idp",

                                    "--idp-user-id=testuser1@ipa.test",

-                                   "--idp=keycloak"]

+                                   "--idp=keycloakidp"]

                        )

+        list_user = self.master.run_command(

+            ["ipa", "user-find", "--idp-user-id=testuser1@ipa.test"]

+        )

+        assert "keycloakuser" in list_user.stdout_text

+        list_by_idp = self.master.run_command(["ipa", "user-find",

+                                               "--idp=keycloakidp"]

+                                              )

+        assert "keycloakuser" in list_by_idp.stdout_text

+        list_by_user = self.master.run_command(

+            ["ipa", "user-find", "--idp-user-id=testuser1@ipa.test", "--all"]

+        )

+        assert "keycloakidp" in list_by_user.stdout_text

+        tasks.clear_sssd_cache(self.master)

+        kinit_idp(self.master, 'keycloakuser', keycloak_server=self.client)

+

+    @pytest.fixture

+    def hbac_setup_teardown(self):

+        # allow sshd only on given host

+        tasks.kinit_admin(self.master)

+        self.master.run_command(["ipa", "hbacrule-disable", "allow_all"])

+        self.master.run_command(["ipa", "hbacrule-add", "rule1"])

+        self.master.run_command(["ipa", "hbacrule-add-user", "rule1",

+                                 "--users=keycloakuser"]

+                                )

+        self.master.run_command(["ipa", "hbacrule-add-host", "rule1",

+                                 "--hosts", self.replica.hostname])

+        self.master.run_command(["ipa", "hbacrule-add-service", "rule1",

+                                 "--hbacsvcs=sshd"]

+                                )

+        tasks.clear_sssd_cache(self.master)

+        tasks.clear_sssd_cache(self.replica)

+        yield

+

+        # cleanup

+        tasks.kinit_admin(self.master)

+        self.master.run_command(["ipa", "hbacrule-enable", "allow_all"])

+        self.master.run_command(["ipa", "hbacrule-del", "rule1"])

+

+    def test_auth_hbac(self, hbac_setup_teardown):

+        """

+        Test case to check that hbacrule is working as

+        expected for user configured with external idp.

+        """

+        kinit_idp(self.master, 'keycloakuser', keycloak_server=self.client)

+        ssh_cmd = "ssh -q -K -l keycloakuser {0} whoami"

+        valid_ssh = self.master.run_command(

+            ssh_cmd.format(self.replica.hostname))

+        assert "keycloakuser" in valid_ssh.stdout_text

+        negative_ssh = self.master.run_command(

+            ssh_cmd.format(self.master.hostname), raiseonerr=False

+        )

+        assert negative_ssh.returncode == 255

+

+    def test_auth_sudo_idp(self):

+        """

+        Test case to check that sudorule is working as

+        expected for user configured with external idp.

+        """

+        tasks.kdestroy_all(self.master)

+        tasks.kinit_admin(self.master)

+        #  rule: keycloakuser are allowed to execute yum on

+        #  the client machine as root.

+        cmdlist = [

+            ["ipa", "sudocmd-add", "/usr/bin/yum"],

+            ["ipa", "sudorule-add", "sudorule"],

+            ['ipa', 'sudorule-add-user', '--users=keycloakuser',

+             'sudorule'],

+            ['ipa', 'sudorule-add-host', '--hosts',

+             self.client.hostname, 'sudorule'],

+            ['ipa', 'sudorule-add-runasuser',

+             '--users=root', 'sudorule'],

+            ['ipa', 'sudorule-add-allow-command',

+             '--sudocmds=/usr/bin/yum', 'sudorule'],

+            ['ipa', 'sudorule-show', 'sudorule', '--all'],

+            ['ipa', 'sudorule-add-option',

+             'sudorule', '--sudooption', "!authenticate"]

+        ]

+        for cmd in cmdlist:

+            self.master.run_command(cmd)

+        tasks.clear_sssd_cache(self.master)

+        tasks.clear_sssd_cache(self.client)

+        try:

+            cmd = 'sudo -ll -U keycloakuser'

+            test = self.client.run_command(cmd).stdout_text

+            assert "User keycloakuser may run the following commands" in test

+            assert "/usr/bin/yum" in test

+            kinit_idp(self.client, 'keycloakuser', self.client)

+            test_sudo = 'su -c "sudo yum list wget" keycloakuser'

+            self.client.run_command(test_sudo)

+            list_fail = self.master.run_command(cmd).stdout_text

+            assert "User keycloakuser is not allowed to run sudo" in list_fail

+        finally:

+            tasks.kinit_admin(self.master)

+            self.master.run_command(['ipa', 'sudorule-del', 'sudorule'])

+            self.master.run_command(["ipa", "sudocmd-del", "/usr/bin/yum"])

+

+    def test_auth_replica(self):

+        """

+        Test case to check that OAuth 2.0 Device

+        Authorization is working as expected on replica.

+        """

+        tasks.clear_sssd_cache(self.master)

+        tasks.clear_sssd_cache(self.replica)

+        tasks.kinit_admin(self.replica)

+        list_user = self.master.run_command(

+            ["ipa", "user-find", "--idp-user-id=testuser1@ipa.test"]

+        )

+        assert "keycloakuser" in list_user.stdout_text

+        list_by_idp = self.replica.run_command(["ipa", "user-find",

+                                                "--idp=keycloakidp"]

+                                               )

+        assert "keycloakuser" in list_by_idp.stdout_text

+        list_by_user = self.replica.run_command(

+            ["ipa", "user-find", "--idp-user-id=testuser1@ipa.test", "--all"]

+        )

+        assert "keycloakidp" in list_by_user.stdout_text

+        kinit_idp(self.replica, 'keycloakuser', keycloak_server=self.client)

+

+    def test_idp_with_services(self):

+        """

+        Test case to check that services can be configured

+        auth indicator as idp.

+        """

         tasks.clear_sssd_cache(self.master)

-        kinit_idp(self.master, 'keycloakuser', keycloak_srv)

+        tasks.kinit_admin(self.master)

+        domain = self.master.domain.name.upper()

+        services = [

+            "DNS/{0}@{1}".format(self.master.hostname, domain),

+            "HTTP/{0}@{1}".format(self.client.hostname, domain),

+            "dogtag/{0}@{1}".format(self.master.hostname, domain),

+            "ipa-dnskeysyncd/{0}@{1}".format(self.master.hostname, domain)

+        ]

+        try:

+            for service in services:

+                test = self.master.run_command(["ipa", "service-mod", service,

+                                                "--auth-ind=idp"]

+                                               )

+                assert "Authentication Indicators: idp" in test.stdout_text

+        finally:

+            for service in services:

+                self.master.run_command(["ipa", "service-mod", service,

+                                         "--auth-ind="])

+

+    def test_idp_backup_restore(self):

+        """

+        Test case to check that after restore data is retrieved

+        with related idp configuration.

+        """

+        tasks.kinit_admin(self.master)

+        user = "backupuser"

+        cmd = ["ipa", "idp-add", "testidp", "--provider=keycloak",

+               "--client-id=ipa_oidc_client", "--org=master",

+               "--base-url={0}:8443/auth".format(self.client.hostname)]

+        self.master.run_command(cmd, stdin_text="{0}\n{0}".format(

+            self.client.config.admin_password))

+

+        tasks.user_add(self.master, user,

+                       extra_args=["--user-auth-type=idp",

+                                   "--idp-user-id=testuser1@ipa.test",

+                                   "--idp=testidp"]

+                       )

+

+        backup_path = tasks.get_backup_dir(self.master)

+        # change data after backup

+        self.master.run_command(['ipa', 'user-del', user])

+        self.master.run_command(['ipa', 'idp-del', 'testidp'])

+        dirman_password = self.master.config.dirman_password

+        self.master.run_command(['ipa-restore', backup_path],

+                                stdin_text=dirman_password + '\nyes')

+        try:

+            list_user = self.master.run_command(

+                ['ipa', 'user-show', 'backupuser', '--all']

+            ).stdout_text

+            assert "External IdP configuration: testidp" in list_user

+            assert "User authentication types: idp" in list_user

+            assert ("External IdP user identifier: "

+                    "testuser1@ipa.test") in list_user

+            list_idp = self.master.run_command(['ipa', 'idp-find', 'testidp'])

+            assert "testidp" in list_idp.stdout_text

+            kinit_idp(self.master, user, self.client)

+        finally:

+            tasks.kdestroy_all(self.master)

+            tasks.kinit_admin(self.master)

+            self.master.run_command(["rm", "-rf", backup_path])

+            self.master.run_command(["ipa", "idp-del", "testidp"])

-- 

2.36.1