optional max-size for download-as-zip/tar

copyparty/__main__.py

@@ -1236,6 +1236,10 @@ def add_optouts(ap):
     ap2.add_argument("-nih", action="store_true", help="no info hostname -- don't show in UI")
     ap2.add_argument("-nid", action="store_true", help="no info disk-usage -- don't show in UI")
     ap2.add_argument("-nb", action="store_true", help="no powered-by-copyparty branding in UI")
+    ap2.add_argument("--zipmaxn", metavar="N", type=u, default="0", help="reject download-as-zip if more than \033[33mN\033[0m files in total; optionally takes a unit suffix: [\033[32m256\033[0m], [\033[32m9K\033[0m], [\033[32m4G\033[0m] (volflag=zipmaxn)")
+    ap2.add_argument("--zipmaxs", metavar="MiB", type=u, default="0", help="reject download-as-zip if total download size exceeds \033[33mMiB\033[0m; assumes megabytes unless a unit suffix is given: [\033[32m256\033[0m], [\033[32m4G\033[0m], [\033[32m2T\033[0m] (volflag=zipmaxs)")
+    ap2.add_argument("--zipmaxt", metavar="TXT", type=u, default="", help="custom errormessage when download size exceeds max (volflag=zipmaxt)")
+    ap2.add_argument("--zipmaxu", action="store_true", help="authenticated users bypass the zip size limit (volflag=zipmaxu)")
     ap2.add_argument("--zip-who", metavar="LVL", type=int, default=3, help="who can download as zip/tar? [\033[32m0\033[0m]=nobody, [\033[32m1\033[0m]=admins, [\033[32m2\033[0m]=authenticated-with-read-access, [\033[32m3\033[0m]=everyone-with-read-access (volflag=zip_who)\n\033[1;31mWARNING:\033[0m if a nested volume has a more restrictive value than a parent volume, then this will be \033[33mignored\033[0m if the download is initiated from the parent, more lenient volume")
     ap2.add_argument("--no-zip", action="store_true", help="disable download as zip/tar; same as \033[33m--zip-who=0\033[0m")
     ap2.add_argument("--no-tarcmp", action="store_true", help="disable download as compressed tar (?tar=gz, ?tar=bz2, ?tar=xz, ?tar=gz:9, ...)")

copyparty/authsrv.py

@@ -47,7 +47,7 @@ from .util import (
 if True:  # pylint: disable=using-constant-test
     from collections.abc import Iterable
 
-    from typing import Any, Generator, Optional, Union
+    from typing import Any, Generator, Optional, Sequence, Union
 
     from .util import NamedLogger, RootLogger
 
@@ -1802,6 +1802,29 @@ class AuthSrv(object):
             rhisttab[histp] = zv
             vfs.histtab[zv.realpath] = histp
 
+        for vol in vfs.all_vols.values():
+            use = False
+            for k, si in [["zipmaxn", ""], ["zipmaxs", "m"]]:
+                try:
+                    zs = vol.flags[k]
+                except:
+                    zs = getattr(self.args, k)
+                if zs in ("", "0"):
+                    vol.flags[k] = 0
+                    continue
+
+                try:
+                    _ = float(zs)
+                    zs = "%s%s" % (zs, si)
+                except:
+                    pass
+                zf = unhumanize(zs)
+                vol.flags[k] = zf
+                if zf:
+                    use = True
+            if use:
+                vol.flags["zipmax"] = True
+
         for vol in vfs.all_vols.values():
             lim = Lim(self.log_func)
             use = False
@@ -2730,7 +2753,7 @@ class AuthSrv(object):
     def dbg_ls(self) -> None:
         users = self.args.ls
         vol = "*"
-        flags: list[str] = []
+        flags: Sequence[str] = []
 
         try:
             users, vol = users.split(",", 1)

copyparty/cfg.py

@@ -55,6 +55,7 @@ def vf_bmap() -> dict[str, str]:
         "xdev",
         "xlink",
         "xvol",
+        "zipmaxu",
     ):
         ret[k] = k
     return ret
@@ -101,6 +102,7 @@ def vf_vmap() -> dict[str, str]:
         "u2ts",
         "ups_who",
         "zip_who",
+        "zipmaxt",
     ):
         ret[k] = k
     return ret
