" + html_escape(msg).encode("utf-8", "replace")
h = {"WWW-Authenticate": 'Basic realm="a"'} if pex.code == 401 else {}
self.reply(zb, status=pex.code, headers=h, volsan=True)
if pex.log:
self.log("additional error context:\n" + pex.log, 6)
return self.keepalive
except Pebkac:
return False
finally:
if self.dl_id:
self.conn.hsrv.dli.pop(self.dl_id, None)
self.conn.hsrv.dls.pop(self.dl_id, None)
def dip(self) :
if self.args.plain_ip:
return self.ip.replace(":", ".")
else:
return self.conn.iphash.s(self.ip)
def cbonk(self, g , v , reason , descr ) :
cond = self.args.dont_ban
if (
cond == "any"
or (cond == "auth" and self.uname != "*")
or (cond == "aa" and self.avol)
or (cond == "av" and self.can_admin)
or (cond == "rw" and self.can_read and self.can_write)
):
return False
self.conn.hsrv.nsus += 1
if not g.lim:
return False
bonk, ip = g.bonk(self.ip, v + self.gctx)
if not bonk:
return False
xban = self.vn.flags.get("xban")
if xban:
hr = runhook(
self.log,
self.conn.hsrv.broker,
None,
"xban",
xban,
self.vn.canonical(self.rem),
self.vpath,
self.host,
self.uname,
"",
time.time(),
0,
self.ip,
time.time(),
[reason, reason],
)
if hr.get("rv") == 0:
return False
self.log("client banned: %s" % (descr,), 1)
self.conn.hsrv.bans[ip] = bonk
self.conn.hsrv.nban += 1
return True
def is_banned(self) :
if not self.conn.bans:
return False
bans = self.conn.bans
ip = ipnorm(self.ip)
if ip not in bans:
return False
rt = bans[ip] - time.time()
if rt < 0:
del bans[ip]
self.log("client unbanned", 3)
return False
self.log("banned for {:.0f} sec".format(rt), 6)
self.terse_reply(self.args.banmsg_b, 403)
return True
def permit_caching(self) :
cache = self.uparam.get("cache")
if cache is None:
self.out_headers["Cache-Control"] = self.vn.flags["cachectl"]
return
n = 69 if not cache else 604869 if cache == "i" else int(cache)
self.out_headers["Cache-Control"] = "max-age=" + str(n)
def k304(self) :
k304 = self.cookies.get("k304")
return k304 == "y" or (self.args.k304 == 2 and k304 != "n")
def no304(self) :
no304 = self.cookies.get("no304")
return no304 == "y" or (self.args.no304 == 2 and no304 != "n")
def _build_html_head(self, kv ) :
html = str(self.vn.flags["html_head_d"])
is_jinja = html[:2] in "%@%"
if is_jinja:
html = html.replace("%", "", 1)
if html.startswith("@"):
html = read_utf8(self.log, html[1:], True)
if html.startswith("%"):
html = html[1:]
is_jinja = True
if is_jinja:
with self.conn.hsrv.mutex:
if html not in self.conn.hsrv.j2:
j2env = jinja2.Environment()
tpl = j2env.from_string(html)
self.conn.hsrv.j2[html] = tpl
html = self.conn.hsrv.j2[html].render(**kv)
self.html_head += html + "\n"
def send_headers(
self,
length ,
status = 200,
mime = None,
headers = None,
) :
response = ["%s %s %s" % (self.http_ver, status, HTTPCODE[status])]
if headers:
self.out_headers.update(headers)
if status == 304:
self.out_headers.pop("Content-Length", None)
self.out_headers.pop("Content-Type", None)
self.out_headerlist[:] = []
if self.k304():
self.keepalive = False
else:
if length is not None:
response.append("Content-Length: " + unicode(length))
if mime:
self.out_headers["Content-Type"] = mime
elif "Content-Type" not in self.out_headers:
self.out_headers["Content-Type"] = "text/html; charset=utf-8"
response.append(H_CONN_KEEPALIVE if self.keepalive else H_CONN_CLOSE)
response.append("Date: " + formatdate())
for k, zs in list(self.out_headers.items()) + self.out_headerlist:
response.append("%s: %s" % (k, zs))
ptn_cc = RE_CC
for zs in response:
m = ptn_cc.search(zs)
if m:
t = "malicious user; Cc in out-hdr; req(%r) hdr(%r) => %r"
self.log(t % (self.req, zs, zs[m.span()[0] :]), 1)
self.cbonk(self.conn.hsrv.gmal, zs, "cc_hdr", "Cc in out-hdr")
raise Pebkac(999)
if self.args.ohead and self.do_log:
keys = self.args.ohead
if "*" in keys:
lines = response[1:]
else:
lines = []
for zs in response[1:]:
if zs.split(":")[0].lower() in keys:
lines.append(zs)
for zs in lines:
hk, hv = zs.split(": ")
self.log("[O] {}: \033[33m[{}]".format(hk, hv), 5)
response.append("\r\n")
try:
self.s.sendall("\r\n".join(response).encode("utf-8"))
except:
raise Pebkac(400, "client d/c while replying headers")
def reply(
self,
body ,
status = 200,
mime = None,
headers = None,
volsan = False,
) :
if (
status > 400
and status in (403, 404, 422)
and (
status != 422
or (
not body.startswith(b"partial upload exists")
and not body.startswith(b"source file busy")
)
)
and (status != 404 or (self.can_get and not self.can_read))
):
if status == 404:
g = self.conn.hsrv.g404
elif status == 403:
g = self.conn.hsrv.g403
else:
g = self.conn.hsrv.g422
gurl = self.conn.hsrv.gurl
if (
gurl.lim
and (not g.lim or gurl.lim < g.lim)
and self.args.sus_urls.search(self.vpath)
):
g = self.conn.hsrv.gurl
if g.lim and (
g == self.conn.hsrv.g422
or not self.args.nonsus_urls
or not self.args.nonsus_urls.search(self.vpath)
):
self.cbonk(g, self.vpath, str(status), "%ss" % (status,))
if volsan:
vols = list(self.asrv.vfs.all_vols.values())
body = vol_san(vols, body)
try:
zs = absreal(__file__).rsplit(os.path.sep, 2)[0]
body = body.replace(zs.encode("utf-8"), b"PP")
except:
pass
self.send_headers(len(body), status, mime, headers)
try:
if self.mode != "HEAD":
self.s.sendall(body)
except:
raise Pebkac(400, "client d/c while replying body")
return body
def loud_reply(self, body , *args , **kwargs ) :
if not kwargs.get("mime"):
kwargs["mime"] = "text/plain; charset=utf-8"
self.log(body.rstrip())
self.reply(body.encode("utf-8") + b"\r\n", *list(args), **kwargs)
def terse_reply(self, body , status = 200) :
self.keepalive = False
lines = [
"%s %s %s" % (self.http_ver or "HTTP/1.1", status, HTTPCODE[status]),
H_CONN_CLOSE,
]
if body:
lines.append(
"Content-Type: text/html; charset=utf-8\r\nContent-Length: "
+ unicode(len(body))
)
lines.append("\r\n")
self.s.sendall("\r\n".join(lines).encode("utf-8") + body)
def urlq(self, add , rm ) :
"a"
if self.is_rclone:
return ""
kv = {k: zs for k, zs in self.uparam.items() if k not in rm}
if "pw" in kv:
pw = self.cookies.get("cppws") or self.cookies.get("cppwd")
if kv["pw"] == pw:
del kv["pw"]
kv.update(add)
if not kv:
return ""
r = ["%s=%s" % (quotep(k), quotep(zs)) if zs else k for k, zs in kv.items()]
return "?" + "&".join(r)
def ourlq(self) :
skip = ("pw", "h", "k")
ret = []
for k, v in self.ouparam.items():
if k in skip:
continue
t = "%s=%s" % (quotep(k), quotep(v))
ret.append(t.replace(" ", "+").rstrip("="))
if not ret:
return ""
return "?" + "&".join(ret)
def redirect(
self,
vpath ,
suf = "",
msg = "aight",
flavor = "go to",
click = True,
status = 200,
use302 = False,
) :
vp = self.args.SRS + vpath
html = self.j2s(
"msg",
h2='{} {}'.format(
quotep(vp) + suf, flavor, html_escape(vp, crlf=True) + suf
),
pre=msg,
click=click,
).encode("utf-8", "replace")
if use302:
self.reply(html, status=302, headers={"Location": vp})
else:
self.reply(html, status=status)
return True
def _cors(self) :
ih = self.headers
origin = ih.get("origin")
if not origin:
sfsite = ih.get("sec-fetch-site")
if sfsite and sfsite.lower().startswith("cross"):
origin = ":|"
else:
return True
host = self.host.lower()
if host.startswith("["):
if "]:" in host:
host = host.split("]:")[0] + "]"
else:
host = host.split(":")[0]
oh = self.out_headers
origin = origin.lower()
proto = "https" if self.is_https else "http"
good_origins = self.args.acao + ["%s://%s" % (proto, host)]
if (
self.args.pw_hdr in ih
or re.sub(r"(:[0-9]{1,5})?/?$", "", origin) in good_origins
):
good_origin = True
bad_hdrs = ("",)
else:
good_origin = False
bad_hdrs = ("", self.args.pw_hdr)
acah = ih.get("access-control-request-headers", "")
acao = (origin if good_origin else None) or (
"*" if "*" in good_origins else None
)
if self.args.allow_csrf:
acao = origin or acao or "*"
acam = ", ".join(self.conn.hsrv.mallow)
oh["Access-Control-Allow-Credentials"] = "true"
good_origin = True
else:
acam = ", ".join(self.args.acam)
if "range" not in acah.lower():
acah += ",Range"
req_h = acah.split(",")
req_h = [x.strip() for x in req_h]
req_h = [x for x in req_h if x.lower() not in bad_hdrs]
acah = ", ".join(req_h)
if not acao:
return False
oh["Access-Control-Allow-Origin"] = acao
oh["Access-Control-Allow-Methods"] = acam.upper()
if acah:
oh["Access-Control-Allow-Headers"] = acah
return good_origin
def handle_get(self) :
if self.do_log:
logmsg = "%-4s %s @%s" % (self.mode, self.req, self.uname)
if "range" in self.headers:
try:
rval = self.headers["range"].split("=", 1)[1]
except:
rval = self.headers["range"]
logmsg += " [\033[36m" + rval + "\033[0m]"
self.log(logmsg)
if "%" in self.req:
self.log(" `-- %r" % (self.vpath,))
if self.vpath.startswith(".cpr"):
if self.vpath.startswith(".cpr/ico/"):
return self.tx_ico(self.vpath.split("/")[-1], exact=True)
if self.vpath.startswith(".cpr/ssdp"):
if self.conn.hsrv.ssdp:
return self.conn.hsrv.ssdp.reply(self)
else:
self.reply(b"ssdp is disabled in server config", 404)
return False
if self.vpath == ".cpr/metrics":
return self.conn.hsrv.metrics.tx(self)
res_path = "web/" + self.vpath[5:]
if res_path in RES:
ap = self.E.mod_ + res_path
if bos.path.exists(ap) or bos.path.exists(ap + ".gz"):
return self.tx_file(ap)
else:
return self.tx_res(res_path)
if res_path in RESM:
ap = self.E.mod_ + RESM[res_path]
if (
"txt" not in self.uparam
and "mime" not in self.uparam
and not self.ouparam.get("dl")
):
self.ouparam["dl"] = res_path.split("/")[-1]
if bos.path.exists(ap) or bos.path.exists(ap + ".gz"):
return self.tx_file(ap)
else:
return self.tx_res(res_path)
self.tx_404()
return False
if "cf_challenge" in self.uparam:
self.reply(self.j2s("cf").encode("utf-8", "replace"))
return True
if not self.can_read and not self.can_write and not self.can_get:
t = "@%s has no access to %r"
if self.vn.realpath and "on403" in self.vn.flags:
t += " (on403)"
self.log(t % (self.uname, "/" + self.vpath))
ret = self.on40x(self.vn.flags["on403"], self.vn, self.rem)
if ret == "true":
return True
elif ret == "false":
return False
elif ret == "home":
self.uparam["h"] = ""
elif ret == "allow":
self.log("plugin override; access permitted")
self.can_read = self.can_write = self.can_move = True
self.can_delete = self.can_get = self.can_upget = True
self.can_admin = True
else:
return self.tx_404(True)
else:
if (
self.asrv.badcfg1
and "h" not in self.ouparam
and "hc" not in self.ouparam
):
zs1 = "copyparty refused to start due to a failsafe: invalid server config; check server log"
zs2 = 'you may access the controlpanel but nothing will work until you shutdown the copyparty container and %s config-file (or provide the configuration as command-line arguments)'
if self.asrv.is_lxc and len(self.asrv.cfg_files_loaded) == 1:
zs2 = zs2 % ("add a",)
else:
zs2 = zs2 % ("fix the",)
html = self.j2s("msg", h1=zs1, h2=zs2)
self.reply(html.encode("utf-8", "replace"), 500)
return True
if "ls" in self.uparam:
return self.tx_ls_vols()
if self.vpath:
ptn = self.args.nonsus_urls
if not ptn or not ptn.search(self.vpath):
self.log(t % (self.uname, "/" + self.vpath))
return self.tx_404(True)
self.uparam["h"] = ""
if "smsg" in self.uparam:
return self.handle_smsg()
if "tree" in self.uparam:
return self.tx_tree()
if "scan" in self.uparam:
return self.scanvol()
if self.args.getmod:
if "delete" in self.uparam:
return self.handle_rm([])
if "move" in self.uparam:
return self.handle_mv()
if "copy" in self.uparam:
return self.handle_cp()
if not self.vpath and self.ouparam:
if "reload" in self.uparam:
return self.handle_reload()
if "stack" in self.uparam:
return self.tx_stack()
if "setck" in self.uparam:
return self.setck()
if "reset" in self.uparam:
return self.set_cfg_reset()
if "hc" in self.uparam:
return self.tx_svcs()
if "shares" in self.uparam:
return self.tx_shares()
if "dls" in self.uparam:
return self.tx_dls()
if "ru" in self.uparam:
return self.tx_rups()
if "idp" in self.uparam:
return self.tx_idp()
if "h" in self.uparam:
return self.tx_mounts()
if "ups" in self.uparam:
return self.tx_ups()
if "rss" in self.uparam:
return self.tx_rss()
return self.tx_browser()
def tx_rss(self) :
if self.do_log:
self.log("RSS %s @%s" % (self.req, self.uname))
if not self.can_read:
return self.tx_404(True)
vn = self.vn
if not vn.flags.get("rss"):
raise Pebkac(405, "RSS is disabled in server config")
rem = self.rem
idx = self.conn.get_u2idx()
if not idx or not hasattr(idx, "p_end"):
if not HAVE_SQLITE3:
raise Pebkac(500, "sqlite3 not found on server; rss is disabled")
raise Pebkac(500, "server busy, cannot generate rss; please retry in a bit")
uv = [rem]
if "recursive" in self.uparam:
uq = "up.rd like ?||'%'"
else:
uq = "up.rd == ?"
zs = str(self.uparam.get("fext", self.args.rss_fext))
if zs in ("True", "False"):
zs = ""
if zs:
zsl = []
for ext in zs.split(","):
zsl.append("+up.fn like '%.'||?")
uv.append(ext)
uq += " and ( %s )" % (" or ".join(zsl),)
zs1 = self.uparam.get("sort") or self.args.rss_sort
zs2 = zs1.lower()
zs = RSS_SORT.get(zs2)
if not zs:
raise Pebkac(400, "invalid sort key; must be m/u/n/s")
uq += " order by up." + zs
if zs1 == zs2:
uq += " desc"
nmax = int(self.uparam.get("nf") or self.args.rss_nf)
hits = idx.run_query(self.uname, [self.vn], uq, uv, False, False, nmax)[0]
q_pw = a_pw = ""
pwk = self.args.pw_urlp
if pwk in self.ouparam and "nopw" not in self.ouparam:
zs = self.ouparam[pwk]
q_pw = "?%s=%s" % (pwk, quotep(zs))
a_pw = "&%s=%s" % (pwk, quotep(zs))
for i in hits:
i["rp"] += a_pw if "?" in i["rp"] else q_pw
title = self.uparam.get("title") or self.vpath.split("/")[-1]
etitle = html_escape(title, True, True)
baseurl = "%s://%s/" % (
"https" if self.is_https else "http",
self.host,
)
feed = baseurl + self.req[1:]
if pwk in self.ouparam and self.ouparam.get("nopw") == "a":
feed = re.sub(r"&%s=[^&]*" % (pwk,), "", feed)
if self.is_vproxied:
baseurl += self.args.RS
efeed = html_escape(feed, True, True)
edirlink = efeed.split("?")[0] + q_pw
ret = [
"""\
\t
\t\t%s
\t\tcopyparty-2
"""
% (efeed, etitle, edirlink)
]
q = "select fn from cv where rd=? and dn=?"
crd, cdn = rem.rsplit("/", 1) if "/" in rem else ("", rem)
try:
cfn = idx.cur[self.vn.realpath].execute(q, (crd, cdn)).fetchone()[0]
bos.stat(os.path.join(vn.canonical(rem), cfn))
cv_url = "%s%s?th=jf%s" % (baseurl, vjoin(self.vpath, cfn), a_pw)
cv_url = html_escape(cv_url, True, True)
zs = """\
\t\t
\t\t\t%s
\t\t\t%s
\t\t\t%s
\t\t
"""
ret.append(zs % (cv_url, etitle, edirlink))
except:
pass
ap = ""
use_magic = "rmagic" in self.vn.flags
tpl_t = self.uparam.get("fmt_t") or self.vn.flags["rss_fmt_t"]
tpl_d = self.uparam.get("fmt_d") or self.vn.flags["rss_fmt_d"]
kw_t = [[x, x[1:-1]] for x in RE_RSS_KW.findall(tpl_t)]
kw_d = [[x, x[1:-1]] for x in RE_RSS_KW.findall(tpl_d)]
for i in hits:
if use_magic:
ap = os.path.join(self.vn.realpath, i["rp"])
tags = i["tags"]
iurl = html_escape("%s%s" % (baseurl, i["rp"]), True, True)
fname = tags["fname"] = unquotep(i["rp"].split("?")[0].split("/")[-1])
title = tpl_t
desc = tpl_d
for zs1, zs2 in kw_t:
title = title.replace(zs1, str(tags.get(zs2, "")))
for zs1, zs2 in kw_d:
desc = desc.replace(zs1, str(tags.get(zs2, "")))
title = html_escape(title.strip(), True, True)
if desc.strip(" -,"):
desc = html_escape(desc.strip(), True, True)
else:
desc = title
mime = html_escape(guess_mime(fname, ap))
lmod = formatdate(max(0, i["ts"]))
zsa = (iurl, iurl, title, desc, lmod, iurl, mime, i["sz"])
zs = (
"""\
\t\t-
\t\t\t
%s
\t\t\t%s
\t\t\t%s
\t\t\t%s
\t\t\t%s
\t\t\t%d \n" % (dur,)
ret.append(zs + "\t\t \n")
ret.append("\t \n \n")
bret = "".join(ret).encode("utf-8", "replace")
self.reply(bret, 200, "text/xml; charset=utf-8")
self.log("rss: %d hits, %d bytes" % (len(hits), len(bret)))
return True
def tx_zls(self, abspath) :
if self.do_log:
self.log("zls %s @%s" % (self.req, self.uname))
if self.args.no_zls:
raise Pebkac(405, "zip browsing is disabled in server config")
import zipfile
try:
with zipfile.ZipFile(abspath, "r") as zf:
filelist = [{"fn": f.filename} for f in zf.infolist()]
ret = json.dumps(filelist).encode("utf-8", "replace")
self.reply(ret, mime="application/json")
return True
except (zipfile.BadZipfile, RuntimeError):
raise Pebkac(404, "requested file is not a valid zip file")
def tx_zget(self, abspath) :
maxsz = 1024 * 1024 * 64
inner_path = self.uparam.get("zget")
if not inner_path:
raise Pebkac(405, "inner path is required")
if self.do_log:
self.log(
"zget %s \033[35m%s\033[0m @%s" % (self.req, inner_path, self.uname)
)
if self.args.no_zls:
raise Pebkac(405, "zip browsing is disabled in server config")
import zipfile
try:
with zipfile.ZipFile(abspath, "r") as zf:
zi = zf.getinfo(inner_path)
if zi.file_size >= maxsz:
raise Pebkac(404, "zip bomb defused")
with zf.open(zi, "r") as fi:
self.send_headers(length=zi.file_size, mime=guess_mime(inner_path))
sendfile_py(
self.log,
0,
zi.file_size,
fi,
self.s,
self.args.s_wr_sz,
self.args.s_wr_slp,
not self.args.no_poll,
{},
"",
)
except KeyError:
raise Pebkac(404, "no such file in archive")
except (zipfile.BadZipfile, RuntimeError):
raise Pebkac(404, "requested file is not a valid zip file")
return True
def handle_propfind(self) :
if self.do_log:
self.log("PFIND %s @%s" % (self.req, self.uname))
if "%" in self.req:
self.log(" `-- %r" % (self.vpath,))
if self.args.no_dav:
raise Pebkac(405, "WebDAV is disabled in server config")
vn = self.vn
rem = self.rem
tap = vn.canonical(rem)
if "davauth" in vn.flags and self.uname == "*":
raise Pebkac(401, "authenticate")
from .dxml import parse_xml
enc = "utf-8"
uenc = enc.upper()
props = DAV_ALLPROPS
clen = int(self.headers.get("content-length", 0))
if clen:
buf = b""
for rbuf in self.get_body_reader()[0]:
buf += rbuf
if not rbuf or len(buf) >= 32768:
break
sbuf = buf.decode(enc, "replace")
self.hint = " " + sbuf
xroot = parse_xml(sbuf)
xtag = next((x for x in xroot if x.tag.split("}")[-1] == "prop"), None)
if xtag is not None:
props = set([y.tag.split("}")[-1] for y in xtag])
self.hint = ""
zi = int(time.time())
vst = os.stat_result((16877, -1, -1, 1, 1000, 1000, 8, zi, zi, zi))
try:
st = bos.stat(tap)
except OSError as ex:
if ex.errno not in (errno.ENOENT, errno.ENOTDIR):
raise
if tap:
raise Pebkac(404)
st = vst
topdir = {"vp": "", "st": st}
fgen = []
depth = self.headers.get("depth", "infinity").lower()
if depth == "infinity":
if not self.can_read and (self.vpath or self.asrv.vfs.realpath):
t = "depth:infinity requires read-access in %r"
t = t % ("/" + self.vpath,)
self.log(t, 3)
raise Pebkac(401, t)
if not stat.S_ISDIR(topdir["st"].st_mode):
t = "depth:infinity can only be used on folders; %r is 0o%o"
t = t % ("/" + self.vpath, topdir["st"])
self.log(t, 3)
raise Pebkac(400, t)
if not self.args.dav_inf:
self.log("client wants --dav-inf", 3)
zb = b'\n'
ret = ret.format(uenc)
for x in fgen:
rp = vjoin(vtop, x["vp"])
st = x["st"]
mtime = max(0, st.st_mtime)
if stat.S_ISLNK(st.st_mode):
try:
st = bos.stat(os.path.join(tap, x["vp"]))
except:
continue
isdir = stat.S_ISDIR(st.st_mode)
ret += "/%s%s " % (
quotep(rp),
"/" if isdir and rp else "",
)
pvs = {
"displayname": html_escape(rp.split("/")[-1]),
"getlastmodified": formatdate(mtime),
"resourcetype": '%s " % (k, v, k)
else:
ret += " HTTP/1.1 200 OK {} HTTP/1.1 404 Not Found "
while ret:
ret = self.send_chunk(ret, enc, chunksz)
self.send_chunk("", enc, chunksz)
return True
def handle_proppatch(self) :
if self.do_log:
self.log("PPATCH %s @%s" % (self.req, self.uname))
if "%" in self.req:
self.log(" `-- %r" % (self.vpath,))
if self.args.no_dav:
raise Pebkac(405, "WebDAV is disabled in server config")
if not self.can_write:
self.log("%s tried to proppatch %r" % (self.uname, "/" + self.vpath))
raise Pebkac(401, "authenticate")
from xml.etree import ElementTree as ET
from .dxml import mkenod, mktnod, parse_xml
buf = b""
for rbuf in self.get_body_reader()[0]:
buf += rbuf
if not rbuf or len(buf) >= 128 * 1024:
break
if self._applesan():
return True
txt = buf.decode("ascii", "replace").lower()
enc = self.get_xml_enc(txt)
uenc = enc.upper()
txt = buf.decode(enc, "replace")
self.hint = " " + txt
ET.register_namespace("D", "DAV:")
xroot = mkenod("D:orz")
xroot.insert(0, parse_xml(txt))
xprop = xroot.find(r"./{DAV:}propertyupdate/{DAV:}set/{DAV:}prop")
for ze in xprop:
ze.clear()
self.hint = ""
txt = """HTTP/1.1 403 Forbidden " + txt
ET.register_namespace("D", "DAV:")
lk = parse_xml(txt)
assert lk.tag == "{DAV:}lockinfo"
self.hint = ""
token = str(uuid.uuid4())
if lk.find(r"./{DAV:}depth") is None:
depth = self.headers.get("depth", "infinity")
lk.append(mktnod("D:depth", depth))
lk.append(mktnod("D:timeout", "Second-3310"))
lk.append(mkenod("D:locktoken", mktnod("D:href", token)))
lk.append(
mkenod("D:lockroot", mktnod("D:href", quotep(self.args.SRS + self.vpath)))
)
lk2 = mkenod("D:activelock")
xroot = mkenod("D:prop", mkenod("D:lockdiscovery", lk2))
for a in lk:
lk2.append(a)
ret = '\n'.format(uenc)
ret += ET.tostring(xroot).decode("utf-8")
rc = 200
if self.can_write and not bos.path.isfile(abspath):
with open(fsenc(abspath), "wb") as _:
rc = 201
self.out_headers["Lock-Token"] = "<{}>".format(token)
self.reply(ret.encode(enc, "replace"), rc, "text/xml; charset=" + enc)
return True
def handle_unlock(self) :
if self.do_log:
self.log("UNLOCK %s @%s" % (self.req, self.uname))
if "%" in self.req:
self.log(" `-- %r" % (self.vpath,))
if self.args.no_dav:
raise Pebkac(405, "WebDAV is disabled in server config")
if not self.can_write and "Microsoft-WebDAV" not in self.ua:
self.log("%s tried to lock %r" % (self.uname, "/" + self.vpath))
raise Pebkac(401, "authenticate")
self.send_headers(None, 204)
return True
def handle_mkcol(self) :
if self._applesan():
return True
if self.do_log:
self.log("MKCOL %s @%s" % (self.req, self.uname))
if "%" in self.req:
self.log(" `-- %r" % (self.vpath,))
if self.args.no_dav:
raise Pebkac(405, "WebDAV is disabled in server config")
if not self.can_write:
raise Pebkac(401, "authenticate")
try:
return self._mkdir(self.vpath, True)
except Pebkac as ex:
if ex.code >= 500:
raise
self.reply(b"", ex.code)
return True
def handle_cpmv(self) :
dst = self.headers["destination"]
dst = re.sub("^[a-zA-Z]+://[^/]+", "", dst).lstrip()
if self.is_vproxied and dst.startswith(self.args.SRS):
dst = dst[len(self.args.RS) :]
if self.do_log:
self.log("%s %s --//> %s @%s" % (self.mode, self.req, dst, self.uname))
if "%" in self.req:
self.log(" `-- %r" % (self.vpath,))
if self.args.no_dav:
raise Pebkac(405, "WebDAV is disabled in server config")
dst = unquotep(dst)
zs = self.headers.get("overwrite", "").lower()
overwrite = zs not in ["f", "false"]
try:
fun = self._cp if self.mode == "COPY" else self._mv
return fun(self.vpath, dst.lstrip("/"), overwrite)
except Pebkac as ex:
if ex.code == 403:
ex.code = 401
raise
def _applesan(self) :
if self.args.dav_mac or "Darwin/" not in self.ua:
return False
vp = "/" + self.vpath
if re.search(APPLESAN_RE, vp):
zt = '\n{} 404 not found ┐( ´ -`)┌
or maybe you don\'t have access -- try a password or go home
'
pt = "404 not found ┐( ´ -`)┌ (or maybe you don't have access -- try a password)"
elif is_403:
t = '403 forbiddena ~┻━┻
use a password or go home
'
pt = "403 forbiddena ~┻━┻ (you'll have to log in)"
rc = 403
else:
t = '404 not found ┐( ´ -`)┌
'
pt = "404 not found ┐( ´ -`)┌"
if self.ua.startswith(("curl/", "fetch")):
pt = "# acct: %s\n%s\n" % (self.uname, pt)
self.reply(pt.encode("utf-8"), status=rc)
return True
if "th" in self.ouparam and str(self.ouparam["th"])[:1] in "jw":
return self.tx_svg("e" + pt[:3])
if rc == 403 and self.uname == "*":
sport = self.s.getsockname()[1]
if self.args.dav_port == sport or (
"sec-fetch-site" not in self.headers
and self.cookies.get("js") != "y"
and sport not in self.args.p_nodav
and (
not self.args.ua_nodav.search(self.ua)
or (self.args.dav_ua1 and self.args.dav_ua1.search(self.ua))
)
):
rc = 401
self.out_headers["WWW-Authenticate"] = 'Basic realm="a"'
t = t.format(self.args.SR)
qv = quotep(self.vpaths) + self.ourlq()
html = self.j2s(
"splash",
this=self,
qvpath=qv,
msg=t,
in_shr=self.args.shr and self.vpath.startswith(self.args.shr1),
ahttps="" if self.is_https else "https://" + self.host + self.req,
)
self.reply(html.encode("utf-8"), status=rc)
return True
def on40x(self, mods , vn , rem ) :
for mpath in mods:
try:
mod = loadpy(mpath, self.args.hot_handlers)
except Exception as ex:
self.log("import failed: {!r}".format(ex))
continue
ret = mod.main(self, vn, rem)
if ret:
return ret.lower()
return ""
def scanvol(self) :
if self.args.no_rescan:
raise Pebkac(403, "the rescan feature is disabled in server config")
vpaths = self.uparam["scan"].split(",/")
if vpaths == [""]:
vpaths = [self.vpath]
vols = []
for vpath in vpaths:
vn, _ = self.asrv.vfs.get(vpath, self.uname, True, True)
vols.append(vn.vpath)
if self.uname not in vn.axs.uadmin:
self.log("rejected scanning [%s] => [%s];" % (vpath, vn.vpath), 3)
raise Pebkac(403, "'scanvol' not allowed for user " + self.uname)
self.log("trying to rescan %d volumes: %r" % (len(vols), vols))
args = [self.asrv.vfs.all_vols, vols, False, True]
x = self.conn.hsrv.broker.ask("up2k.rescan", *args)
err = x.get()
if not err:
self.redirect("", "?h")
return True
raise Pebkac(500, err)
def handle_reload(self) :
act = self.uparam.get("reload")
if act != "cfg":
raise Pebkac(400, "only config files ('cfg') can be reloaded rn")
if not self.avol:
raise Pebkac(403, "'reload' not allowed for user " + self.uname)
if self.args.no_reload:
raise Pebkac(403, "the reload feature is disabled in server config")
x = self.conn.hsrv.broker.ask("reload", True, True)
return self.redirect("", "?h", x.get(), "return to", False)
def tx_stack(self) :
zs = self.args.stack_who
if zs == "all" or (
(zs == "a" and self.avol)
or (zs == "rw" and [x for x in self.wvol if x in self.rvol])
):
pass
else:
raise Pebkac(403, "'stack' not allowed for user " + self.uname)
ret = html_escape(alltrace(self.args.stack_v))
if self.args.stack_v:
ret = "%s\n%s" % (time.time(), ret)
else:
ret = "%s" % (ret,)
self.reply(ret.encode("utf-8"))
return True
def tx_tree(self) :
top = self.uparam["tree"] or ""
dst = self.vpath
if top in [".", ".."]:
top = undot(self.vpath + "/" + top)
if top == dst:
dst = ""
elif top:
if not dst.startswith(top + "/"):
raise Pebkac(422, "arg funk")
dst = dst[len(top) + 1 :]
ret = self.gen_tree(top, dst, self.uparam.get("k", ""))
if self.is_vproxied and not self.uparam["tree"]:
parents = self.args.R.split("/")
for parent in reversed(parents):
ret = {"k%s" % (parent,): ret, "a": []}
zs = json.dumps(ret)
self.reply(zs.encode("utf-8"), mime="application/json")
return True
def gen_tree(self, top , target , dk ) :
ret = {}
excl = None
if target:
excl, target = (target.split("/", 1) + [""])[:2]
sub = self.gen_tree("/".join([top, excl]).strip("/"), target, dk)
ret["k" + quotep(excl)] = sub
vfs = self.asrv.vfs
dk_sz = False
if dk:
vn, rem = vfs.get(top, self.uname, False, False)
if vn.flags.get("dks") and self._use_dirkey(vn, vn.canonical(rem)):
dk_sz = vn.flags.get("dk")
dots = False
fsroot = ""
try:
vn, rem = vfs.get(top, self.uname, not dk_sz, False)
fsroot, vfs_ls, vfs_virt = vn.ls(
rem,
self.uname,
not self.args.no_scandir,
PERMS_rwh,
)
dots = self.uname in vn.axs.udot
dk_sz = vn.flags.get("dk")
except:
dk_sz = None
vfs_ls = []
vfs_virt = {}
for v in self.rvol:
d1, d2 = v.rsplit("/", 1) if "/" in v else ["", v]
if d1 == top:
vfs_virt[d2] = vfs
dirs = [x[0] for x in vfs_ls if stat.S_ISDIR(x[1].st_mode)]
if not dots or "dots" not in self.uparam:
dirs = exclude_dotfiles(dirs)
dirs = [quotep(x) for x in dirs if x != excl]
if dk_sz and fsroot:
kdirs = []
fsroot_ = os.path.join(fsroot, "")
for dn in dirs:
ap = fsroot_ + dn
zs = self.gen_fk(2, self.args.dk_salt, ap, 0, 0)[:dk_sz]
kdirs.append(dn + "?k=" + zs)
dirs = kdirs
for x in vfs_virt:
if x != excl:
try:
dvn, drem = vfs.get(vjoin(top, x), self.uname, False, False)
if (
self.uname not in dvn.axs.uread
and self.uname not in dvn.axs.uwrite
and self.uname not in dvn.axs.uhtml
):
raise Exception()
bos.stat(dvn.canonical(drem, False))
except:
x += "\n"
dirs.append(x)
ret["a"] = dirs
return ret
def get_dls(self) :
ret = []
dls = self.conn.hsrv.tdls
enshare = self.args.shr
shrs = enshare[1:]
for dl_id, (t0, sz, vn, vp, uname) in self.conn.hsrv.tdli.items():
t1, sent = dls[dl_id]
if sent > 0x100000:
sent -= 0x100000
if self.uname not in vn.axs.uread:
vp = ""
elif self.uname not in vn.axs.udot and (vp.startswith(".") or "/." in vp):
vp = ""
elif (
enshare
and vp.startswith(shrs)
and self.uname != vn.shr_owner
and self.uname not in vn.axs.uadmin
and self.uname not in self.args.shr_adm
and not dl_id.startswith(self.ip + ":")
):
vp = ""
if self.uname not in vn.axs.uadmin:
dl_id = uname = ""
ret.append([t0, t1, sent, sz, vp, dl_id, uname])
return ret
def tx_dls(self) :
ret = [
{
"t0": x[0],
"t1": x[1],
"sent": x[2],
"size": x[3],
"path": x[4],
"conn": x[5],
"uname": x[6],
}
for x in self.get_dls()
]
zs = json.dumps(ret, separators=(",\n", ": "))
self.reply(zs.encode("utf-8", "replace"), mime="application/json")
return True
def tx_ups(self) :
idx = self.conn.get_u2idx()
if not idx or not hasattr(idx, "p_end"):
if not HAVE_SQLITE3:
raise Pebkac(500, "sqlite3 not found on server; unpost is disabled")
raise Pebkac(500, "server busy, cannot unpost; please retry in a bit")
sfilt = self.uparam.get("filter") or ""
nfi, vfi = str_anchor(sfilt)
lm = "ups %d%r" % (nfi, sfilt)
if self.args.shr and self.vpath.startswith(self.args.shr1):
shr_dbv, shr_vrem = self.vn.get_dbv(self.rem)
else:
shr_dbv = None
wret = {}
ret = []
t0 = time.time()
lim = time.time() - self.args.unpost
fk_vols = {
vol: (vol.flags["fk"], 2 if "fka" in vol.flags else 1)
for vp, vol in self.asrv.vfs.all_vols.items()
if "fk" in vol.flags
and (self.uname in vol.axs.uread or self.uname in vol.axs.upget)
}
if hasattr(self, "bad_xff"):
allvols = []
t = "will not return list of recent uploads" + BADXFF
self.log(t, 1)
if self.avol:
raise Pebkac(500, t)
x = self.conn.hsrv.broker.ask("up2k.get_unfinished_by_user", self.uname, "")
else:
x = self.conn.hsrv.broker.ask(
"up2k.get_unfinished_by_user", self.uname, self.ip
)
zdsa = x.get()
uret = []
if "timeout" in zdsa:
wret["nou"] = 1
else:
uret = zdsa["f"]
nu = len(uret)
if not self.args.unpost:
allvols = []
else:
allvols = list(self.asrv.vfs.all_vols.values())
allvols = [
x
for x in allvols
if "e2d" in x.flags
and ("*" in x.axs.uwrite or self.uname in x.axs.uwrite or x == shr_dbv)
]
q = ""
qp = (0,)
q_c = -1
for vol in allvols:
cur = idx.get_cur(vol)
if not cur:
continue
nfk, fk_alg = fk_vols.get(vol) or (0, 0)
zi = vol.flags["unp_who"]
if q_c != zi:
q_c = zi
q = "select sz, rd, fn, at from up where "
if zi == 1:
q += "ip=? and un=?"
qp = (self.ip, self.uname, lim)
elif zi == 2:
q += "ip=?"
qp = (self.ip, lim)
if zi == 3:
q += "un=?"
qp = (self.uname, lim)
q += " and at>? order by at desc"
n = 2000
for sz, rd, fn, at in cur.execute(q, qp):
vp = "/" + "/".join(x for x in [vol.vpath, rd, fn] if x)
if nfi == 0 or (nfi == 1 and vfi in vp.lower()):
pass
elif nfi == 2:
if not vp.lower().startswith(vfi):
continue
elif nfi == 3:
if not vp.lower().endswith(vfi):
continue
else:
continue
n -= 1
if not n:
break
rv = {"vp": vp, "sz": sz, "at": at, "nfk": nfk}
if nfk:
rv["ap"] = vol.canonical(vjoin(rd, fn))
rv["fk_alg"] = fk_alg
ret.append(rv)
if len(ret) > 3000:
ret.sort(key=lambda x: x["at"], reverse=True)
ret = ret[:2000]
ret.sort(key=lambda x: x["at"], reverse=True)
if len(ret) > 2000:
ret = ret[:2000]
if len(ret) >= 2000:
wret["oc"] = 1
for rv in ret:
rv["vp"] = quotep(rv["vp"])
nfk = rv.pop("nfk")
if not nfk:
continue
alg = rv.pop("fk_alg")
ap = rv.pop("ap")
try:
st = bos.stat(ap)
except:
continue
fk = self.gen_fk(
alg, self.args.fk_salt, ap, st.st_size, 0 if ANYWIN else st.st_ino
)
rv["vp"] += "?k=" + fk[:nfk]
if not allvols:
wret["noc"] = 1
ret = []
nc = len(ret)
ret = uret + ret
if shr_dbv:
vp_shr, vp_vfs = vroots(self.vpath, vjoin(shr_dbv.vpath, shr_vrem))
for v in ret:
vp = v["vp"]
if vp.startswith(vp_vfs):
v["vp"] = vp_shr + vp[len(vp_vfs) :]
if self.is_vproxied:
for v in ret:
v["vp"] = self.args.SR + v["vp"]
wret["f"] = ret
wret["nu"] = nu
wret["nc"] = nc
jtxt = json.dumps(wret, separators=(",\n", ": "))
self.log("%s #%d+%d %.2fsec" % (lm, nu, nc, time.time() - t0))
self.reply(jtxt.encode("utf-8", "replace"), mime="application/json")
return True
def tx_rups(self) :
if self.args.no_ups_page:
raise Pebkac(500, "listing of recent uploads is disabled in server config")
idx = self.conn.get_u2idx()
if not idx or not hasattr(idx, "p_end"):
if not HAVE_SQLITE3:
raise Pebkac(500, "sqlite3 not found on server; recent-uploads n/a")
raise Pebkac(500, "server busy, cannot list recent uploads; please retry")
sfilt = self.uparam.get("filter") or ""
nfi, vfi = str_anchor(sfilt)
lm = "ru %d%r" % (nfi, sfilt)
self.log(lm)
ret = []
t0 = time.time()
allvols = [
x
for x in self.asrv.vfs.all_vols.values()
if "e2d" in x.flags and ("*" in x.axs.uread or self.uname in x.axs.uread)
]
fk_vols = {
vol: (vol.flags["fk"], 2 if "fka" in vol.flags else 1)
for vol in allvols
if "fk" in vol.flags and "*" not in vol.axs.uread
}
for vol in allvols:
cur = idx.get_cur(vol)
if not cur:
continue
nfk, fk_alg = fk_vols.get(vol) or (0, 0)
adm = "*" in vol.axs.uadmin or self.uname in vol.axs.uadmin
dots = "*" in vol.axs.udot or self.uname in vol.axs.udot
lvl = vol.flags["ups_who"]
if not lvl:
continue
elif lvl == 1 and not adm:
continue
n = 1000
q = "select sz, rd, fn, ip, at, un from up where at>0 order by at desc"
for sz, rd, fn, ip, at, un in cur.execute(q):
vp = "/" + "/".join(x for x in [vol.vpath, rd, fn] if x)
if nfi == 0 or (nfi == 1 and vfi in vp.lower()):
pass
elif nfi == 2:
if not vp.lower().startswith(vfi):
continue
elif nfi == 3:
if not vp.lower().endswith(vfi):
continue
else:
continue
if not dots and "/." in vp:
continue
rv = {
"vp": vp,
"sz": sz,
"ip": ip,
"at": at,
"un": un,
"nfk": nfk,
"adm": adm,
}
if nfk:
rv["ap"] = vol.canonical(vjoin(rd, fn))
rv["fk_alg"] = fk_alg
ret.append(rv)
if len(ret) > 2000:
ret.sort(key=lambda x: x["at"], reverse=True)
ret = ret[:1000]
n -= 1
if not n:
break
ret.sort(key=lambda x: x["at"], reverse=True)
if len(ret) > 1000:
ret = ret[:1000]
for rv in ret:
rv["vp"] = quotep(rv["vp"])
nfk = rv.pop("nfk")
if not nfk:
continue
alg = rv.pop("fk_alg")
ap = rv.pop("ap")
try:
st = bos.stat(ap)
except:
continue
fk = self.gen_fk(
alg, self.args.fk_salt, ap, st.st_size, 0 if ANYWIN else st.st_ino
)
rv["vp"] += "?k=" + fk[:nfk]
if self.args.ups_when:
for rv in ret:
adm = rv.pop("adm")
if not adm:
rv["ip"] = "(You)" if rv["ip"] == self.ip else "(?)"
if rv["un"] not in ("*", self.uname):
rv["un"] = "(?)"
else:
for rv in ret:
adm = rv.pop("adm")
if not adm:
rv["ip"] = "(You)" if rv["ip"] == self.ip else "(?)"
rv["at"] = 0
if rv["un"] not in ("*", self.uname):
rv["un"] = "(?)"
if self.is_vproxied:
for v in ret:
v["vp"] = self.args.SR + v["vp"]
now = time.time()
self.log("%s #%d %.2fsec" % (lm, len(ret), now - t0))
ret2 = {"now": int(now), "filter": sfilt, "ups": ret}
jtxt = json.dumps(ret2, separators=(",\n", ": "))
if "j" in self.ouparam:
self.reply(jtxt.encode("utf-8", "replace"), mime="application/json")
return True
html = self.j2s("rups", this=self, v=json_hesc(jtxt))
self.reply(html.encode("utf-8"), status=200)
return True
def tx_idp(self) :
if self.uname.lower() not in self.args.idp_adm_set:
raise Pebkac(403, "'idp' not allowed for user " + self.uname)
cmd = self.uparam["idp"]
if cmd.startswith("rm="):
import sqlite3
db = sqlite3.connect(self.args.idp_db)
db.execute("delete from us where un=?", (cmd[3:],))
db.commit()
db.close()
self.conn.hsrv.broker.ask("reload", False, True).get()
self.redirect("", "?idp")
return True
rows = [
[k, "[%s]" % ("], [".join(v))]
for k, v in sorted(self.asrv.idp_accs.items())
]
html = self.j2s("idp", this=self, rows=rows, now=int(time.time()))
self.reply(html.encode("utf-8"), status=200)
return True
def tx_shares(self) :
if self.uname == "*":
self.loud_reply("you're not logged in")
return True
idx = self.conn.get_u2idx()
if not idx or not hasattr(idx, "p_end"):
if not HAVE_SQLITE3:
raise Pebkac(500, "sqlite3 not found on server; sharing is disabled")
raise Pebkac(500, "server busy, cannot list shares; please retry in a bit")
cur = idx.get_shr()
if not cur:
raise Pebkac(400, "huh, sharing must be disabled in the server config...")
rows = cur.execute("select * from sh").fetchall()
rows = [list(x) for x in rows]
if self.uname != self.args.shr_adm:
rows = [x for x in rows if x[5] == self.uname]
q = "select vp from sf where k=? limit 99"
for r in rows:
if not r[4]:
r[4] = "---"
else:
zstl = cur.execute(q, (r[0],)).fetchall()
zsl = [html_escape(zst[0]) for zst in zstl]
r[4] = "
".join(zsl)
if self.args.shr_site:
site = self.args.shr_site[:-1]
elif self.is_vproxied:
site = self.args.SR
else:
site = ""
html = self.j2s(
"shares",
this=self,
shr=self.args.shr,
site=site,
rows=rows,
now=int(time.time()),
)
self.reply(html.encode("utf-8"), status=200)
return True
def handle_eshare(self) :
idx = self.conn.get_u2idx()
if not idx or not hasattr(idx, "p_end"):
if not HAVE_SQLITE3:
raise Pebkac(500, "sqlite3 not found on server; sharing is disabled")
raise Pebkac(500, "server busy, cannot create share; please retry in a bit")
skey = self.uparam.get("skey") or self.vpath.split("/")[-1]
if self.args.shr_v:
self.log("handle_eshare: " + skey)
cur = idx.get_shr()
if not cur:
raise Pebkac(400, "huh, sharing must be disabled in the server config...")
rows = cur.execute("select un, t1 from sh where k = ?", (skey,)).fetchall()
un = rows[0][0] if rows and rows[0] else ""
if not un:
raise Pebkac(400, "that sharekey didn't match anything")
expiry = rows[0][1]
if un != self.uname and self.uname != self.args.shr_adm:
t = "your username (%r) does not match the sharekey's owner (%r) and you're not admin"
raise Pebkac(400, t % (self.uname, un))
reload = False
act = self.uparam["eshare"]
if act == "rm":
cur.execute("delete from sh where k = ?", (skey,))
if skey in self.asrv.vfs.nodes[self.args.shr.strip("/")].nodes:
reload = True
else:
now = time.time()
if expiry < now:
expiry = now
reload = True
expiry += int(act) * 60
cur.execute("update sh set t1 = ? where k = ?", (expiry, skey))
cur.connection.commit()
if reload:
self.conn.hsrv.broker.ask("reload", False, True).get()
self.conn.hsrv.broker.ask("up2k.wake_rescanner").get()
self.redirect("", "?shares")
return True
def handle_share(self, req ) :
idx = self.conn.get_u2idx()
if not idx or not hasattr(idx, "p_end"):
if not HAVE_SQLITE3:
raise Pebkac(500, "sqlite3 not found on server; sharing is disabled")
raise Pebkac(500, "server busy, cannot create share; please retry in a bit")
if self.args.shr_v:
self.log("handle_share: " + json.dumps(req, indent=4))
skey = req["k"]
vps = req["vp"]
fns = []
if len(vps) == 1:
vp = vps[0]
if not vp.endswith("/"):
vp, zs = vp.rsplit("/", 1)
fns = [zs]
else:
for zs in vps:
if zs.endswith("/"):
t = "you cannot select more than one folder, or mix files and folders in one selection"
raise Pebkac(400, t)
vp = vps[0].rsplit("/", 1)[0]
for zs in vps:
vp2, fn = zs.rsplit("/", 1)
fns.append(fn)
if vp != vp2:
t = "mismatching base paths in selection:\n %r\n %r"
raise Pebkac(400, t % (vp, vp2))
vp = vp.strip("/")
if self.is_vproxied and (vp == self.args.R or vp.startswith(self.args.RS)):
vp = vp[len(self.args.RS) :]
m = re.search(r"([^0-9a-zA-Z_-])", skey)
if m:
raise Pebkac(400, "sharekey has illegal character %r" % (m[1],))
if vp.startswith(self.args.shr1):
raise Pebkac(400, "yo dawg...")
cur = idx.get_shr()
if not cur:
raise Pebkac(400, "huh, sharing must be disabled in the server config...")
q = "select * from sh where k = ?"
qr = cur.execute(q, (skey,)).fetchall()
if qr and qr[0]:
self.log("sharekey taken by %r" % (qr,))
raise Pebkac(400, "sharekey %r is already in use" % (skey,))
s_rd = "read" in req["perms"]
s_wr = "write" in req["perms"]
s_get = "get" in req["perms"]
s_axs = [s_rd, s_wr, False, False, s_get]
if s_axs == [False] * 5:
raise Pebkac(400, "select at least one permission")
try:
vfs, rem = self.asrv.vfs.get(vp, self.uname, *s_axs)
except:
raise Pebkac(400, "you dont have all the perms you tried to grant")
zs = vfs.flags["shr_who"]
if zs == "auth" and self.uname != "*":
pass
elif zs == "a" and self.uname in vfs.axs.uadmin:
pass
else:
raise Pebkac(400, "you dont have perms to create shares from this volume")
ap, reals, _ = vfs.ls(rem, self.uname, not self.args.no_scandir, [s_axs])
rfns = set([x[0] for x in reals])
for fn in fns:
if fn not in rfns:
raise Pebkac(400, "selected file not found on disk: %r" % (fn,))
pw = req.get("pw") or ""
pw = self.asrv.ah.hash(pw)
now = int(time.time())
sexp = req["exp"]
exp = int(sexp) if sexp else 0
exp = now + exp * 60 if exp else 0
pr = "".join(zc for zc, zb in zip("rwmdg", s_axs) if zb)
q = "insert into sh values (?,?,?,?,?,?,?,?)"
cur.execute(q, (skey, pw, vp, pr, len(fns), self.uname, now, exp))
q = "insert into sf values (?,?)"
for fn in fns:
cur.execute(q, (skey, fn))
cur.connection.commit()
self.conn.hsrv.broker.ask("reload", False, True).get()
self.conn.hsrv.broker.ask("up2k.wake_rescanner").get()
fn = quotep(fns[0]) if len(fns) == 1 else ""
if self.args.shr_site:
surl = "created share: %s%s%s/%s" % (
self.args.shr_site,
self.args.shr[1:],
skey,
fn,
)
else:
surl = "created share: %s://%s%s%s%s/%s" % (
"https" if self.is_https else "http",
self.host,
self.args.SR,
self.args.shr,
skey,
fn,
)
self.loud_reply(surl, status=201)
return True
def handle_rm(self, req ) :
if not req and not self.can_delete:
if self.mode == "DELETE" and self.uname == "*":
raise Pebkac(401, "authenticate")
raise Pebkac(403, "'delete' not allowed for user " + self.uname)
if self.args.no_del:
raise Pebkac(403, "the delete feature is disabled in server config")
unpost = "unpost" in self.uparam
if unpost and hasattr(self, "bad_xff"):
self.log("unpost was denied" + BADXFF, 1)
raise Pebkac(403, "the delete feature is disabled in server config")
if not unpost and self.vn.shr_src:
raise Pebkac(403, "files in shares can only be deleted with unpost")
if not req:
req = [self.vpath]
elif self.is_vproxied:
req = [x[len(self.args.SR) :] for x in req]
nlim = int(self.uparam.get("lim") or 0)
lim = [nlim, nlim] if nlim else []
x = self.conn.hsrv.broker.ask(
"up2k.handle_rm", self.uname, self.ip, req, lim, False, unpost
)
self.loud_reply(x.get())
return True
def handle_mv(self) :
dst = self.uparam.get("move")
if self.is_vproxied and dst and dst.startswith(self.args.SR):
dst = dst[len(self.args.RS) :]
if not dst:
raise Pebkac(400, "need dst vpath")
return self._mv(self.vpath, dst.lstrip("/"), False)
def _mv(self, vsrc , vdst , overwrite ) :
if self.args.no_mv:
raise Pebkac(403, "the rename/move feature is disabled in server config")
svn, srem = self.asrv.vfs.get(vsrc, self.uname, True, False, True)
dvn, drem = self.asrv.vfs.get(vdst, self.uname, False, True)
if overwrite:
dabs = dvn.canonical(drem)
if bos.path.exists(dabs):
self.log("overwriting %s" % (dabs,))
self.asrv.vfs.get(vdst, self.uname, False, True, False, True)
wunlink(self.log, dabs, dvn.flags)
x = self.conn.hsrv.broker.ask(
"up2k.handle_mv", self.ouparam.get("akey"), self.uname, self.ip, vsrc, vdst
)
self.loud_reply(x.get(), status=201)
return True
def handle_cp(self) :
dst = self.uparam.get("copy")
if self.is_vproxied and dst and dst.startswith(self.args.SR):
dst = dst[len(self.args.RS) :]
if not dst:
raise Pebkac(400, "need dst vpath")
return self._cp(self.vpath, dst.lstrip("/"), False)
def _cp(self, vsrc , vdst , overwrite ) :
if self.args.no_cp:
raise Pebkac(403, "the copy feature is disabled in server config")
svn, srem = self.asrv.vfs.get(vsrc, self.uname, True, False)
dvn, drem = self.asrv.vfs.get(vdst, self.uname, False, True)
if overwrite:
dabs = dvn.canonical(drem)
if bos.path.exists(dabs):
self.log("overwriting %s" % (dabs,))
self.asrv.vfs.get(vdst, self.uname, False, True, False, True)
wunlink(self.log, dabs, dvn.flags)
x = self.conn.hsrv.broker.ask(
"up2k.handle_cp", self.ouparam.get("akey"), self.uname, self.ip, vsrc, vdst
)
self.loud_reply(x.get(), status=201)
return True
def handle_fs_abrt(self):
if self.args.no_fs_abrt:
t = "aborting an ongoing copy/move is disabled in server config"
raise Pebkac(403, t)
self.conn.hsrv.broker.say("up2k.handle_fs_abrt", self.uparam["fs_abrt"])
self.loud_reply("aborting", status=200)
return True
def tx_ls_vols(self) :
e_d = {}
eses = ["", ""]
rvol = self.rvol
wvol = self.wvol
allvols = self.asrv.vfs.all_nodes
if self.args.have_unlistc:
rvol = [x for x in rvol if "unlistcr" not in allvols[x].flags]
wvol = [x for x in wvol if "unlistcw" not in allvols[x].flags]
vols = [(x, allvols[x]) for x in list(set(rvol + wvol))]
if self.vpath:
zs = "%s/" % (self.vpath,)
vols = [(x[len(zs) :], y) for x, y in vols if x.startswith(zs)]
vols = [(x.split("/", 1)[0], y) for x, y in vols]
vols = list(({x: y for x, y in vols if x}).items())
if not vols and self.vpath:
return self.tx_404(True)
dirs = [
{
"lead": "",
"href": "%s/" % (x,),
"ext": "---",
"sz": 0,
"ts": 0,
"tags": e_d,
"dt": 0,
"name": 0,
"perms": vn.get_perms("", self.uname),
}
for x, vn in sorted(vols)
]
ls = {
"dirs": dirs,
"files": [],
"acct": self.uname,
"perms": [],
"taglist": [],
"logues": eses,
"readmes": eses,
"srvinf": "" if self.args.nih else self.args.name,
}
return self.tx_ls(ls)
def tx_ls(self, ls ) :
dirs = ls["dirs"]
files = ls["files"]
arg = self.uparam["ls"]
if arg in ["v", "t", "txt"]:
try:
biggest = max(ls["files"] + ls["dirs"], key=itemgetter("sz"))["sz"]
except:
biggest = 0
if arg == "v":
fmt = "\033[0;7;36m{{}}{{:>{}}}\033[0m {{}}"
nfmt = "{}"
biggest = 0
f2 = "".join(
"{}{{}}".format(x)
for x in [
"\033[7m",
"\033[27m",
"",
"\033[0;1m",
"\033[0;36m",
"\033[0m",
]
)
ctab = {"B": 6, "K": 5, "M": 1, "G": 3}
for lst in [dirs, files]:
for x in lst:
a = x["dt"].replace("-", " ").replace(":", " ").split(" ")
x["dt"] = f2.format(*list(a))
sz = humansize(x["sz"], True)
x["sz"] = "\033[0;3{}m {:>5}".format(ctab.get(sz[-1:], 0), sz)
else:
fmt = "{{}} {{:{},}} {{}}"
nfmt = "{:,}"
for x in dirs:
n = x["name"] + "/"
if arg == "v":
n = "\033[94m" + n
x["name"] = n
fmt = fmt.format(len(nfmt.format(biggest)))
retl = [
("# %s: %s" % (x, ls[x])).replace(r" // ", " // ")
for x in ["acct", "perms", "srvinf"]
if x in ls
]
retl += [
fmt.format(x["dt"], x["sz"], x["name"])
for y in [dirs, files]
for x in y
]
ret = "\n".join(retl)
mime = "text/plain; charset=utf-8"
else:
[x.pop(k) for k in ["name", "dt"] for y in [dirs, files] for x in y]
if self.args.no_fnugg:
ls["fnugg"] = "nei"
elif "fnugg" in self.headers:
ls["fnugg"] = self.headers["fnugg"]
ret = json.dumps(ls)
mime = "application/json"
ret += "\n\033[0m" if arg == "v" else "\n"
self.reply(ret.encode("utf-8", "replace"), mime=mime)
return True
def tx_browser(self) :
vpath = ""
vpnodes = [["", "/"]]
if self.vpath:
for node in self.vpath.split("/"):
if not vpath:
vpath = node
else:
vpath += "/" + node
vpnodes.append([quotep(vpath) + "/", html_escape(node, crlf=True)])
vn = self.vn
rem = self.rem
abspath = vn.dcanonical(rem)
dbv, vrem = vn.get_dbv(rem)
try:
st = bos.stat(abspath)
except:
if "on404" not in vn.flags:
return self.tx_404(not self.can_read)
ret = self.on40x(vn.flags["on404"], vn, rem)
if ret == "true":
return True
elif ret == "false":
return False
elif ret == "retry":
try:
st = bos.stat(abspath)
except:
return self.tx_404(not self.can_read)
else:
return self.tx_404(not self.can_read)
if rem.startswith(".hist/up2k.") or (
rem.endswith("/dir.txt") and rem.startswith(".hist/th/")
):
raise Pebkac(403)
e2d = "e2d" in vn.flags
e2t = "e2t" in vn.flags
add_og = "og" in vn.flags
if add_og:
if "th" in self.uparam or "raw" in self.uparam or "opds" in self.uparam:
add_og = False
elif vn.flags["og_ua"]:
add_og = vn.flags["og_ua"].search(self.ua)
og_fn = ""
if "v" in self.uparam:
add_og = True
og_fn = ""
if "b" in self.uparam:
self.out_headers["X-Robots-Tag"] = "noindex, nofollow"
is_dir = stat.S_ISDIR(st.st_mode)
is_dk = False
fk_pass = False
icur = None
if (e2t or e2d) and (is_dir or add_og):
idx = self.conn.get_u2idx()
if idx and hasattr(idx, "p_end"):
icur = idx.get_cur(dbv)
if "k" in self.uparam or "dky" in vn.flags:
if is_dir:
use_dirkey = self._use_dirkey(vn, abspath)
use_filekey = False
else:
use_filekey = self._use_filekey(vn, abspath, st)
use_dirkey = False
else:
use_dirkey = use_filekey = False
th_fmt = self.uparam.get("th")
if self.can_read or (
self.can_get
and (use_filekey or use_dirkey or (not is_dir and "fk" not in vn.flags))
):
if th_fmt is not None:
nothumb = "dthumb" in dbv.flags
if is_dir:
vrem = vrem.rstrip("/")
if nothumb:
pass
elif icur and vrem:
q = "select fn from cv where rd=? and dn=?"
crd, cdn = vrem.rsplit("/", 1) if "/" in vrem else ("", vrem)
try:
cfn = icur.execute(q, (crd, cdn)).fetchone()
if cfn:
fn = cfn[0]
fp = os.path.join(abspath, fn)
st = bos.stat(fp)
vrem = "{}/{}".format(vrem, fn).strip("/")
is_dir = False
except:
pass
else:
for fn in self.args.th_covers:
fp = os.path.join(abspath, fn)
try:
st = bos.stat(fp)
vrem = "{}/{}".format(vrem, fn).strip("/")
is_dir = False
break
except:
pass
if is_dir:
return self.tx_svg("folder")
thp = None
if self.thumbcli and not nothumb:
try:
thp = self.thumbcli.get(dbv, vrem, int(st.st_mtime), th_fmt)
except Pebkac as ex:
if ex.code == 500 and th_fmt[:1] in "jw":
self.log("failed to convert [%s]:\n%s" % (abspath, ex), 3)
return self.tx_svg("--error--\ncheck\nserver\nlog")
raise
if thp:
return self.tx_file(thp)
if th_fmt == "p":
raise Pebkac(404)
elif th_fmt in ACODE2_FMT:
raise Pebkac(415)
return self.tx_ico(rem)
elif self.can_write and th_fmt is not None:
return self.tx_svg("upload\nonly")
if not self.can_read and self.can_get and self.avn:
if not self.can_html:
pass
elif is_dir:
for fn in ("index.htm", "index.html"):
ap2 = os.path.join(abspath, fn)
try:
st2 = bos.stat(ap2)
except:
continue
if not stat.S_ISREG(st2.st_mode):
continue
if not self.trailing_slash:
return self.redirect(
self.vpath + "/", flavor="redirecting to", use302=True
)
fk_pass = True
is_dir = False
add_og = False
rem = vjoin(rem, fn)
vrem = vjoin(vrem, fn)
abspath = ap2
break
elif self.vpath.rsplit("/", 1)[-1] in IDX_HTML:
fk_pass = True
if not is_dir and (self.can_read or self.can_get):
if (
not self.can_read
and not fk_pass
and "fk" in vn.flags
and not use_filekey
and not self.vpath.startswith(self.args.shr1 or "\n")
):
return self.tx_404(True)
is_md = abspath.lower().endswith(".md")
if add_og and not is_md:
if self.host not in self.headers.get("referer", ""):
self.vpath, og_fn = vsplit(self.vpath)
vpath = self.vpath
vn, rem = self.asrv.vfs.get(self.vpath, self.uname, False, False)
abspath = vn.dcanonical(rem)
dbv, vrem = vn.get_dbv(rem)
is_dir = stat.S_ISDIR(st.st_mode)
is_dk = True
vpnodes.pop()
if (
(is_md or self.can_delete)
and "nohtml" not in vn.flags
and (
(is_md and "v" in self.uparam)
or "edit" in self.uparam
or "edit2" in self.uparam
)
):
return self.tx_md(vn, abspath)
if "zls" in self.uparam:
return self.tx_zls(abspath)
if "zget" in self.uparam:
return self.tx_zget(abspath)
if not add_og or not og_fn:
if st.st_size or "nopipe" in vn.flags:
return self.tx_file(abspath, None)
else:
return self.tx_file(abspath, vn.get_dbv("")[0].realpath)
elif is_dir and not self.can_read:
if use_dirkey:
is_dk = True
elif self.can_get and "doc" in self.uparam:
zs = vjoin(self.vpath, self.uparam["doc"]) + "?v"
return self.redirect(zs, flavor="redirecting to", use302=True)
elif not self.can_write:
return self.tx_404(True)
srv_info = []
try:
if not self.args.nih:
srv_info.append(self.args.name_html)
except:
self.log("#wow #whoa")
zi = vn.flags["du_iwho"]
if zi and (
zi == 9
or (zi == 7 and self.uname != "*")
or (zi == 5 and self.can_write)
or (zi == 4 and self.can_write and self.can_read)
or (zi == 3 and self.can_admin)
):
free, total, zs = get_df(abspath, False)
if total:
if "vmaxb" in vn.flags:
total = vn.lim.vbmax
if free == vn.lim.c_vb_r:
free = min(free, max(0, vn.lim.vbmax - vn.lim.c_vb_v))
else:
try:
zi, _ = self.conn.hsrv.broker.ask(
"up2k.get_volsizes", [vn.realpath]
).get()[0]
vn.lim.c_vb_v = zi
vn.lim.c_vb_r = free
free = min(free, max(0, vn.lim.vbmax - zi))
except:
pass
h1 = humansize(free or 0)
h2 = humansize(total)
srv_info.append("{} free of {}".format(h1, h2))
elif zs:
self.log("diskfree(%r): %s" % (abspath, zs), 3)
srv_infot = " // ".join(srv_info)
perms = []
if self.can_read or is_dk:
perms.append("read")
if self.can_write:
perms.append("write")
if self.can_move:
perms.append("move")
if self.can_delete:
perms.append("delete")
if self.can_get:
perms.append("get")
if self.can_upget:
perms.append("upget")
if self.can_admin:
perms.append("admin")
url_suf = self.urlq({}, ["k"])
is_ls = "ls" in self.uparam
is_opds = "opds" in self.uparam
is_js = self.args.force_js or self.cookies.get("js") == "y"
if not is_ls and not add_og and self.ua.startswith(("curl/", "fetch")):
self.uparam["ls"] = "v"
is_ls = True
tpl = "browser"
if "b" in self.uparam:
tpl = "browser2"
is_js = False
elif is_opds:
if not (self.args.opds or "opds" in self.vn.flags):
raise Pebkac(405, "OPDS is disabled in server config")
if not self.can_read:
raise Pebkac(401, "OPDS requires read permission")
is_js = is_ls = False
vf = vn.flags
ls_ret = {
"dirs": [],
"files": [],
"taglist": [],
"srvinf": srv_infot,
"acct": self.uname,
"perms": perms,
"cfg": vn.js_ls,
}
cgv = {
"ls0": None,
"acct": self.uname,
"perms": perms,
}
j2a = {
"cgv1": vn.js_htm,
"cgv": cgv,
"vpnodes": vpnodes,
"files": [],
"ls0": None,
"taglist": [],
"have_tags_idx": int(e2t),
"have_b_u": (self.can_write and self.uparam.get("b") == "u"),
"sb_lg": vn.js_ls["sb_lg"],
"url_suf": url_suf,
"title": html_escape("%s %s" % (self.args.bname, self.vpath), crlf=True),
"srv_info": srv_infot,
"dtheme": self.args.theme,
}
if self.args.js_browser:
zs = self.args.js_browser
zs += "&" if "?" in zs else "?"
j2a["js"] = zs
if self.args.css_browser:
zs = self.args.css_browser
zs += "&" if "?" in zs else "?"
j2a["css"] = zs
if not self.conn.hsrv.prism:
j2a["no_prism"] = True
if not self.can_read and not is_dk:
logues, readmes = self._add_logues(vn, abspath, None)
ls_ret["logues"] = j2a["logues"] = logues
ls_ret["readmes"] = cgv["readmes"] = readmes
if is_ls:
return self.tx_ls(ls_ret)
if not stat.S_ISDIR(st.st_mode):
return self.tx_404(True)
if "zip" in self.uparam or "tar" in self.uparam:
raise Pebkac(403)
zsl = j2a["files"] = []
if is_js:
j2a["ls0"] = cgv["ls0"] = {
"dirs": zsl,
"files": zsl,
"taglist": zsl,
}
html = self.j2s(tpl, **j2a)
self.reply(html.encode("utf-8", "replace"))
return True
for k in ["zip", "tar"]:
v = self.uparam.get(k)
if v is not None and (not add_og or not og_fn):
if is_dk and "dks" not in vn.flags:
t = "server config does not allow download-as-zip/tar; only dk is specified, need dks too"
raise Pebkac(403, t)
return self.tx_zip(k, v, self.vpath, vn, rem, [])
fsroot, vfs_ls, vfs_virt = vn.ls(
rem,
self.uname,
not self.args.no_scandir,
PERMS_rwh,
lstat="lt" in self.uparam,
throw=True,
)
stats = {k: v for k, v in vfs_ls}
ls_names = [x[0] for x in vfs_ls]
ls_names.extend(list(vfs_virt.keys()))
if add_og and og_fn and not self.can_read:
ls_names = [og_fn]
is_js = True
hist = {}
try:
if vf["md_hist"] != "s":
raise Exception()
histdir = os.path.join(fsroot, ".hist")
ptn = RE_MDV
for hfn in bos.listdir(histdir):
m = ptn.match(hfn)
if not m:
continue
fn = m.group(1) + m.group(3)
n, ts, _ = hist.get(fn, (0, 0, ""))
hist[fn] = (n + 1, max(ts, float(m.group(2))), hfn)
except:
pass
lnames = {x.lower(): x for x in ls_names}
if not self.can_dot or (
"dots" not in self.uparam and (is_ls or "dots" not in self.cookies)
):
ls_names = exclude_dotfiles(ls_names)
add_dk = vf.get("dk")
add_fk = vf.get("fk")
fk_alg = 2 if "fka" in vf else 1
if add_dk:
if vf.get("dky"):
add_dk = False
else:
zs = self.gen_fk(2, self.args.dk_salt, abspath, 0, 0)[:add_dk]
ls_ret["dk"] = cgv["dk"] = zs
no_zip = bool(self._can_zip(vf))
dirs = []
files = []
ptn_hr = RE_HR
use_abs_url = (
not is_opds
and not is_ls
and not is_js
and not self.trailing_slash
and vpath
)
for fn in ls_names:
base = ""
href = fn
if use_abs_url:
base = "/" + vpath + "/"
href = base + fn
if fn in vfs_virt:
fspath = vfs_virt[fn].realpath
else:
fspath = fsroot + "/" + fn
try:
linf = stats.get(fn) or bos.lstat(fspath)
inf = bos.stat(fspath) if stat.S_ISLNK(linf.st_mode) else linf
except:
self.log("broken symlink: %r" % (fspath,))
continue
is_dir = stat.S_ISDIR(inf.st_mode)
if is_dir:
href += "/"
if no_zip:
margin = "DIR"
elif add_dk:
zs = absreal(fspath)
margin = 'zip' % (
quotep(href),
self.gen_fk(2, self.args.dk_salt, zs, 0, 0)[:add_dk],
)
else:
margin = 'zip' % (
quotep(href),
)
elif fn in hist:
margin = '#%s' % (
base,
html_escape(hist[fn][2], quot=True, crlf=True),
hist[fn][0],
)
else:
margin = "-"
sz = inf.st_size
zd = datetime.fromtimestamp(max(0, linf.st_mtime), UTC)
dt = "%04d-%02d-%02d %02d:%02d:%02d" % (
zd.year,
zd.month,
zd.day,
zd.hour,
zd.minute,
zd.second,
)
if is_dir:
ext = "---"
elif "." in fn:
ext = ptn_hr.sub("@", fn.rsplit(".", 1)[1])
if len(ext) > 16:
ext = ext[:16]
else:
ext = "%"
if add_fk and not is_dir:
href = "%s?k=%s" % (
quotep(href),
self.gen_fk(
fk_alg,
self.args.fk_salt,
fspath,
sz,
0 if ANYWIN else inf.st_ino,
)[:add_fk],
)
elif add_dk and is_dir:
href = "%s?k=%s" % (
quotep(href),
self.gen_fk(2, self.args.dk_salt, fspath, 0, 0)[:add_dk],
)
else:
href = quotep(href)
item = {
"lead": margin,
"href": href,
"name": fn,
"sz": sz,
"ext": ext,
"dt": dt,
"ts": int(linf.st_mtime),
}
if is_dir:
dirs.append(item)
else:
files.append(item)
if is_dk and not vf.get("dks"):
dirs = []
if (
self.cookies.get("idxh") == "y"
and "ls" not in self.uparam
and "v" not in self.uparam
and not is_opds
):
for item in files:
if item["name"] in IDX_HTML:
vp = vjoin(self.vpath.split("?")[0], item["name"])
vn, rem = self.asrv.vfs.get(vp, self.uname, True, False)
ap = vn.canonical(rem)
if not self.trailing_slash and bos.path.isfile(ap):
return self.redirect(
self.vpath + "/", flavor="redirecting to", use302=True
)
return self.tx_file(ap)
if icur:
mte = vn.flags.get("mte") or {}
tagset = set()
rd = vrem
if self.can_admin:
up_q = "select substr(w,1,16), ip, at, un from up where rd=? and fn=?"
up_m = ["w", "up_ip", ".up_at", "up_by"]
else:
up_q, up_m = vn.flags["ls_q_m"]
mt_q = "select mt.k, mt.v from up inner join mt on mt.w = substr(up.w,1,16) where up.rd = ? and up.fn = ? and +mt.k != 'x'"
for fe in files:
fn = fe["name"]
erd_efn = (rd, fn)
try:
r = icur.execute(mt_q, erd_efn)
except Exception as ex:
if "database is locked" in str(ex):
break
try:
erd_efn = s3enc(idx.mem_cur, rd, fn)
r = icur.execute(mt_q, erd_efn)
except:
self.log("tag read error, %r / %r\n%s" % (rd, fn, min_ex()))
break
tags = {k: v for k, v in r}
if up_q:
try:
up_v = icur.execute(up_q, erd_efn).fetchone()
for zs1, zs2 in zip(up_m, up_v):
if zs2:
tags[zs1] = zs2
except:
pass
_ = [tagset.add(k) for k in tags]
fe["tags"] = tags
for fe in dirs:
fe["tags"] = ODict()
lmte = list(mte)
if self.can_admin:
lmte.extend(("w", "up_by", "up_ip", ".up_at"))
if "nodirsz" not in vf:
tagset.add(".files")
vdir = "%s/" % (rd,) if rd else ""
q = "select sz, nf from ds where rd=? limit 1"
for fe in dirs:
try:
hit = icur.execute(q, (vdir + fe["name"],)).fetchone()
(fe["sz"], fe["tags"][".files"]) = hit
except:
pass
taglist = [k for k in lmte if k in tagset]
else:
taglist = []
logues, readmes = self._add_logues(vn, abspath, lnames)
ls_ret["logues"] = j2a["logues"] = logues
ls_ret["readmes"] = cgv["readmes"] = readmes
if (
not files
and not dirs
and not readmes[0]
and not readmes[1]
and not logues[0]
and not logues[1]
):
logues[1] = "this folder is empty"
if "descript.ion" in lnames and os.path.isfile(
os.path.join(abspath, lnames["descript.ion"])
):
rem = []
items = {x["name"].lower(): x for x in files + dirs}
with open(os.path.join(abspath, lnames["descript.ion"]), "rb") as f:
for bln in [x.strip() for x in f]:
try:
if bln.endswith(b"\x04\xc2"):
bln = bln.replace(br"\\n", b" // ")[:-2]
ln = bln.decode("utf-8", "replace")
if ln.startswith('"'):
fn, desc = ln.split('" ', 1)
fn = fn[1:]
else:
fn, desc = ln.split(" ", 1)
try:
items[fn.lower()]["tags"]["descript.ion"] = desc
except:
t = "%s %sdescript.ion
\n"
logues[1] = t + "\n".join(rem) + "
"
if is_ls:
ls_ret["dirs"] = dirs
ls_ret["files"] = files
ls_ret["taglist"] = taglist
return self.tx_ls(ls_ret)
doc = self.uparam.get("doc") if self.can_read else None
if doc:
zp = self.args.ua_nodoc
if zp and zp.search(self.ua):
t = "this URL contains no valuable information for bots/crawlers"
raise Pebkac(403, t)
j2a["docname"] = doc
doctxt = None
dfn = lnames.get(doc.lower())
if dfn and dfn != doc:
dfn = next((x for x in files if x["name"] == doc), None)
if dfn:
docpath = os.path.join(abspath, doc)
sz = bos.path.getsize(docpath)
if sz < 1024 * self.args.txt_max:
doctxt = read_utf8(self.log, fsenc(docpath), False)
if doc.lower().endswith(".md") and "exp" in vn.flags:
doctxt = self._expand(doctxt, vn.flags.get("exp_md") or [])
else:
self.log("doc 2big: %r" % (doc,), 6)
doctxt = "( size of textfile exceeds serverside limit )"
else:
self.log("doc 404: %r" % (doc,), 6)
doctxt = "( textfile not found )"
if doctxt is not None:
j2a["doc"] = doctxt
for d in dirs:
d["name"] += "/"
dirs.sort(key=itemgetter("name"))
if is_opds:
allowed_exts = vf.get("opds_exts") or self.args.opds_exts
if allowed_exts:
files = [
x for x in files if x["name"].rsplit(".", 1)[-1] in allowed_exts
]
for item in dirs:
href = item["href"]
href += ("&" if "?" in href else "?") + "opds"
item["href"] = href
item["iso8601"] = "%sZ" % (item["dt"].replace(" ", "T"),)
for item in files:
href = item["href"]
href += ("&" if "?" in href else "?") + "dl"
item["href"] = href
item["iso8601"] = "%sZ" % (item["dt"].replace(" ", "T"),)
if "rmagic" in self.vn.flags:
ap = "%s/%s" % (fsroot, item["name"])
item["mime"] = guess_mime(item["name"], ap)
else:
item["mime"] = guess_mime(item["name"])
if (
not self.args.th_no_jpg
and self.thumbcli
and "dthumb" not in dbv.flags
and "dithumb" not in dbv.flags
):
item["jpeg_thumb_href"] = href + "&th=jf"
item["jpeg_thumb_href_hires"] = item["jpeg_thumb_href"] + "3"
j2a["files"] = files
j2a["dirs"] = dirs
html = self.j2s("opds", **j2a)
mime = "application/atom+xml;profile=opds-catalog"
self.reply(html.encode("utf-8", "replace"), mime=mime)
return True
if is_js:
j2a["ls0"] = cgv["ls0"] = {
"dirs": dirs,
"files": files,
"taglist": taglist,
}
j2a["files"] = []
else:
j2a["files"] = dirs + files
j2a["taglist"] = taglist
if add_og and "raw" not in self.uparam:
j2a["this"] = self
cgv["og_fn"] = og_fn
if og_fn and vn.flags.get("og_tpl"):
tpl = vn.flags["og_tpl"]
if "EXT" in tpl:
zs = og_fn.split(".")[-1].lower()
tpl2 = tpl.replace("EXT", zs)
if os.path.exists(tpl2):
tpl = tpl2
with self.conn.hsrv.mutex:
if tpl not in self.conn.hsrv.j2:
tdir, tname = os.path.split(tpl)
j2env = jinja2.Environment()
j2env.loader = jinja2.FileSystemLoader(tdir)
self.conn.hsrv.j2[tpl] = j2env.get_template(tname)
thumb = ""
is_pic = is_vid = is_au = False
for fn in self.args.th_coversd:
if fn in lnames:
thumb = lnames[fn]
break
if og_fn:
ext = og_fn.split(".")[-1].lower()
if self.thumbcli and ext in self.thumbcli.thumbable:
is_pic = (
ext in self.thumbcli.fmt_pil
or ext in self.thumbcli.fmt_vips
or ext in self.thumbcli.fmt_ffi
)
is_vid = ext in self.thumbcli.fmt_ffv
is_au = ext in self.thumbcli.fmt_ffa
if not thumb or not is_au:
thumb = og_fn
file = next((x for x in files if x["name"] == og_fn), None)
else:
file = None
url_base = "%s://%s/%s" % (
"https" if self.is_https else "http",
self.host,
self.args.RS + quotep(vpath),
)
j2a["og_is_pic"] = is_pic
j2a["og_is_vid"] = is_vid
j2a["og_is_au"] = is_au
if thumb:
fmt = vn.flags.get("og_th", "j")
th_base = ujoin(url_base, quotep(thumb))
query = "th=%s&cache" % (fmt,)
if use_filekey:
query += "&k=" + self.uparam["k"]
query = ub64enc(query.encode("utf-8")).decode("ascii")
query += "/th.jpg" if "j" in fmt else "/th.webp"
j2a["og_thumb"] = "%s/.uqe/%s" % (th_base, query)
j2a["og_fn"] = og_fn
j2a["og_file"] = file
if og_fn:
og_fn_q = quotep(og_fn)
query = "raw"
if use_filekey:
query += "&k=" + self.uparam["k"]
query = ub64enc(query.encode("utf-8")).decode("ascii")
query += "/%s" % (og_fn_q,)
j2a["og_url"] = ujoin(url_base, og_fn_q)
j2a["og_raw"] = j2a["og_url"] + "/.uqe/" + query
else:
j2a["og_url"] = j2a["og_raw"] = url_base
if not vn.flags.get("og_no_head"):
ogh = {"twitter:card": "summary"}
title = str(vn.flags.get("og_title") or "")
if thumb:
ogh["og:image"] = j2a["og_thumb"]
zso = vn.flags.get("og_desc") or ""
if zso != "-":
ogh["og:description"] = str(zso)
zs = vn.flags.get("og_site") or self.args.name
if zs not in ("", "-"):
ogh["og:site_name"] = zs
try:
zs1, zs2 = file["tags"]["res"].split("x")
file["tags"][".resw"] = zs1
file["tags"][".resh"] = zs2
except:
pass
tagmap = {}
if is_au:
title = str(vn.flags.get("og_title_a") or "")
ogh["og:type"] = "music.song"
ogh["og:audio"] = j2a["og_raw"]
tagmap = {
"artist": "og:music:musician",
"album": "og:music:album",
".dur": "og:music:duration",
}
elif is_vid:
title = str(vn.flags.get("og_title_v") or "")
ogh["og:type"] = "video.other"
ogh["og:video"] = j2a["og_raw"]
tagmap = {
"title": "og:title",
".dur": "og:video:duration",
".resw": "og:video:width",
".resh": "og:video:height",
}
elif is_pic:
title = str(vn.flags.get("og_title_i") or "")
ogh["twitter:card"] = "summary_large_image"
ogh["twitter:image"] = ogh["og:image"] = j2a["og_raw"]
tagmap = {
".resw": "og:image:width",
".resh": "og:image:height",
}
try:
for k, v in file["tags"].items():
zs = "{{ %s }}" % (k,)
title = title.replace(zs, str(v))
except:
pass
title = re.sub(r"\{\{ [^}]+ \}\}", "", title)
while title.startswith(" - "):
title = title[3:]
while title.endswith(" - "):
title = title[:3]
if vn.flags.get("og_s_title") or not title:
title = str(vn.flags.get("og_title") or "")
for tag, hname in tagmap.items():
try:
v = file["tags"][tag]
if not v:
continue
ogh[hname] = int(v) if tag == ".dur" else v
except:
pass
ogh["og:title"] = title
oghs = [
'\t'
% (k, html_escape(str(v), True, True))
for k, v in ogh.items()
]
zs = self.html_head + "\n%s\n" % ("\n".join(oghs),)
self.html_head = zs.replace("\n\n", "\n")
html = self.j2s(tpl, **j2a)
self.reply(html.encode("utf-8", "replace"))
return True