TPM: support instances with host secret security

Start supporting booting instances with the `host` TPM secret
security. This means setting the `ephemeral` and `private` attributes
on the Libvirt secret correctly, and not undefining the secret once
the instance has spawned. The Libvirt fixture's Secret support is
extended to be able to test all that in a functional test.

For functional testing, we need to:

* Extend our libvirt fixture's Secret object to properly set the usage
  id (which is just the instance UUID) when parsing vTPM secret XML.

Related to blueprint vtpm-live-migration

Change-Id: I5a38a0de76a78b28a205a8d19f2374830054e1ab
Signed-off-by: melanie witt <melwittt@gmail.com>

Author: notartom

nova/conf/libvirt.py

@@ -1636,7 +1636,7 @@
 * ``swtpm_user`` must also be set.
 """),
     cfg.ListOpt('supported_tpm_secret_security',
-        default=['user'],
+        default=['user', 'host'],
         help="""
 The list of TPM security policies supported by this compute host. If a value is
 absent, it is not supported by this host, and any instance that requests it
@@ -1648,6 +1648,10 @@
   accessed by anyone else. The Libvirt secret is private and non-persistent.
   The instance cannot be live-migrated or automatically resumed after host
   reboot.
+* ``host``: The Barbican secret is owned by the instance owner and cannot be
+  accessed by anyone else. The Libvirt secret is public and persistent. It
+  can be read by anyone with sufficient access on the host. The instance can
+  be live-migrated and automatically resumed after host reboot.
 """),
 ]
 

nova/scheduler/request_filter.py

@@ -483,6 +483,9 @@ def tpm_secret_security_filter(
     if security == 'user':
         request_spec.root_required.add(
             os_traits.COMPUTE_SECURITY_TPM_SECRET_SECURITY_USER)
+    elif security == 'host':
+        request_spec.root_required.add(
+            os_traits.COMPUTE_SECURITY_TPM_SECRET_SECURITY_HOST)
     else:
         # We can get here if the requested TPM secret security passed extra
         # spec validation but is not otherwise supported in the code at this

nova/tests/fixtures/libvirt.py

@@ -198,6 +198,7 @@ def _reset():
 VIR_SECRET_USAGE_TYPE_VOLUME = 1
 VIR_SECRET_USAGE_TYPE_CEPH = 2
 VIR_SECRET_USAGE_TYPE_ISCSI = 3
+VIR_SECRET_USAGE_TYPE_VTPM = 5
 
 # metadata types
 VIR_DOMAIN_METADATA_DESCRIPTION = 0
@@ -1834,12 +1835,15 @@ def __init__(self, connection, xml):
     def _parse_xml(self, xml):
         tree = etree.fromstring(xml)
         self._uuid = tree.find('./uuid').text
+        self._ephemeral = tree.get('ephemeral') == 'yes'
         self._private = tree.get('private') == 'yes'
         self._usage_id = None
         usage = tree.find('./usage')
         if usage is not None:
             if usage.get('type') == 'volume':
                 self._usage_id = usage.find('volume').text
+            if usage.get('type') == 'vtpm':
+                self._usage_id = usage.find('name').text
 
     def setValue(self, value, flags=0):
         self._value = value

nova/tests/functional/integrated_helpers.py

@@ -516,12 +516,16 @@ def _delete_server(self, server):
         self.api.delete_server(server['id'])
         self._wait_until_deleted(server)
 
-    def _reboot_server(self, server, hard=False, expected_state='ACTIVE'):
+    def _reboot_server(self, server, hard=False, expected_state='ACTIVE',
+                       api=None):
         """Reboot a server."""
-        self.api.post_server_action(
+        api = api or self.api
+        api.post_server_action(
             server['id'], {'reboot': {'type': 'HARD' if hard else 'SOFT'}},
         )
-        self.notifier.wait_for_versioned_notifications('instance.reboot.end')
+        if expected_state != 'ERROR':
+            self.notifier.wait_for_versioned_notifications(
+                    'instance.reboot.end')
         return self._wait_for_state_change(server, expected_state)
 
     def _show_server(self, server):

nova/tests/functional/libvirt/test_vtpm.py

@@ -18,6 +18,7 @@
 from castellan.common import exception as castellan_exc
 from castellan.common.objects import passphrase
 from castellan.key_manager import key_manager
+import ddt
 import fixtures
 from oslo_log import log as logging
 from oslo_utils import uuidutils
@@ -140,6 +141,7 @@ def remove_consumer(self, context, managed_object_id, consumer_data):
         )
 
 
+@ddt.ddt
 class VTPMServersTest(base.ServersTestBase):
 
     # NOTE: ADMIN_API is intentionally not set to True in order to catch key
@@ -197,6 +199,14 @@ def assertInstanceHasNoSecret(self, server):
         self.assertNotIn('vtpm_secret_uuid', instance.system_metadata)
         self.assertEqual(0, len(self.key_mgr._passphrases))
 
+    def _assert_libvirt_has_secret(self, host, instance_uuid):
+        s = host.driver._host.find_secret('vtpm', instance_uuid)
+        self.assertIsNotNone(s)
+        ctx = nova_context.get_admin_context()
+        instance = objects.Instance.get_by_uuid(ctx, instance_uuid)
+        secret_uuid = instance.system_metadata['vtpm_secret_uuid']
+        self.assertEqual(secret_uuid, s.UUIDString())
+
     def _assert_libvirt_had_secret(self, compute, secret_uuid):
         # This assert is for ephemeral private libvirt secrets that we
         # undefine immediately after guest creation. Examples include 'user'
@@ -206,6 +216,10 @@ def _assert_libvirt_had_secret(self, compute, secret_uuid):
         conn = compute.driver._host.get_connection()
         self.assertIn(secret_uuid, conn._removed_secrets)
 
+    def _assert_libvirt_secret_missing(self, host, instance_uuid):
+        s = host.driver._host.find_secret('vtpm', instance_uuid)
+        self.assertIsNone(s)
+
     def test_tpm_secret_security_user(self):
         self.flags(supported_tpm_secret_security=['user'], group='libvirt')
         host = self.start_compute(hostname='tpm-host')
@@ -251,6 +265,38 @@ def test_create_server(self):
         # ensure we deleted the key now that we no longer need it
         self.assertEqual(0, len(self.key_mgr._passphrases))
 
+    def test_create_server_secret_security_host(self):
+        self.flags(supported_tpm_secret_security=['host'], group='libvirt')
+        compute = self.start_compute()
+
+        # ensure we are reporting the correct traits
+        traits = self._get_provider_traits(self.compute_rp_uuids[compute])
+        self.assertIn('COMPUTE_SECURITY_TPM_SECRET_SECURITY_HOST', traits)
+
+        # create a server with vTPM
+        server = self._create_server_with_vtpm(secret_security='host')
+
+        # ensure our instance's system_metadata field and key manager inventory
+        # is correct
+        self.assertInstanceHasSecret(server)
+
+        # ensure the libvirt secret is defined correctly
+        ctx = nova_context.get_admin_context()
+        instance = objects.Instance.get_by_uuid(ctx, server['id'])
+        conn = self.computes[compute].driver._host.get_connection()
+        secret = conn._secrets[instance.system_metadata['vtpm_secret_uuid']]
+        self.assertFalse(secret._ephemeral)
+        self.assertFalse(secret._private)
+
+        # now delete the server
+        self._delete_server(server)
+
+        # ensure we deleted the key and undefined the secret now that we no
+        # longer need it
+        self.assertEqual(0, len(self.key_mgr._passphrases))
+        self.assertNotIn(instance.system_metadata['vtpm_secret_uuid'],
+                         conn._secrets)
+
     def test_suspend_resume_server(self):
         self.start_compute()
 
@@ -300,6 +346,24 @@ def test_hard_reboot_server(self):
         # is still correct
         self.assertInstanceHasSecret(server)
 
+    @ddt.data(None, 'user', 'host')
+    def test_hard_reboot_server_as_admin(self, secret_security):
+        """Test hard rebooting a non-admin user's instance as admin.
+
+        This should only work for the 'host' TPM secret security policy.
+        """
+        self.start_compute()
+
+        # create a server with vTPM
+        server = self._create_server_with_vtpm(secret_security=secret_security)
+
+        # Attempt to reboot the server as admin, should only work for 'host'.
+        if secret_security == 'host':
+            self._reboot_server(server, hard=True, api=self.admin_api)
+        else:
+            self._reboot_server(server, hard=True, expected_state='ERROR',
+                                api=self.admin_api)
+
     def _test_resize_revert_server__vtpm_to_vtpm(self, extra_specs=None):
         """Test behavior of revert when a vTPM is retained across a resize.
 

