import os import errno import sha import shutil import tempfile import unittest from django.core.files.uploadedfile import SimpleUploadedFile from django.test import TestCase, client from django.utils import simplejson from models import FileModel, UPLOAD_ROOT, UPLOAD_TO class FileUploadTests(TestCase): def test_simple_upload(self): post_data = { 'name': 'Ringo', 'file_field': open(__file__), } response = self.client.post('/file_uploads/upload/', post_data) self.assertEqual(response.status_code, 200) def test_large_upload(self): tdir = tempfile.gettempdir() file1 = tempfile.NamedTemporaryFile(suffix=".file1", dir=tdir) file1.write('a' * (2 ** 21)) file1.seek(0) file2 = tempfile.NamedTemporaryFile(suffix=".file2", dir=tdir) file2.write('a' * (10 * 2 ** 20)) file2.seek(0) # This file contains chinese symbols for a name. file3 = open(os.path.join(tdir, u'test_中文_Orl\u00e9ans.jpg'.encode('utf-8')), 'w+b') file3.write('b' * (2 ** 10)) file3.seek(0) post_data = { 'name': 'Ringo', 'file_field1': open(file1.name), 'file_field2': open(file2.name), 'file_unicode': file3, } for key in post_data.keys(): try: post_data[key + '_hash'] = sha.new(post_data[key].read()).hexdigest() post_data[key].seek(0) except AttributeError: post_data[key + '_hash'] = sha.new(post_data[key]).hexdigest() response = self.client.post('/file_uploads/verify/', post_data) try: os.unlink(file3.name) except: pass self.assertEqual(response.status_code, 200) def test_dangerous_file_names(self): """Uploaded file names should be sanitized before ever reaching the view.""" # This test simulates possible directory traversal attacks by a # malicious uploader We have to do some monkeybusiness here to construct # a malicious payload with an invalid file name (containing os.sep or # os.pardir). This similar to what an attacker would need to do when # trying such an attack. scary_file_names = [ "/tmp/hax0rd.txt", # Absolute path, *nix-style. "C:\\Windows\\hax0rd.txt", # Absolute path, win-syle. "C:/Windows/hax0rd.txt", # Absolute path, broken-style. "\\tmp\\hax0rd.txt", # Absolute path, broken in a different way. "/tmp\\hax0rd.txt", # Absolute path, broken by mixing. "subdir/hax0rd.txt", # Descendant path, *nix-style. "subdir\\hax0rd.txt", # Descendant path, win-style. "sub/dir\\hax0rd.txt", # Descendant path, mixed. "../../hax0rd.txt", # Relative path, *nix-style. "..\\..\\hax0rd.txt", # Relative path, win-style. "../..\\hax0rd.txt" # Relative path, mixed. ] payload = [] for i, name in enumerate(scary_file_names): payload.extend([ '--' + client.BOUNDARY, 'Content-Disposition: form-data; name="file%s"; filename="%s"' % (i, name), 'Content-Type: application/octet-stream', '', 'You got pwnd.' ]) payload.extend([ '--' + client.BOUNDARY + '--', '', ]) payload = "\r\n".join(payload) r = { 'CONTENT_LENGTH': len(payload), 'CONTENT_TYPE': client.MULTIPART_CONTENT, 'PATH_INFO': "/file_uploads/echo/", 'REQUEST_METHOD': 'POST', 'wsgi.input': client.FakePayload(payload), } response = self.client.request(**r) # The filenames should have been sanitized by the time it got to the view. recieved = simplejson.loads(response.content) for i, name in enumerate(scary_file_names): got = recieved["file%s" % i] self.assertEqual(got, "hax0rd.txt") def test_filename_overflow(self): """File names over 256 characters (dangerous on some platforms) get fixed up.""" name = "%s.txt" % ("f"*500) payload = "\r\n".join([ '--' + client.BOUNDARY, 'Content-Disposition: form-data; name="file"; filename="%s"' % name, 'Content-Type: application/octet-stream', '', 'Oops.' '--' + client.BOUNDARY + '--', '', ]) r = { 'CONTENT_LENGTH': len(payload), 'CONTENT_TYPE': client.MULTIPART_CONTENT, 'PATH_INFO': "/file_uploads/echo/", 'REQUEST_METHOD': 'POST', 'wsgi.input': client.FakePayload(payload), } got = simplejson.loads(self.client.request(**r).content) self.assert_(len(got['file']) < 256, "Got a long file name (%s characters)." % len(got['file'])) def test_custom_upload_handler(self): # A small file (under the 5M quota) smallfile = tempfile.NamedTemporaryFile() smallfile.write('a' * (2 ** 21)) # A big file (over the quota) bigfile = tempfile.NamedTemporaryFile() bigfile.write('a' * (10 * 2 ** 20)) # Small file posting should work. response = self.client.post('/file_uploads/quota/', {'f': open(smallfile.name)}) got = simplejson.loads(response.content) self.assert_('f' in got) # Large files don't go through. response = self.client.post("/file_uploads/quota/", {'f': open(bigfile.name)}) got = simplejson.loads(response.content) self.assert_('f' not in got) def test_broken_custom_upload_handler(self): f = tempfile.NamedTemporaryFile() f.write('a' * (2 ** 21)) # AttributeError: You cannot alter upload handlers after the upload has been processed. self.assertRaises( AttributeError, self.client.post, '/file_uploads/quota/broken/', {'f': open(f.name)} ) def test_fileupload_getlist(self): file1 = tempfile.NamedTemporaryFile() file1.write('a' * (2 ** 23)) file2 = tempfile.NamedTemporaryFile() file2.write('a' * (2 * 2 ** 18)) file2a = tempfile.NamedTemporaryFile() file2a.write('a' * (5 * 2 ** 20)) response = self.client.post('/file_uploads/getlist_count/', { 'file1': open(file1.name), 'field1': u'test', 'field2': u'test3', 'field3': u'test5', 'field4': u'test6', 'field5': u'test7', 'file2': (open(file2.name), open(file2a.name)) }) got = simplejson.loads(response.content) self.assertEqual(got.get('file1'), 1) self.assertEqual(got.get('file2'), 2) class DirectoryCreationTests(unittest.TestCase): """ Tests for error handling during directory creation via _save_FIELD_file (ticket #6450) """ def setUp(self): self.obj = FileModel() if not os.path.isdir(UPLOAD_ROOT): os.makedirs(UPLOAD_ROOT) def tearDown(self): os.chmod(UPLOAD_ROOT, 0700) shutil.rmtree(UPLOAD_ROOT) def test_readonly_root(self): """Permission errors are not swallowed""" os.chmod(UPLOAD_ROOT, 0500) try: self.obj.save_testfile_file('foo.txt', SimpleUploadedFile('foo.txt', 'x')) except OSError, err: self.assertEquals(err.errno, errno.EACCES) except: self.fail("OSError [Errno %s] not raised" % errno.EACCES) def test_not_a_directory(self): """The correct IOError is raised when the upload directory name exists but isn't a directory""" # Create a file with the upload directory name fd = open(UPLOAD_TO, 'w') fd.close() try: self.obj.save_testfile_file('foo.txt', SimpleUploadedFile('foo.txt', 'x')) except IOError, err: # The test needs to be done on a specific string as IOError # is raised even without the patch (just not early enough) self.assertEquals(err.args[0], "%s exists and is not a directory" % UPLOAD_TO) except: self.fail("IOError not raised")