From 2b79fa82872c55368167fc7433cb28a2369f5191 Mon Sep 17 00:00:00 2001 From: Todd Willey Date: Thu, 7 Apr 2011 01:42:49 -0400 Subject: test provider fw rules at the virt/ipteables layer. lowercase protocol names in admin api to match what the firewall driver expects. add provider fw rule chain in iptables6 as well. fix a couple of small typos and copy-paste errors. --- nova/tests/test_virt.py | 65 ++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 64 insertions(+), 1 deletion(-) (limited to 'nova/tests') diff --git a/nova/tests/test_virt.py b/nova/tests/test_virt.py index 958c8e3e2..34ecd09c6 100644 --- a/nova/tests/test_virt.py +++ b/nova/tests/test_virt.py @@ -566,7 +566,9 @@ class IptablesFirewallTestCase(test.TestCase): self.network = utils.import_object(FLAGS.network_manager) class FakeLibvirtConnection(object): - pass + def nwfilterDefineXML(*args, **kwargs): + """setup_basic_rules in nwfilter calls this.""" + pass self.fake_libvirt_connection = FakeLibvirtConnection() self.fw = libvirt_conn.IptablesFirewallDriver( get_connection=lambda: self.fake_libvirt_connection) @@ -728,6 +730,67 @@ class IptablesFirewallTestCase(test.TestCase): "TCP port 80/81 acceptance rule wasn't added") db.instance_destroy(admin_ctxt, instance_ref['id']) + def test_provider_firewall_rules(self): + # keep from changing state of actual firewall + #def fake_function(*args, **kwargs): + # pass + #self.fw.iptables.apply = fake_function + #self.fw.nwfilter.setup_basic_filtering = fake_function + + # setup basic instance data + instance_ref = db.instance_create(self.context, + {'user_id': 'fake', + 'project_id': 'fake', + 'mac_address': '56:12:12:12:12:12'}) + ip = '10.11.12.13' + network_ref = db.project_get_network(self.context, 'fake') + admin_ctxt = context.get_admin_context() + fixed_ip = {'address': ip, 'network_id': network_ref['id']} + db.fixed_ip_create(admin_ctxt, fixed_ip) + db.fixed_ip_update(admin_ctxt, ip, {'allocated': True, + 'instance_id': instance_ref['id']}) + # FRAGILE: peeks at how the firewall names chains + chain_name = 'inst-%s' % instance_ref['id'] + + # create a firewall via setup_basic_filtering like libvirt_conn.spawn + # should have a chain with 0 rules + self.fw.setup_basic_filtering(instance_ref, network_info=None) + self.assertTrue('provider' in self.fw.iptables.ipv4['filter'].chains) + rules = [rule for rule in self.fw.iptables.ipv4['filter'].rules + if rule.chain == 'provider'] + self.assertEqual(0, len(rules)) + + # add a rule and send the update message, check for 1 rule + provider_fw0 = db.provider_fw_rule_create(admin_ctxt, + {'protocol': 'tcp', + 'cidr': '10.99.99.99/32', + 'from_port': 1, + 'to_port': 65535}) + self.fw.refresh_provider_fw_rules() + rules = [rule for rule in self.fw.iptables.ipv4['filter'].rules + if rule.chain == 'provider'] + self.assertEqual(1, len(rules)) + + # Add another, refresh, and make sure number of rules goes to two + provider_fw1 = db.provider_fw_rule_create(admin_ctxt, + {'protocol': 'udp', + 'cidr': '10.99.99.99/32', + 'from_port': 1, + 'to_port': 65535}) + self.fw.refresh_provider_fw_rules() + rules = [rule for rule in self.fw.iptables.ipv4['filter'].rules + if rule.chain == 'provider'] + self.assertEqual(2, len(rules)) + + # create the instance filter and make sure it has a jump rule + self.fw.prepare_instance_filter(instance_ref, network_info=None) + self.fw.apply_instance_filter(instance_ref) + inst_rules = [rule for rule in self.fw.iptables.ipv4['filter'].rules + if rule.chain == chain_name] + jump_rules = [rule for rule in inst_rules if '-j' in rule.rule] + prov_rules = [rule for rule in jump_rules if 'provider' in rule.rule] + self.assertEqual(1, len(prov_rules)) + class NWFilterTestCase(test.TestCase): def setUp(self): -- cgit