mirrors/copyparty
1
0
Fork 0
mirror of https://github.com/9001/copyparty.git synced 2025-08-17 09:02:15 -06:00
Code Issues Actions Packages Projects Releases Wiki Activity

file access

Browse source
This commit is contained in:
ed 2019-06-10 01:23:13 +00:00
parent 2414766678
commit 36fc87aa6f
4 changed files with 50 additions and 32 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

36
copyparty/authsrv.py
View file

@ -6,7 +6,7 @@ import os
import threading import threading
from .__init__ import PY2 from .__init__ import PY2
from .util import undot from .util import undot, Pebkac
class VFS(object): class VFS(object):
@ -67,32 +67,44 @@ class VFS(object):
return [self, vpath] return [self, vpath]
def can_access(self, vpath, user): def can_access(self, vpath, uname):
"""return [readable,writable]""" """return [readable,writable]"""
vn, _ = self._find(vpath) vn, _ = self._find(vpath)
return [user in vn.uread, user in vn.uwrite] return [uname in vn.uread, uname in vn.uwrite]
def ls(self, vpath, user): def get(self, vpath, uname, will_read, will_write):
"""return user-readable [fsdir,real,virt] items at vpath""" """returns [vfsnode,fs_remainder] if user has the requested permissions"""
vn, rem = self._find(vpath) vn, rem = self._find(vpath)
if user not in vn.uread: if will_read and uname not in vn.uread:
return [vn.realpath, [], []] raise Pebkac("you don't have read-access for this location")
rp = vn.realpath if will_write and uname not in vn.uwrite:
raise Pebkac("you don't have write-access for this location")
return vn, rem
def canonical(self, rem):
"""returns the canonical path (fully-resolved absolute fs path)"""
rp = self.realpath
if rem: if rem:
rp += "/" + rem rp += "/" + rem
real = os.listdir(rp) return os.path.realpath(rp)
def ls(self, rem, uname):
"""return user-readable [fsdir,real,virt] items at vpath"""
abspath = self.canonical(rem)
real = os.listdir(abspath)
real.sort() real.sort()
if rem: if rem:
virt_vis = [] virt_vis = []
else: else:
virt_all = [] # all nodes that exist virt_all = [] # all nodes that exist
virt_vis = [] # nodes readable by user virt_vis = [] # nodes readable by user
for name, vn2 in sorted(vn.nodes.items()): for name, vn2 in sorted(self.nodes.items()):
virt_all.append(name) virt_all.append(name)
if user in vn2.uread: if uname in vn2.uread:
virt_vis.append(name) virt_vis.append(name)
for name in virt_all: for name in virt_all:
@ -101,7 +113,7 @@ class VFS(object):
except ValueError: except ValueError:
pass pass
return [rp, real, virt_vis] return [abspath, real, virt_vis]
def user_tree(self, uname, readable=False, writable=False): def user_tree(self, uname, readable=False, writable=False):
ret = [] ret = []

9
copyparty/httpcli.py
View file

@ -98,7 +98,7 @@ class HttpCli(object):
args[k.lower()] = True args[k.lower()] = True
self.args = args self.args = args
self.vpath = vpath self.vpath = unquote_plus(vpath)
try: try:
if mode == "GET": if mode == "GET":
@ -305,7 +305,12 @@ class HttpCli(object):
vpath += u"/" + node vpath += u"/" + node
vpnodes.append([cgi.escape(vpath) + "/", cgi.escape(node)]) vpnodes.append([cgi.escape(vpath) + "/", cgi.escape(node)])
fsroot, vfs_ls, vfs_virt = self.auth.vfs.ls(self.vpath, self.uname) vn, rem = self.auth.vfs.get(self.vpath, self.uname, True, False)
abspath = vn.canonical(rem)
if not os.path.isdir(abspath):
return self.tx_file(abspath)
fsroot, vfs_ls, vfs_virt = vn.ls(rem, self.uname)
vfs_ls.extend(vfs_virt) vfs_ls.extend(vfs_virt)
dirs = [] dirs = []

7
copyparty/web/browser.html
View file

@ -11,10 +11,10 @@
<body> <body>
<h1 id="path"> <h1 id="path">
{% for n in vpnodes[:-1] %} {%- for n in vpnodes[:-1] %}
<a href="{{ n[0] }}">{{ n[1] }}</a> <a href="{{ n[0] }}">{{ n[1] }}</a>
{%- endfor %} {%- endfor %}
<span>{{ vpnodes[-1][1] }}</li> <span>{{ vpnodes[-1][1] }}</span>
</h1> </h1>
<table id="files"> <table id="files">
<thead> <thead>
@ -27,13 +27,12 @@
</thead> </thead>
<tbody> <tbody>
{% for f in files %} {%- for f in files %}
<tr><td>{{ f[0] }}</td><td><a href="{{ f[1] }}">{{ f[2] }}</a></td><td>{{ f[3] }}</td><td>{{ f[4] }}</td></tr> <tr><td>{{ f[0] }}</td><td><a href="{{ f[1] }}">{{ f[2] }}</a></td><td>{{ f[3] }}</td><td>{{ f[4] }}</td></tr>
{%- endfor %} {%- endfor %}
</tbody> </tbody>
</table> </table>
</div>
<div id="widget"> <div id="widget">
<div id="wtoggle">=</div> <div id="wtoggle">=</div>
<div id="widgeti"> <div id="widgeti">

30
tests/test_vfs.py
View file

