starlingx/openstack-armada-app
Code Issues Proposed changes
openstack-armada-app/enhanced-policies/tests/test_cinder/test_rbac_cinder.py
Heitor Matsui ecb1e24972 RBAC Patch 2: Cinder tests and common test files 
This patch chain aims to suggest a set of default policies for user
management on stx-openstack. We suggest the creation of the project_admin
and project_readonly roles and provide some policies to fine tune the
access control over the Openstack services to those roles, as described
on README.md.

Also, we provide a set of tests to ensure the policies and permissions
are all working as expected on site for the cloud administrators.

This commit includes Cinder related tests and functions, along with
common test functions used by multiple OpenStack services.

Story: 2008910
Task: 42501

Signed-off-by: Heitor Matsui <heitorvieira.matsui@windriver.com>
Signed-off-by: Thiago Brito <thiago.brito@windriver.com>
Co-authored-by: Miriam Yumi Peixoto <miriam.yumipeixoto@windriver.com>
Co-authored-by: Leonardo Zaccarias <leonardo.zaccarias@windriver.com>
Co-authored-by: Rogerio Oliveira Ferraz <rogeriooliveira.ferraz@windriver.com>

Change-Id: I6b43bc584e470f022fb08a8a4cf741c188dfe80d
2021-06-02 13:20:54 -03:00

699 lines
28 KiB
Python
Raw Permalink Blame History

