Эх сурвалжийг харах

HUE-9096 [doc] Adding sharing and permission tests

Romain 6 жил өмнө
parent
commit
91c53a79a6

+ 2 - 2
desktop/core/src/desktop/api2.py

@@ -618,8 +618,8 @@ def share_document(request):
 
   Example of input: {'read': {'user_ids': [1, 2, 3], 'group_ids': [1, 2, 3]}}
   """
-  perms_dict = request.POST.get('data')
   uuid = request.POST.get('uuid')
+  perms_dict = request.POST.get('data')
 
   if not uuid or not perms_dict:
     raise PopupException(_('share_document requires uuid and perms_dict'))
@@ -657,8 +657,8 @@ def share_document_link(request):
 
   Example of input: {'name': 'link_read', 'is_link_on': true}
   """
-  perm = request.POST.get('data')
   uuid = request.POST.get('uuid')
+  perm = request.POST.get('data')
 
   if not uuid or not perm:
     raise PopupException(_('share_document_link requires uuid and permission data'))

+ 382 - 1
desktop/core/src/desktop/api2_tests.py

@@ -22,6 +22,8 @@ import re
 
 from nose.tools import assert_true, assert_false, assert_equal, assert_not_equal, assert_raises
 
+from useradmin.models import get_default_user_group
+
 from desktop.lib.django_test_util import make_logged_in_client
 from desktop.lib.test_utils import grant_access
 from desktop.models import Document2, User
@@ -37,7 +39,12 @@ class TestApi2(object):
     grant_access(self.user.username, self.user.username, "desktop")
 
   def test_search_entities_interactive_xss(self):
