Merge "Add policy check for downloading image."

doc/source/policies.rst

@@ -55,6 +55,10 @@ The actions that may have a rule enforced on them are:
 
 * ``manage_image_cache`` - Allowed to use the image cache management API
 
+* Added in v2:
+
+* ``download_image`` - Allowed to call the ``GET /images/<IMAGE_ID>/file`` API call
+
 
 To limit an action to a particular role or roles, you list the roles like so ::
 

glance/api/v1/images.py

@@ -265,6 +265,7 @@ class Controller(controller.BaseController):
         :raises HTTPNotFound if image is not available to user
         """
         self._enforce(req, 'get_image')
+        self._enforce(req, 'download_image')
         image_meta = self.get_active_image_meta_or_404(req, id)
 
         if image_meta.get('size') == 0:

glance/api/v2/image_data.py

@@ -16,6 +16,7 @@
 import webob.exc
 
 from glance.api import common
+from glance.api import policy
 import glance.api.v2 as v2
 from glance.common import exception
 from glance.common import utils
@@ -26,11 +27,12 @@ import glance.store
 
 
 class ImageDataController(object):
-    def __init__(self, db_api=None, store_api=None):
+    def __init__(self, db_api=None, store_api=None, policy_enforcer=None):
         self.db_api = db_api or glance.db.get_api()
         self.db_api.configure_db()
         self.store_api = store_api or glance.store
         self.store_api.create_stores()
+        self.policy = policy_enforcer or policy.Enforcer()
 
     def _get_image(self, context, image_id):
         try:
@@ -38,6 +40,13 @@ class ImageDataController(object):
         except exception.NotFound:
             raise webob.exc.HTTPNotFound(_("Image does not exist"))
 
+    def _enforce(self, req, action):
+        """Authorize an action against our policies"""
+        try:
+            self.policy.enforce(req.context, action, {})
+        except exception.Forbidden:
+            raise webob.exc.HTTPForbidden()
+
     @utils.mutating
     def upload(self, req, image_id, data, size):
         image = self._get_image(req.context, image_id)
@@ -53,6 +62,7 @@ class ImageDataController(object):
         self.db_api.image_update(req.context, image_id, values)
 
     def download(self, req, image_id):
+        self._enforce(req, 'download_image')
         ctx = req.context
         image = self._get_image(ctx, image_id)
         location = image['location']

glance/tests/unit/base.py

@@ -23,6 +23,9 @@ import stubout
 
 from glance.openstack.common import cfg
 from glance import store
+# NOTE(ameade): this import is necessary. Since we override a cfg opt it
+# registers we must have that opt loaded.
+from glance.store import filesystem
 from glance.store import location
 from glance.tests import stubs
 from glance.tests import utils as test_utils

glance/tests/unit/utils.py

@@ -114,7 +114,7 @@ class FakePolicyEnforcer(object):
     def __init__(self, *_args, **kwargs):
         self.rules = {}
 
-    def enforce(self, _ctxt, action, _target, **kwargs):
+    def enforce(self, _ctxt, action, target=None, **kwargs):
         """Raise Forbidden if a rule for given action is set to false."""
         if self.rules.get(action) is False:
             raise exception.Forbidden()

glance/tests/unit/v1/test_api.py

@@ -2952,6 +2952,13 @@ class TestGlanceAPI(base.IsolatedUnitTest):
         res = req.get_response(self.api)
         self.assertEqual(res.status_int, 403)
 
+    def test_show_image_unauthorized_download(self):
+        rules = {"download_image": [["false:false"]]}
+        self.set_policy_rules(rules)
+        req = webob.Request.blank("/images/%s" % UUID2)
+        res = req.get_response(self.api)
+        self.assertEqual(res.status_int, 403)
+
     def test_delete_image(self):
         req = webob.Request.blank("/images/%s" % UUID2)
         req.method = 'DELETE'

glance/tests/unit/v2/test_image_data_resource.py

@@ -32,7 +32,8 @@ class TestImagesController(base.StoreClearingUnitTest):
 
         self.controller = glance.api.v2.image_data.ImageDataController(
                 db_api=unit_test_utils.FakeDB(),
-                store_api=unit_test_utils.FakeStoreAPI())
+                store_api=unit_test_utils.FakeStoreAPI(),
+                policy_enforcer=unit_test_utils.FakePolicyEnforcer())
 
     def test_download(self):
         request = unit_test_utils.get_fake_request()
@@ -78,6 +79,24 @@ class TestImagesController(base.StoreClearingUnitTest):
         self.assertEqual('YYYY', output['data'])
 
 
+class TestImageDataControllerPolicies(base.IsolatedUnitTest):
+
+    def setUp(self):
+        super(TestImageDataControllerPolicies, self).setUp()
+        self.db = unit_test_utils.FakeDB()
+        self.policy = unit_test_utils.FakePolicyEnforcer()
+        self.controller = glance.api.v2.image_data.ImageDataController(
+                                                self.db,
+                                                policy_enforcer=self.policy)
+
+    def test_download_unauthorized(self):
+        rules = {"download_image": False}
+        self.policy.set_rules(rules)
+        request = unit_test_utils.get_fake_request()
+        self.assertRaises(webob.exc.HTTPForbidden, self.controller.download,
+                          request, image_id=unit_test_utils.UUID2)
+
+
 class TestImageDataDeserializer(test_utils.BaseTestCase):
 
     def setUp(self):