Merge pull request #22 from freelamb/codex/analyze-and-resolve-issues

Fix multipart upload parsing edge cases

Author: freelamb

CHANGELOG.md

@@ -1,3 +1,11 @@
+<a name="0.3.4"></a>
+## [0.3.4] (2026-06-03)
+### Bug Fixes
+
+* preserve uploads that omit an optional part Content-Type header
+* avoid truncating uploads whose content contains boundary-like text
+* remove partial files when uploads end unexpectedly
+
 <a name="0.3.3"></a>
 ## [0.3.3] (2026-06-03)
 ### Security

simple_http_server.py

@@ -6,7 +6,7 @@
 and HEAD requests in a fairly straightforward manner.
 """
 
-__version__ = "0.3.3"
+__version__ = "0.3.4"
 __author__ = "yangyongbao@126.com"
 __all__ = ["SimpleHTTPRequestHandler"]
 
@@ -117,53 +117,82 @@ def deal_post_data(self):
             remain_bytes = int(content_length)
         except ValueError:
             return False, "Invalid content-length header"
+        if remain_bytes < 0:
+            return False, "Invalid content-length header"
         if remain_bytes > self.max_upload_size:
             return False, "Upload exceeds the %d byte limit" % self.max_upload_size
-        line = self.rfile.readline()
-        remain_bytes -= len(line)
-        if boundary not in line:
+
+        def read_upload_line():
+            line = self.rfile.readline()
+            return line, len(line)
+
+        def is_boundary_line(line):
+            stripped = line.rstrip(b'\r\n')
+            delimiter = b'--' + boundary
+            return stripped == delimiter or stripped == delimiter + b'--'
+
+        line, line_length = read_upload_line()
+        remain_bytes -= line_length
+        if not line or not is_boundary_line(line):
             return False, "Content NOT begin with boundary"
-        line = self.rfile.readline()
-        remain_bytes -= len(line)
-        fn = re.findall(r'Content-Disposition.*name="file"; filename="(.*)"', line.decode('utf-8', 'replace'))
+
+        part_headers = []
+        while remain_bytes > 0:
+            line, line_length = read_upload_line()
+            remain_bytes -= line_length
+            if not line:
+                return False, "Unexpected end of multipart headers"
+            if line in (b'\r\n', b'\n'):
+                break
+            part_headers.append(line.decode('utf-8', 'replace'))
+        else:
+            return False, "Unexpected end of multipart headers"
+
+        header_text = "".join(part_headers)
+        fn = re.findall(r'Content-Disposition.*name="file"; filename="([^"]*)"', header_text)
         if not fn:
             return False, "Can't find out file name..."
         fn = sanitize_upload_filename(fn[0])
         if not fn:
             return False, "Unsafe upload file name"
         path = translate_path(self.path)
+        if not os.path.isdir(path):
+            return False, "Upload target is not a directory"
         target_path = os.path.join(path, fn)
         while os.path.exists(target_path):
             target_path += "_"
-        line = self.rfile.readline()
-        remain_bytes -= len(line)
-        line = self.rfile.readline()
-        remain_bytes -= len(line)
         try:
             out = open(target_path, 'wb')
         except IOError:
             return False, "Can't create file to write, do you have permission to write?"
 
+        success = False
         try:
-            pre_line = self.rfile.readline()
-            remain_bytes -= len(pre_line)
+            pre_line = None
             while remain_bytes > 0:
-                line = self.rfile.readline()
-                remain_bytes -= len(line)
-                if boundary in line:
-                    pre_line = pre_line[0:-1]
-                    if pre_line.endswith(b'\r'):
+                line, line_length = read_upload_line()
+                remain_bytes -= line_length
+                if not line:
+                    return False, "Unexpected end of data."
+                if is_boundary_line(line):
+                    if pre_line is not None:
                         pre_line = pre_line[0:-1]
-                    out.write(pre_line)
-                    out.close()
+                        if pre_line.endswith(b'\r'):
+                            pre_line = pre_line[0:-1]
+                        out.write(pre_line)
+                    success = True
                     return True, "File '%s' upload success!" % os.path.basename(target_path)
-                else:
+                if pre_line is not None:
                     out.write(pre_line)
-                    pre_line = line
+                pre_line = line
+            return False, "Unexpected end of data."
         finally:
-            if not out.closed:
-                out.close()
-        return False, "Unexpected end of data."
+            out.close()
+            if not success:
+                try:
+                    os.remove(target_path)
+                except OSError:
+                    pass
 
     def send_head(self):
         """Common code for GET and HEAD commands.

