v1.0.176: upload.py — path traversal + size limit + UUID + data loss

deploy/services/upload.py

@@ -1,12 +1,24 @@
 """Upload service — accept file, store to DB, parse."""
-import uuid, base64, json, io, cgi
+import uuid, base64, json, io, cgi, os
 from db import documents, contracts, supplements
 from .parse import parse_file
 
+MAX_FILE_SIZE = 100 * 1024 * 1024  # 100 MB
+
 
 def handle_upload(rfile, content_type, content_length):
     """Parse multipart upload, store in DB, return result dict."""
-    body = rfile.read(content_length)
+    # Validate content_length
+    try:
+        cl = int(content_length)
+    except (TypeError, ValueError):
+        return {"ok": False, "error": "invalid content-length"}
+    if cl <= 0:
+        return {"ok": False, "error": "empty request"}
+    if cl > MAX_FILE_SIZE:
+        return {"ok": False, "error": f"file too large (max {MAX_FILE_SIZE // 1024 // 1024}MB)"}
+
+    body = rfile.read(cl)
 
     environ = {
         "REQUEST_METHOD": "POST",
@@ -23,13 +35,21 @@ def handle_upload(rfile, content_type, content_length):
         item = fs["files"]
         if isinstance(item, list):
             item = item[0]
-        filename = item.filename
+        filename = os.path.basename(item.filename) if item.filename else None
+        if filename and (".." in filename or "/" in filename or "\\" in filename):
+            return {"ok": False, "error": "invalid filename"}
         file_data = item.file.read() if hasattr(item, "file") else item.value
         if isinstance(file_data, str):
             file_data = file_data.encode("utf-8")
 
     if "contract_id" in fs:
         contract_id = fs.getfirst("contract_id", "")
+        # Validate UUID
+        if contract_id:
+            try:
+                uuid.UUID(contract_id)
+            except (ValueError, AttributeError):
+                contract_id = ""
 
     batch_id = fs.getfirst("batch_id", None) if "batch_id" in fs else None
     # Validate UUID
@@ -46,7 +66,10 @@ def handle_upload(rfile, content_type, content_length):
     b64 = base64.b64encode(file_data).decode()
 
     if contract_id:
+        try:
             supplements.delete_by_document(contract_id, filename)
+        except Exception:
+            pass  # old record may not exist or FK issue — proceed with insert
 
     doc = documents.insert(filename, mime, b64, batch_id=batch_id)