[ci] Implement missing backend tests for admin file handler

Author: evilaliv3

backend/globaleaks/handlers/admin/file.py

@@ -108,13 +108,15 @@ class FileInstance(BaseHandler):
     ]
 
     def permission_check(self, name):
-        if name in ['css', 'favicon', 'script'] and \
-                not self.session.has_permission('can_upload_files'):
-            raise errors.InvalidAuthentication
+        if self.session.role == 'admin':
+            if name not in ['logo'] and \
+                    not self.session.has_permission('can_upload_files'):
+                raise errors.InvalidAuthentication
 
-        if self.session.role != 'admin' and \
-                not self.session.has_permission('can_edit_general_settings'):
-            raise errors.InvalidAuthentication
+        else:
+            if name not in ['logo'] or \
+                    not self.session.has_permission('can_edit_general_settings'):
+                raise errors.InvalidAuthentication
 
     @inlineCallbacks
     def post(self, name):
@@ -130,7 +132,7 @@ def post(self, name):
             self.allowed_mimetypes = ['text/javascript']
 
         if self.uploaded_file['type'] not in self.allowed_mimetypes:
-            raise errors.ForbiddenOperation
+            raise errors.InputValidationError
 
         if name in special_files or re.match(requests.uuid_regexp, name):
             self.uploaded_file['name'] = name

backend/globaleaks/tests/handlers/admin/test_file.py

@@ -2,48 +2,78 @@
 from twisted.internet.defer import inlineCallbacks
 
 from globaleaks.handlers.admin import file
+from globaleaks.rest import errors
 from globaleaks.tests import helpers
 
+files = [
+    {'handler': 'css', 'name': 'file.css'},
+    {'handler': 'script', 'name': 'file.js'},
+    {'handler': 'logo', 'name': 'file.png'},
+    {'handler': 'favicon', 'name': 'file.ico'},
+    {'handler': 'custom', 'name': 'file.txt'},
+]
+
 
 class TestFileInstance(helpers.TestHandler):
     _handler = file.FileInstance
 
     @inlineCallbacks
-    def test_post(self):
-        handler = self.request({}, role='admin')
+    def test_post_prevent_unauthorized_admin_uploads(self):
+        for f in files:
+            handler = self.request({}, role='admin', attachment=self.get_dummy_attachment(f['name']))
+            if f['handler'] == 'logo':
+                yield handler.post(f['handler'])
+            else:
+                yield self.assertFailure(handler.post(f['handler']), errors.InvalidAuthentication)
+
+    @inlineCallbacks
+    def test_post_prevent_wrong_filetypes(self):
+        permissions = {'can_upload_files': True}
 
-        yield handler.post(u'file.pdf')
+        for f in files:
+            handler = self.request({}, role='admin', permissions=permissions, attachment=self.get_dummy_attachment(f['name'] + ".wrong.ext"))
+            yield self.assertFailure(handler.post(f['handler']), errors.InputValidationError)
 
     @inlineCallbacks
-    def test_delete(self):
-        handler = self.request({}, role='admin')
-        yield handler.delete(u'file.pdf')
+    def test_post_accepts_correct_files(self):
+        permissions = {'can_upload_files': True}
 
+        for f in files:
+            handler = self.request({}, role='admin', permissions=permissions, attachment=self.get_dummy_attachment(f['name']))
+            yield handler.post(f['handler'])
 
-class TestFileCollection(helpers.TestHandler):
-    _handler = file.FileCollection
+    @inlineCallbacks
+    def test_post_prevent_unauthorized_recipients_to_upload_any_file(self):
+        for f in files:
+            handler = self.request({}, role='receiver', attachment=self.get_dummy_attachment(f['name']))
+            yield self.assertFailure(handler.post(f['handler']), errors.InvalidAuthentication)
 
     @inlineCallbacks
-    def test_get(self):
-        self._handler = file.FileInstance
-        handler = self.request({}, role='admin')
-        yield handler.post(u'custom')
+    def test_post_enable_authorized_recipients_to_upload_the_logo_and_only_it(self):
+        permissions = {'can_edit_general_settings': True}
 
-        self._handler = file.FileCollection
-        handler = self.request(role='admin')
-        response = yield handler.get()
+        for f in files:
+            handler = self.request({}, role='receiver', permissions=permissions, attachment=self.get_dummy_attachment(f['name']))
+            if f['handler'] == 'logo':
+                yield handler.post(f['handler'])
+            else:
+                yield self.assertFailure(handler.post(f['handler']), errors.InvalidAuthentication)
 
-        self.assertEqual(len(response), 1)
 
+class TestFileCollection(helpers.TestHandler):
+    _handler = file.FileCollection
+
+    @inlineCallbacks
+    def test_get(self):
         self._handler = file.FileInstance
-        handler = self.request({}, role='admin')
-        yield handler.post(u'custom')
 
-        handler = self.request({}, role='admin')
-        yield handler.post(u'custom')
+        permissions = {'can_upload_files': True}
+        for f in files:
+            handler = self.request({}, role='admin', permissions=permissions, attachment=self.get_dummy_attachment(f['name']))
+            yield handler.post(f['handler'])
 
         self._handler = file.FileCollection
-        handler = self.request(role='admin')
+        handler = self.request(role='admin', permissions=permissions)
         response = yield handler.get()
 
-        self.assertEqual(len(response), 3)
+        self.assertEqual(len(response), 5)

backend/globaleaks/tests/handlers/recipient/test_rfile.py

@@ -6,7 +6,7 @@
 from globaleaks.jobs.delivery import Delivery
 from globaleaks.tests import helpers
 
