openstack/nova
Code Issues Proposed changes

Move policy enforcement into REST API layer for v2.1 api volume_attachment

Browse Source 
This patch moves policy enforement into REST API layer
for v2.1 api volume_attachment, and adds unit tests.

Partially implements blueprint v3-api-policy

Change-Id: Ia069d12f5fb9d6efb22e14a8656dc913d5a23cb6
This commit is contained in:
yuntong 2015-02-26 16:49:29 +08:00 committed by He Jie Xu
parent 973ca44db7
commit 73ff0edb2a
2 changed files with 80 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
nova/api/openstack/compute/plugins/v3/volumes.py
View File

@ -30,9 +30,8 @@ from nova import objects
from nova import volume
ALIAS = "os-volumes"
authorize = extensions.extension_authorizer('compute', 'v3:' + ALIAS)
authorize_attach = extensions.extension_authorizer('compute',
'v3:os-volumes-attachments')
authorize = extensions.os_compute_authorizer(ALIAS)
authorize_attach = extensions.os_compute_authorizer('os-volumes-attachments')
def _translate_volume_detail_view(context, vol):
@ -217,7 +216,7 @@ class VolumeAttachmentController(wsgi.Controller):
"""
def __init__(self):
self.compute_api = compute.API()
self.compute_api = compute.API(skip_policy_check=True)
self.volume_api = volume.API()
super(VolumeAttachmentController, self).__init__()

77
nova/tests/unit/api/openstack/compute/contrib/test_volumes.py
View File

@ -856,3 +856,80 @@ class TestAssistedVolumeSnapshotsPolicyEnforcementV21(test.NoDBTestCase):
self.assertEqual(
"Policy doesn't allow %s to be performed." % rule_name,
exc.format_message())
class TestVolumeAttachPolicyEnforcementV21(test.NoDBTestCase):
def setUp(self):
super(TestVolumeAttachPolicyEnforcementV21, self).setUp()
self.controller = volumes_v21.VolumeAttachmentController()
self.req = fakes.HTTPRequest.blank('')
def _common_policy_check(self, rules, rule_name, func, *arg, **kwarg):
self.policy.set_rules(rules)
exc = self.assertRaises(
exception.PolicyNotAuthorized, func, *arg, **kwarg)
self.assertEqual(
"Policy doesn't allow %s to be performed." % rule_name,
exc.format_message())
def test_index_volume_attach_policy_failed(self):
rule_name = "compute_extension:v3:os-volumes-attachments:index"
rules = {rule_name: "project:non_fake"}
self._common_policy_check(rules, rule_name,
self.controller.index, self.req, FAKE_UUID)
def test_show_volume_attach_policy_failed(self):
rule_name = "compute_extension:v3:os-volumes"
rules = {"compute_extension:v3:os-volumes-attachments:show": "@",
rule_name: "project:non_fake"}
self._common_policy_check(rules, rule_name, self.controller.show,
self.req, FAKE_UUID, FAKE_UUID_A)
rule_name = "compute_extension:v3:os-volumes-attachments:show"
rules = {"compute_extension:v3:os-volumes": "@",
rule_name: "project:non_fake"}
self._common_policy_check(rules, rule_name, self.controller.show,
self.req, FAKE_UUID, FAKE_UUID_A)
def test_create_volume_attach_policy_failed(self):
rule_name = "compute_extension:v3:os-volumes"
rules = {"compute_extension:v3:os-volumes-attachments:create": "@",
rule_name: "project:non_fake"}
body = {'volumeAttachment': {'volumeId': FAKE_UUID_A,
'device': '/dev/fake'}}
self._common_policy_check(rules, rule_name, self.controller.create,
self.req, FAKE_UUID, body=body)
rule_name = "compute_extension:v3:os-volumes-attachments:create"
rules = {"compute_extension:v3:os-volumes": "@",
rule_name: "project:non_fake"}
self._common_policy_check(rules, rule_name, self.controller.create,
self.req, FAKE_UUID, body=body)
def test_update_volume_attach_policy_failed(self):
rule_name = "compute_extension:v3:os-volumes"
rules = {"compute_extension:v3:os-volumes-attachments:update": "@",
rule_name: "project:non_fake"}
body = {'volumeAttachment': {'volumeId': FAKE_UUID_B}}
self._common_policy_check(rules, rule_name, self.controller.update,
self.req, FAKE_UUID, FAKE_UUID_A, body=body)
rule_name = "compute_extension:v3:os-volumes-attachments:update"
rules = {"compute_extension:v3:os-volumes": "@",
rule_name: "project:non_fake"}
self._common_policy_check(rules, rule_name, self.controller.update,
self.req, FAKE_UUID, FAKE_UUID_A, body=body)
def test_delete_volume_attach_policy_failed(self):
rule_name = "compute_extension:v3:os-volumes"
rules = {"compute_extension:v3:os-volumes-attachments:delete": "@",
rule_name: "project:non_fake"}
self._common_policy_check(rules, rule_name, self.controller.delete,
self.req, FAKE_UUID, FAKE_UUID_A)
rule_name = "compute_extension:v3:os-volumes-attachments:delete"
rules = {"compute_extension:v3:os-volumes": "@",
rule_name: "project:non_fake"}
self._common_policy_check(rules, rule_name, self.controller.delete,
self.req, FAKE_UUID, FAKE_UUID_A)