From 6fb71453306258f12ad2f77c48c02bb0144bf05c Mon Sep 17 00:00:00 2001
From: Sergey Orlov <sorlov@redhat.com>
Date: Thu, 28 Nov 2019 16:17:48 +0100
Subject: [PATCH 1/2] ipatests: add test for SSSD updating expired cache items

New test checks that sssd updates expired cahce values both for IPA
domain and trusted AD domain.

Related to: https://pagure.io/SSSD/sssd/issue/4012
---
 ipatests/test_integration/test_sssd.py | 86 +++++++++++++++++++-------
 1 file changed, 64 insertions(+), 22 deletions(-)

diff --git a/ipatests/test_integration/test_sssd.py b/ipatests/test_integration/test_sssd.py
index ac31d42104..ede514be88 100644
--- a/ipatests/test_integration/test_sssd.py
+++ b/ipatests/test_integration/test_sssd.py
@@ -8,6 +8,7 @@
 
 import time
 from contextlib import contextmanager
+import re
 
 import pytest
 
@@ -16,11 +17,7 @@
 from ipaplatform.paths import paths
 
 
-class TestSSSDAuthCache(IntegrationTest):
-    """Regression tests for cached_auth_timeout option
-
-       https://bugzilla.redhat.com/show_bug.cgi?id=1685581
-   """
+class TestSSSD(IntegrationTest):
 
     topology = 'star'
     num_ad_domains = 1
@@ -28,22 +25,22 @@ class TestSSSDAuthCache(IntegrationTest):
     users = {
         'ipa': {
             'name': 'user1',
-            'password': 'SecretUser1'
+            'password': 'SecretUser1',
+            'group': 'user1'
         },
         'ad': {
             'name_tmpl': 'testuser@{domain}',
-            'password': 'Secret123'
+            'password': 'Secret123',
+            'group_tmpl': 'testgroup@{domain}'
         },
     }
     ipa_user = 'user1'
     ipa_user_password = 'SecretUser1'
     intermed_user = 'user2'
-    ad_user_tmpl = 'testuser@{domain}'
-    ad_user_password = 'Secret123'
 
     @classmethod
     def install(cls, mh):
-        super(TestSSSDAuthCache, cls).install(mh)
+        super(TestSSSD, cls).install(mh)
 
         cls.ad = cls.ads[0]  # pylint: disable=no-member
 
@@ -53,22 +50,23 @@ def install(cls, mh):
 
         cls.users['ad']['name'] = cls.users['ad']['name_tmpl'].format(
             domain=cls.ad.domain.name)
+        cls.users['ad']['group'] = cls.users['ad']['group_tmpl'].format(
+            domain=cls.ad.domain.name)
         tasks.user_add(cls.master, cls.intermed_user)
         tasks.create_active_user(cls.master, cls.ipa_user,
                                  cls.ipa_user_password)
 
     @contextmanager
-    def config_sssd_cache_auth(self, cached_auth_timeout):
+    def config_sssd(self, opts):
         sssd_conf_backup = tasks.FileBackup(self.master, paths.SSSD_CONF)
         with tasks.remote_ini_file(self.master, paths.SSSD_CONF) as sssd_conf:
-            domain_section = 'domain/{}'.format(self.master.domain.name)
-            if cached_auth_timeout is None:
-                sssd_conf.remove_option(domain_section, 'cached_auth_timeout')
-            else:
-                sssd_conf.set(domain_section, 'cached_auth_timeout',
-                              cached_auth_timeout)
-            sssd_conf.set('pam', 'pam_verbosity', '2')
-
+            for section, option, value in opts:
+                if hasattr(section, 'name'):
+                    section = 'domain/{}'.format(section.name)
+                if value is None:
+                    sssd_conf.remove_option(section, option)
+                else:
+                    sssd_conf.set(section, option, value)
         try:
             tasks.clear_sssd_cache(self.master)
             yield
