Implement project personas for snapshot metadata

This commit updates the policies to use the default roles provided by
keystone allowing for a more consistent authorization experience across
OpenStack services. This commit also includes read-only support for read
APIs.

Also removes the following from tests/unit/policy.yaml:
-"volume:get_snapshot_metadata": ""
-"volume:update_snapshot_metadata": ""
-"volume:delete_snapshot_metadata": ""

Co-Authored-by: Alan Bishop <abishop@redhat.com>

Change-Id: I99b44c87d2061524aa2b806da9b091b5602d726c

cinder/policies/snapshot_metadata.py

@@ -22,11 +22,24 @@ GET_POLICY = 'volume:get_snapshot_metadata'
 DELETE_POLICY = 'volume:delete_snapshot_metadata'
 UPDATE_POLICY = 'volume:update_snapshot_metadata'
 
+deprecated_get_snapshot_metadata = base.CinderDeprecatedRule(
+    name=GET_POLICY,
+    check_str=base.RULE_ADMIN_OR_OWNER
+)
+deprecated_update_snapshot_metadata = base.CinderDeprecatedRule(
+    name=UPDATE_POLICY,
+    check_str=base.RULE_ADMIN_OR_OWNER
+)
+deprecated_delete_snapshot_metadata = base.CinderDeprecatedRule(
+    name=DELETE_POLICY,
+    check_str=base.RULE_ADMIN_OR_OWNER
+)
+
 
 snapshot_metadata_policies = [
     policy.DocumentedRuleDefault(
         name=GET_POLICY,
-        check_str=base.RULE_ADMIN_OR_OWNER,
+        check_str=base.SYSTEM_READER_OR_PROJECT_READER,
         description="Show snapshot's metadata or one specified metadata "
                     "with a given key.",
         operations=[
@@ -38,25 +51,29 @@ snapshot_metadata_policies = [
                 'method': 'GET',
                 'path': '/snapshots/{snapshot_id}/metadata/{key}'
             }
-        ]),
+        ],
+        deprecated_rule=deprecated_get_snapshot_metadata,
+    ),
     policy.DocumentedRuleDefault(
         name=UPDATE_POLICY,
-        check_str=base.RULE_ADMIN_OR_OWNER,
+        check_str=base.SYSTEM_ADMIN_OR_PROJECT_MEMBER,
         description="Update snapshot's metadata or one specified "
                     "metadata with a given key.",
         operations=[
             {
-                'method': 'PUT',
+                'method': 'POST',
                 'path': '/snapshots/{snapshot_id}/metadata'
             },
             {
                 'method': 'PUT',
                 'path': '/snapshots/{snapshot_id}/metadata/{key}'
             }
-        ]),
+        ],
+        deprecated_rule=deprecated_update_snapshot_metadata,
+    ),
     policy.DocumentedRuleDefault(
         name=DELETE_POLICY,
-        check_str=base.RULE_ADMIN_OR_OWNER,
+        check_str=base.SYSTEM_ADMIN_OR_PROJECT_MEMBER,
         description="Delete snapshot's specified metadata "
                     "with a given key.",
         operations=[
@@ -64,7 +81,9 @@ snapshot_metadata_policies = [
                 'method': 'DELETE',
                 'path': '/snapshots/{snapshot_id}/metadata/{key}'
             }
-        ]),
+        ],
+        deprecated_rule=deprecated_delete_snapshot_metadata,
+    ),
 ]
 
 

cinder/tests/unit/policies/test_snapshot_metadata.py

