Implement project personas for snapshots

This commit updates the policies to use the default roles provided by
keystone to provide some level of authoritative consistency across
OpenStack APIs.

Future changes will incorporate system scope when cinder fully supports
it.

Legacy snapshot unit tests have been updated to not rely on overriding
snapshot policies.

Change-Id: If9fbe267954d5e7395972da3cd58d53801ff97ef
This commit is contained in:
Alan Bishop 2021-09-03 08:39:37 -07:00
parent 81e0da35dc
commit 85d64c6474
6 changed files with 418 additions and 89 deletions

View File

@ -26,11 +26,36 @@ DELETE_POLICY = 'volume:delete_snapshot'
UPDATE_POLICY = 'volume:update_snapshot'
EXTEND_ATTRIBUTE = 'volume_extension:extended_snapshot_attributes'
deprecated_get_all_snapshots = base.CinderDeprecatedRule(
name=GET_ALL_POLICY,
check_str=base.RULE_ADMIN_OR_OWNER
)
deprecated_extend_snapshot_attribute = base.CinderDeprecatedRule(
name=EXTEND_ATTRIBUTE,
check_str=base.RULE_ADMIN_OR_OWNER
)
deprecated_create_snapshot = base.CinderDeprecatedRule(
name=CREATE_POLICY,
check_str=base.RULE_ADMIN_OR_OWNER
)
deprecated_get_snapshot = base.CinderDeprecatedRule(
name=GET_POLICY,
check_str=base.RULE_ADMIN_OR_OWNER
)
deprecated_update_snapshot = base.CinderDeprecatedRule(
name=UPDATE_POLICY,
check_str=base.RULE_ADMIN_OR_OWNER
)
deprecated_delete_snapshot = base.CinderDeprecatedRule(
name=DELETE_POLICY,
check_str=base.RULE_ADMIN_OR_OWNER
)
snapshots_policies = [
policy.DocumentedRuleDefault(
name=GET_ALL_POLICY,
check_str=base.RULE_ADMIN_OR_OWNER,
check_str=base.SYSTEM_READER_OR_PROJECT_READER,
description="List snapshots.",
operations=[
{
@ -41,10 +66,12 @@ snapshots_policies = [
'method': 'GET',
'path': '/snapshots/detail'
}
]),
],
deprecated_rule=deprecated_get_all_snapshots,
),
policy.DocumentedRuleDefault(
name=EXTEND_ATTRIBUTE,
check_str=base.RULE_ADMIN_OR_OWNER,
check_str=base.SYSTEM_READER_OR_PROJECT_READER,
description="List or show snapshots with extended attributes.",
operations=[
{
@ -55,47 +82,57 @@ snapshots_policies = [
'method': 'GET',
'path': '/snapshots/detail'
}
]),
],
deprecated_rule=deprecated_extend_snapshot_attribute,
),
policy.DocumentedRuleDefault(
name=CREATE_POLICY,
check_str=base.RULE_ADMIN_OR_OWNER,
check_str=base.SYSTEM_ADMIN_OR_PROJECT_MEMBER,
description="Create snapshot.",
operations=[
{
'method': 'POST',
'path': '/snapshots'
}
]),
],
deprecated_rule=deprecated_create_snapshot,
),
policy.DocumentedRuleDefault(
name=GET_POLICY,
check_str=base.RULE_ADMIN_OR_OWNER,
check_str=base.SYSTEM_READER_OR_PROJECT_READER,
description="Show snapshot.",
operations=[
{
'method': 'GET',
'path': '/snapshots/{snapshot_id}'
}
]),
],
deprecated_rule=deprecated_get_snapshot,
),
policy.DocumentedRuleDefault(
name=UPDATE_POLICY,
check_str=base.RULE_ADMIN_OR_OWNER,
check_str=base.SYSTEM_ADMIN_OR_PROJECT_MEMBER,
description="Update snapshot.",
operations=[
{
'method': 'PUT',
'path': '/snapshots/{snapshot_id}'
}
]),
],
deprecated_rule=deprecated_update_snapshot,
),
policy.DocumentedRuleDefault(
name=DELETE_POLICY,
check_str=base.RULE_ADMIN_OR_OWNER,
check_str=base.SYSTEM_ADMIN_OR_PROJECT_MEMBER,
description="Delete snapshot.",
operations=[
{
'method': 'DELETE',
'path': '/snapshots/{snapshot_id}'
}
]),
],
deprecated_rule=deprecated_delete_snapshot,
),
]

