Set system_scope='all' in elevated context

In case when enforce_new_defaults is set to True and new policy rules
are used, context.is_admin flag isn't really working as it was with old
rules.
But in case when elevated context is needed, it means that we need
context which has full rights to the system. So we should also set
"system_scope" parameter to "all" to be sure that system scope queries
can be done with such elevated context always.

It is needed e.g. when elevated context is used to get some data from
db. In such case we need to have db query which will not be scoped to
the single project_id and with new defaults to achieve that system_scope
has to be set to "all".

Closes-Bug: #1920001
Change-Id: I9bbf9ebe4252b2aa4b2a185bdc040ac8bf9753c7
This commit is contained in:
Slawek Kaplonski
2021-03-19 12:02:35 +01:00
parent d73449dd1c
commit ff96659ef0
2 changed files with 15 additions and 0 deletions

View File

@@ -17,6 +17,7 @@ import copy
import datetime
import warnings
from oslo_config import cfg
from oslo_context import context as oslo_context
from oslo_db.sqlalchemy import enginefacade
@@ -124,6 +125,9 @@ class ContextBase(oslo_context.RequestContext):
if 'admin' not in [x.lower() for x in context.roles]:
context.roles = context.roles + ["admin"]
if cfg.CONF.oslo_policy.enforce_new_defaults:
context.system_scope = 'all'
return context

View File

@@ -12,6 +12,7 @@
from unittest import mock
from oslo_config import cfg
from oslo_context import context as oslo_context
from testtools import matchers
@@ -152,6 +153,16 @@ class TestNeutronContext(_base.BaseTestCase):
elevated2_ctx = elevated_ctx.elevated()
self.assertTrue(elevated2_ctx.is_admin)
def test_neutron_context_elevated_system_scope_for_new_policies(self):
cfg.CONF.set_override(
'enforce_new_defaults', True, group='oslo_policy')
ctx = context.Context('user_id', 'tenant_id')
self.assertFalse(ctx.is_admin)
self.assertNotEqual('all', ctx.system_scope)
elevated_ctx = ctx.elevated()
self.assertTrue(elevated_ctx.is_admin)
self.assertEqual('all', elevated_ctx.system_scope)
def test_neutron_context_overwrite(self):
ctx1 = context.Context('user_id', 'tenant_id')
self.assertEqual(ctx1.request_id,