Show volume attachment host information for services

When fetching the attachments associated with a volume, Cinder
only return the host information (host_name) if the request
is coming from an admin otherwise returns None.
This is the correct behavior but with a little caveat.
When Glance is configured to use Cinder as it's backend, it
also requires the volume attachment information to find out
which attachments exists for a specific host.
This helps make smart decisions on wheather we want to
disconnect a volume or not[1]. The problem currently is, the
``host_name`` field is not exposed to other openstack
services, like glance.
This patch enables services like Glance API to be able
to fetch the ``host_name`` information for attachments
by checking the ``service`` role in the context of the
request. This helps ensures any non-admin user, apart
from the admin or service category shouldn't be able
to view the ``host_name`` information for attachments.

[1] https://review.opendev.org/c/openstack/glance_store/+/962408

Change-Id: Icea868bf6619f3ab0a00c4e72185f07e66e4a932
Signed-off-by: Rajat Dhasmana <rajatdhasmana@gmail.com>
This commit is contained in:
Rajat Dhasmana
2025-09-26 21:32:31 +00:00
parent 8064195521
commit abfb2c6061
2 changed files with 22 additions and 7 deletions

View File

@@ -90,7 +90,7 @@ class ViewBuilder(common.ViewBuilder):
}
ctxt = request.environ['cinder.context']
attachments = self._get_attachments(volume, ctxt.is_admin)
attachments = self._get_attachments(volume, ctxt)
volume_ref['volume']['attachments'] = attachments
if ctxt.is_admin:
@@ -113,7 +113,7 @@ class ViewBuilder(common.ViewBuilder):
"""Determine if volume is encrypted."""
return volume.get('encryption_key_id') is not None
def _get_attachments(self, volume, is_admin):
def _get_attachments(self, volume, ctxt):
"""Retrieve the attachments of the volume object."""
attachments = []
@@ -124,12 +124,17 @@ class ViewBuilder(common.ViewBuilder):
'attachment_id': attachment.get('id'),
'volume_id': attachment.get('volume_id'),
'server_id': attachment.get('instance_uuid'),
'host_name': attachment.get('attached_host'),
'host_name': None,
'device': attachment.get('mountpoint'),
'attached_at': attachment.get('attach_time'),
}
if not is_admin:
a['host_name'] = None
# When glance is cinder backed, we require the
# host_name to determine when to detach a multiattach
# volume. Glance always uses service credentials to
# request Cinder so we are not exposing the host value
# to end users (non-admin).
if ctxt.is_admin or 'service' in ctxt.roles:
a['host_name'] = attachment.get('attached_host')
attachments.append(a)
return attachments

View File

@@ -1086,6 +1086,8 @@ class VolumeApiTest(test.TestCase):
{'revert': {'snapshot_id': fake_snapshot['id']}})
def test_view_get_attachments(self):
req = fakes.HTTPRequest.blank('/v3/volumes')
context = req.environ['cinder.context']
fake_volume = self._fake_create_volume()
fake_volume['attach_status'] = fields.VolumeAttachStatus.ATTACHING
att_time = datetime.datetime(2017, 8, 31, 21, 55, 7,
@@ -1117,7 +1119,8 @@ class VolumeApiTest(test.TestCase):
# get_attachments should only return attachments with the
# attached status = ATTACHED
attachments = ViewBuilder()._get_attachments(fake_volume, True)
context.is_admin = True
attachments = ViewBuilder()._get_attachments(fake_volume, context)
self.assertEqual(1, len(attachments))
self.assertEqual(fake.UUID3, attachments[0]['attachment_id'])
@@ -1128,9 +1131,16 @@ class VolumeApiTest(test.TestCase):
self.assertEqual(att_time, attachments[0]['attached_at'])
# When admin context is false (non-admin), host_name will be None
attachments = ViewBuilder()._get_attachments(fake_volume, False)
context.is_admin = False
attachments = ViewBuilder()._get_attachments(fake_volume, context)
self.assertIsNone(attachments[0]['host_name'])
# When the request is coming from another service (glance),
# We should be able to see 'host_name'
context.roles.append('service')
attachments = ViewBuilder()._get_attachments(fake_volume, context)
self.assertEqual('host1', attachments[0]['host_name'])
@ddt.data(('created_at=gt:', 0), ('created_at=lt:', 2))
@ddt.unpack
def test_volume_index_filter_by_created_at_with_gt_and_lt(self, change,