mirror of
https://github.com/9001/copyparty.git
synced 2026-11-10 12:28:23 +00:00
shadowing + casefolding (GHSA-3c88-p299-7hvr)
This commit is contained in:
1 parent
36aaa24974
commit
e8de421174
9 files changed
+142
-64
No files matched your search
@@ -622,6 +622,8 @@ config file example:
|
||||
|
||||
hiding specific subfolders by mounting another volume on top of them
|
||||
|
||||
> **WARNING:** shadowing is NOT bulletproof when the underlying filesystem is casefolding (case-insensitive), in other words DON'T rely on this when copyparty is running on Windows and/or MacOS. This is also unsafe when combined with certain filesystems on Linux, especially FAT32/exFAT. Pay attention to the volume-listing in the startup log; `casechk:n` indicates probably-OK, but if it says `casechk:y` then DO NOT trust shadowing to fully prevent access to a file/subfolder.
|
||||
|
||||
for example `-v /mnt::r -v /var/empty:web/certs:` (note: no permissions) mounts the server folder `/mnt` as the webroot, but another volume is mounted at `/web/certs` -- so visitors can only see the contents of `/mnt` and `/mnt/web` (at URLs `/` and `/web`), but not `/mnt/web/certs` because URL `/web/certs` is mapped to `/var/empty`
|
||||
|
||||
to fully unmap it from the filesystem, specify `//NULL` instead of a real path such as `/var/empty`, so for example `-v /mnt::r -v //NULL:web/certs:` ensures `/web/certs` will never be accessible by anyone
|
||||
@@ -3047,7 +3049,7 @@ below are some tweaks roughly ordered by usefulness:
|
||||
* and pypy can sometimes crash on startup with `-j0` (TODO make issue)
|
||||
|
||||
* if you are running the copyparty server **on Windows or Macos:**
|
||||
* `--casechk=n` makes it much faster, but also awakens [the usual surprises](https://github.com/9001/copyparty/issues/781) you expect from a case-insensitive filesystem
|
||||
* `--casechk=n` makes it much faster, but [reduces security](https://github.com/9001/copyparty/security/advisories/GHSA-3c88-p299-7hvr) and awakens [the usual surprises](https://github.com/9001/copyparty/issues/781) you expect from a case-insensitive filesystem
|
||||
* this is the same as `casechk: n` in a config-file
|
||||
|
||||
|
||||
|
||||
@@ -42,6 +42,7 @@ from .util import (
|
||||
odfusion,
|
||||
read_utf8,
|
||||
relchk,
|
||||
sanitize_fn,
|
||||
statdir,
|
||||
ub64enc,
|
||||
uncyg,
|
||||
@@ -426,6 +427,7 @@ class VFS(object):
|
||||
self.root = self
|
||||
self.dev = 0 # st_dev
|
||||
self.nodes: dict[str, VFS] = {} # child nodes
|
||||
self.lnodes: Optional[dict[bytes, VFS]] = None # lowercase
|
||||
self.histtab: dict[str, str] = {} # all realpath->histpath
|
||||
self.dbpaths: dict[str, str] = {} # all realpath->dbpath
|
||||
self.dbv: Optional[VFS] = None # closest full/non-jump parent
|
||||
@@ -575,6 +577,11 @@ class VFS(object):
|
||||
if name in self.nodes:
|
||||
return self.nodes[name]._find(rem)
|
||||
|
||||
if self.lnodes:
|
||||
zs = name.casefold().encode("ascii", "ignore")
|
||||
if zs in self.lnodes:
|
||||
return self.lnodes[zs]._find(rem)
|
||||
|
||||
return self, vpath
|
||||
|
||||
def can_access(
|
||||
@@ -784,6 +791,12 @@ class VFS(object):
|
||||
if not rem:
|
||||
# no vfs nodes in the list of real inodes
|
||||
real = [x for x in real if x[0] not in self.nodes]
|
||||
if self.lnodes:
|
||||
real = [
|
||||
x
|
||||
for x in real
|
||||
if x[0].casefold().encode("ascii", "ignore") not in self.lnodes
|
||||
]
|
||||
|
||||
dbv = self.dbv or self
|
||||
for name, vn2 in sorted(self.nodes.items()):
|
||||
@@ -949,6 +962,22 @@ class VFS(object):
|
||||
for d in [{"vp": v, "ap": a, "st": n} for v, a, n in ret2]:
|
||||
yield d
|
||||
|
||||
def sanitize_fn(self, fn: str, rem: Any = None) -> str:
|
||||
fn = sanitize_fn(fn, self.flags["fsnt"])
|
||||
if rem:
|
||||
return fn
|
||||
if fn in self.nodes or (
|
||||
self.lnodes and fn.casefold().encode("ascii", "ignore") in self.lnodes
|
||||
):
|
||||
raise Pebkac(400, "bad filename")
|
||||
return fn
|
||||
|
||||
def chk_fn(self, fn: str) -> None:
|
||||
if fn in self.nodes or (
|
||||
self.lnodes and fn.casefold().encode("ascii", "ignore") in self.lnodes
|
||||
):
|
||||
raise Pebkac(400, "bad filename")
|
||||
|
||||
def chk_ap(self, ap: str, st: Optional[os.stat_result] = None) -> Optional["VFS"]:
|
||||
aps = ap + os.sep
|
||||
if "xdev" in self.flags and not ANYWIN:
|
||||
@@ -3304,6 +3333,34 @@ class AuthSrv(object):
|
||||
cur.close()
|
||||
db.close()
|
||||
|
||||
zb1 = zb2 = False
|
||||
nodes = [self.vfs]
|
||||
while nodes:
|
||||
vn = nodes.pop()
|
||||
nodes += list(vn.nodes.values())
|
||||
if not vn.realpath:
|
||||
continue
|
||||
if "bcasechk" in vn.flags:
|
||||
if ".hist" in vn.nodes:
|
||||
zb1 = True
|
||||
if vn.nodes and (len(vn.nodes) > 1 or ".hist" not in vn.nodes):
|
||||
zb2 = True
|
||||
if PY2:
|
||||
continue
|
||||
vn.lnodes = {
|
||||
k.casefold().encode("ascii", "ignore"): v
|
||||
for k, v in vn.nodes.items()
|
||||
}
|
||||
if zb2 and PY2:
|
||||
t = "SECURITY WARNING: due to python2, shadowing will NOT provide much security:"
|
||||
self.log(t, 1)
|
||||
if zb2:
|
||||
t = "SECURITY WARNING: you have a case-insensitive / casefolding filesystem, and you also have nested volumes (subvolumes). This combination is NOT 100% bulletproof; it *might* be possible for someone to sneakily access a subvolume they're not supposed to have access to, if they have access to the parent volume."
|
||||
self.log(t, 3)
|
||||
if zb1:
|
||||
t = "you have a case-insensitive / casefolding filesystem, so it is recommended to set --hist C:\\copyparty-data\\ or another safe place"
|
||||
self.log(t, 3)
|
||||
|
||||
self.js_ls = {}
|
||||
self.js_htm = ""
|
||||
self.md_htm = ""
|
||||
|
||||
+16
-10
@@ -15,7 +15,7 @@ from pyftpdlib.handlers import FTPHandler
|
||||
from pyftpdlib.ioloop import IOLoop
|
||||
from pyftpdlib.servers import FTPServer
|
||||
|
||||
from .__init__ import PY2, TYPE_CHECKING
|
||||
from .__init__ import ANYWIN, PY2, TYPE_CHECKING
|
||||
from .authsrv import VFS
|
||||
from .bos import bos
|
||||
from .util import (
|
||||
@@ -29,7 +29,6 @@ from .util import (
|
||||
ipnorm,
|
||||
pybin,
|
||||
read_utf8,
|
||||
relchk,
|
||||
runhook,
|
||||
sanitize_fn,
|
||||
set_fperms,
|
||||
@@ -169,16 +168,23 @@ class FtpFs(AbstractedFS):
|
||||
d: bool = False,
|
||||
) -> tuple[str, VFS, str]:
|
||||
try:
|
||||
vpath = vpath.replace("\\", "/").strip("/")
|
||||
rd, fn = os.path.split(vpath)
|
||||
if relchk(rd):
|
||||
logging.warning("malicious vpath: %s", vpath)
|
||||
t = "Unsupported characters in [{}]"
|
||||
raise FSE(t.format(vpath), 1)
|
||||
# because:
|
||||
# * ftp-spec doesn't define sep
|
||||
# * windows filename limitations
|
||||
# need to allow/translate illegal fn,
|
||||
# but require filesystem-correct rd
|
||||
if ANYWIN or ("\\" in vpath and "/" not in vpath):
|
||||
vpath = vpath.replace("\\", "/")
|
||||
vpath = vpath.strip("/\\")
|
||||
if ANYWIN:
|
||||
rd, fn = os.path.split(vpath)
|
||||
fn = sanitize_fn(fn, "win")
|
||||
vpath = vjoin(rd, fn)
|
||||
# end of windows jank
|
||||
|
||||
fn = sanitize_fn(fn or "")
|
||||
vpath = vjoin(rd, fn)
|
||||
vfs, rem = self.hub.asrv.vfs.get(vpath, self.uname, r, w, m, d)
|
||||
vpath = vjoin(vfs.vpath, rem)
|
||||
rd, fn = os.path.split(vpath)
|
||||
if (
|
||||
w
|
||||
and fn.lower() in vfs.flags["emb_all"]
|
||||
|
||||
+15
-17
@@ -110,7 +110,6 @@ from .util import (
|
||||
s2hms,
|
||||
s3enc,
|
||||
safe_mime,
|
||||
sanitize_fn,
|
||||
sanitize_vpath,
|
||||
sendfile_kern,
|
||||
sendfile_py,
|
||||
@@ -2736,7 +2735,7 @@ class HttpCli(object):
|
||||
if rnd:
|
||||
fn = rand_name(fdir, fn, rnd)
|
||||
|
||||
fn = sanitize_fn(fn or "")
|
||||
fn = vfs.sanitize_fn(fn or "", rem)
|
||||
|
||||
path = os.path.join(fdir, fn)
|
||||
|
||||
@@ -3216,19 +3215,20 @@ class HttpCli(object):
|
||||
if "delete" in self.uparam:
|
||||
return self.handle_rm(body)
|
||||
|
||||
name = undot(body["name"])
|
||||
if "/" in name:
|
||||
raise Pebkac(400, "your client is old; press CTRL-SHIFT-R and try again")
|
||||
name = body["name"]
|
||||
if "/" in name or name in (".", ".."):
|
||||
raise Pebkac(400, "filename contains '/' or is invalid")
|
||||
|
||||
vfs, rem = self.asrv.vfs.get(self.vpath, self.uname, False, True)
|
||||
fsnt = vfs.flags["fsnt"]
|
||||
if fsnt != "lin":
|
||||
vpath = self.vpath
|
||||
fsnt = self.vn.flags["fsnt"]
|
||||
if not VPTL_OS and fsnt != "lin":
|
||||
tl = VPTL_WIN if fsnt == "win" else VPTL_MAC
|
||||
rem = rem.translate(tl)
|
||||
name = name.translate(tl)
|
||||
vpath = vpath.translate(tl)
|
||||
|
||||
vfs, rem = self.asrv.vfs.get(vpath, self.uname, False, True)
|
||||
name = vfs.sanitize_fn(name, rem)
|
||||
dbv, vrem = vfs.get_dbv(rem)
|
||||
|
||||
name = sanitize_fn(name)
|
||||
if (
|
||||
not self.can_read
|
||||
and self.can_write
|
||||
@@ -3593,7 +3593,6 @@ class HttpCli(object):
|
||||
return self.tx_404()
|
||||
|
||||
vfs, rem = self.asrv.vfs.get(session["vp"], self.uname, False, True)
|
||||
vpath = vjoin(vfs.vpath, rem)
|
||||
ap = vfs.canonical(rem)
|
||||
st = bos.stat(ap)
|
||||
|
||||
@@ -3776,12 +3775,11 @@ class HttpCli(object):
|
||||
def _mkdir(self, vpath: str, dav: bool = False) -> bool:
|
||||
nullwrite = self.args.nw
|
||||
self.gctx = vpath
|
||||
vpath = sanitize_vpath(undot(vpath))
|
||||
vpath = sanitize_vpath(undot(vpath), self.vn.flags["fsnt"])
|
||||
vfs, rem = self.asrv.vfs.get(vpath, self.uname, False, True)
|
||||
if "nosub" in vfs.flags:
|
||||
raise Pebkac(403, "mkdir is forbidden below this folder")
|
||||
|
||||
rem = sanitize_vpath(rem)
|
||||
fn = vfs.canonical(rem)
|
||||
|
||||
if not nullwrite:
|
||||
@@ -3828,7 +3826,7 @@ class HttpCli(object):
|
||||
t = "you can only create %s files because you don't have the delete-permission"
|
||||
raise Pebkac(400, t % (vfs.flags["rw_edit"].replace(",", "/")))
|
||||
|
||||
sanitized = sanitize_fn(new_file)
|
||||
sanitized = vfs.sanitize_fn(new_file, rem)
|
||||
fdir = vfs.canonical(rem)
|
||||
fn = os.path.join(fdir, sanitized)
|
||||
|
||||
@@ -3983,7 +3981,7 @@ class HttpCli(object):
|
||||
# fallthrough
|
||||
|
||||
fdir = fdir_base
|
||||
fname = sanitize_fn(p_file or "")
|
||||
fname = vfs.sanitize_fn(p_file or "", rem)
|
||||
suffix = "-%.6f-%s" % (time.time(), dip)
|
||||
if p_file and not nullwrite:
|
||||
if rnd:
|
||||
@@ -7272,7 +7270,7 @@ class HttpCli(object):
|
||||
if use_dirkey:
|
||||
is_dk = True
|
||||
elif self.can_get and "doc" in self.uparam:
|
||||
zs = vjoin(self.vpath, self.uparam["doc"]) + "?v"
|
||||
zs = vjoin(self.vpath, self.uparam["doc"].split("//")[-1]) + "?v"
|
||||
return self.redirect(zs, flavor="redirecting to", use302=True)
|
||||
elif not self.can_write:
|
||||
return self.tx_404(True)
|
||||
|
||||
+18
-14
@@ -37,7 +37,6 @@ from .util import (
|
||||
ipnorm,
|
||||
min_ex,
|
||||
read_utf8,
|
||||
relchk,
|
||||
runhook,
|
||||
sanitize_fn,
|
||||
ub64enc,
|
||||
@@ -337,15 +336,23 @@ class SFTP_Srv(paramiko.SFTPServerInterface):
|
||||
m: bool = False,
|
||||
d: bool = False,
|
||||
) -> tuple[str, VFS, str]:
|
||||
vpath = vpath.replace(os.sep, "/").strip("/")
|
||||
rd, fn = os.path.split(vpath)
|
||||
if relchk(rd):
|
||||
self.log("malicious vpath: %s", vpath)
|
||||
raise Exception("Unsupported characters in [%s]" % (vpath,))
|
||||
# because:
|
||||
# * maybe some clients use \
|
||||
# * windows filename limitations
|
||||
# need to allow/translate illegal fn,
|
||||
# but require filesystem-correct rd
|
||||
if ANYWIN or ("\\" in vpath and "/" not in vpath):
|
||||
vpath = vpath.replace("\\", "/")
|
||||
vpath = vpath.strip("/\\")
|
||||
if ANYWIN:
|
||||
rd, fn = os.path.split(vpath)
|
||||
fn = sanitize_fn(fn, "win")
|
||||
vpath = vjoin(rd, fn)
|
||||
# end of windows jank
|
||||
|
||||
fn = sanitize_fn(fn or "")
|
||||
vpath = vjoin(rd, fn)
|
||||
vn, rem = self.hub.asrv.vfs.get(vpath, self.uname, r, w, m, d)
|
||||
vpath = vjoin(vn.vpath, rem)
|
||||
rd, fn = os.path.split(vpath)
|
||||
if (
|
||||
w
|
||||
and fn.lower() in vn.flags["emb_all"]
|
||||
@@ -684,8 +691,7 @@ class SFTP_Srv(paramiko.SFTPServerInterface):
|
||||
def _mkdir(self, vp: str, attr: SATTR) -> int:
|
||||
self.log("mkdir(%s)" % (vp,))
|
||||
try:
|
||||
vn, rem = self.asrv.vfs.get(vp, self.uname, False, True)
|
||||
ap = vn.canonical(rem, False)
|
||||
ap, vn, _ = self.v2a(vp, w=True)
|
||||
bos.makedirs(ap, vf=vn.flags) # filezilla expects this
|
||||
if attr is not None:
|
||||
paramiko.SFTPServer.set_file_attr(ap, attr)
|
||||
@@ -708,8 +714,7 @@ class SFTP_Srv(paramiko.SFTPServerInterface):
|
||||
def _rmdir(self, vp: str) -> int:
|
||||
self.log("rmdir(%s)" % (vp,))
|
||||
try:
|
||||
vn, rem = self.asrv.vfs.get(vp, self.uname, False, False, will_del=True)
|
||||
ap = os.path.join(vn.realpath, rem)
|
||||
ap, _, _ = self.v2a(vp, d=True)
|
||||
bos.rmdir(ap)
|
||||
return SFTP_OK
|
||||
except Pebkac as ex:
|
||||
@@ -730,8 +735,7 @@ class SFTP_Srv(paramiko.SFTPServerInterface):
|
||||
def _chattr(self, vp: str, attr: SATTR) -> int:
|
||||
self.log("chattr(%s, %s)" % (vp, attr))
|
||||
try:
|
||||
vn, rem = self.asrv.vfs.get(vp, self.uname, False, True, will_del=True)
|
||||
ap = os.path.join(vn.realpath, rem)
|
||||
ap, _, _ = self.v2a(vp, w=True, d=True)
|
||||
paramiko.SFTPServer.set_file_attr(ap, attr)
|
||||
return SFTP_OK
|
||||
except Pebkac as ex:
|
||||
|
||||
+1
-1
@@ -104,7 +104,7 @@ def gen_hdr(
|
||||
ret += spack(b"<LL", vsz, vsz)
|
||||
|
||||
# windows support (the "?" replace below too)
|
||||
fn = sanitize_to(fn, VPTL_WIN)
|
||||
fn = sanitize_to(fn, VPTL_WIN, "win")
|
||||
bfn = fn.encode("utf-8" if utf8 else "cp437", "replace").replace(b"?", b"_")
|
||||
|
||||
# add ntfs (0x24) and/or unix (0x10) extrafields for utc, add z64 if requested
|
||||
|
||||
+10
-1
@@ -275,7 +275,16 @@ class Tftpd(object):
|
||||
if not vfs.realpath:
|
||||
raise Exception("unmapped vfs")
|
||||
|
||||
return vfs, vpath, 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):
|
||||
raise Exception("file not found")
|
||||
|
||||
return vfs, vpath, ap
|
||||
|
||||
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
|
||||
|
||||
+2
-2
@@ -3550,7 +3550,7 @@ class Up2k(object):
|
||||
job["ptop"] = vfs.realpath
|
||||
job["vtop"] = vfs.vpath
|
||||
job["prel"] = rem
|
||||
job["name"] = sanitize_fn(job["name"])
|
||||
job["name"] = vfs.sanitize_fn(job["name"], rem)
|
||||
ud2 = (vfs.vpath, job["prel"], job["name"])
|
||||
if ud1 != ud2:
|
||||
# print(json.dumps(job, sort_keys=True, indent=4))
|
||||
@@ -5439,7 +5439,7 @@ class Up2k(object):
|
||||
job["ptop"] = vfs.realpath
|
||||
job["vtop"] = vfs.vpath
|
||||
job["prel"] = rem
|
||||
job["name"] = sanitize_fn(job["name"])
|
||||
job["name"] = vfs.sanitize_fn(job["name"], rem)
|
||||
ud2 = (vfs.vpath, job["prel"], job["name"])
|
||||
if ud1 != ud2:
|
||||
job["vcfg"] = vf
|
||||
|
||||
+20
-18
@@ -365,13 +365,9 @@ if PY2:
|
||||
else:
|
||||
umktrans = str.maketrans
|
||||
|
||||
FNTL_WIN = umktrans('<>:|?*"\\/', "<>:|?*"\/")
|
||||
VPTL_WIN = umktrans('<>:|?*"\\', "<>:|?*"\")
|
||||
APTL_WIN = umktrans('<>:|?*"/', "<>:|?*"/")
|
||||
FNTL_MAC = VPTL_MAC = APTL_MAC = umktrans(":", ":")
|
||||
FNTL_OS = FNTL_WIN if ANYWIN else FNTL_MAC if MACOS else None
|
||||
VPTL_MAC = umktrans(":", ":")
|
||||
VPTL_OS = VPTL_WIN if ANYWIN else VPTL_MAC if MACOS else None
|
||||
APTL_OS = APTL_WIN if ANYWIN else APTL_MAC if MACOS else None
|
||||
|
||||
|
||||
BADNAMES = []
|
||||
@@ -2476,29 +2472,35 @@ else:
|
||||
undot = _undot
|
||||
|
||||
|
||||
def sanitize_fn(fn: str) -> str:
|
||||
fn = fn.replace("\x00", "_")
|
||||
def sanitize_fn(fn: str, fsnt: Optional[str] = None) -> str:
|
||||
fn = fn.replace("\x00", "_").split("/")[-1]
|
||||
if fsnt and fsnt != "lin":
|
||||
fn = fn.translate(VPTL_WIN if fsnt == "win" or ANYWIN else VPTL_MAC)
|
||||
fn2 = fn.replace("\\", "/")
|
||||
if ANYWIN or "/../" in fn2:
|
||||
fn = fn2.rstrip(". ")
|
||||
fn = fn.split("/")[-1]
|
||||
if APTL_OS:
|
||||
fn = sanitize_to(fn, APTL_OS)
|
||||
if ANYWIN or fsnt == "win" or "/../" in fn2:
|
||||
fn = fn2.split("/")[-1].rstrip(". ")
|
||||
if VPTL_OS:
|
||||
fn = sanitize_to(fn, VPTL_OS)
|
||||
return fn.strip()
|
||||
|
||||
|
||||
def sanitize_to(fn: str, tl: dict[int, int]) -> str:
|
||||
def sanitize_to(fn: str, tl: dict[int, int], fsnt: Optional[str] = None) -> str:
|
||||
fn = fn.translate(tl)
|
||||
if ANYWIN and fn.lower().split(".")[0].strip() in BADNAMESET:
|
||||
fn = "_" + fn
|
||||
if ANYWIN or fsnt == "win":
|
||||
fn = fn.rstrip(". ")
|
||||
if fn.lower().split(".")[0].strip() in BADNAMESET:
|
||||
fn = "_" + fn
|
||||
return fn
|
||||
|
||||
|
||||
def sanitize_vpath(vp: str) -> str:
|
||||
if not APTL_OS:
|
||||
def sanitize_vpath(vp: str, fsnt: Optional[str] = None) -> str:
|
||||
vptl = VPTL_OS
|
||||
if fsnt and fsnt != "lin":
|
||||
vptl = VPTL_WIN if fsnt == "win" or ANYWIN else VPTL_MAC
|
||||
if not vptl:
|
||||
return vp
|
||||
parts = vp.replace(os.sep, "/").split("/")
|
||||
ret = [sanitize_to(x, APTL_OS) for x in parts]
|
||||
ret = [sanitize_to(x, vptl, fsnt) for x in parts]
|
||||
return "/".join(ret)
|
||||
|
||||
|
||||
|
||||
Reference in new issue
Block a user