Add vTPM testcases

This commit will add tests to cover vTPM device support for instances.
The vTPM device allows storing secrets at instance level and its managed
by the Barbican backend.

The _vptm_server_creation_check helper method is used to create server
with specific vtpm version and model and assert that it is configured as
needed from the instance xml.

The test_create_server_with_vtpm_tis method will verify creation of
instance with tpm-tis model and supported version 2.0.

Similarly, test_create_server_with_vtpm_crb will verify creation of
instance with tpm-crb model and supported version 2.0.

In addition the Barbican client service was leveraged from the barbican
tempest plugin [1]. This is to allow the vTPM test to communicate with
the barbican client, confirm the secret key found in the guest domain is
present in the client, the key is active, and the keys description
accuratly describes its purpose is vTPM for the guest. Example reply
from barbican below:

{'algorithm': None,
 'bit_length': None,
 'content_types': {'default': 'application/octet-stream'},
 'created': '2021-10-13T18:17:52',
 'creator_id': '4b1cc6071236438c881f9da54657468f',
 'expiration': None,
 'mode': None,
 'name': 'vTPM secret for instance b537c0df-0e39-4af8-94b3-04bcc8262f20',
 'secret_ref': 'http://192.168.24.3:9311/v1/secrets/13a9ae5e-5187-4c0f-acde-b2cda06ae00c',
 'secret_type': 'passphrase',
 'status': 'ACTIVE',
 'updated': '2021-10-13T18:17:52'}

[1] https://github.com/openstack/barbican-tempest-plugin

Related to:
https://review.opendev.org/c/openstack/nova/+/631363/
https://review.opendev.org/c/openstack/glance/+/633256/
https://bugzilla.redhat.com/show_bug.cgi?id=1782128

Change-Id: I7b1a1306beb871a9294884116f6430ead91ce601

.zuul.yaml

@@ -44,6 +44,10 @@
       tox_envlist: all
       tempest_concurrency: 1
       tempest_test_regex: ^whitebox_tempest_plugin\.
+      # NOTE(jparker) in order for guest to boot via UEFI, the host will need the
+      # open source implementation of UEFI for VMs via the OVMF package. In
+      # addition to test vTPM hosts need swtpm as well
+      extra_packages: ovmf,swtpm-tools
       devstack_localrc:
         MAX_COMPUTE_NODES: 2
         LIBVIRT_TYPE: kvm
@@ -73,6 +77,9 @@
               cpu_model_extra_flags: vme,+ssse3,-mmx
               virt_type: kvm
               rx_queue_size: 1024
+              swtpm_enabled: True
+              swtpm_user: swtpm
+              swtpm_group: swtpm
     group-vars:
       subnode:
         num_hugepages: 2048
@@ -93,6 +100,9 @@
                 cpu_model_extra_flags: vme,+ssse3,-mmx
                 virt_type: kvm
                 rx_queue_size: 1024
+                swtpm_enabled: True
+                swtpm_user: swtpm
+                swtpm_group: swtpm
       tempest:
         num_hugepages: 512
         devstack_plugins:

devstack/plugin.sh

@@ -39,6 +39,7 @@ function configure {
     iniset $TEMPEST_CONFIG compute-feature-enabled virtio_rng "$COMPUTE_FEATURE_VIRTIO_RNG"
     iniset $TEMPEST_CONFIG compute-feature-enabled rbd_download "$COMPUTE_FEATURE_RBD_DOWNLOAD"
     iniset $TEMPEST_CONFIG compute-feature-enabled uefi_secure_boot "$COMPUTE_FEATURE_UEFI_SECURE_BOOT"
+    iniset $TEMPEST_CONFIG compute-feature-enabled vtpm_device_supported "$COMPUTE_FEATURE_VTPM_ENABLED"
 }
 
 if [[ "$1" == "stack" ]]; then

devstack/settings

@@ -23,3 +23,4 @@ WHITEBOX_SHARED_CPUS_PER_NUMA=${WHITEBOX_SHARED_CPUS_PER_NUMA:-2}
 COMPUTE_FEATURE_VIRTIO_RNG=${COMPUTE_FEATURE_VIRTIO_RNG:-'True'}
 COMPUTE_FEATURE_RBD_DOWNLOAD=${COMPUTE_FEATURE_RBD_DOWNLOAD:-'False'}
 COMPUTE_FEATURE_UEFI_SECURE_BOOT=${COMPUTE_FEATURE_UEFI_SECURE_BOOT:-'True'}
