Set upload_image policy to control data upload

There was no policy to control data upload.
Up until today, the add_image policy was a all or nothing, from metadata
to the actual data. Now, with the combination of add_image and upload_image
an administrator will have finer control over the whole chain.

Change-Id: I1a7966ffb5c63dd8239a54fe2963b031d9fe1f9a
Closes-bug: #1254521
This commit is contained in:
Yanis Guenane 2013-11-26 15:55:09 +00:00
parent 7ca0174d8c
commit ab7ea6baf7
3 changed files with 105 additions and 3 deletions

View File

@ -52,6 +52,8 @@ The actions that may have a rule enforced on them are:
* ``upload_image`` - Upload binary image data
* ``POST /v1/images``
* ``PUT /v1/images/<IMAGE_ID>``
* ``PUT /v2/images/<IMAGE_ID>/file``
* ``add_image`` - Create an image entity

View File

@ -747,6 +747,8 @@ class Controller(controller.BaseController):
self._enforce(req, 'publicize_image')
if Controller._copy_from(req):
self._enforce(req, 'copy_from')
if image_data or Controller._copy_from(req):
self._enforce(req, 'upload_image')
self._enforce_create_protected_props(image_meta['properties'].keys(),
req)
@ -782,6 +784,8 @@ class Controller(controller.BaseController):
is_public = image_meta.get('is_public')
if is_public:
self._enforce(req, 'publicize_image')
if image_data or Controller._copy_from(req):
self._enforce(req, 'upload_image')
orig_image_meta = self.get_image_meta_or_404(req, id)
orig_status = orig_image_meta['status']

View File

@ -690,7 +690,7 @@ class TestGlanceAPI(base.IsolatedUnitTest):
def test_add_publicize_image_authorized(self):
rules = {"add_image": '@', "modify_image": '@',
"publicize_image": '@'}
"publicize_image": '@', "upload_image": '@'}
self.set_policy_rules(rules)
fixture_headers = {'x-image-meta-store': 'file',
'x-image-meta-is-public': 'true',
@ -727,8 +727,26 @@ class TestGlanceAPI(base.IsolatedUnitTest):
res = req.get_response(self.api)
self.assertEqual(res.status_int, 403)
def test_add_copy_from_image_authorized(self):
rules = {"add_image": '@', "copy_from": '@'}
def test_add_copy_from_upload_image_unauthorized(self):
rules = {"add_image": '@', "copy_from": '@', "upload_image": '!'}
self.set_policy_rules(rules)
fixture_headers = {'x-image-meta-store': 'file',
'x-image-meta-disk-format': 'vhd',
'x-glance-api-copy-from': 'http://glance.com/i.ovf',
'x-image-meta-container-format': 'ovf',
'x-image-meta-name': 'fake image #F'}
req = webob.Request.blank("/images")
req.method = 'POST'
for k, v in fixture_headers.iteritems():
req.headers[k] = v
req.headers['Content-Type'] = 'application/octet-stream'
res = req.get_response(self.api)
self.assertEqual(res.status_int, 403)
def test_add_copy_from_image_authorized_upload_image_authorized(self):
rules = {"add_image": '@', "copy_from": '@', "upload_image": '@'}
self.set_policy_rules(rules)
fixture_headers = {'x-image-meta-store': 'file',
'x-image-meta-disk-format': 'vhd',
@ -816,6 +834,84 @@ class TestGlanceAPI(base.IsolatedUnitTest):
res = req.get_response(self.api)
self.assertEqual(res.status_int, 400)
def test_add_copy_from_upload_image_unauthorized(self):
rules = {"upload_image": '!', "modify_image": '@',
"add_image": '@'}
self.set_policy_rules(rules)
self.config(image_size_cap=512)
fixture_headers = {
'x-image-meta-name': 'fake image #3',
'x-image-meta-container_format': 'ami',
'x-image-meta-disk_format': 'ami',
'transfer-encoding': 'chunked',
'content-type': 'application/octet-stream',
}
req = webob.Request.blank("/images")
req.method = 'POST'
req.body_file = StringIO.StringIO('X' * (CONF.image_size_cap))
for k, v in fixture_headers.iteritems():
req.headers[k] = v
res = req.get_response(self.api)
self.assertEqual(res.status_int, 403)
def test_update_data_upload_image_unauthorized(self):
rules = {"upload_image": '!', "modify_image": '@',
"add_image": '@'}
self.set_policy_rules(rules)
"""Tests creates a queued image for no body and no loc header"""
self.config(image_size_cap=512)
fixture_headers = {'x-image-meta-store': 'file',
'x-image-meta-name': 'fake image #3'}
req = webob.Request.blank("/images")
req.method = 'POST'
for k, v in fixture_headers.iteritems():
req.headers[k] = v
res = req.get_response(self.api)
self.assertEqual(res.status_int, 201)
res_body = json.loads(res.body)['image']
self.assertEqual('queued', res_body['status'])
image_id = res_body['id']
req = webob.Request.blank("/images/%s" % image_id)
req.method = 'PUT'
req.headers['Content-Type'] = 'application/octet-stream'
req.headers['transfer-encoding'] = 'chunked'
req.headers['x-image-disk-format'] = 'vhd'
req.headers['x-image-container-format'] = 'ovf'
req.body_file = StringIO.StringIO('X' * (CONF.image_size_cap))
res = req.get_response(self.api)
self.assertEqual(res.status_int, 403)
def test_update_copy_from_upload_image_unauthorized(self):
rules = {"upload_image": '!', "modify_image": '@',
"add_image": '@', "copy_from": '@'}
self.set_policy_rules(rules)
fixture_headers = {'x-image-meta-disk-format': 'vhd',
'x-image-meta-container-format': 'ovf',
'x-image-meta-name': 'fake image #3'}
req = webob.Request.blank("/images")
req.method = 'POST'
for k, v in fixture_headers.iteritems():
req.headers[k] = v
res = req.get_response(self.api)
self.assertEqual(res.status_int, 201)
res_body = json.loads(res.body)['image']
self.assertEqual('queued', res_body['status'])
image_id = res_body['id']
req = webob.Request.blank("/images/%s" % image_id)
req.method = 'PUT'
req.headers['Content-Type'] = 'application/octet-stream'
req.headers['x-glance-api-copy-from'] = 'http://glance.com/i.ovf'
res = req.get_response(self.api)
self.assertEqual(res.status_int, 403)
def _do_test_post_image_content_missing_format(self, missing):
"""Tests creation of an image with missing format"""
fixture_headers = {'x-image-meta-store': 'file',