-attachment = b'hello world'
+file_content = b'Hello World'
 
 
 class TestWBFileWorkFlow(helpers.TestHandlerWithPopulatedDB):
@@ -19,7 +19,8 @@ def test_get(self):
         self._handler = rtip.ReceiverFileUpload
         rtips_desc = yield self.get_rtips()
         for rtip_desc in rtips_desc:
-            handler = self.request(role='receiver', user_id=rtip_desc['receiver_id'], attached_file=attachment)
+            attachment = self.get_dummy_attachment(content=file_content)
+            handler = self.request(role='receiver', user_id=rtip_desc['receiver_id'], attachment=attachment)
             yield handler.post(rtip_desc['id'])
 
         yield Delivery().run()
@@ -31,7 +32,7 @@ def test_get(self):
             for rfile_desc in rfiles_desc:
                 handler = self.request(role='whistleblower', user_id=wbtip_desc['id'])
                 yield handler.get(rfile_desc['id'])
-                self.assertEqual(handler.request.getResponseBody(), attachment)
+                self.assertEqual(handler.request.getResponseBody(), file_content)
 
         self._handler = rtip.ReceiverFileDownload
         rtips_desc = yield self.get_rtips()

backend/globaleaks/tests/handlers/test_file.py

@@ -16,7 +16,7 @@ def test_get(self):
         yield self.assertFailure(handler.get(u'custom'), ResourceNotFound)
 
         self._handler = admin_file.FileInstance
-        handler = self.request({}, role='admin')
+        handler = self.request({}, role='admin', permissions={'can_upload_files': True})
         yield handler.post('custom')
 
         self._handler = admin_file.FileCollection

backend/globaleaks/tests/helpers.py

@@ -3,6 +3,7 @@
 Utilities and basic TestCases.
 """
 import json
+import mimetypes
 import os
 import shutil
 
@@ -421,13 +422,14 @@ def __init__(self):
         }
 
 
-def get_dummy_file(content=None):
-    filename = generateRandomKey() + ".pdf"
+def get_dummy_attachment(name=None, content=None):
+    if name is None:
+        name = generateRandomKey() + ".pdf"
 
-    content_type = 'application/pdf'
+    content_type, _ = mimetypes.guess_type(name)
 
     if content is None:
-        content = Base64Encoder.decode(VALID_BASE64_IMG)
+        content = name.encode()
 
     temporary_file = SecureTemporaryFile(Settings.tmp_path)
 
@@ -437,9 +439,9 @@ def get_dummy_file(content=None):
     State.TempUploadFiles[os.path.basename(temporary_file.filepath)] = temporary_file
 
     return {
-        'id': filename,
+        'id': name,
         'date': datetime_now(),
-        'name': filename,
+        'name': name,
         'description': 'description',
         'body': temporary_file,
         'size': len(content),
@@ -459,7 +461,7 @@ def check_confirmation(self):
 
 
 def get_file_upload(self):
-    return get_dummy_file()
+    return get_dummy_attachment()
 
 
 BaseHandler.get_file_upload = get_file_upload
@@ -718,8 +720,8 @@ def get_dummy_submission(self, context_id):
             'receipt': GCE.derive_key(GCE.generate_receipt(), VALID_SALT)
         })
 
-    def get_dummy_file(self, content=None):
-        return get_dummy_file(content)
+    def get_dummy_attachment(self, name=None, content=None):
+        return get_dummy_attachment(name=name, content=content)
 
     def get_dummy_redirect(self, x=''):
         return {
@@ -732,7 +734,7 @@ def emulate_file_upload(self, session, n):
         This emulates the file upload of an incomplete submission
         """
         for _ in range(n):
-            session.files.append(self.get_dummy_file())
+            session.files.append(self.get_dummy_attachment())
 
     def pollute_events(self, number_of_times=10):
         for _ in range(number_of_times):
@@ -886,7 +888,7 @@ def perform_submission_start(self):
 
     def perform_submission_uploads(self, submission_id):
         for _ in range(self.population_of_attachments):
-            Sessions.get(submission_id).files.append(self.get_dummy_file())
+            Sessions.get(submission_id).files.append(self.get_dummy_attachment())
 
     @inlineCallbacks
     def perform_submission_actions(self, session_id):
@@ -973,14 +975,16 @@ class TestHandler(TestGLWithPopulatedDB):
     #
     #  }
     # }
+    can_upload_files = True
 
     def setUp(self):
         return TestGL.setUp(self)
 
     def request(self, body='', uri=b'https://www.globaleaks.org/',
-                user_id=None, role=None, multilang=False, headers=None, args=None,
-                client_addr=b'127.0.0.1', handler_cls=None, attached_file=None,
-                kwargs=None, token=False):
+                user_id=None, role=None, multilang=False, headers=None, token=False, permissions=None,
+                client_addr=b'127.0.0.1',
+                handler_cls=None, attachment=None,
+                args=None, kwargs=None):
         """
         Constructs a handler for preforming mock requests using the bag of params described below.
         """
@@ -1012,6 +1016,9 @@ def request(self, body='', uri=b'https://www.globaleaks.org/',
             else:
                 session = Sessions.new(1, user_id, 1, role, USER_PRV_KEY, USER_ESCROW_PRV_KEY if role == 'admin' else '')
 
+            if permissions:
+                session.permissions = permissions
+
             headers[b'x-session'] = session.id
 
         # during unit tests a token is always provided to any handler
@@ -1041,7 +1048,7 @@ def request(self, body='', uri=b'https://www.globaleaks.org/',
             request.language = None
 
         if handler.upload_handler:
-            handler.uploaded_file = self.get_dummy_file(attached_file)
+            handler.uploaded_file = attachment if attachment else self.get_dummy_attachment()
 
         return handler