+COMPUTE_FEATURE_VTPM_ENABLED=${COMPUTE_FEATURE_VTPM_ENABLED:-'True'}

playbooks/whitebox/pre.yaml

@@ -8,13 +8,14 @@
           name: crudini
           state: present
       become: yes
-    # NOTE(jparker) in order for guest to boot via UEFI, the host will need the
-    # open source implementation of UEFI for VMs via the OVMF package.
-    - name: Install ovmf
+    - name: Install additional packages needed by job environment
       package:
-        name: ovmf
+        name: "{{ item }}"
         state: present
       become: yes
+      loop:
+        - "{{ extra_packages }}"
+      when: extra_packages is defined
     # NOTE(artom) The run-tempest role runs as the tempest user, so we need to give
     # the tempest user SSH access to all hosts. Devstack's orchestrate-devstack
     # role should have put a pubkey into the stack user's authorized_keys, so if we

whitebox_tempest_plugin/api/compute/test_vtpm.py

@@ -0,0 +1,119 @@
+# Copyright 2023 Red Hat
+# All Rights Reserved.
+#
+#    Licensed 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.
+
+from tempest import config
+from tempest.exceptions import BuildErrorException
+from whitebox_tempest_plugin.api.compute import base
+
+CONF = config.CONF
+
+
+class VTPMTest(base.BaseWhiteboxComputeTest):
+    """Tests Virtual Trusted Platform Module (vTPM) device support for instance.
+    Creating instance with a variety of device versions and module types are
+    tested. Tests require creating instance flavor with extra specs about the
+    tpm version and model to be specified and Barbican Key manager must enabled
+    in the environement to manage the instance secrets.
+    """
+
+    @classmethod
+    def skip_checks(cls):
+        super(VTPMTest, cls).skip_checks()
+        if (CONF.compute_feature_enabled.vtpm_device_supported is False):
+            msg = "CONF.compute_feature_enabled.vtpm_device_supported must " \
+                "be set."
+            raise cls.skipException(msg)
+
+    @classmethod
+    def setup_clients(cls):
+        super(VTPMTest, cls).setup_clients()
+        os = getattr(cls, 'os_primary')
+        cls.secret_client = os.secret_v1.SecretClient(
+            service='key-manager'
+        )
+
+    def _vptm_server_creation_check(self, vtpm_model, vtpm_version):
+        """Test to verify creating server with vTPM device
+
+        This test creates a server with specific tpm version and model
+        and verifies the same is configured by fetching instance xml.
+        """
+
+        flavor_specs = {'hw:tpm_version': vtpm_version,
+                        'hw:tpm_model': vtpm_model}
+        vtpm_flavor = self.create_flavor(extra_specs=flavor_specs)
+
+        # Create server with vtpm device and fetch xml data
+        server = self.create_test_server(flavor=vtpm_flavor['id'],
+                                         wait_until="ACTIVE")
+        server_xml = self.get_server_xml(server['id'])
+
+        # Assert tpm model found in vTPM XML element is correct
+        vtpm_element = server_xml.find('./devices/tpm[@model]')
+        vtpm_model_found = vtpm_element.get('model')
+        self.assertEqual(
+            vtpm_model, vtpm_model_found, 'Expected vTPM model %s not found '
+            'instead found: %s' % (vtpm_model, vtpm_model_found))
+
+        # Assert tpm version found in vTPM element is correct
+        vtpm_version_found = \
+            vtpm_element.find('.backend[@version]').get('version')
+        self.assertEqual(
+            vtpm_version, vtpm_version_found, 'Expeted vTPM version %s not '
+            'found instead found: %s' % (vtpm_version, vtpm_version_found))
+
+        # Assert secret is present in the vTPM XML element
+        vtpm_secret_element = vtpm_element.find('.backend/encryption')
+        self.assertIsNotNone(
+            vtpm_secret_element.get('secret'), 'Secret not found on vTPM '
+            'element')
+
+        # Get the secret uuid and get secret details from barbican
+        secret_uuid = secret_uuid = vtpm_secret_element.get('secret')
+        secret_info = self.secret_client.get_secret(secret_uuid)
+
+        # Confirm the secret is ACTIVE and its description mentions the
+        # respective server uuid and it is used for vTPM
+        self.assertEqual(
+            'ACTIVE', secret_info.get('status'), 'Secret is not ACTIVE, '
+            'current status: %s' % secret_info.get('status'))
+        self.assertTrue(
+            server['id'] in secret_info.get('name'), 'Server id not present '
+            'in secret key information: %s' % secret_info.get('name'))
+        self.assertTrue(
+            'vtpm' in secret_info.get('name').lower(), 'No mention of vTPM in '
+            'secret description: %s' % secret_info.get('name'))
+
+        # Delete server after test
+        self.delete_server(server['id'])
+
+    def test_create_server_with_vtpm_tis(self):
+        # Test creating server with tpm-tis model and versions supported
+        self._vptm_server_creation_check('tpm-tis', '2.0')
+
+    def test_create_server_with_vtpm_crb(self):
+        # Test creating server with tpm-crb model and versions supported
+        self._vptm_server_creation_check('tpm-crb', '2.0')
+
+    def test_invalid_model_version_creation(self):
+        # Test attempting to create a server with an invalid model/version
+        # combination model
+        flavor_specs = {'hw:tpm_version': '1.2',
+                        'hw:tpm_model': '2.0'}
+        vtpm_flavor = self.create_flavor(extra_specs=flavor_specs)
+        self.assertRaises(BuildErrorException,
+                          self.create_test_server,
+                          flavor=vtpm_flavor['id'],
+                          wait_until='ACTIVE')