@@ -299,6 +301,10 @@ flagcats = {
         "rss": "allow '?rss' URL suffix (experimental)",
         "ups_who=2": "restrict viewing the list of recent uploads",
         "zip_who=2": "restrict access to download-as-zip/tar",
+        "zipmaxn=9k": "reject download-as-zip if more than 9000 files",
+        "zipmaxs=2g": "reject download-as-zip if size over 2 GiB",
+        "zipmaxt=no": "reply with 'no' if download-as-zip exceeds max",
+        "zipmaxu": "zip-size-limit does not apply to authenticated users",
         "nopipe": "disable race-the-beam (download unfinished uploads)",
         "mv_retry": "ms-windows: timeout for renaming busy files",
         "rm_retry": "ms-windows: timeout for deleting busy files",

copyparty/httpcli.py

@@ -20,9 +20,9 @@ import time
 import uuid
 from datetime import datetime
 from operator import itemgetter
-from ipaddress import IPv6Network
 
 import jinja2  # typechk
+from ipaddress import IPv6Network
 
 try:
     if os.environ.get("PRTY_NO_LZMA"):
@@ -87,10 +87,10 @@ from .util import (
     quotep,
     rand_name,
     read_header,
-    read_utf8,
     read_socket,
     read_socket_chunked,
     read_socket_unbounded,
+    read_utf8,
     relchk,
     ren_open,
     runhook,
@@ -4366,6 +4366,33 @@ class HttpCli(object):
         else:
             fn = self.host.split(":")[0]
 
+        if vn.flags.get("zipmax") and (not self.uname or not "zipmaxu" in vn.flags):
+            maxs = vn.flags.get("zipmaxs") or 0
+            maxn = vn.flags.get("zipmaxn") or 0
+            nf = 0
+            nb = 0
+            fgen = vn.zipgen(
+                vpath, rem, set(items), self.uname, False, not self.args.no_scandir
+            )
+            t = "total size exceeds a limit specified in server config"
+            t = vn.flags.get("zipmaxt") or t
+            if maxs and maxn:
+                for zd in fgen:
+                    nf += 1
+                    nb += zd["st"].st_size
+                    if maxs < nb or maxn < nf:
+                        raise Pebkac(400, t)
+            elif maxs:
+                for zd in fgen:
+                    nb += zd["st"].st_size
+                    if maxs < nb:
+                        raise Pebkac(400, t)
+            elif maxn:
+                for zd in fgen:
+                    nf += 1
+                    if maxn < nf:
+                        raise Pebkac(400, t)
+
         safe = (string.ascii_letters + string.digits).replace("%", "")
         afn = "".join([x if x in safe.replace('"', "") else "_" for x in fn])
         bascii = unicode(safe).encode("utf-8")

copyparty/up2k.py

@@ -1119,7 +1119,7 @@ class Up2k(object):
         ft = "\033[0;32m{}{:.0}"
         ff = "\033[0;35m{}{:.0}"
         fv = "\033[0;36m{}:\033[90m{}"
-        zs = "ext_th_d html_head mv_re_r mv_re_t rm_re_r rm_re_t srch_re_dots srch_re_nodot"
+        zs = "ext_th_d html_head mv_re_r mv_re_t rm_re_r rm_re_t srch_re_dots srch_re_nodot zipmax"
         fx = set(zs.split())
         fd = vf_bmap()
         fd.update(vf_cmap())

tests/util.py

@@ -129,7 +129,7 @@ class Cfg(Namespace):
     def __init__(self, a=None, v=None, c=None, **ka0):
         ka = {}
 
-        ex = "chpw daw dav_auth dav_mac dav_rt e2d e2ds e2dsa e2t e2ts e2tsr e2v e2vu e2vp early_ban ed emp exp force_js getmod grid gsel hardlink ih ihead magic hardlink_only nid nih no_acode no_athumb no_bauth no_clone no_cp no_dav no_db_ip no_del no_dirsz no_dupe no_lifetime no_logues no_mv no_pipe no_poll no_readme no_robots no_sb_md no_sb_lg no_scandir no_tarcmp no_thumb no_vthumb no_zip nrand nsort nw og og_no_head og_s_title ohead q rand re_dirsz rss smb srch_dbg srch_excl stats uqe vague_403 vc ver write_uplog xdev xlink xvol zs"
+        ex = "chpw daw dav_auth dav_mac dav_rt e2d e2ds e2dsa e2t e2ts e2tsr e2v e2vu e2vp early_ban ed emp exp force_js getmod grid gsel hardlink ih ihead magic hardlink_only nid nih no_acode no_athumb no_bauth no_clone no_cp no_dav no_db_ip no_del no_dirsz no_dupe no_lifetime no_logues no_mv no_pipe no_poll no_readme no_robots no_sb_md no_sb_lg no_scandir no_tarcmp no_thumb no_vthumb no_zip nrand nsort nw og og_no_head og_s_title ohead q rand re_dirsz rss smb srch_dbg srch_excl stats uqe vague_403 vc ver write_uplog xdev xlink xvol zipmaxu zs"
         ka.update(**{k: False for k in ex.split()})
 
         ex = "dav_inf dedup dotpart dotsrch hook_v no_dhash no_fastboot no_fpool no_htp no_rescan no_sendfile no_ses no_snap no_up_list no_voldump re_dhash plain_ip"
@@ -144,10 +144,10 @@ class Cfg(Namespace):
         ex = "au_vol dl_list mtab_age reg_cap s_thead s_tbody th_convt ups_who zip_who"
         ka.update(**{k: 9 for k in ex.split()})
 
-        ex = "db_act forget_ip k304 loris no304 re_maxage rproxy rsp_jtr rsp_slp s_wr_slp snap_wri theme themes turbo u2ow"
+        ex = "db_act forget_ip k304 loris no304 re_maxage rproxy rsp_jtr rsp_slp s_wr_slp snap_wri theme themes turbo u2ow zipmaxn zipmaxs"
         ka.update(**{k: 0 for k in ex.split()})
 
-        ex = "ah_alg bname chpw_db doctitle df exit favico idp_h_usr ipa html_head lg_sba lg_sbf log_fk md_sba md_sbf name og_desc og_site og_th og_title og_title_a og_title_v og_title_i shr tcolor textfiles unlist vname xff_src R RS SR"
+        ex = "ah_alg bname chpw_db doctitle df exit favico idp_h_usr ipa html_head lg_sba lg_sbf log_fk md_sba md_sbf name og_desc og_site og_th og_title og_title_a og_title_v og_title_i shr tcolor textfiles unlist vname xff_src zipmaxt R RS SR"
         ka.update(**{k: "" for k in ex.split()})
 
         ex = "ban_403 ban_404 ban_422 ban_pw ban_url spinner"