| # 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. | |
| """ | |
| Lifecycle integration tests for the KMS (Key Management Service) feature. | |
| Covers: | |
| - HSM profile CRUD (add / list / update / delete) | |
| - KMS key CRUD (create / list / update / delete) | |
| - Key rotation | |
| - Multi-tenancy / access isolation | |
| - Negative scenarios (delete key in use, duplicate name, delete profile with keys) | |
| All tests use the built-in *database* KMS provider so that they can run in any | |
| CI environment without real HSM hardware. | |
| """ | |
| import random | |
| import string | |
| from marvin.cloudstackTestCase import cloudstackTestCase | |
| from marvin.lib.base import ( | |
| Account, | |
| Domain, | |
| HSMProfile, | |
| KMSKey, | |
| VirtualMachine, | |
| ServiceOffering, | |
| Volume, | |
| ) | |
| from marvin.lib.common import get_zone, get_domain, get_template | |
| from marvin.lib.utils import cleanup_resources | |
| from nose.plugins.attrib import attr | |
| def _random_name(prefix="test-kms"): | |
| suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=6)) | |
| return f"{prefix}-{suffix}" | |
| class TestKMSLifecycle(cloudstackTestCase): | |
| """ | |
| End-to-end lifecycle tests for KMS keys and HSM profiles. | |
| Each test gets a fresh Domain + Account pair via setUp so that | |
| domain-level isolation is guaranteed and cleanup is straightforward. | |
| """ | |
| @classmethod | |
| def setUpClass(cls): | |
| cls.test_client = super(TestKMSLifecycle, cls).getClsTestClient() | |
| cls.apiclient = cls.test_client.getApiClient() | |
| cls.zone = get_zone(cls.apiclient, cls.test_client.getZoneForTests()) | |
| cls.domain = get_domain(cls.apiclient) | |
| cls.hypervisor = cls.test_client.getHypervisorInfo() | |
| cls._cleanup = [] | |
| # The built-in database provider is seeded as a public, system-owned profile named | |
| # "HSM Database Provider". listHSMProfiles has no server-side name filter, and a root | |
| # admin's listing without listall is restricted to admin-owned profiles, so we must | |
| # pass listall=True and match the name client-side. | |
| cls.default_profile = cls._find_default_profile(cls.apiclient) | |
| if cls.default_profile and not cls.default_profile.enabled: | |
| hsm = HSMProfile({"id": cls.default_profile.id}) | |
| hsm.update(cls.apiclient, enabled=True) | |
| # Re-fetch to get updated state | |
| cls.default_profile = cls._find_default_profile(cls.apiclient) | |
| @staticmethod | |
| def _find_default_profile(apiclient): | |
| """Locate the seeded public 'HSM Database Provider' profile.""" | |
| profiles = HSMProfile.list( | |
| apiclient, listall=True, keyword="HSM Database Provider" | |
| ) or [] | |
| for profile in profiles: | |
| if profile.name == "HSM Database Provider": | |
| return profile | |
| return None | |
| @classmethod | |
| def tearDownClass(cls): | |
| super(TestKMSLifecycle, cls).tearDownClass() | |
| # ------------------------------------------------------------------ | |
| # Per-test helpers | |
| # ------------------------------------------------------------------ | |
| def setUp(self): | |
| self.apiclient = self.test_client.getApiClient() | |
| self.cleanup = [] | |
| self._create_domain_and_account() | |
| def tearDown(self): | |
| self.cleanup.reverse() | |
| cleanup_resources(self.apiclient, self.cleanup) | |
| def _create_domain_and_account(self, is_domain_admin=False): | |
| """Create a child domain + account and register them for cleanup.""" | |
| self.child_domain = Domain.create( | |
| self.apiclient, | |
| {"name": _random_name("kms-dom")}, | |
| parentdomainid=self.domain.id, | |
| ) | |
| self.cleanup.append(self.child_domain) | |
| self.user_account = Account.create( | |
| self.apiclient, | |
| { | |
| "firstname": "KMS", | |
| "lastname": "Test", | |
| "email": "kmstest@example.com", | |
| "username": _random_name("kmsuser"), | |
| "password": "password", | |
| }, | |
| domainid=self.child_domain.id, | |
| admin=is_domain_admin, | |
| ) | |
| self.cleanup.append(self.user_account) | |
| # API client scoped to the new user | |
| self.user_apiclient = self.test_client.getUserApiClient( | |
| UserName=self.user_account.name, | |
| DomainName=self.child_domain.name, | |
| ) | |
| def _create_kms_key(self, name, profile_id, apiclient=None, zoneid=None, purpose=None): | |
| api_client = apiclient or self.apiclient | |
| zone_id = zoneid or self.zone.id | |
| key = KMSKey.create( | |
| api_client, | |
| name=name, | |
| zoneid=zone_id, | |
| hsmprofileid=profile_id, | |
| purpose=purpose | |
| ) | |
| self.cleanup.append(key) | |
| return key | |
| def _create_hsm_profile(self, name, protocol="database", zoneid=None): | |
| zone_id = zoneid or self.zone.id | |
| profile = HSMProfile.create( | |
| self.apiclient, | |
| name=name, | |
| protocol=protocol, | |
| zoneid=zone_id, | |
| ) | |
| self.cleanup.append(profile) | |
| return profile | |
| # ================================================================== | |
| # HSM Profile lifecycle tests (01 – 03, 11, 13) | |
| # ================================================================== | |
| @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], | |
| required_hardware="false") | |
| def test_01_add_hsm_profile_admin(self): | |
| """Test: admin creates a system-wide database HSM profile.""" | |
| profile_name = _random_name("hsm-prof") | |
| profile = self._create_hsm_profile(name=profile_name) | |
| self.assertIsNotNone(profile, "HSM profile creation returned None") | |
| self.assertEqual( | |
| profile.name, profile_name, | |
| "HSM profile name does not match the requested name" | |
| ) | |
| self.assertEqual( | |
| profile.protocol.lower(), "database", | |
| "HSM profile protocol should be 'database'" | |
| ) | |
| @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], | |
| required_hardware="false") | |
| def test_02_list_hsm_profiles(self): | |
| """Test: list HSM profiles and verify a created profile is present.""" | |
| profile_name = _random_name("hsm-list") | |
| profile = self._create_hsm_profile(name=profile_name) | |
| profiles = HSMProfile.list(self.apiclient, id=profile.id) | |
| self.assertIsNotNone(profiles, "listHSMProfiles returned None") | |
| self.assertEqual(len(profiles), 1, "Expected exactly one HSM profile matching the ID") | |
| self.assertEqual(profiles[0].id, profile.id, "Profile IDs do not match") | |
| @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], | |
| required_hardware="false") | |
| def test_03_update_hsm_profile(self): | |
| """Test: update the name of an existing HSM profile.""" | |
| profile = self._create_hsm_profile(name=_random_name("hsm-upd")) | |
| new_name = _random_name("hsm-renamed") | |
| updated = profile.update(self.apiclient, name=new_name) | |
| self.assertIsNotNone(updated, "updateHSMProfile returned None") | |
| self.assertEqual( | |
| updated.name, new_name, | |
| "Profile name was not updated" | |
| ) | |
| # ================================================================== | |
| # KMS Key CRUD tests (tests 04 – 09) | |
| # ================================================================== | |
| @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], | |
| required_hardware="false") | |
| def test_04_create_kms_key_admin(self): | |
| """Test: admin creates a KMS key in the zone, verifies fields.""" | |
| if not self.default_profile: | |
| self.skipTest("Default HSM profile 'default' not found") | |
| profile = self.default_profile | |
| key_name = _random_name("kms-key") | |
| key = self._create_kms_key(name=key_name, profile_id=profile.id, purpose="volume") | |
| self.assertIsNotNone(key, "createKMSKey returned None") | |
| self.assertEqual(key.name, key_name, "Key name does not match") | |
| self.assertEqual( | |
| key.zoneid, self.zone.id, | |
| "Key zone ID does not match the requested zone" | |
| ) | |
| self.assertTrue(key.enabled, "Newly created key should be enabled") | |
| self.assertIsNotNone(key.id, "Key UUID should not be None") | |
| @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], | |
| required_hardware="false") | |
| def test_05_list_kms_keys(self): | |
| """Test: list KMS keys filtered by zone and by id.""" | |
| if not self.default_profile: | |
| self.skipTest("Default HSM profile 'default' not found") | |
| profile = self.default_profile | |
| key = self._create_kms_key(name=_random_name("key"), profile_id=profile.id) | |
| # Filter by explicit key ID | |
| keys_by_id = KMSKey.list(self.apiclient, id=key.id) | |
| self.assertIsNotNone(keys_by_id, "listKMSKeys by id returned None") | |
| self.assertEqual(len(keys_by_id), 1, "Expected exactly one key matching given ID") | |
| self.assertEqual(keys_by_id[0].id, key.id, "Key IDs do not match") | |
| # Filter by zone | |
| keys_by_zone = KMSKey.list(self.apiclient, zoneid=self.zone.id) | |
| self.assertIsNotNone(keys_by_zone, "listKMSKeys by zone returned None") | |
| found = any(k.id == key.id for k in keys_by_zone) | |
| self.assertTrue(found, "Newly created key not found when listing by zone") | |
| @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], | |
| required_hardware="false") | |
| def test_06_update_kms_key(self): | |
| """Test: update key name, description, and enabled status.""" | |
| if not self.default_profile: | |
| self.skipTest("Default HSM profile 'default' not found") | |
| profile = self.default_profile | |
| key = self._create_kms_key(name=_random_name("key-upd"), profile_id=profile.id) | |
| new_name = _random_name("key-renamed") | |
| new_desc = "Updated description" | |
| updated = key.update( | |
| self.apiclient, | |
| name=new_name, | |
| description=new_desc, | |
| enabled=False, | |
| ) | |
| self.assertIsNotNone(updated, "updateKMSKey returned None") | |
| self.assertEqual(updated.name, new_name, "Key name was not updated") | |
| self.assertEqual(updated.description, new_desc, "Key description was not updated") | |
| self.assertFalse(updated.enabled, "Key should be disabled after update") | |
| @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], | |
| required_hardware="false") | |
| def test_07_create_kms_key_user(self): | |
| """Test: domain user creates their own KMS key; verifies ownership.""" | |
| # Admin creates the system HSM profile first | |
| if not self.default_profile: | |
| self.skipTest("Default HSM profile 'default' not found") | |
| profile = self.default_profile | |
| key_name = _random_name("user-key") | |
| key = self._create_kms_key(name=key_name, profile_id=profile.id, apiclient=self.user_apiclient) | |
| self.assertIsNotNone(key, "User-level createKMSKey returned None") | |
| self.assertEqual(key.name, key_name, "Key name does not match") | |
| self.assertEqual( | |
| key.account, self.user_account.name, | |
| "Key account should belong to the creating user account" | |
| ) | |
| @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], | |
| required_hardware="false") | |
| def test_08_list_kms_keys_user_isolation(self): | |
| """Test: User A's keys are NOT visible to User B.""" | |
| if not self.default_profile: | |
| self.skipTest("Default HSM profile 'default' not found") | |
| profile = self.default_profile | |
| # User A key (self.user_account) | |
| key_a = self._create_kms_key(name=_random_name("key-a"), profile_id=profile.id, apiclient=self.user_apiclient) | |
| # Create User B in a separate child domain | |
| domain_b = Domain.create( | |
| self.apiclient, | |
| {"name": _random_name("dom-b")}, | |
| parentdomainid=self.domain.id, | |
| ) | |
| self.cleanup.append(domain_b) | |
| account_b = Account.create( | |
| self.apiclient, | |
| { | |
| "firstname": "UserB", | |
| "lastname": "Test", | |
| "email": "userb@example.com", | |
| "username": _random_name("userb"), | |
| "password": "password", | |
| }, | |
| domainid=domain_b.id, | |
| admin=False, | |
| ) | |
| self.cleanup.append(account_b) | |
| apiclient_b = self.test_client.getUserApiClient( | |
| UserName=account_b.name, | |
| DomainName=domain_b.name, | |
| ) | |
| # User B should not be able to see User A's key | |
| keys_for_b = KMSKey.list(apiclient_b, id=key_a.id) | |
| if keys_for_b: | |
| self.assertNotEqual( | |
| keys_for_b[0].accountid, self.user_account.id, | |
| "User B should not see User A's KMS keys" | |
| ) | |
| @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], | |
| required_hardware="false") | |
| def test_09_delete_kms_key(self): | |
| """Test: delete a KMS key that is not in use; verify it is gone.""" | |
| if not self.default_profile: | |
| self.skipTest("Default HSM profile 'default' not found") | |
| profile = self.default_profile | |
| key = self._create_kms_key(name=_random_name("key-del"), profile_id=profile.id) | |
| key.delete(self.apiclient) | |
| self.cleanup.remove(key) | |
| # Verify the key no longer appears in listings | |
| keys = KMSKey.list(self.apiclient, id=key.id) | |
| self.assertTrue( | |
| not keys, | |
| "Deleted KMS key should not appear in listKMSKeys" | |
| ) | |
| # ================================================================== | |
| # Key rotation (test 10) | |
| # ================================================================== | |
| @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], | |
| required_hardware="false") | |
| def test_10_rotate_kms_key(self): | |
| """Test: rotate a KMS key; verify the key version increments.""" | |
| if not self.default_profile: | |
| self.skipTest("Default HSM profile 'default' not found") | |
| profile = self.default_profile | |
| key = self._create_kms_key(name=_random_name("key-rot"), profile_id=profile.id) | |
| initial_version = key.version | |
| key.rotate(self.apiclient) | |
| # Fetch the updated key details and confirm version incremented | |
| keys = KMSKey.list(self.apiclient, id=key.id) | |
| self.assertIsNotNone(keys, "listKMSKeys after rotation returned None") | |
| self.assertEqual(len(keys), 1, "Expected exactly one key after rotation") | |
| rotated_key = keys[0] | |
| self.assertGreater( | |
| rotated_key.version, | |
| initial_version, | |
| f"Key version should increase after rotation (was {initial_version}, " | |
| f"got {rotated_key.version})" | |
| ) | |
| # ================================================================== | |
| # Negative tests (11) | |
| # ================================================================== | |
| @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], | |
| required_hardware="false") | |
| def test_11_delete_hsm_profile_with_keys_negative(self): | |
| """ | |
| Negative test: deleting an HSM profile that still has associated KMS | |
| keys should be rejected. | |
| """ | |
| profile = self._create_hsm_profile(name=_random_name("hsm-with-key")) | |
| key = self._create_kms_key(name=_random_name("key-blocks-del"), profile_id=profile.id) | |
| with self.assertRaises(Exception, | |
| msg="Deleting HSM profile with active keys should fail"): | |
| profile.delete(self.apiclient) | |
| # ================================================================== | |
| # VM Encryption tests (12) | |
| # ================================================================== | |
| @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], | |
| required_hardware="false") | |
| def test_12_deploy_vm_with_root_disk_encryption(self): | |
| """ | |
| Test: deploy a VM with its root disk encrypted using a KMS key. | |
| Verify that the VM starts and the root volume has the KMS key ID. | |
| """ | |
| if self.hypervisor.lower() != 'kvm': | |
| raise self.skipTest("Skipping test case for non-kvm hypervisor") | |
| # 1. Create a KMS key for the user | |
| key = self._create_kms_key(name=_random_name("vm-root-key"), profile_id=self.default_profile.id, apiclient=self.user_apiclient) | |
| # 2. Get a template and a service offering | |
| template = get_template( | |
| self.apiclient, | |
| self.zone.id, | |
| self.test_client.getParsedTestDataConfig().get("ostype", "CentOS 7.0 (64-bit)") | |
| ) | |
| if template == -1 or template is None: | |
| self.fail("Check for template failed") | |
| service_offering = ServiceOffering.create( | |
| self.apiclient, | |
| self.test_client.getParsedTestDataConfig()["service_offering"] | |
| ) | |
| self.cleanup.append(service_offering) | |
| # 3. Deploy VM with root disk encryption | |
| vm = VirtualMachine.create( | |
| self.user_apiclient, | |
| self.test_client.getParsedTestDataConfig()["virtual_machine"], | |
| templateid=template.id, | |
| accountid=self.user_account.name, | |
| domainid=self.child_domain.id, | |
| serviceofferingid=service_offering.id, | |
| zoneid=self.zone.id, | |
| rootdiskkmskeyid=key.id | |
| ) | |
| self.cleanup.append(vm) | |
| self.assertEqual( | |
| vm.state, | |
| "Running", | |
| "VM should be in Running state after deployment" | |
| ) | |
| # 4. Verify the root volume has the KMS key ID | |
| volumes = Volume.list( | |
| self.user_apiclient, | |
| virtualmachineid=vm.id, | |
| type='ROOT', | |
| listall=True | |
| ) | |
| self.assertTrue( | |
| volumes and len(volumes) > 0, | |
| "VM should have at least one ROOT volume" | |
| ) | |
| root_volume = volumes[0] | |
| self.assertEqual( | |
| str(root_volume.kmskeyid), | |
| str(key.id), | |
| f"Root volume should have KMS key ID {key.id}, found {root_volume.kmskeyid}" | |
| ) | |
| # ================================================================== | |
| # HSM Profile cleanup (13) | |
| # ================================================================== | |
| @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], | |
| required_hardware="false") | |
| def test_13_delete_hsm_profile(self): | |
| """Test: delete an HSM profile that has no associated keys; verify it is gone.""" | |
| profile = self._create_hsm_profile(name=_random_name("hsm-gone")) | |
| profile.delete(self.apiclient) | |
| self.cleanup.remove(profile) | |
| profiles = HSMProfile.list(self.apiclient, id=profile.id) | |
| self.assertTrue( | |
| not profiles, | |
| "Deleted HSM profile should not appear in listHSMProfiles" | |
| ) |