@@ -0,0 +1,202 @@
+# Copyright 2021 Red Hat, Inc.
+# 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
+
+from cinder.api import microversions as mv
+from cinder.api.v3 import snapshot_metadata
+from cinder import exception
+from cinder.policies import snapshot_metadata as policy
+from cinder.policies import snapshots as snapshots_policy
+from cinder.tests.unit.api import fakes as fake_api
+from cinder.tests.unit.policies import base
+from cinder.tests.unit import utils as test_utils
+
+
+@ddt.ddt
+class SnapshotMetadataPolicyTest(base.BasePolicyTest):
+    authorized_readers = [
+        'legacy_admin',
+        'legacy_owner',
+        'system_admin',
+        'project_admin',
+        'project_member',
+        'project_reader',
+        'project_foo',
+    ]
+
+    unauthorized_readers = [
+        'system_member',
+        'system_reader',
+        'system_foo',
+        'other_project_member',
+        'other_project_reader',
+    ]
+
+    authorized_members = [
+        'legacy_admin',
+        'legacy_owner',
+        'system_admin',
+        'project_admin',
+        'project_member',
+        'project_reader',
+        'project_foo',
+    ]
+
+    unauthorized_members = [
+        'system_member',
+        'system_reader',
+        'system_foo',
+        'other_project_member',
+        'other_project_reader',
+    ]
+
+    # DB validations will throw SnapshotNotFound for some contexts
+    unauthorized_exceptions = [
+        exception.SnapshotNotFound,
+    ]
+
+    # Basic policy test is without enforcing scope (which cinder doesn't
+    # yet support) and deprecated rules enabled.
+    def setUp(self, enforce_scope=False, enforce_new_defaults=False,
+              *args, **kwargs):
+        super().setUp(enforce_scope, enforce_new_defaults, *args, **kwargs)
+        self.controller = snapshot_metadata.Controller()
+        self.api_path = '/v3/%s/snapshots' % (self.project_id)
+        self.api_version = mv.BASE_VERSION
+        self.vol_type = test_utils.create_volume_type(
+            self.project_admin_context,
+            name='fake_vol_type', testcase_instance=self)
+        # Relax the snapshots GET_POLICY in order to get past that check.
+        self.policy.set_rules({snapshots_policy.GET_POLICY: ""},
+                              overwrite=False)
+
+    def _create_volume(self, **kwargs):
+        volume = test_utils.create_volume(self.project_member_context,
+                                          volume_type_id=self.vol_type.id,
+                                          testcase_instance=self, **kwargs)
+        return volume
+
+    def _create_snapshot(self, **kwargs):
+        volume = self._create_volume(**kwargs)
+        snapshot = test_utils.create_snapshot(self.project_member_context,
+                                              volume_id=volume.id,
+                                              testcase_instance=self, **kwargs)
+        return snapshot
+
+    @ddt.data(*base.all_users)
+    def test_get_policy(self, user_id):
+        metadata = {'inside': 'out'}
+        snapshot = self._create_snapshot(metadata=metadata)
+        rule_name = policy.GET_POLICY
+        url = '%s/%s/metadata' % (self.api_path, snapshot.id)
+        req = fake_api.HTTPRequest.blank(url, version=self.api_version)
+
+        response = self.common_policy_check(
+            user_id, self.authorized_readers,
+            self.unauthorized_readers,
+            self.unauthorized_exceptions,
+            rule_name, self.controller.show,
+            req, snapshot_id=snapshot.id, id='inside')
+
+        if user_id in self.authorized_readers:
+            self.assertDictEqual(metadata, response['meta'])
+
+    @ddt.data(*base.all_users)
+    def test_update_policy(self, user_id):
+        snapshot = self._create_snapshot()
+        rule_name = policy.UPDATE_POLICY
+        url = '%s/%s/metadata' % (self.api_path, snapshot.id)
+        req = fake_api.HTTPRequest.blank(url, version=self.api_version)
+        req.method = 'POST'
+        metadata = {
+            'inside': 'out',
+            'outside': 'in'
+        }
+        body = {
+            "metadata": {**metadata}
+        }
+
+        response = self.common_policy_check(
+            user_id, self.authorized_members,
+            self.unauthorized_members,
+            self.unauthorized_exceptions,
+            rule_name, self.controller.update_all,
+            req, snapshot_id=snapshot.id, body=body)
+
+        if user_id in self.authorized_members:
+            self.assertDictEqual(metadata, response['metadata'])
+
+    @ddt.data(*base.all_users)
+    def test_delete_policy(self, user_id):
+        metadata = {'inside': 'out'}
+        snapshot = self._create_snapshot(metadata=metadata)
+        rule_name = policy.DELETE_POLICY
+        url = '%s/%s/metadata/inside' % (self.api_path, snapshot.id)
+        req = fake_api.HTTPRequest.blank(url, version=self.api_version)
+        req.method = 'DELETE'
+
+        # Relax the GET_POLICY in order to get past that check.
+        self.policy.set_rules({policy.GET_POLICY: ""},
+                              overwrite=False)
+
+        self.common_policy_check(user_id, self.authorized_members,
+                                 self.unauthorized_members,
+                                 self.unauthorized_exceptions,
+                                 rule_name, self.controller.delete, req,
+                                 snapshot_id=snapshot.id, id='inside')
+
+
+class SnapshotMetadataPolicySecureRbacTest(SnapshotMetadataPolicyTest):
+    authorized_readers = [
+        'legacy_admin',
+        'system_admin',
+        'project_admin',
+        'project_member',
+        'project_reader',
+    ]
+
+    unauthorized_readers = [
+        'legacy_owner',
+        'system_member',
+        'system_reader',
+        'system_foo',
+        'project_foo',
+        'other_project_member',
+        'other_project_reader',
+    ]
+
+    authorized_members = [
+        'legacy_admin',
+        'system_admin',
+        'project_admin',
+        'project_member',
+    ]
+
+    unauthorized_members = [
+        'legacy_owner',
+        'system_member',
+        'system_reader',
+        'system_foo',
+        'project_reader',
+        'project_foo',
+        'other_project_member',
+        'other_project_reader',
+    ]
+
+    def setUp(self, *args, **kwargs):
+        # Test secure RBAC by disabling deprecated policy rules (scope
+        # is still not enabled).
+        super().setUp(enforce_scope=False, enforce_new_defaults=True,
+                      *args, **kwargs)

cinder/tests/unit/policy.yaml

@@ -4,21 +4,6 @@
 # Default rule for most Admin APIs.
 "admin_api": "is_admin:True"
 
-# Show snapshot's metadata or one specified metadata with a given key.
-# GET  /snapshots/{snapshot_id}/metadata
-# GET  /snapshots/{snapshot_id}/metadata/{key}
-"volume:get_snapshot_metadata": ""
-
-# Update snapshot's metadata or one specified metadata with a given
-# key.
-# PUT  /snapshots/{snapshot_id}/metadata
-# PUT  /snapshots/{snapshot_id}/metadata/{key}
-"volume:update_snapshot_metadata": ""
-
-# Delete snapshot's specified metadata with a given key.
-# DELETE  /snapshots/{snapshot_id}/metadata/{key}
-"volume:delete_snapshot_metadata": ""
-
 # Reset status of group snapshot.
 # POST  /group_snapshots/{g_snapshot_id}/action (reset_status)
 "group:reset_group_snapshot_status": ""