# Copyright 2015 NEC Corporation # All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import subprocess import mock from oslo_config import cfg from ironic_inspector import firewall from ironic_inspector import node_cache from ironic_inspector.test import base as test_base from ironic_inspector import utils CONF = cfg.CONF @mock.patch.object(firewall, '_iptables') @mock.patch.object(utils, 'get_client') class TestFirewall(test_base.NodeTest): def test_update_filters_without_manage_firewall(self, mock_get_client, mock_iptables): CONF.set_override('manage_firewall', False, 'firewall') firewall.update_filters() self.assertEqual(0, mock_iptables.call_count) @mock.patch.object(subprocess, 'check_call') def test_init_args(self, mock_call, mock_get_client, mock_iptables): firewall.init() init_expected_args = [ ('-D', 'INPUT', '-i', 'br-ctlplane', '-p', 'udp', '--dport', '67', '-j', CONF.firewall.firewall_chain), ('-F', CONF.firewall.firewall_chain), ('-X', CONF.firewall.firewall_chain), ('-N', CONF.firewall.firewall_chain)] call_args_list = mock_iptables.call_args_list for (args, call) in zip(init_expected_args, call_args_list): self.assertEqual(args, call[0]) self.assertEqual(('iptables', '-w'), firewall.BASE_COMMAND) @mock.patch.object(subprocess, 'check_call') def test_init_args_old_iptables(self, mock_call, mock_get_client, mock_iptables): mock_call.side_effect = subprocess.CalledProcessError(2, '') firewall.init() init_expected_args = [ ('-D', 'INPUT', '-i', 'br-ctlplane', '-p', 'udp', '--dport', '67', '-j', CONF.firewall.firewall_chain), ('-F', CONF.firewall.firewall_chain), ('-X', CONF.firewall.firewall_chain), ('-N', CONF.firewall.firewall_chain)] call_args_list = mock_iptables.call_args_list for (args, call) in zip(init_expected_args, call_args_list): self.assertEqual(args, call[0]) self.assertEqual(('iptables',), firewall.BASE_COMMAND) def test_init_kwargs(self, mock_get_client, mock_iptables): firewall.init() init_expected_kwargs = [ {'ignore': True}, {'ignore': True}, {'ignore': True}] call_args_list = mock_iptables.call_args_list for (kwargs, call) in zip(init_expected_kwargs, call_args_list): self.assertEqual(kwargs, call[1]) def test_update_filters_args(self, mock_get_client, mock_iptables): firewall.init() update_filters_expected_args = [ ('-D', 'INPUT', '-i', 'br-ctlplane', '-p', 'udp', '--dport', '67', '-j', CONF.firewall.firewall_chain), ('-F', CONF.firewall.firewall_chain), ('-X', CONF.firewall.firewall_chain), ('-N', CONF.firewall.firewall_chain), ('-D', 'INPUT', '-i', 'br-ctlplane', '-p', 'udp', '--dport', '67', '-j', firewall.NEW_CHAIN), ('-F', firewall.NEW_CHAIN), ('-X', firewall.NEW_CHAIN), ('-N', firewall.NEW_CHAIN), ('-A', firewall.NEW_CHAIN, '-j', 'ACCEPT'), ('-I', 'INPUT', '-i', 'br-ctlplane', '-p', 'udp', '--dport', '67', '-j', firewall.NEW_CHAIN), ('-D', 'INPUT', '-i', 'br-ctlplane', '-p', 'udp', '--dport', '67', '-j', CONF.firewall.firewall_chain), ('-F', CONF.firewall.firewall_chain), ('-X', CONF.firewall.firewall_chain), ('-E', firewall.NEW_CHAIN, CONF.firewall.firewall_chain) ] firewall.update_filters() call_args_list = mock_iptables.call_args_list for (args, call) in zip(update_filters_expected_args, call_args_list): self.assertEqual(args, call[0]) def test_update_filters_kwargs(self, mock_get_client, mock_iptables): firewall.init() update_filters_expected_kwargs = [ {'ignore': True}, {'ignore': True}, {'ignore': True}, {}, {'ignore': True}, {'ignore': True}, {'ignore': True}, {}, {}, {}, {'ignore': True}, {'ignore': True}, {'ignore': True} ] firewall.update_filters() call_args_list = mock_iptables.call_args_list for (kwargs, call) in zip(update_filters_expected_kwargs, call_args_list): self.assertEqual(kwargs, call[1]) def test_update_filters_with_blacklist(self, mock_get_client, mock_iptables): active_macs = ['11:22:33:44:55:66', '66:55:44:33:22:11'] inactive_mac = ['AA:BB:CC:DD:EE:FF'] self.macs = active_macs + inactive_mac self.ports = [mock.Mock(address=m) for m in self.macs] mock_get_client.port.list.return_value = self.ports node_cache.add_node(self.node.uuid, mac=active_macs, bmc_address='1.2.3.4', foo=None) firewall.init() update_filters_expected_args = [ ('-D', 'INPUT', '-i', 'br-ctlplane', '-p', 'udp', '--dport', '67', '-j', CONF.firewall.firewall_chain), ('-F', CONF.firewall.firewall_chain), ('-X', CONF.firewall.firewall_chain), ('-N', CONF.firewall.firewall_chain), ('-D', 'INPUT', '-i', 'br-ctlplane', '-p', 'udp', '--dport', '67', '-j', firewall.NEW_CHAIN), ('-F', firewall.NEW_CHAIN), ('-X', firewall.NEW_CHAIN), ('-N', firewall.NEW_CHAIN), # Blacklist ('-A', firewall.NEW_CHAIN, '-m', 'mac', '--mac-source', inactive_mac[0], '-j', 'DROP'), ('-A', firewall.NEW_CHAIN, '-j', 'ACCEPT'), ('-I', 'INPUT', '-i', 'br-ctlplane', '-p', 'udp', '--dport', '67', '-j', firewall.NEW_CHAIN), ('-D', 'INPUT', '-i', 'br-ctlplane', '-p', 'udp', '--dport', '67', '-j', CONF.firewall.firewall_chain), ('-F', CONF.firewall.firewall_chain), ('-X', CONF.firewall.firewall_chain), ('-E', firewall.NEW_CHAIN, CONF.firewall.firewall_chain) ] firewall.update_filters(mock_get_client) call_args_list = mock_iptables.call_args_list for (args, call) in zip(update_filters_expected_args, call_args_list): self.assertEqual(args, call[0])