6 files changed, 266 insertions, 0 deletions
diff --git a/tests/regressiontests/file_uploads/__init__.py b/tests/regressiontests/file_uploads/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
--- /dev/null
+++ b/tests/regressiontests/file_uploads/__init__.py
diff --git a/tests/regressiontests/file_uploads/models.py b/tests/regressiontests/file_uploads/models.py
new file mode 100644
index 0000000000..2d5607b2a7
--- /dev/null
+++ b/tests/regressiontests/file_uploads/models.py
@@ -0,0 +1,2 @@
+# This file unintentionally left blank.
+# Oops.
+\ No newline at end of file
diff --git a/tests/regressiontests/file_uploads/tests.py b/tests/regressiontests/file_uploads/tests.py
new file mode 100644
index 0000000000..8992298470
--- /dev/null
+++ b/tests/regressiontests/file_uploads/tests.py
@@ -0,0 +1,158 @@
+import os
+import sha
+import tempfile
+from django.test import TestCase, client
+from django.utils import simplejson
+
+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_&#20013;&#25991;_Orl\u00e9ans.jpg'), '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)}
+        )        
+        
+\ No newline at end of file
diff --git a/tests/regressiontests/file_uploads/uploadhandler.py b/tests/regressiontests/file_uploads/uploadhandler.py
new file mode 100644
index 0000000000..54f82f626c
--- /dev/null
+++ b/tests/regressiontests/file_uploads/uploadhandler.py
@@ -0,0 +1,26 @@
+"""
+Upload handlers to test the upload API.
+"""
+
+from django.core.files.uploadhandler import FileUploadHandler, StopUpload
+
+class QuotaUploadHandler(FileUploadHandler):
+    """
+    This test upload handler terminates the connection if more than a quota
+    (5MB) is uploaded.
+    """
+    
+    QUOTA = 5 * 2**20 # 5 MB
+    
+    def __init__(self, request=None):
+        super(QuotaUploadHandler, self).__init__(request)
+        self.total_upload = 0
+        
+    def receive_data_chunk(self, raw_data, start):
+        self.total_upload += len(raw_data)
+        if self.total_upload >= self.QUOTA:
+            raise StopUpload(connection_reset=True)
+        return raw_data
+            
+    def file_complete(self, file_size):
+        return None
+\ No newline at end of file
diff --git a/tests/regressiontests/file_uploads/urls.py b/tests/regressiontests/file_uploads/urls.py
new file mode 100644
index 0000000000..529bee312d
--- /dev/null
+++ b/tests/regressiontests/file_uploads/urls.py
@@ -0,0 +1,10 @@
+from django.conf.urls.defaults import *
+import views
+
+urlpatterns = patterns('',
+    (r'^upload/$',          views.file_upload_view),
+    (r'^verify/$',          views.file_upload_view_verify),
+    (r'^echo/$',            views.file_upload_echo),
+    (r'^quota/$',           views.file_upload_quota),
+    (r'^quota/broken/$',    views.file_upload_quota_broken),
+)
diff --git a/tests/regressiontests/file_uploads/views.py b/tests/regressiontests/file_uploads/views.py
new file mode 100644
index 0000000000..833cf90531
--- /dev/null
+++ b/tests/regressiontests/file_uploads/views.py
@@ -0,0 +1,70 @@
+import os
+import sha
+from django.core.files.uploadedfile import UploadedFile
+from django.http import HttpResponse, HttpResponseServerError
+from django.utils import simplejson
+from uploadhandler import QuotaUploadHandler
+
+def file_upload_view(request):
+    """
+    Check that a file upload can be updated into the POST dictionary without
+    going pear-shaped.
+    """
+    form_data = request.POST.copy()
+    form_data.update(request.FILES)
+    if isinstance(form_data.get('file_field'), UploadedFile) and isinstance(form_data['name'], unicode):
+        # If a file is posted, the dummy client should only post the file name,
+        # not the full path.
+        if os.path.dirname(form_data['file_field'].file_name) != '':
+            return HttpResponseServerError()            
+        return HttpResponse('')
+    else:
+        return HttpResponseServerError()
+
+def file_upload_view_verify(request):
+    """
+    Use the sha digest hash to verify the uploaded contents.
+    """
+    form_data = request.POST.copy()
+    form_data.update(request.FILES)
+
+    # Check to see if unicode names worked out.
+    if not request.FILES['file_unicode'].file_name.endswith(u'test_\u4e2d\u6587_Orl\xe9ans.jpg'):
+        return HttpResponseServerError()
+
+    for key, value in form_data.items():
+        if key.endswith('_hash'):
+            continue
+        if key + '_hash' not in form_data:
+            continue
+        submitted_hash = form_data[key + '_hash']
+        if isinstance(value, UploadedFile):
+            new_hash = sha.new(value.read()).hexdigest()
+        else:
+            new_hash = sha.new(value).hexdigest()
+        if new_hash != submitted_hash:
+            return HttpResponseServerError()
+
+    return HttpResponse('')
+
+def file_upload_echo(request):
+    """
+    Simple view to echo back info about uploaded files for tests.
+    """
+    r = dict([(k, f.file_name) for k, f in request.FILES.items()])
+    return HttpResponse(simplejson.dumps(r))
+    
+def file_upload_quota(request):
+    """
+    Dynamically add in an upload handler.
+    """
+    request.upload_handlers.insert(0, QuotaUploadHandler())
+    return file_upload_echo(request)
+        
+def file_upload_quota_broken(request):
+    """
+    You can't change handlers after reading FILES; this view shouldn't work.
+    """
+    response = file_upload_echo(request)
+    request.upload_handlers.insert(0, QuotaUploadHandler())
+    return response
+\ No newline at end of file