19a5985f removed the restriction on uploading logues, as it was

too restrictive, blocking editing through webdav and ftp

but since logues and readmes can be used as helptext for users
with write-only access, it makes sense to block logue/readme
uploads from write-only users

users with write-only access can still upload any file as before,
but the filename prefix `_wo_` is added onto files named either
README.md | PREADME.md | .prologue.html | .epilogue.html

the new option `--wo-up-readme` restores previous behavior, and
will not add the filename-prefix for readmes/logues

copyparty/__main__.py

@@ -1222,6 +1222,7 @@ def add_yolo(ap):
     ap2 = ap.add_argument_group('yolo options')
     ap2.add_argument("--allow-csrf", action="store_true", help="disable csrf protections; let other domains/sites impersonate you through cross-site requests")
     ap2.add_argument("--getmod", action="store_true", help="permit ?move=[...] and ?delete as GET")
+    ap2.add_argument("--wo-up-readme", action="store_true", help="allow users with write-only access to upload logues and readmes without adding the _wo_ filename prefix (volflag=wo_up_readme)")
 
 
 def add_optouts(ap):

copyparty/cfg.py

@@ -52,6 +52,7 @@ def vf_bmap() -> dict[str, str]:
         "og_s_title",
         "rand",
         "rss",
+        "wo_up_readme",
         "xdev",
         "xlink",
         "xvol",
