openstack/neutron-lib
Code Issues Proposed changes

Add "has_global_access" attribute to the context object

Browse Source 
In case when API policies with custom roles has to be defined by the
operator and such custom role should have granted access to the
resources from all projects, like for example some kind of
"admin_reader" or "auditor" role, it was not possible to achieve so far.
The problem was that for all non-admin and not service users, SQL
queries were scoped to the own project only always so such "auditor"
couldn't even get data from different projects from the database.

This patch introduces new API policy rule called
`context_with_global_access` and attribute `has_global_access` to the
neutron_lib.context.ContextBase class.
By default `context_with_global_access` rule is granted to nobody but it
can be defined in the neutron policy file like e.g.:

    "context_with_global_access": "role:auditor"

and then `neutron_context` object for API requests made by someone with
such role granted will be able to fetch all data from the database.

This doesn't mean that anyone with such role will be able to do or get
everything through the API because there is still policy engine with
defined API policies which prevents that.
So to e.g. grant such auditor user permission to list all networks in
the cluster, additional rule would be needed in policy file and it can
looks for example like:

    "get_network": "role:admin_only) or
                    (role:reader and project_id:%(project_id)s) or
                    rule:shared or rule:external or
                    rule:context_is_advsvc or
                    role:auditor"

Closes-Bug: #2115184
Change-Id: I90149b0212dafa8f469dc329cc4b45042cded38c
Signed-off-by: Slawek Kaplonski <skaplons@redhat.com>
This commit is contained in:
Slawek Kaplonski
2025-07-03 16:40:44 +02:00
parent d1a6237259
commit b510f7feb8
5 changed files with 65 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
neutron_lib/context.py
View File

@@ -36,7 +36,7 @@ class ContextBase(oslo_context.RequestContext):
def __init__(self, user_id=None, project_id=None, is_admin=None,
timestamp=None, project_name=None, user_name=None,
is_advsvc=None, tenant_id=None, tenant_name=None,
**kwargs):
has_global_access=False, **kwargs):
# NOTE(jamielennox): We maintain this argument order for tests that
# pass arguments positionally.
@@ -54,6 +54,7 @@ class ContextBase(oslo_context.RequestContext):
'project_name instead')
kwargs.setdefault('project_id', project_id)
kwargs.setdefault('project_name', project_name)
self._has_global_access = has_global_access
super().__init__(
is_admin=is_admin, user_id=user_id, **kwargs)
@@ -66,6 +67,9 @@ class ContextBase(oslo_context.RequestContext):
self._is_service_role = policy_engine.check_is_service_role(self)
if self.is_admin is None:
self.is_admin = policy_engine.check_is_admin(self)
if not self._has_global_access:
self._has_global_access = policy_engine.check_has_global_access(
self)
@property
def tenant_id(self):
@@ -100,6 +104,10 @@ class ContextBase(oslo_context.RequestContext):
"Please use method 'is_service_role' instead.")
return self.is_service_role
@property
def has_global_access(self):
return self.is_admin or self._has_global_access
def to_dict(self):
context = super().to_dict()
context.update({
@@ -110,6 +118,7 @@ class ContextBase(oslo_context.RequestContext):
'tenant_name': self.project_name,
'project_name': self.project_name,
'user_name': self.user_name,
'has_global_access': self.has_global_access,
})
return context
@@ -117,6 +126,7 @@ class ContextBase(oslo_context.RequestContext):
values = super().to_policy_values()
values['tenant_id'] = self.project_id
values['is_admin'] = self.is_admin
values['has_global_access'] = self.has_global_access
# NOTE(jamielennox): These are almost certainly unused and non-standard
# but kept for backwards compatibility. Remove them in Pike

11
neutron_lib/db/utils.py
View File

@@ -169,8 +169,8 @@ def model_query_scope_is_project(context, model):
:param context: The context to check for admin and advsvc rights.
:param model: The model to check the project_id of.
:returns: True if the context is not admin and not advsvc and the model
has a project_id. False otherwise.
:returns: True if the context has no global access and is not advsvc
and the model has a project_id. False otherwise.
"""
if not hasattr(model, 'project_id'):
# If model doesn't have project_id, there is no need to scope query to
@@ -180,9 +180,10 @@ def model_query_scope_is_project(context, model):
# For context which has 'advanced-service' rights the
# query will not be scoped to a single project_id
return False
# Unless context has 'admin' rights the
# query will be scoped to a single project_id
return not context.is_admin
# Unless context has 'global' access the
# resources from the database query will be scoped to a single project_id
# context with 'admin' rights is treated as it has global access always.
return not context.has_global_access
def model_query(context, model):

21
neutron_lib/policy/_engine.py
View File

@@ -19,6 +19,7 @@ from oslo_policy import policy
_ROLE_ENFORCER = None
_ADMIN_CTX_POLICY = 'context_is_admin'
_GLOBAL_CTX_POLICY = 'context_with_global_access'
_ADVSVC_CTX_POLICY = 'context_is_advsvc'
_SERVICE_ROLE = 'service_api'
@@ -31,6 +32,16 @@ _BASE_RULES = [
_ADMIN_CTX_POLICY,
'role:admin',
description='Rule for cloud admin access'),
policy.RuleDefault(
# By default, no one has global access to the resources.
# That is special meaning of the "!" in rule, see
# https://docs.openstack.org/oslo.policy/latest/admin/policy-yaml-file.html#examples.
# This policy rule should be overridden by the cloud administrator if
# there is need to have any custom role with global access to the
# resources from all projects.
_GLOBAL_CTX_POLICY,
'!',
description='Rule for context with global access to the resources'),
policy.RuleDefault(
_ADVSVC_CTX_POLICY,
'role:advsvc',
@@ -84,6 +95,16 @@ def check_is_admin(context):
return _check_rule(context, _ADMIN_CTX_POLICY)
def check_has_global_access(context):
"""Verify context has rights to fetch resources no matter of the owner
:param context: The context object.
:returns: True if the context has global rights (as per the global
enforcer) and False otherwise.
"""
return _check_rule(context, _GLOBAL_CTX_POLICY)
def check_is_advsvc(context):
"""Verify context has advsvc rights according to global policy settings.

