dont trust zip headers

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:

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,))

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