Merge "Improve is_image_available"

This commit is contained in:
Zuul 2025-04-15 17:22:53 +00:00 committed by Gerrit Code Review
commit 621338e967
4 changed files with 125 additions and 14 deletions

View File

@ -22,6 +22,7 @@ from oslo_utils import timeutils
from oslo_utils import uuidutils from oslo_utils import uuidutils
from ironic.common import exception from ironic.common import exception
from ironic.common import keystone
from ironic.conf import CONF from ironic.conf import CONF
_IMAGE_ATTRIBUTES = ['size', 'disk_format', 'owner', _IMAGE_ATTRIBUTES = ['size', 'disk_format', 'owner',
@ -34,6 +35,7 @@ _IMAGE_ATTRIBUTES = ['size', 'disk_format', 'owner',
LOG = log.getLogger(__name__) LOG = log.getLogger(__name__)
_GLANCE_SESSION = None
def _extract_attributes(image): def _extract_attributes(image):
@ -123,28 +125,40 @@ def is_image_available(context, image):
This check is needed in case Nova and Glance are deployed This check is needed in case Nova and Glance are deployed
without authentication turned on. without authentication turned on.
""" """
# NOTE: Any support for private/shared images in Ironic requires a secure
# way for ironic to know the original requester:
# - If we trust node[instance_info][project_id], we are susceptible to a
# node.owner stealing another project's private image by lying in
# instance_info.
# - As of 2025.1, the project_id attached to the auth context at this
# point is more likely to be the nova-computes service user rather
# than the original requester. This is a missing feature from the
# Ironic/Nova virt driver.
auth_token = getattr(context, 'auth_token', None) auth_token = getattr(context, 'auth_token', None)
conductor_project_id = get_conductor_project_id()
image_visibility = getattr(image, 'visibility', None) image_visibility = getattr(image, 'visibility', None)
image_owner = getattr(image, 'owner', None) image_owner = getattr(image, 'owner', None)
image_id = getattr(image, 'id', 'unknown') image_id = getattr(image, 'id', 'unknown')
is_admin = 'admin' in getattr(context, 'roles', []) is_admin = 'admin' in getattr(context, 'roles', [])
project_id = getattr(context, 'project_id', None)
project = getattr(context, 'project', 'unknown') project = getattr(context, 'project', 'unknown')
# The presence of an auth token implies this is an authenticated # If an auth token is present and the config allows access via auth token,
# request and we need not handle the noauth use-case. # allow image access.
if auth_token: if CONF.allow_image_access_via_auth_token and auth_token:
# We return true here since we want the *user* request context to # We return true here since we want the *user* request context to
# be able to be used. # be able to be used.
return True return True
# If the image visibility is public or community, allow access.
if image_visibility == 'public': if image_visibility in ['public', 'community']:
return True return True
# If the user is an admin and the config allows ignoring project checks for
if project_id and image_owner == project_id: # admin tasks, allow access.
return True
if is_admin and CONF.ignore_project_check_for_admin_tasks: if is_admin and CONF.ignore_project_check_for_admin_tasks:
return True return True
# If the image is private and the owner is the conductor project,
# allow access.
if image_visibility == 'private' and image_owner == conductor_project_id:
return True
LOG.info( LOG.info(
'Access to %s owned by %s denied to requester %s', 'Access to %s owned by %s denied to requester %s',
@ -167,3 +181,20 @@ def is_glance_image(image_href):
return False return False
return (image_href.startswith('glance://') return (image_href.startswith('glance://')
or uuidutils.is_uuid_like(image_href)) or uuidutils.is_uuid_like(image_href))
def get_conductor_project_id():
global _GLANCE_SESSION
if not _GLANCE_SESSION:
_GLANCE_SESSION = keystone.get_session('glance')
session = _GLANCE_SESSION
service_auth = keystone.get_auth('glance')
try:
if service_auth and hasattr(service_auth, 'get_project_id'):
return service_auth.get_project_id(session)
elif hasattr(session, 'get_project_id') and session.auth:
return session.get_project_id()
except Exception as e:
LOG.debug("Error getting conductor project ID: %s", str(e))
return None

View File

@ -69,6 +69,12 @@ api_opts = [
default='/etc/ironic/htpasswd', default='/etc/ironic/htpasswd',
help=_('Path to Apache format user authentication file used ' help=_('Path to Apache format user authentication file used '
'when auth_strategy=http_basic')), 'when auth_strategy=http_basic')),
cfg.BoolOpt('allow_image_access_via_auth_token',
default=False,
deprecated_for_removal=True,
help=_('If True, Ironic allows access to Glance images if an '
'auth_token is present in the request context.')
),
cfg.BoolOpt( cfg.BoolOpt(
'ignore_project_check_for_admin_tasks', 'ignore_project_check_for_admin_tasks',
default=True, default=True,

View File

@ -102,7 +102,8 @@ class TestGlanceImageService(base.TestCase):
fixture = {'name': None, fixture = {'name': None,
'owner': None, 'owner': None,
'properties': {}, 'properties': {},
'status': "active"} 'status': "active",
'visibility': "public"}
fixture.update(kwargs) fixture.update(kwargs)
return openstack.image.v2.image.Image.new(**fixture) return openstack.image.v2.image.Image.new(**fixture)
@ -146,7 +147,7 @@ class TestGlanceImageService(base.TestCase):
'status': "active", 'status': "active",
'tags': [], 'tags': [],
'updated_at': None, 'updated_at': None,
'visibility': None, 'visibility': "public",
'os_hash_algo': None, 'os_hash_algo': None,
'os_hash_value': None, 'os_hash_value': None,
} }
@ -188,6 +189,7 @@ class TestGlanceImageService(base.TestCase):
class MyGlanceStubClient(stubs.StubGlanceClient): class MyGlanceStubClient(stubs.StubGlanceClient):
"""A client that fails the first time, then succeeds.""" """A client that fails the first time, then succeeds."""
def get_image(self, image_id): def get_image(self, image_id):
if tries[0] == 0: if tries[0] == 0:
tries[0] = 1 tries[0] = 1
@ -229,10 +231,12 @@ class TestGlanceImageService(base.TestCase):
'image contains no data', 'image contains no data',
self.service.download, image_id) self.service.download, image_id)
@mock.patch.object(service_utils, '_GLANCE_SESSION', autospec=True)
@mock.patch('os.sendfile', autospec=True) @mock.patch('os.sendfile', autospec=True)
@mock.patch('os.path.getsize', autospec=True) @mock.patch('os.path.getsize', autospec=True)
@mock.patch('%s.open' % __name__, new=mock.mock_open(), create=True) @mock.patch('%s.open' % __name__, new=mock.mock_open(), create=True)
def test_download_file_url(self, mock_getsize, mock_sendfile): def test_download_file_url(self, mock_getsize, mock_sendfile,
mock_serviceutils_glance):
# NOTE: only in v2 API # NOTE: only in v2 API
class MyGlanceStubClient(stubs.StubGlanceClient): class MyGlanceStubClient(stubs.StubGlanceClient):
@ -241,8 +245,9 @@ class TestGlanceImageService(base.TestCase):
s_tmpfname = '/whatever/source' s_tmpfname = '/whatever/source'
def get_image(self, image_id): def get_image(self, image_id):
direct_url = "file://%s" + self.s_tmpfname
return type('GlanceTestDirectUrlMeta', (object,), return type('GlanceTestDirectUrlMeta', (object,),
{'direct_url': 'file://%s' + self.s_tmpfname}) dict(visibility='public', direct_url=direct_url))
stub_context = context.RequestContext(auth_token=True) stub_context = context.RequestContext(auth_token=True)
stub_context.user_id = 'fake' stub_context.user_id = 'fake'
@ -251,6 +256,7 @@ class TestGlanceImageService(base.TestCase):
stub_service = image_service.GlanceImageService(stub_client, stub_service = image_service.GlanceImageService(stub_client,
context=stub_context) context=stub_context)
mock_serviceutils_glance.return_value = stub_service
image_id = uuidutils.generate_uuid() image_id = uuidutils.generate_uuid()
self.config(allowed_direct_url_schemes=['file'], group='glance') self.config(allowed_direct_url_schemes=['file'], group='glance')
@ -278,6 +284,7 @@ class TestGlanceImageService(base.TestCase):
def test_client_forbidden_converts_to_imagenotauthed(self): def test_client_forbidden_converts_to_imagenotauthed(self):
class MyGlanceStubClient(stubs.StubGlanceClient): class MyGlanceStubClient(stubs.StubGlanceClient):
"""A client that raises a Forbidden exception.""" """A client that raises a Forbidden exception."""
def get_image(self, image_id): def get_image(self, image_id):
raise openstack_exc.ForbiddenException() raise openstack_exc.ForbiddenException()
@ -295,6 +302,7 @@ class TestGlanceImageService(base.TestCase):
def test_client_notfound_converts_to_imagenotfound(self): def test_client_notfound_converts_to_imagenotfound(self):
class MyGlanceStubClient(stubs.StubGlanceClient): class MyGlanceStubClient(stubs.StubGlanceClient):
"""A client that raises a NotFound exception.""" """A client that raises a NotFound exception."""
def get_image(self, image_id): def get_image(self, image_id):
raise openstack_exc.NotFoundException() raise openstack_exc.NotFoundException()
@ -995,3 +1003,55 @@ class TestServiceUtils(base.TestCase):
self.assertFalse(service_utils.is_glance_image(image_href)) self.assertFalse(service_utils.is_glance_image(image_href))
image_href = None image_href = None
self.assertFalse(service_utils.is_glance_image(image_href)) self.assertFalse(service_utils.is_glance_image(image_href))
class TestIsImageAvailable(base.TestCase):
def setUp(self):
super(TestIsImageAvailable, self).setUp()
self.image = mock.Mock()
self.context = context.RequestContext()
self.context.roles = []
def test_allow_access_via_auth_token_enabled(self):
self.context.auth_token = 'fake-token'
self.config(allow_image_access_via_auth_token=True)
self.assertTrue(service_utils.is_image_available(
self.context, self.image))
def test_allow_public_image(self):
self.image.visibility = 'public'
self.assertTrue(service_utils.is_image_available(
self.context, self.image))
def test_allow_community_image(self):
self.image.visibility = 'community'
self.assertTrue(service_utils.is_image_available(
self.context, self.image))
def test_allow_admin_if_config_enabled(self):
self.context.roles = ['admin']
self.config(ignore_project_check_for_admin_tasks=True)
self.assertTrue(service_utils.is_image_available(
self.context, self.image))
def test_allow_private_image_owned_by_conductor(self):
self.image.visibility = 'private'
self.image.owner = service_utils.get_conductor_project_id()
self.assertTrue(service_utils.is_image_available(
self.context, self.image))
def test_deny_private_image_different_owner(self):
self.config(allow_image_access_via_auth_token=False)
self.config(ignore_project_check_for_admin_tasks=False)
self.image.visibility = 'private'
self.image.owner = 'other-owner'
self.image.id = 'fake-id'
self.context.project = 'test-project'
self.context.roles = []
self.context.auth_token = None
result = service_utils.is_image_available(self.context, self.image)
self.assertFalse(result)

View File

@ -0,0 +1,14 @@
---
features:
- |
If `allow_image_access_via_auth_token` is set to `True`, Ironic allows access to
Glance images if an auth_token is present in the request context.
upgrade:
- |
CONF.allow_image_access_via_auth_token is set to `True` in this Ironic release.
OpenStack integrated operators should ensure images for Ironic use are using image
visibility "public" or "community" for the most reliable results.
deprecation:
- |
CONF.allow_image_access_via_auth_token is deprecated, and will be removed,
along with legacy image access logic, in or after the OpenStack 2026.2 release.