ipatests: refactor and extend tests for IPA-Samba integration

Add tests for following scenarios:
* running `ipa-client-samba --uninstall` without prior installation
* mount and access Samba share by IPA user
* mount and access Samba share by AD user
* mount samba share by one IPA user and access it by another one
* try mount samba share without kerberos authentication
* uninstall and reinstall ipa-client-samba

Relates: https://pagure.io/freeipa/issue/3999
Reviewed-By: Michal Polovka <mpolovka@redhat.com>
Reviewed-By: Alexander Bokovoy <abbra@users.noreply.github.com>

ipatests/prci_definitions/gating.yaml

@@ -11,6 +11,10 @@ topologies:
     name: master_1repl_1client
     cpu: 4
     memory: 7400
+  ad_master_2client: &ad_master_2client
+    name: ad_master_2client
+    cpu: 4
+    memory: 12000
 
 jobs:
   fedora-30/build:
@@ -223,13 +227,13 @@ jobs:
     requires: [fedora-30/build]
     priority: 50
     job:
-      class: RunPytest
+      class: RunADTests
       args:
         build_url: '{fedora-30/build_url}'
         test_suite: test_integration/test_smb.py
         template: *ci-master-f30
         timeout: 4800
-        topology: *master_1repl_1client
+        topology: *ad_master_2client
 
   fedora-30/replica_promotion:
     requires: [fedora-30/build]

ipatests/prci_definitions/nightly_f29.yaml

@@ -27,6 +27,10 @@ topologies:
     name: master_3repl_1client
     cpu: 6
     memory: 12900
+  ad_master_2client: &ad_master_2client
+    name: ad_master_2client
+    cpu: 4
+    memory: 12000
 
 jobs:
   fedora-29/build:
@@ -239,13 +243,13 @@ jobs:
     requires: [fedora-29/build]
     priority: 50
     job:
-      class: RunPytest
+      class: RunADtests
       args:
         build_url: '{fedora-29/build_url}'
         test_suite: test_integration/test_smb.py
         template: *ci-master-f29
         timeout: 4800
-        topology: *master_1repl_1client
+        topology: *ad_master_2client
 
   fedora-29/test_server_del:
     requires: [fedora-29/build]

ipatests/prci_definitions/nightly_master.yaml

@@ -27,6 +27,10 @@ topologies:
     name: master_3repl_1client
     cpu: 6
     memory: 12900
+  ad_master_2client: &ad_master_2client
+    name: ad_master_2client
+    cpu: 4
+    memory: 12000
 
 jobs:
   fedora-30/build:
@@ -239,13 +243,13 @@ jobs:
     requires: [fedora-30/build]
     priority: 50
     job:
-      class: RunPytest
+      class: RunADTests
       args:
         build_url: '{fedora-30/build_url}'
         test_suite: test_integration/test_smb.py
         template: *ci-master-f30
         timeout: 4800
-        topology: *master_1repl_1client
+        topology: *ad_master_2client
 
   fedora-30/test_server_del:
     requires: [fedora-30/build]

ipatests/prci_definitions/nightly_rawhide.yaml

@@ -27,6 +27,10 @@ topologies:
     name: master_3repl_1client
     cpu: 6
     memory: 12900
+  ad_master_2client: &ad_master_2client
+    name: ad_master_2client
+    cpu: 4
+    memory: 12000
 
 jobs:
   fedora-rawhide/build:
@@ -239,13 +243,13 @@ jobs:
     requires: [fedora-rawhide/build]
     priority: 50
     job:
-      class: RunPytest
+      class: RunADtest
       args:
         build_url: '{fedora-rawhide/build_url}'
         test_suite: test_integration/test_smb.py
         template: *ci-master-frawhide
         timeout: 4800
-        topology: *master_1repl_1client
+        topology: *ad_master_2client
 
   fedora-rawhide/test_server_del:
     requires: [fedora-rawhide/build]

ipatests/prci_definitions/temp_commit.yaml

@@ -33,6 +33,10 @@ topologies:
     name: master_3repl_1client
     cpu: 6
     memory: 12900