View File

@ -250,9 +250,8 @@ class SnapshotApiTest(test.TestCase):
'display_description': 'Default description',
'expected_attrs': ['metadata'],
}
ctx = context.RequestContext(fake.PROJECT_ID, fake.USER_ID, True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
fake_volume_obj = fake_volume.fake_volume_obj(ctx)
snapshot_obj = fake_snapshot.fake_snapshot_obj(self.ctx, **snapshot)
fake_volume_obj = fake_volume.fake_volume_obj(self.ctx)
snapshot_get_by_id.return_value = snapshot_obj
volume_get.return_value = fake_volume_obj
@ -261,6 +260,7 @@ class SnapshotApiTest(test.TestCase):
}
body = {"snapshot": updates}
req = fakes.HTTPRequest.blank('/v2/snapshots/%s' % UUID)
req.environ['cinder.context'] = self.ctx
res_dict = self.controller.update(req, UUID, body=body)
expected = {
'snapshot': {
@ -297,9 +297,8 @@ class SnapshotApiTest(test.TestCase):
'description': 'Default description',
'expected_attrs': ['metadata'],
}
ctx = context.RequestContext(fake.PROJECT_ID, fake.USER_ID, True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
fake_volume_obj = fake_volume.fake_volume_obj(ctx)
snapshot_obj = fake_snapshot.fake_snapshot_obj(self.ctx, **snapshot)
fake_volume_obj = fake_volume.fake_volume_obj(self.ctx)
snapshot_get_by_id.return_value = snapshot_obj
volume_get.return_value = fake_volume_obj
@ -309,6 +308,7 @@ class SnapshotApiTest(test.TestCase):
}
body = {"snapshot": updates}
req = fakes.HTTPRequest.blank('/v2/snapshots/%s' % UUID)
req.environ['cinder.context'] = self.ctx
res_dict = self.controller.update(req, UUID, body=body)
self.assertEqual(fields.SnapshotStatus.AVAILABLE,
@ -356,9 +356,8 @@ class SnapshotApiTest(test.TestCase):
'display_description': 'Default description',
'expected_attrs': ['metadata'],
}
ctx = context.RequestContext(fake.PROJECT_ID, fake.USER_ID, True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
fake_volume_obj = fake_volume.fake_volume_obj(ctx)
snapshot_obj = fake_snapshot.fake_snapshot_obj(self.ctx, **snapshot)
fake_volume_obj = fake_volume.fake_volume_obj(self.ctx)
snapshot_get_by_id.return_value = snapshot_obj
volume_get.return_value = fake_volume_obj
@ -368,6 +367,7 @@ class SnapshotApiTest(test.TestCase):
}
body = {"snapshot": updates}
req = fakes.HTTPRequest.blank('/v2/snapshots/%s' % UUID)
req.environ['cinder.context'] = self.ctx
res_dict = self.controller.update(req, UUID, body=body)
expected = {
'snapshot': {
@ -402,14 +402,14 @@ class SnapshotApiTest(test.TestCase):
'display_description': 'Default description',
'expected_attrs': ['metadata'],
}
ctx = context.RequestContext(fake.PROJECT_ID, fake.USER_ID, True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
fake_volume_obj = fake_volume.fake_volume_obj(ctx)
snapshot_obj = fake_snapshot.fake_snapshot_obj(self.ctx, **snapshot)
fake_volume_obj = fake_volume.fake_volume_obj(self.ctx)
snapshot_get_by_id.return_value = snapshot_obj
volume_get_by_id.return_value = fake_volume_obj
snapshot_id = UUID
req = fakes.HTTPRequest.blank('/v2/snapshots/%s' % snapshot_id)
req.environ['cinder.context'] = self.ctx
resp = self.controller.delete(req, snapshot_id)
self.assertEqual(HTTPStatus.ACCEPTED, resp.status_int)
@ -435,12 +435,12 @@ class SnapshotApiTest(test.TestCase):
'display_description': 'Default description',
'expected_attrs': ['metadata'],
}
ctx = context.RequestContext(fake.PROJECT_ID, fake.USER_ID, True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
fake_volume_obj = fake_volume.fake_volume_obj(ctx)
snapshot_obj = fake_snapshot.fake_snapshot_obj(self.ctx, **snapshot)
fake_volume_obj = fake_volume.fake_volume_obj(self.ctx)
snapshot_get_by_id.return_value = snapshot_obj
volume_get_by_id.return_value = fake_volume_obj
req = fakes.HTTPRequest.blank('/v2/snapshots/%s' % UUID)
req.environ['cinder.context'] = self.ctx
resp_dict = self.controller.show(req, UUID)
self.assertIn('snapshot', resp_dict)

View File

@ -130,6 +130,7 @@ class SnapshotMetaDataTest(test.TestCase):
self.req_id = str(uuid.uuid4())
self.url = '/v3/%s/snapshots/%s/metadata' % (
fake.PROJECT_ID, self.req_id)
self.ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
snap = {"volume_id": fake.VOLUME_ID,
"display_name": "Volume Test Name",
@ -137,6 +138,7 @@ class SnapshotMetaDataTest(test.TestCase):
"metadata": {}}
body = {"snapshot": snap}
req = fakes.HTTPRequest.blank('/v3/snapshots')
req.environ['cinder.context'] = self.ctx
self.snapshot_controller.create(req, body=body)
@mock.patch('cinder.objects.Snapshot.get_by_id')
@ -145,14 +147,14 @@ class SnapshotMetaDataTest(test.TestCase):
'id': self.req_id,
'expected_attrs': ['metadata']
}
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
snapshot_obj = fake_snapshot.fake_snapshot_obj(self.ctx, **snapshot)
snapshot_obj['metadata'] = {'key1': 'value1',
'key2': 'value2',
'key3': 'value3'}
snapshot_get_by_id.return_value = snapshot_obj
req = fakes.HTTPRequest.blank(self.url)
req.environ['cinder.context'] = self.ctx
res_dict = self.controller.index(req, self.req_id)
expected = {
@ -170,6 +172,7 @@ class SnapshotMetaDataTest(test.TestCase):
exception.SnapshotNotFound(snapshot_id=self.req_id)
req = fakes.HTTPRequest.blank(self.url)
req.environ['cinder.context'] = self.ctx
self.assertRaises(exception.SnapshotNotFound,
self.controller.index, req, self.url)
@ -179,11 +182,11 @@ class SnapshotMetaDataTest(test.TestCase):
'id': self.req_id,
'expected_attrs': ['metadata']
}
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
snapshot_obj = fake_snapshot.fake_snapshot_obj(self.ctx, **snapshot)
snapshot_get_by_id.return_value = snapshot_obj
req = fakes.HTTPRequest.blank(self.url)
req.environ['cinder.context'] = self.ctx
res_dict = self.controller.index(req, self.req_id)
expected = {'metadata': {}}
self.assertEqual(expected, res_dict)
@ -194,12 +197,12 @@ class SnapshotMetaDataTest(test.TestCase):
'id': self.req_id,
'expected_attrs': ['metadata']
}
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
snapshot_obj = fake_snapshot.fake_snapshot_obj(self.ctx, **snapshot)
snapshot_obj['metadata'] = {'key2': 'value2'}
snapshot_get_by_id.return_value = snapshot_obj
req = fakes.HTTPRequest.blank(self.url + '/key2')
req.environ['cinder.context'] = self.ctx
res_dict = self.controller.show(req, self.req_id, 'key2')
expected = {'meta': {'key2': 'value2'}}
self.assertEqual(expected, res_dict)
@ -210,6 +213,7 @@ class SnapshotMetaDataTest(test.TestCase):
exception.SnapshotNotFound(snapshot_id=self.req_id)
req = fakes.HTTPRequest.blank(self.url + '/key2')
req.environ['cinder.context'] = self.ctx
self.assertRaises(exception.SnapshotNotFound,
self.controller.show, req, self.req_id, 'key2')
@ -219,11 +223,11 @@ class SnapshotMetaDataTest(test.TestCase):
'id': self.req_id,
'expected_attrs': ['metadata']
}
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
snapshot_obj = fake_snapshot.fake_snapshot_obj(self.ctx, **snapshot)
snapshot_get_by_id.return_value = snapshot_obj
req = fakes.HTTPRequest.blank(self.url + '/key6')
req.environ['cinder.context'] = self.ctx
self.assertRaises(exception.SnapshotMetadataNotFound,
self.controller.show, req, self.req_id, 'key6')
@ -234,12 +238,12 @@ class SnapshotMetaDataTest(test.TestCase):
'id': self.req_id,
'expected_attrs': ['metadata']
}
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
snapshot_obj = fake_snapshot.fake_snapshot_obj(self.ctx, **snapshot)
snapshot_obj['metadata'] = {'key2': 'value2'}
snapshot_get_by_id.return_value = snapshot_obj
req = fakes.HTTPRequest.blank(self.url + '/key2')
req.environ['cinder.context'] = self.ctx
req.method = 'DELETE'
res = self.controller.delete(req, self.req_id, 'key2')
@ -249,6 +253,7 @@ class SnapshotMetaDataTest(test.TestCase):
self.mock_object(cinder.db, 'snapshot_get',
return_snapshot_nonexistent)
req = fakes.HTTPRequest.blank(self.url + '/key1')
req.environ['cinder.context'] = self.ctx
req.method = 'DELETE'
self.assertRaises(exception.SnapshotNotFound,
self.controller.delete, req, self.req_id, 'key1')
@ -259,11 +264,11 @@ class SnapshotMetaDataTest(test.TestCase):
'id': self.req_id,
'expected_attrs': ['metadata']
}
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
snapshot_obj = fake_snapshot.fake_snapshot_obj(self.ctx, **snapshot)
snapshot_get_by_id.return_value = snapshot_obj
req = fakes.HTTPRequest.blank(self.url + '/key6')
req.environ['cinder.context'] = self.ctx
req.method = 'DELETE'
self.assertRaises(exception.SnapshotMetadataNotFound,
self.controller.delete, req, self.req_id, 'key6')
@ -277,9 +282,8 @@ class SnapshotMetaDataTest(test.TestCase):
'id': self.req_id,
'expected_attrs': ['metadata']
}
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
fake_volume_obj = fake_volume.fake_volume_obj(ctx)
snapshot_obj = fake_snapshot.fake_snapshot_obj(self.ctx, **snapshot)
fake_volume_obj = fake_volume.fake_volume_obj(self.ctx)
snapshot_get_by_id.return_value = snapshot_obj
volume_get_by_id.return_value = fake_volume_obj
@ -287,6 +291,7 @@ class SnapshotMetaDataTest(test.TestCase):
return_create_snapshot_metadata)
req = fakes.HTTPRequest.blank('/v3/snapshot_metadata')
req.environ['cinder.context'] = self.ctx
req.method = 'POST'
req.content_type = "application/json"
body = {"metadata": {"key1": "value1",
@ -304,8 +309,7 @@ class SnapshotMetaDataTest(test.TestCase):
'id': self.req_id,
'expected_attrs': ['metadata']
}
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
snapshot_obj = fake_snapshot.fake_snapshot_obj(self.ctx, **snapshot)
snapshot_get_by_id.return_value = snapshot_obj
# if the keys in uppercase_and_lowercase, should return the one
@ -314,6 +318,7 @@ class SnapshotMetaDataTest(test.TestCase):
return_create_snapshot_metadata_insensitive)
req = fakes.HTTPRequest.blank('/v3/snapshot_metadata')
req.environ['cinder.context'] = self.ctx
req.method = 'POST'
req.content_type = "application/json"
body = {"metadata": {"key1": "value1",
@ -334,6 +339,7 @@ class SnapshotMetaDataTest(test.TestCase):
self.mock_object(cinder.db, 'snapshot_metadata_update',
return_create_snapshot_metadata)
req = fakes.HTTPRequest.blank(self.url)
req.environ['cinder.context'] = self.ctx
req.method = 'POST'
req.headers["content-type"] = "application/json"
@ -344,6 +350,7 @@ class SnapshotMetaDataTest(test.TestCase):
self.mock_object(cinder.db, 'snapshot_metadata_update',
return_create_snapshot_metadata)
req = fakes.HTTPRequest.blank(self.url + '/key1')
req.environ['cinder.context'] = self.ctx
req.method = 'PUT'
body = {"meta": {"": "value1"}}
req.body = jsonutils.dump_as_bytes(body)
@ -356,6 +363,7 @@ class SnapshotMetaDataTest(test.TestCase):
self.mock_object(cinder.db, 'snapshot_metadata_update',
return_create_snapshot_metadata)
req = fakes.HTTPRequest.blank(self.url + '/key1')
req.environ['cinder.context'] = self.ctx
req.method = 'PUT'
body = {"meta": {("a" * 260): "value1"}}
req.body = jsonutils.dump_as_bytes(body)
@ -372,6 +380,7 @@ class SnapshotMetaDataTest(test.TestCase):
return_create_snapshot_metadata)
req = fakes.HTTPRequest.blank('/v3/snapshot_metadata')
req.environ['cinder.context'] = self.ctx
req.method = 'POST'
req.content_type = "application/json"
body = {"metadata": {"key9": "value9"}}
@ -386,13 +395,13 @@ class SnapshotMetaDataTest(test.TestCase):
'id': self.req_id,
'expected_attrs': []
}
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
snapshot_obj = fake_snapshot.fake_snapshot_obj(self.ctx, **snapshot)
snapshot_get_by_id.return_value = snapshot_obj
self.mock_object(cinder.db, 'snapshot_metadata_update',
return_new_snapshot_metadata)
req = fakes.HTTPRequest.blank(self.url)
req.environ['cinder.context'] = self.ctx
req.method = 'PUT'
req.content_type = "application/json"
expected = {
@ -418,13 +427,13 @@ class SnapshotMetaDataTest(test.TestCase):
'id': self.req_id,
'expected_attrs': ['metadata']
}
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
snapshot_obj = fake_snapshot.fake_snapshot_obj(self.ctx, **snapshot)
snapshot_get_by_id.return_value = snapshot_obj
self.mock_object(cinder.db, 'snapshot_metadata_update',
return_new_snapshot_metadata)
req = fakes.HTTPRequest.blank(self.url)
req.environ['cinder.context'] = self.ctx
req.method = 'PUT'
req.content_type = "application/json"
body = {
@ -455,13 +464,13 @@ class SnapshotMetaDataTest(test.TestCase):
'id': self.req_id,
'expected_attrs': []
}
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
snapshot_obj = fake_snapshot.fake_snapshot_obj(self.ctx, **snapshot)
snapshot_get_by_id.return_value = snapshot_obj
self.mock_object(cinder.db, 'snapshot_metadata_update',
return_value={})
req = fakes.HTTPRequest.blank(self.url)
req.environ['cinder.context'] = self.ctx
req.method = 'PUT'
req.content_type = "application/json"
expected = {'metadata': {}}
@ -474,6 +483,7 @@ class SnapshotMetaDataTest(test.TestCase):
self.mock_object(cinder.db, 'snapshot_metadata_update',
return_create_snapshot_metadata)
req = fakes.HTTPRequest.blank(self.url)
req.environ['cinder.context'] = self.ctx
req.method = 'PUT'
req.content_type = "application/json"
expected = {'meta': {}}
@ -487,6 +497,7 @@ class SnapshotMetaDataTest(test.TestCase):
self.mock_object(cinder.db, 'snapshot_metadata_update',
return_create_snapshot_metadata)
req = fakes.HTTPRequest.blank(self.url)
req.environ['cinder.context'] = self.ctx
req.method = 'PUT'
req.content_type = "application/json"
expected = {'metadata': ['asdf']}
@ -500,6 +511,7 @@ class SnapshotMetaDataTest(test.TestCase):
self.mock_object(cinder.db, 'snapshot_get',
return_snapshot_nonexistent)
req = fakes.HTTPRequest.blank(self.url)
req.environ['cinder.context'] = self.ctx
req.method = 'PUT'
req.content_type = "application/json"
body = {'metadata': {'key10': 'value10'}}
@ -517,11 +529,11 @@ class SnapshotMetaDataTest(test.TestCase):
'id': self.req_id,
'expected_attrs': ['metadata']
}
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
snapshot_obj = fake_snapshot.fake_snapshot_obj(self.ctx, **snapshot)
snapshot_get_by_id.return_value = snapshot_obj
req = fakes.HTTPRequest.blank(self.url + '/key1')
req.environ['cinder.context'] = self.ctx
req.method = 'PUT'
body = {"meta": {"key1": "value1"}}
req.body = jsonutils.dump_as_bytes(body)
@ -535,6 +547,7 @@ class SnapshotMetaDataTest(test.TestCase):
return_snapshot_nonexistent)
req = fakes.HTTPRequest.blank(
'/v3/%s/snapshots/asdf/metadata/key1' % fake.PROJECT_ID)
req.environ['cinder.context'] = self.ctx
req.method = 'PUT'
body = {"meta": {"key1": "value1"}}
req.body = jsonutils.dump_as_bytes(body)
@ -548,6 +561,7 @@ class SnapshotMetaDataTest(test.TestCase):
self.mock_object(cinder.db, 'snapshot_metadata_update',
return_create_snapshot_metadata)
req = fakes.HTTPRequest.blank(self.url + '/key1')
req.environ['cinder.context'] = self.ctx
req.method = 'PUT'
req.headers["content-type"] = "application/json"
@ -560,6 +574,7 @@ class SnapshotMetaDataTest(test.TestCase):
def test_update_item_empty_key(self, metadata_update, snapshot_get):
snapshot_get.return_value = fake_get
req = fakes.HTTPRequest.blank(self.url + '/key1')
req.environ['cinder.context'] = self.ctx
req.method = 'PUT'
body = {"meta": {"": "value1"}}
req.body = jsonutils.dump_as_bytes(body)
@ -574,13 +589,13 @@ class SnapshotMetaDataTest(test.TestCase):
'id': self.req_id,
'expected_attrs': ['metadata']
}
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
snapshot_obj = fake_snapshot.fake_snapshot_obj(self.ctx, **snapshot)
snapshot_get_by_id.return_value = snapshot_obj
self.mock_object(cinder.db, 'snapshot_metadata_update',
return_create_snapshot_metadata)
req = fakes.HTTPRequest.blank(self.url + '/key1')
req.environ['cinder.context'] = self.ctx
req.method = 'PUT'
body = {"meta": {("a" * 260): "value1"}}
req.body = jsonutils.dump_as_bytes(body)
@ -596,13 +611,13 @@ class SnapshotMetaDataTest(test.TestCase):
'id': self.req_id,
'expected_attrs': ['metadata']
}
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
snapshot_obj = fake_snapshot.fake_snapshot_obj(self.ctx, **snapshot)
snapshot_get_by_id.return_value = snapshot_obj
self.mock_object(cinder.db, 'snapshot_metadata_update',
return_create_snapshot_metadata)
req = fakes.HTTPRequest.blank(self.url + '/key1')
req.environ['cinder.context'] = self.ctx
req.method = 'PUT'
body = {"meta": {"key1": ("a" * 260)}}
req.body = jsonutils.dump_as_bytes(body)
@ -622,10 +637,10 @@ class SnapshotMetaDataTest(test.TestCase):
}
self.mock_object(cinder.db, 'snapshot_metadata_update',
return_create_snapshot_metadata)
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
snapshot_obj = fake_snapshot.fake_snapshot_obj(self.ctx, **snapshot)
snapshot_get_by_id.return_value = snapshot_obj
req = fakes.HTTPRequest.blank(self.url + '/key1')
req.environ['cinder.context'] = self.ctx
req.method = 'PUT'
req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
@ -638,6 +653,7 @@ class SnapshotMetaDataTest(test.TestCase):
self.mock_object(cinder.db, 'snapshot_metadata_update',
return_create_snapshot_metadata)
req = fakes.HTTPRequest.blank(self.url + '/bad')
req.environ['cinder.context'] = self.ctx
req.method = 'PUT'
body = {"meta": {"key1": "value1"}}
req.body = jsonutils.dump_as_bytes(body)
@ -657,13 +673,13 @@ class SnapshotMetaDataTest(test.TestCase):
'id': self.req_id,
'expected_attrs': ['metadata']
}
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
snapshot_obj = fake_snapshot.fake_snapshot_obj(self.ctx, **snapshot)
snapshot_get_by_id.return_value = snapshot_obj
self.mock_object(cinder.db, 'snapshot_metadata_update',
return_create_snapshot_metadata)
req = fakes.HTTPRequest.blank(self.url)
req.environ['cinder.context'] = self.ctx
req.method = 'POST'
req.headers["content-type"] = "application/json"

