Add is_service_role property to the context class

As we are moving to the new S-RBAC policies, we want to use "service"
role for all service to service communication. See [1] for details.

This require from Context class property similar to old "is_advsvc" but
with new naming convention and using new policy rule.

This patch adds this new property together with all required policies
and rules.
For now "ContextBase.is_advsvc" property will return True if one of the
advsvc OR service_role will be True to make it working in the same way
with both old and new policies but once we will get rid of the old
policies we should also remove is_advsvc property from the ContextBase
class.

[1] https://governance.openstack.org/tc/goals/selected/consistent-and-secure-rbac.html#phase-2

Change-Id: Ic401db8b4e2745234e61fe2c05afd5b4ab719a03

neutron_lib/context.py

@@ -18,10 +18,13 @@ import datetime
 
 from oslo_context import context as oslo_context
 from oslo_db.sqlalchemy import enginefacade
+from oslo_log import log as logging
 
 from neutron_lib.db import api as db_api
 from neutron_lib.policy import _engine as policy_engine
 
+LOG = logging.getLogger(__name__)
+
 
 class ContextBase(oslo_context.RequestContext):
     """Security context and request information.
@@ -47,10 +50,11 @@ class ContextBase(oslo_context.RequestContext):
         if not timestamp:
             timestamp = datetime.datetime.utcnow()
         self.timestamp = timestamp
-        self.is_advsvc = is_advsvc
-        if self.is_advsvc is None:
-            self.is_advsvc = (self.is_admin or
-                              policy_engine.check_is_advsvc(self))
+        self._is_advsvc = is_advsvc
+        if self._is_advsvc is None:
+            self._is_advsvc = (self.is_admin or
+                               policy_engine.check_is_advsvc(self))
+        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)
 
@@ -70,6 +74,23 @@ class ContextBase(oslo_context.RequestContext):
     def tenant_name(self, tenant_name):
         self.project_name = tenant_name
 
+    @property
+    def is_service_role(self):
+        # TODO(slaweq): we will not need to check ContextBase._is_advsvc once
+        # we will get rid of the old API policies
+        return self._is_service_role or self._is_advsvc
+
+    @property
+    def is_advsvc(self):
+        # TODO(slaweq): this property should be removed once we will get rid
+        # of old policy rules and only new, S-RBAC rules will be available as
+        # then we will use ContextBase.is_service_role instead
+        LOG.warning("Method 'is_advsvc' is deprecated since 2023.2 release "
+                    "(neutron-lib 3.8.0) and will be removed once support for "
+                    "the old RBAC API policies will be removed from Neutron. "
+                    "Please use method 'is_service_role' instead.")
+        return self.is_service_role
+
     def to_dict(self):
         context = super().to_dict()
         context.update({

neutron_lib/db/utils.py

@@ -176,7 +176,7 @@ def model_query_scope_is_project(context, model):
         # If model doesn't have project_id, there is no need to scope query to
         # just one project
         return False
-    if context.is_advsvc:
+    if context.is_service_role:
         # For context which has 'advanced-service' rights the
         # query will not be scoped to a single project_id
         return False

neutron_lib/policy/__init__.py

@@ -23,5 +23,6 @@ RULE_ADMIN_OR_OWNER = 'rule:admin_or_owner'
 RULE_ADMIN_ONLY = 'rule:admin_only'
 RULE_ANY = 'rule:regular_user'
 RULE_ADVSVC = 'rule:context_is_advsvc'
+RULE_SERVICE_ROLE = 'rule:service_api'
 RULE_ADMIN_OR_NET_OWNER = 'rule:admin_or_network_owner'
 RULE_ADMIN_OR_PARENT_OWNER = 'rule:admin_or_ext_parent_owner'

neutron_lib/policy/_engine.py

@@ -20,6 +20,7 @@ from oslo_policy import policy
 _ROLE_ENFORCER = None
 _ADMIN_CTX_POLICY = 'context_is_admin'
 _ADVSVC_CTX_POLICY = 'context_is_advsvc'
+_SERVICE_ROLE = 'service_api'
 
 
 # TODO(gmann): Remove setting the default value of config policy_file
@@ -38,6 +39,10 @@ _BASE_RULES = [
         _ADVSVC_CTX_POLICY,
         'role:advsvc',
         description='Rule for advanced service role access'),
+    policy.RuleDefault(
+        _SERVICE_ROLE,
+        'role:service',
+        description='Default rule for the service-to-service APIs.'),
 ]
 
 
@@ -91,6 +96,16 @@ def check_is_advsvc(context):
     return _check_rule(context, _ADVSVC_CTX_POLICY)
 
 
+def check_is_service_role(context):
+    """Verify context is service role according to global policy settings.
+
+    :param context: The context object.
+    :returns: True if the context is service role (as per the global
+    enforcer) and False otherwise.
+    """
+    return _check_rule(context, _SERVICE_ROLE)
+
+
 def list_rules():
     return _BASE_RULES
 

neutron_lib/tests/etc/dummy_policy.yaml

@@ -1,2 +1,3 @@
 "context_is_admin": "role:dummy"
 "context_is_advsvc": "role:dummy"
+"service_api": "role:dummy"

neutron_lib/tests/etc/policy.yaml

@@ -1,3 +1,4 @@
 "context_is_admin": "role:admin"
 "context_is_advsvc": "role:advsvc"
+"service_api": "role:service"
 "default": "rule:admin_or_owner"

neutron_lib/tests/unit/api/test_attributes.py

@@ -322,6 +322,16 @@ class TestValidatePriviliges(base.BaseTestCase):
         except exc.HTTPBadRequest:
             self.fail("HTTPBadRequest exception should not be raised.")
 
+    def test__validate_privileges_service_role_other_tenant(self):
+        project_id = 'fake_project'
+        ctx = context.Context(project_id='fake_project2',
+                              roles=['service'])
+        res_dict = {'project_id': project_id}
+        try:
+            attributes._validate_privileges(ctx, res_dict)
+        except exc.HTTPBadRequest:
+            self.fail("HTTPBadRequest exception should not be raised.")
+
 
 class TestRetrieveValidSortKeys(base.BaseTestCase):
 

neutron_lib/tests/unit/db/test_utils.py

@@ -124,6 +124,21 @@ class TestUtilsLegacyPolicies(base.BaseTestCase):
         self.assertFalse(
             utils.model_query_scope_is_project(ctx, model))
 
+    def test_model_query_scope_is_project_service_role(self):
+        ctx = context.Context(
+            project_id='some project',
+            is_admin=False,
+            roles=['service'])
+        model = mock.Mock(project_id='project')
+
+        self.assertFalse(
+            utils.model_query_scope_is_project(ctx, model))
+
+        # Ensure that project_id isn't mocked
+        del model.project_id
+        self.assertFalse(
+            utils.model_query_scope_is_project(ctx, model))
+
     def test_model_query_scope_is_project_regular_user(self):
         ctx = context.Context(
             project_id='some project',

neutron_lib/tests/unit/policy/test__engine.py

@@ -72,3 +72,27 @@ class TestPolicyEnforcer(base.BaseTestCase):
         policy_engine.init(policy_file='no_policy.yaml')
         ctx = context.Context('me', 'my_project', roles=['advsvc'])
         self.assertTrue(policy_engine.check_is_advsvc(ctx))
+
+    def test_check_is_service_role(self):
+        ctx = context.Context('me', 'my_project', roles=['service'])
+        self.assertTrue(policy_engine.check_is_service_role(ctx))
+
+    def test_check_is_not_service_role_user(self):
+        ctx = context.Context('me', 'my_project', roles=['member'])
+        self.assertFalse(policy_engine.check_is_service_role(ctx))
+
+    def test_check_is_not_service_role_admin(self):
+        ctx = context.Context('me', 'my_project').elevated()
+        self.assertTrue(policy_engine.check_is_admin(ctx))
+        self.assertFalse(policy_engine.check_is_service_role(ctx))
+
+    def test_check_is_service_role_no_roles_no_service_role(self):
+        policy_engine.init(policy_file='dummy_policy.yaml')
+        ctx = context.Context('me', 'my_project', roles=['service'])
+        # No service role in the policy file, so cannot assume the role.
+        self.assertFalse(policy_engine.check_is_service_role(ctx))
+
+    def test_check_is_service_role_with_default_policy(self):
+        policy_engine.init(policy_file='no_policy.yaml')
+        ctx = context.Context('me', 'my_project', roles=['service'])
+        self.assertTrue(policy_engine.check_is_service_role(ctx))

neutron_lib/tests/unit/test_context.py

@@ -76,6 +76,11 @@ class TestNeutronContext(_base.BaseTestCase):
         self.assertFalse(ctx.is_admin)
         self.assertTrue(ctx.is_advsvc)
 
+    def test_neutron_context_create_is_service_role(self):
+        ctx = context.Context('user_id', 'tenant_id', roles=['service'])
+        self.assertFalse(ctx.is_admin)
+        self.assertTrue(ctx.is_service_role)
+
     def test_neutron_context_create_with_auth_token(self):
         ctx = context.Context('user_id', 'tenant_id',
                               auth_token='auth_token_xxx')

releasenotes/notes/service_role-added-to-context-class-4b9583aaaabd37e0.yaml

@@ -0,0 +1,11 @@
+---
+features:
+  - |
+    New attribute ``is_service_role`` is added to the
+    ``neutron_lib.context.ContextBase`` class. This attribute indicates if the
+    context belongs to the service user which is used in the new secure RBAC
+    policies for service to service communication.
+deprecations:
+  - |
+    Atrribute ``is_advscv`` from the ``neutron_lib.context.ContextBase`` class
+    is deprecated and ``is_service_role`` should be used instead.