1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
|
import os
import errno
import shutil
import unittest
from django.core.files import temp as tempfile
from django.core.files.uploadedfile import SimpleUploadedFile
from django.test import TestCase, client
from django.utils import simplejson
from django.utils.hashcompat import sha_constructor
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_constructor(post_data[key].read()).hexdigest()
post_data[key].seek(0)
except AttributeError:
post_data[key + '_hash'] = sha_constructor(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")
|