@@ -76,6 +74,9 @@ def config_sssd_cache_auth(self, cached_auth_timeout):
             sssd_conf_backup.restore()
             tasks.clear_sssd_cache(self.master)
 
+    # Check user authentication is cached for domain and its subdomains
+    # Regression tests for https://bugzilla.redhat.com/show_bug.cgi?id=1685581
+
     def is_auth_cached(self, user):
         cmd = ['su', '-l', user['name'], '-c', 'true']
         res = tasks.run_command_as_user(self.master, self.intermed_user, cmd,
@@ -84,20 +85,29 @@ def is_auth_cached(self, user):
 
     @pytest.mark.parametrize('user', ['ipa', 'ad'])
     def test_auth_cache_disabled_by_default(self, user):
-        with self.config_sssd_cache_auth(cached_auth_timeout=None):
+        sssd_options = [
+            (self.master.domain, 'cached_auth_timeout',  None),
+            ('pam', 'pam_verbosity', '2')]
+        with self.config_sssd_cache_auth(sssd_options):
             assert not self.is_auth_cached(self.users[user])
             assert not self.is_auth_cached(self.users[user])
 
     @pytest.mark.parametrize('user', ['ipa', 'ad'])
     def test_auth_cache_disabled_with_value_0(self, user):
-        with self.config_sssd_cache_auth(cached_auth_timeout=0):
+        sssd_options = [
+            (self.master.domain, 'cached_auth_timeout',  0),
+            ('pam', 'pam_verbosity', '2')]
+        with self.config_sssd_cache_auth(sssd_options):
             assert not self.is_auth_cached(self.users[user])
             assert not self.is_auth_cached(self.users[user])
 
     @pytest.mark.parametrize('user', ['ipa', 'ad'])
     def test_auth_cache_enabled_when_configured(self, user):
         timeout = 30
-        with self.config_sssd_cache_auth(cached_auth_timeout=timeout):
+        sssd_options = [
+            (self.master.domain, 'cached_auth_timeout',  timeout),
+            ('pam', 'pam_verbosity', '2')]
+        with self.config_sssd_cache_auth(sssd_options):
             start = time.time()
             # check auth is cached after first login
             assert not self.is_auth_cached(self.users[user])
@@ -108,3 +118,35 @@ def test_auth_cache_enabled_when_configured(self, user):
             assert self.is_auth_cached(self.users[user])
             time.sleep(10)
             assert not self.is_auth_cached(self.users[user])
+
+    @pytest.mark.parametrize('user_origin', ['ipa', 'ad'])
+    def test_sssd_cache_refresh(self, user_origin):
+        """Check SSSD updates expired cache items for domain and its subdomains
+
+        Regression test for https://pagure.io/SSSD/sssd/issue/4012
+        """
+        def get_cache_update_time(obj_kind, user):
+            res = self.master.run_command(
+                ['sssctl', '{}-show'.format(obj_kind), user])
+            m = re.search(r'Cache entry last update time:\s+([^\n]+)',
+                          res.stdout_text)
+            update_time = m.group(1).strip()
+            assert update_time
+            return update_time
+
+        # by design, sssd does first update of expired records in 30 seconds
+        # since start
+        refresh_time = 30
+        user = self.users[user_origin]['name']
+        group = self.users[user_origin]['group']
+        sssd_options = [
+            (self.master.domain, 'refresh_expired_interval',  1),
+            (self.master.domain, 'entry_cache_timeout',  1)]
+        with self.config_sssd_cache_auth(sssd_options):
+            start = time.time()
+            self.master.run_command(['id', user])
+            user_update_time = get_cache_update_time('user', user)
+            group_update_time = get_cache_update_time('group', group)
+            time.sleep(start + refresh_time + 5 - time.time())
+            assert user_update_time != get_cache_update_time('user', user)
+            assert group_update_time != get_cache_update_time('group', group)

From cae5809dde2291899ff623088f86ed22abee3b5e Mon Sep 17 00:00:00 2001
From: Sergey Orlov <sorlov@redhat.com>
Date: Thu, 28 Nov 2019 16:21:22 +0100
Subject: [PATCH 2/2] ipatests: refactor sssd tests for cached_auth_timeout
 option

No changes in test scenarios or coverage. Changes:
* Renamed parameter user => user_origin for consistency with new tests
* evaluate user name once per testcase to reduce copy-pastes
* evauate wait time in a more natural way
---
 ipatests/test_integration/test_sssd.py | 36 ++++++++++++++------------
 1 file changed, 19 insertions(+), 17 deletions(-)

diff --git a/ipatests/test_integration/test_sssd.py b/ipatests/test_integration/test_sssd.py
index ede514be88..05398e0695 100644
--- a/ipatests/test_integration/test_sssd.py
+++ b/ipatests/test_integration/test_sssd.py
@@ -83,41 +83,43 @@ def is_auth_cached(self, user):
                                         stdin_text=user['password'] + '\n')
         return 'Authenticated with cached credentials.' in res.stdout_text
 
