freeipa/ipatests/test_integration/test_http_kdc_proxy.py

#
# Copyright (C) 2016  FreeIPA Contributors see COPYING for license
#

from __future__ import absolute_import

import re
from contextlib import contextmanager

import pytest

from ipatests.pytest_ipa.integration import tasks
from ipatests.pytest_ipa.integration.firewall import Firewall
from ipatests.test_integration.base import IntegrationTest
from ipaplatform.paths import paths


class TestHttpKdcProxy(IntegrationTest):
    topology = "line"
    num_clients = 1
    num_ad_domains = 1

    @classmethod
    def install(cls, mh):
        super().install(mh)

        cls.client = cls.clients[0]
        cls.ad = cls.ads[0]

        tasks.kinit_admin(cls.master)
        cls.master.run_command(['ipa', 'pwpolicy-mod', '--minlife=0'])

        tasks.install_adtrust(cls.master)
        tasks.configure_dns_for_trust(cls.master, cls.ad)
        tasks.establish_trust_with_ad(cls.master, cls.ad.domain.name)

    @classmethod
    def uninstall(cls, mh):
        tasks.remove_trust_info_from_ad(
            cls.master, cls.ad.domain.name, cls.ad.hostname)
        super().uninstall(mh)

    @pytest.fixture(autouse=True, scope='function')
    def cleanup_credentials(self):
        tasks.kdestroy_all(self.client)
        tasks.clear_sssd_cache(self.client)

    @pytest.fixture(scope='class')
    def users(self, mh):
        master = mh.master
        ad = mh.ads[0]
        users = {
            'ipa': {
                'name': 'ipa_test_user',
                'password': 'SecretIpaTestUser',
                'domain': mh.master.domain,
                'test_service': 'HTTP/{}@{}'
                .format(master.hostname, master.domain.realm),
            },
            'ad': {
                'name': 'testuser@{}'.format(ad.domain.realm),
                'password': 'Secret123',
                'domain': ad.domain,
                'test_service': 'HTTP/{}@{}'
                .format(ad.hostname, ad.domain.realm),
            }
        }
        tasks.kinit_admin(mh.master)
        tasks.create_active_user(
            mh.master, users['ipa']['name'], users['ipa']['password'])
        yield users
        tasks.kinit_admin(mh.master)
        mh.master.run_command(['ipa', 'user-del', users['ipa']['name']])

    @pytest.fixture()
    def restrict_network_for_client(self, mh):
        fw_rules_allow = [
            ['OUTPUT', '-p', 'udp', '--dport', '53', '-j', 'ACCEPT'],
            ['OUTPUT', '-p', 'tcp', '--dport', '80', '-j', 'ACCEPT'],
            ['OUTPUT', '-p', 'tcp', '--dport', '443', '-j', 'ACCEPT'],
            ['OUTPUT', '-p', 'tcp', '--sport', '22', '-j', 'ACCEPT']]
        fw = Firewall(self.client)
        fw.prepend_passthrough_rules(fw_rules_allow)
        fw.passthrough_rule(['-P', 'OUTPUT', 'DROP'])
        yield
        fw.passthrough_rule(['-P', 'OUTPUT', 'ACCEPT'])
        fw.remove_passthrough_rules(fw_rules_allow)

    @pytest.fixture()
    def client_use_kdcproxy(self, mh):
        """Configure client for using kdcproxy for IPA and AD domains."""
        def replace_regexp_once(pattern, repl, string):
            res, n = re.subn(pattern, repl, string)
            assert n == 1
            return res

        krb5conf_backup = tasks.FileBackup(self.client, paths.KRB5_CONF)
        krb5conf = self.client.get_file_contents(
            paths.KRB5_CONF, encoding='utf-8')
        kdc_url = 'https://{}/KdcProxy'.format(self.master.hostname)

        # configure kdc proxy for IPA realm
        krb5conf = replace_regexp_once(
            r' kdc = .+', ' kdc = {}'.format(kdc_url), krb5conf)
        krb5conf = replace_regexp_once(
            r'kpasswd_server = .+', 'kpasswd_server = {}'.format(kdc_url),
            krb5conf)

        # configure kdc proxy for Windows AD realm
        ad_realm_config = '''
        {realm} = {{
            kdc = {kdc_url}
            kpasswd_server = {kdc_url}
        }}
        '''.format(realm=self.ad.domain.realm, kdc_url=kdc_url)
        krb5conf = replace_regexp_once(
            r'\[realms\]',
            '[realms]' + ad_realm_config,
            krb5conf
        )

        self.client.put_file_contents(paths.KRB5_CONF, krb5conf)
        self.client.run_command(['systemctl', 'restart', 'sssd.service'])
        yield
        krb5conf_backup.restore()
        self.client.run_command(['systemctl', 'restart', 'sssd.service'])

    @contextmanager
    def configure_kdc_proxy_for_ad_trust(self, use_tcp):
        backup = tasks.FileBackup(self.master, paths.KDCPROXY_CONFIG)
        with tasks.remote_ini_file(self.master, paths.KDCPROXY_CONFIG) as conf:
            conf.set('global', 'use_dns', 'true')
            conf.set('global', 'configs', 'mit')
            if use_tcp:
                conf.add_section(self.ad.domain.realm)
                conf.set(self.ad.domain.realm, 'kerberos',
                         'kerberos+tcp://{}:88'.format(self.ad.hostname))
                conf.set(self.ad.domain.realm, 'kpasswd',
                         'kpasswd+tcp://{}:464'.format(self.ad.hostname))
        try:
            self.master.run_command(['ipactl', 'restart'])
            yield
        finally:
            backup.restore()
            self.master.run_command(['ipactl', 'restart'])

    def check_kerberos_requests(self, user, skip_kpasswd_check=False):
        # KDC AS request
        tasks.kinit_as_user(self.client, user['name'], user['password'])

        # KDC TGS requests
        self.client.run_command(['kvno', user['test_service']])

        # KDC AS requests and kpasswd requests

        # Changing password on Windows AD can not be done now because
        # of default password policy mandating that minimal password lifetime
        # is one day.
        # Once we switch to dynamically creating test users in Windows AD
        # and create an utility for modifying Group Policy Objects then
        # we should update test setup and remove this condition.
        def set_password(old_pass, new_pass):
            with self.client.spawn_expect(['kpasswd', user['name']]) as e:
                e.expect('Password for .+:')
                e.sendline(old_pass)
                e.expect_exact('Enter new password:')
                e.sendline(new_pass)
                e.expect_exact('Enter it again:')
                e.sendline(new_pass)
                e.expect_exit(ignore_remaining_output=True)

        if not skip_kpasswd_check:
            test_password = 'Secret123456'
            set_password(user['password'], test_password)
            # Restore password:
            set_password(test_password, user['password'])

    @pytest.mark.parametrize('user_origin', ['ipa', 'ad'])
    def test_user_login_on_client_without_firewall(self, users, user_origin):
        """Basic check for test setup."""
        self.check_kerberos_requests(users[user_origin],
                                     skip_kpasswd_check=user_origin == 'ad')

    @pytest.mark.usefixtures('restrict_network_for_client')
    @pytest.mark.parametrize('user_origin', ['ipa', 'ad'])
    def test_access_blocked_on_client_without_kdcproxy(
            self, users, user_origin):
        """Check for test firewall setup."""
        user = users[user_origin]
        result = tasks.kinit_as_user(
            self.client, user['name'], user['password'], raiseonerr=False)
        expected_errors = [
            ("Cannot contact any KDC for realm '{}' while getting initial "
             "credentials".format(user['domain'].realm)),
            ('Cannot find KDC for realm "{}" while getting initial '
             "credentials".format(user['domain'].realm)),
        ]
        assert (result.returncode == 1
                and any(s in result.stderr_text for s in expected_errors))

    @pytest.mark.usefixtures('restrict_network_for_client',
                             'client_use_kdcproxy')
    def test_ipa_user_login_on_client_with_kdcproxy(self, users):
        self.check_kerberos_requests(users['ipa'])

    @pytest.mark.usefixtures('restrict_network_for_client',
                             'client_use_kdcproxy')
    @pytest.mark.parametrize('use_tcp', [True, False])
    def test_ad_user_login_on_client_with_kdcproxy(self, users, use_tcp):
        with self.configure_kdc_proxy_for_ad_trust(use_tcp):
            self.check_kerberos_requests(users['ad'], skip_kpasswd_check=True)

    @pytest.fixture()
    def windows_small_mtu_size(self, mh):
        new_mtu = 70

        def get_iface_name():
            result = self.ad.run_command([
                'powershell', '-c',
                '(Get-NetIPAddress -IPAddress {}).InterfaceAlias'.format(
                    self.ad.ip)])
            return result.stdout_text.strip()

        def get_mtu(iface_name):
            result = self.ad.run_command([
                'netsh', 'interface', 'ipv4', 'show', 'subinterface',
                iface_name])
            mtu = result.stdout_text.strip().splitlines()[-1].split()[0]
            return int(mtu)

        def set_mtu(iface_name, mtu):
            self.ad.run_command([
                'netsh', 'interface', 'ipv4', 'set', 'subinterface',
                iface_name, 'mtu={}'.format(mtu)])

        iface_name = get_iface_name()
        original_mtu = get_mtu(iface_name)
        set_mtu(iface_name, new_mtu)
        # `netsh` does not report failures with return code so we check
        # it was successful by inspecting the actual value of MTU
        assert get_mtu(iface_name) == new_mtu
        yield
        set_mtu(iface_name, original_mtu)
        assert get_mtu(iface_name) == original_mtu

    @pytest.mark.usefixtures('restrict_network_for_client',
                             'client_use_kdcproxy',
                             'windows_small_mtu_size')
    def test_kdcproxy_handles_small_packets_from_ad(self, users):
        """Check that kdcproxy handles AD response split to several TCP packets

        This is a regression test for the bug in python-kdcproxy:
        https://github.com/latchset/kdcproxy/pull/44
          When the reply from AD is split into several TCP packets the kdc
          proxy software cannot handle it and returns a false error message
          indicating it cannot contact the KDC server.
        """
        with self.configure_kdc_proxy_for_ad_trust(use_tcp=True):
            self.check_kerberos_requests(users['ad'], skip_kpasswd_check=True)