+  ad_master_2client: &ad_master_2client
+    name: ad_master_2client
+    cpu: 4
+    memory: 12000
 
 jobs:
   fedora-30/build:

ipatests/test_integration/test_smb.py

@@ -8,153 +8,327 @@
 
 from __future__ import absolute_import
 
-import time
+from functools import partial
-import os
+import textwrap
+
+import pytest
 
 from ipatests.test_integration.base import IntegrationTest
 from ipatests.pytest_ipa.integration import tasks
 from ipaplatform.paths import paths
 
-# give some time for units to stabilize
+
-# otherwise we get transient errors
+def wait_smbd_functional(host):
-WAIT_AFTER_INSTALL = 5
+    """Wait smbd is functional after (re)start
-WAIT_AFTER_UNINSTALL = WAIT_AFTER_INSTALL
+
-user_password = "Secret123"
+    After start of smbd there is a 2-3 seconds delay before daemon is
-users = {
+    fully functional and clients can successfuly mount a share.
-    "athena": "p",
+    The ping command effectively blocks until the daemon is ready.
-    "euripides": "s"
+    """
-}
+    host.run_command(['smbcontrol', 'smbd', 'ping'])
 
 
 class TestSMB(IntegrationTest):
+    topology = 'star'
+    num_clients = 2
+    num_ad_domains = 1
 
-    num_replicas = 1
+    ipa_user1 = 'user1'
-    num_clients = 1
+    ipa_user1_password = 'SecretUser1'
-
+    ipa_user2 = 'user2'
-    @classmethod
+    ipa_user2_password = 'SecretUser2'
-    def fix_resolv_conf(cls, client, server):
+    ad_user_login = 'testuser'
-
+    ad_user_password = 'Secret123'
-        contents = client.get_file_contents(paths.RESOLV_CONF,
+    ipa_test_group = 'ipa_testgroup'
-                                            encoding='utf-8')
+    ad_test_group = 'testgroup'
-        nameserver = 'nameserver %s\n' % server.ip
-        if not contents.startswith(nameserver):
-            contents = nameserver + contents.replace(nameserver, '')
-            client.run_command([
-                '/usr/bin/cp', paths.RESOLV_CONF,
-                '%s.sav' % paths.RESOLV_CONF
-            ])
-            client.put_file_contents(paths.RESOLV_CONF, contents)
-
-    @classmethod
-    def restore_resolv_conf(cls, client):
-        client.run_command([
-            '/usr/bin/cp',
-            '%s.sav' % paths.RESOLV_CONF,
-            paths.RESOLV_CONF
-        ])
 
     @classmethod
     def install(cls, mh):
-        tasks.install_master(cls.master, setup_dns=True)
+        if cls.domain_level is not None:
+            domain_level = cls.domain_level
+        else:
+            domain_level = cls.master.config.domain_level
+        tasks.install_topo(cls.topology,
+                           cls.master, cls.replicas,
+                           cls.clients, domain_level,
+                           clients_extra_args=('--mkhomedir',))
+
+        cls.ad = cls.ads[0]  # pylint: disable=no-member
+        cls.smbserver = cls.clients[0]
+        cls.smbclient = cls.clients[1]
+        cls.ad_user = '{}@{}'.format(cls.ad_user_login, cls.ad.domain.name)
+
+        tasks.config_host_resolvconf_with_master_data(cls.master,
+                                                      cls.smbclient)
         tasks.install_adtrust(cls.master)
+        tasks.configure_dns_for_trust(cls.master, cls.ad)
+        tasks.configure_windows_dns_for_trust(cls.ad, cls.master)
+        tasks.establish_trust_with_ad(cls.master, cls.ad.domain.name,
+                                      extra_args=['--two-way=true'])
 