whitebox_tempest_plugin/config.py

@@ -310,5 +310,13 @@
                 help="If false, skip standard uefi boot tests"),
     cfg.BoolOpt('uefi_secure_boot',
                 default=False,
-                help="If false, skip uefi secure boot tests")
+                help="If false, skip uefi secure boot tests"),
+    cfg.BoolOpt('vtpm_device_supported',
+                default=False,
+                help='vTPM device support for guest instances to store '
+                     'secrets. Requires these flags set in nova.conf'
+                     '[libvirt]/swtpm_enabled=True'
+                     '[libvirt]/swtpm_user=tss'
+                     '[libvirt]/swtpm_group=tss'
+                     '[key_manager]/backend=barbican')
 ]

whitebox_tempest_plugin/plugin.py

@@ -56,3 +56,14 @@ def get_opt_lists(self):
                  whitebox_config.database_opts),
                 (whitebox_config.hardware_group.name,
                  whitebox_config.hardware_opts)]
+
+    def get_service_clients(self):
+        v1_params = {
+            'name': 'secret_v1',
+            'service_version': 'secret.v1',
+            'module_path': 'whitebox_tempest_plugin.services.key_manager.json',
+            'client_names': [
+                'SecretClient'
+            ]
+        }
+        return [v1_params]

whitebox_tempest_plugin/services/key_manager/json/__init__.py

@@ -0,0 +1,22 @@
+# Copyright 2021 Red Hat Inc.
+# All Rights Reserved.
+#
+#    Licensed 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.
+
+
+from whitebox_tempest_plugin.services.key_manager.json.secret_client \
+    import SecretClient
+
+__all__ = [
+    'SecretClient',
+]

whitebox_tempest_plugin/services/key_manager/json/base.py

@@ -0,0 +1,30 @@
+# Copyright 2021 Red Hat Inc.
+# All Rights Reserved.
+#
+#    Licensed 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.
+
+from tempest.lib.common import rest_client
+
+
+_DEFAULT_SERVICE_TYPE = 'key-manager'
+
+
+class BarbicanTempestClient(rest_client.RestClient):
+
+    def __init__(self, *args, **kwargs):
+        kwargs['service'] = _DEFAULT_SERVICE_TYPE
+        super().__init__(*args, **kwargs)
+
+    @classmethod
+    def ref_to_uuid(cls, href):
+        return href.split('/')[-1]

whitebox_tempest_plugin/services/key_manager/json/secret_client.py

@@ -0,0 +1,48 @@
+# Copyright 2021 Red Hat Inc.
+# All Rights Reserved.
+#
+#    Licensed 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.
+
+from tempest import config
+
+from whitebox_tempest_plugin.services.key_manager.json import base
+
+
+CONF = config.CONF
+
+
+class SecretClient(base.BarbicanTempestClient):
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._secret_ids = set()
+
+    def get_secret(self, secret_id):
+        resp, body = self.get("v1/secrets/%s" % secret_id)
+        self.expected_success(200, resp.status)
+        return self._parse_resp(body)
+
+    def list_secrets(self, **kwargs):
+        uri = "v1/secrets"
+        if kwargs is not None:
+            uri = '{base}?'.format(base=uri)
+
+            for key in kwargs.keys():
+                uri = '{base}&{name}={value}'.format(
+                    base=uri,
+                    name=key,
+                    value=kwargs[key]
+                )
+        resp, body = self.get(uri)
+        self.expected_success(200, resp.status)
+        return self._parse_resp(body)