mirror of
https://github.com/Abdess/retroarch_system.git
synced 2026-10-18 17:33:30 -05:00
321 lines
12 KiB
Python
321 lines
12 KiB
Python
"""Reading archives that may be hostile.
|
|
|
|
Every limit here bounds what a crafted archive can cost without
|
|
capping a real romset: the largest pack is already several gigabytes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import functools
|
|
import hashlib
|
|
import os
|
|
import re
|
|
import stat
|
|
import sys
|
|
import tempfile
|
|
import zipfile
|
|
|
|
from hashing import md5sum
|
|
|
|
|
|
def check_inside_zip(container: str, file_name: str, expected_md5: str) -> str:
|
|
"""Check a ROM inside a ZIP, as Batocera's checkInsideZip() does.
|
|
|
|
Returns "ok", "untested", "not_in_zip", or "error".
|
|
"""
|
|
try:
|
|
info = os.stat(container)
|
|
except OSError:
|
|
return "error"
|
|
found = _member_md5(
|
|
container, info.st_size, info.st_mtime_ns, file_name.casefold(),
|
|
expected_md5 != "",
|
|
)
|
|
if found in ("not_in_zip", "error"):
|
|
return found
|
|
if expected_md5 == "":
|
|
return "ok"
|
|
return "ok" if found == expected_md5 else "untested"
|
|
|
|
|
|
@functools.lru_cache(maxsize=8192)
|
|
def _member_md5(
|
|
container: str, size: int, mtime_ns: int, folded: str, hashed: bool
|
|
) -> str:
|
|
"""The md5 of the member named `folded` (case-folded), or why there is none.
|
|
|
|
One archive member is asked about by every md5 a platform accepts for it
|
|
and by every step that checks it: `verify --all` decompressed 355
|
|
members 4 788 times. Keyed by the archive's size and mtime, so a
|
|
rewritten archive is read again.
|
|
"""
|
|
try:
|
|
with zipfile.ZipFile(container) as archive:
|
|
for fname in archive.namelist():
|
|
if fname.casefold() != folded:
|
|
continue
|
|
if archive.getinfo(fname).file_size > 512 * 1024 * 1024:
|
|
return "error"
|
|
if not hashed:
|
|
return "present"
|
|
with archive.open(fname) as entry:
|
|
return md5sum(entry)
|
|
return "not_in_zip"
|
|
except (zipfile.BadZipFile, OSError, KeyError):
|
|
return "error"
|
|
|
|
|
|
@functools.lru_cache(maxsize=4096)
|
|
def _member_crcs(path: str, mtime: float) -> frozenset[str]:
|
|
try:
|
|
with zipfile.ZipFile(path) as archive:
|
|
return frozenset(
|
|
format(info.CRC, "08x") for info in archive.infolist() if not info.is_dir()
|
|
)
|
|
except (zipfile.BadZipFile, OSError):
|
|
return frozenset()
|
|
|
|
|
|
def zip_member_crcs(path: str) -> frozenset[str]:
|
|
"""CRC-32 of each member, read from the central directory."""
|
|
try:
|
|
mtime = os.path.getmtime(path)
|
|
except OSError:
|
|
return frozenset()
|
|
return _member_crcs(path, mtime)
|
|
|
|
|
|
_zip_contents_cache: tuple[tuple, dict] | None = None
|
|
|
|
# Member md5s per archive, keyed by the archive's sha1: content addressed, so
|
|
# a cached answer is never stale. Every pipeline step that resolves a
|
|
# zipped_file used to decompress the 54 845 members again, about 23 s each.
|
|
MEMBER_CACHE = os.path.join(
|
|
os.path.dirname(os.path.dirname(os.path.abspath(__file__))), ".cache", "zip_member_md5.json"
|
|
)
|
|
|
|
|
|
# No caller asks for a member above this bound, the index's default.
|
|
MAX_HASHED_MEMBER = 512 * 1024 * 1024
|
|
|
|
|
|
def _read_member_cache(path: str) -> dict[str, list[list]]:
|
|
import json # noqa: PLC0415
|
|
|
|
try:
|
|
with open(path, encoding="utf-8") as fh:
|
|
data = json.load(fh)
|
|
except FileNotFoundError:
|
|
return {}
|
|
except (OSError, ValueError) as exc:
|
|
print(f"warning: ignoring unreadable {path}: {exc}", file=sys.stderr)
|
|
return {}
|
|
return data if isinstance(data, dict) else {}
|
|
|
|
|
|
def _write_member_cache(path: str, data: dict[str, list[list]]) -> None:
|
|
import json # noqa: PLC0415
|
|
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
fd, scratch = tempfile.mkstemp(dir=os.path.dirname(path), suffix=".tmp")
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8") as fh:
|
|
json.dump(data, fh, separators=(",", ":"))
|
|
os.replace(scratch, path)
|
|
except OSError as exc:
|
|
print(f"warning: could not write {path}: {exc}", file=sys.stderr)
|
|
if os.path.exists(scratch):
|
|
os.unlink(scratch)
|
|
|
|
|
|
def _members_md5(path: str) -> list[list] | None:
|
|
"""[md5, size] of each member, or None for an archive that cannot be read."""
|
|
members: list[list] = []
|
|
try:
|
|
with zipfile.ZipFile(path, "r") as zf:
|
|
for info in zf.infolist():
|
|
if info.is_dir():
|
|
continue
|
|
if info.file_size > MAX_HASHED_MEMBER:
|
|
members.append([None, info.file_size])
|
|
continue
|
|
h = hashlib.md5()
|
|
with zf.open(info.filename) as inner:
|
|
for chunk in iter(lambda: inner.read(65536), b""):
|
|
h.update(chunk)
|
|
members.append([h.hexdigest(), info.file_size])
|
|
except (zipfile.BadZipFile, OSError) as exc:
|
|
print(f"warning: unreadable archive {path}: {exc}", file=sys.stderr)
|
|
return None
|
|
return members
|
|
|
|
|
|
def build_zip_contents_index(
|
|
db: dict,
|
|
max_entry_size: int = 512 * 1024 * 1024,
|
|
cache_path: str | None = MEMBER_CACHE,
|
|
) -> dict:
|
|
"""Build {inner_rom_md5: zip_file_sha1} for ROMs inside ZIP files.
|
|
|
|
Cached in-process, and on disk per archive sha1 when cache_path is set.
|
|
"""
|
|
global _zip_contents_cache
|
|
|
|
zip_entries: list[tuple[str, str]] = []
|
|
for sha1, entry in db.get("files", {}).items():
|
|
path = entry["path"]
|
|
if path.endswith(".zip") and os.path.exists(path):
|
|
zip_entries.append((path, sha1))
|
|
|
|
# The bound and the sha1 each path carries are part of the answer: a
|
|
# second call with a smaller bound got members the first one admitted,
|
|
# and a regenerated database got the sha1 of the old one.
|
|
fingerprint = (
|
|
max_entry_size,
|
|
frozenset((path, os.path.getmtime(path), sha1) for path, sha1 in zip_entries),
|
|
)
|
|
if _zip_contents_cache is not None and _zip_contents_cache[0] == fingerprint:
|
|
return _zip_contents_cache[1]
|
|
|
|
cached = _read_member_cache(cache_path) if cache_path else {}
|
|
added = False
|
|
index: dict[str, str] = {}
|
|
for path, sha1 in zip_entries:
|
|
members = cached.get(sha1)
|
|
if members is None:
|
|
members = _members_md5(path)
|
|
if members is None:
|
|
continue
|
|
cached[sha1] = members
|
|
added = True
|
|
for md5, size in members:
|
|
if md5 is not None and size <= max_entry_size:
|
|
index[md5] = sha1
|
|
if added and cache_path:
|
|
_write_member_cache(cache_path, cached)
|
|
|
|
_zip_contents_cache = (fingerprint, index)
|
|
return index
|
|
|
|
|
|
MAX_ZIP_MEMBERS = 100_000
|
|
|
|
MAX_ZIP_MEMBER_SIZE = 8 * 1024 * 1024 * 1024
|
|
|
|
# The largest generated pack is already ~5 GB uncompressed and the collection
|
|
# only grows; this bounds a malicious archive without capping a real one.
|
|
MAX_ZIP_TOTAL_SIZE = 64 * 1024 * 1024 * 1024
|
|
|
|
# DEFLATE cannot exceed roughly 1,032:1, so this rejects a declared ratio no
|
|
# real DEFLATE member can reach. Methods with a higher ceiling (bzip2, LZMA)
|
|
# are exempt and bounded by the per-member and per-archive size limits alone.
|
|
MAX_ZIP_COMPRESSION_RATIO = 1_100
|
|
_BOUNDED_RATIO_METHODS = (zipfile.ZIP_STORED, zipfile.ZIP_DEFLATED)
|
|
|
|
|
|
def safe_extract_zip(
|
|
zip_path: str,
|
|
dest_dir: str,
|
|
*,
|
|
max_members: int = MAX_ZIP_MEMBERS,
|
|
max_member_size: int = MAX_ZIP_MEMBER_SIZE,
|
|
max_total_size: int = MAX_ZIP_TOTAL_SIZE,
|
|
max_compression_ratio: int = MAX_ZIP_COMPRESSION_RATIO,
|
|
) -> None:
|
|
"""Extract a ZIP with traversal, link and resource-limit protection.
|
|
|
|
Files are streamed to a temporary sibling and atomically installed only
|
|
after their declared length and CRC have been checked by ``zipfile``.
|
|
"""
|
|
dest = os.path.realpath(dest_dir)
|
|
os.makedirs(dest, exist_ok=True)
|
|
# Reading the umask sets it; it is put back at once.
|
|
umask = os.umask(0)
|
|
os.umask(umask)
|
|
with zipfile.ZipFile(zip_path, "r") as zf:
|
|
members = zf.infolist()
|
|
if len(members) > max_members:
|
|
raise ValueError(
|
|
f"ZIP has {len(members)} members; limit is {max_members}"
|
|
)
|
|
|
|
declared_total = 0
|
|
seen: set[str] = set()
|
|
for member in members:
|
|
# Archives written on Windows store a backslash separator. It is a
|
|
# separator, not a filename character, so it is normalized before
|
|
# the component checks rather than rejected.
|
|
name = member.filename.replace("\\", "/")
|
|
if not name or "\x00" in name:
|
|
raise ValueError(f"Unsafe ZIP member name: {member.filename!r}")
|
|
if name.startswith("/") or re.match(r"^[A-Za-z]:", name):
|
|
raise ValueError(f"Absolute ZIP member path: {name}")
|
|
parts = [part for part in name.split("/") if part]
|
|
if any(part in (".", "..") for part in parts):
|
|
raise ValueError(f"ZIP traversal detected: {name}")
|
|
normalized = "/".join(parts)
|
|
if normalized in seen:
|
|
raise ValueError(f"Duplicate ZIP member path: {name}")
|
|
seen.add(normalized)
|
|
|
|
mode = (member.external_attr >> 16) & 0xFFFF
|
|
file_type = stat.S_IFMT(mode)
|
|
if file_type not in (0, stat.S_IFREG, stat.S_IFDIR):
|
|
raise ValueError(f"ZIP link or special file rejected: {name}")
|
|
if member.flag_bits & 0x1:
|
|
raise ValueError(f"Encrypted ZIP member rejected: {name}")
|
|
if member.file_size > max_member_size:
|
|
raise ValueError(
|
|
f"ZIP member {name} is {member.file_size} bytes; "
|
|
f"limit is {max_member_size}"
|
|
)
|
|
declared_total += member.file_size
|
|
if declared_total > max_total_size:
|
|
raise ValueError(
|
|
f"ZIP expands to {declared_total} bytes; limit is {max_total_size}"
|
|
)
|
|
if member.file_size and member.compress_type in _BOUNDED_RATIO_METHODS:
|
|
if member.compress_size == 0:
|
|
raise ValueError(f"Invalid compression size for ZIP member: {name}")
|
|
if member.file_size / member.compress_size > max_compression_ratio:
|
|
raise ValueError(f"Suspicious compression ratio for ZIP member: {name}")
|
|
|
|
target = os.path.realpath(os.path.join(dest, *parts))
|
|
if not target.startswith(dest + os.sep) and target != dest:
|
|
raise ValueError(f"ZIP traversal detected: {name}")
|
|
if member.is_dir() or name.endswith("/"):
|
|
os.makedirs(target, exist_ok=True)
|
|
continue
|
|
|
|
os.makedirs(os.path.dirname(target), exist_ok=True)
|
|
tmp_path = ""
|
|
try:
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="wb", dir=os.path.dirname(target), delete=False
|
|
) as tmp_file:
|
|
tmp_path = tmp_file.name
|
|
actual_size = 0
|
|
with zf.open(member, "r") as source:
|
|
while True:
|
|
chunk = source.read(1024 * 1024)
|
|
if not chunk:
|
|
break
|
|
actual_size += len(chunk)
|
|
if actual_size > member.file_size or actual_size > max_member_size:
|
|
raise ValueError(
|
|
f"ZIP member exceeded declared or configured size: {name}"
|
|
)
|
|
tmp_file.write(chunk)
|
|
if actual_size != member.file_size:
|
|
raise ValueError(
|
|
f"ZIP member size mismatch for {name}: "
|
|
f"{actual_size} != {member.file_size}"
|
|
)
|
|
# The member's own mode, else what a plain open() gives; the
|
|
# temporary file is private and would keep 0600 otherwise.
|
|
os.chmod(tmp_path, ((mode & 0o777) or 0o666) & ~umask)
|
|
os.replace(tmp_path, target)
|
|
tmp_path = ""
|
|
finally:
|
|
if tmp_path and os.path.exists(tmp_path):
|
|
os.unlink(tmp_path)
|