nova/tests/unit/scheduler/test_request_filter.py

@@ -749,8 +749,13 @@ def test_virtio_sound_filter(self):
             reqspec.root_required)
         self.assertEqual(set(), reqspec.root_forbidden)
 
-    @ddt.data('flavor', 'image')
-    def test_tpm_secret_security_filter(self, source):
+    @ddt.data(
+        ('flavor', 'user', ot.COMPUTE_SECURITY_TPM_SECRET_SECURITY_USER),
+        ('flavor', 'host', ot.COMPUTE_SECURITY_TPM_SECRET_SECURITY_HOST),
+        ('image', 'user', ot.COMPUTE_SECURITY_TPM_SECRET_SECURITY_USER),
+        ('image', 'host', ot.COMPUTE_SECURITY_TPM_SECRET_SECURITY_HOST))
+    @ddt.unpack
+    def test_tpm_secret_security_filter(self, source, secret_security, trait):
         # First ensure that tpm_secret_security_filter is included
         self.assertIn(request_filter.tpm_secret_security_filter,
                       request_filter.ALL_REQUEST_FILTERS)
@@ -761,14 +766,14 @@ def test_tpm_secret_security_filter(self, source):
                     extra_specs={
                         'hw:tpm_model': 'tpm-tis',
                         'hw:tpm_version': '1.2',
-                        'hw:tpm_secret_security': 'user',
+                        'hw:tpm_secret_security': secret_security,
                     }),
                 image=objects.ImageMeta(properties=objects.ImageMetaProps()))
         elif source == 'image':
             reqspec = objects.RequestSpec(
                 flavor=objects.Flavor(
                     extra_specs={
-                        'hw:tpm_secret_security': 'user',
+                        'hw:tpm_secret_security': secret_security,
                     }),
                 image=objects.ImageMeta(
                     properties=objects.ImageMetaProps(hw_tpm_model='tpm-tis',
@@ -778,9 +783,7 @@ def test_tpm_secret_security_filter(self, source):
         self.assertEqual(set(), reqspec.root_forbidden)
         self.assertTrue(
             request_filter.tpm_secret_security_filter(self.context, reqspec))
-        self.assertEqual(
-            {ot.COMPUTE_SECURITY_TPM_SECRET_SECURITY_USER},
-            reqspec.root_required)
+        self.assertEqual({trait}, reqspec.root_required)
         self.assertEqual(set(), reqspec.root_forbidden)
 
     def test_tpm_secret_security_filter_skip(self):

nova/tests/unit/virt/libvirt/test_driver.py

@@ -20978,6 +20978,55 @@ def test_create_guest__with_vtpm_error(
         # ...and undefined it after, despite the error
         drvr._host.create_secret.return_value.undefine.assert_called_once()
 
+    @mock.patch('nova.virt.libvirt.host.Host')
+    @mock.patch('nova.crypto.ensure_vtpm_secret')
+    def test_get_or_create_secret_for_vtpm_host_security_found(
+            self, mock_secret, mock_host):
+        """Test that the key manager service API is not called.
+
+        If the secret can be found locally from libvirt with 'host' TPM secret
+        security, there should be no call to the key manager API.
+        """
+        instance = objects.Instance(**self.test_instance)
+        instance.flavor.extra_specs = {'hw:tpm_secret_security': 'host'}
+        mock_host.return_value.find_secret.return_value = mock.sentinel.secret
+
+        drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
+        secret, security = drvr._get_or_create_secret_for_vtpm(self.context,
+                                                               instance)
+
+        mock_secret.assert_not_called()
+        self.assertEqual(mock.sentinel.secret, secret)
+        self.assertEqual('host', security)
+
+    @mock.patch('nova.virt.libvirt.host.Host')
+    @mock.patch('nova.crypto.ensure_vtpm_secret')
+    def test_get_or_create_secret_for_vtpm_host_security_not_found(
+            self, mock_secret, mock_host):
+        """Test that the key manager service API is called.
+
+        If the secret is not found locally from libvirt with 'host' TPM secret
+        security, there should be a call to the key manager API.
+        """
+        instance = objects.Instance(**self.test_instance)
+        instance.flavor.extra_specs = {'hw:tpm_secret_security': 'host'}
+        mock_host.return_value.find_secret.return_value = None
+        mock_secret.return_value = (uuids.secret, mock.sentinel.passphrase)
+
+        drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
+        secret, security = drvr._get_or_create_secret_for_vtpm(self.context,
+                                                               instance)
+
+        mock_secret.assert_called_once_with(self.context, instance)
+        # ensure_vtpm_secret() returns (secret_uuid, passphrase)
+        mock_host.return_value.create_secret.assert_called_once_with(
+            'vtpm', uuids.instance, password=mock.sentinel.passphrase,
+            uuid=uuids.secret, ephemeral=False, private=False)
+
+        self.assertEqual(
+            mock_host.return_value.create_secret.return_value, secret)
+        self.assertEqual('host', security)
+
     @mock.patch('nova.virt.disk.api.clean_lxc_namespace')
     @mock.patch('nova.virt.libvirt.driver.LibvirtDriver.get_info')
     @mock.patch('nova.virt.disk.api.setup_container')
@@ -27807,7 +27856,7 @@ def test_cleanup_lvm_encrypted(self, mock_save, mock_undefine_domain,
         instance = objects.Instance(
             uuid=uuids.instance, id=1,
             ephemeral_key_uuid=uuids.ephemeral_key_uuid,
-            resources=None)
+            resources=None, flavor=objects.Flavor())
         instance.system_metadata = {}
         block_device_info = {'root_device_name': '/dev/vda',
                              'ephemerals': [],

nova/tests/unit/virt/libvirt/test_host.py

@@ -21,6 +21,7 @@
 import eventlet
 from eventlet import greenthread
 from eventlet import tpool
+from lxml import etree
 from oslo_serialization import jsonutils
 from oslo_utils.fixture import uuidsentinel as uuids
 from oslo_utils import uuidutils
@@ -1051,6 +1052,21 @@ def test_create_secret(self, mock_sec):
         self.host.create_secret('iscsi', 'iscsivol', password="foo")
         secret.setValue.assert_called_once_with("foo")
 
+    @mock.patch.object(fakelibvirt.virConnect, "secretDefineXML")
+    def test_create_secret_vtpm_ephemeral_private_default(self, mock_sec):
+        self.host.create_secret('vtpm', uuids.instance)
+        xml = etree.fromstring(mock_sec.call_args.args[0])
+        self.assertEqual('yes', xml.get('ephemeral'))
+        self.assertEqual('yes', xml.get('private'))
+
+    @mock.patch.object(fakelibvirt.virConnect, "secretDefineXML")
+    def test_create_secret_vtpm_not_ephemeral_private(self, mock_sec):
+        self.host.create_secret('vtpm', uuids.instance, ephemeral=False,
+                                private=False)
+        xml = etree.fromstring(mock_sec.call_args.args[0])
+        self.assertEqual('no', xml.get('ephemeral'))
+        self.assertEqual('no', xml.get('private'))
+
     @mock.patch('nova.virt.libvirt.host.Host.find_secret')
     def test_delete_secret(self, mock_find_secret):
         """deleting secret."""