瀏覽代碼

[doc2] Apply permission checks to doc2 API and add tests

Jenny Kim 9 年之前
父節點
當前提交
dcc5c57
共有 3 個文件被更改,包括 111 次插入38 次删除
  1. 17 2
      desktop/core/src/desktop/api2.py
  2. 38 36
      desktop/core/src/desktop/models.py
  3. 56 0
      desktop/core/src/desktop/tests_doc2.py

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

@@ -79,7 +79,8 @@ def get_documents(request):
   else:  # Find by path
     document = Document2.objects.get_by_path(user=request.user, path=path)
 
-  # TODO perms
+  # Check if user has read permissions
+  document.can_read_or_exception(request.user)
 
   # Get querystring filters if any
   page = int(request.GET.get('page', 1))
@@ -122,6 +123,9 @@ def get_document(request):
   else:
     doc = Document2.objects.get_by_uuid(uuid=request.GET['uuid'])
 
+  # Check if user has read permissions
+  doc.can_read_or_exception(request.user)
+
   return JsonResponse(doc.to_dict())
 
 
@@ -136,6 +140,11 @@ def move_document(request):
 
   source = Directory.objects.get_by_uuid(uuid=source_doc_uuid)
   destination = Directory.objects.get_by_uuid(uuid=destination_doc_uuid)