@@ -173,6 +174,7 @@ flagcats = {
         "vmaxb=1g": "total volume size max 1 GiB (suffixes: b, k, m, g, t)",
         "vmaxn=4k": "max 4096 files in volume (suffixes: b, k, m, g, t)",
         "medialinks": "return medialinks for non-up2k uploads (not hotlinks)",
+        "wo_up_readme": "write-only users can upload logues without getting renamed",
         "rand": "force randomized filenames, 9 chars long by default",
         "nrand=N": "randomized filenames are N chars long",
         "u2ow=N": "overwrite existing files? 0=no 1=if-older 2=always",

copyparty/ftpd.py

@@ -19,6 +19,7 @@ from .__init__ import PY2, TYPE_CHECKING
 from .authsrv import VFS
 from .bos import bos
 from .util import (
+    FN_EMB,
     VF_CAREFUL,
     Daemon,
     ODict,
@@ -170,6 +171,16 @@ class FtpFs(AbstractedFS):
             fn = sanitize_fn(fn or "", "")
             vpath = vjoin(rd, fn)
             vfs, rem = self.hub.asrv.vfs.get(vpath, self.uname, r, w, m, d)
+            if (
+                w
+                and fn.lower() in FN_EMB
+                and self.h.uname not in vfs.axs.uread
+                and "wo_up_readme" not in vfs.flags
+            ):
+                fn = "_wo_" + fn
+                vpath = vjoin(rd, fn)
+                vfs, rem = self.hub.asrv.vfs.get(vpath, self.uname, r, w, m, d)
+
             if not vfs.realpath:
                 t = "No filesystem mounted at [{}]"
                 raise FSE(t.format(vpath))

copyparty/httpcli.py

@@ -46,6 +46,7 @@ from .util import (
     APPLESAN_RE,
     BITNESS,
     DAV_ALLPROPS,
+    FN_EMB,
     HAVE_SQLITE3,
     HTTPCODE,
     META_NOBOTS,
@@ -2550,6 +2551,16 @@ class HttpCli(object):
         vfs, rem = self.asrv.vfs.get(self.vpath, self.uname, False, True)
         dbv, vrem = vfs.get_dbv(rem)
 
+        name = sanitize_fn(name, "")
+        if (
+            not self.can_read
+            and self.can_write
+            and name.lower() in FN_EMB
+            and "wo_up_readme" not in dbv.flags
+        ):
+            name = "_wo_" + name
+
+        body["name"] = name
         body["vtop"] = dbv.vpath
         body["ptop"] = dbv.realpath
         body["prel"] = vrem

copyparty/tftpd.py

@@ -36,7 +36,19 @@ from partftpy.TftpShared import TftpException
 from .__init__ import EXE, PY2, TYPE_CHECKING
 from .authsrv import VFS
 from .bos import bos
-from .util import UTC, BytesIO, Daemon, ODict, exclude_dotfiles, min_ex, runhook, undot
+from .util import (
+    FN_EMB,
+    UTC,
+    BytesIO,
+    Daemon,
+    ODict,
+    exclude_dotfiles,
+    min_ex,
+    runhook,
+    undot,
+    vjoin,
+    vsplit,
+)
 
 if True:  # pylint: disable=using-constant-test
     from typing import Any, Union
@@ -244,16 +256,25 @@ class Tftpd(object):
         for srv in srvs:
             srv.stop()
 
-    def _v2a(self, caller: str, vpath: str, perms: list, *a: Any) -> tuple[VFS, str]:
+    def _v2a(
+        self, caller: str, vpath: str, perms: list, *a: Any
+    ) -> tuple[VFS, str, str]:
         vpath = vpath.replace("\\", "/").lstrip("/")
         if not perms:
             perms = [True, True]
 
         debug('%s("%s", %s) %s\033[K\033[0m', caller, vpath, str(a), perms)
         vfs, rem = self.asrv.vfs.get(vpath, "*", *perms)
+        if perms[1] and "*" not in vfs.axs.uread and "wo_up_readme" not in vfs.flags:
+            zs, fn = vsplit(vpath)
+            if fn.lower() in FN_EMB:
+                vpath = vjoin(zs, "_wo_" + fn)
+                vfs, rem = self.asrv.vfs.get(vpath, "*", *perms)
+
         if not vfs.realpath:
             raise Exception("unmapped vfs")
-        return vfs, vfs.canonical(rem)
+
+        return vfs, vpath, vfs.canonical(rem)
 
     def _ls(self, vpath: str, raddress: str, rport: int, force=False) -> Any:
         # generate file listing if vpath is dir.txt and return as file object
@@ -331,7 +352,7 @@ class Tftpd(object):
         else:
             raise Exception("bad mode %s" % (mode,))
 
-        vfs, ap = self._v2a("open", vpath, [rd, wr])
+        vfs, vpath, ap = self._v2a("open", vpath, [rd, wr])
         if wr:
             if "*" not in vfs.axs.uwrite:
                 yeet("blocked write; folder not world-writable: /%s" % (vpath,))
@@ -368,7 +389,7 @@ class Tftpd(object):
         return open(ap, mode, *a, **ka)
 
     def _mkdir(self, vpath: str, *a) -> None:
-        vfs, ap = self._v2a("mkdir", vpath, [])
+        vfs, _, ap = self._v2a("mkdir", vpath, [False, True])
         if "*" not in vfs.axs.uwrite:
             yeet("blocked mkdir; folder not world-writable: /%s" % (vpath,))
 
@@ -376,7 +397,7 @@ class Tftpd(object):
 
     def _unlink(self, vpath: str) -> None:
         # return bos.unlink(self._v2a("stat", vpath, *a)[1])
-        vfs, ap = self._v2a("delete", vpath, [True, False, False, True])
+        vfs, _, ap = self._v2a("delete", vpath, [True, False, False, True])
 
         try:
             inf = bos.stat(ap)
@@ -400,7 +421,7 @@ class Tftpd(object):
 
     def _p_exists(self, vpath: str) -> bool:
         try:
-            ap = self._v2a("p.exists", vpath, [False, False])[1]
+            ap = self._v2a("p.exists", vpath, [False, False])[2]
             bos.stat(ap)
             return True
         except:
@@ -408,7 +429,7 @@ class Tftpd(object):
 
     def _p_isdir(self, vpath: str) -> bool:
         try:
-            st = bos.stat(self._v2a("p.isdir", vpath, [False, False])[1])
+            st = bos.stat(self._v2a("p.isdir", vpath, [False, False])[2])
             ret = stat.S_ISDIR(st.st_mode)
             return ret
         except:

copyparty/up2k.py

@@ -2918,7 +2918,6 @@ class Up2k(object):
             if ptop not in self.registry:
                 raise Pebkac(410, "location unavailable")
 
-        cj["name"] = sanitize_fn(cj["name"], "")
         cj["poke"] = now = self.db_act = self.vol_act[ptop] = time.time()
         wark = dwark = self._get_wark(cj)
         job = None
@@ -3236,6 +3235,7 @@ class Up2k(object):
                                     job["ptop"] = vfs.realpath
                                     job["vtop"] = vfs.vpath
                                     job["prel"] = rem
+                                    job["name"] = sanitize_fn(job["name"], "")
                                     if zvfs.vpath != vfs.vpath:
                                         # print(json.dumps(job, sort_keys=True, indent=4))
                                         job["hash"] = cj["hash"]
@@ -4996,6 +4996,7 @@ class Up2k(object):
                     job["ptop"] = vfs.realpath
                     job["vtop"] = vfs.vpath
                     job["prel"] = rem
+                    job["name"] = sanitize_fn(job["name"], "")
                     if zvfs.vpath != vfs.vpath:
                         self.log("xbu reloc2:%d..." % (depth,), 6)
                         return self._handle_json(job, depth + 1)

copyparty/util.py

@@ -448,6 +448,8 @@ UNHUMANIZE_UNITS = {
 
 VF_CAREFUL = {"mv_re_t": 5, "rm_re_t": 5, "mv_re_r": 0.1, "rm_re_r": 0.1}
 
+FN_EMB = set([".prologue.html", ".epilogue.html", "readme.md", "preadme.md"])
+
 
 def read_ram() -> tuple[float, float]:
     a = b = 0

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 zipmaxu 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 wo_up_readme 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"