mirror of
https://github.com/9001/copyparty.git
synced 2025-08-17 09:02:15 -06:00
vfs ls
This commit is contained in:
ed
parent
250d0bdf57
commit
df276b6a84
|
|
@ -2,7 +2,7 @@
|
|||
# coding: utf-8
|
||||
from __future__ import print_function
|
||||
|
||||
import pprint
|
||||
import os
|
||||
import threading
|
||||
|
||||
from .__init__ import *
|
||||
|
|
@ -30,14 +30,14 @@ class VFS(object):
|
|||
# exists; do not manipulate permissions
|
||||
return self.nodes[name].add(src, dst)
|
||||
|
||||
n = VFS(
|
||||
vn = VFS(
|
||||
"{}/{}".format(self.realpath, name),
|
||||
"{}/{}".format(self.vpath, name).lstrip("/"),
|
||||
self.uread,
|
||||
self.uwrite,
|
||||
)
|
||||
self.nodes[name] = n
|
||||
return n.add(src, dst)
|
||||
self.nodes[name] = vn
|
||||
return vn.add(src, dst)
|
||||
|
||||
if dst in self.nodes:
|
||||
# leaf exists; return as-is
|
||||
|
|
@ -45,9 +45,76 @@ class VFS(object):
|
|||
|
||||
# leaf does not exist; create and keep permissions blank
|
||||
vp = "{}/{}".format(self.vpath, dst).lstrip("/")
|
||||
n = VFS(src, vp)
|
||||
self.nodes[dst] = n
|
||||
return n
|
||||
vn = VFS(src, vp)
|
||||
self.nodes[dst] = vn
|
||||
return vn
|
||||
|
||||
def undot(self, path):
|
||||
ret = []
|
||||
for node in path.split("/"):
|
||||
if node in ["", "."]:
|
||||
continue
|
||||
|
||||
if node == "..":
|
||||
if ret:
|
||||
ret.pop()
|
||||
continue
|
||||
|
||||
ret.append(node)
|
||||
|
||||
return "/".join(ret)
|
||||
|
||||
def _find(self, vpath):
|
||||
"""return [vfs,remainder]"""
|
||||
vpath = self.undot(vpath)
|
||||
if vpath == "":
|
||||
return [self, ""]
|
||||
|
||||
if "/" in vpath:
|
||||
name, rem = vpath.split("/", 1)
|
||||
else:
|
||||
name = vpath
|
||||
rem = ""
|
||||
|
||||
if name in self.nodes:
|
||||
return self.nodes[name]._find(rem)
|
||||
|
||||
return [self, vpath]
|
||||
|
||||
def ls(self, vpath, user):
|
||||
"""return user-readable [virt,real] items at vpath"""
|
||||
vn, rem = self._find(vpath)
|
||||
|
||||
if user not in vn.uread:
|
||||
return [[], []]
|
||||
|
||||
rp = vn.realpath
|
||||
if rem:
|
||||
rp += "/" + rem
|
||||
|
||||
real = os.listdir(rp)
|
||||
real.sort()
|
||||
if rem:
|
||||
virt_vis = []
|
||||
else:
|
||||
virt_all = [] # all nodes that exist
|
||||
virt_vis = [] # nodes readable by user
|
||||
for name, vn2 in sorted(vn.nodes.items()):
|
||||
virt_all.append(name)
|
||||
if user in vn2.uread:
|
||||
virt_vis.append(name)
|
||||
|
||||
for name in virt_all:
|
||||
try:
|
||||
real.remove(name)
|
||||
except:
|
||||
pass
|
||||
|
||||
absreal = []
|
||||
for p in real:
|
||||
absreal.append("{}/{}".format(rp, p).replace("//", "/"))
|
||||
|
||||
return [absreal, virt_vis]
|
||||
|
||||
|
||||
class AuthSrv(object):
|
||||
|
|
@ -136,4 +203,5 @@ class AuthSrv(object):
|
|||
self.user = user
|
||||
self.iuser = self.invert(user)
|
||||
|
||||
# import pprint
|
||||
# pprint.pprint({"usr": user, "rd": mread, "wr": mwrite, "mnt": mount})
|
||||
|
|
|
|||
|
|
@ -33,3 +33,10 @@ para() { for s in 1 2 3 4 5 6 7 8 12 16 24 32 48 64; do echo $s; for r in {1..4}
|
|||
## usage: avg logfile
|
||||
|
||||
avg() { awk 'function pr(ncsz) {if (nsmp>0) {printf "%3s %s\n", csz, sum/nsmp} csz=$1;sum=0;nsmp=0} {sub(/\r$/,"")} /^[0-9]+$/ {pr($1);next} / MiB/ {sub(/ MiB.*/,"");sub(/.* /,"");sum+=$1;nsmp++} END {pr(0)}' "$1"; }
|
||||
|
||||
|
||||
##
|
||||
## vscode
|
||||
|
||||
# replace variable name
|
||||
# (^|[^\w])oldname([^\w]|$) => $1newname$2
|
||||
|
|
|
|||
|
|
@ -15,6 +15,21 @@ class TestVFS(unittest.TestCase):
|
|||
def dump(self, vfs):
|
||||
print(json.dumps(vfs, indent=4, sort_keys=True, default=lambda o: o.__dict__))
|
||||
|
||||
def unfoo(self, foo):
|
||||
for k, v in {"foo": "a", "bar": "b", "baz": "c", "qux": "d"}.items():
|
||||
foo = foo.replace(k, v)
|
||||
|
||||
return foo
|
||||
|
||||
def undot(self, vfs, query, response):
|
||||
self.assertEqual(vfs.undot(query), response)
|
||||
query = self.unfoo(query)
|
||||
response = self.unfoo(response)
|
||||
self.assertEqual(vfs.undot(query), response)
|
||||
|
||||
def absify(self, root, names):
|
||||
return ["{}/{}".format(root, x).replace("//", "/") for x in names]
|
||||
|
||||
def test(self):
|
||||
td = "/dev/shm/vfs"
|
||||
try:
|
||||
|
|
@ -59,33 +74,63 @@ class TestVFS(unittest.TestCase):
|
|||
self.assertEqual(vfs.uread, ["*"])
|
||||
self.assertEqual(vfs.uwrite, [])
|
||||
|
||||
# read-only rootfs with write-only subdirectory
|
||||
# read-only rootfs with write-only subdirectory (read-write for k)
|
||||
vfs = AuthSrv(
|
||||
Namespace(c=None, a=[], v=[".::r", "a/ac/acb:a/ac/acb:w"]), None
|
||||
Namespace(c=None, a=["k:k"], v=[".::r:ak", "a/ac/acb:a/ac/acb:w:ak"]), None
|
||||
).vfs
|
||||
self.assertEqual(len(vfs.nodes), 1)
|
||||
self.assertEqual(vfs.vpath, "")
|
||||
self.assertEqual(vfs.realpath, td)
|
||||
self.assertEqual(vfs.uread, ["*"])
|
||||
self.assertEqual(vfs.uwrite, [])
|
||||
self.assertEqual(vfs.uread, ["*", "k"])
|
||||
self.assertEqual(vfs.uwrite, ["k"])
|
||||
n = vfs.nodes["a"]
|
||||
self.assertEqual(len(vfs.nodes), 1)
|
||||
self.assertEqual(n.vpath, "a")
|
||||
self.assertEqual(n.realpath, td + "/a")
|
||||
self.assertEqual(n.uread, ["*"])
|
||||
self.assertEqual(n.uwrite, [])
|
||||
self.assertEqual(n.uread, ["*", "k"])
|
||||
self.assertEqual(n.uwrite, ["k"])
|
||||
n = n.nodes["ac"]
|
||||
self.assertEqual(len(vfs.nodes), 1)
|
||||
self.assertEqual(n.vpath, "a/ac")
|
||||
self.assertEqual(n.realpath, td + "/a/ac")
|
||||
self.assertEqual(n.uread, ["*"])
|
||||
self.assertEqual(n.uwrite, [])
|
||||
self.assertEqual(n.uread, ["*", "k"])
|
||||
self.assertEqual(n.uwrite, ["k"])
|
||||
n = n.nodes["acb"]
|
||||
self.assertEqual(n.nodes, {})
|
||||
self.assertEqual(n.vpath, "a/ac/acb")
|
||||
self.assertEqual(n.realpath, td + "/a/ac/acb")
|
||||
self.assertEqual(n.uread, [])
|
||||
self.assertEqual(n.uwrite, ["*"])
|
||||
self.assertEqual(n.uread, ["k"])
|
||||
self.assertEqual(n.uwrite, ["*", "k"])
|
||||
|
||||
real, virt = vfs.ls("/", "*")
|
||||
self.assertEqual(real, self.absify(td, ["b", "c"]))
|
||||
self.assertEqual(virt, ["a"])
|
||||
|
||||
real, virt = vfs.ls("a", "*")
|
||||
self.assertEqual(real, self.absify(td, ["a/aa", "a/ab"]))
|
||||
self.assertEqual(virt, ["ac"])
|
||||
|
||||
real, virt = vfs.ls("a/ab", "*")
|
||||
self.assertEqual(real, self.absify(td, ["a/ab/aba", "a/ab/abb", "a/ab/abc"]))
|
||||
self.assertEqual(virt, [])
|
||||
|
||||
real, virt = vfs.ls("a/ac", "*")
|
||||
self.assertEqual(real, self.absify(td, ["a/ac/aca", "a/ac/acc"]))
|
||||
self.assertEqual(virt, [])
|
||||
|
||||
real, virt = vfs.ls("a/ac", "k")
|
||||
self.assertEqual(real, self.absify(td, ["a/ac/aca", "a/ac/acc"]))
|
||||
self.assertEqual(virt, ["acb"])
|
||||
|
||||
real, virt = vfs.ls("a/ac/acb", "*")
|
||||
self.assertEqual(real, [])
|
||||
self.assertEqual(virt, [])
|
||||
|
||||
real, virt = vfs.ls("a/ac/acb", "k")
|
||||
self.assertEqual(
|
||||
real, self.absify(td, ["a/ac/acb/acba", "a/ac/acb/acbb", "a/ac/acb/acbc"])
|
||||
)
|
||||
self.assertEqual(virt, [])
|
||||
|
||||
# breadth-first construction
|
||||
vfs = AuthSrv(
|
||||
|
|
@ -103,10 +148,29 @@ class TestVFS(unittest.TestCase):
|
|||
None,
|
||||
).vfs
|
||||
|
||||
# sanitizing relative paths
|
||||
self.undot(vfs, "foo/bar/../baz/qux", "foo/baz/qux")
|
||||
self.undot(vfs, "foo/../bar", "bar")
|
||||
self.undot(vfs, "foo/../../bar", "bar")
|
||||
self.undot(vfs, "foo/../../", "")
|
||||
self.undot(vfs, "./.././foo/", "foo")
|
||||
self.undot(vfs, "./.././foo/..", "")
|
||||
|
||||
# shadowing
|
||||
# crossreferences
|
||||
# loops
|
||||
# listdir mapping
|
||||
# access reduction
|
||||
vfs = AuthSrv(Namespace(c=None, a=[], v=[".::r", "b:a/ac:r"]), None).vfs
|
||||
|
||||
r1, v1 = vfs.ls("", "*")
|
||||
self.assertEqual(r1, self.absify(td, ["b", "c"]))
|
||||
self.assertEqual(v1, ["a"])
|
||||
|
||||
r1, v1 = vfs.ls("a", "*")
|
||||
self.assertEqual(r1, self.absify(td, ["a/aa", "a/ab"]))
|
||||
self.assertEqual(v1, ["ac"])
|
||||
|
||||
r1, v1 = vfs.ls("a/ac", "*")
|
||||
r2, v2 = vfs.ls("b", "*")
|
||||
self.assertEqual(r1, self.absify(td, ["b/ba", "b/bb", "b/bc"]))
|
||||
self.assertEqual(r1, r2)
|
||||
self.assertEqual(v1, v2)
|
||||
|
||||
shutil.rmtree(td)
|
||||
|
|
|
|||
Loading…
Reference in a new issue