diff --git a/nova/tests/unit/api/openstack/compute/test_volumes.py b/nova/tests/unit/api/openstack/compute/test_volumes.py index 3d6ca62e90..c8f5a0beec 100644 --- a/nova/tests/unit/api/openstack/compute/test_volumes.py +++ b/nova/tests/unit/api/openstack/compute/test_volumes.py @@ -1506,57 +1506,6 @@ class AssistedSnapshotDeleteTestCaseV275(AssistedSnapshotDeleteTestCaseV21): self.controller.delete, req, 1) -class TestVolumeAttachPolicyEnforcementV21(test.NoDBTestCase): - - def setUp(self): - super(TestVolumeAttachPolicyEnforcementV21, self).setUp() - self.controller = volumes_v21.VolumeAttachmentController() - self.req = fakes.HTTPRequest.blank('') - - self.stub_out('nova.compute.api.API.get', fake_get_instance) - - 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 = "os_compute_api: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 = "os_compute_api:os-volumes-attachments:show" - rules = {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 = "os_compute_api:os-volumes-attachments:create" - rules = {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) - - def test_update_volume_attach_policy_failed(self): - rule_name = "os_compute_api:os-volumes-attachments:update" - rules = {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) - - def test_delete_volume_attach_policy_failed(self): - rule_name = "os_compute_api:os-volumes-attachments:delete" - rules = {rule_name: "project:non_fake"} - self._common_policy_check(rules, rule_name, self.controller.delete, - self.req, FAKE_UUID, FAKE_UUID_A) - - class TestVolumesAPIDeprecation(test.NoDBTestCase): def setUp(self): diff --git a/nova/tests/unit/policies/test_volumes.py b/nova/tests/unit/policies/test_volumes.py new file mode 100644 index 0000000000..ae738a0e45 --- /dev/null +++ b/nova/tests/unit/policies/test_volumes.py @@ -0,0 +1,179 @@ +# 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 fixtures +import mock +from oslo_utils.fixture import uuidsentinel as uuids +from oslo_utils import timeutils + +from nova.api.openstack.compute import volumes as volumes_v21 +from nova.compute import vm_states +from nova import exception +from nova import objects +from nova.tests.unit.api.openstack import fakes +from nova.tests.unit import fake_block_device +from nova.tests.unit import fake_instance +from nova.tests.unit.policies import base + +# This is the server ID. +FAKE_UUID = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa' +# This is the old volume ID (to swap from). +FAKE_UUID_A = '00000000-aaaa-aaaa-aaaa-000000000000' +# This is the new volume ID (to swap to). +FAKE_UUID_B = 'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb' + + +def fake_bdm_get_by_volume_and_instance(cls, ctxt, volume_id, instance_uuid): + if volume_id != FAKE_UUID_A: + raise exception.VolumeBDMNotFound(volume_id=volume_id) + db_bdm = fake_block_device.FakeDbBlockDeviceDict( + {'id': 1, + 'instance_uuid': instance_uuid, + 'device_name': '/dev/fake0', + 'delete_on_termination': 'False', + 'source_type': 'volume', + 'destination_type': 'volume', + 'snapshot_id': None, + 'volume_id': volume_id, + 'volume_size': 1}) + return objects.BlockDeviceMapping._from_db_object( + ctxt, objects.BlockDeviceMapping(), db_bdm) + + +def fake_get_volume(self, context, id): + if id == FAKE_UUID_A: + status = 'in-use' + attach_status = 'attached' + elif id == FAKE_UUID_B: + status = 'available' + attach_status = 'detached' + else: + raise exception.VolumeNotFound(volume_id=id) + return {'id': id, 'status': status, 'attach_status': attach_status} + + +class VolumeAttachPolicyTest(base.BasePolicyTest): + """Test os-volumes-attachments APIs policies with all possible context. + + This class defines the set of context with different roles + which are allowed and not allowed to pass the policy checks. + With those set of context, it will call the API operation and + verify the expected behaviour. + """ + + def setUp(self): + super(VolumeAttachPolicyTest, self).setUp() + self.controller = volumes_v21.VolumeAttachmentController() + self.req = fakes.HTTPRequest.blank('') + self.stub_out('nova.objects.BlockDeviceMapping' + '.get_by_volume_and_instance', + fake_bdm_get_by_volume_and_instance) + self.stub_out('nova.volume.cinder.API.get', fake_get_volume) + + self.mock_get = self.useFixture( + fixtures.MockPatch('nova.api.openstack.common.get_instance')).mock + uuid = uuids.fake_id + self.instance = fake_instance.fake_instance_obj( + self.project_member_context, + id=1, uuid=uuid, project_id=self.project_id, + vm_state=vm_states.ACTIVE, + task_state=None, launched_at=timeutils.utcnow()) + self.mock_get.return_value = self.instance + + # Check that admin or owner is able to list/create/show/delete + # the attached volume. + self.admin_or_owner_authorized_contexts = [ + self.legacy_admin_context, self.system_admin_context, + self.project_admin_context, self.project_foo_context, + self.project_reader_context, self.project_member_context + ] + + self.admin_or_owner_unauthorized_contexts = [ + self.system_member_context, self.system_reader_context, + self.system_foo_context, + self.other_project_member_context + ] + + # Check that admin is able to update the attached volume + self.admin_authorized_contexts = [ + self.legacy_admin_context, + self.system_admin_context, + self.project_admin_context + ] + # Check that non-admin is not able to change the service + self.admin_unauthorized_contexts = [ + self.system_member_context, + self.system_reader_context, + self.system_foo_context, + self.project_member_context, + self.other_project_member_context, + self.project_foo_context, + self.project_reader_context + ] + + @mock.patch.object(objects.BlockDeviceMappingList, 'get_by_instance_uuid') + def test_index_volume_attach_policy(self, mock_get_instance): + rule_name = "os_compute_api:os-volumes-attachments:index" + self.common_policy_check(self.admin_or_owner_authorized_contexts, + self.admin_or_owner_unauthorized_contexts, + rule_name, self.controller.index, + self.req, FAKE_UUID) + + def test_show_volume_attach_policy(self): + rule_name = "os_compute_api:os-volumes-attachments:show" + self.common_policy_check(self.admin_or_owner_authorized_contexts, + self.admin_or_owner_unauthorized_contexts, + rule_name, self.controller.show, + self.req, FAKE_UUID, FAKE_UUID_A) + + @mock.patch('nova.compute.api.API.attach_volume') + def test_create_volume_attach_policy(self, mock_attach_volume): + rule_name = "os_compute_api:os-volumes-attachments:create" + body = {'volumeAttachment': {'volumeId': FAKE_UUID_B, + 'device': '/dev/fake'}} + self.common_policy_check(self.admin_or_owner_authorized_contexts, + self.admin_or_owner_unauthorized_contexts, + rule_name, self.controller.create, + self.req, FAKE_UUID, body=body) + + @mock.patch('nova.compute.api.API.detach_volume') + def test_delete_volume_attach_policy(self, mock_detach_volume): + rule_name = "os_compute_api:os-volumes-attachments:delete" + self.common_policy_check(self.admin_or_owner_authorized_contexts, + self.admin_or_owner_unauthorized_contexts, + rule_name, self.controller.delete, + self.req, FAKE_UUID, FAKE_UUID_A) + + @mock.patch('nova.compute.api.API.swap_volume') + def test_update_volume_attach_policy(self, mock_swap_volume): + rule_name = "os_compute_api:os-volumes-attachments:update" + body = {'volumeAttachment': {'volumeId': FAKE_UUID_B}} + self.common_policy_check(self.admin_authorized_contexts, + self.admin_unauthorized_contexts, + rule_name, self.controller.update, + self.req, FAKE_UUID, FAKE_UUID_A, body=body) + + +class VolumeAttachScopeTypePolicyTest(VolumeAttachPolicyTest): + """Test os-volume-attachments APIs policies with system scope enabled. + + This class set the nova.conf [oslo_policy] enforce_scope to True + so that we can switch on the scope checking on oslo policy side. + It defines the set of context with scoped token + which are allowed and not allowed to pass the policy checks. + With those set of context, it will run the API operation and + verify the expected behaviour. + """ + + def setUp(self): + super(VolumeAttachScopeTypePolicyTest, self).setUp() + self.flags(enforce_scope=True, group="oslo_policy")