-        for client in cls.replicas + cls.clients:
+        tasks.create_active_user(cls.master, cls.ipa_user1,
-            cls.fix_resolv_conf(client, cls.master)
+                                 password=cls.ipa_user1_password)
-            tasks.install_client(cls.master, client,
+        tasks.create_active_user(cls.master, cls.ipa_user2,
-                                 extra_args=['--mkhomedir'])
+                                 password=cls.ipa_user2_password)
+        # Trigger creation of home directories on the SMB server
+        for user in [cls.ipa_user1, cls.ipa_user2, cls.ad_user]:
+            tasks.run_command_as_user(cls.smbserver, user, ['stat', '.'])
 
-        cls.replicas[0].collect_log('/var/log/samba/')
+    @pytest.yield_fixture
-        cls.master.collect_log('/var/log/samba/')
+    def enable_smb_client_dns_lookup_kdc(self):
+        smbclient = self.smbclient
+        save_file = tasks.create_temp_file(smbclient)
+        smbclient.run_command(['cp', paths.KRB5_CONF, save_file])
+        krb5_conf = smbclient.get_file_contents(
+            paths.KRB5_CONF, encoding='utf-8')
+        krb5_conf = krb5_conf.replace(
+            'dns_lookup_kdc = false', 'dns_lookup_kdc = true')
+        smbclient.put_file_contents(paths.KRB5_CONF, krb5_conf)
+        yield
+        smbclient.run_command(['mv', save_file, paths.KRB5_CONF])
 
-    @classmethod
+    @pytest.yield_fixture
-    def uninstall(cls, mh):
+    def samba_share_public(self):
-        for client in cls.clients + cls.replicas:
+        """Setup share outside /home on samba server."""
-            tasks.uninstall_client(client)
+        share_name = 'shared'
-            cls.restore_resolv_conf(client)
+        share_path = '/srv/samba_shared'
-        tasks.uninstall_master(cls.master)
+        smbserver = self.smbserver
 
-    def test_prepare_users(self):
+        smbserver.run_command(['mkdir', share_path])
-        smbsrv = self.replicas[0]
+        smbserver.run_command(['chmod', '777', share_path])
+        # apply selinux context only if selinux is enabled
+        res = smbserver.run_command('selinuxenabled', ok_returncode=(0, 1))
+        selinux_enabled = res.returncode == 0
+        if selinux_enabled:
+            smbserver.run_command(['chcon', '-t', 'samba_share_t', share_path])
+        smbconf_save_file = tasks.create_temp_file(smbserver)
+        smbserver.run_command(['cp', paths.SMB_CONF, smbconf_save_file])
+        smb_conf = smbserver.get_file_contents(
+            paths.SMB_CONF, encoding='utf-8')
+        smb_conf += textwrap.dedent('''
+        [{name}]
+            path = {path}
+            writable = yes
+            browsable=yes
+        '''.format(name=share_name, path=share_path))
+        smbserver.put_file_contents(paths.SMB_CONF, smb_conf)
+        smbserver.run_command(['systemctl', 'restart', 'smb'])
+        wait_smbd_functional(smbserver)
+        yield {
+            'name': share_name,
+            'server_path': share_path,
+            'unc': '//{}/{}'.format(smbserver.hostname, share_name)
+        }
+        smbserver.run_command(['mv', smbconf_save_file, paths.SMB_CONF])
+        smbserver.run_command(['systemctl', 'restart', 'smb'])
+        wait_smbd_functional(smbserver)
+        smbserver.run_command(['rmdir', share_path])
 
-        temp_pass = "t3mp!p4ss"
+    def mount_smb_share(self, user, password, share, mountpoint):
-        user_kinit = "%s\n%s\n%s\n" % (temp_pass,
+        tasks.kdestroy_all(self.smbclient)
-                                       user_password, user_password)
+        tasks.kinit_as_user(self.smbclient, user, password)
-        user_addpass = "%s\n%s\n" % (temp_pass, temp_pass)
+        self.smbclient.run_command(['mkdir', '-p', mountpoint])
-        for user in users:
+        self.smbclient.run_command([
-            self.master.run_command([
+            'mount', '-t', 'cifs', share['unc'], mountpoint,
-                "ipa", "user-add",
+            '-o', 'sec=krb5i,multiuser'
-                "%s" % user, "--first", "%s" % user,
+        ])
-                "--last", "%s" % users[user],
+        tasks.kdestroy_all(self.smbclient)
-                '--password'], stdin_text=user_addpass
-            )
-            self.master.run_command(['kdestroy', '-A'])
-            self.master.run_command(
-                ['kinit', user], stdin_text=user_kinit
-            )
-            # Force creation of home directories on the SMB server
-            smbsrv.run_command(['su', '-l', '-', user, '-c', 'stat .'])
 
-            # Switch back to admin
+    def smb_sanity_check(self, user, client_mountpoint, share):
-            self.master.run_command(['kdestroy', '-A'])
+        test_dir = 'testdir_{}'.format(user)
-            tasks.kinit_admin(self.master)
+        test_file = 'testfile_{}'.format(user)
+        test_file_path = '{}/{}'.format(test_dir, test_file)
+        test_string = 'Hello, world!'
+
+        run_smb_client = partial(tasks.run_command_as_user, self.smbclient,
+                                 user, cwd=client_mountpoint)
+        run_smb_server = partial(self.smbserver.run_command,
+                                 cwd=share['server_path'])
+
+        try:
+            # check creation of directory from client side
+            run_smb_client(['mkdir', test_dir])
+            # check dir properties at client side
+            res = run_smb_client(['stat', '-c', '%n %U %G', test_dir])
+            assert res.stdout_text == '{0} {1} {1}\n'.format(test_dir, user)
+            # check dir properties at server side
+            res = run_smb_server(['stat', '-c', '%n %U %G', test_dir])
+            assert res.stdout_text == '{0} {1} {1}\n'.format(test_dir, user)
+
+            # check creation of file from client side
+            run_smb_client('printf "{}" > {}'.format(
+                test_string, test_file_path))
+            # check file is listed at client side
+            res = run_smb_client(['ls', test_dir])
+            assert res.stdout_text == test_file + '\n'
+            # check file is listed at server side
+            res = run_smb_server(['ls', test_dir])
+            assert res.stdout_text == test_file + '\n'
+            # check file properties at server side
+            res = run_smb_server(['stat', '-c', '%n %s %U %G', test_file_path])
+            assert res.stdout_text == '{0} {1} {2} {2}\n'.format(
+                test_file_path, len(test_string), user)
+            # check file properties at client side
+            res = run_smb_client(['stat', '-c', '%n %s %U %G', test_file_path])
+            assert res.stdout_text == '{0} {1} {2} {2}\n'.format(
+                test_file_path, len(test_string), user)
+            # check file contents at client side
+            res = run_smb_client(['cat', test_file_path])
+            assert res.stdout_text == test_string
+            # check file contents at server side
+            file_contents_at_server = self.smbserver.get_file_contents(
+                '{}/{}'.format(share['server_path'], test_file_path),
+                encoding='utf-8')
+            assert file_contents_at_server == test_string
+
+            # check access using smbclient utility
+            res = run_smb_client(
+                ['smbclient', '-k', share['unc'], '-c', 'dir'])
+            assert test_dir in res.stdout_text
+
+            # check file and dir removal from client side
+            run_smb_client(['rm', test_file_path])
+            run_smb_client(['rmdir', test_dir])
+            # check dir does not exist at client side
+            res = run_smb_client(['stat', test_dir], raiseonerr=False)
+            assert res.returncode == 1
+            assert 'No such file or directory' in res.stderr_text
+            # check dir does not exist at server side
+            res = run_smb_server(['stat', test_dir], raiseonerr=False)
+            assert res.returncode == 1
+            assert 'No such file or directory' in res.stderr_text
+        finally:
+            run_smb_server(['rm', '-rf', test_dir], raiseonerr=False)
+
+    def cleanup_mount(self, mountpoint):
+        self.smbclient.run_command(['umount', mountpoint], raiseonerr=False)
+        self.smbclient.run_command(['rmdir', mountpoint], raiseonerr=False)
+
+    def test_samba_uninstallation_without_installation(self):
+        res = self.smbserver.run_command(
+            ['ipa-client-samba', '--uninstall', '-U'])
+        assert res.stdout_text == 'Samba domain member is not configured yet\n'
 
     def test_install_samba(self):
