openstack
/
manila-tempest-plugin
Code Issues Proposed changes

Add access rules restriction tests 

Adds new test cases and updates current test cases for access allow
and deny APIs. Such APIs now allow users to restrict the visibility
of some fields during the access creation, or even restricting the
deletion of access rules.

Depends-On: https://review.opendev.org/c/openstack/manila/+/887725
Change-Id: If0f7655dc6c11c6c12eeb76bd1cc853466235cca
This commit is contained in:
silvacarloss 2023-07-24 13:53:32 -03:00
parent 646fcd2e23
commit d10819b89b
5 changed files with 207 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
manila_tempest_tests/config.py
View File

@ -40,7 +40,7 @@ ShareGroup = [
"This value is only used to validate the versions "
"response from Manila."),
cfg.StrOpt("max_api_microversion",
default="2.81",
default="2.82",
help="The maximum api microversion is configured to be the "
"value of the latest microversion supported by Manila."),
cfg.StrOpt("region",

12
manila_tempest_tests/services/share/v2/json/shares_client.py
View File

@ -764,7 +764,8 @@ class SharesV2Client(shares_client.SharesClient):
def create_access_rule(self, share_id, access_type="ip",
access_to="0.0.0.0", access_level=None,
version=LATEST_MICROVERSION, metadata=None,
action_name=None):
action_name=None, lock_visibility=False,
lock_deletion=False):
post_body = {
self._get_access_action_name(version, 'os-allow_access'): {
"access_type": access_type,
@ -774,6 +775,10 @@ class SharesV2Client(shares_client.SharesClient):
}
if metadata is not None:
post_body['allow_access']['metadata'] = metadata
if lock_visibility:
post_body['allow_access']['lock_visibility'] = True
if lock_deletion:
post_body['allow_access']['lock_deletion'] = True
body = json.dumps(post_body)
resp, body = self.post(
"shares/%s/action" % share_id, body, version=version,
@ -817,12 +822,15 @@ class SharesV2Client(shares_client.SharesClient):
return rest_client.ResponseBody(resp, body)
def delete_access_rule(self, share_id, rule_id,
version=LATEST_MICROVERSION, action_name=None):
version=LATEST_MICROVERSION, action_name=None,
unrestrict=False):
post_body = {
self._get_access_action_name(version, 'os-deny_access'): {
"access_id": rule_id,
}
}
if unrestrict:
post_body['deny_access']['unrestrict'] = True
body = json.dumps(post_body)
resp, body = self.post(
"shares/%s/action" % share_id, body, version=version)

14
manila_tempest_tests/tests/api/base.py
View File

@ -1017,7 +1017,8 @@ class BaseSharesTest(test.BaseTestCase):
def allow_access(self, share_id, client=None, access_type=None,
access_level='rw', access_to=None, metadata=None,
version=LATEST_MICROVERSION, status='active',
raise_rule_in_error_state=True, cleanup=True):
raise_rule_in_error_state=True, lock_visibility=False,
lock_deletion=False, cleanup=True):
client = client or self.shares_v2_client
a_type, a_to = utils.get_access_rule_data_from_config(
@ -1030,8 +1031,15 @@ class BaseSharesTest(test.BaseTestCase):
'access_to': access_to,
'access_level': access_level
}
delete_kwargs = (
{'unrestrict': True} if lock_deletion else {}
)
if client is self.shares_v2_client:
kwargs.update({'metadata': metadata, 'version': version})
if lock_visibility:
kwargs.update({'lock_visibility': True})
if lock_deletion:
kwargs.update({'lock_deletion': True})
rule = client.create_access_rule(share_id, **kwargs)['access']
waiters.wait_for_resource_status(
@ -1042,7 +1050,9 @@ class BaseSharesTest(test.BaseTestCase):
self.addCleanup(
client.wait_for_resource_deletion, rule_id=rule['id'],
share_id=share_id, version=version)
self.addCleanup(client.delete_access_rule, share_id, rule['id'])
self.addCleanup(
client.delete_access_rule, share_id, rule['id'],
**delete_kwargs)
return rule

91
manila_tempest_tests/tests/api/test_rules.py
View File

@ -28,6 +28,7 @@ from manila_tempest_tests import utils
CONF = config.CONF
LATEST_MICROVERSION = CONF.share.max_api_microversion
RESTRICTED_RULES_VERSION = '2.82'
def _create_delete_ro_access_rule(self, version):
@ -444,6 +445,10 @@ class ShareRulesTest(base.BaseSharesMixedTest):
cls.share_type = cls.create_share_type()
cls.share_type_id = cls.share_type['id']
cls.share = cls.create_share(share_type_id=cls.share_type_id)
cls.user_project = cls.os_admin.projects_client.show_project(
cls.shares_v2_client.project_id)['project']
cls.new_user = cls.create_user_and_get_client(
project=cls.user_project)
@decorators.idempotent_id('c52e95cc-d6ea-4d02-9b52-cd7c1913dfff')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
@ -519,6 +524,92 @@ class ShareRulesTest(base.BaseSharesMixedTest):
msg = "expected id lists %s times in rule list" % (len(gen))
self.assertEqual(1, len(gen), msg)
@decorators.idempotent_id('3bca373e-e54f-49e1-8789-99a383cf4df3')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
@utils.skip_if_microversion_not_supported(RESTRICTED_RULES_VERSION)
@ddt.data(
*itertools.product(utils.deduplicate(
["2.81", CONF.share.max_api_microversion]), (True, False))
)
@ddt.unpack
def test_list_restricted_rules_from_other_user(
self, older_version, lock_visibility):
# create rule
self.allow_access(
self.share["id"], client=self.shares_v2_client,
access_type=self.access_type, access_to=self.access_to,
lock_visibility=lock_visibility, lock_deletion=True)
rule = (
self.shares_v2_client.list_access_rules(
self.share["id"])["access_list"][0])
rules = self.new_user.shares_v2_client.list_access_rules(
self.share['id'])['access_list']
rules_get_lower_version = (
self.new_user.shares_v2_client.list_access_rules(
self.share['id'], version=older_version)['access_list'])
expected_access_to = '******' if lock_visibility else self.access_to
expected_access_key = (
'******' if lock_visibility else rule['access_key'])
# verify values
rule_latest_rules_api = [r for r in rules if r['id'] == rule['id']][0]
rule_lower_version_rules_api = [r for r in rules_get_lower_version
if r['id'] == rule['id']][0]
self.assertEqual(
expected_access_to, rule_latest_rules_api["access_to"])
self.assertEqual(
expected_access_to,
rule_lower_version_rules_api['access_to'])
if self.access_type == 'cephx':
self.assertEqual(expected_access_key,
rule_latest_rules_api['access_key'])
self.assertEqual(
expected_access_key,
rule_lower_version_rules_api['access_key'])
@decorators.idempotent_id('4829265a-eb32-400d-91a0-be06ce31a2ef')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
def test_admin_listing_restricted_rules(self):
utils.check_skip_if_microversion_not_supported(
RESTRICTED_RULES_VERSION)
# create rule
self.allow_access(
self.share["id"], client=self.shares_v2_client,
access_type=self.access_type, access_to=self.access_to,
lock_visibility=True)
rules = self.admin_shares_v2_client.list_access_rules(
self.share["id"])['access_list']
# ensure admin can see rules even if locked
self.assertEqual(self.access_to, rules[0]["access_to"])
if self.access_type == 'cephx':
self.assertIsNotNone(rules[0]['access_key'])
self.assertFalse(rules[0]['access_key'] == '******')
else:
self.assertIsNone(rules[0]['access_key'])
@decorators.idempotent_id('00202c6c-b4c7-4fa6-933a-562fbffde405')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
def test_admin_delete_restricted_rules(self):
utils.check_skip_if_microversion_not_supported(
RESTRICTED_RULES_VERSION)
# create rule
rule = self.allow_access(
self.share["id"], client=self.shares_v2_client,
access_type=self.access_type, access_to=self.access_to,
lock_visibility=True, lock_deletion=True, cleanup=False)
self.admin_shares_v2_client.delete_access_rule(
self.share['id'], rule['id'], unrestrict=True)
self.shares_v2_client.wait_for_resource_deletion(
rule_id=rule['id'], share_id=self.share['id'])
@decorators.idempotent_id('b77bcbda-9754-48f0-9be6-79341ad1af64')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
@ddt.data(*utils.deduplicate(['1.0', '2.9', '2.27', '2.28',

93
manila_tempest_tests/tests/api/test_rules_negative.py
View File

@ -28,6 +28,7 @@ from manila_tempest_tests import utils
CONF = config.CONF
LATEST_MICROVERSION = CONF.share.max_api_microversion
RESTRICTED_RULES_VERSION = '2.82'
@ddt.ddt
@ -60,6 +61,11 @@ class ShareIpRulesForNFSNegativeTest(base.BaseSharesMixedTest):
# create snapshot
cls.snap = cls.create_snapshot_wait_for_active(cls.share["id"])
cls.user_project = cls.os_admin.projects_client.show_project(
cls.shares_v2_client.project_id)['project']
cls.new_user = cls.create_user_and_get_client(
project=cls.user_project)
@decorators.idempotent_id('16781b45-d2bb-4891-aa97-c28c0769d5bd')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
@ddt.data('1.2.3.256',
@ -170,6 +176,93 @@ class ShareIpRulesForNFSNegativeTest(base.BaseSharesMixedTest):
self.admin_client.create_access_rule,
share["id"], access_type, access_to)
@decorators.idempotent_id('478d3c84-b0ea-41c8-a860-e87f182d991c')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
def test_deny_access_unrestrict_other_user_rule(self):
utils.check_skip_if_microversion_not_supported(
RESTRICTED_RULES_VERSION)
access_type, access_to = utils.get_access_rule_data_from_config(
self.protocol)
# create rule
rule = self.allow_access(
self.share["id"], client=self.shares_v2_client,
access_type=access_type, access_to=access_to,
lock_visibility=True, lock_deletion=True)
self.assertRaises(
lib_exc.Forbidden,
self.new_user.shares_v2_client.delete_access_rule,
self.share['id'],
rule['id'],
unrestrict=True
)
@decorators.idempotent_id('c107b0b7-7a3e-4114-af64-ca8fe6e836c9')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
@ddt.data(True, False)
def test_deny_access_without_unrestrict_as_owner_user(self, same_user):
utils.check_skip_if_microversion_not_supported(
RESTRICTED_RULES_VERSION)
access_type, access_to = utils.get_access_rule_data_from_config(
self.protocol)
# create rule
rule = self.allow_access(
self.share["id"], client=self.shares_v2_client,
access_type=access_type, access_to=access_to,
lock_visibility=True, lock_deletion=True)
client = (
self.shares_v2_client
if same_user else self.new_user.shares_v2_client
)
self.assertRaises(
lib_exc.Forbidden,
client.delete_access_rule,
self.share['id'],
rule['id'])
self.assertRaises(
lib_exc.Forbidden,
client.delete_access_rule,
self.share['id'],
rule['id'],
version='2.81'
)
@decorators.idempotent_id('f5b9e7c9-7e6b-4918-a1c4-e03c8d82c46a')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_allow_access_multiple_visibility_locks_not_allowed(self):
utils.check_skip_if_microversion_not_supported(
RESTRICTED_RULES_VERSION)
access_type, access_to = utils.get_access_rule_data_from_config(
self.protocol)
# create rule
rule = self.allow_access(
self.share["id"], client=self.shares_v2_client,
access_type=access_type, access_to=access_to,
lock_visibility=True, lock_deletion=True)
self.assertRaises(
lib_exc.Conflict,
self.shares_v2_client.create_resource_lock,
rule['id'],
"access_rule",
resource_action="show",
lock_reason="locked for testing"
)
self.assertRaises(
lib_exc.Conflict,
self.new_user.shares_v2_client.create_resource_lock,
rule['id'],
"access_rule",
resource_action="show",
lock_reason="locked for testing"
)
@ddt.ddt
class ShareIpRulesForCIFSNegativeTest(ShareIpRulesForNFSNegativeTest):