+
+  # Check if user has write permissions for both source and destination
+  source.can_write_or_exception(request.user)
+  destination.can_write_or_exception(request.user)
+
   doc = source.move(destination, request.user)
 
   return JsonResponse({
@@ -154,7 +163,10 @@ def create_directory(request):
     raise PopupException(_('create_directory requires parent_uuid and name'))
 
   parent_dir = Directory.objects.get_by_uuid(uuid=parent_uuid)
-  # TODO: Check permissions and move to manager
+
+  # Check if user has write permissions for parent directory
+  parent_dir.can_write_or_exception(request.user)
+
   directory = Directory.objects.create(name=name, owner=request.user, parent_directory=parent_dir)
 
   return JsonResponse({
@@ -183,6 +195,9 @@ def delete_document(request):
 
   document = Document2.objects.get_by_uuid(uuid=uuid)
 
+  # Check if user has write permissions for given document
+  document.can_write_or_exception(request.user)
+
   if skip_trash:
     # TODO: check if document is in the .Trash folder, if not raise exception
     if document.is_directory and document.has_children:

+ 38 - 36
desktop/core/src/desktop/models.py

@@ -777,6 +777,7 @@ class Document2Manager(models.Manager):
     """
     Since UUID is not a unique field, but part of a composite unique key, this returns the latest version by UUID
     This should always be used in place of Document2.objects.get(uuid=) when a single document is expected
+    WARNING: This does not check for read/write pernissions!
     """
     docs = self.filter(uuid=uuid).order_by('-last_modified')
     if not docs.exists():
@@ -795,6 +796,7 @@ class Document2Manager(models.Manager):
   def get_by_path(self, user, path):
     """
     This can be an expensive operation b/c we have to traverse the path tree, so if possible, request a document by UUID
+    NOTE: get_by_path only works for the owner's documents since it is based off the user's home directory
     """
     cleaned_path = path.rstrip('/')
     doc = Document2.objects.get_home_directory(user)
@@ -956,34 +958,6 @@ class Document2(models.Model):
       'absoluteUrl': self.get_absolute_url()
     }
 
-  def can_read(self, user):
-    has_read_permissions = False
-    perm = self.list_permissions('read')
-    if perm:
-      has_read_permissions = perm.groups.filter(id__in=user.groups.all()).exists() or user in perm.users.all()
-    return user.is_superuser or self.owner == user or self.can_write(user) or has_read_permissions
-
-  def can_read_or_exception(self, user):
-    if self.doc.get():
-      self.doc.get().can_read_or_exception(user)
-    elif self.can_read(user):
-      return True
-    else:
-      raise PopupException(_("Document does not exist or you don't have the permission to access it."))
-
-  def can_write(self, user):
-    has_write_permissions = False
-    perm = self.list_permissions('write')
-    if perm:
-      has_write_permissions = perm.groups.filter(id__in=user.groups.all()).exists() or user in perm.users.all()
-    return user.is_superuser or self.owner == user or has_write_permissions
-
-  def can_write_or_exception(self, user):
-    if self.can_write(user):
-      return True
-    else:
-      raise PopupException(_("Document does not exist or you don't have the permission to access it."))
-
   def get_history(self):
     return self.dependencies.filter(is_history=True).order_by('-last_modified')
 
@@ -1054,6 +1028,34 @@ class Document2(models.Model):
 
   # TODO: restore
 
+  def can_read(self, user):
+    perm = self.get_permission('read')
+    has_read_permissions = perm.user_has_access(user) if perm else False
+    return user.is_superuser or self.owner == user or self.can_write(user) or has_read_permissions
+
+  def can_read_or_exception(self, user):
+    if self.can_read(user):
+      return True
+    else:
+      raise PopupException(_("Document does not exist or you don't have the permission to access it."))
+
+  def can_write(self, user):
+    perm = self.get_permission('write')
+    has_write_permissions = perm.user_has_access(user) if perm else False
+    return user.is_superuser or self.owner == user or has_write_permissions
+
+  def can_write_or_exception(self, user):
+    if self.can_write(user):
+      return True
+    else:
+      raise PopupException(_("Document does not exist or you don't have the permission to access it."))
+
+  def get_permission(self, perm='read'):
+    try:
+      return Document2Permission.objects.get(doc=self, perms=perm)
+    except Document2Permission.DoesNotExist:
+      return None
+
   def share(self, user, name='read', users=None, groups=None, all=False):
     with transaction.atomic():
       self.update_permission(user, name, users, groups, all)
@@ -1079,12 +1081,6 @@ class Document2(models.Model):
 
     perm.save()
 
-  def list_permissions(self, perm='read'):
-    try:
-      return Document2Permission.objects.get(doc=self, perms=perm)
-    except Document2Permission.DoesNotExist:
-      return None
-
   def _massage_permissions(self):
     """
     Returns the permissions for a given document as a dictionary
@@ -1094,8 +1090,8 @@ class Document2(models.Model):
       'write': {'users': [], 'groups': [], 'all': False}
     }
 
-    read_perms = self.list_permissions(perm='read')
-    write_perms = self.list_permissions(perm='write')
+    read_perms = self.get_permission(perm='read')
+    write_perms = self.get_permission(perm='write')
 
     if read_perms:
       permissions.update(read_perms.to_dict())
@@ -1200,6 +1196,12 @@ class Document2Permission(models.Model):
       }
     }
 
+  def user_has_access(self, user):
+    """
+    Returns true if the given user has permissions based on users, groups, or all flag
+    """
+    return self.groups.filter(id__in=user.groups.all()).exists() or user in self.users.all() or self.all
+
 
 def get_data_link(meta):
   link = None

+ 56 - 0
desktop/core/src/desktop/tests_doc2.py

@@ -92,6 +92,12 @@ class TestDocument2(object):
     assert_true('uuid' in data)
     assert_equal(doc.uuid, data['uuid'])
 
+    # Invalid uuid returns error
+    response = self.client.get('/desktop/api2/doc/get', {'uuid': '1234-5678-9'})
+    data = json.loads(response.content)
+    assert_equal(-1, data['status'])
+    assert_true('not found' in data['message'])
+
 
   def test_directory_create(self):
     response = self.client.post('/desktop/api2/doc/mkdir', {'parent_uuid': json.dumps(self.home_dir.uuid), 'name': json.dumps('test_mkdir')})
@@ -283,6 +289,16 @@ class TestDocument2Permissions(object):
   def test_share_document_read_by_user(self):
     doc = Document2.objects.create(name='new_doc', type='query-hive', owner=self.user, data={}, parent_directory=self.home_dir)
 
+    # owner can view document
+    response = self.client.get('/desktop/api2/doc/get', {'uuid': doc.uuid})
+    data = json.loads(response.content)
+    assert_equal(doc.uuid, data['uuid'], data)
+
+    # other user cannot view document
+    response = self.client_not_me.get('/desktop/api2/doc/get', {'uuid': doc.uuid})
+    data = json.loads(response.content)
+    assert_equal(-1, data['status'])
+
     # Share read perm by users
     response = self.client.post("/desktop/api2/doc/share", {
       'uuid': json.dumps(doc.uuid),
@@ -309,10 +325,25 @@ class TestDocument2Permissions(object):
     assert_true(doc.can_read(self.user_not_me))
     assert_false(doc.can_write(self.user_not_me))
 
+    # other user can view document
+    response = self.client_not_me.get('/desktop/api2/doc/get', {'uuid': doc.uuid})
+    data = json.loads(response.content)
+    assert_equal(doc.uuid, data['uuid'], data)
+
 
   def test_share_document_read_by_group(self):
     doc = Document2.objects.create(name='new_doc', type='query-hive', owner=self.user, data={}, parent_directory=self.home_dir)
 
+    # owner can view document
+    response = self.client.get('/desktop/api2/doc/get', {'uuid': doc.uuid})
+    data = json.loads(response.content)
+    assert_equal(doc.uuid, data['uuid'], data)
+
+    # other user cannot view document
+    response = self.client_not_me.get('/desktop/api2/doc/get', {'uuid': doc.uuid})
+    data = json.loads(response.content)
+    assert_equal(-1, data['status'])
+
     response = self.client.post("/desktop/api2/doc/share", {
       'uuid': json.dumps(doc.uuid),
       'data': json.dumps({
@@ -337,10 +368,20 @@ class TestDocument2Permissions(object):
     assert_true(doc.can_read(self.user_not_me))
     assert_false(doc.can_write(self.user_not_me))
 
+    # other user can view document
+    response = self.client_not_me.get('/desktop/api2/doc/get', {'uuid': doc.uuid})
+    data = json.loads(response.content)
+    assert_equal(doc.uuid, data['uuid'], data)
+
 
   def test_share_document_write_by_user(self):
     doc = Document2.objects.create(name='new_doc', type='query-hive', owner=self.user, data={}, parent_directory=self.home_dir)
 
+    # other user cannot modify document
+    response = self.client_not_me.post('/desktop/api2/doc/delete', {'uuid': json.dumps(doc.uuid)})
+    data = json.loads(response.content)
+    assert_equal(-1, data['status'])
+
     # Share write perm by user
     response = self.client.post("/desktop/api2/doc/share", {
       'uuid': json.dumps(doc.uuid),
@@ -366,10 +407,20 @@ class TestDocument2Permissions(object):
     assert_true(doc.can_read(self.user_not_me))
     assert_true(doc.can_write(self.user_not_me))
 
+    # other user can modify document
+    response = self.client_not_me.post('/desktop/api2/doc/delete', {'uuid': json.dumps(doc.uuid)})
+    data = json.loads(response.content)
+    assert_equal(0, data['status'])
+
 
   def test_share_document_write_by_group(self):
     doc = Document2.objects.create(name='new_doc', type='query-hive', owner=self.user, data={}, parent_directory=self.home_dir)
 
+    # other user cannot modify document
+    response = self.client_not_me.post('/desktop/api2/doc/delete', {'uuid': json.dumps(doc.uuid)})
+    data = json.loads(response.content)
+    assert_equal(-1, data['status'])
+
     # Share write perm by group
     response = self.client.post("/desktop/api2/doc/share", {
       'uuid': json.dumps(doc.uuid),
@@ -395,6 +446,11 @@ class TestDocument2Permissions(object):
     assert_true(doc.can_read(self.user_not_me))
     assert_true(doc.can_write(self.user_not_me))
 
+    # other user can modify document
+    response = self.client_not_me.post('/desktop/api2/doc/delete', {'uuid': json.dumps(doc.uuid)})
+    data = json.loads(response.content)
+    assert_equal(0, data['status'])
+
 
   def test_share_directory(self):
     # Test that updating the permissions for a directory updates all nested documents accordingly, with file structure: