Adopt Detached VMware Root VMDKs as FCD Volumes

cinder/context.py

@@ -261,6 +261,7 @@ def to_policy_values(self) -> dict:
         policy = super(RequestContext, self).to_policy_values()
 
         policy['is_admin'] = self.is_admin
+        policy['service_roles'] = self.service_roles
 
         return policy
 

cinder/policies/base.py

@@ -18,6 +18,8 @@
 from oslo_log import versionutils
 from oslo_policy import policy
 
+from cinder import service_token
+
 # General observations
 # --------------------
 # - This file uses the three "default roles" provided by Keystone during
@@ -145,6 +147,7 @@
 # as RULE_ADMIN_API in Xena and won't be deprecated until Yoga development.
 # XENA_SYSTEM_ADMIN_ONLY = "rule:xena_system_admin_only"
 RULE_ADMIN_API = "rule:admin_api"
+RULE_ADMIN_OR_SERVICE_API = "rule:admin_api or is_service_request:True"
 
 # TODO: xena rules to be removed in AA
 xena_rule_defaults = [
@@ -279,6 +282,29 @@ def __init__(self,
         )
 
 
+@policy.register('is_service_request')
+class IsServiceRequestCheck(policy.Check):
+    """oslo.policy check: `is_service_request:True`.
+
+    Evaluates to True when the request context carries a service token
+    with at least one role listed in
+    `[keystone_authtoken]/service_token_roles`.
+
+    Usage in policy rules::
+
+        "volume_extension:volume_manage":
+            "rule:admin_api or is_service_request:True"
+    """
+
+    def __init__(self, kind, match):
+        self.expected = match.lower() == 'true'
+        super(IsServiceRequestCheck, self).__init__(kind, str(self.expected))
+
+    def __call__(self, target, creds, enforcer):
+        service_roles = creds.get('service_roles') or []
+        return service_token.is_service_request(service_roles) == self.expected
+
+
 # This is used by the deprecated rules in the individual policy files
 # in Xena.
 # TODO: remove in Yoga

cinder/policies/manageable_volumes.py

@@ -41,7 +41,7 @@
         ]),
     policy.DocumentedRuleDefault(
         name=MANAGE_POLICY,
-        check_str=base.RULE_ADMIN_API,
+        check_str=base.RULE_ADMIN_OR_SERVICE_API,
         description="Manage existing volumes.",
         operations=[
             {

cinder/service_token.py

@@ -0,0 +1,47 @@
+# Copyright (c) 2026 SAP SE
+#
+# 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.
+
+"""Utilities for verifying incoming service tokens.
+
+This module handles the *incoming* side of service-token authentication:
+checking whether a request context carries a service token whose roles match
+the operator-configured `[keystone_authtoken]/service_token_roles` list.
+
+See also `cinder/service_auth.py` for the *outgoing* side (attaching a
+service token when Cinder calls other services).
+"""
+
+from oslo_config import cfg
+
+CONF = cfg.CONF
+
+
+def is_service_request(service_roles):
+    """Return True if *service_roles* satisfy the configured service roles.
+
+    A request counts as a service request when it carries a secondary service
+    token (populated into `RequestContext.service_roles` by Keystone
+    middleware) and at least one of those roles is listed in
+    `[keystone_authtoken]/service_token_roles`.
+
+    :param service_roles: list of role strings from the service token
+        (`RequestContext.service_roles`).
+    :returns: True when the request carries an accepted service token.
+    """
+    if not service_roles:
+        return False
+    configured = set(CONF.keystone_authtoken.service_token_roles)
+    return bool(configured.intersection(service_roles))

cinder/tests/unit/api/contrib/test_volume_manage.py

@@ -640,3 +640,127 @@ def test_manage_with_default_type(self,
                          called_with['volume_type']['name'])
         self.assertEqual(default_vt['id'],
                          called_with['volume_type']['id'])
+
+    @mock.patch('cinder.volume.api.API.manage_existing', wraps=api_manage)
+    def test_manage_volume_service_token_bypasses_policy(self,
+                                                         mock_api_manage):
+        """A valid service token allows non-admin users to manage volumes.
+
+        When another service (e.g. Nova) calls manage_existing on behalf of a
+        user, the request carries a service token.  The manage policy
+        (admin-only) should be bypassed in this case, following the same
+        pattern as attachment_deletion_allowed.
+        """
+        ctxt = context.RequestContext(fake.USER_ID, fake.PROJECT_ID,
+                                      is_admin=False)
+        ctxt.service_roles = ['service']
+
+        body = {'volume': {'host': 'host_ok',
+                           'ref': 'fake_ref'}}
+        req = webob.Request.blank('/v3/%s/os-volume-manage' % fake.PROJECT_ID)
+        req.method = 'POST'
+        req.headers['Content-Type'] = 'application/json'
+        req.environ['cinder.context'] = ctxt
+        req.headers["OpenStack-API-Version"] = "volume 3.11"
+        req.api_version_request = api_version.APIVersionRequest('3.11')
+        req.body = jsonutils.dump_as_bytes(body)
+        res = req.get_response(app())
+        self.assertEqual(HTTPStatus.ACCEPTED, res.status_int)
+        self.assertEqual(1, mock_api_manage.call_count)
+
+    def test_manage_volume_non_admin_without_service_token_rejected(self):
+        """Non-admin users without a service token are still rejected.
+
+        The admin-only policy must still be enforced when there is no service
+        token on the request.
+        """
+        ctxt = context.RequestContext(fake.USER_ID, fake.PROJECT_ID,
+                                      is_admin=False)
+        # No service_roles set (or empty)
+        ctxt.service_roles = []
+
+        body = {'volume': {'host': 'host_ok',
+                           'ref': 'fake_ref'}}
+        req = webob.Request.blank('/v3/%s/os-volume-manage' % fake.PROJECT_ID)
+        req.method = 'POST'
+        req.headers['Content-Type'] = 'application/json'
+        req.environ['cinder.context'] = ctxt
+        req.headers["OpenStack-API-Version"] = "volume 3.11"
+        req.api_version_request = api_version.APIVersionRequest('3.11')
+        req.body = jsonutils.dump_as_bytes(body)
+        res = req.get_response(app())
+        self.assertEqual(HTTPStatus.FORBIDDEN, res.status_int)
+
+    @mock.patch('cinder.volume.api.API.manage_existing', wraps=api_manage)
+    def test_manage_volume_admin_without_service_token_still_works(
+            self, mock_api_manage):
+        """Admin users can still manage volumes without a service token.
+
+        The existing admin path must remain unchanged.
+        """
+        ctxt = context.RequestContext(fake.USER_ID, fake.PROJECT_ID,
+                                      is_admin=True)
+
+        body = {'volume': {'host': 'host_ok',
+                           'ref': 'fake_ref'}}
+        req = webob.Request.blank('/v3/%s/os-volume-manage' % fake.PROJECT_ID)
+        req.method = 'POST'
+        req.headers['Content-Type'] = 'application/json'
+        req.environ['cinder.context'] = ctxt
+        req.headers["OpenStack-API-Version"] = "volume 3.11"
+        req.api_version_request = api_version.APIVersionRequest('3.11')
+        req.body = jsonutils.dump_as_bytes(body)
+        res = req.get_response(app())
+        self.assertEqual(HTTPStatus.ACCEPTED, res.status_int)
+        self.assertEqual(1, mock_api_manage.call_count)
+
+    @mock.patch('cinder.volume.api.API.manage_existing', wraps=api_manage)
+    def test_manage_volume_custom_service_role(self, mock_api_manage):
+        """Custom service_token_roles config is respected.
+
+        If the operator configures a custom role name for service tokens,
+        the manage API should accept requests with that role.
+        """
+        self.override_config('service_token_roles',
+                             ['service', 'nova_service'],
+                             group='keystone_authtoken')
+
+        ctxt = context.RequestContext(fake.USER_ID, fake.PROJECT_ID,
+                                      is_admin=False)
+        ctxt.service_roles = ['nova_service']
+
+        body = {'volume': {'host': 'host_ok',
+                           'ref': 'fake_ref'}}
+        req = webob.Request.blank('/v3/%s/os-volume-manage' % fake.PROJECT_ID)
+        req.method = 'POST'
+        req.headers['Content-Type'] = 'application/json'
+        req.environ['cinder.context'] = ctxt
+        req.headers["OpenStack-API-Version"] = "volume 3.11"
+        req.api_version_request = api_version.APIVersionRequest('3.11')
+        req.body = jsonutils.dump_as_bytes(body)
+        res = req.get_response(app())
+        self.assertEqual(HTTPStatus.ACCEPTED, res.status_int)
+        self.assertEqual(1, mock_api_manage.call_count)
+
+    def test_manage_volume_unconfigured_service_role_rejected(self):
+        """A service token with an unconfigured role is still rejected.
+
+        Only roles listed in [keystone_authtoken]/service_token_roles
+        should be accepted.  An arbitrary role on the service token does
+        not grant manage access.
+        """
+        ctxt = context.RequestContext(fake.USER_ID, fake.PROJECT_ID,
+                                      is_admin=False)
+        ctxt.service_roles = ['not_a_configured_service_role']
+
+        body = {'volume': {'host': 'host_ok',
+                           'ref': 'fake_ref'}}
+        req = webob.Request.blank('/v3/%s/os-volume-manage' % fake.PROJECT_ID)
+        req.method = 'POST'
+        req.headers['Content-Type'] = 'application/json'
+        req.environ['cinder.context'] = ctxt
+        req.headers["OpenStack-API-Version"] = "volume 3.11"
+        req.api_version_request = api_version.APIVersionRequest('3.11')
+        req.body = jsonutils.dump_as_bytes(body)
+        res = req.get_response(app())
+        self.assertEqual(HTTPStatus.FORBIDDEN, res.status_int)

cinder/tests/unit/test_context.py

@@ -133,6 +133,18 @@ def test_request_context_with_roles(self):
                                       roles=roles)
         self.assertEqual(roles, ctxt.roles)
 
+    def test_to_policy_values_includes_service_roles(self):
+        ctxt = context.RequestContext('111', '222')
+        ctxt.service_roles = ['service']
+        values = ctxt.to_policy_values()
+        self.assertEqual(['service'], values['service_roles'])
+
+    def test_to_policy_values_empty_service_roles(self):
+        ctxt = context.RequestContext('111', '222')
+        ctxt.service_roles = []
+        values = ctxt.to_policy_values()
+        self.assertEqual([], values['service_roles'])
+
 
 @ddt.ddt
 class ContextAuthorizeTestCase(test.TestCase):