diff --git a/nova/api/openstack/common.py b/nova/api/openstack/common.py index 7ca7688dacf..64af8ebe59e 100644 --- a/nova/api/openstack/common.py +++ b/nova/api/openstack/common.py @@ -30,6 +30,7 @@ from nova import exception from nova.i18n import _ from nova import objects +from nova.objects import fields from nova import quota from nova import utils @@ -499,9 +500,14 @@ def raise_feature_not_supported(msg=None): raise webob.exc.HTTPNotImplemented(explanation=msg) -def get_flavor(context, flavor_id): +def get_flavor(context, flavor_id, + domain_permission=fields.FlavorPermissionRuleEffect.ALLOW, + project_permission=fields.FlavorPermissionRuleEffect.ALLOW): try: - return objects.Flavor.get_by_flavor_id(context, flavor_id) + return objects.Flavor.get_by_flavor_id( + context, flavor_id, + domain_permission=domain_permission, + project_permission=project_permission) except exception.FlavorNotFound as error: raise exc.HTTPNotFound(explanation=error.format_message()) diff --git a/nova/api/openstack/compute/flavor_permission_rules.py b/nova/api/openstack/compute/flavor_permission_rules.py new file mode 100644 index 00000000000..7a60f3a267a --- /dev/null +++ b/nova/api/openstack/compute/flavor_permission_rules.py @@ -0,0 +1,251 @@ +# 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. + +from __future__ import annotations + +import typing as ty + +import webob.exc + +from oslo_utils import strutils +from oslo_utils import uuidutils + +from nova.api.openstack import common +from nova.api.openstack.compute.schemas import ( + flavor_permission_rules as schema) +from nova.api.openstack.compute.views import ( + flavor_permission_rules as views_fpr) +from nova.api.openstack import wsgi +from nova.api import validation +from nova import exception +from nova import objects +from nova.objects import fields +from nova.policies import flavor_permission_rules as fpr_policies + +if ty.TYPE_CHECKING: + from nova.objects.flavor import Flavor + from nova.objects.flavor_permission_rule import FlavorPermissionRule + + +class FlavorPermissionRulesController(wsgi.Controller): + """Controller for flavor permission rules.""" + + _view_builder_class = views_fpr.ViewBuilder + + @staticmethod + def _policy_target( + domain_id: str, + project_id: str | None, + ) -> dict[str, str]: + """Build a policy target dict for a flavor permission rule. + + Allows to restrict rule access to either the context project or the + context project domain, depending on the rule's scope. + """ + target = {'project_domain_id': domain_id} + if project_id is not None: + target['project_id'] = project_id + return target + + @staticmethod + def _resolve_flavor(context, flavor_ref: str) -> Flavor: + """Resolve a flavor reference ignoring permission rules.""" + return common.get_flavor( + context, flavor_ref, + domain_permission=None, + project_permission=None) + + @staticmethod + def _flavor_refs( + context, + rules: ty.Iterable[FlavorPermissionRule], + ) -> dict[int, str]: + """Map internal flavor ids referenced by rules to public flavorids.""" + flavor_ids = {r.flavor_id for r in rules if r.flavor_id is not None} + if not flavor_ids: + return {} + return objects.FlavorList.get_flavor_ids_by_ids( + context, flavor_ids, + domain_permission=None, + project_permission=None) + + @wsgi.expected_errors((400, 403, 404)) + @validation.query_schema(schema.index_query) + @validation.response_body_schema(schema.index_response) + def index(self, req: wsgi.Request) -> dict[str, ty.Any]: + context = req.environ['nova.context'] + root = fpr_policies.POLICY_ROOT + access_all = context.can( + root % 'index:all', target={}, fatal=False) + access_domain = access_all or context.can( + root % 'index:domain', target={}, fatal=False) + # Verify access to flavor permission rules + access_domain or context.can(root % 'index:project', target={}) + + filter_by_context_project = not access_domain + project_id = req.GET.get('project_id') + scope = req.GET.get('scope') + if not access_domain: + if scope == fields.FlavorPermissionRuleScope.DOMAIN: + raise webob.exc.HTTPForbidden( + explanation="Policy doesn't allow filtering by domain " + "scope") + scope = fields.FlavorPermissionRuleScope.PROJECT + + filter_by_context_domain = not access_all + domain_id = req.GET.get('domain_id') + if domain_id and not access_all: + raise webob.exc.HTTPForbidden( + explanation="Policy doesn't allow filtering by domain_id") + + limit, marker = common.get_limit_and_marker(req) + flavor_id = None + flavor_ref = req.GET.get('flavor_id') + if flavor_ref is not None: + flavor = self._resolve_flavor(context, flavor_ref) + flavor_id = flavor.id + + has_flavor_str = req.GET.get('has_flavor') + has_flavor = (strutils.bool_from_string(has_flavor_str, strict=True) + if has_flavor_str is not None else None) + + if flavor_id is not None and has_flavor is not None: + raise webob.exc.HTTPBadRequest( + explanation="'flavor_id' and 'has_flavor' are mutually " + "exclusive") + + try: + rules = objects.FlavorPermissionRuleList.get_all( + context, + filter_by_context_domain=filter_by_context_domain, + filter_by_context_project=filter_by_context_project, + domain_id=domain_id, project_id=project_id, scope=scope, + effect=req.GET.get('effect'), flavor_id=flavor_id, + has_flavor=has_flavor, limit=limit, marker=marker) + except exception.MarkerNotFound as e: + raise webob.exc.HTTPBadRequest(explanation=e.format_message()) + + flavor_refs = self._flavor_refs(context, rules) + return self._view_builder.index(req, rules, flavor_refs) + + @wsgi.expected_errors((403, 404)) + @validation.response_body_schema(schema.create_show_update_response) + def show(self, req: wsgi.Request, id: str) -> dict[str, ty.Any]: + context = req.environ['nova.context'] + try: + rule = objects.FlavorPermissionRule.get_by_uuid(context, id) + except exception.FlavorPermissionRuleNotFound as e: + raise webob.exc.HTTPNotFound(explanation=e.format_message()) + + if not context.can( + fpr_policies.POLICY_ROOT % ('show:%s' % rule.scope), + target=self._policy_target(rule.domain_id, rule.project_id), + fatal=False): + # Return Not Found rather than Forbidden to avoid leaking the + # rule's existence to callers who don't have access to it. + raise webob.exc.HTTPNotFound() + + flavor_ref = self._flavor_refs(context, [rule]).get(rule.flavor_id) + return self._view_builder.show(req, rule, flavor_ref) + + @wsgi.response(201) + @wsgi.expected_errors((400, 403, 404, 409)) + @validation.schema(schema.create) + @validation.response_body_schema(schema.create_show_update_response) + def create( + self, + req: wsgi.Request, + body: dict[str, ty.Any], + ) -> dict[str, ty.Any]: + context = req.environ['nova.context'] + data = body['flavor_permission_rule'] + domain_id = data['domain_id'] + project_id = data.get('project_id') + scope = (fields.FlavorPermissionRuleScope.PROJECT + if project_id else fields.FlavorPermissionRuleScope.DOMAIN) + + context.can( + fpr_policies.POLICY_ROOT % ('create:%s' % scope), + target=self._policy_target(domain_id, project_id)) + + flavor_id = None + flavor_ref = None + if 'flavor_id' in data and data['flavor_id'] is not None: + flavor = self._resolve_flavor(context, str(data['flavor_id'])) + if not flavor.is_public: + raise webob.exc.HTTPBadRequest( + explanation="Flavor permission rules only apply to public " + "flavors.") + flavor_id = flavor.id + flavor_ref = flavor.flavorid + + rule = objects.FlavorPermissionRule( + context=context, uuid=uuidutils.generate_uuid(), + domain_id=domain_id, project_id=project_id, flavor_id=flavor_id, + effect=data['effect']) + try: + rule.create() + except exception.FlavorPermissionRuleExists as e: + raise webob.exc.HTTPConflict(explanation=e.format_message()) + + return self._view_builder.show(req, rule, flavor_ref) + + @wsgi.response(204) + @wsgi.expected_errors(404) + @validation.response_body_schema(schema.delete_response) + def delete(self, req: wsgi.Request, id: str) -> None: + context = req.environ['nova.context'] + try: + rule = objects.FlavorPermissionRule.get_by_uuid(context, id) + except exception.FlavorPermissionRuleNotFound as e: + raise webob.exc.HTTPNotFound(explanation=e.format_message()) + + if not context.can( + fpr_policies.POLICY_ROOT % ('delete:%s' % rule.scope), + target=self._policy_target(rule.domain_id, rule.project_id), + fatal=False): + # Return Not Found rather than Forbidden to avoid leaking the + # rule's existence to callers who don't have access to it. + raise webob.exc.HTTPNotFound() + + rule.destroy() + + @wsgi.expected_errors((400, 404)) + @validation.schema(schema.update) + @validation.response_body_schema(schema.create_show_update_response) + def update( + self, + req: wsgi.Request, + id: str, + body: dict[str, ty.Any], + ) -> dict[str, ty.Any]: + context = req.environ['nova.context'] + try: + rule = objects.FlavorPermissionRule.get_by_uuid(context, id) + except exception.FlavorPermissionRuleNotFound as e: + raise webob.exc.HTTPNotFound(explanation=e.format_message()) + + if not context.can( + fpr_policies.POLICY_ROOT % ('update:%s' % rule.scope), + target=self._policy_target(rule.domain_id, rule.project_id), + fatal=False): + # Return Not Found rather than Forbidden to avoid leaking the + # rule's existence to callers who don't have access to it. + raise webob.exc.HTTPNotFound() + + rule.effect = body['flavor_permission_rule']['effect'] + rule.save() + flavor_ref = self._flavor_refs(context, [rule]).get(rule.flavor_id) + return self._view_builder.show(req, rule, flavor_ref) diff --git a/nova/api/openstack/compute/flavors.py b/nova/api/openstack/compute/flavors.py index be950240d9b..3d53e226f8d 100644 --- a/nova/api/openstack/compute/flavors.py +++ b/nova/api/openstack/compute/flavors.py @@ -26,8 +26,10 @@ from nova import exception from nova.i18n import _ from nova import objects +from nova.objects import fields from nova.policies import flavor_extra_specs as fes_policies from nova.policies import flavor_manage as fm_policies +from nova.policies import flavor_permission_rules as fpr_policies from nova import utils @@ -137,7 +139,7 @@ def update(self, req, id, body): return self._view_builder.show(req, flavor, include_description=True, include_extra_specs=include_extra_specs) - @wsgi.expected_errors(400) + @wsgi.expected_errors((400, 403)) @validation.query_schema(schema.index_query, '2.0', '2.74') @validation.query_schema(schema.index_query_275, '2.75') @validation.response_body_schema(schema.index_response, '2.0', '2.54') @@ -147,7 +149,7 @@ def index(self, req): limited_flavors = self._get_flavors(req) return self._view_builder.index(req, limited_flavors) - @wsgi.expected_errors(400) + @wsgi.expected_errors((400, 403)) @validation.query_schema(schema.index_query, '2.0', '2.74') @validation.query_schema(schema.index_query_275, '2.75') @validation.response_body_schema(schema.detail_response, '2.0', '2.54') @@ -162,8 +164,18 @@ def detail(self, req): req, flavors_view.FLAVOR_EXTRA_SPECS_MICROVERSION): include_extra_specs = context.can( fes_policies.POLICY_ROOT % 'index', fatal=False) + include_domain = context.can( + fpr_policies.POLICY_ROOT % 'index:domain', fatal=False) + include_project = include_domain or context.can( + fpr_policies.POLICY_ROOT % 'index:project', fatal=False) + flavor_permissions = None + if include_domain or include_project: + flavor_permissions = limited_flavors.get_permissions( + include_domain=include_domain, + include_project=include_project) return self._view_builder.detail( - req, limited_flavors, include_extra_specs=include_extra_specs) + req, limited_flavors, include_extra_specs=include_extra_specs, + flavor_permissions=flavor_permissions) @wsgi.expected_errors(404) @validation.query_schema(schema.show_query) @@ -174,8 +186,19 @@ def detail(self, req): def show(self, req, id): """Return data about the given flavor id.""" context = req.environ['nova.context'] + include_domain = context.can( + fpr_policies.POLICY_ROOT % 'index:domain', fatal=False) + include_project = include_domain or context.can( + fpr_policies.POLICY_ROOT % 'index:project', fatal=False) try: - flavor = flavors.get_flavor_by_flavor_id(id, ctxt=context) + flavor = flavors.get_flavor_by_flavor_id( + id, ctxt=context, + domain_permission=( + None if include_domain + else fields.FlavorPermissionRuleEffect.ALLOW), + project_permission=( + None if include_project + else fields.FlavorPermissionRuleEffect.ALLOW)) except exception.FlavorNotFound as e: raise webob.exc.HTTPNotFound(explanation=e.format_message()) @@ -186,9 +209,15 @@ def show(self, req, id): fes_policies.POLICY_ROOT % 'index', fatal=False) include_description = api_version_request.is_supported( req, flavors_view.FLAVOR_DESCRIPTION_MICROVERSION) + flavor_permission = None + if include_domain or include_project: + flavor_permission = flavor.get_permission( + include_domain=include_domain, + include_project=include_project) return self._view_builder.show( req, flavor, include_description=include_description, - include_extra_specs=include_extra_specs) + include_extra_specs=include_extra_specs, + flavor_permission=flavor_permission) def _parse_is_public(self, is_public): """Parse is_public into something usable.""" @@ -236,10 +265,39 @@ def _get_flavors(self, req): req.params['minDisk']) raise webob.exc.HTTPBadRequest(explanation=msg) + can_filter_domain = context.can( + fpr_policies.POLICY_ROOT % 'index:domain', fatal=False) + domain_permission = req.params.get('domain_permission') + if domain_permission and not can_filter_domain: + raise webob.exc.HTTPForbidden( + explanation="Policy doesn't allow filtering by " + "domain_permission.") + + can_filter_project = can_filter_domain or context.can( + fpr_policies.POLICY_ROOT % 'index:project', fatal=False) + project_permission = req.params.get('project_permission') + if project_permission and not can_filter_project: + raise webob.exc.HTTPForbidden( + explanation="Policy doesn't allow filtering by " + "project_permission.") + + # Force permission filtering for admin users when explicitly requested + force_permission_filter = bool(domain_permission or project_permission) + # Map 'all' to None and default to enforcing flavor permission rules. + domain_permission = ( + None if domain_permission == 'all' + else domain_permission or fields.FlavorPermissionRuleEffect.ALLOW) + project_permission = ( + None if project_permission == 'all' + else project_permission or fields.FlavorPermissionRuleEffect.ALLOW) + try: limited_flavors = objects.FlavorList.get_all( context, filters=filters, sort_key=sort_key, sort_dir=sort_dir, - limit=limit, marker=marker) + limit=limit, marker=marker, + domain_permission=domain_permission, + project_permission=project_permission, + force_permission_filter=force_permission_filter) except exception.MarkerNotFound: msg = _('marker [%s] not found') % marker raise webob.exc.HTTPBadRequest(explanation=msg) diff --git a/nova/api/openstack/compute/routes.py b/nova/api/openstack/compute/routes.py index c47f40847c7..3dce7eb340f 100644 --- a/nova/api/openstack/compute/routes.py +++ b/nova/api/openstack/compute/routes.py @@ -37,6 +37,7 @@ from nova.api.openstack.compute import extension_info from nova.api.openstack.compute import fixed_ips from nova.api.openstack.compute import flavor_access +from nova.api.openstack.compute import flavor_permission_rules from nova.api.openstack.compute import flavors from nova.api.openstack.compute import flavors_extraspecs from nova.api.openstack.compute import floating_ip_dns @@ -157,6 +158,10 @@ def _create_controller(main_controller, action_controller_list): flavors_extraspecs.FlavorExtraSpecsController, []) +flavor_permission_rules_controller = functools.partial(_create_controller, + flavor_permission_rules.FlavorPermissionRulesController, []) + + floating_ip_dns_controller = functools.partial(_create_controller, floating_ip_dns.FloatingIPDNSDomainController, []) @@ -416,6 +421,15 @@ def _create_controller(main_controller, action_controller_list): ('/flavors/{flavor_id}/os-flavor-access', { 'GET': [flavor_access_controller, 'index'] }), + ('/flavor-permission-rules', { + 'GET': [flavor_permission_rules_controller, 'index'], + 'POST': [flavor_permission_rules_controller, 'create'] + }), + ('/flavor-permission-rules/{id}', { + 'GET': [flavor_permission_rules_controller, 'show'], + 'PUT': [flavor_permission_rules_controller, 'update'], + 'DELETE': [flavor_permission_rules_controller, 'delete'] + }), ('/images', { 'GET': [images_controller, 'index'] }), diff --git a/nova/api/openstack/compute/schemas/flavor_permission_rules.py b/nova/api/openstack/compute/schemas/flavor_permission_rules.py new file mode 100644 index 00000000000..a48f1a61cca --- /dev/null +++ b/nova/api/openstack/compute/schemas/flavor_permission_rules.py @@ -0,0 +1,133 @@ +# 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. + +from nova.api.validation import parameter_types +from nova.objects import fields + + +_RULE_EFFECTS = list(fields.FlavorPermissionRuleEffect.ALL) +_RULE_SCOPES = list(fields.FlavorPermissionRuleScope.ALL) + +_links = { + 'type': 'array', + 'items': { + 'type': 'object', + 'properties': { + 'href': {'type': 'string', 'format': 'uri'}, + 'rel': {'type': 'string'}, + }, + 'required': ['href', 'rel'], + 'additionalProperties': False, + }, +} + +_rule = { + 'type': 'object', + 'properties': { + 'id': {'type': 'string'}, + 'domain_id': {'type': 'string'}, + 'project_id': {'type': ['string', 'null']}, + 'flavor_id': {'type': ['string', 'null']}, + 'effect': {'type': 'string', 'enum': _RULE_EFFECTS}, + 'scope': {'type': 'string', 'enum': _RULE_SCOPES}, + 'links': _links, + }, + 'required': ['id', 'domain_id', 'project_id', 'flavor_id', + 'effect', 'scope', 'links'], + 'additionalProperties': False, +} + +create_show_update_response = { + 'type': 'object', + 'properties': { + 'flavor_permission_rule': _rule, + }, + 'required': ['flavor_permission_rule'], + 'additionalProperties': False, +} + +index_response = { + 'type': 'object', + 'properties': { + 'flavor_permission_rules': { + 'type': 'array', + 'items': _rule, + }, + 'flavor_permission_rules_links': _links, + }, + 'required': ['flavor_permission_rules'], + 'additionalProperties': False, +} + +delete_response = { + 'type': 'null', +} + +create = { + 'type': 'object', + 'properties': { + 'flavor_permission_rule': { + 'type': 'object', + 'properties': { + 'domain_id': parameter_types.project_id, + 'project_id': parameter_types.project_id, + 'flavor_id': parameter_types.flavor_ref, + 'effect': {'type': 'string', 'enum': _RULE_EFFECTS}, + }, + 'required': ['domain_id', 'effect'], + 'additionalProperties': False, + }, + }, + 'required': ['flavor_permission_rule'], + 'additionalProperties': False, +} + +update = { + 'type': 'object', + 'properties': { + 'flavor_permission_rule': { + 'type': 'object', + 'properties': { + 'effect': {'type': 'string', 'enum': _RULE_EFFECTS}, + }, + 'required': ['effect'], + 'additionalProperties': False, + }, + }, + 'required': ['flavor_permission_rule'], + 'additionalProperties': False, +} + +index_query = { + 'type': 'object', + 'properties': { + 'limit': parameter_types.multi_params( + parameter_types.non_negative_integer), + 'marker': parameter_types.multi_params({'type': 'string'}), + 'scope': parameter_types.multi_params( + {'type': 'string', 'enum': _RULE_SCOPES}), + 'domain_id': parameter_types.multi_params( + parameter_types.project_id), + 'project_id': parameter_types.multi_params( + parameter_types.project_id), + 'flavor_id': parameter_types.multi_params( + parameter_types.flavor_ref), + 'has_flavor': parameter_types.multi_params( + parameter_types.boolean), + 'effect': parameter_types.multi_params( + {'type': 'string', 'enum': _RULE_EFFECTS}), + }, + 'additionalProperties': False, +} diff --git a/nova/api/openstack/compute/schemas/flavors.py b/nova/api/openstack/compute/schemas/flavors.py index 0246a099d71..f7a448d0d3f 100644 --- a/nova/api/openstack/compute/schemas/flavors.py +++ b/nova/api/openstack/compute/schemas/flavors.py @@ -15,6 +15,7 @@ import copy from nova.api.validation import parameter_types +from nova.objects.fields import FlavorPermissionRuleEffect # NOTE(takashin): The following sort keys are defined for backward # compatibility. If they are changed, the API microversion should be bumped. @@ -26,6 +27,8 @@ VALID_SORT_DIR = ['asc', 'desc'] +VALID_PERMISSION_VALUES = [*FlavorPermissionRuleEffect.ALL, 'all'] + create = { 'type': 'object', 'properties': { @@ -127,7 +130,11 @@ 'sort_key': parameter_types.multi_params({'type': 'string', 'enum': VALID_SORT_KEYS}), 'sort_dir': parameter_types.multi_params({'type': 'string', - 'enum': VALID_SORT_DIR}) + 'enum': VALID_SORT_DIR}), + 'domain_permission': parameter_types.multi_params( + {'type': 'string', 'enum': VALID_PERMISSION_VALUES}), + 'project_permission': parameter_types.multi_params( + {'type': 'string', 'enum': VALID_PERMISSION_VALUES}), }, # NOTE(gmann): This is kept True to keep backward compatibility. # As of now Schema validation stripped out the additional parameters and @@ -201,6 +208,10 @@ 'vcpus': {'type': 'integer'}, 'OS-FLV-EXT-DATA:ephemeral': {'type': 'integer'}, 'OS-FLV-DISABLED:disabled': {'type': 'boolean'}, + 'permissions': { + 'type': 'object', + 'additionalProperties': {'type': 'string'}, + }, }, 'required': [ 'disk', diff --git a/nova/api/openstack/compute/views/flavor_permission_rules.py b/nova/api/openstack/compute/views/flavor_permission_rules.py new file mode 100644 index 00000000000..20c750a4e70 --- /dev/null +++ b/nova/api/openstack/compute/views/flavor_permission_rules.py @@ -0,0 +1,74 @@ +# 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. + +from __future__ import annotations + +import typing as ty + +from nova.api.openstack import common + +if ty.TYPE_CHECKING: + from nova.api.openstack import wsgi + from nova.objects.flavor_permission_rule import FlavorPermissionRule + from nova.objects.flavor_permission_rule import FlavorPermissionRuleList + + +class ViewBuilder(common.ViewBuilder): + + _collection_name = 'flavor_permission_rules' + + def _rule_dict( + self, + request: wsgi.Request, + rule: FlavorPermissionRule, + flavor_ref: str | None, + ) -> dict[str, ty.Any]: + return { + 'id': rule.uuid, + 'domain_id': rule.domain_id, + 'project_id': rule.project_id, + 'flavor_id': flavor_ref, + 'effect': rule.effect, + 'scope': rule.scope, + 'links': self._get_links( + request, rule.uuid, self._collection_name), + } + + def show( + self, + request: wsgi.Request, + rule: FlavorPermissionRule, + flavor_ref: str | None, + ) -> dict[str, ty.Any]: + return { + 'flavor_permission_rule': self._rule_dict( + request, rule, flavor_ref), + } + + def index( + self, + request: wsgi.Request, + rules: FlavorPermissionRuleList, + flavor_refs: dict[int, str], + ) -> dict[str, ty.Any]: + rules_list = [ + self._rule_dict(request, r, flavor_refs.get(r.flavor_id)) + for r in rules] + response = {'flavor_permission_rules': rules_list} + links = self._get_collection_links( + request, rules_list, self._collection_name) + if links: + response['flavor_permission_rules_links'] = links + return response diff --git a/nova/api/openstack/compute/views/flavors.py b/nova/api/openstack/compute/views/flavors.py index 1df4f093ba0..20f8a9fc743 100644 --- a/nova/api/openstack/compute/views/flavors.py +++ b/nova/api/openstack/compute/views/flavors.py @@ -25,7 +25,7 @@ class ViewBuilder(common.ViewBuilder): _collection_name = "flavors" def basic(self, request, flavor, include_description=False, - include_extra_specs=False): + include_extra_specs=False, flavor_permission=None): # include_extra_specs is placeholder param which is not used in # this method as basic() method is used by index() (GET /flavors) # which does not return those keys in response. @@ -42,10 +42,13 @@ def basic(self, request, flavor, include_description=False, if include_description: flavor_dict['flavor']['description'] = flavor.description + if flavor_permission is not None: + flavor_dict['flavor']['permissions'] = flavor_permission + return flavor_dict def show(self, request, flavor, include_description=False, - include_extra_specs=False): + include_extra_specs=False, flavor_permission=None): flavor_dict = { "flavor": { "id": flavor["flavorid"], @@ -73,6 +76,9 @@ def show(self, request, flavor, include_description=False, if api_version_request.is_supported(request, '2.75'): flavor_dict['flavor']['swap'] = flavor["swap"] or 0 + if flavor_permission is not None: + flavor_dict['flavor']['permissions'] = flavor_permission + return flavor_dict def index(self, request, flavors): @@ -83,17 +89,20 @@ def index(self, request, flavors): return self._list_view(self.basic, request, flavors, coll_name, include_description=include_description) - def detail(self, request, flavors, include_extra_specs=False): + def detail(self, request, flavors, include_extra_specs=False, + flavor_permissions=None): """Return the 'detail' view of flavors.""" coll_name = self._collection_name + '/detail' include_description = api_version_request.is_supported( request, FLAVOR_DESCRIPTION_MICROVERSION) return self._list_view(self.show, request, flavors, coll_name, include_description=include_description, - include_extra_specs=include_extra_specs) + include_extra_specs=include_extra_specs, + flavor_permissions=flavor_permissions) def _list_view(self, func, request, flavors, coll_name, - include_description=False, include_extra_specs=False): + include_description=False, include_extra_specs=False, + flavor_permissions=None): """Provide a view for a list of flavors. :param func: Function used to format the flavor data @@ -105,11 +114,17 @@ def _list_view(self, func, request, flavors, coll_name, included in the response dict. :param include_extra_specs: If the flavor.extra_specs should be included in the response dict. + :param flavor_permissions: Dict of flavor id → permission dict, or + None when permissions are not requested. :returns: Flavor reply data in dictionary format """ flavor_list = [func(request, flavor, include_description, - include_extra_specs)["flavor"] + include_extra_specs, + flavor_permission=( + None if flavor_permissions is None + else flavor_permissions.get( + flavor['id'], {})))["flavor"] for flavor in flavors] flavors_links = self._get_collection_links(request, flavors, diff --git a/nova/compute/flavors.py b/nova/compute/flavors.py index c12b0de65ab..3ad356c2807 100644 --- a/nova/compute/flavors.py +++ b/nova/compute/flavors.py @@ -29,6 +29,7 @@ from nova import exception from nova.i18n import _ from nova import objects +from nova.objects import fields from nova import utils CONF = nova.conf.CONF @@ -128,7 +129,11 @@ def create(name, memory, vcpus, root_gb, ephemeral_gb=0, flavorid=None, # TODO(termie): flavor-specific code should probably be in the API that uses # flavors. -def get_flavor_by_flavor_id(flavorid, ctxt=None, read_deleted="yes"): +def get_flavor_by_flavor_id(flavorid, ctxt=None, read_deleted="yes", + domain_permission=( + fields.FlavorPermissionRuleEffect.ALLOW), + project_permission=( + fields.FlavorPermissionRuleEffect.ALLOW)): """Retrieve flavor by flavorid. :raises: FlavorNotFound @@ -136,7 +141,10 @@ def get_flavor_by_flavor_id(flavorid, ctxt=None, read_deleted="yes"): if ctxt is None: ctxt = context.get_admin_context(read_deleted=read_deleted) - return objects.Flavor.get_by_flavor_id(ctxt, flavorid, read_deleted) + return objects.Flavor.get_by_flavor_id( + ctxt, flavorid, read_deleted, + domain_permission=domain_permission, + project_permission=project_permission) # NOTE(danms): This method is deprecated, do not use it! diff --git a/nova/db/api/migrations/versions/f177f53f978b_add_flavor_permission_rules.py b/nova/db/api/migrations/versions/f177f53f978b_add_flavor_permission_rules.py new file mode 100644 index 00000000000..3686f5dfb93 --- /dev/null +++ b/nova/db/api/migrations/versions/f177f53f978b_add_flavor_permission_rules.py @@ -0,0 +1,57 @@ +# 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. + +"""add_flavor_permission_rules + +Revision ID: f177f53f978b +Revises: cdeec0c85668 +Create Date: 2025-10-24 15:43:26.554412 +""" + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = 'f177f53f978b' +down_revision = 'cdeec0c85668' +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + 'flavor_permission_rules', + sa.Column('created_at', sa.DateTime), + sa.Column('updated_at', sa.DateTime), + sa.Column('id', sa.Integer, primary_key=True), + sa.Column('uuid', sa.String(36), nullable=False), + sa.Column('domain_id', sa.String(255), nullable=False), + sa.Column('project_id', sa.String(255), nullable=False, + server_default=''), + sa.Column('flavor_id', sa.Integer, nullable=False, + server_default='-1'), + sa.Column( + 'effect', + sa.Enum('allow', 'deny', name='flavor_permission_rules0effect'), + nullable=False), + sa.UniqueConstraint('uuid', name='uniq_flavor_permission_rules0uuid'), + sa.UniqueConstraint( + 'domain_id', 'project_id', 'flavor_id', + name='uniq_flavor_permission_rules0domain_id0project_id0flavor_id' + ), + sa.Index('flavor_permission_rules_uuid_idx', 'uuid'), + sa.Index('flavor_permission_rules_domain_id_project_id_flavor_id_idx', + 'domain_id', 'project_id', 'flavor_id'), + mysql_engine='InnoDB', + mysql_charset='utf8', + ) diff --git a/nova/db/api/models.py b/nova/db/api/models.py index 5a236e98f16..1f044ac4ca9 100644 --- a/nova/db/api/models.py +++ b/nova/db/api/models.py @@ -281,6 +281,37 @@ class FlavorProjects(BASE): flavor = orm.relationship(Flavors, back_populates='projects') +class FlavorPermissionRule(BASE): + """Represents a flavor permission rule for a project""" + + __tablename__ = 'flavor_permission_rules' + __table_args__ = ( + schema.UniqueConstraint( + 'uuid', name='uniq_flavor_permission_rules0uuid'), + schema.UniqueConstraint( + 'domain_id', 'project_id', 'flavor_id', + name='uniq_flavor_permission_rules0domain_id0project_id0flavor_id' + ), + sa.Index('flavor_permission_rules_uuid_idx', 'uuid'), + sa.Index( + 'flavor_permission_rules_domain_id_project_id_flavor_id_idx', + 'domain_id', 'project_id', 'flavor_id'), + ) + + id = sa.Column(sa.Integer, primary_key=True) + uuid = sa.Column(sa.String(36), nullable=False) + domain_id = sa.Column(sa.String(255), nullable=False) + # Empty string sentinel means domain-scope rule. We do not use NULL to + # properly enforce the unique constraint. + project_id = sa.Column(sa.String(255), nullable=False, default='') + # -1 sentinel means default rule (applies to all flavors). We do not use + # NULL to properly enforce the unique constraint. + flavor_id = sa.Column(sa.Integer, nullable=False, default=-1) + effect = sa.Column( + sa.Enum('allow', 'deny', name='flavor_permission_rules0effect'), + nullable=False) + + class BuildRequest(BASE): """Represents the information passed to the scheduler.""" diff --git a/nova/exception.py b/nova/exception.py index 15a17de4c6e..38b015f298f 100644 --- a/nova/exception.py +++ b/nova/exception.py @@ -1295,6 +1295,21 @@ class FlavorAccessExists(NovaException): "and project %(project_id)s combination.") +class FlavorPermissionRuleExists(NovaException): + msg_fmt = _("Flavor permission rule already exists for uuid %(uuid)s or " + "combination of project %(project_id)s and flavor " + "%(flavor_id)s.") + + +class FlavorPermissionRuleNotFound(NotFound): + msg_fmt = _("Flavor permission rule not found for id %(id)s") + + +class FlavorPermissionRuleNotFoundForProjectFlavor(NotFound): + msg_fmt = _("Flavor permission rule not found for project %(project_id)s " + "and flavor %(flavor_id)s combination.") + + class InvalidSharedStorage(NovaException): msg_fmt = _("%(path)s is not on shared storage: %(reason)s") diff --git a/nova/objects/__init__.py b/nova/objects/__init__.py index 4b51ea3e466..001d2cb4afb 100644 --- a/nova/objects/__init__.py +++ b/nova/objects/__init__.py @@ -34,6 +34,7 @@ def register_all(): __import__('nova.objects.ec2') __import__('nova.objects.external_event') __import__('nova.objects.flavor') + __import__('nova.objects.flavor_permission_rule') __import__('nova.objects.host_mapping') __import__('nova.objects.hv_spec') __import__('nova.objects.image_meta') diff --git a/nova/objects/fields.py b/nova/objects/fields.py index 4b619faa6dc..f0ed0ff7006 100644 --- a/nova/objects/fields.py +++ b/nova/objects/fields.py @@ -1075,6 +1075,20 @@ class InstanceTaskState(BaseNovaEnum): SHELVING_OFFLOADING, UNSHELVING, IN_CLUSTER_VMOTION) +class FlavorPermissionRuleEffect(BaseNovaEnum): + ALLOW = 'allow' + DENY = 'deny' + + ALL = (ALLOW, DENY) + + +class FlavorPermissionRuleScope(BaseNovaEnum): + DOMAIN = 'domain' + PROJECT = 'project' + + ALL = (DOMAIN, PROJECT) + + class InstancePowerState(Enum): _UNUSED = '_unused' NOSTATE = 'pending' @@ -1432,6 +1446,10 @@ class InstanceTaskStateField(BaseEnumField): AUTO_TYPE = InstanceTaskState() +class FlavorPermissionRuleEffectField(BaseEnumField): + AUTO_TYPE = FlavorPermissionRuleEffect() + + class InstancePowerStateField(BaseEnumField): AUTO_TYPE = InstancePowerState() diff --git a/nova/objects/flavor.py b/nova/objects/flavor.py index 2e9e41706ff..cba0d1f153d 100644 --- a/nova/objects/flavor.py +++ b/nova/objects/flavor.py @@ -14,6 +14,7 @@ from copy import deepcopy from operator import itemgetter +import typing as ty from oslo_db import exception as db_exc from oslo_utils import versionutils @@ -32,6 +33,8 @@ from nova import objects from nova.objects import base from nova.objects import fields +from nova.objects.flavor_permission_rule import ( + DB_NONE_SENTINELS as _FPR_SENTINELS) from nova import utils @@ -44,6 +47,14 @@ CONF = nova.conf.CONF +# Define a constant for the 'allow' effect because: +# - we cannot directly use 'fields.FlavorPermissionRuleEffect' for method +# default parameters since it is shadowed by the Flavor.fields dict +# - we do not want to rename the existing 'fields' import +# For consistency we also add a 'deny' constant +_PERM_ALLOW = fields.FlavorPermissionRuleEffect.ALLOW +_PERM_DENY = fields.FlavorPermissionRuleEffect.DENY + def _dict_with_extra_specs(flavor_model): extra_specs = {x['key']: x['value'] @@ -191,6 +202,8 @@ def _flavor_destroy(context, flavor_id=None, flavorid=None): filter_by(flavor_id=result.id).delete() context.session.query(api_models.FlavorExtraSpecs).\ filter_by(flavor_id=result.id).delete() + context.session.query(api_models.FlavorPermissionRule).\ + filter_by(flavor_id=result.id).delete() context.session.delete(result) return result @@ -308,9 +321,87 @@ def _from_db_object(context, flavor, db_flavor, expected_attrs=None): return flavor @staticmethod - def _flavor_get_query_from_db(context): + def _flavor_query_allowed_clause( + project_id: str, + scope: str, + ) -> sa.sql.ColumnElement: + """Return a SQL column element that is true when the flavor is allowed. + + :param project_id: Must be a domain ID for scope DOMAIN and a + non-domain project ID for scope PROJECT. + + A flavor is NOT allowed for a project_id at the given scope if: + - There is a 'deny' rule matching the flavor_id, OR + - There is a 'deny' rule with flavor_id=-1 (default deny) AND there + is no 'allow' rule matching the flavor_id. + """ + Rule = api_models.FlavorPermissionRule + if scope == fields.FlavorPermissionRuleScope.DOMAIN: + scope_filter = sa.and_( + Rule.domain_id == project_id, + Rule.project_id == _FPR_SENTINELS['project_id']) + else: + scope_filter = Rule.project_id == project_id + specific_deny = sa.exists().where(sa.and_( + scope_filter, + Rule.effect == _PERM_DENY, + Rule.flavor_id == api_models.Flavors.id, + )) + default_deny = sa.exists().where(sa.and_( + scope_filter, + Rule.effect == _PERM_DENY, + Rule.flavor_id == _FPR_SENTINELS['flavor_id'], + )) + specific_allow = sa.exists().where(sa.and_( + scope_filter, + Rule.effect == _PERM_ALLOW, + Rule.flavor_id == api_models.Flavors.id, + )) + return ~sa.or_(specific_deny, sa.and_(default_deny, ~specific_allow)) + + @staticmethod + def _flavor_query_permission_filter( + query: orm.Query, + project_id: str, + scope: str, + permission: str | None, + ) -> orm.Query: + """Filter a flavor query by flavor permission for one scope. + + None bypasses all filtering by flavor permission rules. ALLOW enforces + permission rules by returning only non-public flavors and public + flavors allowed by flavor permission rules. DENY returns only public + flavors denied by flavor permission rules. + """ + if permission is None: + return query + allowed = Flavor._flavor_query_allowed_clause( + project_id, scope) + if permission == _PERM_DENY: + return query.filter( + api_models.Flavors.is_public == sql.true(), ~allowed) + return query.filter( + sa.or_(api_models.Flavors.is_public == sql.false(), allowed)) + + @staticmethod + def _flavor_get_query_from_db( + context, + domain_permission: str | None = _PERM_ALLOW, + project_permission: str | None = _PERM_ALLOW, + force_permission_filter: bool = False, + ) -> orm.Query: # We don't use a database context decorator on this method because this # method is not executing a query, it's only building one. + # + # Permission filtering is controlled via: + # - domain_permission: ALLOW/DENY only returns flavors allowed/denied + # at domain scope. None bypasses domain scope filtering. + # - project_permission: ALLOW/DENY only returns flavors allowed/denied + # at project scope. None bypasses project scope filtering. + # - force_permission_filter: If True, apply permission filtering + # also for admin contexts. + # + # The default values enforce permission rules for non-admin users. query = context.session.query(api_models.Flavors).options( orm.joinedload(api_models.Flavors.extra_specs) ) @@ -320,6 +411,18 @@ def _flavor_get_query_from_db(context): api_models.Flavors.projects.any(project_id=context.project_id) ]) query = query.filter(sa.or_(*the_filter)) + if not context.is_admin or force_permission_filter: + if context.project_domain_id: + query = Flavor._flavor_query_permission_filter( + query, context.project_domain_id, + fields.FlavorPermissionRuleScope.DOMAIN, + domain_permission) + if context.project_id: + query = Flavor._flavor_query_permission_filter( + query, context.project_id, + fields.FlavorPermissionRuleScope.PROJECT, + project_permission) + return query @staticmethod @@ -349,13 +452,17 @@ def _flavor_get_by_name_from_db(context, name): @staticmethod @db_utils.require_context @api_db_api.context_manager.reader - def _flavor_get_by_flavor_id_from_db(context, flavor_id): + def _flavor_get_by_flavor_id_from_db( + context, flavor_id, domain_permission, + project_permission): """Returns a dict describing specific flavor_id.""" flavor_id = _unaliased_flavor_id(flavor_id) - result = Flavor._flavor_get_query_from_db(context).\ - filter_by(flavorid=flavor_id).\ - order_by(expression.asc(api_models.Flavors.id)).\ - first() + result = Flavor._flavor_get_query_from_db( + context, domain_permission=domain_permission, + project_permission=project_permission).\ + filter_by(flavorid=flavor_id).\ + order_by(expression.asc(api_models.Flavors.id)).\ + first() if not result: raise exception.FlavorNotFound(flavor_id=flavor_id) return _dict_with_extra_specs(result) @@ -430,12 +537,26 @@ def get_by_name(cls, context, name): expected_attrs=['extra_specs']) @base.remotable_classmethod - def get_by_flavor_id(cls, context, flavor_id, read_deleted=None): - db_flavor = cls._flavor_get_by_flavor_id_from_db(context, - flavor_id) + def get_by_flavor_id( + cls, context, flavor_id, read_deleted=None, + domain_permission=_PERM_ALLOW, project_permission=_PERM_ALLOW): + db_flavor = cls._flavor_get_by_flavor_id_from_db( + context, flavor_id, domain_permission=domain_permission, + project_permission=project_permission) return cls._from_db_object(context, cls(context), db_flavor, expected_attrs=['extra_specs']) + @base.remotable + def get_permission( + self, + include_domain: bool = False, + include_project: bool = False, + ) -> dict[str, str]: + result = _get_flavor_permissions_from_db( + self._context, {self.id}, include_domain=include_domain, + include_project=include_project) + return result.get(self.id, {}) + @staticmethod def _flavor_add_project(context, flavor_id, project_id): return _flavor_add_project(context, flavor_id, project_id) @@ -630,12 +751,16 @@ def _send_notification(self, action): @api_db_api.context_manager.reader def _flavor_get_all_from_db(context, inactive, filters, sort_key, sort_dir, - limit, marker): + limit, marker, domain_permission, + project_permission, force_permission_filter): """Returns all flavors. """ filters = filters or {} - query = Flavor._flavor_get_query_from_db(context) + query = Flavor._flavor_get_query_from_db( + context, domain_permission=domain_permission, + project_permission=project_permission, + force_permission_filter=force_permission_filter) if 'min_memory_mb' in filters: query = query.filter( @@ -680,6 +805,69 @@ def _flavor_get_all_from_db(context, inactive, filters, sort_key, sort_dir, return all_flavors +@api_db_api.context_manager.reader +def _get_flavor_ids_by_ids_from_db( + context, + ids: ty.Collection[int], + domain_permission: str | None, + project_permission: str | None, +) -> dict[int, str]: + """Map internal flavor ids to their public flavorid.""" + query = Flavor._flavor_get_query_from_db( + context, domain_permission=domain_permission, + project_permission=project_permission).\ + filter(api_models.Flavors.id.in_(ids)) + return {row.id: row.flavorid for row in query} + + +@api_db_api.context_manager.reader +def _get_flavor_permissions_from_db( + context, + flavor_ids: ty.Collection[int], + include_domain: bool, + include_project: bool, +) -> dict[int, dict[str, str]]: + """Return effective flavor permission per flavor ID and scope. + + Only includes scopes whose context attribute is present. Returns an + empty dict if neither scope resolves to a context value. + """ + include_domain = include_domain and bool(context.project_domain_id) + include_project = include_project and bool(context.project_id) + if not include_domain and not include_project: + return {} + + columns = [api_models.Flavors.id] + if include_domain: + columns.append( + Flavor._flavor_query_allowed_clause( + context.project_domain_id, + fields.FlavorPermissionRuleScope.DOMAIN, + ).label('domain_allowed')) + if include_project: + columns.append( + Flavor._flavor_query_allowed_clause( + context.project_id, + fields.FlavorPermissionRuleScope.PROJECT, + ).label('project_allowed')) + + rows = context.session.query(*columns).filter( + api_models.Flavors.id.in_(flavor_ids), + api_models.Flavors.is_public == sql.true()).all() + + permissions = {} + for row in rows: + permission = {} + if include_domain: + permission[fields.FlavorPermissionRuleScope.DOMAIN] = ( + _PERM_ALLOW if row.domain_allowed else _PERM_DENY) + if include_project: + permission[fields.FlavorPermissionRuleScope.PROJECT] = ( + _PERM_ALLOW if row.project_allowed else _PERM_DENY) + permissions[row.id] = permission + return permissions + + @base.NovaObjectRegistry.register class FlavorList(base.ObjectListBase, base.NovaObject): VERSION = '1.1' @@ -690,14 +878,15 @@ class FlavorList(base.ObjectListBase, base.NovaObject): @base.remotable_classmethod def get_all(cls, context, inactive=False, filters=None, - sort_key='flavorid', sort_dir='asc', limit=None, marker=None): - api_db_flavors = _flavor_get_all_from_db(context, - inactive=inactive, - filters=filters, - sort_key=sort_key, - sort_dir=sort_dir, - limit=limit, - marker=marker) + sort_key='flavorid', sort_dir='asc', limit=None, marker=None, + domain_permission=_PERM_ALLOW, project_permission=_PERM_ALLOW, + force_permission_filter=False): + api_db_flavors = _flavor_get_all_from_db( + context, inactive=inactive, filters=filters, sort_key=sort_key, + sort_dir=sort_dir, limit=limit, marker=marker, + domain_permission=domain_permission, + project_permission=project_permission, + force_permission_filter=force_permission_filter) return base.obj_make_list(context, cls(context), objects.Flavor, api_db_flavors, expected_attrs=['extra_specs']) @@ -722,3 +911,29 @@ def get_by_id(cls, context, ids): } res[x.id] = flavor_info return res + + @base.remotable_classmethod + def get_flavor_ids_by_ids( + cls, context, + ids: ty.Collection[int], + domain_permission: str | None = _PERM_ALLOW, + project_permission: str | None = _PERM_ALLOW, + ) -> dict[int, str]: + """Map internal flavor ids to their public flavorid. + + Internal ids without a matching flavor are omitted. + """ + return _get_flavor_ids_by_ids_from_db( + context, ids, domain_permission=domain_permission, + project_permission=project_permission) + + @base.remotable + def get_permissions( + self, + include_domain: bool = False, + include_project: bool = False, + ) -> dict[int, dict[str, str]]: + flavor_ids = {f.id for f in self.objects} + return _get_flavor_permissions_from_db( + self._context, flavor_ids, include_domain=include_domain, + include_project=include_project) diff --git a/nova/objects/flavor_permission_rule.py b/nova/objects/flavor_permission_rule.py new file mode 100644 index 00000000000..98a9a37e3ef --- /dev/null +++ b/nova/objects/flavor_permission_rule.py @@ -0,0 +1,369 @@ +# 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. + +from __future__ import annotations + +import typing as ty + +from oslo_db import exception as db_exc +from oslo_db.sqlalchemy.utils import paginate_query +from oslo_utils import uuidutils + +from nova.db.api import api as api_db_api +from nova.db.api import models as api_models +from nova.db import utils as db_utils +from nova import exception +from nova.objects import base +from nova.objects import fields + +if ty.TYPE_CHECKING: + from nova import context as nova_context + + +# Maps each DB field that uses a sentinel value to that sentinel. The sentinel +# stands in for None in the object layer: '' means domain-scope (no project), +# -1 means default rule (no specific flavor). +DB_NONE_SENTINELS = { + 'project_id': '', + 'flavor_id': -1, +} + + +@base.NovaObjectRegistry.register +class FlavorPermissionRule(base.NovaPersistentObject, base.NovaObject): + """Restricts public flavors to specific domains and projects. + + A flavor permission rule has a 'domain_id', an optional 'project_id', an + optional 'flavor_id' and an 'effect' ('allow' or 'deny'). + + The scope of a flavor permission rule is derived from 'project_id': + - 'domain' scope rules have no 'project_id'. They apply to all projects + within domain 'domain_id'. + - 'project' scope rules have a 'project_id' and apply only to that project. + They are not inherited by sub-projects. 'domain_id' is the domain of the + project. + + A flavor is permitted for a project if it is permitted at both the domain + scope and the project scope. Rules without a 'flavor_id' define the + domain's or project's default behavior for flavors without a + flavor-specific rule. If a domain or project does not have a default + behavior rule, then all flavors are permitted at that scope by default. + + There is at most one flavor permission rule for each combination of + 'domain_id', 'project_id' and 'flavor_id'. Consequently, a public flavor is + only denied at a scope if for the corresponding project: + - there is a 'deny' rule matching the 'flavor_id' + - OR there is a 'deny' rule without a 'flavor_id' AND there is no 'allow' + rule matching the 'flavor_id' + + Note: Flavor permission rules only apply to public flavors. Private flavors + are controlled exclusively via the flavor access list. + """ + # Version 1.0: Initial version + VERSION = '1.0' + + fields = { + 'id': fields.IntegerField(), + 'uuid': fields.UUIDField(), + 'domain_id': fields.StringField(), + 'project_id': fields.StringField(nullable=True), + 'flavor_id': fields.IntegerField(nullable=True), + 'effect': fields.FlavorPermissionRuleEffectField(), + } + + @property + def scope(self) -> str: + """Derived scope: 'domain' if project_id is None, else 'project'.""" + if self.project_id is None: + return fields.FlavorPermissionRuleScope.DOMAIN + return fields.FlavorPermissionRuleScope.PROJECT + + @staticmethod + def _from_db_object( + context: nova_context.RequestContext, + rule: FlavorPermissionRule, + db_rule: api_models.FlavorPermissionRule, + ) -> FlavorPermissionRule: + # NOTE(sebkro) Delete fields are not implemented in the API DB models, + # but are inherited from NovaPersistentObject + ignore = {'deleted': False, 'deleted_at': None} + for field in rule.fields: + if field in ignore and not hasattr(db_rule, field): + setattr(rule, field, ignore[field]) + if field in db_rule: + value = db_rule[field] + # Translate DB sentinels to None in the object layer + if (field in DB_NONE_SENTINELS and + value == DB_NONE_SENTINELS[field]): + value = None + setattr(rule, field, value) + rule._context = context + rule.obj_reset_changes() + return rule + + @staticmethod + @db_utils.require_context + @api_db_api.context_manager.reader + def _get_by_id_from_db( + context: nova_context.RequestContext, + id: int, + ) -> api_models.FlavorPermissionRule: + query = context.session.query( + api_models.FlavorPermissionRule).filter_by(id=id) + db_rule = query.first() + if not db_rule: + raise exception.FlavorPermissionRuleNotFound(id=id) + return db_rule + + @staticmethod + @db_utils.require_context + @api_db_api.context_manager.reader + def _get_by_uuid_from_db( + context: nova_context.RequestContext, + uuid: str, + ) -> api_models.FlavorPermissionRule: + query = context.session.query( + api_models.FlavorPermissionRule).filter_by(uuid=uuid) + db_rule = query.first() + if not db_rule: + raise exception.FlavorPermissionRuleNotFound(id=uuid) + return db_rule + + @staticmethod + def _to_db_values( + values: dict[str, ty.Any], + ) -> dict[str, ty.Any]: + """Translate object-layer None sentinels to DB sentinel values.""" + values = values.copy() + for field, sentinel in DB_NONE_SENTINELS.items(): + if values.get(field) is None: + values[field] = sentinel + return values + + @staticmethod + @db_utils.require_context + @api_db_api.context_manager.writer + def _create_in_db( + context: nova_context.RequestContext, + values: dict[str, ty.Any], + ) -> api_models.FlavorPermissionRule: + db_rule = api_models.FlavorPermissionRule() + db_values = FlavorPermissionRule._to_db_values(values) + db_rule.update(db_values) + try: + db_rule.save(context.session) + except db_exc.DBDuplicateEntry: + raise exception.FlavorPermissionRuleExists( + uuid=values.get('uuid'), + project_id=values.get('project_id'), + flavor_id=values.get('flavor_id')) + return db_rule + + @staticmethod + @db_utils.require_context + @api_db_api.context_manager.writer + def _destroy_in_db( + context: nova_context.RequestContext, + id: int, + ) -> None: + result = context.session.query( + api_models.FlavorPermissionRule).filter_by(id=id).delete() + if not result: + raise exception.FlavorPermissionRuleNotFound(id=id) + + @staticmethod + @db_utils.require_context + @api_db_api.context_manager.writer + def _save( + context: nova_context.RequestContext, + id: int, + values: dict[str, ty.Any], + ) -> api_models.FlavorPermissionRule: + db_rule = FlavorPermissionRule._get_by_id_from_db(context, id) + values = FlavorPermissionRule._to_db_values(values) + db_rule.update(values) + try: + db_rule.save(context.session) + except db_exc.DBDuplicateEntry: + raise exception.FlavorPermissionRuleExists( + uuid=values.get('uuid'), + project_id=values.get('project_id'), + flavor_id=values.get('flavor_id')) + return db_rule + + @base.remotable_classmethod + def get_by_id( + cls, + context: nova_context.RequestContext, + id: int, + ) -> FlavorPermissionRule: + db_rule = cls._get_by_id_from_db(context, id) + return cls._from_db_object(context, cls(context), db_rule) + + @base.remotable_classmethod + def get_by_uuid( + cls, + context: nova_context.RequestContext, + uuid: str, + ) -> FlavorPermissionRule: + db_rule = cls._get_by_uuid_from_db(context, uuid) + return cls._from_db_object(context, cls(context), db_rule) + + @base.remotable + def create(self) -> None: + if not self.obj_attr_is_set('uuid'): + self.uuid = uuidutils.generate_uuid() + updates = self.obj_get_changes() + db_rule = self._create_in_db(self._context, updates) + self._from_db_object(self._context, self, db_rule) + + @base.remotable + def destroy(self) -> None: + self._destroy_in_db(self._context, self.id) + + @base.remotable + def save(self) -> None: + updates = self.obj_get_changes() + if updates: + db_rule = self._save(self._context, self.id, updates) + # Refresh updated_at. + self._from_db_object(self._context, self, db_rule) + + +@base.NovaObjectRegistry.register +class FlavorPermissionRuleList(base.ObjectListBase, base.NovaObject): + # Version 1.0: Initial version + VERSION = '1.0' + + fields = { + 'objects': fields.ListOfObjectsField('FlavorPermissionRule'), + } + + @staticmethod + @api_db_api.context_manager.reader + def _get_from_db( + context: nova_context.RequestContext, + filter_by_context_domain: bool = True, + filter_by_context_project: bool = True, + domain_id: str | None = None, + project_id: str | None = None, + scope: str | None = None, + effect: str | None = None, + flavor_id: int | None = None, + has_flavor: bool | None = None, + limit: int | None = None, + marker: str | None = None, + ) -> list[api_models.FlavorPermissionRule]: + """Get flavor permission rules from the database. + + :param filter_by_context_domain: If True, restrict rules to + 'context.project_domain_id'. + :param filter_by_context_project: If True, restrict rules to + 'context.project_id'. + :param domain_id: Filter by 'domain_id'. None means no filter. + :param project_id: Filter by 'project_id'. None means no filter. + :param scope: Filter by 'scope' ('domain' or 'project'). + :param effect: Filter by 'effect' ('allow' or 'deny'). + :param flavor_id: Filter by 'flavor_id'. None means no filter. + :param has_flavor: If True, return only flavor-specific rules. If + False, return only default rules. None means no filter. + :param limit: Maximum number of rules to return. + :param marker: UUID of the last rule in the previous page. + """ + Rule = api_models.FlavorPermissionRule + query = context.session.query(Rule) + if scope == fields.FlavorPermissionRuleScope.DOMAIN: + query = query.filter( + Rule.project_id == DB_NONE_SENTINELS['project_id']) + elif scope == fields.FlavorPermissionRuleScope.PROJECT: + query = query.filter( + Rule.project_id != DB_NONE_SENTINELS['project_id']) + + if filter_by_context_domain: + if not context.project_domain_id: + return [] + query = query.filter(Rule.domain_id == context.project_domain_id) + if filter_by_context_project: + if not context.project_id: + return [] + query = query.filter(Rule.project_id == context.project_id) + if domain_id is not None: + query = query.filter(Rule.domain_id == domain_id) + if project_id is not None: + query = query.filter(Rule.project_id == project_id) + if effect is not None: + query = query.filter(Rule.effect == effect) + if flavor_id is not None: + query = query.filter( + Rule.flavor_id == flavor_id) + + if has_flavor is True: + query = query.filter( + Rule.flavor_id != DB_NONE_SENTINELS['flavor_id']) + elif has_flavor is False: + query = query.filter( + Rule.flavor_id == DB_NONE_SENTINELS['flavor_id']) + + marker_rule = None + if marker is not None: + marker_query = context.session.query(Rule).filter_by(uuid=marker) + marker_rule = marker_query.first() + if not marker_rule: + raise exception.MarkerNotFound(marker=marker) + + query = paginate_query(query, Rule, limit, ['id'], marker=marker_rule) + return query.all() + + @base.remotable_classmethod + def get_all( + cls, + context: nova_context.RequestContext, + filter_by_context_domain: bool = True, + filter_by_context_project: bool = True, + domain_id: str | None = None, + project_id: str | None = None, + scope: str | None = None, + effect: str | None = None, + flavor_id: int | None = None, + has_flavor: bool | None = None, + limit: int | None = None, + marker: str | None = None, + ) -> FlavorPermissionRuleList: + """Get flavor permission rules. + + :param filter_by_context_domain: If True, restrict rules to + 'context.project_domain_id'. + :param filter_by_context_project: If True, restrict rules to + 'context.project_id'. + :param domain_id: Filter by 'domain_id'. None means no filter. + :param project_id: Filter by 'project_id'. None means no filter. + :param scope: Filter by 'scope' ('domain' or 'project'). + :param effect: Filter by 'effect' ('allow' or 'deny'). + :param flavor_id: Filter by 'flavor_id'. None means no filter. + :param has_flavor: If True, return only flavor-specific rules. If + False, return only default rules. None means no filter. + :param limit: Maximum number of rules to return. + :param marker: UUID of the last rule in the previous page. + """ + db_rules = cls._get_from_db( + context, + filter_by_context_domain=filter_by_context_domain, + filter_by_context_project=filter_by_context_project, + domain_id=domain_id, project_id=project_id, scope=scope, + effect=effect, flavor_id=flavor_id, has_flavor=has_flavor, + limit=limit, marker=marker) + return base.obj_make_list( + context, cls(context), FlavorPermissionRule, db_rules + ) diff --git a/nova/policies/__init__.py b/nova/policies/__init__.py index f6fe028f737..0ab1404a5a9 100644 --- a/nova/policies/__init__.py +++ b/nova/policies/__init__.py @@ -31,6 +31,7 @@ from nova.policies import flavor_access from nova.policies import flavor_extra_specs from nova.policies import flavor_manage +from nova.policies import flavor_permission_rules from nova.policies import floating_ip_pools from nova.policies import floating_ips from nova.policies import hosts @@ -91,6 +92,7 @@ def list_rules(): flavor_access.list_rules(), flavor_extra_specs.list_rules(), flavor_manage.list_rules(), + flavor_permission_rules.list_rules(), floating_ip_pools.list_rules(), floating_ips.list_rules(), hosts.list_rules(), diff --git a/nova/policies/flavor_permission_rules.py b/nova/policies/flavor_permission_rules.py new file mode 100644 index 00000000000..973dfb7bf5c --- /dev/null +++ b/nova/policies/flavor_permission_rules.py @@ -0,0 +1,150 @@ +# 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. + +from oslo_policy import policy + +from nova.policies import base + +POLICY_ROOT = 'os_compute_api:os-flavor-permission-rules:%s' + +flavor_permission_rules_policies = [ + policy.DocumentedRuleDefault( + name=POLICY_ROOT % 'index:all', + check_str=base.ADMIN, + description='List all flavor permission rules.', + operations=[{'method': 'GET', + 'path': '/flavor-permission-rules'}], + scope_types=['project']), + policy.DocumentedRuleDefault( + name=POLICY_ROOT % 'index:domain', + check_str=base.ADMIN, + description='List flavor permission rules for the context domain. ' + 'Include context domain permission annotations in flavor ' + 'detail/show responses and allow filtering flavor ' + 'index/detail by context domain permission.', + operations=[{'method': 'GET', + 'path': '/flavor-permission-rules'}, + {'method': 'GET', + 'path': '/flavors'}, + {'method': 'GET', + 'path': '/flavors/detail'}, + {'method': 'GET', + 'path': '/flavors/{flavor_id}'}], + scope_types=['project']), + policy.DocumentedRuleDefault( + name=POLICY_ROOT % 'index:project', + check_str=base.ADMIN, + description='List flavor permission rules for the context project. ' + 'Include context project permission annotations in flavor ' + 'detail/show responses and allow filtering flavor ' + 'index/detail by context project permission.', + operations=[{'method': 'GET', + 'path': '/flavor-permission-rules'}, + {'method': 'GET', + 'path': '/flavors'}, + {'method': 'GET', + 'path': '/flavors/detail'}, + {'method': 'GET', + 'path': '/flavors/{flavor_id}'}], + scope_types=['project']), + policy.DocumentedRuleDefault( + name=POLICY_ROOT % 'show:domain', + check_str=base.ADMIN, + description=( + 'Show a domain-scoped flavor permission rule. ' + 'Supports %(project_domain_id)s in check_str to restrict access ' + 'to users whose context matches the rule domain.'), + operations=[{'method': 'GET', + 'path': '/flavor-permission-rules/{id}'}], + scope_types=['project']), + policy.DocumentedRuleDefault( + name=POLICY_ROOT % 'show:project', + check_str=base.ADMIN, + description=( + 'Show a project-scoped flavor permission rule. ' + 'Supports %(project_domain_id)s and %(project_id)s in check_str ' + 'to restrict access to users whose context matches the rule ' + 'domain or project.'), + operations=[{'method': 'GET', + 'path': '/flavor-permission-rules/{id}'}], + scope_types=['project']), + policy.DocumentedRuleDefault( + name=POLICY_ROOT % 'create:domain', + check_str=base.ADMIN, + description=( + 'Create a domain-scoped flavor permission rule. ' + 'Supports %(project_domain_id)s in check_str to restrict access ' + 'to users whose context matches the rule domain.'), + operations=[{'method': 'POST', + 'path': '/flavor-permission-rules'}], + scope_types=['project']), + policy.DocumentedRuleDefault( + name=POLICY_ROOT % 'create:project', + check_str=base.ADMIN, + description=( + 'Create a project-scoped flavor permission rule. ' + 'Supports %(project_domain_id)s and %(project_id)s in check_str ' + 'to restrict access to users whose context matches the rule ' + 'domain or project.'), + operations=[{'method': 'POST', + 'path': '/flavor-permission-rules'}], + scope_types=['project']), + policy.DocumentedRuleDefault( + name=POLICY_ROOT % 'delete:domain', + check_str=base.ADMIN, + description=( + 'Delete a domain-scoped flavor permission rule. ' + 'Supports %(project_domain_id)s in check_str to restrict access ' + 'to users whose context matches the rule domain.'), + operations=[{'method': 'DELETE', + 'path': '/flavor-permission-rules/{id}'}], + scope_types=['project']), + policy.DocumentedRuleDefault( + name=POLICY_ROOT % 'delete:project', + check_str=base.ADMIN, + description=( + 'Delete a project-scoped flavor permission rule. ' + 'Supports %(project_domain_id)s and %(project_id)s in check_str ' + 'to restrict access to users whose context matches the rule ' + 'domain or project.'), + operations=[{'method': 'DELETE', + 'path': '/flavor-permission-rules/{id}'}], + scope_types=['project']), + policy.DocumentedRuleDefault( + name=POLICY_ROOT % 'update:domain', + check_str=base.ADMIN, + description=( + 'Update a domain-scoped flavor permission rule. ' + 'Supports %(project_domain_id)s in check_str to restrict access ' + 'to users whose context matches the rule domain.'), + operations=[{'method': 'PUT', + 'path': '/flavor-permission-rules/{id}'}], + scope_types=['project']), + policy.DocumentedRuleDefault( + name=POLICY_ROOT % 'update:project', + check_str=base.ADMIN, + description=( + 'Update a project-scoped flavor permission rule. ' + 'Supports %(project_domain_id)s and %(project_id)s in check_str ' + 'to restrict access to users whose context matches the rule ' + 'domain or project.'), + operations=[{'method': 'PUT', + 'path': '/flavor-permission-rules/{id}'}], + scope_types=['project']), +] + + +def list_rules() -> list[policy.DocumentedRuleDefault]: + return flavor_permission_rules_policies diff --git a/nova/tests/functional/db/test_flavor.py b/nova/tests/functional/db/test_flavor.py index c0435b5cbc6..eafa2722dec 100644 --- a/nova/tests/functional/db/test_flavor.py +++ b/nova/tests/functional/db/test_flavor.py @@ -15,6 +15,7 @@ from nova.db.api import models as api_models from nova import exception from nova import objects +from nova.objects import fields from nova import test from nova.tests import fixtures @@ -175,3 +176,280 @@ def test_get_all_with_marker_not_found(self): flavor.create() self.assertRaises(exception.MarkerNotFound, self._test_get_all, 2, marker='noflavoratall') + + def test_get_flavor_ids_by_ids_maps_internal_ids(self): + flavor = objects.Flavor(context=self.context, **fake_api_flavor) + flavor.create() + result = objects.FlavorList.get_flavor_ids_by_ids( + self.context, {flavor.id}) + self.assertEqual({flavor.id: flavor.flavorid}, result) + + def test_get_flavor_ids_by_ids_omits_missing(self): + flavor = objects.Flavor(context=self.context, **fake_api_flavor) + flavor.create() + result = objects.FlavorList.get_flavor_ids_by_ids( + self.context, {flavor.id, 999999}) + self.assertEqual({flavor.id: flavor.flavorid}, result) + + def test_get_flavor_ids_by_ids_empty(self): + self.assertEqual( + {}, objects.FlavorList.get_flavor_ids_by_ids(self.context, set())) + + +class FlavorPermissionTestCase(test.NoDBTestCase): + """Tests for the flavor permission rule handling. + """ + USES_DB_SELF = True + + PROJECT_ID = 'fake-project' + DOMAIN_ID = 'fake-domain' + ALLOW = fields.FlavorPermissionRuleEffect.ALLOW + DENY = fields.FlavorPermissionRuleEffect.DENY + + def _get_context(self, include_domain=True): + return context.RequestContext( + 'fake-user', self.PROJECT_ID, + project_domain_id=self.DOMAIN_ID if include_domain else None) + + def setUp(self): + super().setUp() + self.useFixture(fixtures.Database()) + self.useFixture(fixtures.Database(database='api')) + self.context = self._get_context() + flavor = objects.Flavor(context=self.context, + **dict(fake_api_flavor, is_public=True)) + flavor.create() + self.flavor = flavor + + def _create_rule(self, domain_id, effect, project_id=None, + flavor_id=None): + rule = objects.FlavorPermissionRule( + context=self.context, + domain_id=domain_id, + project_id=project_id, + effect=effect, + flavor_id=flavor_id, + ) + rule.create() + return rule + + def _flavor_visible( + self, + domain_permission=fields.FlavorPermissionRuleEffect.ALLOW, + project_permission=fields.FlavorPermissionRuleEffect.ALLOW, + context=None, + ): + flavors = objects.FlavorList.get_all( + context or self.context, + domain_permission=domain_permission, + project_permission=project_permission) + return any(f.id == self.flavor.id for f in flavors) + + def test_no_rules_flavor_accessible(self): + self.assertTrue(self._flavor_visible()) + + def test_domain_specific_deny_hides_flavor(self): + self._create_rule(self.DOMAIN_ID, self.DENY, flavor_id=self.flavor.id) + self.assertFalse(self._flavor_visible()) + # domain_permission does not affect project-level denial + self.assertFalse(self._flavor_visible( + project_permission=None)) + + def test_domain_default_deny_hides_flavor(self): + self._create_rule(self.DOMAIN_ID, self.DENY) + self.assertFalse(self._flavor_visible()) + # domain_permission does not affect project-level denial + self.assertFalse(self._flavor_visible( + project_permission=None)) + + def test_domain_default_deny_overridden_by_specific_allow(self): + self._create_rule(self.DOMAIN_ID, self.DENY) + self._create_rule(self.DOMAIN_ID, self.ALLOW, flavor_id=self.flavor.id) + self.assertTrue(self._flavor_visible()) + + def test_project_specific_deny_hides_flavor(self): + self._create_rule(self.DOMAIN_ID, self.DENY, + project_id=self.PROJECT_ID, + flavor_id=self.flavor.id) + self.assertFalse(self._flavor_visible()) + self.assertTrue(self._flavor_visible( + project_permission=None)) + + def test_project_default_deny_hides_flavor(self): + self._create_rule(self.DOMAIN_ID, self.DENY, + project_id=self.PROJECT_ID) + self.assertFalse(self._flavor_visible()) + self.assertTrue(self._flavor_visible( + project_permission=None)) + + def test_project_default_deny_overridden_by_specific_allow(self): + self._create_rule(self.DOMAIN_ID, self.DENY, + project_id=self.PROJECT_ID) + self._create_rule(self.DOMAIN_ID, self.ALLOW, + project_id=self.PROJECT_ID, + flavor_id=self.flavor.id) + self.assertTrue(self._flavor_visible()) + + def test_no_domain_id_in_context_skips_domain_filter(self): + self._create_rule(self.DOMAIN_ID, self.DENY, flavor_id=self.flavor.id) + self.assertTrue(self._flavor_visible( + context=self._get_context(include_domain=False))) + + def test_deny_for_other_project_does_not_affect_visibility(self): + self._create_rule(self.DOMAIN_ID, self.DENY, + project_id='other-project', + flavor_id=self.flavor.id) + self._create_rule(self.DOMAIN_ID, self.DENY, + project_id='other-project') + self.assertTrue(self._flavor_visible()) + + def test_deny_for_other_domain_does_not_affect_visibility(self): + self._create_rule('other-domain', self.DENY, flavor_id=self.flavor.id) + self._create_rule('other-domain', self.DENY) + self.assertTrue(self._flavor_visible()) + + def test_denied_domain_filter_returns_denied_flavors(self): + # Flavor not in denied set without domain scope deny rule + self.assertFalse(self._flavor_visible( + domain_permission=fields.FlavorPermissionRuleEffect.DENY, + project_permission=None)) + # Flavor in denied set with domain scope deny rule + self._create_rule(self.DOMAIN_ID, self.DENY, flavor_id=self.flavor.id) + self.assertTrue(self._flavor_visible( + domain_permission=fields.FlavorPermissionRuleEffect.DENY, + project_permission=None)) + + def test_denied_project_filter_returns_denied_flavors(self): + # Flavor not in denied set without project scope deny rule + self.assertFalse(self._flavor_visible( + domain_permission=None, + project_permission=fields.FlavorPermissionRuleEffect.DENY)) + # Flavor in denied set with project scope deny rule + self._create_rule(self.DOMAIN_ID, self.DENY, + project_id=self.PROJECT_ID, + flavor_id=self.flavor.id) + self.assertTrue(self._flavor_visible( + domain_permission=None, + project_permission=fields.FlavorPermissionRuleEffect.DENY)) + + def test_private_flavor_not_affected_by_fpr(self): + """Flavor permission rules do not hide PRIVATE flavors""" + private = objects.Flavor( + context=self.context, + **dict(fake_api_flavor, + name='m1.private', flavorid='m1.private', + is_public=False, projects=[self.PROJECT_ID])) + private.create() + + # Domain-scope deny for all flavors + self._create_rule(self.DOMAIN_ID, self.DENY) + # Project-scope deny for all flavors + self._create_rule(self.DOMAIN_ID, self.DENY, + project_id=self.PROJECT_ID) + + # Public flavor is denied by the rules + self.assertFalse(self._flavor_visible()) + # Private flavor is NOT affected by FPRs; it passes through + flavors = objects.FlavorList.get_all(self.context) + self.assertIn(private.id, {f.id for f in flavors}) + + def test_admin_context_bypasses_permission_rules(self): + self._create_rule(self.DOMAIN_ID, self.DENY, flavor_id=self.flavor.id) + self._create_rule(self.DOMAIN_ID, self.DENY, + project_id=self.PROJECT_ID, + flavor_id=self.flavor.id) + admin_ctx = context.get_admin_context() + self.assertTrue(self._flavor_visible(context=admin_ctx)) + + def test_no_project_id_in_context_skips_project_filter(self): + self._create_rule(self.DOMAIN_ID, self.DENY, + project_id=self.PROJECT_ID, + flavor_id=self.flavor.id) + ctx = context.RequestContext( + 'fake-user', None, project_domain_id=self.DOMAIN_ID) + self.assertTrue(self._flavor_visible(context=ctx)) + + def test_get_permission_db(self): + # Flavor allowed at both scopes without deny rules + result = self.flavor.get_permission( + include_domain=True, include_project=True) + self.assertEqual({ + fields.FlavorPermissionRuleScope.DOMAIN: + fields.FlavorPermissionRuleEffect.ALLOW, + fields.FlavorPermissionRuleScope.PROJECT: + fields.FlavorPermissionRuleEffect.ALLOW, + }, result) + # Flavor denied at domain scope with a domain deny rule + self._create_rule(self.DOMAIN_ID, self.DENY, flavor_id=self.flavor.id) + result = self.flavor.get_permission( + include_domain=True, include_project=True) + self.assertEqual({ + fields.FlavorPermissionRuleScope.DOMAIN: + fields.FlavorPermissionRuleEffect.DENY, + fields.FlavorPermissionRuleScope.PROJECT: + fields.FlavorPermissionRuleEffect.ALLOW, + }, result) + + def test_get_permissions_db(self): + flavor_list = objects.FlavorList( + context=self.context, objects=[self.flavor]) + result = flavor_list.get_permissions( + include_domain=True, include_project=True) + self.assertEqual({ + self.flavor.id: { + fields.FlavorPermissionRuleScope.DOMAIN: + fields.FlavorPermissionRuleEffect.ALLOW, + fields.FlavorPermissionRuleScope.PROJECT: + fields.FlavorPermissionRuleEffect.ALLOW, + } + }, result) + + def test_get_permission_private_flavor_returns_empty(self): + private = objects.Flavor( + context=self.context, + **dict(fake_api_flavor, + name='m1.private', flavorid='m1.private', + is_public=False, projects=[])) + private.create() + self._create_rule(self.DOMAIN_ID, self.DENY, flavor_id=private.id) + flavor = objects.Flavor(context=self.context, id=private.id) + result = flavor.get_permission( + include_domain=True, include_project=True) + self.assertEqual({}, result) + + def test_get_permissions_omits_private_flavors(self): + private = objects.Flavor( + context=self.context, + **dict(fake_api_flavor, + name='m1.private', flavorid='m1.private', + is_public=False, projects=[])) + private.create() + self._create_rule(self.DOMAIN_ID, self.DENY, flavor_id=private.id) + flavor_list = objects.FlavorList( + context=self.context, objects=[self.flavor, private]) + result = flavor_list.get_permissions( + include_domain=True, include_project=True) + self.assertIn(self.flavor.id, result) + self.assertNotIn(private.id, result) + + def test_get_permission_empty_context(self): + # Context with no project_domain_id and no project_id returns {} + ctx = context.RequestContext('fake-user', None) + flavor = objects.Flavor(context=ctx, id=self.flavor.id) + result = flavor.get_permission( + include_domain=True, include_project=True) + self.assertEqual({}, result) + + def test_get_flavor_ids_by_ids_enforces_permission(self): + self._create_rule(self.DOMAIN_ID, self.DENY, flavor_id=self.flavor.id) + # Enforcing (default): the denied flavor is omitted + self.assertEqual( + {}, objects.FlavorList.get_flavor_ids_by_ids( + self.context, {self.flavor.id})) + # ALL bypasses permission filtering: the flavor is returned + self.assertEqual( + {self.flavor.id: self.flavor.flavorid}, + objects.FlavorList.get_flavor_ids_by_ids( + self.context, {self.flavor.id}, + domain_permission=None, + project_permission=None)) diff --git a/nova/tests/unit/api/openstack/compute/test_flavor_access.py b/nova/tests/unit/api/openstack/compute/test_flavor_access.py index 2e137a78446..6622e3d5b49 100644 --- a/nova/tests/unit/api/openstack/compute/test_flavor_access.py +++ b/nova/tests/unit/api/openstack/compute/test_flavor_access.py @@ -24,6 +24,7 @@ from nova.api.openstack.compute import flavors as flavors_api from nova import context from nova import exception +from nova.objects import fields from nova import test from nova.tests.unit.api.openstack import fakes @@ -71,7 +72,11 @@ def fake_get_flavor_access_by_flavor_id(context, flavorid): return res -def fake_get_flavor_by_flavor_id(context, flavorid): +def fake_get_flavor_by_flavor_id(context, flavorid, + domain_permission=( + fields.FlavorPermissionRuleEffect.ALLOW), + project_permission=( + fields.FlavorPermissionRuleEffect.ALLOW)): return FLAVORS[flavorid] @@ -83,9 +88,12 @@ def _has_flavor_access(flavorid, projectid): return False -def fake_get_all_flavors_sorted_list(context, inactive=False, - filters=None, sort_key='flavorid', - sort_dir='asc', limit=None, marker=None): +def fake_get_all_flavors_sorted_list( + context, inactive=False, filters=None, sort_key='flavorid', + sort_dir='asc', limit=None, marker=None, + domain_permission=fields.FlavorPermissionRuleEffect.ALLOW, + project_permission=fields.FlavorPermissionRuleEffect.ALLOW, + force_permission_filter=False): if filters is None or filters['is_public'] is None: return sorted(FLAVORS.values(), key=lambda item: item[sort_key]) diff --git a/nova/tests/unit/api/openstack/compute/test_flavor_permission_rules.py b/nova/tests/unit/api/openstack/compute/test_flavor_permission_rules.py new file mode 100644 index 00000000000..53bf6e3b864 --- /dev/null +++ b/nova/tests/unit/api/openstack/compute/test_flavor_permission_rules.py @@ -0,0 +1,465 @@ +# 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. + +from unittest import mock + +import webob.exc + +from nova.api.openstack.compute import ( + flavor_permission_rules as fpr_api) +import nova.conf +from nova import exception +from nova.objects import fields +from nova.policies import flavor_permission_rules as fpr_policies +from nova import test +from nova.tests.unit.api.openstack import fakes +from nova.tests.unit import fake_flavor_permission_rule as fake_rule + + +CONF = nova.conf.CONF + +ROOT = fpr_policies.POLICY_ROOT +DOMAIN_ID = 'fake-domain' +PROJECT_ID = fakes.FAKE_PROJECT_ID +ALLOW = fields.FlavorPermissionRuleEffect.ALLOW +DENY = fields.FlavorPermissionRuleEffect.DENY +SCOPE_DOMAIN = fields.FlavorPermissionRuleScope.DOMAIN +SCOPE_PROJECT = fields.FlavorPermissionRuleScope.PROJECT + + +def _make_can(allowed_suffixes): + """Return a context.can side_effect allowing specific policy suffixes.""" + def can(action, target=None, fatal=True): + if any(action == ROOT % s for s in allowed_suffixes): + return True + if fatal: + raise exception.PolicyNotAuthorized(action=action) + return False + return can + + +class FlavorPermissionRulesControllerTest(test.NoDBTestCase): + + def setUp(self): + super().setUp() + self.controller = fpr_api.FlavorPermissionRulesController() + patcher = mock.patch( + 'nova.objects.FlavorList.get_flavor_ids_by_ids', return_value={}) + self.mock_flavorids = patcher.start() + self.addCleanup(patcher.stop) + + def _req(self, *allowed_policies, qs=''): + """Build a request whose context allows the given policy suffixes.""" + url = '/flavor-permission-rules' + if qs: + url += '?' + qs + req = fakes.HTTPRequestV21.blank(url) + ctx = req.environ['nova.context'] + ctx.project_domain_id = DOMAIN_ID + ctx.can = mock.Mock(side_effect=_make_can(allowed_policies)) + return req + + def _rule_obj(self, req, **db_updates): + ctx = req.environ['nova.context'] + db = fake_rule.fake_db_flavor_permission_rule(**db_updates) + return fake_rule.fake_flavor_permission_rule_obj(ctx, db) + + @mock.patch('nova.objects.FlavorPermissionRuleList.get_all') + def test_index_all_access(self, mock_get_all): + mock_get_all.return_value = [] + self.controller.index(self._req('index:all')) + mock_get_all.assert_called_once_with( + mock.ANY, + filter_by_context_domain=False, + filter_by_context_project=False, + domain_id=None, project_id=None, + scope=None, effect=None, + flavor_id=None, has_flavor=None, + limit=CONF.api.max_limit, marker=None) + + @mock.patch('nova.objects.FlavorPermissionRuleList.get_all') + def test_index_domain_access(self, mock_get_all): + mock_get_all.return_value = [] + self.controller.index(self._req('index:domain')) + mock_get_all.assert_called_once_with( + mock.ANY, + filter_by_context_domain=True, + filter_by_context_project=False, + domain_id=None, project_id=None, + scope=None, effect=None, + flavor_id=None, has_flavor=None, + limit=CONF.api.max_limit, marker=None) + + @mock.patch('nova.objects.FlavorPermissionRuleList.get_all') + def test_index_project_access(self, mock_get_all): + mock_get_all.return_value = [] + self.controller.index(self._req('index:project')) + mock_get_all.assert_called_once_with( + mock.ANY, + filter_by_context_domain=True, + filter_by_context_project=True, + domain_id=None, project_id=None, + scope=SCOPE_PROJECT, effect=None, + flavor_id=None, has_flavor=None, + limit=CONF.api.max_limit, marker=None) + + def test_index_unauthorized(self): + self.assertRaises( + exception.PolicyNotAuthorized, + self.controller.index, self._req()) + + @mock.patch('nova.objects.FlavorPermissionRuleList.get_all') + def test_index_scope_domain_forbidden_for_project_only( + self, mock_get_all): + req = self._req('index:project', qs='scope=' + SCOPE_DOMAIN) + self.assertRaises(webob.exc.HTTPForbidden, + self.controller.index, req) + + @mock.patch('nova.objects.FlavorPermissionRuleList.get_all') + def test_index_domain_id_forbidden_for_non_all(self, mock_get_all): + req = self._req('index:domain', qs='domain_id=some-domain') + self.assertRaises(webob.exc.HTTPForbidden, + self.controller.index, req) + + @mock.patch('nova.objects.FlavorPermissionRuleList.get_all') + def test_index_domain_id_allowed_for_all(self, mock_get_all): + mock_get_all.return_value = [] + self.controller.index( + self._req('index:all', qs='domain_id=some-domain')) + mock_get_all.assert_called_once_with( + mock.ANY, + filter_by_context_domain=False, + filter_by_context_project=False, + domain_id='some-domain', + project_id=None, scope=None, effect=None, + flavor_id=None, has_flavor=None, + limit=CONF.api.max_limit, marker=None) + + @mock.patch('nova.objects.FlavorPermissionRuleList.get_all') + def test_index_project_id_filter(self, mock_get_all): + mock_get_all.return_value = [] + self.controller.index( + self._req('index:all', qs='project_id=some-project')) + mock_get_all.assert_called_once_with( + mock.ANY, + filter_by_context_domain=False, + filter_by_context_project=False, + domain_id=None, project_id='some-project', + scope=None, effect=None, + flavor_id=None, has_flavor=None, + limit=CONF.api.max_limit, marker=None) + + @mock.patch('nova.objects.FlavorPermissionRuleList.get_all') + def test_index_effect_filter(self, mock_get_all): + mock_get_all.return_value = [] + self.controller.index( + self._req('index:all', qs='effect=' + DENY)) + mock_get_all.assert_called_once_with( + mock.ANY, + filter_by_context_domain=False, + filter_by_context_project=False, + domain_id=None, project_id=None, + scope=None, effect=DENY, + flavor_id=None, has_flavor=None, + limit=CONF.api.max_limit, marker=None) + + @mock.patch('nova.api.openstack.common.get_flavor') + @mock.patch('nova.objects.FlavorPermissionRuleList.get_all') + def test_index_flavor_id_resolved(self, mock_get_all, mock_get_flavor): + mock_get_all.return_value = [] + fake_flavor = mock.Mock() + fake_flavor.id = 42 + mock_get_flavor.return_value = fake_flavor + self.controller.index( + self._req('index:all', qs='flavor_id=1')) + mock_get_flavor.assert_called_once_with( + mock.ANY, '1', + domain_permission=None, + project_permission=None) + mock_get_all.assert_called_once_with( + mock.ANY, + filter_by_context_domain=False, + filter_by_context_project=False, + domain_id=None, project_id=None, + scope=None, effect=None, + flavor_id=42, has_flavor=None, + limit=CONF.api.max_limit, marker=None) + + @mock.patch('nova.api.openstack.common.get_flavor') + def test_index_flavor_id_not_found(self, mock_get_flavor): + mock_get_flavor.side_effect = webob.exc.HTTPNotFound + req = self._req('index:all', qs='flavor_id=nonexistent') + self.assertRaises(webob.exc.HTTPNotFound, + self.controller.index, req) + + @mock.patch('nova.objects.FlavorPermissionRuleList.get_all') + def test_index_marker_not_found(self, mock_get_all): + mock_get_all.side_effect = exception.MarkerNotFound( + marker='bad-marker') + req = self._req('index:all', qs='marker=bad-marker') + self.assertRaises(webob.exc.HTTPBadRequest, + self.controller.index, req) + + @mock.patch('nova.objects.FlavorPermissionRuleList.get_all') + def test_index_has_flavor_false(self, mock_get_all): + mock_get_all.return_value = [] + self.controller.index( + self._req('index:all', qs='has_flavor=false')) + mock_get_all.assert_called_once_with( + mock.ANY, + filter_by_context_domain=False, + filter_by_context_project=False, + domain_id=None, project_id=None, + scope=None, effect=None, + flavor_id=None, has_flavor=False, + limit=CONF.api.max_limit, marker=None) + + @mock.patch('nova.objects.FlavorPermissionRuleList.get_all') + def test_index_has_flavor_true(self, mock_get_all): + mock_get_all.return_value = [] + self.controller.index( + self._req('index:all', qs='has_flavor=true')) + mock_get_all.assert_called_once_with( + mock.ANY, + filter_by_context_domain=False, + filter_by_context_project=False, + domain_id=None, project_id=None, + scope=None, effect=None, + flavor_id=None, has_flavor=True, + limit=CONF.api.max_limit, marker=None) + + @mock.patch('nova.api.openstack.common.get_flavor') + def test_index_has_flavor_and_flavor_id_exclusive(self, mock_get_flavor): + fake_flavor = mock.Mock() + fake_flavor.id = 42 + mock_get_flavor.return_value = fake_flavor + req = self._req('index:all', qs='flavor_id=1&has_flavor=true') + self.assertRaises(webob.exc.HTTPBadRequest, + self.controller.index, req) + + @mock.patch('nova.objects.FlavorPermissionRuleList.get_all') + def test_index_returns_rules(self, mock_get_all): + req = self._req('index:all') + rule = self._rule_obj(req) + mock_get_all.return_value = [rule] + result = self.controller.index(req) + rules = result['flavor_permission_rules'] + self.assertEqual(1, len(rules)) + self.assertEqual(rule.uuid, rules[0]['id']) + self.assertEqual(rule.domain_id, rules[0]['domain_id']) + self.assertEqual(rule.effect, rules[0]['effect']) + + @mock.patch('nova.objects.FlavorPermissionRule.get_by_uuid') + def test_show_domain_rule(self, mock_get): + req = self._req('show:domain') + rule = self._rule_obj(req, project_id=None) + mock_get.return_value = rule + result = self.controller.show(req, rule.uuid) + rule_dict = result['flavor_permission_rule'] + self.assertEqual(rule.uuid, rule_dict['id']) + self.assertEqual(SCOPE_DOMAIN, rule_dict['scope']) + self.assertIsNone(rule_dict['project_id']) + + @mock.patch('nova.objects.FlavorPermissionRule.get_by_uuid') + def test_show_project_rule(self, mock_get): + req = self._req('show:project') + rule = self._rule_obj(req) + mock_get.return_value = rule + result = self.controller.show(req, rule.uuid) + self.assertEqual(SCOPE_PROJECT, + result['flavor_permission_rule']['scope']) + + @mock.patch('nova.objects.FlavorPermissionRule.get_by_uuid') + def test_show_translates_flavor_id(self, mock_get): + req = self._req('show:project') + rule = self._rule_obj(req, flavor_id=123) + mock_get.return_value = rule + self.mock_flavorids.return_value = {123: '1'} + result = self.controller.show(req, rule.uuid) + rule_dict = result['flavor_permission_rule'] + self.assertEqual('1', rule_dict['flavor_id']) + self.mock_flavorids.assert_called_once_with( + mock.ANY, {123}, + domain_permission=None, + project_permission=None) + + @mock.patch('nova.objects.FlavorPermissionRule.get_by_uuid') + def test_show_nonexistent(self, mock_get): + mock_get.side_effect = exception.FlavorPermissionRuleNotFound( + id='nonexistent-uuid') + self.assertRaises(webob.exc.HTTPNotFound, + self.controller.show, + self._req('show:domain', 'show:project'), + 'nonexistent-uuid') + + @mock.patch('nova.objects.FlavorPermissionRule.get_by_uuid') + def test_show_unauthorized(self, mock_get): + req = self._req() # no show policies + rule = self._rule_obj(req) + mock_get.return_value = rule + self.assertRaises(webob.exc.HTTPNotFound, + self.controller.show, req, rule.uuid) + + @mock.patch('nova.objects.FlavorPermissionRule.create') + def test_create_domain_scope(self, mock_create): + req = self._req('create:domain') + body = {'flavor_permission_rule': { + 'domain_id': DOMAIN_ID, + 'effect': ALLOW, + }} + result = self.controller.create(req, body=body) + rule_dict = result['flavor_permission_rule'] + self.assertEqual(DOMAIN_ID, rule_dict['domain_id']) + self.assertIsNone(rule_dict['project_id']) + self.assertEqual(SCOPE_DOMAIN, rule_dict['scope']) + self.assertEqual(ALLOW, rule_dict['effect']) + + @mock.patch('nova.objects.FlavorPermissionRule.create') + def test_create_project_scope(self, mock_create): + req = self._req('create:project') + body = {'flavor_permission_rule': { + 'domain_id': DOMAIN_ID, + 'project_id': PROJECT_ID, + 'effect': DENY, + }} + result = self.controller.create(req, body=body) + rule_dict = result['flavor_permission_rule'] + self.assertEqual(DOMAIN_ID, rule_dict['domain_id']) + self.assertEqual(PROJECT_ID, rule_dict['project_id']) + self.assertEqual(SCOPE_PROJECT, rule_dict['scope']) + self.assertEqual(DENY, rule_dict['effect']) + + @mock.patch('nova.api.openstack.common.get_flavor') + @mock.patch('nova.objects.FlavorPermissionRule.create') + def test_create_with_flavor_id(self, mock_create, mock_get_flavor): + fake_flavor = mock.Mock() + fake_flavor.id = 99 + fake_flavor.flavorid = '1' + fake_flavor.is_public = True + mock_get_flavor.return_value = fake_flavor + req = self._req('create:domain') + body = {'flavor_permission_rule': { + 'domain_id': DOMAIN_ID, + 'effect': DENY, + 'flavor_id': '1', + }} + result = self.controller.create(req, body=body) + mock_get_flavor.assert_called_once_with( + mock.ANY, '1', + domain_permission=None, + project_permission=None) + self.assertEqual( + '1', result['flavor_permission_rule']['flavor_id']) + + @mock.patch('nova.api.openstack.common.get_flavor') + def test_create_flavor_not_found(self, mock_get_flavor): + mock_get_flavor.side_effect = webob.exc.HTTPNotFound + req = self._req('create:domain') + body = {'flavor_permission_rule': { + 'domain_id': DOMAIN_ID, + 'effect': ALLOW, + 'flavor_id': 'nonexistent', + }} + self.assertRaises(webob.exc.HTTPNotFound, + self.controller.create, req, body=body) + + @mock.patch('nova.api.openstack.common.get_flavor') + def test_create_private_flavor_raises_bad_request(self, mock_get_flavor): + fake_flavor = mock.Mock() + fake_flavor.is_public = False + mock_get_flavor.return_value = fake_flavor + req = self._req('create:domain') + body = {'flavor_permission_rule': { + 'domain_id': DOMAIN_ID, + 'effect': DENY, + 'flavor_id': '2', + }} + self.assertRaises(webob.exc.HTTPBadRequest, + self.controller.create, req, body=body) + + @mock.patch('nova.objects.FlavorPermissionRule.create') + def test_create_duplicate(self, mock_create): + mock_create.side_effect = exception.FlavorPermissionRuleExists( + uuid='fake-uuid', project_id=None, flavor_id=None) + req = self._req('create:domain') + body = {'flavor_permission_rule': { + 'domain_id': DOMAIN_ID, + 'effect': ALLOW, + }} + self.assertRaises(webob.exc.HTTPConflict, + self.controller.create, req, body=body) + + def test_create_unauthorized(self): + req = self._req() # no create policies + body = {'flavor_permission_rule': { + 'domain_id': DOMAIN_ID, + 'effect': ALLOW, + }} + self.assertRaises(exception.PolicyNotAuthorized, + self.controller.create, req, body=body) + + @mock.patch('nova.objects.FlavorPermissionRule.save') + @mock.patch('nova.objects.FlavorPermissionRule.get_by_uuid') + def test_update_success(self, mock_get, mock_save): + req = self._req('update:project') + rule = self._rule_obj(req) + mock_get.return_value = rule + body = {'flavor_permission_rule': {'effect': DENY}} + result = self.controller.update(req, rule.uuid, body=body) + self.assertEqual(DENY, result['flavor_permission_rule']['effect']) + + @mock.patch('nova.objects.FlavorPermissionRule.get_by_uuid') + def test_update_nonexistent(self, mock_get): + mock_get.side_effect = exception.FlavorPermissionRuleNotFound( + id='nonexistent-uuid') + req = self._req('update:domain', 'update:project') + body = {'flavor_permission_rule': {'effect': DENY}} + self.assertRaises(webob.exc.HTTPNotFound, + self.controller.update, req, + 'nonexistent-uuid', body=body) + + @mock.patch('nova.objects.FlavorPermissionRule.get_by_uuid') + def test_update_unauthorized(self, mock_get): + req = self._req() # no update policies + rule = self._rule_obj(req) + mock_get.return_value = rule + body = {'flavor_permission_rule': {'effect': DENY}} + self.assertRaises(webob.exc.HTTPNotFound, + self.controller.update, req, rule.uuid, body=body) + + @mock.patch('nova.objects.FlavorPermissionRule.destroy') + @mock.patch('nova.objects.FlavorPermissionRule.get_by_uuid') + def test_delete_success(self, mock_get, mock_destroy): + req = self._req('delete:project') + rule = self._rule_obj(req) + mock_get.return_value = rule + self.controller.delete(req, rule.uuid) + mock_destroy.assert_called_once_with() + + @mock.patch('nova.objects.FlavorPermissionRule.get_by_uuid') + def test_delete_nonexistent(self, mock_get): + mock_get.side_effect = exception.FlavorPermissionRuleNotFound( + id='nonexistent-uuid') + req = self._req('delete:domain', 'delete:project') + self.assertRaises(webob.exc.HTTPNotFound, + self.controller.delete, req, 'nonexistent-uuid') + + @mock.patch('nova.objects.FlavorPermissionRule.get_by_uuid') + def test_delete_unauthorized(self, mock_get): + req = self._req() # no delete policies + rule = self._rule_obj(req) + mock_get.return_value = rule + self.assertRaises(webob.exc.HTTPNotFound, + self.controller.delete, req, rule.uuid) diff --git a/nova/tests/unit/api/openstack/compute/test_flavors.py b/nova/tests/unit/api/openstack/compute/test_flavors.py index c7fbf5c468a..64c282219f4 100644 --- a/nova/tests/unit/api/openstack/compute/test_flavors.py +++ b/nova/tests/unit/api/openstack/compute/test_flavors.py @@ -23,6 +23,8 @@ from nova import context from nova import exception from nova import objects +from nova.objects import fields +from nova.policies import flavor_permission_rules as fpr_policies from nova import test from nova.tests.unit.api.openstack import fakes from nova.tests.unit import matchers @@ -40,7 +42,7 @@ def fake_get_limit_and_marker(request, max_limit=1): return limit, marker -def return_flavor_not_found(context, flavor_id, read_deleted=None): +def return_flavor_not_found(context, flavor_id, *args, **kwargs): raise exception.FlavorNotFound(flavor_id=flavor_id) @@ -144,6 +146,21 @@ def test_get_flavor_with_custom_link_prefix(self): self._set_expected_body(expected['flavor'], fakes.FLAVORS['1']) self.assertEqual(expected, flavor) + @mock.patch('nova.objects.Flavor.get_permission') + def test_show_includes_permissions(self, mock_perm): + perm = { + fields.FlavorPermissionRuleScope.DOMAIN: + fields.FlavorPermissionRuleEffect.ALLOW, + fields.FlavorPermissionRuleScope.PROJECT: + fields.FlavorPermissionRuleEffect.ALLOW, + } + mock_perm.return_value = perm + req = self.fake_request.blank( + self._prefix + '/flavors/1', + use_admin_context=True, version=self.microversion) + result = self.controller.show(req, '1') + self.assertEqual(perm, result['flavor']['permissions']) + def test_get_flavor_list(self): req = self._build_request('/flavors') flavor = self.controller.index(req) @@ -416,6 +433,24 @@ def test_get_flavor_list_detail(self): self._set_expected_body(expected['flavors'][1], fakes.FLAVORS['2']) self.assertEqual(expected, flavor) + @mock.patch('nova.objects.FlavorList.get_permissions') + def test_detail_includes_permissions(self, mock_perms): + perms = { + 1: {fields.FlavorPermissionRuleScope.DOMAIN: + fields.FlavorPermissionRuleEffect.ALLOW}, + 2: {fields.FlavorPermissionRuleScope.DOMAIN: + fields.FlavorPermissionRuleEffect.DENY}, + } + mock_perms.return_value = perms + req = self.fake_request.blank( + self._prefix + '/flavors/detail', + use_admin_context=True, version=self.microversion) + result = self.controller.detail(req) + self.assertEqual(perms[fakes.FLAVORS['1'].id], + result['flavors'][0]['permissions']) + self.assertEqual(perms[fakes.FLAVORS['2'].id], + result['flavors'][1]['permissions']) + @mock.patch('nova.objects.FlavorList.get_all', return_value=objects.FlavorList()) def test_get_empty_flavor_list(self, mock_get): @@ -1012,3 +1047,151 @@ def test_string_none(self): def test_other(self): self.assertRaises( webob.exc.HTTPBadRequest, self.assertPublic, None, 'other') + + +class FlavorPermissionRulesIntegrationTest(test.NoDBTestCase): + """Tests for permission filtering and annotation.""" + + FPR_ROOT = fpr_policies.POLICY_ROOT + + def setUp(self): + super().setUp() + self.controller = flavors_v21.FlavorsController() + + def _req(self, *fpr_suffixes, qs='', use_admin_context=False): + url = '/flavors' + if qs: + url += '?' + qs + req = fakes.HTTPRequestV21.blank(url) + ctx = req.environ['nova.context'] + ctx.is_admin = use_admin_context + allowed = {self.FPR_ROOT % s for s in fpr_suffixes} + + def can(action, target=None, fatal=True): + if action in allowed: + return True + if fatal: + raise exception.PolicyNotAuthorized(action=action) + return False + + ctx.can = mock.Mock(side_effect=can) + return req + + @mock.patch('nova.objects.FlavorList.get_all') + def test_domain_permission_forbidden(self, mock_get_all): + self.assertRaises( + webob.exc.HTTPForbidden, self.controller.index, + self._req(qs='domain_permission=' + + fields.FlavorPermissionRuleEffect.ALLOW)) + + @mock.patch('nova.objects.FlavorList.get_all') + def test_project_permission_forbidden(self, mock_get_all): + self.assertRaises( + webob.exc.HTTPForbidden, self.controller.index, + self._req(qs='project_permission=' + + fields.FlavorPermissionRuleEffect.ALLOW)) + + @mock.patch('nova.objects.FlavorList.get_all') + def test_domain_permission_forwarded(self, mock_get_all): + mock_get_all.return_value = objects.FlavorList() + self.controller.index( + self._req('index:domain', + qs='domain_permission=' + + fields.FlavorPermissionRuleEffect.DENY)) + self.assertEqual( + fields.FlavorPermissionRuleEffect.DENY, + mock_get_all.call_args[1]['domain_permission']) + + @mock.patch('nova.objects.FlavorList.get_all') + def test_project_permission_forwarded(self, mock_get_all): + mock_get_all.return_value = objects.FlavorList() + self.controller.index( + self._req('index:project', qs='project_permission=all')) + self.assertIsNone(mock_get_all.call_args[1]['project_permission']) + + @mock.patch('nova.objects.FlavorList.get_all') + def test_admin_explicit_permission_forces_filter(self, mock_get_all): + mock_get_all.return_value = objects.FlavorList() + req = self._req( + 'index:domain', + qs='domain_permission=' + fields.FlavorPermissionRuleEffect.DENY, + use_admin_context=True) + self.controller.index(req) + self.assertTrue(mock_get_all.call_args[1]['force_permission_filter']) + self.assertEqual( + fields.FlavorPermissionRuleEffect.DENY, + mock_get_all.call_args[1]['domain_permission']) + + @mock.patch('nova.objects.FlavorList.get_all') + def test_admin_no_permission_param_no_force(self, mock_get_all): + mock_get_all.return_value = objects.FlavorList() + req = self._req('index:domain', use_admin_context=True) + self.controller.index(req) + self.assertFalse(mock_get_all.call_args[1]['force_permission_filter']) + + @mock.patch('nova.objects.FlavorList.get_all') + @mock.patch('nova.objects.FlavorList.get_permissions') + def test_detail_no_fpr_policy_skips_get_permissions( + self, mock_gp, mock_get_all): + mock_get_all.return_value = objects.FlavorList() + self.controller.detail(self._req()) + mock_gp.assert_not_called() + + @mock.patch('nova.objects.FlavorList.get_all') + @mock.patch('nova.objects.FlavorList.get_permissions') + def test_detail_domain_policy_calls_get_permissions( + self, mock_gp, mock_get_all): + mock_gp.return_value = {} + mock_get_all.return_value = objects.FlavorList() + self.controller.detail(self._req('index:domain')) + mock_gp.assert_called_once_with( + include_domain=True, include_project=True) + + @mock.patch('nova.objects.FlavorList.get_all') + @mock.patch('nova.objects.FlavorList.get_permissions') + def test_detail_project_policy_calls_get_permissions( + self, mock_gp, mock_get_all): + mock_gp.return_value = {} + mock_get_all.return_value = objects.FlavorList() + self.controller.detail(self._req('index:project')) + mock_gp.assert_called_once_with( + include_domain=False, include_project=True) + + @mock.patch('nova.compute.flavors.get_flavor_by_flavor_id') + @mock.patch('nova.objects.Flavor.get_permission') + def test_show_no_fpr_policy_no_permission_filters( + self, mock_perm, mock_get): + mock_get.return_value = fakes.FLAVORS['1'] + self.controller.show(self._req(), 'm1.tiny') + mock_get.assert_called_once_with( + 'm1.tiny', ctxt=mock.ANY, + domain_permission=fields.FlavorPermissionRuleEffect.ALLOW, + project_permission=fields.FlavorPermissionRuleEffect.ALLOW) + mock_perm.assert_not_called() + + @mock.patch('nova.compute.flavors.get_flavor_by_flavor_id') + @mock.patch('nova.objects.Flavor.get_permission') + def test_show_domain_policy_uses_all_filter(self, mock_perm, mock_get): + mock_perm.return_value = {} + mock_get.return_value = fakes.FLAVORS['1'] + self.controller.show(self._req('index:domain'), 'm1.tiny') + mock_get.assert_called_once_with( + 'm1.tiny', ctxt=mock.ANY, + domain_permission=None, + project_permission=None) + mock_perm.assert_called_once_with( + include_domain=True, include_project=True) + + @mock.patch('nova.compute.flavors.get_flavor_by_flavor_id') + @mock.patch('nova.objects.Flavor.get_permission') + def test_show_project_policy_project_filter_only( + self, mock_perm, mock_get): + mock_perm.return_value = {} + mock_get.return_value = fakes.FLAVORS['1'] + self.controller.show(self._req('index:project'), 'm1.tiny') + mock_get.assert_called_once_with( + 'm1.tiny', ctxt=mock.ANY, + domain_permission=fields.FlavorPermissionRuleEffect.ALLOW, + project_permission=None) + mock_perm.assert_called_once_with( + include_domain=False, include_project=True) diff --git a/nova/tests/unit/api/openstack/fakes.py b/nova/tests/unit/api/openstack/fakes.py index 1cb48613d21..9462de50645 100644 --- a/nova/tests/unit/api/openstack/fakes.py +++ b/nova/tests/unit/api/openstack/fakes.py @@ -38,6 +38,7 @@ from nova import exception as exc from nova import objects from nova.objects import base +from nova.objects import fields from nova import quota from nova.tests.unit import fake_block_device from nova.tests.unit.objects import test_keypair @@ -741,7 +742,11 @@ def fake_not_implemented(*args, **kwargs): def stub_out_flavor_get_by_flavor_id(test): @classmethod - def fake_get_by_flavor_id(cls, context, flavor_id, read_deleted=None): + def fake_get_by_flavor_id(cls, context, flavor_id, read_deleted=None, + domain_permission=( + fields.FlavorPermissionRuleEffect.ALLOW), + project_permission=( + fields.FlavorPermissionRuleEffect.ALLOW)): return FLAVORS[flavor_id] test.stub_out('nova.objects.Flavor.get_by_flavor_id', @@ -752,7 +757,12 @@ def stub_out_flavor_get_all(test): @staticmethod def fake_get_all(context, inactive=False, filters=None, sort_key='flavorid', sort_dir='asc', limit=None, - marker=None): + marker=None, + domain_permission=( + fields.FlavorPermissionRuleEffect.ALLOW), + project_permission=( + fields.FlavorPermissionRuleEffect.ALLOW), + force_permission_filter=False): if marker in ['99999']: raise exc.MarkerNotFound(marker) diff --git a/nova/tests/unit/db/api/test_migrations.py b/nova/tests/unit/db/api/test_migrations.py index 37565208071..0827eea73b0 100644 --- a/nova/tests/unit/db/api/test_migrations.py +++ b/nova/tests/unit/db/api/test_migrations.py @@ -196,6 +196,36 @@ def _check_cdeec0c85668(self, connection): # removal without creating it first, which is dumb pass + def _pre_upgrade_f177f53f978b(self, connection): + # we use the inspector here rather than oslo_db.utils.column_exists, + # since the latter will create a new connection + inspector = sqlalchemy.inspect(connection) + self.assertFalse(inspector.has_table('flavor_permission_rules')) + + def _check_f177f53f978b(self, connection): + # we use the inspector here rather than oslo_db.utils.column_exists, + # since the latter will create a new connection + inspector = sqlalchemy.inspect(connection) + self.assertTrue(inspector.has_table('flavor_permission_rules')) + columns = {x['name'] for x in + inspector.get_columns('flavor_permission_rules')} + expected_columns = {'id', 'uuid', 'project_id', 'flavor_id', 'type', + 'scope', 'created_at', 'updated_at'} + self.assertEqual(expected_columns, columns) + unique_constraints = [ + set(c['column_names']) for c in inspector.get_unique_constraints( + 'flavor_permission_rules')] + expected_unique_constraints = [ + {'uuid'}, {'flavor_id', 'project_id', 'scope'}] + for c in expected_unique_constraints: + self.assertIn(c, unique_constraints) + indexes = [ + set(idx['column_names']) for idx in inspector.get_indexes( + 'flavor_permission_rules')] + expected_indexes = [{'uuid'}, {'project_id', 'scope', 'flavor_id'}] + for idx in expected_indexes: + self.assertIn(idx, indexes) + def test_single_base_revision(self): """Ensure we only have a single base revision. diff --git a/nova/tests/unit/fake_flavor_permission_rule.py b/nova/tests/unit/fake_flavor_permission_rule.py new file mode 100644 index 00000000000..55ccd85a365 --- /dev/null +++ b/nova/tests/unit/fake_flavor_permission_rule.py @@ -0,0 +1,54 @@ +# Copyright (c) 2025 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. + +from oslo_utils import uuidutils + +from nova import objects +from nova.objects import fields +from nova.objects.flavor_permission_rule import DB_NONE_SENTINELS + + +def fake_db_flavor_permission_rule(**updates): + for field, sentinel in DB_NONE_SENTINELS.items(): + if field in updates and updates[field] is None: + updates[field] = sentinel + db_rule = { + 'id': 1, + 'uuid': uuidutils.generate_uuid(), + 'domain_id': 'fake-domain', + 'project_id': 'fake-project', + 'flavor_id': 123, + 'effect': fields.FlavorPermissionRuleEffect.ALLOW, + } | updates + + for name, field in objects.FlavorPermissionRule.fields.items(): + if name in db_rule: + continue + if field.nullable: + db_rule[name] = None + elif field.default != fields.UnspecifiedDefault: + db_rule[name] = field.default + else: + raise Exception( + f'fake_db_flavor_permission_rule needs help with {name}') + + return db_rule + + +def fake_flavor_permission_rule_obj(context, db_rule=None, **updates): + if db_rule is None: + db_rule = fake_db_flavor_permission_rule() + return objects.FlavorPermissionRule._from_db_object( + context, objects.FlavorPermissionRule(), db_rule | updates) diff --git a/nova/tests/unit/objects/test_flavor.py b/nova/tests/unit/objects/test_flavor.py index 4172d3fda3a..12e85f25e18 100644 --- a/nova/tests/unit/objects/test_flavor.py +++ b/nova/tests/unit/objects/test_flavor.py @@ -13,6 +13,7 @@ # under the License. import datetime +import ddt from unittest import mock from oslo_db import exception as db_exc @@ -88,7 +89,10 @@ def test_get_by_flavor_id_from_api(self, mock_get): mock_get.return_value = fake_flavor flavor = flavor_obj.Flavor.get_by_flavor_id(self.context, 'm1.foo') self._compare(self, fake_flavor, flavor) - mock_get.assert_called_once_with(self.context, 'm1.foo') + mock_get.assert_called_once_with( + self.context, 'm1.foo', + domain_permission=fields.FlavorPermissionRuleEffect.ALLOW, + project_permission=fields.FlavorPermissionRuleEffect.ALLOW) @staticmethod @api_db_api.context_manager.writer @@ -279,6 +283,24 @@ def test_destroy_api_by_flavorid(self, mock_destroy): mock_destroy.assert_called_once_with(self.context, flavorid=flavor.flavorid) + @mock.patch('nova.objects.flavor._get_flavor_permissions_from_db') + def test_get_permission(self, mock_get_perms): + flavor = flavor_obj.Flavor( + context=self.context, id=fake_flavor['id']) + expected = { + fields.FlavorPermissionRuleScope.DOMAIN: + fields.FlavorPermissionRuleEffect.ALLOW, + fields.FlavorPermissionRuleScope.PROJECT: + fields.FlavorPermissionRuleEffect.DENY, + } + mock_get_perms.return_value = {fake_flavor['id']: expected} + result = flavor.get_permission( + include_domain=True, include_project=True) + self.assertEqual(expected, result) + mock_get_perms.assert_called_once_with( + self.context, {fake_flavor['id']}, + include_domain=True, include_project=True) + def test_load_projects_from_api(self): mock_get_projects = mock.Mock(return_value=['a', 'b']) objects.Flavor._get_projects_from_db = mock_get_projects @@ -383,7 +405,8 @@ def test_get_all_from_db(self): api_flavors = flavor_obj._flavor_get_all_from_db(self.context, False, None, 'flavorid', 'asc', - None, None) + None, None, None, + None, False) flavors = objects.FlavorList.get_all(self.context) # Make sure we're getting all flavors from the api @@ -408,10 +431,13 @@ def test_get_all(self, mock_api_get): sort_dir='asc') self.assertEqual(1, len(flavors)) _TestFlavor._compare(self, _fake_flavor, flavors[0]) - mock_api_get.assert_called_once_with(self.context, inactive=False, - filters=filters, sort_key='id', - sort_dir='asc', limit=None, - marker=None) + mock_api_get.assert_called_once_with( + self.context, inactive=False, + filters=filters, sort_key='id', + sort_dir='asc', limit=None, marker=None, + domain_permission=fields.FlavorPermissionRuleEffect.ALLOW, + project_permission=fields.FlavorPermissionRuleEffect.ALLOW, + force_permission_filter=False) @mock.patch('nova.objects.flavor._flavor_get_all_from_db') def test_get_all_limit_applied_to_api(self, mock_api_get): @@ -428,10 +454,33 @@ def test_get_all_limit_applied_to_api(self, mock_api_get): sort_dir='asc') self.assertEqual(1, len(flavors)) _TestFlavor._compare(self, _fake_flavor, flavors[0]) - mock_api_get.assert_called_once_with(self.context, inactive=False, - filters=filters, sort_key='id', - sort_dir='asc', limit=1, - marker=None) + mock_api_get.assert_called_once_with( + self.context, inactive=False, + filters=filters, sort_key='id', + sort_dir='asc', limit=1, marker=None, + domain_permission=fields.FlavorPermissionRuleEffect.ALLOW, + project_permission=fields.FlavorPermissionRuleEffect.ALLOW, + force_permission_filter=False) + + @mock.patch('nova.objects.flavor._get_flavor_permissions_from_db') + def test_get_permissions(self, mock_get_perms): + flavor = flavor_obj.Flavor( + context=self.context, id=fake_flavor['id']) + flavor_list = flavor_obj.FlavorList( + context=self.context, objects=[flavor]) + expected = { + fake_flavor['id']: { + fields.FlavorPermissionRuleScope.DOMAIN: + fields.FlavorPermissionRuleEffect.ALLOW, + }, + } + mock_get_perms.return_value = expected + result = flavor_list.get_permissions( + include_domain=True, include_project=False) + self.assertEqual(expected, result) + mock_get_perms.assert_called_once_with( + self.context, {fake_flavor['id']}, + include_domain=True, include_project=False) def test_get_no_marker_in_api(self): self.assertRaises(exception.MarkerNotFound, @@ -573,3 +622,22 @@ def test_min_memory_mb_AND_root_gb_filter(self): filters = {'min_memory_mb': 16384, 'min_root_gb': 80} expected = ['m1.xlarge'] self.assertFilterResults(filters, expected) + + +@ddt.ddt +class TestFlavorQueryPermissionFilter(test.NoDBTestCase): + + @ddt.data( + (None, False), + (fields.FlavorPermissionRuleEffect.ALLOW, True), + (fields.FlavorPermissionRuleEffect.DENY, True), + ) + @ddt.unpack + @mock.patch.object(flavor_obj.Flavor, '_flavor_query_allowed_clause') + def test_filter(self, permission, expect_filter, mock_clause): + mock_clause.return_value = True + query = mock.MagicMock() + flavor_obj.Flavor._flavor_query_permission_filter( + query, 'proj1', 'project', permission) + self.assertEqual(expect_filter, mock_clause.called) + self.assertEqual(expect_filter, query.filter.called) diff --git a/nova/tests/unit/objects/test_flavor_permission_rule.py b/nova/tests/unit/objects/test_flavor_permission_rule.py new file mode 100644 index 00000000000..71852e2a391 --- /dev/null +++ b/nova/tests/unit/objects/test_flavor_permission_rule.py @@ -0,0 +1,464 @@ +# Copyright (c) 2025 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. + +from datetime import datetime +import ddt +from unittest import mock + +from nova import context +from nova.db.api import api as api_db_api +from nova.db.api import models as api_models +from nova import exception +from nova import objects +from nova.objects import fields +from nova.objects.flavor_permission_rule import DB_NONE_SENTINELS +from nova import test +from nova.tests.unit import fake_flavor_permission_rule as fake_rule +from nova.tests.unit.objects import test_objects + + +class _TestFlavorPermissionRuleObject: + """Mixin for FlavorPermissionRule object test cases.""" + + _fake_db_rule = fake_rule.fake_db_flavor_permission_rule() + + def _get_fake_db_rule(self, **updates): + return fake_rule.fake_db_flavor_permission_rule(**updates) + + def _get_fake_rule_obj(self, db_rule=None): + if db_rule is None: + db_rule = self._fake_db_rule + return fake_rule.fake_flavor_permission_rule_obj( + self.context, db_rule) + + def _compare_obj(self, rule_obj, db_rule, db_allow_missing=None, + db_allow_none=None): + if db_allow_none is None: + db_allow_none = [] + if db_allow_missing is None: + db_allow_missing = [] + for field in rule_obj.fields: + if field not in db_rule and field in db_allow_missing: + self.assertTrue(rule_obj.obj_attr_is_set(field)) + continue + + db_val = db_rule[field] + # Mirror the sentinel translation that _from_db_object performs + if (field in DB_NONE_SENTINELS and + db_val == DB_NONE_SENTINELS[field]): + db_val = None + + if db_val is None and field in db_allow_none: + self.assertTrue(rule_obj.obj_attr_is_set(field)) + continue + + obj_val = getattr(rule_obj, field) + if isinstance(obj_val, datetime): + obj_val = obj_val.replace(tzinfo=None) + # Indirection API loses microsecond precision + if rule_obj.indirection_api: + db_val = db_val.replace(microsecond=0) + + self.assertEqual(db_val, obj_val, f'field: {field}') + + @staticmethod + @api_db_api.context_manager.writer + def _create_context_db_rule(context, fake_db_rule=None): + if fake_db_rule is None: + fake_db_rule = _TestFlavorPermissionRuleObject._fake_db_rule + fake_db_rule = fake_db_rule.copy() + fake_db_rule.pop('id') + db_rule = api_models.FlavorPermissionRule() + db_rule.update(fake_db_rule) + db_rule.save(context.session) + return db_rule + + def _create_db_rule(self, fake_db_rule=None): + return self._create_context_db_rule(self.context, fake_db_rule) + + +class TestFlavorPermissionRuleObjectNoDB(test.NoDBTestCase, + _TestFlavorPermissionRuleObject): + + def setUp(self): + super().setUp() + # Set up context like _BaseTestCase does + self.user_id = 'fake-user' + self.project_id = 'fake-project' + self.context = context.RequestContext(self.user_id, self.project_id) + + @mock.patch('nova.objects.FlavorPermissionRule._get_by_id_from_db') + def test_get_by_id(self, mock_get): + mock_get.return_value = self._fake_db_rule + rule = objects.FlavorPermissionRule.get_by_id( + self.context, self._fake_db_rule['id']) + self._compare_obj(rule, self._fake_db_rule) + mock_get.assert_called_once_with( + self.context, self._fake_db_rule['id']) + + @mock.patch('nova.objects.FlavorPermissionRule._get_by_uuid_from_db') + def test_get_by_uuid(self, mock_get): + mock_get.return_value = self._fake_db_rule + rule = objects.FlavorPermissionRule.get_by_uuid( + self.context, self._fake_db_rule['uuid']) + self._compare_obj(rule, self._fake_db_rule) + mock_get.assert_called_once_with( + self.context, self._fake_db_rule['uuid']) + + @mock.patch('nova.objects.FlavorPermissionRule._create_in_db') + def test_create(self, mock_create): + mock_create.return_value = self._fake_db_rule + fake_db_rule = self._fake_db_rule.copy() + fake_db_rule.pop('id') + # Create rule with tracked changes + rule = objects.FlavorPermissionRule(context=self.context, + **fake_db_rule) + rule.create() + mock_create.assert_called_once_with(self.context, fake_db_rule) + self._compare_obj(rule, self._fake_db_rule) + + @mock.patch('nova.objects.FlavorPermissionRule._destroy_in_db') + def test_destroy(self, mock_destroy): + mock_destroy.return_value = self._fake_db_rule + rule = self._get_fake_rule_obj() + rule.destroy() + mock_destroy.assert_called_once_with(self.context, 1) + + @mock.patch('nova.objects.FlavorPermissionRule._save') + def test_save(self, mock_save): + new_effect = fields.FlavorPermissionRuleEffect.DENY + mock_save.return_value = self._fake_db_rule | {'effect': new_effect} + rule = self._get_fake_rule_obj() + rule.effect = new_effect + rule.save() + mock_save.assert_called_once_with( + self.context, rule.id, {'effect': new_effect}) + + @mock.patch('nova.objects.FlavorPermissionRule._save') + def test_save_no_changes(self, mock_save): + rule = self._get_fake_rule_obj() + rule.obj_reset_changes() + rule.save() + mock_save.assert_not_called() + + def test_scope_domain(self): + rule = self._get_fake_rule_obj( + self._get_fake_db_rule(project_id=None)) + self.assertEqual( + fields.FlavorPermissionRuleScope.DOMAIN, rule.scope) + + def test_scope_project(self): + rule = self._get_fake_rule_obj() + self.assertEqual( + fields.FlavorPermissionRuleScope.PROJECT, rule.scope) + + +class _TestFlavorPermissionRuleObjectDB(_TestFlavorPermissionRuleObject): + """Mixin for FlavorPermissionRule object test cases with DB.""" + + def test_get_by_id(self): + db_rule = self._create_db_rule() + rule = objects.FlavorPermissionRule.get_by_id(self.context, db_rule.id) + self._compare_obj(rule, db_rule) + + def test_get_by_uuid(self): + db_rule = self._create_db_rule() + rule = objects.FlavorPermissionRule.get_by_uuid( + self.context, db_rule.uuid) + self._compare_obj(rule, db_rule) + + def test_create(self): + fake_db_rule = self._fake_db_rule.copy() + fake_db_rule.pop('id') + # Create rule with tracked changes + rule = objects.FlavorPermissionRule(context=self.context, + **fake_db_rule) + rule.create() + self.assertIsNotNone(rule.id) + self.assertIsNotNone(rule.created_at) + self._compare_obj(rule, fake_db_rule, db_allow_missing=['id'], + db_allow_none=['created_at']) + updated_rule = objects.FlavorPermissionRule.get_by_id( + self.context, rule.id) + self._compare_obj(updated_rule, fake_db_rule, db_allow_missing=['id'], + db_allow_none=['created_at']) + + def test_destroy(self): + db_rule = self._create_db_rule() + rule = objects.FlavorPermissionRule.get_by_id(self.context, db_rule.id) + rule.destroy() + self.assertRaises( + exception.FlavorPermissionRuleNotFound, + objects.FlavorPermissionRule.get_by_id, + self.context, db_rule.id) + + def test_save(self): + db_rule = self._create_db_rule() + rule = objects.FlavorPermissionRule.get_by_id(self.context, db_rule.id) + new_effect = fields.FlavorPermissionRuleEffect.DENY + rule.effect = new_effect + rule.save() + updated_rule = objects.FlavorPermissionRule.get_by_id( + self.context, db_rule.id) + self.assertEqual(new_effect, updated_rule.effect) + + def test_get_by_id_not_found(self): + self.assertRaises( + exception.FlavorPermissionRuleNotFound, + objects.FlavorPermissionRule.get_by_id, + self.context, 99999) + + def test_get_by_uuid_not_found(self): + self.assertRaises( + exception.FlavorPermissionRuleNotFound, + objects.FlavorPermissionRule.get_by_uuid, + self.context, 'nonexistent-uuid') + + def test_create_duplicate(self): + fake_db_rule = self._fake_db_rule.copy() + fake_db_rule.pop('id') + rule1 = objects.FlavorPermissionRule( + context=self.context, **fake_db_rule) + rule1.create() + rule2 = objects.FlavorPermissionRule( + context=self.context, **fake_db_rule) + self.assertRaises( + exception.FlavorPermissionRuleExists, rule2.create) + + +class TestFlavorPermissionRuleObject( + test_objects._LocalTest, _TestFlavorPermissionRuleObjectDB): + pass + + +class TestFlavorPermissionRuleObjectRemote( + test_objects._RemoteTest, _TestFlavorPermissionRuleObjectDB): + pass + + +@ddt.ddt +class TestFlavorPermissionRuleListObjectNoDB(test.NoDBTestCase, + _TestFlavorPermissionRuleObject): + + def setUp(self): + super().setUp() + # Set up context like _BaseTestCase does + self.user_id = 'fake-user' + self.project_id = 'fake-project' + self.context = context.RequestContext( + self.user_id, self.project_id, project_domain_id='fake-domain') + + @ddt.data( + ({}, True, True), + ({'filter_by_context_domain': False}, False, True), + ({'filter_by_context_project': False}, True, False), + ) + @ddt.unpack + @mock.patch('nova.objects.FlavorPermissionRuleList._get_from_db') + def test_get_all_context_filters( + self, call_kwargs, exp_filter_domain, exp_filter_project, + mock_get): + mock_get.return_value = [] + objects.FlavorPermissionRuleList.get_all( + self.context, **call_kwargs) + mock_get.assert_called_once_with( + self.context, + filter_by_context_domain=exp_filter_domain, + filter_by_context_project=exp_filter_project, + domain_id=None, project_id=None, scope=None, effect=None, + flavor_id=None, has_flavor=None, limit=None, marker=None) + + @ddt.data( + ('scope', fields.FlavorPermissionRuleScope.PROJECT), + ('effect', fields.FlavorPermissionRuleEffect.DENY), + ('flavor_id', 123), + ('domain_id', 'dom-a'), + ('project_id', 'proj-a'), + ('has_flavor', True), + ('has_flavor', False), + ) + @ddt.unpack + @mock.patch('nova.objects.FlavorPermissionRuleList._get_from_db') + def test_get_all_filter(self, arg_name, value, mock_get): + mock_get.return_value = [] + objects.FlavorPermissionRuleList.get_all( + self.context, + filter_by_context_domain=False, + filter_by_context_project=False, + **{arg_name: value}) + self.assertEqual(value, mock_get.call_args[1][arg_name]) + + +class _TestFlavorPermissionRuleListObject(_TestFlavorPermissionRuleObject): + + def test_get_all(self): + db_rule1 = self._create_db_rule() + db_rule2 = self._create_db_rule( + self._get_fake_db_rule(project_id='project-2')) + rules = objects.FlavorPermissionRuleList.get_all( + self.context, + filter_by_context_domain=False, + filter_by_context_project=False) + self.assertEqual(2, len(rules)) + rules = objects.FlavorPermissionRuleList.get_all( + self.context, + filter_by_context_domain=False, + filter_by_context_project=False, + limit=1) + self.assertEqual(1, len(rules)) + self.assertEqual(db_rule1.id, rules[0].id) + rules = objects.FlavorPermissionRuleList.get_all( + self.context, + filter_by_context_domain=False, + filter_by_context_project=False, + limit=1, marker=db_rule1.uuid) + self.assertEqual(1, len(rules)) + self.assertEqual(db_rule2.id, rules[0].id) + + def test_get_all_domain_and_project_filters(self): + ctx = context.RequestContext( + 'fake-user', 'fake-project', + project_domain_id='fake-domain') + self._create_db_rule() + self._create_db_rule( + self._get_fake_db_rule(project_id=None)) + self._create_db_rule( + self._get_fake_db_rule(project_id='other-project')) + self._create_db_rule( + self._get_fake_db_rule(domain_id='other-domain', + project_id='other-project')) + + def _get_all(**kwargs): + return {(r.domain_id, r.project_id) + for r in objects.FlavorPermissionRuleList.get_all( + ctx, **kwargs)} + + self.assertEqual( + {('fake-domain', 'fake-project')}, + _get_all()) + self.assertEqual( + {('fake-domain', None), ('fake-domain', 'fake-project'), + ('fake-domain', 'other-project')}, + _get_all(filter_by_context_project=False)) + self.assertEqual( + {('fake-domain', 'fake-project'), + ('fake-domain', 'other-project')}, + _get_all(scope=fields.FlavorPermissionRuleScope.PROJECT, + filter_by_context_project=False)) + self.assertEqual( + {('fake-domain', None)}, + _get_all(scope=fields.FlavorPermissionRuleScope.DOMAIN, + filter_by_context_project=False)) + self.assertEqual( + {('fake-domain', None), ('fake-domain', 'fake-project'), + ('fake-domain', 'other-project'), + ('other-domain', 'other-project')}, + _get_all(filter_by_context_domain=False, + filter_by_context_project=False)) + + def test_get_all_marker_not_found(self): + self.assertRaises( + exception.MarkerNotFound, + objects.FlavorPermissionRuleList.get_all, + self.context, + filter_by_context_domain=False, + filter_by_context_project=False, + marker='nonexistent-uuid') + + def test_get_all_empty_context_domain(self): + self._create_db_rule() + ctx = context.RequestContext('fake-user', 'fake-project') + rules = objects.FlavorPermissionRuleList.get_all( + ctx, + filter_by_context_domain=True, + filter_by_context_project=False) + self.assertEqual(0, len(rules)) + + def test_get_all_empty_context_project(self): + self._create_db_rule() + ctx = context.RequestContext( + 'fake-user', None, project_domain_id='fake-domain') + rules = objects.FlavorPermissionRuleList.get_all( + ctx, + filter_by_context_domain=False, + filter_by_context_project=True) + self.assertEqual(0, len(rules)) + + def test_get_all_effect_filter(self): + allow_rule = self._create_db_rule() + deny_rule = self._create_db_rule( + self._get_fake_db_rule( + project_id='other-project', + effect=fields.FlavorPermissionRuleEffect.DENY)) + rules = objects.FlavorPermissionRuleList.get_all( + self.context, + filter_by_context_domain=False, + filter_by_context_project=False, + effect=fields.FlavorPermissionRuleEffect.ALLOW) + self.assertEqual(1, len(rules)) + self.assertEqual(allow_rule.id, rules[0].id) + rules = objects.FlavorPermissionRuleList.get_all( + self.context, + filter_by_context_domain=False, + filter_by_context_project=False, + effect=fields.FlavorPermissionRuleEffect.DENY) + self.assertEqual(1, len(rules)) + self.assertEqual(deny_rule.id, rules[0].id) + + def test_get_all_flavor_id_filter(self): + self._create_db_rule() + flavor_id = 456 + rule = self._create_db_rule( + self._get_fake_db_rule(flavor_id=flavor_id)) + rules = objects.FlavorPermissionRuleList.get_all( + self.context, + filter_by_context_domain=False, + filter_by_context_project=False, + flavor_id=flavor_id) + self.assertEqual(1, len(rules)) + self.assertEqual(rule.id, rules[0].id) + + def test_get_all_has_flavor_filter(self): + flavor_rule = self._create_db_rule() + default_rule = self._create_db_rule( + self._get_fake_db_rule(flavor_id=None)) + + rules = objects.FlavorPermissionRuleList.get_all( + self.context, + filter_by_context_domain=False, + filter_by_context_project=False, + has_flavor=False) + self.assertEqual(1, len(rules)) + self.assertEqual(default_rule.id, rules[0].id) + self.assertIsNone(rules[0].flavor_id) + + rules = objects.FlavorPermissionRuleList.get_all( + self.context, + filter_by_context_domain=False, + filter_by_context_project=False, + has_flavor=True) + self.assertEqual(1, len(rules)) + self.assertEqual(flavor_rule.id, rules[0].id) + self.assertIsNotNone(rules[0].flavor_id) + + +class TestFlavorPermissionRuleListObject( + test_objects._LocalTest, _TestFlavorPermissionRuleListObject): + pass + + +class TestRemoteFlavorPermissionRuleListObject( + test_objects._RemoteTest, _TestFlavorPermissionRuleListObject): + pass diff --git a/nova/tests/unit/objects/test_objects.py b/nova/tests/unit/objects/test_objects.py index f0f5321f76a..b1f71a30f16 100644 --- a/nova/tests/unit/objects/test_objects.py +++ b/nova/tests/unit/objects/test_objects.py @@ -1097,8 +1097,10 @@ def obj_name(cls): 'DiskMetadata': '1.0-e7a0f1ccccf10d26a76b28e7492f3788', 'EC2Ids': '1.0-474ee1094c7ec16f8ce657595d8c49d9', 'EC2InstanceMapping': '1.0-a4556eb5c5e94c045fe84f49cf71644f', - 'Flavor': '1.2-4ce99b41327bb230262e5a8f45ff0ce3', - 'FlavorList': '1.1-912b5ce24d48bc60cc9db96104f95581', + 'Flavor': '1.2-341b2de4d558bc4a426b0e34d2491bb9', + 'FlavorList': '1.1-e63c7fa460da017c13274b4253fa7b5b', + 'FlavorPermissionRule': '1.0-3514a71e53a3df7aac502ef1867ecfbb', + 'FlavorPermissionRuleList': '1.0-f9b0e3e518c3e6654a390e41b94aa6fb', 'HostMapping': '1.0-1a3390a696792a552ab7bd31a77ba9ac', 'HostMappingList': '1.1-18ac2bfb8c1eb5545bed856da58a79bc', 'HVSpec': '1.2-de06bcec472a2f04966b855a49c46b41', diff --git a/nova/tests/unit/policies/test_flavor_permission_rules.py b/nova/tests/unit/policies/test_flavor_permission_rules.py new file mode 100644 index 00000000000..17c76c0121c --- /dev/null +++ b/nova/tests/unit/policies/test_flavor_permission_rules.py @@ -0,0 +1,233 @@ +# 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. + +import ddt +import fixtures +from oslo_utils.fixture import uuidsentinel as uuids +import webob.exc + +from nova.api.openstack.compute import ( + flavor_permission_rules as fpr_api) +from nova import context as nova_context +from nova.objects import fields +from nova.policies import flavor_permission_rules as fpr_policies +from nova.tests.unit.api.openstack import fakes +from nova.tests.unit import fake_flavor_permission_rule as fake_fpr +from nova.tests.unit.policies import base + + +RULE_DOMAIN_ID = 'test-domain' +EFFECT_ALLOW = fields.FlavorPermissionRuleEffect.ALLOW +EFFECT_DENY = fields.FlavorPermissionRuleEffect.DENY + + +@ddt.ddt +class FlavorPermissionRulesPolicyTest(base.BasePolicyTest): + """Test flavor permission rules API policies with all possible contexts. + + Only admin contexts are authorized by default. + """ + + def setUp(self): + super().setUp() + self.controller = fpr_api.FlavorPermissionRulesController() + self.req = fakes.HTTPRequest.blank('') + # Create flavor permission rules with domain and project scope + ctx = self.project_admin_context + self.domain_rule = fake_fpr.fake_flavor_permission_rule_obj( + ctx, + fake_fpr.fake_db_flavor_permission_rule( + domain_id=RULE_DOMAIN_ID, project_id=None)) + self.project_rule = fake_fpr.fake_flavor_permission_rule_obj( + ctx, + fake_fpr.fake_db_flavor_permission_rule( + domain_id=RULE_DOMAIN_ID, project_id=self.project_id)) + # Mock the get_by_uuid method to return the domain rule by default + self.mock_get_rule = self.useFixture(fixtures.MockPatch( + 'nova.objects.FlavorPermissionRule.get_by_uuid')).mock + self.mock_get_rule.return_value = self.domain_rule + # Mock other FlavorPermissionRuleList and FlavorList methods + self.useFixture(fixtures.MockPatch( + 'nova.objects.FlavorPermissionRuleList.get_all', + return_value=[])) + self.useFixture(fixtures.MockPatch( + 'nova.objects.FlavorList.get_flavor_ids_by_ids', + return_value={})) + self.useFixture(fixtures.MockPatch( + 'nova.objects.FlavorPermissionRule.create')) + self.useFixture(fixtures.MockPatch( + 'nova.objects.FlavorPermissionRule.destroy')) + self.useFixture(fixtures.MockPatch( + 'nova.objects.FlavorPermissionRule.save')) + # With legacy rules and no scope check, all admin contexts can manage + # flavor permission rules. + self.admin_authorized_contexts = [ + self.legacy_admin_context, self.system_admin_context, + self.project_admin_context] + + def _scope_rule(self, scope: str): + return self.domain_rule if scope == 'domain' else self.project_rule + + def _authorized_contexts(self, scope: str) -> list: + return self.admin_authorized_contexts + + def _check_not_found_auth(self, authorized_contexts, func, + *args, **kwargs): + """Check a policy where unauthorized access raises HTTPNotFound""" + unauth = list(set(self.all_contexts) - set(authorized_contexts)) + for ctx in authorized_contexts: + self.req.environ['nova.context'] = ctx + func(self.req, *args, **kwargs) + for ctx in unauth: + self.req.environ['nova.context'] = ctx + self.assertRaises( + webob.exc.HTTPNotFound, func, self.req, *args, **kwargs) + + @ddt.data('all', 'domain', 'project') + def test_index(self, scope): + # Block access via the other scopes' policies to isolate the test scope + self.policy.set_rules({ + fpr_policies.POLICY_ROOT % f'index:{s}': '!' + for s in ('all', 'domain', 'project') if s != scope + }, overwrite=False) + # index:project is the terminal fatal check if index:all and + # index:domain are not authorized + rule_name = fpr_policies.POLICY_ROOT % 'index:project' + self.common_policy_auth( + self._authorized_contexts(scope), + rule_name, self.controller.index, self.req) + + @ddt.data('domain', 'project') + def test_show(self, scope): + rule = self._scope_rule(scope) + self.mock_get_rule.return_value = rule + self._check_not_found_auth( + self._authorized_contexts(scope), + self.controller.show, rule.uuid) + + @ddt.data('domain', 'project') + def test_create(self, scope): + rule_name = fpr_policies.POLICY_ROOT % f'create:{scope}' + data = {'domain_id': RULE_DOMAIN_ID, 'effect': EFFECT_ALLOW} + if scope == 'project': + data['project_id'] = self.project_id + self.common_policy_auth( + self._authorized_contexts(scope), + rule_name, self.controller.create, self.req, + body={'flavor_permission_rule': data}) + + @ddt.data('domain', 'project') + def test_delete(self, scope): + rule = self._scope_rule(scope) + self.mock_get_rule.return_value = rule + self._check_not_found_auth( + self._authorized_contexts(scope), + self.controller.delete, rule.uuid) + + @ddt.data('domain', 'project') + def test_update(self, scope): + rule = self._scope_rule(scope) + self.mock_get_rule.return_value = rule + body = {'flavor_permission_rule': {'effect': EFFECT_DENY}} + self._check_not_found_auth( + self._authorized_contexts(scope), + self.controller.update, rule.uuid, body=body) + + +class FlavorPermissionRulesNoLegacyPolicyTest( + FlavorPermissionRulesPolicyTest): + """Test flavor permission rules API policies without deprecated rules""" + + without_deprecated_rules = True + + +class FlavorPermissionRulesScopePolicyTest( + FlavorPermissionRulesPolicyTest): + """Test flavor permission rules API policies with scope enforcement""" + + def setUp(self): + super().setUp() + self.flags(enforce_scope=True, group="oslo_policy") + # system_admin_context is unauthorized because it is system scoped and + # the policies require project scope + self.admin_authorized_contexts = [ + self.legacy_admin_context, + self.project_admin_context] + + +class FlavorPermissionRulesScopeNoLegacyPolicyTest( + FlavorPermissionRulesScopePolicyTest): + """Test flavor permission rules API policies with scope enforcement and + without deprecated rules + """ + + without_deprecated_rules = True + + +class FlavorPermissionRulesAdminOverridePolicyTest( + FlavorPermissionRulesPolicyTest): + """Test flavor permission rules API policies with custom overrides for + delegating flavor permission rule management to scope-based global_admin, + domain_admin and project_admin roles. + """ + + def setUp(self): + super().setUp() + self.flags(enforce_scope=True, group="oslo_policy") + # Create domain and project admin context + self.domain_admin_context = nova_context.RequestContext( + user_id='domain_admin', + project_id=uuids.domain_admin_project, + project_domain_id=RULE_DOMAIN_ID, + roles=['domain_admin']) + self.all_contexts.add(self.domain_admin_context) + self.project_custom_context = nova_context.RequestContext( + user_id='project_admin_user', + project_id=self.project_id, + project_domain_id=RULE_DOMAIN_ID, + roles=['project_admin']) + self.all_contexts.add(self.project_custom_context) + self.global_admin_context = nova_context.RequestContext( + user_id='global_admin', + project_id=uuids.global_admin_project, + project_domain_id=uuids.global_admin_domain, + roles=['global_admin']) + self.all_contexts.add(self.global_admin_context) + # Override the default policy rules to restrict access to the custom + # role of the respective scope + domain_admin_check = ( + 'role:domain_admin and project_domain_id:%(project_domain_id)s') + project_admin_check = ( + 'role:project_admin and project_id:%(project_id)s') + self.policy.set_rules({ + **{fpr_policies.POLICY_ROOT % rule: domain_admin_check + for rule in [ + 'show:domain', 'create:domain', + 'delete:domain', 'update:domain']}, + **{fpr_policies.POLICY_ROOT % rule: project_admin_check + for rule in [ + 'show:project', 'create:project', + 'delete:project', 'update:project']}, + fpr_policies.POLICY_ROOT % 'index:all': 'role:global_admin', + fpr_policies.POLICY_ROOT % 'index:domain': 'role:domain_admin', + fpr_policies.POLICY_ROOT % 'index:project': 'role:project_admin', + }, overwrite=False) + + def _authorized_contexts(self, scope: str) -> list: + return { + 'all': [self.global_admin_context], + 'domain': [self.domain_admin_context], + 'project': [self.project_custom_context], + }[scope]