View File

@ -94,12 +94,12 @@ class SnapshotApiTest(test.TestCase):
'expected_attrs': ['metadata'],
'group_snapshot_id': None,
}
ctx = context.RequestContext(fake.PROJECT_ID, fake.USER_ID, True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
fake_volume_obj = fake_volume.fake_volume_obj(ctx)
snapshot_obj = fake_snapshot.fake_snapshot_obj(self.ctx, **snapshot)
fake_volume_obj = fake_volume.fake_volume_obj(self.ctx)
snapshot_get_by_id.return_value = snapshot_obj
volume_get_by_id.return_value = fake_volume_obj
req = fakes.HTTPRequest.blank('/v3/snapshots/%s' % UUID)
req.environ['cinder.context'] = self.ctx
req.api_version_request = mv.get_api_version(max_ver)
resp_dict = self.controller.show(req, UUID)
@ -124,6 +124,7 @@ class SnapshotApiTest(test.TestCase):
def _create_snapshot(self, name=None, metadata=None):
"""Creates test snapshopt with provided metadata"""
req = fakes.HTTPRequest.blank('/v3/snapshots')
req.environ['cinder.context'] = self.ctx
snap = {"volume_id": fake.VOLUME_ID,
"display_name": name or "Volume Test Name",
"description": "Volume Test Desc"

View File

@ -0,0 +1,301 @@
# 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.
from unittest import mock
import ddt
from cinder.api.contrib import extended_snapshot_attributes as snapshot_attr
from cinder.api import microversions as mv
from cinder.api.v3 import snapshots
from cinder import exception
from cinder.policies import snapshots as 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 SnapshotsPolicyTest(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',
]
authorized_admins = [
'legacy_admin',
'system_admin',
'project_admin',
]
unauthorized_admins = [
'legacy_owner',
'system_member',
'system_reader',
'system_foo',
'project_member',
'project_reader',
'project_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 = snapshots.SnapshotsController()
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)
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_all_policy(self, user_id):
self._create_snapshot()
rule_name = policy.GET_ALL_POLICY
url = self.api_path
req = fake_api.HTTPRequest.blank(url, version=self.api_version)
# Generally, any logged in user can list all volumes.
authorized_readers = [user_id]
unauthorized_readers = []
# The exception is when deprecated rules are disabled, in which case
# roles are enforced. Users without the 'reader' role should be
# blocked.
if self.enforce_new_defaults:
context = self.create_context(user_id)
if 'reader' not in context.roles:
authorized_readers = []
unauthorized_readers = [user_id]
response = self.common_policy_check(user_id, authorized_readers,
unauthorized_readers,
self.unauthorized_exceptions,
rule_name, self.controller.index,
req)
# For some users, even if they're authorized, the list of snapshots
# will be empty if they are not in the snapshots's project.
empty_response_users = [
*self.unauthorized_readers,
# legacy_admin and system_admin do not have a project_id, and
# so the list of snapshots returned will be empty.
'legacy_admin',
'system_admin',
]
snapshots = response['snapshots'] if response else []
snapshot_count = 0 if user_id in empty_response_users else 1
self.assertEqual(snapshot_count, len(snapshots))
@ddt.data(*base.all_users)
def test_get_policy(self, user_id):
snapshot = self._create_snapshot()
rule_name = policy.GET_POLICY
url = '%s/%s' % (self.api_path, snapshot.id)
req = fake_api.HTTPRequest.blank(url, version=self.api_version)
self.common_policy_check(user_id, self.authorized_readers,
self.unauthorized_readers,
self.unauthorized_exceptions,
rule_name, self.controller.show, req,
id=snapshot.id)
@ddt.data(*base.all_users)
def test_extend_attribute_policy(self, user_id):
snapshot = self._create_snapshot()
rule_name = policy.EXTEND_ATTRIBUTE
url = '%s/%s' % (self.api_path, snapshot.id)
req = fake_api.HTTPRequest.blank(url, version=self.api_version)
snapshot_dict = snapshot.obj_to_primitive()['versioned_object.data']
req.get_db_snapshot = mock.MagicMock()
req.get_db_snapshot.return_value = snapshot_dict
resp_obj = mock.MagicMock(obj={'snapshot': snapshot_dict})
self.assertNotIn('os-extended-snapshot-attributes:project_id',
snapshot_dict.keys())
controller = snapshot_attr.ExtendedSnapshotAttributesController()
self.common_policy_check(user_id, self.authorized_readers,
self.unauthorized_readers,
self.unauthorized_exceptions,
rule_name, controller.show, req,
resp_obj=resp_obj,
id=snapshot.id, fatal=False)
if user_id in self.authorized_readers:
self.assertIn('os-extended-snapshot-attributes:project_id',
snapshot_dict.keys())
@ddt.data(*base.all_users)
def test_create_policy(self, user_id):
volume = self._create_volume()
rule_name = policy.CREATE_POLICY
url = self.api_path
req = fake_api.HTTPRequest.blank(url, version=self.api_version)
req.method = 'POST'
body = {
"snapshot": {
"name": "snap-001",
"volume_id": volume.id,
}
}
unauthorized_exceptions = [
exception.VolumeNotFound,
]
self.common_policy_check(user_id, self.authorized_members,
self.unauthorized_members,
unauthorized_exceptions,
rule_name, self.controller.create, req,
body=body)
@ddt.data(*base.all_users)
def test_update_policy(self, user_id):
snapshot = self._create_snapshot()
rule_name = policy.UPDATE_POLICY
url = '%s/%s' % (self.api_path, snapshot.id)
req = fake_api.HTTPRequest.blank(url, version=self.api_version)
req.method = 'PUT'
body = {
"snapshot": {
"description": "This is yet another snapshot."
}
}
# 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.update, req,
id=snapshot.id, body=body)
@ddt.data(*base.all_users)
def test_delete_policy(self, user_id):
snapshot = self._create_snapshot(status='available')
rule_name = policy.DELETE_POLICY
url = '%s/%s' % (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,
id=snapshot.id)
class SnapshotsPolicySecureRbacTest(SnapshotsPolicyTest):
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)

View File

@ -19,32 +19,6 @@
# DELETE /snapshots/{snapshot_id}/metadata/{key}
"volume:delete_snapshot_metadata": ""
# List snapshots.
# GET /snapshots
# GET /snapshots/detail
"volume:get_all_snapshots": ""
# List or show snapshots with extended attributes.
# GET /snapshots/{snapshot_id}
# GET /snapshots/detail
"volume_extension:extended_snapshot_attributes": ""
# Create snapshot.
# POST /snapshots
"volume:create_snapshot": ""
# Show snapshot.
# GET /snapshots/{snapshot_id}
"volume:get_snapshot": ""
# Update snapshot.
# PUT /snapshots/{snapshot_id}
"volume:update_snapshot": ""
# Delete snapshot.
# DELETE /snapshots/{snapshot_id}
"volume:delete_snapshot": ""
# List backups.
# GET /backups
# GET /backups/detail