Merge pull request #18 from Kinto/7-check-permissions

CHANGELOG.rst

@@ -15,6 +15,7 @@ Changelog
 **Bug fixes**
 
 - Only index records if the storage transaction is committed (fixes #15)
+- Do not allow to search if no read permission on collection or bucket (fixes #7)
 
 **Internal changes**
 

kinto_elasticsearch/views.py

@@ -1,17 +1,29 @@
 import logging
 
 import elasticsearch
+from kinto.core import authorization
 from kinto.core import Service
+from kinto.core import utils
 
 
 logger = logging.getLogger(__name__)
 
+
+class RouteFactory(authorization.RouteFactory):
+    def __init__(self, request):
+        super().__init__(request)
+        records_plural = utils.strip_uri_prefix(request.path.replace("/search", "/records"))
+        self.permission_object_id = records_plural
+        self.required_permission = "read"
+
+
 search = Service(name="search",
                  path='/buckets/{bucket_id}/collections/{collection_id}/search',
-                 description="Search")
+                 description="Search",
+                 factory=RouteFactory)
 
 
-@search.post()
+@search.post(permission=authorization.DYNAMIC)
 def get_search(request):
     bucket_id = request.matchdict['bucket_id']
     collection_id = request.matchdict['collection_id']

tests/__init__.py

@@ -19,6 +19,7 @@ def __init__(self, *args, **kwargs):
         self.headers.update(get_user_headers('mat'))
 
     def tearDown(self):
+        super().tearDown()
         self.app.app.registry.indexer.flush()
 
     @classmethod

tests/test_elasticsearch.py

@@ -4,6 +4,7 @@
 
 import elasticsearch
 from kinto.core import utils as core_utils
+from kinto.core.testing import get_user_headers
 
 from kinto_elasticsearch import __version__ as elasticsearch_version
 from . import BaseWebTest
@@ -74,6 +75,8 @@ def test_response_is_served_if_indexer_fails(self):
 
 class SearchView(BaseWebTest, unittest.TestCase):
     def test_search_response_is_empty_if_indexer_fails(self):
+        self.app.put("/buckets/bid", headers=self.headers)
+        self.app.put("/buckets/bid/collections/cid", headers=self.headers)
         with mock.patch("kinto_elasticsearch.indexer.Indexer.search",
                         side_effect=elasticsearch.ElasticsearchException):
             resp = self.app.post("/buckets/bid/collections/cid/search",
@@ -88,3 +91,51 @@ def test_search_on_empty_collection_returns_empty_list(self):
                              headers=self.headers)
         result = resp.json
         assert len(result["hits"]["hits"]) == 0
+
+
+class PermissionsCheck(BaseWebTest, unittest.TestCase):
+    def test_search_is_allowed_if_write_on_bucket(self):
+        body = {"permissions": {"write": ["system.Everyone"]}}
+        self.app.put_json("/buckets/bid", body, headers=self.headers)
+        self.app.put("/buckets/bid/collections/cid", headers=self.headers)
+
+        self.app.post("/buckets/bid/collections/cid/search", status=200)
+
+    def test_search_is_allowed_if_read_on_bucket(self):
+        body = {"permissions": {"read": ["system.Everyone"]}}
+        self.app.put_json("/buckets/bid", body, headers=self.headers)
+        self.app.put("/buckets/bid/collections/cid", headers=self.headers)
+
+        self.app.post("/buckets/bid/collections/cid/search", status=200)
+
+    def test_search_is_allowed_if_write_on_collection(self):
+        self.app.put("/buckets/bid", headers=self.headers)
+        body = {"permissions": {"write": ["system.Everyone"]}}
+        self.app.put_json("/buckets/bid/collections/cid", body, headers=self.headers)
+
+        self.app.post("/buckets/bid/collections/cid/search", status=200)
+
+    def test_search_is_allowed_if_read_on_collection(self):
+        self.app.put("/buckets/bid", headers=self.headers)
+        body = {"permissions": {"read": ["system.Everyone"]}}
+        self.app.put_json("/buckets/bid/collections/cid", body, headers=self.headers)
+
+        self.app.post("/buckets/bid/collections/cid/search", status=200)
+
+    def test_search_is_not_allowed_by_default(self):
+        self.app.put("/buckets/bid", headers=self.headers)
+        self.app.put("/buckets/bid/collections/cid", headers=self.headers)
+
+        self.app.post("/buckets/bid/collections/cid/search", status=401)
+        headers = get_user_headers("cual", "quiera")
+        self.app.post("/buckets/bid/collections/cid/search", status=403, headers=headers)
+
+    def test_search_is_not_allowed_if_only_read_on_certain_records(self):
+        self.app.put("/buckets/bid", headers=self.headers)
+        body = {"permissions": {"record:create": ["system.Authenticated"]}}
+        self.app.put_json("/buckets/bid/collections/cid", body, headers=self.headers)
+        headers = get_user_headers("toto")
+        self.app.post_json("/buckets/bid/collections/cid/records", {"data": {"pi": 42}},
+                           headers=headers)
+
+        self.app.post("/buckets/bid/collections/cid/search", status=403, headers=headers)