tests/test_simple_http_server.py

@@ -1,4 +1,5 @@
 import os
+from io import BytesIO
 import shutil
 import sys
 import tempfile
@@ -54,6 +55,94 @@ def test_translate_path_stays_under_current_directory(self):
         )
 
 
+class DummyUploadHandler(object):
+    max_upload_size = simple_http_server.MAX_UPLOAD_SIZE
+
+    def __init__(
+        self,
+        body,
+        content_length=None,
+        content_type='multipart/form-data; boundary=test-boundary',
+        path='/',
+    ):
+        self.headers = {
+            'Content-Type': content_type,
+            'content-length': str(len(body) if content_length is None else content_length),
+        }
+        self.rfile = BytesIO(body)
+        self.path = path
+
+    def deal_post_data(self):
+        return simple_http_server.SimpleHTTPRequestHandler.deal_post_data(self)
+
+
+class UploadParsingTests(unittest.TestCase):
+    boundary = b'test-boundary'
+
+    def make_body(self, filename, content, include_content_type=False, close=True):
+        headers = [
+            b'--' + self.boundary,
+            b'Content-Disposition: form-data; name="file"; filename="' + filename + b'"',
+        ]
+        if include_content_type:
+            headers.append(b'Content-Type: application/octet-stream')
+        headers.append(b'')
+        terminator = b'--' + self.boundary + (b'--' if close else b'')
+        return b'\r\n'.join(headers) + b'\r\n' + content + b'\r\n' + terminator + b'\r\n'
+
+    def run_upload(self, body, content_length=None):
+        old_cwd = os.getcwd()
+        temp_dir = tempfile.mkdtemp()
+        try:
+            os.chdir(temp_dir)
+            handler = DummyUploadHandler(body, content_length=content_length)
+            result = handler.deal_post_data()
+            files = {}
+            for name in os.listdir(temp_dir):
+                with open(os.path.join(temp_dir, name), 'rb') as uploaded:
+                    files[name] = uploaded.read()
+            return result, files
+        finally:
+            os.chdir(old_cwd)
+            shutil.rmtree(temp_dir)
+
+    def test_upload_without_part_content_type_preserves_file_content(self):
+        body = self.make_body(b'example.txt', b'first line\r\nsecond line')
+
+        result, files = self.run_upload(body)
+
+        self.assertTrue(result[0], result[1])
+        self.assertEqual(files, {'example.txt': b'first line\r\nsecond line'})
+
+    def test_upload_content_may_contain_boundary_text(self):
+        body = self.make_body(
+            b'example.txt',
+            b'before\r\nnot a delimiter: --test-boundary\r\nafter',
+        )
+
+        result, files = self.run_upload(body)
+
+        self.assertTrue(result[0], result[1])
+        self.assertEqual(
+            files,
+            {'example.txt': b'before\r\nnot a delimiter: --test-boundary\r\nafter'},
+        )
+
+    def test_interrupted_upload_fails_and_removes_partial_file(self):
+        body = b'\r\n'.join([
+            b'--' + self.boundary,
+            b'Content-Disposition: form-data; name="file"; filename="partial.txt"',
+            b'',
+            b'partial content without closing boundary',
+        ])
+
+        result, files = self.run_upload(body, content_length=len(body) + 20)
+
+        self.assertFalse(result[0])
+        self.assertEqual(result[1], 'Unexpected end of data.')
+        self.assertEqual(files, {})
+
+
 class ArgumentParserTests(unittest.TestCase):
     def test_default_bind_is_localhost(self):
         old_argv = sys.argv