xvol/xdev (GHSA-hg99-jqfp-rf76)

This commit is contained in:
ed committed 2026-10-14 21:49:04 +00:00
1 parent 0dc1cb3291
commit 82eab3d2f8
10 files changed
+175 -74

No files matched your search

+7 -2
View File
@@ -1805,12 +1805,17 @@ avoid traversing into other filesystems using `--xdev` / volflag `:c,xdev`, ski
and/or you can `--xvol` / `:c,xvol` to ignore all symlinks leaving the volume's top directory, but still allow bind-mounts pointing elsewhere and/or you can `--xvol` / `:c,xvol` to ignore all symlinks leaving the volume's top directory, but still allow bind-mounts pointing elsewhere
* symlinks are permitted with `xvol` if they point into another volume where the user also has some sort of access, keeping permissions from outer/initial volume * symlinks are permitted with `xvol` if they point into another volume where the user also has some sort of access
* usually keeps permissions from outer/initial volume, but don't take that for granted
these options will reduce performance; unlikely worst-case estimates are 14% reduction for directory listings, 35% for download-as-tar these options will reduce performance; unlikely worst-case estimates are 27% reduction for directory listings, 35% for download-as-tar
* if the underlying storage is [a network disk](#using-the-cloud-as-storage) then xvol/xdev will be **very** slow (not to mention pointless/useless)
as of copyparty v1.7.0 these options also prevent file access at runtime -- in previous versions it was just hints for the indexer as of copyparty v1.7.0 these options also prevent file access at runtime -- in previous versions it was just hints for the indexer
> **WARNING:** the `xvol` and `xdev` options should NOT be relied on for security; these are originally and primarily intended for the [e2ds indexer](#file-indexing). Certain underlying filesystems (especially FUSE-based and casefolding ones) will be problematic, and symlinks inherently have various TOCTOUs, and there's probably also bugs in copyparty where xvol/xdev is not fully enforced for certain APIs. The developer does not care about these options and they do not receive much testing.
### periodic rescan ### periodic rescan
filesystem monitoring; if copyparty is not the only software doing stuff on your filesystem, you may want to enable periodic rescans to keep the index up to date filesystem monitoring; if copyparty is not the only software doing stuff on your filesystem, you may want to enable periodic rescans to keep the index up to date
+50 -3
View File
@@ -470,6 +470,7 @@ class VFS(object):
self.canonical = self._canonical_null self.canonical = self._canonical_null
self.dcanonical = self._dcanonical_null self.dcanonical = self._dcanonical_null
self.get = self._get
self.get_dbv = self._get_dbv self.get_dbv = self._get_dbv
self.ls = self._ls self.ls = self._ls
@@ -603,7 +604,7 @@ class VFS(object):
ret += "A" ret += "A"
return ret return ret
def get( def _get(
self, self,
vpath: str, vpath: str,
uname: str, uname: str,
@@ -645,6 +646,42 @@ class VFS(object):
return vn, rem return vn, rem
def _get_xvd(
self,
vpath: str,
uname: str,
will_read: bool,
will_write: bool,
will_move: bool = False,
will_del: bool = False,
will_get: bool = False,
err: int = 403,
) -> tuple["VFS", str]:
zvr = self._get(
vpath, uname, will_read, will_write, will_move, will_del, will_get, err
)
if not zvr:
return None
vn, rem = zvr
if "xvd" in vn.flags and vn.realpath:
avn = vn.chk_ap(vn.canonical(rem))
if avn and avn.vpath == vn.vpath:
return vn, rem
if (
not avn
or (will_read and uname not in avn.axs.uread)
or (will_write and uname not in avn.axs.uwrite)
or (will_move and uname not in avn.axs.umove)
or (will_del and uname not in avn.axs.udel)
or (will_get and uname not in avn.axs.uget)
):
if not err:
return None
raise Pebkac(
err, "permission denied for [/%s]" % (vjoin(vn.vpath, rem))
)
return vn, rem
def _get_share_src(self, vrem: str) -> tuple["VFS", str]: def _get_share_src(self, vrem: str) -> tuple["VFS", str]:
src = self.shr_src src = self.shr_src
if not src: if not src:
@@ -870,12 +907,16 @@ class VFS(object):
self.log("vfs.walk", t % (seen[-1], fsroot, self.vpath, rem), 3) self.log("vfs.walk", t % (seen[-1], fsroot, self.vpath, rem), 3)
return return
if "xdev" in self.flags or "xvol" in self.flags: if "xvd" in self.flags:
ps = (permsets[0] + [False] * 5)[:5] # rwmdg
rm1 = [] rm1 = []
for le in vfs_ls: for le in vfs_ls:
ap = absreal(os.path.join(fsroot, le[0])) ap = absreal(os.path.join(fsroot, le[0]))
vn2 = self.chk_ap(ap) vn2 = self.chk_ap(ap)
if not vn2 or not vn2.get("", uname, True, False, err=0): try:
if not vn2.get("", uname, *ps):
raise Exception()
except:
rm1.append(le) rm1.append(le)
_ = [vfs_ls.remove(x) for x in rm1] # type: ignore _ = [vfs_ls.remove(x) for x in rm1] # type: ignore
@@ -2979,6 +3020,12 @@ class AuthSrv(object):
excl.extend([r"^\.", r"/\."]) excl.extend([r"^\.", r"/\."])
vol.flags["srch_re_nodot"] = re.compile("|".join(excl)) vol.flags["srch_re_nodot"] = re.compile("|".join(excl))
for vol in list(vfs.all_nodes.values()):
if "xvd" in vol.flags:
for vol in vfs.all_nodes.values():
vol.get = vol._get_xvd
break
have_daw = False have_daw = False
for vol in vfs.all_nodes.values(): for vol in vfs.all_nodes.values():
daw = vol.flags.get("daw") or self.args.daw daw = vol.flags.get("daw") or self.args.daw
+6 -13
View File
@@ -199,18 +199,7 @@ class FtpFs(AbstractedFS):
t = "No filesystem mounted at [{}]" t = "No filesystem mounted at [{}]"
raise FSE(t.format(vpath)) raise FSE(t.format(vpath))
if "xdev" in vfs.flags or "xvol" in vfs.flags: ap = vfs.canonical(rem, False)
ap = vfs.canonical(rem)
avfs = vfs.chk_ap(ap)
t = "Permission denied in [{}]"
if not avfs:
raise FSE(t.format(vpath), 1)
cr, cw, cm, cd, _, _, _, _, _ = avfs.uaxs[self.h.uname]
if r and not cr or w and not cw or m and not cm or d and not cd:
raise FSE(t.format(vpath), 1)
else:
ap = vfs.canonical(rem, False)
if "bcasechk" in vfs.flags and not vfs.casechk(rem, True): if "bcasechk" in vfs.flags and not vfs.casechk(rem, True):
raise FSE("No such file or directory", 1) raise FSE("No such file or directory", 1)
@@ -320,7 +309,11 @@ class FtpFs(AbstractedFS):
def chdir(self, path: str) -> None: def chdir(self, path: str) -> None:
nwd = join(self.cwd, path) nwd = join(self.cwd, path)
vfs, rem = self.hub.asrv.vfs.get(nwd, self.uname, False, False) try:
vfs, rem = self.hub.asrv.vfs.get(nwd, self.uname, False, False)
except Exception as ex:
raise FSE(str(ex))
if not vfs.realpath: if not vfs.realpath:
self.cwd = nwd self.cwd = nwd
return return
+18 -5
View File
@@ -885,10 +885,9 @@ class HttpCli(object):
uparam["b"] = "" uparam["b"] = ""
cookies["b"] = "" cookies["b"] = ""
vn, rem = self.asrv.vfs.get(self.vpath, self.uname, False, False) vn, rem = self.asrv.vfs._get(self.vpath, self.uname, False, False)
if vn.realpath and ("xdev" in vn.flags or "xvol" in vn.flags): if vn.realpath and "xvd" in vn.flags:
ap = vn.canonical(rem) avn = vn.chk_ap(vn.canonical(rem))
avn = vn.chk_ap(ap)
else: else:
avn = vn avn = vn
@@ -2475,6 +2474,9 @@ class HttpCli(object):
if "%" in self.req: if "%" in self.req:
self.log(" `-- %r" % (self.vpath,)) self.log(" `-- %r" % (self.vpath,))
if not self.avn:
raise Pebkac(403, "no access")
if self.headers.get("expect", "").lower() == "100-continue": if self.headers.get("expect", "").lower() == "100-continue":
try: try:
self.s.sendall(b"HTTP/1.1 100 Continue\r\n\r\n") self.s.sendall(b"HTTP/1.1 100 Continue\r\n\r\n")
@@ -3830,6 +3832,8 @@ class HttpCli(object):
fdir = vfs.canonical(rem) fdir = vfs.canonical(rem)
fn = os.path.join(fdir, sanitized) fn = os.path.join(fdir, sanitized)
self.asrv.vfs.get(vjoins(vfs.vpath, rem, sanitized), self.uname, False, True)
for hn in ("xbu", "xau"): for hn in ("xbu", "xau"):
xxu = vfs.flags.get(hn) xxu = vfs.flags.get(hn)
if xxu: if xxu:
@@ -3987,6 +3991,13 @@ class HttpCli(object):
if rnd: if rnd:
fname = rand_name(fdir, fname, rnd) fname = rand_name(fdir, fname, rnd)
self.can_delete = (
self.uname
in self.asrv.vfs.get(
vjoin(upload_vpath, fname), self.uname, False, True
)[0].axs.udel
)
abspath = os.path.join(fdir, fname) abspath = os.path.join(fdir, fname)
open_args = {"fdir": fdir, "suffix": suffix, "vf": vfs.flags} open_args = {"fdir": fdir, "suffix": suffix, "vf": vfs.flags}
@@ -4344,7 +4355,7 @@ class HttpCli(object):
bos.makedirs(fp, vf=vfs.flags) bos.makedirs(fp, vf=vfs.flags)
fp = os.path.join(fp, fn) fp = os.path.join(fp, fn)
rem = "{}/{}".format(rp, fn).strip("/") rem = vjoin(rp, fn)
dbv, vrem = vfs.get_dbv(rem) dbv, vrem = vfs.get_dbv(rem)
if not self.can_delete and ( if not self.can_delete and (
@@ -4362,6 +4373,8 @@ class HttpCli(object):
self.reply(response.encode("utf-8")) self.reply(response.encode("utf-8"))
return True return True
self.asrv.vfs.get(vjoin(vfs.vpath, rem), self.uname, True, True)
srv_lastmod = -1.0 srv_lastmod = -1.0
srv_lastmod3 = -1 srv_lastmod3 = -1
try: try:
+1 -12
View File
@@ -378,18 +378,7 @@ class SFTP_Srv(paramiko.SFTPServerInterface):
# return "", vn, rem # return "", vn, rem
raise OSError(errno.ENOENT, "no filesystem mounted at [/%s]" % (vpath,)) raise OSError(errno.ENOENT, "no filesystem mounted at [/%s]" % (vpath,))
if "xdev" in vn.flags or "xvol" in vn.flags: ap = vn.canonical(rem, False)
ap = vn.canonical(rem)
avn = vn.chk_ap(ap)
t = "Permission denied in [{}]"
if not avn:
raise OSError(errno.EPERM, "permission denied in [/%s]" % (vpath,))
cr, cw, cm, cd, _, _, _, _, _ = avn.uaxs[self.uname]
if r and not cr or w and not cw or m and not cm or d and not cd:
raise OSError(errno.EPERM, "permission denied in [/%s]" % (vpath,))
else:
ap = vn.canonical(rem, False)
if "bcasechk" in vn.flags and not vn.casechk(rem, True): if "bcasechk" in vn.flags and not vn.casechk(rem, True):
raise OSError(errno.ENOENT, "file does not exist case-sensitively") raise OSError(errno.ENOENT, "file does not exist case-sensitively")
-5
View File
@@ -276,11 +276,6 @@ class Tftpd(object):
raise Exception("unmapped vfs") raise Exception("unmapped vfs")
ap = vfs.canonical(rem) ap = vfs.canonical(rem)
if "xdev" in vfs.flags or "xvol" in vfs.flags:
avn = vfs.chk_ap(ap)
for n, zb in enumerate(perms):
if not avn or zb and not avn.uaxs["*"][n]:
raise Exception("no access")
if "bcasechk" in vfs.flags and not vfs.casechk(rem, True): if "bcasechk" in vfs.flags and not vfs.casechk(rem, True):
raise Exception("file not found") raise Exception("file not found")
+1 -1
View File
@@ -1179,7 +1179,7 @@ class Up2k(object):
ft = "\033[0;32m{}{:.0}" ft = "\033[0;32m{}{:.0}"
ff = "\033[0;35m{}{:.0}" ff = "\033[0;35m{}{:.0}"
fv = "\033[0;36m{}:\033[90m{}" fv = "\033[0;36m{}:\033[90m{}"
zs = "bcasechk du_iwho emb_all emb_lgs emb_mds ext_th_d html_head html_head_d html_head_s ls_q_m put_name2 mv_re_r mv_re_t rm_re_r rm_re_t oh_f oh_g rw_edit_set srch_re_dots srch_re_nodot th_coversd th_coversl th_covers_set th_coversd_set zipmax zipmaxn_v zipmaxs_v" zs = "bcasechk du_iwho emb_all emb_lgs emb_mds ext_th_d html_head html_head_d html_head_s ls_q_m put_name2 mv_re_r mv_re_t rm_re_r rm_re_t oh_f oh_g rw_edit_set srch_re_dots srch_re_nodot th_coversd th_coversl th_covers_set th_coversd_set xvd zipmax zipmaxn_v zipmaxs_v"
fx = set(zs.split()) fx = set(zs.split())
fd = vf_bmap() fd = vf_bmap()
fd.update(vf_cmap()) fd.update(vf_cmap())
+2
View File
@@ -469,6 +469,8 @@ IMPLICATIONS = [
["smbvvv", "smbvv"], ["smbvvv", "smbvv"],
["smbvv", "smbv"], ["smbvv", "smbv"],
["smbv", "smb"], ["smbv", "smb"],
["xdev", "xvd"],
["xvol", "xvd"],
["zv", "zmv"], ["zv", "zmv"],
["zv", "zsv"], ["zv", "zsv"],
["z", "zm"], ["z", "zm"],
+87 -31
View File
@@ -42,7 +42,9 @@ class TestShr(unittest.TestCase):
if edl is None: if edl is None:
edl = els[1] edl = els[1]
can_dl = [] can_dl = []
for fn in b[1]: zs = "f1 f2 f3 f4 l1 l2 l3 l4"
known_fns = list(sorted(list(set(b[1] + zs.split()))))
for fn in known_fns:
if fn == "a.db": if fn == "a.db":
continue continue
furl = url + "/" + fn furl = url + "/" + fn
@@ -51,7 +53,9 @@ class TestShr(unittest.TestCase):
h, zb = self.curl(furl, True) h, zb = self.curl(furl, True)
if h.startswith("HTTP/1.1 200 "): if h.startswith("HTTP/1.1 200 "):
can_dl.append(fn) can_dl.append(fn)
self.assertEqual(edl, can_dl) elif zb.startswith(b">>fb:"):
can_dl.append(fn)
self.assertEqual(edl, list(sorted(list(set(can_dl)))))
def setUp(self): def setUp(self):
self.td = tu.get_ramdisk() self.td = tu.get_ramdisk()
@@ -61,12 +65,15 @@ class TestShr(unittest.TestCase):
os.mkdir("d1") os.mkdir("d1")
os.mkdir("d2") os.mkdir("d2")
os.mkdir("d2/d3") os.mkdir("d2/d3")
for zs in ("d1/f1", "d2/f2", "d2/d3/f3"): os.mkdir("d2/d3/d4")
for zs in ("d1/f1", "d2/f2", "d2/d3/f3", "d2/d3/d4/f4"):
with open(zs, "wb") as f: with open(zs, "wb") as f:
f.write(zs.encode("utf-8")) f.write(b">>fb:" + zs.encode("utf-8"))
for dst in ("d1", "d2", "d2/d3"): for dst in ("d1", "d2", "d2/d3", "d2/d3/d4"):
src, fn = zs.rsplit("/", 1) src, fn = zs.rsplit("/", 1)
os.symlink(absreal(zs), dst + "/l" + fn[-1:]) os.symlink(absreal(zs), dst + "/l" + fn[-1:])
os.symlink(absreal("d2/d3"), absreal("d2/ld"))
os.symlink(absreal("d2/d3"), absreal("d2/d3/d4/ld"))
db = sqlite3.connect("a.db") db = sqlite3.connect("a.db")
with db: with db:
@@ -92,12 +99,14 @@ class TestShr(unittest.TestCase):
shr_v=False, shr_v=False,
) )
self.cinit() self.cinit()
L1234 = ["l1", "l2", "l3", "l4"]
self.assertLD("", True, [["d1", "d2", "v1"], ["a.db"]], []) self.assertLD("", True, [["d1", "d2", "v1"], ["a.db"]], [])
self.assertLD("d1", True, [[], ["f1", "l1", "l2", "l3"]], None) self.assertLD("d1", True, [[], ["f1"] + L1234], None)
self.assertLD("v1", True, [[], ["f1", "l1", "l2", "l3"]], None) self.assertLD("v1", True, [[], ["f1"] + L1234], None)
self.assertLD("d2", True, [["d3"], ["f2", "l1", "l2", "l3"]], None) self.assertLD("d2", True, [["d3", "ld"], ["f2"] + L1234], None)
self.assertLD("d2/d3", True, [[], ["f3", "l1", "l2", "l3"]], None) self.assertLD("d2/d3", True, [["d4"], ["f3"] + L1234], None)
self.assertLD("d2/d3/d4", True, [["ld"], ["f4"] + L1234], None)
self.assertLD("d3", True, [], []) self.assertLD("d3", True, [], [])
jt = { jt = {
@@ -105,7 +114,7 @@ class TestShr(unittest.TestCase):
"vp": ["/"], "vp": ["/"],
"pw": "", "pw": "",
"exp": "99", "exp": "99",
"perms": ["read"], "perms": ["read", "write"],
} }
print(self.post_json("?pw=p1&share", jt)[1]) print(self.post_json("?pw=p1&share", jt)[1])
jt = { jt = {
@@ -113,26 +122,48 @@ class TestShr(unittest.TestCase):
"vp": ["/d2/"], "vp": ["/d2/"],
"pw": "", "pw": "",
"exp": "99", "exp": "99",
"perms": ["read"], "perms": ["read", "write"],
}
print(self.post_json("?pw=p1&share", jt)[1])
jt = {
"k": "d4",
"vp": ["/d2/d3/d4/"],
"pw": "",
"exp": "99",
"perms": ["read", "write"],
} }
print(self.post_json("?pw=p1&share", jt)[1]) print(self.post_json("?pw=p1&share", jt)[1])
self.conn.shutdown() self.conn.shutdown()
self.cinit() self.cinit()
self.assertLD("", True, [["d1", "d2", "v1"], ["a.db"]], []) self.assertLD("", True, [["d1", "d2", "v1"], ["a.db"]], [])
self.assertLD("d1", True, [[], ["f1", "l1", "l2", "l3"]], None) self.assertLD("d1", True, [[], ["f1"] + L1234], None)
self.assertLD("v1", True, [[], ["f1", "l1", "l2", "l3"]], None) self.assertLD("v1", True, [[], ["f1"] + L1234], None)
self.assertLD("d2", True, [["d3"], ["f2", "l1", "l2", "l3"]], None) self.assertLD("d2", True, [["d3", "ld"], ["f2"] + L1234], None)
self.assertLD("d2/d3", True, [[], ["f3", "l1", "l2", "l3"]], None) self.assertLD("d2/d3", True, [["d4"], ["f3"] + L1234], None)
self.assertLD("d2/d3/d4", True, [["ld"], ["f4"] + L1234], None)
self.assertLD("d3", True, [], []) self.assertLD("d3", True, [], [])
self.assertLD("shr/d2", False, [[], ["f2", "l1", "l2", "l3"]], None) self.assertLD("shr/d2", False, [["ld"], ["f2"] + L1234], None)
self.assertLD("shr/d2/d3", False, [], None) self.assertLD("shr/d2/d3", False, [], None)
self.assertLD("shr/d2/d3/d4", False, [], None)
self.assertLD("shr/d4", False, [["ld"], ["f4"] + L1234], None)
self.assertLD("shr/r", False, [["d1"], ["a.db"]], []) self.assertLD("shr/r", False, [["d1"], ["a.db"]], [])
self.assertLD("shr/r/d1", False, [[], ["f1", "l1", "l2", "l3"]], None) self.assertLD("shr/r/d1", False, [[], ["f1"] + L1234], None)
self.assertLD("shr/r/d2", False, [], None) # unfortunate self.assertLD("shr/r/d2", False, [], None) # unfortunate
self.assertLD("shr/r/d2/d3", False, [], None) self.assertLD("shr/r/d2/d3", False, [], None)
self.assertLD("shr/r/d2/d3/d4", False, [], None)
self.put("shr/d2/a", 201)
os.unlink("d2/a")
self.put("shr/d2/ld/a", 201)
os.unlink("d2/d3/a")
self.put("shr/d4/ld/a", 201)
os.unlink("d2/d3/a")
self.conn.shutdown() self.conn.shutdown()
@@ -147,12 +178,14 @@ class TestShr(unittest.TestCase):
xvol=True, xvol=True,
) )
self.cinit() self.cinit()
L1234 = ["l1", "l2", "l3", "l4"]
self.assertLD("", True, [["d1", "d2", "v1"], ["a.db"]], []) self.assertLD("", True, [["d1", "d2", "v1"], ["a.db"]], [])
self.assertLD("d1", True, [[], ["f1", "l1", "l2", "l3"]], None) self.assertLD("d1", True, [[], ["f1"] + L1234], None)
self.assertLD("v1", True, [[], ["f1", "l1", "l2", "l3"]], None) self.assertLD("v1", True, [[], ["f1"] + L1234], None)
self.assertLD("d2", True, [["d3"], ["f2", "l1", "l2", "l3"]], None) self.assertLD("d2", True, [["d3", "ld"], ["f2"] + L1234], None)
self.assertLD("d2/d3", True, [[], ["f3", "l1", "l2", "l3"]], None) self.assertLD("d2/d3", True, [["d4"], ["f3"] + L1234], None)
self.assertLD("d2/d3/d4", True, [["ld"], ["f4"] + L1234], None)
self.assertLD("d3", True, [], []) self.assertLD("d3", True, [], [])
jt = { jt = {
@@ -160,7 +193,7 @@ class TestShr(unittest.TestCase):
"vp": ["/"], "vp": ["/"],
"pw": "", "pw": "",
"exp": "99", "exp": "99",
"perms": ["read"], "perms": ["read", "write"],
} }
print(self.post_json("?pw=p1&share", jt)[1]) print(self.post_json("?pw=p1&share", jt)[1])
jt = { jt = {
@@ -168,28 +201,45 @@ class TestShr(unittest.TestCase):
"vp": ["/d2/"], "vp": ["/d2/"],
"pw": "", "pw": "",
"exp": "99", "exp": "99",
"perms": ["read"], "perms": ["read", "write"],
}
print(self.post_json("?pw=p1&share", jt)[1])
jt = {
"k": "d4",
"vp": ["/d2/d3/d4/"],
"pw": "",
"exp": "99",
"perms": ["read", "write"],
} }
print(self.post_json("?pw=p1&share", jt)[1]) print(self.post_json("?pw=p1&share", jt)[1])
self.conn.shutdown() self.conn.shutdown()
self.cinit() self.cinit()
self.assertLD("", True, [["d1", "d2", "v1"], ["a.db"]], []) self.assertLD("", True, [["d1", "d2", "v1"], ["a.db"]], [])
self.assertLD("d1", True, [[], ["f1", "l1", "l2", "l3"]], None) self.assertLD("d1", True, [[], ["f1"] + L1234], None)
self.assertLD("v1", True, [[], ["f1", "l1", "l2", "l3"]], None) self.assertLD("v1", True, [[], ["f1"] + L1234], None)
self.assertLD("d2", True, [["d3"], ["f2", "l1", "l2", "l3"]], None) self.assertLD("d2", True, [["d3", "ld"], ["f2"] + L1234], None)
self.assertLD("d2/d3", True, [[], ["f3", "l1", "l2", "l3"]], None) self.assertLD("d2/d3", True, [["d4"], ["f3"] + L1234], None)
self.assertLD("d2/d3/d4", True, [["ld"], ["f4"] + L1234], None)
self.assertLD("d3", True, [], []) self.assertLD("d3", True, [], [])
self.assertLD("shr/d2", False, [[], ["f2", "l1", "l2", "l3"]], ["f2", "l2"]) self.assertLD("shr/d2", False, [["ld"], ["f2"] + L1234], ["f2", "l2"])
self.assertLD("shr/d2/d3", False, [], []) self.assertLD("shr/d2/d3", False, [], [])
self.assertLD("shr/d2/d3/d4", False, [], None)
self.assertLD("shr/d4", False, [["ld"], ["f4"] + L1234], ["f4", "l4"])
self.assertLD("shr/r", False, [["d1"], ["a.db"]], []) self.assertLD("shr/r", False, [["d1"], ["a.db"]], [])
self.assertLD( self.assertLD("shr/r/d1", False, [[], ["f1"] + L1234], ["f1", "l1", "l2"])
"shr/r/d1", False, [[], ["f1", "l1", "l2", "l3"]], ["f1", "l1", "l2"]
)
self.assertLD("shr/r/d2", False, [], []) # unfortunate self.assertLD("shr/r/d2", False, [], []) # unfortunate
self.assertLD("shr/r/d2/d3", False, [], []) self.assertLD("shr/r/d2/d3", False, [], [])
self.assertLD("shr/r/d2/d3/d4", False, [], None)
self.put("shr/d2/a", 201)
os.unlink("d2/a")
self.put("shr/d2/ld/a", 401)
self.put("shr/d4/ld/a", 401)
self.conn.shutdown() self.conn.shutdown()
@@ -227,3 +277,9 @@ class TestShr(unittest.TestCase):
print("PUT -->", buf) print("PUT -->", buf)
HttpCli(self.conn.setbuf(buf)).run() HttpCli(self.conn.setbuf(buf)).run()
return self.conn.s._reply.decode("utf-8").split("\r\n\r\n", 1) return self.conn.s._reply.decode("utf-8").split("\r\n\r\n", 1)
def put(self, url: str, expect: str):
h = "PUT /%s HTTP/1.1\r\nContent-Length: 1\r\n\r\n`"
HttpCli(self.conn.setbuf((h % (url,)).encode("utf-8"))).run()
h, r = self.conn.s._reply.decode("utf-8").split("\r\n\r\n", 1)
self.assertEqual(expect, int(h.split(" ")[1]))
+3 -2
View File
@@ -163,7 +163,7 @@ class Cfg(Namespace):
ex = "ac_convt au_vol dl_list du_iwho mcr mtab_age reg_cap s_thead s_tbody tail_tmax tail_who th_convt th_qv th_qvx ups_who ver_iwho zip_who" ex = "ac_convt au_vol dl_list du_iwho mcr mtab_age reg_cap s_thead s_tbody tail_tmax tail_who th_convt th_qv th_qvx ups_who ver_iwho zip_who"
ka.update(**{k: 9 for k in ex.split()}) ka.update(**{k: 9 for k in ex.split()})
ex = "ctl_re db_act forget_ip gauto idp_cookie idp_store k304 loris md_nhist no304 nosubtle qr_pin qr_wait re_maxage rproxy rsp_jtr rsp_slp s_wr_slp snap_wri tftp theme themes turbo u2ow zipmaxn zipmaxs" ex = "ctl_re dav_port db_act forget_ip gauto idp_cookie idp_store k304 loris md_nhist no304 nosubtle qr_pin qr_wait re_maxage rproxy rsp_jtr rsp_slp s_wr_slp snap_wri tftp theme themes turbo u2ow zipmaxn zipmaxs"
ka.update(**{k: 0 for k in ex.split()}) ka.update(**{k: 0 for k in ex.split()})
ex = "ah_alg bname chdir chmod_f chpw_db csp_dl csp_ui db_xattr doctitle df epilogues exit favico fika ipa ipar html_head html_head_d html_head_s idp_login idp_logout lg_sba lg_sbf log_date log_fk md_sba md_sbf name og_desc og_site og_th og_title og_title_a og_title_v og_title_i opds_exts preadmes prologues readmes redup shr shr1 shr_site site smsg tcolor textfiles th_pregen txt_eol ufavico ufavico_h unlist up_site vc_url vname xff_src zipmaxt R RS SR" ex = "ah_alg bname chdir chmod_f chpw_db csp_dl csp_ui db_xattr doctitle df epilogues exit favico fika ipa ipar html_head html_head_d html_head_s idp_login idp_logout lg_sba lg_sbf log_date log_fk md_sba md_sbf name og_desc og_site og_th og_title og_title_a og_title_v og_title_i opds_exts preadmes prologues readmes redup shr shr1 shr_site site smsg tcolor textfiles th_pregen txt_eol ufavico ufavico_h unlist up_site vc_url vname xff_src zipmaxt R RS SR"
@@ -172,7 +172,7 @@ class Cfg(Namespace):
ex = "apnd_who ban_403 ban_404 ban_422 ban_pw ban_pwc ban_url dont_ban cachectl http_vary rcm rss_fmt_d rss_fmt_t spinner" ex = "apnd_who ban_403 ban_404 ban_422 ban_pw ban_pwc ban_url dont_ban cachectl http_vary rcm rss_fmt_d rss_fmt_t spinner"
ka.update(**{k: "no" for k in ex.split()}) ka.update(**{k: "no" for k in ex.split()})
ex = "ext_th grp idp_h_usr idp_hm_usr ipr on403 on404 qr_file xac xad xar xau xban xbc xbd xbr xbu xiu xm" ex = "ext_th grp idp_h_usr idp_hm_usr ipr on403 on404 p_nodav qr_file xac xad xar xau xban xbc xbd xbr xbu xiu xm"
ka.update(**{k: [] for k in ex.split()}) ka.update(**{k: [] for k in ex.split()})
ex = "exp_lg exp_md" ex = "exp_lg exp_md"
@@ -227,6 +227,7 @@ class Cfg(Namespace):
th_crop="y", th_crop="y",
th_size="320x256", th_size="320x256",
th_x3="n", th_x3="n",
ua_nodav=re.compile("\n"),
u2sort="s", u2sort="s",
u2ts="c", u2ts="c",
ui_filesz="1", ui_filesz="1",