74 lines
2.5 KiB
Python
74 lines
2.5 KiB
Python
"""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}
|