17
neutron_lib/tests/unit/test_context.py
View File

@@ -116,11 +116,13 @@ class TestNeutronContext(_base.BaseTestCase):
ctx = context.Context('user_id', 'project_id', is_advsvc=True)
self.assertFalse(ctx.is_admin)
self.assertTrue(ctx.is_advsvc)
self.assertFalse(ctx.has_global_access)
def test_neutron_context_create_is_service_role(self):
ctx = context.Context('user_id', 'project_id', roles=['service'])
self.assertFalse(ctx.is_admin)
self.assertTrue(ctx.is_service_role)
self.assertFalse(ctx.has_global_access)
def test_neutron_context_create_with_auth_token(self):
ctx = context.Context('user_id', 'project_id',
@@ -223,6 +225,7 @@ class TestNeutronContext(_base.BaseTestCase):
self.assertIsNone(ctx_dict['tenant_id'])
self.assertIsNone(ctx_dict['auth_token'])
self.assertTrue(ctx_dict['is_admin'])
self.assertTrue(ctx_dict['has_global_access'])
self.assertIn('admin', ctx_dict['roles'])
self.assertIsNotNone(ctx.session)
self.assertNotIn('session', ctx_dict)
@@ -270,6 +273,7 @@ class TestNeutronContext(_base.BaseTestCase):
self.assertNotEqual('all', ctx.system_scope)
elevated_ctx = ctx.elevated()
self.assertTrue(elevated_ctx.is_admin)
self.assertTrue(elevated_ctx.has_global_access)
for expected_role in expected_roles:
self.assertIn(expected_role, elevated_ctx.roles)
# make sure we do not set the system scope in context
@@ -280,6 +284,7 @@ class TestNeutronContext(_base.BaseTestCase):
custom_roles = ['custom_role']
ctx = context.Context('user_id', 'project_id', roles=custom_roles)
self.assertFalse(ctx.is_admin)
self.assertFalse(ctx.has_global_access)
self.assertNotEqual('all', ctx.system_scope)
for expected_admin_role in expected_admin_roles:
self.assertNotIn(expected_admin_role, ctx.roles)
@@ -288,6 +293,7 @@ class TestNeutronContext(_base.BaseTestCase):
elevated_ctx = ctx.elevated()
self.assertTrue(elevated_ctx.is_admin)
self.assertTrue(elevated_ctx.has_global_access)
for expected_admin_role in expected_admin_roles:
self.assertIn(expected_admin_role, elevated_ctx.roles)
for custom_role in custom_roles:
@@ -319,6 +325,17 @@ class TestNeutronContext(_base.BaseTestCase):
self.assertEqual(req_id_before, oslo_context.get_current().request_id)
self.assertNotEqual(req_id_before, ctx_admin.request_id)
def test_neutron_context_has_global_access(self):
with mock.patch('neutron_lib.policy._engine.check_has_global_access',
return_value=False):
ctx = context.Context('user_id', 'project_id')
self.assertFalse(ctx.has_global_access)
with mock.patch('neutron_lib.policy._engine.check_has_global_access',
return_value=True):
ctx = context.Context('user_id', 'project_id')
self.assertTrue(ctx.has_global_access)
def test_to_policy_values(self):
values = {
'user_id': 'user_id',

10
releasenotes/notes/add-has_global_access-to-the-context-object-672af662b46be0a3.yaml Normal file
View File

@@ -0,0 +1,10 @@
---
features:
- |
New attribute ``has_global_access`` is added to the context object. Value of
this attribute is set based on the API policy rule
``context_with_global_access`` and should be used in case when there are
custom roles with access to the resources from all projects. For example,
``auditor`` role which should have read only access to all of the resources
in the cloud.
By default ``context.has_global_access`` is granted to no one.