@ -33,6 +33,11 @@ class TestVFS(unittest.TestCase):
def absify(self, root, names): def absify(self, root, names):
return ["{}/{}".format(root, x).replace("//", "/") for x in names] return ["{}/{}".format(root, x).replace("//", "/") for x in names]
def ls(self, vfs, vpath, uname):
"""helper for resolving and listing a folder"""
vn, rem = vfs.get(vpath, uname, True, False)
return vn.ls(rem, uname)
def runcmd(self, *argv): def runcmd(self, *argv):
p = sp.Popen(argv, stdout=sp.PIPE, stderr=sp.PIPE) p = sp.Popen(argv, stdout=sp.PIPE, stderr=sp.PIPE)
stdout, stderr = p.communicate() stdout, stderr = p.communicate()
@ -131,37 +136,34 @@ class TestVFS(unittest.TestCase):
self.assertEqual(n.uread, ["k"]) self.assertEqual(n.uread, ["k"])
self.assertEqual(n.uwrite, ["*", "k"]) self.assertEqual(n.uwrite, ["*", "k"])
fsdir, real, virt = vfs.ls("/", "*") fsdir, real, virt = self.ls(vfs, "/", "*")
self.assertEqual(fsdir, td) self.assertEqual(fsdir, td)
self.assertEqual(real, ["b", "c"]) self.assertEqual(real, ["b", "c"])
self.assertEqual(virt, ["a"]) self.assertEqual(virt, ["a"])
fsdir, real, virt = vfs.ls("a", "*") fsdir, real, virt = self.ls(vfs, "a", "*")
self.assertEqual(fsdir, td + "/a") self.assertEqual(fsdir, td + "/a")
self.assertEqual(real, ["aa", "ab"]) self.assertEqual(real, ["aa", "ab"])
self.assertEqual(virt, ["ac"]) self.assertEqual(virt, ["ac"])
fsdir, real, virt = vfs.ls("a/ab", "*") fsdir, real, virt = self.ls(vfs, "a/ab", "*")
self.assertEqual(fsdir, td + "/a/ab") self.assertEqual(fsdir, td + "/a/ab")
self.assertEqual(real, ["aba", "abb", "abc"]) self.assertEqual(real, ["aba", "abb", "abc"])
self.assertEqual(virt, []) self.assertEqual(virt, [])
fsdir, real, virt = vfs.ls("a/ac", "*") fsdir, real, virt = self.ls(vfs, "a/ac", "*")
self.assertEqual(fsdir, td + "/a/ac") self.assertEqual(fsdir, td + "/a/ac")
self.assertEqual(real, ["aca", "acc"]) self.assertEqual(real, ["aca", "acc"])
self.assertEqual(virt, []) self.assertEqual(virt, [])
fsdir, real, virt = vfs.ls("a/ac", "k") fsdir, real, virt = self.ls(vfs, "a/ac", "k")
self.assertEqual(fsdir, td + "/a/ac") self.assertEqual(fsdir, td + "/a/ac")
self.assertEqual(real, ["aca", "acc"]) self.assertEqual(real, ["aca", "acc"])
self.assertEqual(virt, ["acb"]) self.assertEqual(virt, ["acb"])
fsdir, real, virt = vfs.ls("a/ac/acb", "*") self.assertRaises(util.Pebkac, vfs.get, "a/ac/acb", "*", True, False)
self.assertEqual(fsdir, td + "/a/ac/acb")
self.assertEqual(real, [])
self.assertEqual(virt, [])
fsdir, real, virt = vfs.ls("a/ac/acb", "k") fsdir, real, virt = self.ls(vfs, "a/ac/acb", "k")
self.assertEqual(fsdir, td + "/a/ac/acb") self.assertEqual(fsdir, td + "/a/ac/acb")
self.assertEqual(real, ["acba", "acbb", "acbc"]) self.assertEqual(real, ["acba", "acbb", "acbc"])
self.assertEqual(virt, []) self.assertEqual(virt, [])
@ -193,18 +195,18 @@ class TestVFS(unittest.TestCase):
# shadowing # shadowing
vfs = AuthSrv(Namespace(c=None, a=[], v=[".::r", "b:a/ac:r"]), None).vfs vfs = AuthSrv(Namespace(c=None, a=[], v=[".::r", "b:a/ac:r"]), None).vfs
fsp, r1, v1 = vfs.ls("", "*") fsp, r1, v1 = self.ls(vfs, "", "*")
self.assertEqual(fsp, td) self.assertEqual(fsp, td)
self.assertEqual(r1, ["b", "c"]) self.assertEqual(r1, ["b", "c"])
self.assertEqual(v1, ["a"]) self.assertEqual(v1, ["a"])
fsp, r1, v1 = vfs.ls("a", "*") fsp, r1, v1 = self.ls(vfs, "a", "*")
self.assertEqual(fsp, td + "/a") self.assertEqual(fsp, td + "/a")
self.assertEqual(r1, ["aa", "ab"]) self.assertEqual(r1, ["aa", "ab"])
self.assertEqual(v1, ["ac"]) self.assertEqual(v1, ["ac"])
fsp1, r1, v1 = vfs.ls("a/ac", "*") fsp1, r1, v1 = self.ls(vfs, "a/ac", "*")
fsp2, r2, v2 = vfs.ls("b", "*") fsp2, r2, v2 = self.ls(vfs, "b", "*")
self.assertEqual(fsp1, td + "/b") self.assertEqual(fsp1, td + "/b")
self.assertEqual(fsp2, td + "/b") self.assertEqual(fsp2, td + "/b")
self.assertEqual(r1, ["ba", "bb", "bc"]) self.assertEqual(r1, ["ba", "bb", "bc"])