#
# Copyright (c) 2021 Wind River Systems, Inc.
#
# SPDX-License-Identifier: Apache-2.0
#
# All Rights Reserved.
#
from tests import rbac_test_base
class TestBlockStorage(rbac_test_base.TestClass):
def test_uc_volume_1(self):
"""
1. user11 can create a volume from an image
2. usr11 can list/detail the volume
3. user11 can create metadata on the volume
4. user11 can update/delete metadata
5. user11 cant extend the volume when detached
6. user11 can update readonly flag of the volume
7. user11 can retype the volume
8. user11 can attach/dettach the volume
"""
self.set_connections_for_user(self.user11)
# 1. user11 can create a volume from an image
self._create_volume("volume11")
# 2. usr11 can list/detail the volume
volumes = self._list_volumes()
self.assertIn("volume11", [v.name for v in volumes])
# 3. user11 can create metadata on the volume
self._update_volume_metadata("volume11", metadata={"my": "test"})
volume11 = self._get_volume("volume11")
self.assertIn("my", volume11.metadata)
self.assertEqual(volume11.metadata.get("my"), "test")
# 4. user11 can update/delete metadata
self._update_volume_metadata("volume11", metadata={"my": "test2"})
volume11 = self._get_volume("volume11")
self.assertIn("my", volume11.metadata)
self.assertEqual(volume11.metadata.get("my"), "test2")
self._delete_volume_metadata("volume11", ["my"])
volume11 = self._get_volume("volume11")
self.assertNotIn("my", volume11.metadata)
# 5. user11 cant extend the volume when detached
volume11 = self._extend_volume("volume11", 2)
self.assertEqual(volume11.size, 2)
# 6. user11 can update readonly flag of the volume
# TODO(tbrito): Fix after merge of https://review.opendev.org/c/openstack/openstacksdk/+/776266
self._set_volume_readonly_flag("volume11", readonly=True)
volume11 = self._get_volume("volume11")
self.assertTrue(volume11.metadata.get("readonly"))
# 7. user11 can retype the volume
# TODO(tbrito): Fix after merge of https://review.opendev.org/c/openstack/openstacksdk/+/776272
self._retype_volume("volume11", volume_type="rbd1")
volume11 = self._get_volume("volume11")
# TODO(tbrito): Req accepted but volume doesn't change? Figure out why
# self.assertEquals(volume11.volume_type, "rbd1")
# 8. user11 can attach/detach the volume
self._create_server("instance11", image_name="cirros", flavor_name="m1.tiny")
self._add_volume_to_server("instance11", "volume11")
instance11 = self._get_server("instance11")
self.assertIn(volume11.id, [v.get("id") for v in instance11.attached_volumes])
self._remove_volume_from_server("volume11", "instance11")
instance11 = self._get_server("instance11")
self.assertEqual(instance11.attached_volumes, [])
def test_uc_volume_2(self):
"""
1. user12 can create volume from an image
2. user12 cannot delete the volume
3. use12 can list/details the volume it created
4. user 12 can create metadata of the volumes
5. user 12 can not update/delete metadata of the volumes
6. user12 can get list/detail of metadata of volumes of project1
7. user12 cannot extend the volume
8. user12 can attach/detach the volume to an instance
"""
self.set_connections_for_user(self.user12)
# 1. user12 can create volume form an image
self._create_volume("volume12")
volumes = self._list_volumes()
self.assertIn("volume12", [v.name for v in volumes])
# 2. user12 cannot delete the volume
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume:delete to be performed",
self._delete_volume,
"volume12"
)
# 3. use12 can list/details the volume it created
volumes = self._list_volumes()
self.assertIn("volume12", [v.name for v in volumes])
self._get_volume("volume12")
# 4. user 12 can create metadata of the volumes
self._update_volume_metadata("volume12", metadata={"my": "test"})
volume12 = self._get_volume("volume12")
self.assertIn("my", volume12.metadata)
self.assertEqual(volume12.metadata.get("my"), "test")
# 5. user12 can not update/delete metadata of the volumes
# NOTE(tbrito): cinderclient.set_metadata uses the POST endpoint, so it's not possible to verify that atm
# self.assertRaises(
# Exception,
# self._update_volume_metadata,
# "volume12",
# metadata={"my": "test2"}
# )
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume:delete_volume_metadata to be performed",
self._delete_volume_metadata,
"volume12",
["my"]
)
# 6. user12 can get list/detail of metadata of volumes of project1
metadata = self._get_volume_metadata("volume12")
self.assertIn("my", volume12.metadata)
self.assertEqual(metadata.get("my"), "test")
# 7. user12 cannot extend the volume
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume:extend to be performed",
self._extend_volume,
"volume12",
2
)
# 8. user12 can attach/detach the volume to an instance
self._create_server("instance12", image_name="cirros", flavor_name="m1.tiny")
self._add_volume_to_server("instance12", "volume12")
instance12 = self._get_server("instance12")
self.assertIn(volume12.id, [v.get("id") for v in instance12.attached_volumes])
self._remove_volume_from_server("volume12", "instance12")
instance12 = self._get_server("instance12")
self.assertEqual(instance12.attached_volumes, [])
def test_uc_volume_3(self):
"""
1. user13 cannot create/delete/update volumes of project1
2. user13 can list/details the volumes of project1
3. user13 cannot add/update/delete metadata of volumes
4. user13 can show metadata of volumes
5. user13 cannot update readonly flag of the volumes
"""
self.set_connections_for_user(self.user11)
self._create_volume("volume11")
self._update_volume_metadata("volume11", metadata={"my-11": "test-11"})
self.set_connections_for_user(self.user13)
# 1. user13 cannot create/delete/update volumes of project1
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume:create to be performed",
self._create_volume,
"volume13"
)
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume:delete to be performed",
self._delete_volume,
"volume11"
)
# NOTE(tbrito): cinderclient.set_metadata uses the POST endpoint, so it's not possible to verify that atm
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume:update to be performed",
self._update_volume,
"volume11",
name="THIS IS VOLUME 13"
)
# 2. user13 can list/details the volumes of project1
volumes = self._list_volumes()
self.assertIn("volume11", [v.name for v in volumes])
volume11 = self._get_volume("volume11")
self.assertEqual(volume11.status, "available")
# 3. user13 cannot add/update/delete metadata of volumes
# NOTE(tbrito): cinderclient.set_metadata uses the POST endpoint, so
# it's not possible to verify that atm
# self.assertRaises(
# Exception,
# self._update_volume_metadata,
# "volume11",
# metadata={"my": "test"}
# )
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume:delete_volume_metadata to be performed",
self._delete_volume_metadata,
"volume11",
["my-11"]
)
# 4. user13 can show metadata of volumes
volume11 = self._get_volume("volume11")
self.assertDictEqual(volume11.metadata, {"my-11": "test-11"})
# 5. user13 cannot update readonly flag of the volumes
# TODO(tbrito): Fix after merge of https://review.opendev.org/c/openstack/openstacksdk/+/776266
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume:update_readonly_flag to be performed",
self._set_volume_readonly_flag,
"volume11",
readonly=True
)
def test_uc_volume_4(self):
"""
user11/12/13 as members of project1,
1. cannot get list/detail of volumes of project2
2. cannot update/delete volumes of project2
3. cannot force delete volumes of project2
"""
self.set_connections_for_user(self.user21)
self._create_volume("volume21")
for user in (self.user11, self.user12, self.user13):
self.set_connections_for_user(user)
# 1. cannot get list/detail of volumes of project2,
self.assertNotIn("volume21", [v.name for v in self._list_volumes()])
self.assertRaisesRegex(
Exception,
"No Volume found for volume21",
self._get_volume,
"volume21"
)
# 2. cannot update/delete volumes of project2
self.assertRaisesRegex(
Exception,
"No Volume found for volume21",
self._update_volume_metadata,
"volume21",
metadata={"my": "test"}
)
self.assertRaisesRegex(
Exception,
"No Volume found for volume21",
self._delete_volume,
"volume21"
)
# 3. cannot force delete volumes of project2
self.assertRaisesRegex(
Exception,
"No Volume found for volume21",
self._delete_volume,
"volume21",
force=True
)
def test_uc_snapshot_1(self):
"""
1. user11 create a snapshot of volume with metadata when the voluem is detached.
2. user11 can list/detail metadata of snapshot of project1
3. user11 can update/delete the metadata of snapshot
4. user11 can detail the snapshot of project1
5. user11 can update/delete snapshot
6. user11 can create a snapshot of the volume when it is attached
"""
self.set_connections_for_user(self.user11)
self._create_volume("volume11")
# 1. user11 create a snapshot of volume with metadata when the volume is detached.
self._create_snapshot(volume_name="volume11", name="snapshot11", description="snapshot11yeah",
metadata={"my": "test"})
# TODO(tbrito): https://review.opendev.org/c/openstack/openstacksdk/+/778757
# 2. user11 can list/detail metadata of snapshot of project1
metadata = self._get_snapshot_metadata("snapshot11")
self.assertIn("my", [k for k, v in metadata.items()])
# 3. user11 can update/delete the metadata of snapshot
self._update_snapshot_metadata("snapshot11", metadata={"my": "test2"})
metadata = self._get_snapshot_metadata("snapshot11")
self.assertIn("test2", metadata.get("my"))
self._delete_snapshot_metadata("snapshot11", "my")
metadata = self._get_snapshot_metadata("snapshot11")
self.assertNotIn("my", metadata)
# 4. user11 can detail the snapshot of project1
snapshot = self._get_snapshot("snapshot11")
self.assertEqual("snapshot11yeah", snapshot.description)
# 5. user11 can update/delete snapshot
# TODO(tbrito):
self._update_snapshot("snapshot11", description="My test description")
snapshot = self._get_snapshot("snapshot11")
self.assertEqual("My test description", snapshot.description)
self._delete_snapshot("snapshot11")
# 6. user11 can create a snapshot of the volume when it is attached
self._create_server("instance11", image_name="cirros", flavor_name="m1.tiny")
self._add_volume_to_server("instance11", "volume11")
self._create_snapshot(volume_name="volume11", name="snapshot11.2", force=True)
def test_uc_snapshot_2(self):
"""
1. user12 create a snapshot of volume with metadata when the volume is detached
2. user12 can list/detail metadata of snapshot of project1
3. user12 cannot update/delete the metadata of snapshot
4. user12 can detail the snapshot of project1
5. user12 cannot update/delete snapshot
6. user12 can create a snapshot of the volume wht it is attached
"""
self.set_connections_for_user(self.user11)
self._create_volume("volume11")
self._create_snapshot(volume_name="volume11", name="snapshot11", description="snapshot11yeah",
metadata={"my": "test"})
self.set_connections_for_user(self.user12)
self._create_volume("volume12")
# 1. user12 create a snapshot of volume with metadata when the volume is detached.
self._create_snapshot(volume_name="volume12", name="snapshot12", metadata={"my2": "test2"})
# TODO(tbrito): https://review.opendev.org/c/openstack/openstacksdk/+/778757
# 2. user12 can list/detail metadata of snapshot of project1
metadata = self._get_snapshot_metadata("snapshot11")
self.assertIn("my", [k for k, v in metadata.items()])
# 3. user12 cannot update/delete the metadata of snapshot
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume:update_snapshot_metadata to be performed",
self._update_snapshot_metadata,
"snapshot11",
metadata={"my": "test2"}
)
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume:delete_snapshot_metadata to be performed",
self._delete_snapshot_metadata,
"snapshot11",
"my"
)
# 4. user12 can detail the snapshot of project1
snapshot = self._get_snapshot("snapshot11")
self.assertEqual("snapshot11yeah", snapshot.description)
# 5. user12 cannot update/delete snapshot
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume:update_snapshot to be performed",
self._update_snapshot,
"snapshot11",
description="My test description"
)
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume:delete_snapshot to be performed",
self._delete_snapshot,
"snapshot11"
)
# 6. user12 can create a snapshot of the volume when it is attached
self._create_server("instance12", image_name="cirros", flavor_name="m1.tiny")
# NOTE: user12 cannot create attachment due to os_compute_api:os-volumes-attachmentos_compute_api:os-volumes-attachments:create
# Using user11 instead
self.set_connections_for_user(self.user11)
self._add_volume_to_server("instance12", "volume12")
self.set_connections_for_user(self.user12)
self._create_snapshot(volume_name="volume12", name="snapshot12.2", force=True)
def test_uc_snapshot_3(self):
"""
1. user13 cannot create a snapshot of volume with metadata when the volume is detached
2. user13 can list/detail metadata of snapshot of project1
3. user13 cannot update/delete the metadata of snapshot
4. user13 can detail the snapshot of project1
5. user13 cannot update/delete snapshot
6. user13 cannot create a snapshot of the volume when it is attached
"""
self.set_connections_for_user(self.user11)
self._create_volume("volume11")
self._create_snapshot(volume_name="volume11", name="snapshot11", description="snapshot11yeah",
metadata={"my": "test"})
# 1. user13 cannot create a snapshot of volume with metadata when the volume is detached.
self.set_connections_for_user(self.user13)
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume:create_snapshot to be performed",
self._create_snapshot,
volume_name="volume11",
name="snapshot13",
metadata={"my3": "test3"}
)
# TODO(tbrito):https://review.opendev.org/c/openstack/openstacksdk/+/778757
# 2. user13 can list/detail metadata of snapshot of project1
metadata = self._get_snapshot_metadata("snapshot11")
self.assertIn("my", metadata)
# 3. user13 cannot update/delete the metadata of snapshot
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume:update_snapshot_metadata to be performed",
self._update_snapshot_metadata,
"snapshot11",
metadata={"my": "test2"}
)
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume:delete_snapshot_metadata to be performed",
self._delete_snapshot_metadata,
"snapshot11",
"my"
)
# 4. user13 can detail the snapshot of project1
snapshot = self._get_snapshot("snapshot11")
self.assertEqual("snapshot11yeah", snapshot.description)
# 5. user13 cannot update/delete snapshot
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume:update_snapshot to be performed",
self._update_snapshot,
"snapshot11",
description="My test description"
)
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume:delete_snapshot to be performed",
self._delete_snapshot,
"snapshot11"
)
# 6. user13 cannot create a snapshot of the volume when it is attached
self.set_connections_for_user(self.user11)
self._create_server("instance11", image_name="cirros", flavor_name="m1.tiny")
self._add_volume_to_server("instance11", "volume11")
self.set_connections_for_user(self.user13)
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume:create_snapshot to be performed",
self._create_snapshot,
volume_name="volume11",
name="snapshot13"
)
def test_uc_snapshot_4(self):
"""
1. user21 create a snapshot of a volume of project2
2. user11/user12/user13 cannot list/details the volume
"""
# 1. user21 create a snapshot of a volume of project2
self.set_connections_for_user(self.user21)
self._create_volume("volume21")
self._create_snapshot(volume_name="volume21", name="snapshot21", metadata={"my": "test"})
# 2. user11/user12/user13 cannot list/details the volume
for user in (self.user11, self.user12, self.user13):
self.set_connections_for_user(user)
self.assertNotIn("volume21", self._list_snapshots())
self.assertRaisesRegex(
Exception,
"No Snapshot found for volume21",
self._get_snapshot,
"volume21"
)
def test_uc_volumeupload_1(self):
"""
1. user 11 can upload an image from volume of project1
2. user11 can show the new image it uploaded
"""
# 1. user 11 can upload an image from volume of project1
self.set_connections_for_user(self.user11)
self._create_volume("volume11")
self._create_image_from_volume(volume_name="volume11", image_name="image11")
# 2. user11 can show the new image it uploaded
self._get_image_by_name("image11")
def test_uc_volumeupload_2(self):
"""
1. user12/user13 cannot upload image from volume of project1
"""
self.set_connections_for_user(self.user11)
self._create_volume("volume11")
self.set_connections_for_user(self.user12)
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume_extension:volume_actions:upload_image to be performed",
self._create_image_from_volume,
volume_name="volume11",
image_name="image11"
)
self.set_connections_for_user(self.user13)
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume_extension:volume_actions:upload_image to be performed",
self._create_image_from_volume,
volume_name="volume11",
image_name="image11"
)
def test_uc_volumetransfer_1(self):
""""
1. user11 can start a volume transfer of volume of project1
2. user11 can list/details the transfer it started
3. user21 can accept the transfer in project2
4. user11 can delete the transfer it started
"""
self.set_connections_for_user(self.user11)
self._create_volume("volume11")
# 1. user11 can start a volume transfer of volume of project1
original_transfer = self._start_volume_transfer("volume11", "Transfer volume11")
# 2. user11 can list/details the transfer it started
all_transfers = self._list_volume_transfers()
self.assertIn("Transfer volume11", [t.name for t in all_transfers])
transfer = self._get_volume_transfer("Transfer volume11")
self.assertEqual("Transfer volume11", transfer.name)
# 3. user21 can accept the transfer in project2
self.set_connections_for_user(self.user21)
self._accept_volume_transfer(transfer.id, original_transfer.auth_key)
# 4. user11 can delete the transfer it started
self.set_connections_for_user(self.user11)
self._create_volume("volume11.2")
self._start_volume_transfer("volume11.2", "Transfer volume11.2")
self._delete_volume_transfer("Transfer volume11.2")
def test_uc_volumetransfer_2(self):
"""
1. user12 cannot start a volume transfer of volume of project1
2. user21 start a volume transfer in project2
3. user12 can list/detail the volume transfer in project1
4. user12 can accept the transfer in project1
"""
self.set_connections_for_user(self.user12)
self._create_volume("volume12")
# 1. user12 cannot start a volume transfer of volume of project1
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume:create_transfer to be performed",
self._start_volume_transfer,
"volume12",
"Transfer volume12"
)
# 2. user21 start a volume transfer in project2
self.set_connections_for_user(self.user21)
self._create_volume("volume21")
transfer21 = self._start_volume_transfer("volume21", "Transfer volume21")
# 3. user12 can list/detail the volume transfer in project1
all_transfers = self._list_volume_transfers()
self.assertIn("Transfer volume21", [t.name for t in all_transfers])
transfer = self._get_volume_transfer("Transfer volume21")
self.assertEqual(transfer.id, transfer21.id)
# 4. user12 can accept the transfer in project1
self._accept_volume_transfer(transfer.id, transfer21.auth_key)
def test_uc_volumetransfer_3(self):
"""
1. user11 start a volume transfer of volume of project1
2. user13 can list/details the transfer it started
3. user13 cannot start the transfer in project1
4. user21 start a volume transfer in project2
5. user13 cannot accept the transfer in project1
"""
self.set_connections_for_user(self.user11)
self._create_volume("volume11")
# 1. user11 start a volume transfer of volume of project1
self._start_volume_transfer("volume11", "Transfer volume11")
# 2. user13 can list/details the transfer it started
self.set_connections_for_user(self.user13)
all_transfers = self._list_volume_transfers()
self.assertIn("Transfer volume11", [t.name for t in all_transfers])
transfer = self._get_volume_transfer("Transfer volume11")
self.assertEqual("Transfer volume11", transfer.name)
# 3. user13 cannot start the transfer in project1
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume:create_transfer to be performed",
self._start_volume_transfer,
"volume11",
"Another transfer volume11"
)
# 4. user21 start a volume transfer in project2
self.set_connections_for_user(self.user21)
self._create_volume("volume21")
transfer21 = self._start_volume_transfer("volume21", "Transfer volume21")
# 5. user13 cannot accept the transfer in project1
self.set_connections_for_user(self.user13)
self.assertRaisesRegex(
Exception,
"Policy doesn't allow volume:accept_transfer to be performed",
self._accept_volume_transfer,
transfer21.id,
transfer21.auth_key
)
def test_uc_volumebackup_1(self):
"""
1. user11/12 can create a volume backup of project1,
2. user13 cannot create a volume backup of project1,
3. user11/user12/user13 can list/details the created backup
4. user 11 can restore the backup
5. user12 CAN restore the backup (or else nova migration will fail)
6. user12/user13 cannot restore the backup
7. user11 can delete the backup
8. use12/user13 cannot delete the backup
"""
self.set_connections_for_user(self.user11)
self._create_volume("volume11")
# 1. user11/12 can create a volume backup of project1
self._create_volume_backup("volume11", "volume11-bkp11")
self.set_connections_for_user(self.user12)
self._create_volume_backup("volume11", "volume11-bkp12")
# 2. user13 cannot create a volume backup of project1
self.set_connections_for_user(self.user13)
self.assertRaisesRegex(
Exception,
"Policy doesn't allow backup:create to be performed",
self._create_volume_backup,
"volume11",
"volume11-bkp13"
)
# 3. user11/user12/user13 can list/details the created backup
for user in (self.user11, self.user12, self.user13):
self.set_connections_for_user(user)
self._get_volume_backup("volume11-bkp11")
# 4. user 11 can restore the backup
self.set_connections_for_user(self.user11)
self._restore_volume_backup("volume11-bkp12", "restored-volume11")
self.assertIn("restored-volume11", [v.name for v in self._list_volumes()])
# 5. user12 CAN restore the backup (or else nova migration will fail)
self.set_connections_for_user(self.user12)
self.assertRaisesRegex(
Exception,
"Policy doesn't allow backup:restore to be performed",
self._restore_volume_backup, "volume11-bkp12", "restored-volume12"
)
self.assertNotIn("restored-volume12", [v.name for v in self._list_volumes()])
# 6. user13 cannot restore the backup
self.set_connections_for_user(self.user13)
self.assertRaisesRegex(
Exception,
"Policy doesn't allow backup:restore to be performed",
self._restore_volume_backup, "volume11-bkp12", "restored-volume13"
)
self.assertNotIn("restored-volume13", [v.name for v in self._list_volumes()])
# 7. user11 can delete the backup
self.set_connections_for_user(self.user11)
self._delete_volume_backup("volume11-bkp12")
# 8. use12/user13 cannot delete the backup
for user in (self.user12, self.user13):
self.set_connections_for_user(user)
self.assertRaisesRegex(
Exception,
"Policy doesn't allow backup:delete to be performed",
self._delete_volume_backup,
"volume11-bkp11"
)
View Git Blame Copy Permalink