diff --git a/copyparty/httpcli.py b/copyparty/httpcli.py index abd7aab77..0e14bd970 100644 --- a/copyparty/httpcli.py +++ b/copyparty/httpcli.py @@ -130,6 +130,7 @@ from .util import ( vsplit, wunlink, yieldfile, + zip_fi, ) if True: # pylint: disable=using-constant-test @@ -1921,9 +1922,7 @@ class HttpCli(object): try: with zipfile.ZipFile(abspath, "r") as zf: - zi = zf.getinfo(inner_path) - if zi.file_size >= maxsz: - raise Pebkac(404, "zip bomb defused") + zi = zip_fi(zf, inner_path, maxsz) with zf.open(zi, "r") as fi: mime = guess_mime(inner_path) if mime not in SAFE_MIMES and "nohtml" in self.vn.flags: diff --git a/copyparty/mtag.py b/copyparty/mtag.py index 3bfd40432..af1377b30 100644 --- a/copyparty/mtag.py +++ b/copyparty/mtag.py @@ -30,6 +30,9 @@ from .util import ( sfsenc, uncyg, wunlink, + zip_fi, + zip_lim, + zip_readf, ) if True: # pylint: disable=using-constant-test @@ -174,7 +177,7 @@ def au_unpk( zil = [x for x in zil if x.filename.lower().split(".")[-1] == au] if not zil: raise Exception("no audio inside zip") - fi = zf.open(zil[0]) + fi = zf.open(zip_lim(zil[0], maxsz)) elif pk == "cbz": import zipfile @@ -188,17 +191,18 @@ def au_unpk( t = "cbz: %d files, %d hits" % (nf, len(znil)) if not znil: raise Exception("no images inside cbz") - using = sorted(znil)[0][1].filename + zi = sorted(znil)[0][1] if znil: - t += ", using " + using + t += ", using " + zi.filename log(t) - fi = zf.open(using) + fi = zf.open(zip_lim(zi, maxsz)) elif pk == "kra" or pk == "ora": import zipfile zf = zipfile.ZipFile(abspath, "r") - fi = zf.open("mergedimage.png") + zi = zip_fi(zf, "mergedimage.png", maxsz) + fi = zf.open(zi) elif pk == "epub": fi = get_cover_from_epub(log, abspath) @@ -438,7 +442,8 @@ def get_cover_from_epub(log: "NamedLogger", abspath: str) -> Optional[IO[bytes]] with zipfile.ZipFile(abspath, "r") as z: # First open the container file to find the package document (.opf file) try: - container_root = parse_xml(z.read("META-INF/container.xml").decode()) + zb = zip_readf(z, "META-INF/container.xml", 2 << 20) + container_root = parse_xml(zb.decode()) except KeyError: log("epub: no container file found in %s" % (abspath,)) return None @@ -452,7 +457,8 @@ def get_cover_from_epub(log: "NamedLogger", abspath: str) -> Optional[IO[bytes]] # Then open the first package document to find the path of the cover image try: - package_root = parse_xml(z.read(rootfile_path).decode()) + zb = zip_readf(z, rootfile_path, 2 << 20) + package_root = parse_xml(zb.decode()) except KeyError: log("epub: no package document found in %s" % (abspath,)) return None @@ -476,7 +482,8 @@ def get_cover_from_epub(log: "NamedLogger", abspath: str) -> Optional[IO[bytes]] adjusted_cover_path = urljoin(rootfile_path, coverimage_path) try: - return z.open(adjusted_cover_path) + zi = zip_fi(z, adjusted_cover_path, 2 << 24) + return z.open(zi) except KeyError: t = "epub: cover specified in package document, but doesn't exist: %s" log(t % (adjusted_cover_path,)) diff --git a/copyparty/util.py b/copyparty/util.py index 1d7329978..a38fe1705 100644 --- a/copyparty/util.py +++ b/copyparty/util.py @@ -4399,6 +4399,36 @@ def gzip_file_orig_sz(f) -> int: return sunpack(b"I", rv)[0] # type: ignore +def zip_fi(zf, fp, max_sz): + zi = zf.getinfo(fp) + if max_sz and zi.file_size > max_sz: + raise Pebkac(404, "zip bomb defused") + return zi + + +def zip_lim(zi, max_sz): + if zi.file_size > max_sz: + raise Pebkac(404, "zip bomb defused") + return zi + + +def zip_read(zf, zi, max_sz): + ret = b"" + with zf.open(zi) as f: + while True: + buf = f.read(max_sz) + if not buf: + break + ret += buf + if len(ret) >= max_sz: + raise Pebkac(404, "zip bomb defused") + return ret + + +def zip_readf(zf, zp, max_sz): + return zip_read(zf, zip_fi(zf, zp, max_sz), max_sz) + + def align_tab(lines: list[str]) -> list[str]: rows = [] ncols = 0