瀏覽代碼

[desktop] Only allow owners to change document permissions

Erick Tryzelaar 10 年之前
父節點
當前提交
45287b5
共有 3 個文件被更改,包括 124 次插入15 次删除
  1. 2 1
      desktop/core/src/desktop/api.py
  2. 113 7
      desktop/core/src/desktop/api_tests.py
  3. 9 7
      desktop/core/src/desktop/models.py

+ 2 - 1
desktop/core/src/desktop/api.py

@@ -288,8 +288,9 @@ def update_permissions(request):
     data = json.loads(request.POST['data'])
     doc_id = request.POST['doc_id']
     try:
-      doc = Document.objects.get_doc(doc_id, request.user)
+      doc = Document.objects.get_doc_for_writing(doc_id, request.user)
       doc.sync_permissions(data)
+
       response['message'] = _('Permissions updated!')
       response['status'] = 0
       response['doc'] = massage_doc_for_json(doc, request.user)

+ 113 - 7
desktop/core/src/desktop/api_tests.py

@@ -47,18 +47,44 @@ class TestDocModelTags():
 
   def add_tag(self, name):
     response = self.client.post("/desktop/api/tag/add_tag", {'name': name})
-    assert_equal(0, json.loads(response.content)['status'], response.content)
-    return json.loads(response.content)['id']
+    content = json.loads(response.content)
+    assert_equal(content['status'], 0, content)
+
+    return content['id']
 
   def add_doc(self, name):
     script = PigScript.objects.create(owner=self.user)
     doc = Document.objects.link(script, owner=script.owner, name=name)
     return script, doc
 
+  def share_doc(self, doc, permissions):
+    response = self.client.post("/desktop/api/doc/update_permissions", {
+        'doc_id': doc.id,
+        '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_add_tag(self):
     response = self.client.get("/desktop/api/tag/add_tag")
     assert_equal(-1, json.loads(response.content)['status'])
 
+    response = self.client.post("/desktop/api/tag/add_tag")
+    content = json.loads(response.content)
+    assert_equal(content['status'], -1, content)
+
     tag_id = self.add_tag('my_tag')
 
     assert_true(DocumentTag.objects.filter(id=tag_id, owner=self.user, tag='my_tag').exists())
@@ -91,8 +117,11 @@ class TestDocModelTags():
     response = self.client.get("/desktop/api/tag/remove_tag")
     assert_equal(-1, json.loads(response.content)['status'])
 
+    # Only the owner can remove tags.
     response = self.client_not_me.post("/desktop/api/tag/remove_tag", {'tag_id': tag_id})
-    assert_equal(-1, json.loads(response.content)['status'], response.content)
+    content = json.loads(response.content)
+    assert_equal(content['status'], -1, content)
+    assert_equal(content['message'], "DocumentTag matching query does not exist.", content)
 
     response = self.client.post("/desktop/api/tag/remove_tag", {'tag_id': tag_id})
     assert_equal(0, json.loads(response.content)['status'], response.content)
@@ -109,9 +138,27 @@ class TestDocModelTags():
     docs = _get_docs(self.user)
     assert_not_equal({}, massaged_documents_for_json(docs, self.user))
 
+  def test_tag_errors(self):
+    script, doc = self.add_doc('tag_pig_errors')
+
+    # Users without permission cannot see docs.
+    response = self.client_not_me.post("/desktop/api/doc/tag", {'data': json.dumps({'doc_id': doc.id, 'tag': 'pig'})})
+    content = json.loads(response.content)
+    assert_equal(content['status'], -1, content)
+    assert_equal(content['message'], "Document matching query does not exist.", content)
+
+    # Users with permission cannot tag docs.
+    self.share_doc_read_only(doc)
+
+    response = self.client_not_me.post("/desktop/api/doc/tag", {'data': json.dumps({'doc_id': doc.id, 'tag': 'pig'})})
+    content = json.loads(response.content)
+    assert_equal(content['status'], -1, content)
+    assert_equal(content['message'], "Document matching query does not exist.", content)
+
   def test_tag(self):
     script, doc = self.add_doc('tag_pig')
 
+    # Owners can add tags.
     response = self.client.post("/desktop/api/doc/tag", {'data': json.dumps({'doc_id': doc.id, 'tag': 'pig'})})
     assert_equal(0, json.loads(response.content)['status'], response.content)
 
@@ -137,16 +184,15 @@ class TestDocModelTags():
         {"id": tag2_id, "name": "update_tags_2"}
       ], sorted(content['doc']['tags'], key=lambda t: t['id']))
 