-
+        self.smbserver.run_command(['ipa-client-samba', '-U'])
-        smbsrv = self.replicas[0]
+        # smb and winbind are expected to be not running
-
+        for service in ['smb', 'winbind']:
-        smbsrv.run_command([
+            result = self.smbserver.run_command(
-            "ipa-client-samba", "-U"
+                ['systemctl', 'status', service], raiseonerr=False)
+            assert result.returncode == 3
+        self.smbserver.run_command([
+            'systemctl', 'enable', '--now', 'smb', 'winbind'
         ])
+        wait_smbd_functional(self.smbserver)
+        # check that smb and winbind started successfully
+        for service in ['smb', 'winbind']:
+            self.smbserver.run_command(['systemctl', 'status', service])
+        # print status for debugging purposes
+        self.smbserver.run_command(['smbstatus'])
 
-        smbsrv.run_command([
+    def test_samba_service_listed(self):
-            "systemctl", "enable", "--now", "smb", "winbind"
+        """Check samba service is listed.
-        ])
-        time.sleep(WAIT_AFTER_INSTALL)
 
-        smbsrv.run_command(['smbstatus'])
+        Regression test for https://bugzilla.redhat.com/show_bug.cgi?id=1731433
-
-    def test_access_homes_smbclient(self):
-        """Access user home directory via smb3.ko and smbclient
-           Test checks that both kernel SMB3 driver and userspace
-           smbclient utility work against IPA-enrolled Samba server
         """
-        smbsrv = self.replicas[0]
+        service_name = 'cifs/{}@{}'.format(
-        smbclt = self.clients[0]
+            self.smbserver.hostname, self.smbserver.domain.name.upper())
+        tasks.kinit_admin(self.master)
+        res = self.master.run_command(
+            ['ipa', 'service-show', '--raw', service_name])
+        expected_output = 'krbprincipalname: {}\n'.format(service_name)
+        assert expected_output in res.stdout_text
+        res = self.master.run_command(
+            ['ipa', 'service-find', '--raw', service_name])
+        assert expected_output in res.stdout_text
 
-        remote_uri = '//{smbsrv}/homes'.format(smbsrv=smbsrv.hostname)
+    def check_smb_access_at_ipa_client(self, user, password, samba_share):
+        mount_point = '/mnt/smb'
 
-        for user in users:
+        self.mount_smb_share(user, password, samba_share, mount_point)
-            smbclt.run_command(['kinit', user], stdin_text=user_password)
+        try:
-            mntpoint = '/mnt/{user}'.format(user=user)
+            tasks.run_command_as_user(self.smbclient, user, ['kdestroy', '-A'])
-            userfile = '{user}.dat'.format(user=user)
+            tasks.run_command_as_user(self.smbclient, user, ['kinit', user],
+                                      stdin_text=password + '\n')
+            self.smb_sanity_check(user, mount_point, samba_share)
+        finally:
+            self.cleanup_mount(mount_point)
 
-            smbclt.run_command(['mkdir', '-p', mntpoint])
+    def test_smb_access_for_ipa_user_at_ipa_client(self):
-            smbclt.run_command(['mount', '-t', 'cifs',
+        samba_share = {
-                                remote_uri, mntpoint, '-o',
+            'name': 'homes',
-                                'user={user},sec=krb5i'.format(user=user)])
+            'server_path': '/home/{}'.format(self.ipa_user1),
-            smbclt.run_command(['dd', 'count=1024', 'bs=1K', 'if=/dev/zero',
+            'unc': '//{}/homes'.format(self.smbserver.hostname)
-                                'of={path}'.format(
+        }
-                                    path=os.path.join(mntpoint, userfile))])
+        self.check_smb_access_at_ipa_client(
-            smbclt.run_command(['findmnt', '-t', 'cifs'])
+            self.ipa_user1, self.ipa_user1_password, samba_share)
-            smbclt.run_command(['ls', '-laZ',
+
-                                os.path.join(mntpoint, userfile)])
+    def test_smb_access_for_ad_user_at_ipa_client(
-            smbsrv.run_command(['smbstatus'])
+            self, enable_smb_client_dns_lookup_kdc):
-            smbclt.run_command(['umount', '-a', '-t', 'cifs'])
+        samba_share = {
-            smbclt.run_command(['smbclient', '-k', remote_uri,
+            'name': 'homes',
-                                '-c', 'allinfo {path}'.format(path=userfile)])
+            'server_path': '/home/{}/{}'.format(self.ad.domain.name,
-            smbclt.run_command(['kdestroy', '-A'])
+                                                self.ad_user_login),
+            'unc': '//{}/homes'.format(self.smbserver.hostname)
+        }
+        self.check_smb_access_at_ipa_client(
+            self.ad_user, self.ad_user_password, samba_share)
+
+    def test_smb_mount_and_access_by_different_users(self, samba_share_public):
+        user1 = self.ipa_user1
+        password1 = self.ipa_user1_password
+        user2 = self.ipa_user2
+        password2 = self.ipa_user2_password
+        mount_point = '/mnt/smb'
+
+        try:
+            self.mount_smb_share(user1, password1, samba_share_public,
+                                 mount_point)
+            tasks.run_command_as_user(self.smbclient, user2,
+                                      ['kdestroy', '-A'])
+            tasks.run_command_as_user(self.smbclient, user2, ['kinit', user2],
+                                      stdin_text=password2 + '\n')
+            self.smb_sanity_check(user2, mount_point, samba_share_public)
+        finally:
+            self.cleanup_mount(mount_point)
+
+    def test_smb_mount_fails_without_kerberos_ticket(self, samba_share_public):
+        mountpoint = '/mnt/smb'
+        try:
+            tasks.kdestroy_all(self.smbclient)
+            self.smbclient.run_command(['mkdir', '-p', mountpoint])
+            res = self.smbclient.run_command([
+                'mount', '-t', 'cifs', samba_share_public['unc'], mountpoint,
+                '-o', 'sec=krb5i,multiuser'
+            ], raiseonerr=False)
+            assert res.returncode == 32
+        finally:
+            self.cleanup_mount(mountpoint)
 
     def test_uninstall_samba(self):
-        for user in users:
+        self.smbserver.run_command(['ipa-client-samba', '--uninstall', '-U'])
-            self.master.run_command(['ipa', 'user-del', user])
+        res = self.smbserver.run_command(
+            ['systemctl', 'status', 'winbind'], raiseonerr=False)
+        assert res.returncode == 3
+        res = self.smbserver.run_command(
+            ['systemctl', 'status', 'smb'], raiseonerr=False)
+        assert res.returncode == 3
 
-        smbsrv = self.replicas[0]
+    def test_repeated_uninstall_samba(self):
-        smbsrv.run_command(['ipa-client-samba', '--uninstall', '-U'])
+        """Test samba uninstallation after successful uninstallation.
-        # test for https://pagure.io/freeipa/issue/8019
+
-        # try another uninstall after the first one:
+        Test for bug https://pagure.io/freeipa/issue/8019.
-        smbsrv.run_command(['ipa-client-samba', '--uninstall', '-U'])
+        """
-        # test for https://pagure.io/freeipa/issue/8021
+        self.smbserver.run_command(['ipa-client-samba', '--uninstall', '-U'])
-        # try to install again:
+
-        smbsrv.run_command(["ipa-client-samba", "-U"])
+    def test_samba_reinstall(self):
-        # cleanup:
+        """Test samba can be reinstalled.
-        smbsrv.run_command(['ipa-client-samba', '--uninstall', '-U'])
+
+        Test installation after uninstallation and do some sanity checks.
+        Test for bug https://pagure.io/freeipa/issue/8021
+        """
+        self.test_install_samba()
+        self.test_smb_access_for_ipa_user_at_ipa_client()
+
+    def test_cleanup(self):
+        tasks.unconfigure_windows_dns_for_trust(self.ad, self.master)