# Authors: # Ana Krivokapic # # Copyright (C) 2013 Red Hat # see file 'COPYING' for use and warranty information # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . import os import subprocess from ipatests.test_integration.base import IntegrationTest from ipatests.test_integration import tasks CLIENT_KEYTAB = '/etc/krb5.keytab' class TestForcedClientReenrollment(IntegrationTest): """ Forced client re-enrollment http://www.freeipa.org/page/V3/Forced_client_re-enrollment#Test_Plan """ num_replicas = 1 num_clients = 1 @classmethod def install(cls): super(TestForcedClientReenrollment, cls).install() tasks.install_master(cls.master) tasks.install_replica(cls.master, cls.replicas[0], setup_ca=False) cls.BACKUP_KEYTAB = os.path.join( cls.master.config.test_dir, 'krb5.keytab' ) def setUp(self): tasks.prepare_host(self.clients[0]) tasks.install_client(self.master, self.clients[0]) def tearDown(self): tasks.uninstall_client(self.clients[0]) self.delete_client_host_entry() def test_reenroll_with_force_join(self): """ Client re-enrollment using admin credentials (--force-join) """ sshfp_record_pre = self.get_sshfp_record() self.restore_client() self.check_client_host_entry() self.reenroll_client(force_join=True) sshfp_record_post = self.get_sshfp_record() assert sshfp_record_pre == sshfp_record_post def test_reenroll_with_keytab(self): """ Client re-enrollment using keytab """ self.backup_keytab() sshfp_record_pre = self.get_sshfp_record() self.restore_client() self.check_client_host_entry() self.restore_keytab() self.reenroll_client(keytab=self.BACKUP_KEYTAB) sshfp_record_post = self.get_sshfp_record() assert sshfp_record_pre == sshfp_record_post def test_reenroll_with_both_force_join_and_keytab(self): """ Client re-enrollment using both --force-join and --keytab options """ self.backup_keytab() sshfp_record_pre = self.get_sshfp_record() self.restore_client() self.check_client_host_entry() self.restore_keytab() self.reenroll_client(force_join=True, keytab=self.BACKUP_KEYTAB) sshfp_record_post = self.get_sshfp_record() assert sshfp_record_pre == sshfp_record_post def test_reenroll_to_replica(self): """ Client re-enrollment using keytab, to a replica """ self.backup_keytab() sshfp_record_pre = self.get_sshfp_record() self.restore_client() self.check_client_host_entry() self.restore_keytab() self.reenroll_client(keytab=self.BACKUP_KEYTAB, to_replica=True) sshfp_record_post = self.get_sshfp_record() assert sshfp_record_pre == sshfp_record_post def test_try_to_reenroll_with_disabled_host(self): """ Client re-enrollment using keytab, with disabled host """ self.backup_keytab() self.disable_client_host_entry() self.restore_client() self.check_client_host_entry(enabled=False) self.restore_keytab() self.reenroll_client(keytab=self.BACKUP_KEYTAB, expect_fail=True) def test_try_to_reenroll_with_uninstalled_host(self): """ Client re-enrollment using keytab, with uninstalled host """ self.backup_keytab() self.uninstall_client() self.restore_client() self.check_client_host_entry(enabled=False) self.restore_keytab() self.reenroll_client(keytab=self.BACKUP_KEYTAB, expect_fail=True) def test_try_to_reenroll_with_deleted_host(self): """ Client re-enrollment using keytab, with deleted host """ self.backup_keytab() self.delete_client_host_entry() self.restore_client() self.check_client_host_entry(not_found=True) self.restore_keytab() self.reenroll_client(keytab=self.BACKUP_KEYTAB, expect_fail=True) def test_try_to_reenroll_with_incorrect_keytab(self): """ Client re-enrollment using keytab, with incorrect keytab file """ EMPTY_KEYTAB = os.path.join( self.clients[0].config.test_dir, 'empty.keytab' ) self.restore_client() self.check_client_host_entry() self.clients[0].run_command(['touch', EMPTY_KEYTAB]) self.reenroll_client(keytab=EMPTY_KEYTAB, expect_fail=True) def uninstall_client(self): self.clients[0].run_command( ['ipa-client-install', '--uninstall', '-U'], set_env=False, raiseonerr=False ) def restore_client(self): client = self.clients[0] client.run_command([ 'iptables', '-A', 'INPUT', '-j', 'ACCEPT', '-p', 'tcp', '--dport', '22' ]) client.run_command([ 'iptables', '-A', 'INPUT', '-j', 'REJECT', '-p', 'all', '--source', self.master.ip ]) self.uninstall_client() client.run_command(['iptables', '-F']) def reenroll_client(self, keytab=None, to_replica=False, force_join=False, expect_fail=False): server = self.replicas[0] if to_replica else self.master client = self.clients[0] self.fix_resolv_conf(client, server) args = [ 'ipa-client-install', '-U', '--server', server.hostname, '--domain', server.domain.name ] if force_join: args.append('--force-join') if keytab: args.extend(['--keytab', keytab]) else: args.extend([ '-p', client.config.admin_name, '-w', client.config.admin_password ]) result = client.run_command( args, set_env=False, raiseonerr=not expect_fail ) assert 'IPA Server: %s' % server.hostname in result.stderr_text if expect_fail: err_msg = 'Kerberos authentication failed using keytab' assert result.returncode == 1 assert err_msg in result.stderr_text elif force_join and keytab: warn_msg = ("Option 'force-join' has no additional effect " "when used with together with option 'keytab'.") assert warn_msg in result.stderr_text def check_client_host_entry(self, enabled=True, not_found=False): result = self.master.run_command( ['ipa', 'host-show', self.clients[0].hostname], raiseonerr=not not_found ) if not_found: assert result.returncode == 2 assert 'host not found' in result.stderr_text elif enabled: assert 'Certificate:' in result.stdout_text assert 'Keytab: True' in result.stdout_text else: assert 'Certificate:' not in result.stdout_text assert 'Keytab: False' in result.stdout_text def disable_client_host_entry(self): self.master.run_command( ['ipa', 'host-disable', self.clients[0].hostname] ) def delete_client_host_entry(self): try: self.master.run_command( ['ipa', 'host-del', self.clients[0].hostname] ) except subprocess.CalledProcessError as e: if e.returncode != 2: raise def get_sshfp_record(self): sshfp_record = '' client_host = self.clients[0].hostname.split('.')[0] result = self.master.run_command( ['ipa', 'dnsrecord-show', self.master.domain.name, client_host] ) lines = result.stdout_text.splitlines() for line in lines: if 'SSHFP record:' in line: sshfp_record = line.replace('SSHFP record:', '').strip() assert sshfp_record, 'SSHFP record not found' sshfp_record = set(sshfp_record.split(', ')) self.log.debug("SSHFP record for host %s: %s", client_host, str(sshfp_record)) return sshfp_record def backup_keytab(self): contents = self.clients[0].get_file_contents(CLIENT_KEYTAB) self.master.put_file_contents(self.BACKUP_KEYTAB, contents) def restore_keytab(self): contents = self.master.get_file_contents(self.BACKUP_KEYTAB) self.clients[0].put_file_contents(self.BACKUP_KEYTAB, contents) def fix_resolv_conf(self, client, server): """ Put server's ip address at the top of resolv.conf """ contents = client.get_file_contents('/etc/resolv.conf') nameserver = 'nameserver %s\n' % server.ip if not contents.startswith(nameserver): contents = nameserver + contents.replace(nameserver, '') client.put_file_contents('/etc/resolv.conf', contents)