-    query = Document2.objects.create(name='<script>alert(5)</script>', description='<script>alert(5)</script>', type='query-hive', owner=self.user)
+    query = Document2.objects.create(
+        name='<script>alert(5)</script>',
+        description='<script>alert(5)</script>',
+        type='query-hive',
+        owner=self.user
+    )
 
     try:
       response = self.client.post('/desktop/api/search/entities_interactive/', data={
@@ -55,3 +62,377 @@ class TestApi2(object):
       assert_true('&gt;' in result_json, result_json)
     finally:
       query.delete()
+
+
+class TestDocumentApiSharingPermissions(object):
+
+  def setUp(self):
+    self.client = make_logged_in_client(username="perm_user", groupname="default", recreate=True, is_superuser=False)
+    self.client_not_me = make_logged_in_client(username="not_perm_user", groupname="default", recreate=True, is_superuser=False)
+
+    self.user = User.objects.get(username="perm_user")
+    self.user_not_me = User.objects.get(username="not_perm_user")
+
+    grant_access(self.user.username, self.user.username, "desktop")
+    grant_access(self.user_not_me.username, self.user_not_me.username, "desktop")
+
+
+  def _add_doc(self, name):
+    return Document2.objects.create(
+        name=name,
+        type='query-hive',
+        owner=self.user
+    )
+
+  def share_doc(self, doc, permissions, client=None):
+    if client is None:
+      client = self.client
+
+    return client.post("/desktop/api2/doc/share", {
+        'uuid': json.dumps(doc.uuid),
+        'data': json.dumps(permissions)
+    })
+
+  def share_doc_read_only(self, doc):
+    return self.share_doc(doc, {
+      'read': {
+        'user_ids': [
+          self.user.id
+        ],
+        'group_ids': []
+      },
+      'write': {
+        'user_ids': [],
+        'group_ids': []
+      }
+    })
+
+  def test_update_permissions(self):
+    doc = self._add_doc('test_update_permissions')
+
+    response = self.share_doc(
+        doc,
+        {
+          'read': {
+            'user_ids': [self.user_not_me.id],
+            'group_ids': []
+          }
+        }
+    )
+
+    assert_equal(0, json.loads(response.content)['status'], response.content)
+
+  def test_share_document_permissions(self):
+    # No doc
+    response = self.client.get('/desktop/api2/docs/')
+    data = json.loads(response.content)
+    assert_false(data['documents'])
+
+    response = self.client_not_me.get('/desktop/api2/docs/')
+    data = json.loads(response.content)
+    assert_false(data['documents'])
+
+    # Add doc
+    doc = self._add_doc('test_update_permissions')
+    doc_id = '%s' % doc.id
+
+    response = self.client.get('/desktop/api2/docs/')
+    data = json.loads(response.content)
+    assert_true(data['documents'])
+
+    response = self.client_not_me.get('/desktop/api2/docs/')
+    data = json.loads(response.content)
+    assert_false(data['documents'])
+
+    assert_true(doc.can_read(self.user))
+    assert_true(doc.can_write(self.user))
+    assert_false(doc.can_read(self.user_not_me))
+    assert_false(doc.can_write(self.user_not_me))
+
+    # Share by user
+    response = self.share_doc(doc, {
+        'read': {
+          'user_ids': [
+            self.user_not_me.id
+          ],
+          'group_ids': []
+        },
+        'write': {
+          'user_ids': [],
+          'group_ids': []
+        }
+      }
+    )
+
+    assert_equal(0, json.loads(response.content)['status'], response.content)
+
+    assert_true(doc.can_read(self.user))
+    assert_true(doc.can_write(self.user))
+    assert_true(doc.can_read(self.user_not_me))
+    assert_false(doc.can_write(self.user_not_me))
+
+    response = self.client.get('/desktop/api2/docs/')
+    data = json.loads(response.content)
+    assert_true(data['documents'])
+
+    response = self.client_not_me.get('/desktop/api2/docs/')
+    data = json.loads(response.content)
+    assert_true(data['documents'])
+
+    # Un-share
+    response = self.share_doc(doc, {
+        'read': {
+          'user_ids': [
+          ],
+          'group_ids': []
+        },
+        'write': {
+          'user_ids': [],
+          'group_ids': []
+        }
+      }
+    )
+
+    assert_equal(0, json.loads(response.content)['status'], response.content)
+
+    assert_true(doc.can_read(self.user))
+    assert_true(doc.can_write(self.user))
+    assert_false(doc.can_read(self.user_not_me))
+    assert_false(doc.can_write(self.user_not_me))
+
+    response = self.client.get('/desktop/api2/docs/')
+    data = json.loads(response.content)
+    assert_true(data['documents'])
+
+    response = self.client_not_me.get('/desktop/api2/docs/')
+    data = json.loads(response.content)
+    assert_false(data['documents'])
+
+    # Share by group
+    default_group = get_default_user_group()
+
+    response = self.share_doc(doc, {
+        'read': {
+          'user_ids': [
+          ],
+          'group_ids': []
+        },
+        'write': {
+          'user_ids': [],
+          'group_ids': [default_group.id]
+        }
+      }
+    )
+
+    assert_equal(0, json.loads(response.content)['status'], response.content)
+
+    assert_true(doc.can_read(self.user))
+    assert_true(doc.can_write(self.user))
+    assert_true(doc.can_read(self.user_not_me))
+    assert_true(doc.can_write(self.user_not_me))
+
+    response = self.client.get('/desktop/api2/docs/')
+    data = json.loads(response.content)
+    assert_true(data['documents'])
+
+    response = self.client_not_me.get('/desktop/api2/docs/')
+    data = json.loads(response.content)
+    assert_true(data['documents'])
+
+    # Un-share
+    response = self.share_doc(doc, {
+        'read': {
+          'user_ids': [
+          ],
+          'group_ids': []
+        },
+        'write': {
+          'user_ids': [],
+          'group_ids': []
+        }
+      }
+    )
+
+    assert_equal(0, json.loads(response.content)['status'], response.content)
+
+    assert_true(doc.can_read(self.user))
+    assert_true(doc.can_write(self.user))
+    assert_false(doc.can_read(self.user_not_me))
+    assert_false(doc.can_write(self.user_not_me))
+
+    response = self.client.get('/desktop/api2/docs/')
+    data = json.loads(response.content)
+    assert_true(data['documents'])
+
+    response = self.client_not_me.get('/desktop/api2/docs/')
+    data = json.loads(response.content)
+    assert_false(data['documents'])
+
+    # Modify by other user
+    response = self.share_doc(doc, {
+        'read': {
+          'user_ids': [
+          ],
+          'group_ids': []
+        },
+        'write': {
+          'user_ids': [self.user_not_me.id],
+          'group_ids': []
+        }
+      }
+    )
+
+    assert_equal(0, json.loads(response.content)['status'], response.content)
+
+    assert_true(doc.can_read(self.user))
+    assert_true(doc.can_write(self.user))
+    assert_true(doc.can_read(self.user_not_me))
+    assert_true(doc.can_write(self.user_not_me))
+
+    response = self.client.get('/desktop/api2/docs/')
+    data = json.loads(response.content)
+    assert_true(data['documents'])
+
+    response = self.client_not_me.get('/desktop/api2/docs/')
+    data = json.loads(response.content)
+    assert_true(data['documents'])
+
+    # Un-share
+    response = self.share_doc(doc, {
+        'read': {
+          'user_ids': [
+          ],
+          'group_ids': []
+        },
+        'write': {
+          'user_ids': [],
+          'group_ids': []
+        }
+      }
+    )
+
+    assert_equal(0, json.loads(response.content)['status'], response.content)
+
+    assert_true(doc.can_read(self.user))
+    assert_true(doc.can_write(self.user))
+    assert_false(doc.can_read(self.user_not_me))
+    assert_false(doc.can_write(self.user_not_me))
+
+    response = self.client.get('/desktop/api2/docs/')
+    data = json.loads(response.content)
+    assert_true(data['documents'])
+
+    response = self.client_not_me.get('/desktop/api2/docs/')
+    data = json.loads(response.content)
+    assert_false(data['documents'])
+
+    # Modify by group
+    response = self.share_doc(doc, {
+        'read': {
+          'user_ids': [
+          ],
+          'group_ids': []
+        },
+        'write': {
+          'user_ids': [],
+          'group_ids': [default_group.id]
+        }
+      }
+    )
+
+    assert_equal(0, json.loads(response.content)['status'], response.content)
+
+    assert_true(doc.can_read(self.user))
+    assert_true(doc.can_write(self.user))
+    assert_true(doc.can_read(self.user_not_me))
+    assert_true(doc.can_write(self.user_not_me))
+
+    response = self.client.get('/desktop/api2/docs/')
+    data = json.loads(response.content)
+    assert_true(data['documents'])
+
+    response = self.client_not_me.get('/desktop/api2/docs/')
+    data = json.loads(response.content)
+    assert_true(data['documents'])
+
+    # Un-share
+    response = self.share_doc(doc, {
+        'read': {
+          'user_ids': [
+          ],
+          'group_ids': []
+        },
+        'write': {
+          'user_ids': [],
+          'group_ids': []
+        }
+      }
+    )
+
+    assert_equal(0, json.loads(response.content)['status'], response.content)
+
+    assert_true(doc.can_read(self.user))
+    assert_true(doc.can_write(self.user))
+    assert_false(doc.can_read(self.user_not_me))
+    assert_false(doc.can_write(self.user_not_me))
+
+    response = self.client.get('/desktop/api2/docs/')
+    data = json.loads(response.content)
+    assert_true(data['documents'])
+
+    response = self.client_not_me.get('/desktop/api2/docs/')
+    data = json.loads(response.content)
+    assert_false(data['documents'])
+
+
+  def test_update_permissions_cannot_escalate_privileges(self):
+    doc = self._add_doc('test_update_permissions_cannot_escape_privileges')
+
+    # Share read permissions
+    response = self.share_doc(doc, {
+        'read': {
+          'user_ids': [
+            self.user_not_me.id
+          ],
+          'group_ids': []
+        },
+        'write': {
+          'user_ids': [],
+          'group_ids': []
+        }
+      }
+    )
+
+    assert_equal(0, json.loads(response.content)['status'], response.content)
+
+    assert_true(doc.can_read(self.user))
+    assert_true(doc.can_write(self.user))
+    assert_true(doc.can_read(self.user_not_me))
+    assert_false(doc.can_write(self.user_not_me))
+
+    # Try, and fail to escalate privileges.
+    response = self.share_doc(doc, {
+        'read': {
+          'user_ids': [
+            self.user_not_me.id
+          ],
+          'group_ids': []
+        },
+        'write': {
+          'user_ids': [
+            self.user_not_me.id,
+          ],
+          'group_ids': []
+        }
+      },
+      self.client_not_me
+    )
+
+    content = json.loads(response.content)
+    assert_equal(content['status'], -1)
+    assert_true("Document does not exist or you don\'t have the permission to access it." in content['message'], content['message'])
+
+    assert_true(doc.can_read(self.user))
+    assert_true(doc.can_write(self.user))
+    assert_true(doc.can_read(self.user_not_me))
+    assert_false(doc.can_write(self.user_not_me))