| # 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. |
| |
| import os |
| import subprocess |
| import socket |
| |
| from nose.plugins.attrib import attr |
| |
| from marvin.cloudstackTestCase import cloudstackTestCase |
| from marvin.lib.base import (Account, ServiceOffering, VirtualMachine, NetworkOffering, Network |
| , SecurityGroup) |
| from marvin.cloudstackAPI import (authorizeSecurityGroupIngress |
| , authorizeSecurityGroupEgress, revokeSecurityGroupIngress, revokeSecurityGroupEgress) |
| from marvin.lib.common import get_zone, get_template |
| from marvin.lib.decoratorGenerators import skipTestIf |
| from .common import cleanup_resources, not_tungsten_fabric_zone |
| |
| |
| class Services: |
| """Test Tungsten Plugin |
| """ |
| |
| def __init__(self): |
| self.services = { |
| "network_offering": { |
| "name": "TungstenSharedNetworkOffering", |
| "displaytext": "TungstenSharedNetworkOffering", |
| "guestiptype": "Shared", |
| "traffictype": "GUEST", |
| "supportedservices": [ |
| "Dhcp", |
| "Dns", |
| "Connectivity", |
| "SecurityGroup" |
| ], |
| "serviceProviderList": { |
| "Dhcp": "Tungsten", |
| "Dns": "Tungsten", |
| "Connectivity": "Tungsten", |
| "SecurityGroup": "Tungsten" |
| }, |
| "useTungsten": "on", |
| "specifyVlan": "true", |
| "specifyIpRanges": "true" |
| }, |
| "networks": { |
| "network1": { |
| "name": "TF-Network1", |
| "displaytext": "TF-Network1", |
| "gateway": "192.168.6.1", |
| "netmask": "255.255.255.0", |
| "startip": "192.168.6.100", |
| "endip": "192.168.6.200", |
| "vlan": "untagged" |
| } |
| }, |
| "account": { |
| "email": "test@test.com", |
| "firstname": "Test", |
| "lastname": "User", |
| "username": "test", |
| "password": "password", |
| }, |
| "virtual_machines": { |
| "vm1": { |
| "name": "vm1", |
| "displayname": "vm1", |
| "username": "root", |
| "password": "password", |
| "ssh_port": 22, |
| "hypervisor": 'KVM', |
| "privateport": 22, |
| "publicport": 22, |
| "protocol": 'TCP' |
| |
| }, |
| "vm2": { |
| "name": "vm2", |
| "displayname": "vm2", |
| "username": "root", |
| "password": "password", |
| "ssh_port": 22, |
| "hypervisor": 'KVM', |
| "privateport": 22, |
| "publicport": 22, |
| "protocol": 'TCP' |
| }, |
| "vm3": { |
| "name": "vm3", |
| "displayname": "vm3", |
| "username": "root", |
| "password": "password", |
| "ssh_port": 22, |
| "hypervisor": 'KVM', |
| "privateport": 22, |
| "publicport": 22, |
| "protocol": 'TCP' |
| } |
| }, |
| "security_groups": { |
| "security_group1": { |
| "name": "sg1" |
| }, |
| "security_group2": { |
| "name": "sg2" |
| }, |
| "security_group3": { |
| "name": "sg3" |
| } |
| }, |
| "ostype": 'CentOS 5.6 (64-bit)' |
| } |
| |
| |
| class TestSharedZone(cloudstackTestCase): |
| @classmethod |
| def setUpClass(cls): |
| cls.api_client = cls.clstestclient.getApiClient() |
| cls.services = Services().services |
| cls.test_data = cls.clstestclient.getParsedTestDataConfig() |
| cls.zone = get_zone(cls.api_client, cls.clstestclient.getZoneForTests()) |
| cls.template = get_template(cls.api_client, cls.zone.id, cls.test_data["ostype"]) |
| cls.not_tungsten_fabric_zone = not_tungsten_fabric_zone(cls.api_client, cls.zone.id) |
| cls._cleanup = [] |
| |
| @classmethod |
| def tearDownClass(self): |
| try: |
| cleanup_resources(self.api_client, self.zone.id, self._cleanup) |
| except Exception as e: |
| raise Exception("Warning: Exception during cleanup : %s" % e) |
| return |
| |
| def setUp(self): |
| self.account = Account.create(self.api_client, self.services["account"], admin=True) |
| self.service_offering = ServiceOffering.create(self.api_client, self.test_data["service_offering"]) |
| self.cleanup = [self.account, self.service_offering] |
| return |
| |
| def tearDown(self): |
| try: |
| cleanup_resources(self.api_client, self.zone.id, reversed(self.cleanup)) |
| except Exception as e: |
| raise Exception("Warning: Exception during cleanup : %s" % e) |
| return |
| |
| def getLocalMachineIpAddress(self): |
| """ Get IP address of the machine on which test case is running """ |
| socket_ = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) |
| socket_.connect(('8.8.8.8', 0)) |
| return socket_.getsockname()[0] |
| |
| @skipTestIf("not_tungsten_fabric_zone") |
| @attr(tags=["tungstenfabric"], required_hardware="false") |
| def test_01_test_default_security_group_rule(self): |
| |
| # Validate the following: |
| # 1. deploy vm |
| # 2. ping vm from test machine ( should fail as default security group rule ) |
| # 3. create icmp ingress security group rule with test machine ip address cidr |
| # 4. ping vm from test machine again ( should success as icmp ingress security group rule) |
| # 5. create ssh ingress security group rule with test machine ip address cidr |
| # 6. ssh to vm from test machine |
| # 7. ping google from vm ( should success as default security group rule ) |
| |
| network_offering = NetworkOffering.create(self.api_client, self.services["network_offering"]) |
| network_offering.update(self.api_client, state='Enabled') |
| self.cleanup.append(network_offering) |
| |
| # create network |
| network = Network.create( |
| self.api_client, |
| self.services["networks"]["network1"], |
| networkofferingid=network_offering.id, |
| zoneid=self.zone.id, |
| accountid=self.account.name, |
| domainid=self.account.domainid |
| ) |
| self.cleanup.append(network) |
| |
| # create security group |
| security_group = SecurityGroup.create( |
| self.api_client, |
| self.services["security_groups"]["security_group1"], |
| account=self.account.name, |
| domainid=self.account.domainid |
| ) |
| self.cleanup.append(security_group) |
| |
| # create virtual machine |
| vm = VirtualMachine.create( |
| self.api_client, |
| self.services["virtual_machines"]["vm1"], |
| serviceofferingid=self.service_offering.id, |
| templateid=self.template.id, |
| networkids=[network.id], |
| zoneid=self.zone.id, |
| accountid=self.account.name, |
| domainid=self.account.domainid, |
| securitygroupids=[security_group.id], |
| mode='advanced' |
| ) |
| self.cleanup.append(vm) |
| |
| # ping vm from test machine |
| if os.name == 'nt': |
| result = subprocess.call( |
| ['ping', '-n', '1', vm.ipaddress]) |
| else: |
| result = subprocess.call( |
| ['ping', '-c', '1', vm.ipaddress]) |
| |
| self.assertEqual( |
| result, |
| 1, |
| "Ping should be fail as default security group deny all incoming traffic" |
| ) |
| |
| test_machine_ip_address = self.getLocalMachineIpAddress() |
| cidr = test_machine_ip_address + "/32" |
| |
| # create icmp ingress security group rule with test machine ip address cidr |
| cmd = authorizeSecurityGroupIngress.authorizeSecurityGroupIngressCmd() |
| cmd.securitygroupid = security_group.id |
| cmd.protocol = 'ICMP' |
| cmd.icmptype = "-1" |
| cmd.icmpcode = "-1" |
| cmd.cidrlist = cidr |
| icmp_ingress_rule = self.api_client.authorizeSecurityGroupIngress(cmd) |
| |
| # ping vm from test machine again |
| if os.name == 'nt': |
| result = subprocess.call( |
| ['ping', '-n', '1', vm.ipaddress]) |
| else: |
| result = subprocess.call( |
| ['ping', '-c', '1', vm.ipaddress]) |
| |
| self.assertEqual( |
| result, |
| 0, |
| "Ping should be successful" |
| ) |
| |
| # create ssh ingress security group rule with test machine ip address cidr |
| cmd = authorizeSecurityGroupIngress.authorizeSecurityGroupIngressCmd() |
| cmd.securitygroupid = security_group.id |
| cmd.protocol = 'TCP' |
| cmd.startport = 22 |
| cmd.endport = 22 |
| cmd.cidrlist = cidr |
| ssh_ingress_rule = self.api_client.authorizeSecurityGroupIngress(cmd) |
| |
| # ssh to vm from test machine |
| ssh = vm.get_ssh_client() |
| |
| # ping google from vm |
| res = ssh.execute("ping -c1 google.com") |
| self.assertEqual( |
| str(res).count("1 received"), |
| 1, |
| "Ping google.com from vm should be successful " |
| "as default security group permit all outcoming traffic" |
| ) |
| |
| # revoke icmp ingress security group rule |
| cmd = revokeSecurityGroupIngress.revokeSecurityGroupIngressCmd() |
| cmd.id = icmp_ingress_rule.ingressrule[0].ruleid |
| self.api_client.revokeSecurityGroupIngress(cmd) |
| |
| # revoke ssh ingress security group rule |
| cmd = revokeSecurityGroupIngress.revokeSecurityGroupIngressCmd() |
| cmd.id = ssh_ingress_rule.ingressrule[0].ruleid |
| self.api_client.revokeSecurityGroupIngress(cmd) |
| |
| @skipTestIf("not_tungsten_fabric_zone") |
| @attr(tags=["tungstenfabric"], required_hardware="false") |
| def test_01_test_ingress_rule_security_group_with_cidr(self): |
| |
| # Validate the following: |
| # 1. deploy 2 vm : vm1 and vm2 |
| # 2. create ssh ingress security group rule with test machine ip address cidr |
| # 3. ssh to vm1 |
| # 4. ssh to vm2 from vm1 ( should fail as ssh ingress security group rule ) |
| # 5. create ssh ingress security group rule with vm1 ip address cidr |
| # 4. ssh to vm2 from vm1 again ( should success as ssh ingress security group rule ) |
| |
| network_offering = NetworkOffering.create(self.api_client, self.services["network_offering"]) |
| network_offering.update(self.api_client, state='Enabled') |
| self.cleanup.append(network_offering) |
| |
| # create network |
| network = Network.create( |
| self.api_client, |
| self.services["networks"]["network1"], |
| networkofferingid=network_offering.id, |
| zoneid=self.zone.id, |
| accountid=self.account.name, |
| domainid=self.account.domainid |
| ) |
| self.cleanup.append(network) |
| |
| # create security group |
| security_group = SecurityGroup.create( |
| self.api_client, |
| self.services["security_groups"]["security_group1"], |
| account=self.account.name, |
| domainid=self.account.domainid |
| ) |
| self.cleanup.append(security_group) |
| |
| # create virtual machine 1 |
| vm1 = VirtualMachine.create( |
| self.api_client, |
| self.services["virtual_machines"]["vm1"], |
| serviceofferingid=self.service_offering.id, |
| templateid=self.template.id, |
| networkids=[str(network.id)], |
| zoneid=self.zone.id, |
| accountid=self.account.name, |
| domainid=self.account.domainid, |
| securitygroupids=[security_group.id], |
| mode='advanced' |
| ) |
| self.cleanup.append(vm1) |
| |
| # create virtual machine 2 |
| vm2 = VirtualMachine.create( |
| self.api_client, |
| self.services["virtual_machines"]["vm2"], |
| serviceofferingid=self.service_offering.id, |
| templateid=self.template.id, |
| networkids=[str(network.id)], |
| zoneid=self.zone.id, |
| accountid=self.account.name, |
| domainid=self.account.domainid, |
| securitygroupids=[security_group.id], |
| mode='advanced' |
| ) |
| self.cleanup.append(vm2) |
| |
| # create ssh ingress security group rule with test machine ip address cidr |
| test_machine_ip_address = self.getLocalMachineIpAddress() |
| cidr = test_machine_ip_address + "/32" |
| cmd = authorizeSecurityGroupIngress.authorizeSecurityGroupIngressCmd() |
| cmd.securitygroupid = security_group.id |
| cmd.protocol = 'TCP' |
| cmd.startport = 22 |
| cmd.endport = 22 |
| cmd.cidrlist = cidr |
| ingress_rule1 = self.api_client.authorizeSecurityGroupIngress(cmd) |
| |
| ssh = vm1.get_ssh_client() |
| res1 = ssh.execute("ssh %s@%s -v" % (vm2.username, vm2.ssh_ip)) |
| self.debug("Response is :%s" % res1) |
| self.assertFalse("connection established" in str(res1).lower(), "SSH to VM 2 should fail") |
| |
| # create ssh ingress security group rule with vm1 ip address cidr |
| cmd = authorizeSecurityGroupIngress.authorizeSecurityGroupIngressCmd() |
| cmd.securitygroupid = security_group.id |
| cmd.protocol = 'TCP' |
| cmd.startport = 22 |
| cmd.endport = 22 |
| cmd.cidrlist = vm1.ipaddress + "/32" |
| ingress_rule2 = self.api_client.authorizeSecurityGroupIngress(cmd) |
| |
| res2 = ssh.execute("ssh %s@%s -v" % (vm2.username, vm2.ssh_ip)) |
| self.debug("Response is :%s" % res2) |
| self.assertTrue("connection established" in str(res2).lower(), "SSH to VM 2 should success") |
| |
| # revoke ssh ingress security group rule 1 |
| cmd = revokeSecurityGroupIngress.revokeSecurityGroupIngressCmd() |
| cmd.id = ingress_rule1.ingressrule[0].ruleid |
| self.api_client.revokeSecurityGroupIngress(cmd) |
| |
| # revoke ssh ingress security group rule 2 |
| cmd = revokeSecurityGroupIngress.revokeSecurityGroupIngressCmd() |
| cmd.id = ingress_rule2.ingressrule[0].ruleid |
| self.api_client.revokeSecurityGroupIngress(cmd) |
| |
| @skipTestIf("not_tungsten_fabric_zone") |
| @attr(tags=["tungstenfabric"], required_hardware="false") |
| def test_01_test_ingress_rule_security_group_with_account(self): |
| |
| # Validate the following: |
| # 1. create network 1 |
| # 2. create security group 1 |
| # 3. create security group 2 |
| # 4. deploy vm 1 |
| # 5. deploy vm 2 |
| # 6. create ssh ingress security group 1 rule with test machine ip address cidr |
| # 7. ssh to vm1 from test machine |
| # 8. ping vm2 from vm1 ( should fail ) |
| # 9. create icmp ingress security group 2 rule with account and security group 1 |
| # 10. ping vm2 from vm1 again ( should success ) |
| # 11. create ssh ingress security group 2 rule with test machine ip address cidr |
| # 12. ssh to vm2 from test machine |
| # 13. ping vm1 from vm2 ( should fail ) |
| # 14. create icmp ingress security group 1 rule with account and security group 2 |
| # 15. ping vm1 from vm2 again ( should success ) |
| # 16. revoke ssh ingress security group rule 1 |
| # 17. revoke ssh ingress security group rule 2 |
| # 18. revoke icmp ingress security group rule 1 |
| # 19. revoke icmp ingress security group rule 2 |
| |
| network_offering = NetworkOffering.create(self.api_client, self.services["network_offering"]) |
| network_offering.update(self.api_client, state='Enabled') |
| self.cleanup.append(network_offering) |
| |
| # 1. create network 1 |
| network = Network.create( |
| self.api_client, |
| self.services["networks"]["network1"], |
| networkofferingid=network_offering.id, |
| zoneid=self.zone.id, |
| accountid=self.account.name, |
| domainid=self.account.domainid |
| ) |
| self.cleanup.append(network) |
| |
| # 2. create security group 1 |
| security_group1 = SecurityGroup.create( |
| self.api_client, |
| self.services["security_groups"]["security_group1"], |
| account=self.account.name, |
| domainid=self.account.domainid |
| ) |
| self.cleanup.append(security_group1) |
| |
| # 3. create security group 2 |
| security_group2 = SecurityGroup.create( |
| self.api_client, |
| self.services["security_groups"]["security_group2"], |
| account=self.account.name, |
| domainid=self.account.domainid |
| ) |
| self.cleanup.append(security_group2) |
| |
| # 4. deploy vm 1 |
| vm1 = VirtualMachine.create( |
| self.api_client, |
| self.services["virtual_machines"]["vm1"], |
| serviceofferingid=self.service_offering.id, |
| templateid=self.template.id, |
| networkids=[str(network.id)], |
| zoneid=self.zone.id, |
| accountid=self.account.name, |
| domainid=self.account.domainid, |
| securitygroupids=[security_group1.id], |
| mode='advanced' |
| ) |
| self.cleanup.append(vm1) |
| |
| # 5. deploy vm 2 |
| vm2 = VirtualMachine.create( |
| self.api_client, |
| self.services["virtual_machines"]["vm2"], |
| serviceofferingid=self.service_offering.id, |
| templateid=self.template.id, |
| networkids=[str(network.id)], |
| zoneid=self.zone.id, |
| accountid=self.account.name, |
| domainid=self.account.domainid, |
| securitygroupids=[security_group2.id], |
| mode='advanced' |
| ) |
| self.cleanup.append(vm2) |
| |
| # 6. create ssh ingress security group 1 rule with test machine ip address cidr |
| test_machine_ip_address = self.getLocalMachineIpAddress() |
| cidr = test_machine_ip_address + "/32" |
| cmd = authorizeSecurityGroupIngress.authorizeSecurityGroupIngressCmd() |
| cmd.securitygroupid = security_group1.id |
| cmd.protocol = 'TCP' |
| cmd.startport = 22 |
| cmd.endport = 22 |
| cmd.cidrlist = cidr |
| ssh_ingress_rule1 = self.api_client.authorizeSecurityGroupIngress(cmd) |
| |
| # 7. ssh to vm1 from test machine |
| ssh1 = vm1.get_ssh_client() |
| |
| # 8. ping vm2 from vm1 ( should fail ) |
| res1 = ssh1.execute("ping -c1 %s" % vm2.ipaddress) |
| self.assertEqual( |
| str(res1).count("0 received"), |
| 1, |
| "Ping vm2 from vm1 should be fail" |
| ) |
| |
| # 9. create icmp ingress security group 2 rule with account and security group 1 |
| cmd = authorizeSecurityGroupIngress.authorizeSecurityGroupIngressCmd() |
| cmd.securitygroupid = security_group2.id |
| cmd.protocol = 'ICMP' |
| cmd.icmptype = "-1" |
| cmd.icmpcode = "-1" |
| cmd.usersecuritygrouplist = [{'account':str(self.account.name),'group': str(security_group1.name)}] |
| icmp_ingress_rule1 = self.api_client.authorizeSecurityGroupIngress(cmd) |
| |
| # 10. ping vm2 from vm1 again ( should success ) |
| res2 = ssh1.execute("ping -c1 %s" % vm2.ipaddress) |
| self.assertEqual( |
| str(res2).count("1 received"), |
| 1, |
| "Ping vm2 from vm1 should be success" |
| ) |
| |
| # 11. create ssh ingress security group 2 rule with test machine ip address cidr |
| cmd = authorizeSecurityGroupIngress.authorizeSecurityGroupIngressCmd() |
| cmd.securitygroupid = security_group2.id |
| cmd.protocol = 'TCP' |
| cmd.startport = 22 |
| cmd.endport = 22 |
| cmd.cidrlist = cidr |
| ssh_ingress_rule2 = self.api_client.authorizeSecurityGroupIngress(cmd) |
| |
| # 12. ssh to vm2 from test machine |
| ssh2 = vm2.get_ssh_client() |
| |
| # 13. ping vm1 from vm2 ( should fail ) |
| res3 = ssh2.execute("ping -c1 %s" % vm1.ipaddress) |
| self.assertEqual( |
| str(res3).count("0 received"), |
| 1, |
| "Ping vm1 from vm2 should be fail" |
| ) |
| |
| # 14. create icmp ingress security group 1 rule with account and security group 2 |
| cmd = authorizeSecurityGroupIngress.authorizeSecurityGroupIngressCmd() |
| cmd.securitygroupid = security_group1.id |
| cmd.protocol = 'ICMP' |
| cmd.icmptype = "-1" |
| cmd.icmpcode = "-1" |
| cmd.usersecuritygrouplist = [{'account':str(self.account.name),'group': str(security_group2.name)}] |
| icmp_ingress_rule2 = self.api_client.authorizeSecurityGroupIngress(cmd) |
| |
| # 15. ping vm1 from vm2 again ( should success ) |
| res4 = ssh2.execute("ping -c1 %s" % vm2.ipaddress) |
| self.assertEqual( |
| str(res4).count("1 received"), |
| 1, |
| "Ping vm1 from vm2 should be success" |
| ) |
| |
| # 16. revoke ssh ingress security group rule 1 |
| cmd = revokeSecurityGroupIngress.revokeSecurityGroupIngressCmd() |
| cmd.id = ssh_ingress_rule1.ingressrule[0].ruleid |
| self.api_client.revokeSecurityGroupIngress(cmd) |
| |
| # 17. revoke ssh ingress security group rule 2 |
| cmd = revokeSecurityGroupIngress.revokeSecurityGroupIngressCmd() |
| cmd.id = ssh_ingress_rule2.ingressrule[0].ruleid |
| self.api_client.revokeSecurityGroupIngress(cmd) |
| |
| # 18. revoke icmp ingress security group rule 1 |
| cmd = revokeSecurityGroupIngress.revokeSecurityGroupIngressCmd() |
| cmd.id = icmp_ingress_rule1.ingressrule[0].ruleid |
| self.api_client.revokeSecurityGroupIngress(cmd) |
| |
| # 19. revoke icmp ingress security group rule 2 |
| cmd = revokeSecurityGroupIngress.revokeSecurityGroupIngressCmd() |
| cmd.id = icmp_ingress_rule2.ingressrule[0].ruleid |
| self.api_client.revokeSecurityGroupIngress(cmd) |
| |
| @skipTestIf("not_tungsten_fabric_zone") |
| @attr(tags=["tungstenfabric"], required_hardware="false") |
| def test_01_test_egress_rule_security_group_with_cidr(self): |
| |
| # Validate the following: |
| # 1. create network offering |
| # 2. create network 1 |
| # 3. create security group 1 |
| # 4. deploy vm1 |
| # 5. create ssh ingress security group 1 rule with test machine ip address cidr |
| # 6. ssh to vm1 from test machine |
| # 7. ping 8.8.8.8 from vm1 ( should success ) |
| # 8. ping 4.4.4.4 from vm1 ( should success ) |
| # 9. create icmp egress security group 1 rule with 8.8.8.8/32 cidr |
| # 10. ping 8.8.8.8 from vm 1 ( should success ) |
| # 11. ping 8.8.4.4 from vm1 ( should fail ) |
| # 12. create icmp egress security group 1 rule with 8.8.4.4/32 cidr |
| # 13. ping 8.8.4.4 from vm1 ( should success ) |
| |
| # 1. create network offering |
| network_offering = NetworkOffering.create( |
| self.api_client, |
| self.services["network_offering"] |
| ) |
| network_offering.update(self.api_client, state='Enabled') |
| self.cleanup.append(network_offering) |
| |
| # 2. create network 1 |
| network = Network.create( |
| self.api_client, |
| self.services["networks"]["network1"], |
| networkofferingid=network_offering.id, |
| zoneid=self.zone.id, |
| accountid=self.account.name, |
| domainid=self.account.domainid |
| ) |
| self.cleanup.append(network) |
| |
| # 3. create security group 1 |
| security_group1 = SecurityGroup.create( |
| self.api_client, |
| self.services["security_groups"]["security_group1"], |
| account=self.account.name, |
| domainid=self.account.domainid |
| ) |
| self.cleanup.append(security_group1) |
| |
| # 4. deploy vm 1 |
| vm1 = VirtualMachine.create( |
| self.api_client, |
| self.services["virtual_machines"]["vm1"], |
| serviceofferingid=self.service_offering.id, |
| templateid=self.template.id, |
| networkids=[str(network.id)], |
| zoneid=self.zone.id, |
| accountid=self.account.name, |
| domainid=self.account.domainid, |
| securitygroupids=[security_group1.id], |
| mode='advanced' |
| ) |
| self.cleanup.append(vm1) |
| |
| # 5. create ssh ingress security group 1 rule with test machine ip address cidr |
| test_machine_ip_address = self.getLocalMachineIpAddress() |
| cidr = test_machine_ip_address + "/32" |
| cmd = authorizeSecurityGroupIngress.authorizeSecurityGroupIngressCmd() |
| cmd.securitygroupid = security_group1.id |
| cmd.protocol = 'TCP' |
| cmd.startport = 22 |
| cmd.endport = 22 |
| cmd.cidrlist = cidr |
| ssh_ingress_rule1 = self.api_client.authorizeSecurityGroupIngress(cmd) |
| |
| # 6. ssh to vm1 from test machine |
| ssh1 = vm1.get_ssh_client() |
| |
| # 7. ping 8.8.8.8 from vm 1 ( should success ) |
| res1 = ssh1.execute("ping -c1 8.8.8.8") |
| self.assertEqual( |
| str(res1).count("1 received"), |
| 1, |
| "Ping 8.8.8.8 from vm1 should be success" |
| ) |
| |
| # 8. ping 8.8.4.4 from vm1 ( should success ) |
| res2 = ssh1.execute("ping -c1 8.8.4.4") |
| self.assertEqual( |
| str(res2).count("1 received"), |
| 1, |
| "Ping 8.8.4.4 from vm1 should be success" |
| ) |
| |
| # 9. create icmp egress security group 1 rule with 8.8.8.8/32 cidr |
| cmd = authorizeSecurityGroupEgress.authorizeSecurityGroupEgressCmd() |
| cmd.securitygroupid = security_group1.id |
| cmd.protocol = 'ICMP' |
| cmd.icmptype = "-1" |
| cmd.icmpcode = "-1" |
| cmd.cidrlist = '8.8.8.8/32' |
| icmp_egress_rule1 = self.api_client.authorizeSecurityGroupEgress(cmd) |
| |
| # 10. ping 8.8.8.8 from vm 1 ( should success ) |
| res3 = ssh1.execute("ping -c1 8.8.8.8") |
| self.assertEqual( |
| str(res3).count("1 received"), |
| 1, |
| "Ping 8.8.8.8 from vm1 should be success" |
| ) |
| |
| # 11. ping 8.8.4.4 from vm1 ( should fail ) |
| res4 = ssh1.execute("ping -c1 8.8.4.4") |
| self.assertEqual( |
| str(res4).count("0 received"), |
| 1, |
| "Ping 8.8.4.4 from vm1 should be fail" |
| ) |
| |
| # 12. create icmp egress security group 1 rule with 8.8.4.4 cidr |
| cmd = authorizeSecurityGroupEgress.authorizeSecurityGroupEgressCmd() |
| cmd.securitygroupid = security_group1.id |
| cmd.protocol = 'ICMP' |
| cmd.icmptype = "-1" |
| cmd.icmpcode = "-1" |
| cmd.cidrlist = "8.8.4.4/32" |
| icmp_egress_rule2 = self.api_client.authorizeSecurityGroupEgress(cmd) |
| |
| # 13. ping vm2 from vm1 ( should success ) |
| res5 = ssh1.execute("ping -c1 8.8.4.4") |
| self.assertEqual( |
| str(res5).count("1 received"), |
| 1, |
| "Ping 8.8.4.4 from vm1 should be success" |
| ) |
| |
| # 14. revoke ssh ingress security group rule 1 |
| cmd = revokeSecurityGroupIngress.revokeSecurityGroupIngressCmd() |
| cmd.id = ssh_ingress_rule1.ingressrule[0].ruleid |
| self.api_client.revokeSecurityGroupIngress(cmd) |
| |
| # 15. revoke icmp egress security group rule 1 |
| cmd = revokeSecurityGroupEgress.revokeSecurityGroupEgressCmd() |
| cmd.id = icmp_egress_rule1.egressrule[0].ruleid |
| self.api_client.revokeSecurityGroupEgress(cmd) |
| |
| # 16. revoke icmp egress security group rule 2 |
| cmd = revokeSecurityGroupEgress.revokeSecurityGroupEgressCmd() |
| cmd.id = icmp_egress_rule2.egressrule[0].ruleid |
| self.api_client.revokeSecurityGroupEgress(cmd) |
| |
| @skipTestIf("not_tungsten_fabric_zone") |
| @attr(tags=["tungstenfabric"], required_hardware="false") |
| def test_01_test_egress_rule_security_group_with_account(self): |
| |
| # Validate the following: |
| # 1. create network offering |
| # 2. create network 1 |
| # 3. create security group 1 |
| # 4. create security group 2 |
| # 5. create security group 3 |
| # 6. deploy vm1 |
| # 7. deploy vm2 |
| # 8. deploy vm3 |
| # 9. create ssh ingress security group 1 rule with test machine ip address cidr |
| # 10. create icmp ingress security group 2 rule with security group 1 |
| # 11. create icmp ingress security group 3 rule with security group 1 |
| # 12. ssh to vm1 from test machine |
| # 13. ping vm2 from vm1 ( should success ) |
| # 14. ping vm3 from vm1 ( should success ) |
| # 15. create icmp egress security group 1 rule with security group 2 |
| # 16. ping vm2 from vm1 ( should success ) |
| # 17. ping vm3 from vm1 ( should fail ) |
| # 18. create icmp egress security group 1 rule with security group 3 |
| # 19. ping vm2 from vm1 ( should success ) |
| # 20. ping vm3 from vm1 ( should success ) |
| # 21. revoke ssh ingress security group rule 1 |
| # 22. revoke icmp ingress security group rule 1 |
| # 23. revoke icmp ingress security group rule 2 |
| # 24. revoke icmp egress security group rule 1 |
| # 25. revoke icmp egress security group rule 2 |
| |
| # 1. create network offering |
| network_offering = NetworkOffering.create( |
| self.api_client, |
| self.services["network_offering"] |
| ) |
| network_offering.update(self.api_client, state='Enabled') |
| self.cleanup.append(network_offering) |
| |
| # 2. create network 1 |
| network = Network.create( |
| self.api_client, |
| self.services["networks"]["network1"], |
| networkofferingid=network_offering.id, |
| zoneid=self.zone.id, |
| accountid=self.account.name, |
| domainid=self.account.domainid |
| ) |
| self.cleanup.append(network) |
| |
| # 3. create security group 1 |
| security_group1 = SecurityGroup.create( |
| self.api_client, |
| self.services["security_groups"]["security_group1"], |
| account=self.account.name, |
| domainid=self.account.domainid |
| ) |
| self.cleanup.append(security_group1) |
| |
| # 4. create security group 2 |
| security_group2 = SecurityGroup.create( |
| self.api_client, |
| self.services["security_groups"]["security_group2"], |
| account=self.account.name, |
| domainid=self.account.domainid |
| ) |
| self.cleanup.append(security_group2) |
| |
| # 5. create security group 3 |
| security_group3 = SecurityGroup.create( |
| self.api_client, |
| self.services["security_groups"]["security_group3"], |
| account=self.account.name, |
| domainid=self.account.domainid |
| ) |
| self.cleanup.append(security_group3) |
| |
| # 6. deploy vm1 |
| vm1 = VirtualMachine.create( |
| self.api_client, |
| self.services["virtual_machines"]["vm1"], |
| serviceofferingid=self.service_offering.id, |
| templateid=self.template.id, |
| networkids=[str(network.id)], |
| zoneid=self.zone.id, |
| accountid=self.account.name, |
| domainid=self.account.domainid, |
| securitygroupids=[security_group1.id], |
| mode='advanced' |
| ) |
| self.cleanup.append(vm1) |
| |
| # 7. deploy vm2 |
| vm2 = VirtualMachine.create( |
| self.api_client, |
| self.services["virtual_machines"]["vm2"], |
| serviceofferingid=self.service_offering.id, |
| templateid=self.template.id, |
| networkids=[str(network.id)], |
| zoneid=self.zone.id, |
| accountid=self.account.name, |
| domainid=self.account.domainid, |
| securitygroupids=[security_group2.id], |
| mode='advanced' |
| ) |
| self.cleanup.append(vm2) |
| |
| # 8. deploy vm3 |
| vm3 = VirtualMachine.create( |
| self.api_client, |
| self.services["virtual_machines"]["vm3"], |
| serviceofferingid=self.service_offering.id, |
| templateid=self.template.id, |
| networkids=[str(network.id)], |
| zoneid=self.zone.id, |
| accountid=self.account.name, |
| domainid=self.account.domainid, |
| securitygroupids=[security_group3.id], |
| mode='advanced' |
| ) |
| self.cleanup.append(vm3) |
| |
| # 9. create ssh ingress security group 1 rule with test machine ip address cidr |
| test_machine_ip_address = self.getLocalMachineIpAddress() |
| cidr = test_machine_ip_address + "/32" |
| cmd = authorizeSecurityGroupIngress.authorizeSecurityGroupIngressCmd() |
| cmd.securitygroupid = security_group1.id |
| cmd.protocol = 'TCP' |
| cmd.startport = 22 |
| cmd.endport = 22 |
| cmd.cidrlist = cidr |
| ssh_ingress_rule1 = self.api_client.authorizeSecurityGroupIngress(cmd) |
| |
| # 10. create icmp ingress security group 2 rule with security group 1 |
| cmd = authorizeSecurityGroupIngress.authorizeSecurityGroupIngressCmd() |
| cmd.securitygroupid = security_group2.id |
| cmd.protocol = 'ICMP' |
| cmd.icmptype = "-1" |
| cmd.icmpcode = "-1" |
| cmd.usersecuritygrouplist = [{'account':str(self.account.name),'group': str(security_group1.name)}] |
| icmp_ingress_rule1 = self.api_client.authorizeSecurityGroupIngress(cmd) |
| |
| # 11. create icmp ingress security group 3 rule with security group 1 |
| cmd = authorizeSecurityGroupIngress.authorizeSecurityGroupIngressCmd() |
| cmd.securitygroupid = security_group3.id |
| cmd.protocol = 'ICMP' |
| cmd.icmptype = "-1" |
| cmd.icmpcode = "-1" |
| cmd.usersecuritygrouplist = [{'account':str(self.account.name),'group': str(security_group1.name)}] |
| icmp_ingress_rule2 = self.api_client.authorizeSecurityGroupIngress(cmd) |
| |
| # 12. ssh to vm1 from test machine |
| ssh1 = vm1.get_ssh_client() |
| |
| # 13. ping vm2 from vm1 ( should success ) |
| res1 = ssh1.execute("ping -c1 %s" % vm2.ipaddress) |
| self.assertEqual( |
| str(res1).count("1 received"), |
| 1, |
| "Ping vm2 from vm1 should be success" |
| ) |
| |
| # 14. ping vm3 from vm1 ( should success ) |
| res2 = ssh1.execute("ping -c1 %s" % vm3.ipaddress) |
| self.assertEqual( |
| str(res2).count("1 received"), |
| 1, |
| "Ping vm3 from vm1 should be success" |
| ) |
| |
| # 15. create icmp egress security group 1 rule with security group 2 |
| cmd = authorizeSecurityGroupEgress.authorizeSecurityGroupEgressCmd() |
| cmd.securitygroupid = security_group1.id |
| cmd.protocol = 'ICMP' |
| cmd.icmptype = "-1" |
| cmd.icmpcode = "-1" |
| cmd.usersecuritygrouplist = [{'account':str(self.account.name),'group': str(security_group2.name)}] |
| icmp_egress_rule1 = self.api_client.authorizeSecurityGroupEgress(cmd) |
| |
| # 16. ping vm2 from vm1 ( should success ) |
| res3 = ssh1.execute("ping -c1 %s" % vm2.ipaddress) |
| self.assertEqual( |
| str(res3).count("1 received"), |
| 1, |
| "Ping vm2 from vm1 should be success" |
| ) |
| |
| # 17. ping vm3 from vm1 ( should fail ) |
| res4 = ssh1.execute("ping -c1 %s" % vm3.ipaddress) |
| self.assertEqual( |
| str(res4).count("0 received"), |
| 1, |
| "Ping vm3 from vm1 should be fail" |
| ) |
| |
| # 18. create icmp egress security group 1 rule with security group 3 |
| cmd = authorizeSecurityGroupEgress.authorizeSecurityGroupEgressCmd() |
| cmd.securitygroupid = security_group1.id |
| cmd.protocol = 'ICMP' |
| cmd.icmptype = "-1" |
| cmd.icmpcode = "-1" |
| cmd.usersecuritygrouplist = [{'account':str(self.account.name),'group': str(security_group3.name)}] |
| icmp_egress_rule2 = self.api_client.authorizeSecurityGroupEgress(cmd) |
| |
| # 19. ping vm2 from vm1 ( should success ) |
| res5 = ssh1.execute("ping -c1 %s" % vm2.ipaddress) |
| self.assertEqual( |
| str(res5).count("1 received"), |
| 1, |
| "Ping vm2 from vm1 should be success" |
| ) |
| |
| # 20. ping vm3 from vm1 ( should success ) |
| res6 = ssh1.execute("ping -c1 %s" % vm3.ipaddress) |
| self.assertEqual( |
| str(res6).count("1 received"), |
| 1, |
| "Ping vm3 from vm1 should be success" |
| ) |
| |
| # 21. revoke ssh ingress security group rule 1 |
| cmd = revokeSecurityGroupIngress.revokeSecurityGroupIngressCmd() |
| cmd.id = ssh_ingress_rule1.ingressrule[0].ruleid |
| self.api_client.revokeSecurityGroupIngress(cmd) |
| |
| # 22. revoke icmp ingress security group rule 1 |
| cmd = revokeSecurityGroupIngress.revokeSecurityGroupIngressCmd() |
| cmd.id = icmp_ingress_rule1.ingressrule[0].ruleid |
| self.api_client.revokeSecurityGroupIngress(cmd) |
| |
| # 23. revoke icmp ingress security group rule 2 |
| cmd = revokeSecurityGroupIngress.revokeSecurityGroupIngressCmd() |
| cmd.id = icmp_ingress_rule2.ingressrule[0].ruleid |
| self.api_client.revokeSecurityGroupIngress(cmd) |
| |
| # 24. revoke icmp egress security group rule 1 |
| cmd = revokeSecurityGroupEgress.revokeSecurityGroupEgressCmd() |
| cmd.id = icmp_egress_rule1.egressrule[0].ruleid |
| self.api_client.revokeSecurityGroupEgress(cmd) |
| |
| # 25. revoke icmp egress security group rule 2 |
| cmd = revokeSecurityGroupEgress.revokeSecurityGroupEgressCmd() |
| cmd.id = icmp_egress_rule2.egressrule[0].ruleid |
| self.api_client.revokeSecurityGroupEgress(cmd) |
