| # Licensed to the Apache Software Foundation (ASF) under one | |
| # or more contributor license agreements. See the NOTICE file | |
| # distributed with this work for additional information | |
| # regarding copyright ownership. The ASF licenses this file | |
| # to you 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. | |
| """Smoke tests for firewall rules on VPC public IPs.""" | |
| from nose.plugins.attrib import attr | |
| from marvin.cloudstackTestCase import cloudstackTestCase | |
| from marvin.lib.base import Account, FireWallRule, Network, NetworkOffering, PublicIPAddress, VPC, VpcOffering | |
| from marvin.lib.common import get_domain, get_zone, list_publicIP | |
| from marvin.lib.utils import cleanup_resources, wait_until | |
| class TestVpcFirewallRules(cloudstackTestCase): | |
| @classmethod | |
| def setUpClass(cls): | |
| cls.testClient = super(TestVpcFirewallRules, cls).getClsTestClient() | |
| cls.apiclient = cls.testClient.getApiClient() | |
| cls.services = cls.testClient.getParsedTestDataConfig() | |
| cls.zone = get_zone(cls.apiclient, cls.testClient.getZoneForTests()) | |
| cls.domain = get_domain(cls.apiclient) | |
| cls._cleanup = [] | |
| cls.account = Account.create( | |
| cls.apiclient, | |
| cls.services["account"], | |
| domainid=cls.domain.id | |
| ) | |
| cls._cleanup.append(cls.account) | |
| cls.services["vpc_offering"]["supportedservices"] = ( | |
| "Vpn,Dhcp,Dns,SourceNat,Lb,UserData,StaticNat," | |
| "NetworkACL,PortForwarding,Firewall" | |
| ) | |
| cls.services["vpc_offering"]["serviceProviderList"] = { | |
| "Vpn": "VpcVirtualRouter", | |
| "Dhcp": "VpcVirtualRouter", | |
| "Dns": "VpcVirtualRouter", | |
| "SourceNat": "VpcVirtualRouter", | |
| "Lb": "VpcVirtualRouter", | |
| "UserData": "VpcVirtualRouter", | |
| "StaticNat": "VpcVirtualRouter", | |
| "NetworkACL": "VpcVirtualRouter", | |
| "PortForwarding": "VpcVirtualRouter", | |
| "Firewall": "VpcVirtualRouter" | |
| } | |
| cls.vpc_offering = VpcOffering.create( | |
| cls.apiclient, | |
| cls.services["vpc_offering"] | |
| ) | |
| cls.vpc_offering.update(cls.apiclient, state="Enabled") | |
| cls._cleanup.append(cls.vpc_offering) | |
| network_offering = NetworkOffering.list( | |
| cls.apiclient, | |
| name="DefaultIsolatedNetworkOfferingForVpcNetworks" | |
| ) | |
| cls.assertTrue( | |
| network_offering is not None and len(network_offering) > 0, | |
| "No VPC tier network offering found" | |
| ) | |
| cls.network_offering = network_offering[0] | |
| cls.services["vpc"]["cidr"] = "10.20.30.0/24" | |
| cls.vpc = VPC.create( | |
| cls.apiclient, | |
| cls.services["vpc"], | |
| vpcofferingid=cls.vpc_offering.id, | |
| zoneid=cls.zone.id, | |
| account=cls.account.name, | |
| domainid=cls.account.domainid | |
| ) | |
| cls.tier = Network.create( | |
| cls.apiclient, | |
| services={ | |
| "name": "vpc-fw-tier", | |
| "displaytext": "vpc-fw-tier" | |
| }, | |
| accountid=cls.account.name, | |
| domainid=cls.account.domainid, | |
| networkofferingid=cls.network_offering.id, | |
| zoneid=cls.zone.id, | |
| vpcid=cls.vpc.id, | |
| gateway="10.20.30.1", | |
| netmask="255.255.255.0" | |
| ) | |
| @classmethod | |
| def tearDownClass(cls): | |
| try: | |
| cleanup_resources(cls.apiclient, cls._cleanup) | |
| except Exception as e: | |
| raise Exception("Warning: Exception during cleanup: %s" % e) | |
| def setUp(self): | |
| self.apiclient = self.testClient.getApiClient() | |
| self.cleanup = [] | |
| def tearDown(self): | |
| cleanup_resources(self.apiclient, self.cleanup) | |
| def _wait_for_firewall_rule(self, rule_id): | |
| rules = FireWallRule.list(self.apiclient, id=rule_id, listall=True) | |
| if rules and len(rules) == 1: | |
| return True, rules[0] | |
| return False, None | |
| @attr(tags=["advanced", "advancedns", "smoke"], required_hardware="false") | |
| def test_01_create_firewall_rule_on_vpc_public_ip(self): | |
| """Verify firewall rule can be created and listed on a dedicated VPC public IP.""" | |
| public_ip = PublicIPAddress.create( | |
| self.apiclient, | |
| zoneid=self.zone.id, | |
| accountid=self.account.name, | |
| domainid=self.account.domainid, | |
| vpcid=self.vpc.id | |
| ) | |
| self.cleanup.append(public_ip) | |
| firewall_rule = FireWallRule.create( | |
| self.apiclient, | |
| ipaddressid=public_ip.ipaddress.id, | |
| protocol="tcp", | |
| cidrlist=["0.0.0.0/0"], | |
| startport=19090, | |
| endport=19090, | |
| vpcid=self.vpc.id | |
| ) | |
| self.cleanup.insert(0, firewall_rule) | |
| result, listed_rule = wait_until(2, 10, self._wait_for_firewall_rule, firewall_rule.id) | |
| self.assertTrue(result, "Firewall rule was not listed for the VPC public IP") | |
| self.assertEqual(listed_rule.id, firewall_rule.id) | |
| self.assertEqual(listed_rule.ipaddressid, public_ip.ipaddress.id) | |
| self.assertEqual(listed_rule.vpcid, self.vpc.id) | |
| self.assertEqual(listed_rule.protocol.lower(), "tcp") | |
| self.assertEqual(int(listed_rule.startport), 19090) | |
| self.assertEqual(int(listed_rule.endport), 19090) | |
| @attr(tags=["advanced", "advancedns", "smoke"], required_hardware="false") | |
| def test_02_create_firewall_rule_on_vpc_source_nat_ip(self): | |
| """Verify firewall rule can be created and listed on the VPC source NAT IP.""" | |
| source_nat_ips = list_publicIP( | |
| self.apiclient, | |
| vpcid=self.vpc.id, | |
| listall=True, | |
| issourcenat=True | |
| ) | |
| self.assertTrue(source_nat_ips is not None and len(source_nat_ips) > 0, | |
| "No source NAT IP found for the VPC") | |
| source_nat_ip = source_nat_ips[0] | |
| firewall_rule = FireWallRule.create( | |
| self.apiclient, | |
| ipaddressid=source_nat_ip.id, | |
| protocol="tcp", | |
| cidrlist=["0.0.0.0/0"], | |
| startport=19443, | |
| endport=19443, | |
| vpcid=self.vpc.id | |
| ) | |
| self.cleanup.append(firewall_rule) | |
| result, listed_rule = wait_until(2, 10, self._wait_for_firewall_rule, firewall_rule.id) | |
| self.assertTrue(result, "Firewall rule was not listed for the VPC source NAT IP") | |
| self.assertEqual(listed_rule.id, firewall_rule.id) | |
| self.assertEqual(listed_rule.ipaddressid, source_nat_ip.id) | |
| self.assertEqual(listed_rule.vpcid, self.vpc.id) | |
| self.assertEqual(listed_rule.protocol.lower(), "tcp") | |
| self.assertEqual(int(listed_rule.startport), 19443) | |
| self.assertEqual(int(listed_rule.endport), 19443) |