fix: ZIP — лимиты файлов/размера, ZIP-бомба, санитизация имён
This commit is contained in:
@@ -1,5 +1,23 @@
|
|||||||
"""Unzip service — extract files from ZIP archive (no DB, no parse)."""
|
"""Unzip service — extract files from ZIP archive (no DB, no parse)."""
|
||||||
import zipfile, io, base64
|
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):
|
def handle_unzip(rfile, content_length):
|
||||||
@@ -19,11 +37,31 @@ def handle_unzip(rfile, content_length):
|
|||||||
zip_data = raw[zip_start:]
|
zip_data = raw[zip_start:]
|
||||||
|
|
||||||
files = []
|
files = []
|
||||||
|
total_uncompressed = 0
|
||||||
with zipfile.ZipFile(io.BytesIO(zip_data)) as zf:
|
with zipfile.ZipFile(io.BytesIO(zip_data)) as zf:
|
||||||
for name in zf.namelist():
|
namelist = zf.namelist()
|
||||||
if name.endswith("/"):
|
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
|
continue
|
||||||
data = zf.read(name)
|
|
||||||
|
# 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 ""
|
ext = name.rsplit(".", 1)[-1].lower() if "." in name else ""
|
||||||
files.append({
|
files.append({
|
||||||
"filename": name,
|
"filename": name,
|
||||||
|
|||||||
Reference in New Issue
Block a user