You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cloudstack.apache.org by GitBox <gi...@apache.org> on 2022/04/21 12:14:36 UTC
[GitHub] [cloudstack] weizhouapache commented on a diff in pull request #4774: Added configuration and Integration test to restrict public template …
weizhouapache commented on code in PR #4774:
URL: https://github.com/apache/cloudstack/pull/4774#discussion_r855120318
##########
test/integration/component/test_template_access_across_domains.py:
##########
@@ -0,0 +1,626 @@
+# 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 Local Modules
+from nose.plugins.attrib import attr
+from marvin.cloudstackTestCase import cloudstackTestCase, unittest
+from marvin.cloudstackAPI import (listZones,
+ deleteTemplate,
+ listConfigurations,
+ updateConfiguration)
+from marvin.lib.utils import (cleanup_resources)
+from marvin.lib.base import (Account,
+ Domain,
+ Network,
+ NetworkOffering,
+ Template,
+ ServiceOffering,
+ VirtualMachine,
+ Snapshot,
+ Volume)
+from marvin.lib.common import (get_domain,
+ get_zone,
+ get_template,
+ get_builtin_template_info)
+# Import System modules
+import time
+import logging
+
+class TestTemplateAccessAcrossDomains(cloudstackTestCase):
+ @classmethod
+ def setUpClass(cls):
+ cls.testClient = super(TestTemplateAccessAcrossDomains, cls).getClsTestClient()
+ cls.apiclient = cls.testClient.getApiClient()
+
+ cls.services = cls.testClient.getParsedTestDataConfig()
+ # Get Zone, Domain and templates
+ cls.domain = get_domain(cls.apiclient)
+ cls.zone = get_zone(cls.apiclient, cls.testClient.getZoneForTests())
+ cls.services['mode'] = cls.zone.networktype
+ cls.logger = logging.getLogger("TestRouterResources")
+ cls._cleanup = []
+ cls.unsupportedHypervisor = False
+ cls.hypervisor = cls.testClient.getHypervisorInfo()
+ if cls.hypervisor.lower() in ['lxc']:
+ cls.unsupportedHypervisor = True
+ return
+ cls.services["virtual_machine"]["zoneid"] = cls.zone.id
+
+ # Create new domain1
+ cls.domain1 = Domain.create(
+ cls.apiclient,
+ services=cls.services["acl"]["domain1"],
+ parentdomainid=cls.domain.id)
+ cls._cleanup.append(cls.domain1)
+
+ # Create account1
+ cls.account1 = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD1"],
+ domainid=cls.domain1.id
+ )
+ cls._cleanup.append(cls.account1)
+
+ # Create new sub-domain
+ cls.sub_domain = Domain.create(
+ cls.apiclient,
+ services=cls.services["acl"]["domain11"],
+ parentdomainid=cls.domain1.id)
+ cls._cleanup.append(cls.sub_domain)
+
+ # Create account for sub-domain
+ cls.sub_account = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD11"],
+ domainid=cls.sub_domain.id
+ )
+ cls._cleanup.append(cls.sub_account)
+
+ # Create new domain2
+ cls.domain2 = Domain.create(
+ cls.apiclient,
+ services=cls.services["acl"]["domain2"],
+ parentdomainid=cls.domain.id)
+ cls._cleanup.append(cls.domain2)
+
+ # Create account2
+ cls.account2 = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD2"],
+ domainid=cls.domain2.id
+ )
+ cls._cleanup.append(cls.account2)
+
+ cls.service_offering = ServiceOffering.create(
+ cls.apiclient,
+ cls.services["service_offering"]
+ )
+ cls._cleanup.append(cls.service_offering)
+ if cls.hypervisor.lower() in ['kvm']:
+ # register template under ROOT domain
+ cls.root_template = Template.register(cls.apiclient,
+ cls.services["test_templates"]["kvm"],
+ zoneid=cls.zone.id,
+ domainid=cls.domain.id,
+ hypervisor=cls.hypervisor.lower())
+ cls.root_template.download(cls.apiclient)
+ cls._cleanup.append(cls.root_template)
+ cls.services["test_templates"]["kvm"]["name"] = cls.account1.name
+ cls.template1 = Template.register(cls.apiclient,
+ cls.services["test_templates"]["kvm"],
+ zoneid=cls.zone.id,
+ account=cls.account1.name,
+ domainid=cls.domain1.id,
+ hypervisor=cls.hypervisor.lower())
+ cls.template1.download(cls.apiclient)
+ cls._cleanup.append(cls.template1)
+ cls.services["test_templates"]["kvm"]["name"] = cls.sub_account.name
+ cls.sub_template = Template.register(cls.apiclient,
+ cls.services["test_templates"]["kvm"],
+ zoneid=cls.zone.id,
+ account=cls.sub_account.name,
+ domainid=cls.sub_domain.id,
+ hypervisor=cls.hypervisor.lower())
+ cls.sub_template.download(cls.apiclient)
+ cls._cleanup.append(cls.sub_template)
+ cls.template2 = Template.register(cls.apiclient,
+ cls.services["test_templates"]["kvm"],
+ zoneid=cls.zone.id,
+ account=cls.account2.name,
+ domainid=cls.domain2.id,
+ hypervisor=cls.hypervisor.lower())
+ cls.template2.download(cls.apiclient)
+ cls._cleanup.append(cls.template2)
+ else:
+ return
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestTemplateAccessAcrossDomains, cls).tearDownClass()
+
+ def setUp(self):
+ self.apiclient = self.testClient.getApiClient()
+ self.domain1_config = self.get_restrict_template_configuration(self.domain1.id)
+ self.domain2_config = self.get_restrict_template_configuration(self.domain2.id)
+ self.sub_domain_config = self.get_restrict_template_configuration(self.sub_domain.id)
+ self.cleanup = []
+ return
+
+ def tearDown(self):
+ try:
+ self.update_restrict_template_configuration(self.domain1.id, self.domain1_config)
+ self.update_restrict_template_configuration(self.domain2.id, self.domain2_config)
+ self.update_restrict_template_configuration(self.sub_domain.id, self.sub_domain_config)
+ super(TestTemplateAccessAcrossDomains, self).tearDown()
+ except Exception as e:
+ raise Exception("Warning: Exception during cleanup : %s" % e)
+ return
+
+ @attr(tags=["advanced", "basic", "sg"], required_hardware="false")
+ def test_01_check_cross_domain_template_access(self):
+ """
+ Verify that templates belonging to one domain should not be accessible
+ by other domains except for parent and ROOT domains
+
+ Steps:
+ 1. Set global setting restrict.public.access.to.templates to true
+ 2. Make sure template of domain2 should not be accessible by domain1
+ 3. Make sure template of domain1 should not be accessible by domain2
+ 4. Make sure parent and ROOT domain can still access above templates
+ :return:
+ """
+
+ # Step 1
+ self.update_restrict_template_configuration(self.domain1.id, "true")
+ self.update_restrict_template_configuration(self.domain2.id, "true")
+ self.validate_uploaded_template(self.apiclient, self.template1.id)
+
+ # Step 2
+ self.validate_template_ownership(self.template2, self.domain1, self.domain2, False)
+
+ self.validate_uploaded_template(self.apiclient, self.template2.id)
+
+ # Step 3
+ self.validate_template_ownership(self.template1, self.domain2, self.domain1, False)
+
+ # Make sure root domain can still access all subdomain templates
+ # Step 4
+ self.validate_template_ownership(self.template1, self.domain, self.domain1, True)
+ self.validate_template_ownership(self.template2, self.domain, self.domain2, True)
+
+
+ @attr(tags=["advanced", "basic", "sg"], required_hardware="false")
+ def test_02_create_template(self):
+ """
+ Verify that templates belonging to one domain can be accessible
+ by other domains by default
+
+ Steps:
+ 1. Set global setting restrict.public.access.to.templates to false (default behavior)
+ 2. Make sure template of domain2 can be accessible by domain1
+ 3. Make sure template of domain1 can be accessible by domain2
+ 4. Make sure parent and ROOT domain can still access above templates
+ 5. Deploy virtual machine in domain1 using template from domain2
+ 6. Make sure that virtual machine can be deployed and is in running state
+ :return:
+ """
+
+ # Step 1
+ self.update_restrict_template_configuration(self.domain1.id, "false")
+ self.update_restrict_template_configuration(self.domain2.id, "false")
+
+ # Step 2
+ self.validate_template_ownership(self.template2, self.domain1, self.domain2, True)
+
+ # Step 3
+ self.validate_template_ownership(self.template1, self.domain2, self.domain1, True)
+
+ # Step 4
+ # Make sure root domain can still access all subdomain templates
+ self.validate_template_ownership(self.template1, self.domain, self.domain1, True)
+ self.validate_template_ownership(self.template2, self.domain, self.domain2, True)
+
+ # Step 5
+ # Deploy new virtual machine using template
+ self.virtual_machine = VirtualMachine.create(
+ self.apiclient,
+ self.services["virtual_machine"],
+ templateid=self.template2.id,
+ accountid=self.account1.name,
+ domainid=self.account1.domainid,
+ serviceofferingid=self.service_offering.id,
+ )
+ self.cleanup.append(self.virtual_machine)
+ self.debug("creating an instance with template ID: %s" % self.template2.id)
+ vm_response = VirtualMachine.list(self.apiclient,
+ id=self.virtual_machine.id,
+ account=self.account1.name,
+ domainid=self.account1.domainid)
+ self.assertEqual(
+ isinstance(vm_response, list),
+ True,
+ "Check for list VMs response after VM deployment"
+ )
+ # Verify VM response to check whether VM deployment was successful
+ self.assertNotEqual(
+ len(vm_response),
+ 0,
+ "Check VMs available in List VMs response"
+ )
+
+ # Step 6
+ vm = vm_response[0]
+ self.assertEqual(
+ vm.state,
+ 'Running',
+ "Check the state of VM created from Template"
+ )
+
+ @attr(tags=["advanced", "basic", "sg"], required_hardware="false")
+ def test_03_check_subdomain_template_access(self):
+ """
+ Verify that templates belonging to parent domain can be accessible
+ by sub domains
+
+ Steps:
+ 1. Set global setting restrict.public.access.to.templates to true
+ 2. Make sure template of ROOT domain can be accessible by domain1
+ 3. Make sure template of ROOT domain can be accessible by domain2
+ """
+
+ # Step 1
+ self.update_restrict_template_configuration(self.domain1.id, "true")
+ self.update_restrict_template_configuration(self.domain2.id, "true")
+ # Make sure child domains can still access parent domain templates
+ self.validate_uploaded_template(self.apiclient, self.root_template.id)
+
+ # Step 2
+ self.validate_template_ownership(self.root_template, self.domain1, self.domain, True)
+
+ # Step 3
+ self.validate_template_ownership(self.root_template, self.domain2, self.domain, True)
+
+ @attr(tags=["advanced", "basic", "sg"], required_hardware="false")
+ def test_04_check_non_public_template_access(self):
+ """
+ Verify that non public templates belonging to one domain
+ should not be accessible by other domains by default
+
+ Steps:
+ 1. Set global setting restrict.public.access.to.templates to true
+ 2. Change the permission level of "ispublic" of template to false
+ 3. Make sure other domains should not be able to access the template
+ 4. Make sure that ONLY ROOT domain can access the non public template
+ 5. Set global setting restrict.public.access.to.templates to false
+ 6. Repeat the steps 3 and 4
+ """
+
+ # Step 1
+ self.update_restrict_template_configuration(self.domain1.id, "true")
+ self.update_restrict_template_configuration(self.domain2.id, "true")
+
+ # Step 2
+ self.template2.updatePermissions(self.apiclient,
+ ispublic="False")
+
+ list_template_response = self.list_templates('all', self.domain2)
+ self.assertEqual(
+ isinstance(list_template_response, list),
+ True,
+ "Check list response returns a valid list"
+ )
+ for template_response in list_template_response:
+ if template_response.id == self.template2.id:
+ break
+
+ self.assertIsNotNone(
+ template_response,
+ "Check template %s failed" % self.template2.id
+ )
+ self.assertEqual(
+ template_response.ispublic,
+ int(False),
+ "Check ispublic permission of template"
+ )
+
+ # Step 3
+ # Other domains should not access non public template
+ self.validate_template_ownership(self.template2, self.domain1, self.domain2, False)
+
+ # Step 4
+ # Only ROOT domain can access non public templates of child domain
+ self.validate_template_ownership(self.template2, self.domain, self.domain2, True)
+
+ # Step 5
+ self.update_restrict_template_configuration(self.domain1.id, "false")
+ self.update_restrict_template_configuration(self.domain2.id, "false")
+
+ # Step 6
+ self.validate_template_ownership(self.template2, self.domain1, self.domain2, False)
+ self.validate_template_ownership(self.template2, self.domain, self.domain2, True)
+
+ @attr(tags=["advanced", "basic", "sg"], required_hardware="false")
+ def test_05_check_non_public_template_subdomain_access(self):
+ """
+ Verify that non public templates belonging to ROOT domain
+ should not be accessible by sub domains by default
+
+ Steps:
+ 1. Set global setting restrict.public.access.to.templates to true
+ 2. Change the permission level of "ispublic" of template to false
+ 3. Make sure other domains should not be able to access the template
+ 4. Make sure that ONLY ROOT domain can access the non public template
+ 5. Set global setting restrict.public.access.to.templates to false
+ 6. Repeat the steps 3 and 4
+ """
+ self.update_restrict_template_configuration(self.domain1.id, "true")
+ self.update_restrict_template_configuration(self.domain2.id, "true")
+ self.root_template.updatePermissions(self.apiclient,
+ ispublic="False")
+
+ list_template_response = self.list_templates('all', self.domain)
+ self.assertEqual(
+ isinstance(list_template_response, list),
+ True,
+ "Check list response returns a valid list"
+ )
+ for template_response in list_template_response:
+ if template_response.id == self.root_template.id:
+ break
+
+ self.assertIsNotNone(
+ template_response,
+ "Check template %s failed" % self.root_template.id
+ )
+ self.assertEqual(
+ template_response.ispublic,
+ int(False),
+ "Check ispublic permission of template"
+ )
+
+ # Other domains should not access non public template
+ self.validate_template_ownership(self.root_template, self.domain1, self.domain, False)
+ # Only ROOT domain can access non public templates of child domain
+ self.validate_template_ownership(self.root_template, self.domain2, self.domain, False)
+
+ self.update_restrict_template_configuration(self.domain1.id, "false")
+ self.update_restrict_template_configuration(self.domain2.id, "false")
+ self.validate_template_ownership(self.root_template, self.domain1, self.domain2, False)
+ self.validate_template_ownership(self.root_template, self.domain2, self.domain2, False)
+
+ @attr(tags=["advanced", "basic", "sg"], required_hardware="false")
+ def test_06_check_sub_public_template_sub_domain_access(self):
+ """
+ Verify that non root admin sub-domains can access parents templates
+
+ Steps:
+ 1. Set global setting restrict.public.access.to.templates to true
+ 2. Make sure that sub-domain account can access root templates
+ 3. Make sure that sub-domain account can access parent templates
+ 4. Make sure that ROOT domain can access the sub-domain template
+ 5. Make sure that sibling domain cannot access templates of sub-domain
+ """
+
+ self.root_template.updatePermissions(self.apiclient,
+ ispublic="True")
+ # Step 1
+ self.update_restrict_template_configuration(self.domain1.id, "true")
+ self.update_restrict_template_configuration(self.domain2.id, "true")
+ # Make sure child domains can still access parent domain templates
+ self.validate_uploaded_template(self.apiclient, self.sub_template.id)
+
+ # Step 2
+ self.validate_template_ownership(self.root_template, self.sub_domain, self.domain, True)
+
+ # Step 3
+ self.validate_template_ownership(self.template1, self.sub_domain, self.domain1, True)
+
+ # Step 4
+ self.validate_template_ownership(self.sub_template, self.domain, self.sub_domain, True)
+
+ # Step 5
+ self.validate_template_ownership(self.sub_template, self.domain2, self.sub_domain, False)
+
+ @attr(tags=["advanced", "basic", "sg"], required_hardware="false")
+ def test_07_check_default_public_template_sub_domain_access(self):
+ """
+ Verify that non root admin sub-domains can access parents templates by default
+
+ Steps:
+ 1. Set global setting restrict.public.access.to.templates to false
+ 2. Make sure that sub-domain account can access root templates
+ 3. Make sure that sub-domain account can access parent templates
+ 4. Make sure that ROOT domain can access the sub-domain template
+ 5. Make sure that sibling domain cannot access templates of sub-domain
+ """
+
+ # Step 1
+ self.update_restrict_template_configuration(self.domain1.id, "false")
+ self.update_restrict_template_configuration(self.domain2.id, "false")
+ # Make sure child domains can still access parent domain templates
+ self.validate_uploaded_template(self.apiclient, self.sub_template.id)
+
+ # Step 2
+ self.validate_template_ownership(self.root_template, self.sub_domain, self.domain, True)
+
+ # Step 3
+ self.validate_template_ownership(self.template1, self.sub_domain, self.domain1, True)
+
+ # Step 4
+ self.validate_template_ownership(self.sub_template, self.domain, self.sub_domain, True)
+
+ # Step 5
+ self.validate_template_ownership(self.sub_template, self.domain2, self.sub_domain, True)
+
+ @attr(tags=["advanced", "basic", "sg"], required_hardware="false")
+ def test_08_check_non_public_template_sub_domain_access(self):
+ """
+ Verify that non public templates belonging to one domain
+ should not be accessible by other domains by default except ROOT domain
+
+ Steps:
+ 1. Set global setting restrict.public.access.to.templates to true
+ 2. Change the permission level of "ispublic" of template1 to false
+ 3. Make sure other domains should not be able to access the template
+ 4. Make sure that ONLY ROOT domain can access the non public template
+ 5. Set global setting restrict.public.access.to.templates to false
+ 6. Repeat the steps 3 and 4
+ """
+
+ # Step 1
+ self.update_restrict_template_configuration(self.domain1.id, "true")
+ self.update_restrict_template_configuration(self.domain2.id, "true")
+
+ # Step 2
+ self.template1.updatePermissions(self.apiclient,
+ ispublic="False")
+
+ list_template_response = self.list_templates('all', self.domain1)
+ for template_response in list_template_response:
+ if template_response.id == self.template1.id:
+ break
+
+ self.assertEqual(
+ isinstance(list_template_response, list),
+ True,
+ "Check list response returns a valid list"
+ )
+ self.assertIsNotNone(
+ template_response,
+ "Check template %s failed" % self.template1.id
+ )
+ self.assertEqual(
+ template_response.ispublic,
+ int(False),
+ "Check ispublic permission of template"
+ )
+
+ # Step 3
+ # Other domains should not access non public template
+ self.validate_template_ownership(self.template1, self.domain2, self.domain1, False)
+
+ # Even child domain should not access non public template
+ self.validate_template_ownership(self.template1, self.sub_domain, self.domain1, False)
+
+ # Step 4
+ # Only ROOT domain can access non public templates of child domain
+ self.validate_template_ownership(self.template1, self.domain, self.domain1, True)
+
+ # Step 5
+ self.update_restrict_template_configuration(self.domain1.id, "false")
+ self.update_restrict_template_configuration(self.domain2.id, "false")
+
+ # Step 6
+ self.validate_template_ownership(self.template1, self.domain2, self.domain1, False)
+ self.validate_template_ownership(self.template1, self.sub_domain, self.domain1, False)
+ self.validate_template_ownership(self.template1, self.domain, self.domain1, True)
+
+ def validate_uploaded_template(self, apiclient, template_id, retries=70, interval=5):
+ """Check if template download will finish in 1 minute"""
+ while retries > -1:
+ time.sleep(interval)
+ template_response = Template.list(
+ apiclient,
+ id=template_id,
+ zoneid=self.zone.id,
+ templatefilter='self'
+ )
+
+ if isinstance(template_response, list):
+ template = template_response[0]
+ if not hasattr(template, 'status') or not template or not template.status:
+ retries = retries - 1
+ continue
+ if 'Failed' in template.status:
+ raise Exception(
+ "Failed to download template: status - %s" %
+ template.status)
+
+ elif template.status == 'Download Complete' and template.isready:
+ return
+
+ elif 'Downloaded' in template.status:
+ retries = retries - 1
+ continue
+
+ elif 'Installing' not in template.status:
+ if retries >= 0:
+ retries = retries - 1
+ continue
+ raise Exception(
+ "Error in downloading template: status - %s" %
+ template.status)
+
+ else:
+ retries = retries - 1
+ raise Exception("Template download failed exception.")
+
+ def list_templates(self, templatefilter, domain):
+ return Template.list(
+ self.apiclient,
+ templatefilter=templatefilter,
+ zoneid=self.zone.id,
+ domainid=domain.id)
+
+ def validate_template_ownership(self, template, owner, nonowner, include_cross_domain_template):
+ """List the template belonging to domain which created it
+ Make sure that other domain can't access it.
+ """
+ list_template_response = self.list_templates('all', owner)
+ if list_template_response is not None:
+ """If global setting is false then public templates of any domain should
+ be accessible by any other domain
+ """
+ if include_cross_domain_template:
+ for temp in list_template_response:
+ if template.name == temp.name:
+ return
+
+ raise Exception("Template %s belonging to domain %s should "
+ "be accessible by domain %s"
+ % (template.name, nonowner.name, owner.name))
+ else:
+ """If global setting is true then public templates of any domain should not
+ be accessible by any other domain except for root domain
+ """
+ for temp in list_template_response:
+ if template.name == temp.name:
+ raise Exception("Template %s belonging to domain %s should "
+ "not be accessible by domain %s"
+ % (template.name, nonowner.name, owner.name))
+
+ def get_restrict_template_configuration(self, domain_id):
+ """
+ Function to get the global setting "restrict.public.access.to.templates" for domain
+ """
+ list_configurations_cmd = listConfigurations.listConfigurationsCmd()
+ list_configurations_cmd.name = "restrict.public.template.access.to.domain"
+ list_configurations_cmd.scopename = "domain"
+ list_configurations_cmd.scopeid = domain_id
+ response = self.apiclient.listConfigurations(list_configurations_cmd)
+ return response[0].value
+
+ def update_restrict_template_configuration(self, domain_id, value):
+ """
+ Function to update the global setting "restrict.public.access.to.templates" for domain
+ """
+ update_configuration_cmd = updateConfiguration.updateConfigurationCmd()
+ update_configuration_cmd.name = "restrict.public.template.access.to.domain"
Review Comment:
@soreana
since you have changed the name of global configuration to `share.public.templates.with.other.domains`, can you change the name in component test ? (maybe the value as well)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@cloudstack.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org