-    # No perms
+    # Only the owner can update tags.
     response = self.client_not_me.post("/desktop/api/doc/update_tags", {'data': json.dumps({'doc_id': doc.id, 'tag_ids': [tag1_id, tag2_id]})})
     content = json.loads(response.content)
-
-    assert_equal(-1, content['status'])
+    assert_equal(content['status'], -1, response.content)
+    assert_equal(content['message'], "Document matching query does not exist.", content)
 
     # todo no default tag on test user?
 
 
-
 class TestDocModelPermissions():
 
   def setUp(self):
@@ -438,3 +484,63 @@ class TestDocModelPermissions():
     assert_true(doc_id in json.loads(response.context['json_documents']))
     response = self.client_not_me.get('/home')
     assert_false(doc_id in json.loads(response.context['json_documents']))
+
+  def test_update_permissions_cannot_escalate_privileges(self):
+    script, doc = self._add_doc('test_update_permissions_cannot_escape_privileges')
+
+    # Share read permissions
+    response = self.client.post("/desktop/api/doc/update_permissions", {
+      'doc_id': doc.id,
+      'data': json.dumps({
+        'read': {
+          'user_ids': [
+            self.user.id,
+            self.user_not_me.id,
+          ],
+          'group_ids': []
+        },
+        'write': {
+          'user_ids': [
+            self.user.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_false(doc.can_write(self.user_not_me))
+
+    # Try, and fail to escalate privileges.
+    response = self.client_not_me.post("/desktop/api/doc/update_permissions", {
+      'doc_id': doc.id,
+      'data': json.dumps({
+        'read': {
+          'user_ids': [
+            self.user.id,
+            self.user_not_me.id,
+          ],
+          'group_ids': []
+        },
+        'write': {
+          'user_ids': [
+            self.user_not_me.id,
+            self.user_not_me.id,
+          ],
+          'group_ids': []
+        }
+      })
+    })
+
+    content = json.loads(response.content)
+    assert_equal(content['status'], -1)
+    assert_equal(content['message'], "Document does not exist or you don\'t have the permission to access it.")
+
+    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))

+ 9 - 7
desktop/core/src/desktop/models.py

@@ -96,8 +96,9 @@ class DocumentTagManager(models.Manager):
     except DocumentTag.DoesNotExist:
       tag = self._get_tag(user=owner, name=tag_name)
 
-    doc = Document.objects.get_doc(doc_id, owner)
+    doc = Document.objects.get_doc_for_writing(doc_id, owner)
     doc.add_tag(tag)
+
     return tag
 
   def untag(self, tag_id, owner, doc_id):
@@ -106,8 +107,7 @@ class DocumentTagManager(models.Manager):
     if tag.tag in DocumentTag.RESERVED:
       raise Exception(_("Can't remove %s: it is a reserved tag.") % tag)
 
-    doc = Document.objects.get_doc(doc_id, owner=owner)
-    doc.can_write_or_exception(owner)
+    doc = Document.objects.get_doc_for_writing(doc_id, owner=owner)
     doc.remove_tag(tag)
 
   def delete_tag(self, tag_id, owner):
@@ -123,8 +123,7 @@ class DocumentTagManager(models.Manager):
       doc.add_tag(default_tag)
 
   def update_tags(self, owner, doc_id, tag_ids):
-    doc = Document.objects.get_doc(doc_id, owner)
-    doc.can_write_or_exception(owner)
+    doc = Document.objects.get_doc_for_writing(doc_id, owner)
 
     for tag in doc.tags.all():
       if tag.tag not in DocumentTag.RESERVED:
@@ -183,8 +182,11 @@ class DocumentManager(models.Manager):
 
     return docs
 
-  def get_doc(self, doc_id, user):
-    return Document.objects.documents(user).get(id=doc_id)
+  def get_doc_for_writing(self, doc_id, user):
+    """Fetch a document and confirm that this user can write to it."""
+    doc = Document.objects.documents(user).get(id=doc_id)
+    doc.can_write_or_exception(user)
+    return doc
 
   def trashed_docs(self, model_class, user):
     tag = DocumentTag.objects.get_trash_tag(user=user)