-    @pytest.mark.parametrize('user', ['ipa', 'ad'])
-    def test_auth_cache_disabled_by_default(self, user):
+    @pytest.mark.parametrize('user_origin', ['ipa', 'ad'])
+    def test_auth_cache_disabled_by_default(self, user_origin):
+        user = self.users[user_origin]
         sssd_options = [
             (self.master.domain, 'cached_auth_timeout',  None),
             ('pam', 'pam_verbosity', '2')]
         with self.config_sssd_cache_auth(sssd_options):
-            assert not self.is_auth_cached(self.users[user])
-            assert not self.is_auth_cached(self.users[user])
+            assert not self.is_auth_cached(user)
+            assert not self.is_auth_cached(user)
 
-    @pytest.mark.parametrize('user', ['ipa', 'ad'])
-    def test_auth_cache_disabled_with_value_0(self, user):
+    @pytest.mark.parametrize('user_origin', ['ipa', 'ad'])
+    def test_auth_cache_disabled_with_value_0(self, user_origin):
+        user = self.users[user_origin]
         sssd_options = [
             (self.master.domain, 'cached_auth_timeout',  0),
             ('pam', 'pam_verbosity', '2')]
         with self.config_sssd_cache_auth(sssd_options):
-            assert not self.is_auth_cached(self.users[user])
-            assert not self.is_auth_cached(self.users[user])
+            assert not self.is_auth_cached(user)
+            assert not self.is_auth_cached(user)
 
-    @pytest.mark.parametrize('user', ['ipa', 'ad'])
-    def test_auth_cache_enabled_when_configured(self, user):
+    @pytest.mark.parametrize('user_origin', ['ipa', 'ad'])
+    def test_auth_cache_enabled_when_configured(self, user_origin):
         timeout = 30
+        user = self.users[user_origin]
         sssd_options = [
             (self.master.domain, 'cached_auth_timeout',  timeout),
             ('pam', 'pam_verbosity', '2')]
         with self.config_sssd_cache_auth(sssd_options):
             start = time.time()
             # check auth is cached after first login
-            assert not self.is_auth_cached(self.users[user])
-            assert self.is_auth_cached(self.users[user])
+            assert not self.is_auth_cached(user)
+            assert self.is_auth_cached(user)
             # check cache expires after configured timeout
-            elapsed = time.time() - start
-            time.sleep(timeout - 5 - elapsed)
-            assert self.is_auth_cached(self.users[user])
-            time.sleep(10)
-            assert not self.is_auth_cached(self.users[user])
+            time.sleep(start + timeout - 5 - time.time())
+            assert self.is_auth_cached(user)
+            time.sleep(start + timeout + 5 - time.time())
+            assert not self.is_auth_cached(user)
 
     @pytest.mark.parametrize('user_origin', ['ipa', 'ad'])
     def test_sssd_cache_refresh(self, user_origin):
