refactor: services/ → compare/ + drhider/ (разделение Сверки и Хайдера)

deploy/compare/unzip.py

@@ -0,0 +1,73 @@
+"""Unzip service — extract files from ZIP archive (no DB, no parse)."""
+import zipfile, io, base64, os
+
+MAX_FILES = 500
+MAX_UNCOMPRESSED = 500 * 1024 * 1024  # 500 MB total
+MAX_RATIO = 100  # max compression ratio (uncompressed/compressed)
+
+
+def _sanitize_zip_name(name: str) -> str | None:
+    """Decode cp437→utf8, strip path traversal, return safe name or None."""
+    # Decode cp437 filenames (common in ZIPs from Windows)
+    try:
+        name = name.encode("cp437").decode("utf-8", errors="replace")
+    except (UnicodeDecodeError, UnicodeEncodeError):
+        pass
+    # Strip directory prefixes and path traversal
+    name = os.path.basename(name)
+    if not name or name.endswith("/") or ".." in name or "/" in name or "\\" in name:
+        return None
+    return name
+
+
+def handle_unzip(rfile, content_length):
+    """Parse ZIP upload, return list of extracted files as base64."""
+    body = rfile.read(content_length)
+
+    # Find file data in raw multipart
+    idx = body.find(b"\r\n\r\n")
+    if idx < 0:
+        return {"ok": False, "error": "invalid multipart"}
+
+    raw = body[idx + 4:]
+    zip_start = raw.find(b"PK\x03\x04")
+    if zip_start < 0:
+        return {"ok": False, "error": "not a ZIP file"}
+
+    zip_data = raw[zip_start:]
+
+    files = []
+    total_uncompressed = 0
+    with zipfile.ZipFile(io.BytesIO(zip_data)) as zf:
+        namelist = zf.namelist()
+        if len(namelist) > MAX_FILES:
+            return {"ok": False, "error": f"too many files in ZIP (max {MAX_FILES})"}
+
+        for info in zf.infolist():
+            if info.is_dir():
+                continue
+
+            # ZIP bomb: compression ratio check
+            if info.compress_size > 0:
+                ratio = info.file_size / info.compress_size
+                if ratio > MAX_RATIO:
+                    return {"ok": False, "error": f"suspicious compression ratio ({ratio:.0f}:1) — possible ZIP bomb"}
+
+            name = _sanitize_zip_name(info.filename)
+            if not name:
+                continue  # skip unsafe names
+
+            data = zf.read(info)
+            total_uncompressed += len(data)
+            if total_uncompressed > MAX_UNCOMPRESSED:
+                return {"ok": False, "error": f"uncompressed size exceeds {MAX_UNCOMPRESSED // 1024 // 1024}MB"}
+
+            ext = name.rsplit(".", 1)[-1].lower() if "." in name else ""
+            files.append({
+                "filename": name,
+                "data_b64": base64.b64encode(data).decode(),
+                "ext": ext,
+                "size": len(data),
+            })
+
+    return {"ok": True, "files": files}