webdav: add tests + fix minor edgecases

* allow depth:0 at top of unmapped root

* cannot use the Referer header to identify
   graphical browsers since rclone sends it

copyparty/httpcli.py

@@ -1424,7 +1424,8 @@ def handle_propfind(self) -> bool:
 
         depth = self.headers.get("depth", "infinity").lower()
         if depth == "infinity":
-            if not self.can_read:
+            # allow depth:0 from unmapped root, but require read-axs otherwise
+            if not self.can_read and (self.vpath or self.asrv.vfs.realpath):
                 t = "depth:infinity requires read-access in /%s"
                 t = t % (self.vpath,)
                 self.log(t, 3)
@@ -4632,7 +4633,6 @@ def tx_404(self, is_403: bool = False) -> bool:
             and not self.pw
             and not self.ua.startswith("Mozilla/")
             and "sec-fetch-site" not in self.headers
-            and "referer" not in self.headers
         ):
             rc = 401
             self.out_headers["WWW-Authenticate"] = 'Basic realm="a"'

tests/test_httpcli.py

@@ -15,7 +15,7 @@
 from copyparty.authsrv import AuthSrv
 from copyparty.httpcli import HttpCli
 from tests import util as tu
-from tests.util import Cfg, eprint
+from tests.util import Cfg, eprint, pfind2ls
 
 
 def hdr(query):
@@ -129,6 +129,24 @@ def test(self):
                 else:
                     ref = []
 
+                h, b = self.propfind(durl, 1)
+                fns = [x for x in pfind2ls(b) if not x.endswith("/")]
+                if ref:
+                    self.assertIn("<D:propstat>", b)
+                elif not rok and not wok:
+                    self.assertListEqual([], fns)
+                else:
+                    self.assertIn("<D:multistatus", b)
+
+                h, b = self.propfind(durl, 0)
+                fns = [x for x in pfind2ls(b) if not x.endswith("/")]
+                if ref:
+                    self.assertIn("<D:propstat>", b)
+                elif not rok:
+                    self.assertListEqual([], fns)
+                else:
+                    self.assertIn("<D:multistatus", b)
+
                 if test_tar:
                     url = durl + "?tar"
                     h, b = self.curl(url, True)
@@ -225,5 +243,12 @@ def curl(self, url, binary=False):
 
         return conn.s._reply.decode("utf-8").split("\r\n\r\n", 1)
 
+    def propfind(self, url, depth=1):
+        zs = "PROPFIND /%s HTTP/1.1\r\nDepth: %d\r\nPW: o\r\nConnection: close\r\n\r\n"
+        buf = zs % (url, depth)
+        conn = self.conn.setbuf(buf.encode("utf-8"))
+        HttpCli(conn).run()
+        return conn.s._reply.decode("utf-8").split("\r\n\r\n", 1)
+
     def log(self, src, msg, c=0):
         print(msg)

tests/test_webdav.py

@@ -0,0 +1,264 @@
+#!/usr/bin/env python3
+# coding: utf-8
+from __future__ import print_function, unicode_literals
+
+import os
+import shutil
+import tempfile
+import time
+import unittest
+
+from copyparty.authsrv import AuthSrv
+from copyparty.httpcli import HttpCli
+from tests import util as tu
+from tests.util import TC, Cfg, pfind2ls
+
+# tcpdump of `rclone ls dav:`
+RCLONE_PROPFIND = """PROPFIND /%s HTTP/1.1
+Host: 127.0.0.1:3923
+User-Agent: rclone/v1.67.0
+Content-Length: 308
+Authorization: Basic azp1
+Depth: 1
+Referer: http://127.0.0.1:3923/
+Accept-Encoding: gzip
+
+<?xml version="1.0"?>
+<d:propfind  xmlns:d="DAV:" xmlns:oc="http://owncloud.org/ns" xmlns:nc="http://nextcloud.org/ns">
+ <d:prop>
+  <d:displayname />
+  <d:getlastmodified />
+  <d:getcontentlength />
+  <d:resourcetype />
+  <d:getcontenttype />
+  <oc:checksums />
+  <oc:permissions />
+ </d:prop>
+</d:propfind>
+"""
+
+
+# tcpdump of `rclone copy fa dav:/a/`  (it does a mkcol first)
+RCLONE_MKCOL = """MKCOL /%s HTTP/1.1
+Host: 127.0.0.1:3923
+User-Agent: rclone/v1.67.0
+Authorization: Basic azp1
+Referer: http://127.0.0.1:3923/
+Accept-Encoding: gzip
+\n"""
+
+
+# tcpdump of `rclone copy fa dav:/a/`  (the actual upload)
+RCLONE_PUT = """PUT /%s HTTP/1.1
+Host: 127.0.0.1:3923
+User-Agent: rclone/v1.67.0
+Content-Length: 6
+Authorization: Basic azp1
+Content-Type: application/octet-stream
+Oc-Checksum: SHA1:f5e3dc3fb27af53cd0005a1184e2df06481199e8
+Referer: http://127.0.0.1:3923/
+X-Oc-Mtime: 1689453578
+Accept-Encoding: gzip
+
+fgsfds"""
+
+
+# tcpdump of `rclone delete dav:/a/d1/`  (it does propfind recursively and then this on each file)
+# (note: `rclone rmdirs dav:/a/d1/` does the same thing but just each folder after asserting they're empty)
+RCLONE_DELETE = """DELETE /%s HTTP/1.1
+Host: 127.0.0.1:3923
+User-Agent: rclone/v1.67.0
+Authorization: Basic azp1
+Referer: http://127.0.0.1:3923/
+Accept-Encoding: gzip
+\n"""
+
+
+# tcpdump of `rclone move dav:/a/d1/d2 /a/d1/d3`  (it does a lot of boilerplate propfinds/mkcols before)
+RCLONE_MOVE = """MOVE /%s HTTP/1.1
+Host: 127.0.0.1:3923
+User-Agent: rclone/v1.67.0
+Authorization: Basic azp1
+Destination: http://127.0.0.1:3923/%s
+Overwrite: T
+Referer: http://127.0.0.1:3923/
+Accept-Encoding: gzip
+\n"""
+
+
+class TestHttpCli(TC):
+    def setUp(self):
+        self.td = tu.get_ramdisk()
+        self.maxDiff = 99999
+
+    def tearDown(self):
+        self.conn.shutdown()
+        os.chdir(tempfile.gettempdir())
+        shutil.rmtree(self.td)
+
+    def test(self):
+        td = os.path.join(self.td, "vfs")
+        os.mkdir(td)
+        os.chdir(td)
+
+        self.fn = "g{:x}g".format(int(time.time() * 3))
+        vcfg = [
+            "r:r:r,u",
+            "w:w:w,u",
+            "a:a:A,u",
+            "x:x:r,u2",
+            "x/r:x/r:r,u",
+            "x/x:x/x:r,u2",
+        ]
+        self.args = Cfg(v=vcfg, a=["u:u", "u2:u2"])
+        self.asrv = AuthSrv(self.args, self.log)
+        self.conn = tu.VHttpConn(self.args, self.asrv, self.log, b"", True)
+
+        self.fns = ["%s/%s" % (zs.split(":")[0], self.fn) for zs in vcfg]
+        for fp in self.fns:
+            try:
+                os.makedirs(os.path.dirname(fp))
+            except:
+                pass
+            with open(fp, "wb") as f:
+                f.write(("ok %s\n" % (fp,)).encode("utf-8"))
+
+        ##
+        ## depth:1 (regular listing)
+
+        # unmapped root; should return list of volumes
+        h, b = self.req(RCLONE_PROPFIND % ("",))
+        fns = pfind2ls(b)
+        self.assertStart("HTTP/1.1 207 Multi-Status\r", h)
+        self.assertListEqual(fns, ["/", "/a/", "/r/"])
+
+        # toplevel of a volume; has one file
+        h, b = self.req(RCLONE_PROPFIND % ("a",))
+        fns = pfind2ls(b)
+        self.assertStart("HTTP/1.1 207 Multi-Status\r", h)
+        self.assertListEqual(fns, ["/a/", "/a/" + self.fn])
+
+        # toplevel of a volume; has one file
+        h, b = self.req(RCLONE_PROPFIND % ("r",))
+        fns = pfind2ls(b)
+        self.assertStart("HTTP/1.1 207 Multi-Status\r", h)
+        self.assertListEqual(fns, ["/r/", "/r/" + self.fn])
+
+        # toplevel of write-only volume; has one file, will not list
+        h, b = self.req(RCLONE_PROPFIND % ("w",))
+        fns = pfind2ls(b)
+        self.assertStart("HTTP/1.1 207 Multi-Status\r", h)
+        self.assertListEqual(fns, ["/w/"])
+
+        ##
+        ## auth challenge
+
+        bad_pfind = RCLONE_PROPFIND.replace("Authorization: Basic azp1\n", "")
+        bad_put = RCLONE_PUT.replace("Authorization: Basic azp1\n", "")
+        urls = ["", "r", "w", "a"]
+        urls += [x + "/" + self.fn for x in urls[1:]]
+        for url in urls:
+            for q in (bad_pfind, bad_put):
+                h, b = self.req(q % (url,))
+                self.assertStart("HTTP/1.1 401 Unauthorized\r", h)
+                self.assertIn('\nWWW-Authenticate: Basic realm="a"\r', h)
+
+        ##
+        ## depth:0 (recursion)
+
+        # depth:0 from unmapped root should work;
+        # will NOT list contents of /x/r/ due to current limitations
+        # (stops descending at first non-accessible volume)
+        recursive = RCLONE_PROPFIND.replace("Depth: 1\n", "")
+        h, b = self.req(recursive % ("",))
+        fns = pfind2ls(b)
+        expect = ["/", "/a/", "/r/"]
+        expect += [x + self.fn for x in expect[1:]]
+        self.assertListEqual(fns, expect)
+
+        # same thing here...
+        h, b = self.req(recursive % ("/x",))
+        fns = pfind2ls(b)
+        self.assertListEqual(fns, [])
+
+        # but this obviously works
+        h, b = self.req(recursive % ("/x/r",))
+        fns = pfind2ls(b)
+        self.assertListEqual(fns, ["/x/r/", "/x/r/" + self.fn])
+
+        ##
+        ## uploading
+
+        # rclone does a propfind on the target file first; expects 404
+        h, b = self.req(RCLONE_PROPFIND % ("a/fa",))
+        self.assertStart("HTTP/1.1 404 Not Found\r", h)
+
+        # then it does a mkcol (mkdir), expecting 405 (exists)
+        h, b = self.req(RCLONE_MKCOL % ("a",))
+        self.assertStart("HTTP/1.1 405 Method Not Allowed\r", h)
+
+        # then it uploads the file
+        h, b = self.req(RCLONE_PUT % ("a/fa",))
+        self.assertStart("HTTP/1.1 201 Created\r", h)
+
+        # then it does a propfind to confirm
+        h, b = self.req(RCLONE_PROPFIND % ("a/fa",))
+        fns = pfind2ls(b)
+        self.assertStart("HTTP/1.1 207 Multi-Status\r", h)
+        self.assertListEqual(fns, ["/a/fa"])
+
+        ##
+        ## upload into set of subfolders that don't exist yet
+
+        # rclone does this:
+        # propfind /a/d1/d2/fa => 404
+        # mkcol /a/d1/d2/ => 409
+        # propfind /a/d1/d2/ => 404
+        # mkcol /a/d1/ => 201
+        # mkcol /a/d1/d2/ => 201
+        # put /a/d1/d2/fa => 201
+        # propfind /a/d1/d2/fa => 207
+        # ...some of which already tested above;
+
+        h, b = self.req(RCLONE_PROPFIND % ("/a/d1/d2/",))
+        self.assertStart("HTTP/1.1 404 Not Found\r", h)
+
+        h, b = self.req(RCLONE_PROPFIND % ("/a/d1/",))
+        self.assertStart("HTTP/1.1 404 Not Found\r", h)
+
+        h, b = self.req(RCLONE_MKCOL % ("/a/d1/d2/",))
+        self.assertStart("HTTP/1.1 409 Conflict\r", h)
+
+        h, b = self.req(RCLONE_MKCOL % ("/a/d1/",))
+        self.assertStart("HTTP/1.1 201 Created\r", h)
+
+        h, b = self.req(RCLONE_MKCOL % ("/a/d1/d2/",))
+        self.assertStart("HTTP/1.1 201 Created\r", h)
+
+        h, b = self.req(RCLONE_PUT % ("a/d1/d2/fa",))
+        self.assertStart("HTTP/1.1 201 Created\r", h)
+
+        ##
+        ## rename
+
+        h, b = self.req(RCLONE_MOVE % ("a/d1/d2/", "a/d1/d3/"))
+        self.assertStart("HTTP/1.1 201 Created\r", h)
+        self.assertListEqual(os.listdir("a/d1"), ["d3"])
+
+        ##
+        ## delete
+
+        h, b = self.req(RCLONE_DELETE % ("a/d1",))
+        self.assertStart("HTTP/1.1 200 OK\r", h)
+        if os.path.exists("a/d1"):
+            self.fail("a/d1 still exists")
+
+    def req(self, q):
+        h, b = q.split("\n\n", 1)
+        q = h.replace("\n", "\r\n") + "\r\n\r\n" + b
+        conn = self.conn.setbuf(q.encode("utf-8"))
+        HttpCli(conn).run()
+        return conn.s._reply.decode("utf-8").split("\r\n\r\n", 1)
+
+    def log(self, src, msg, c=0):
+        print(msg)

tests/util.py

@@ -12,6 +12,7 @@
 import tempfile
 import threading
 import time
+import unittest
 from argparse import Namespace
 
 import jinja2
@@ -118,14 +119,28 @@ def subdir(top):
     return subdir(ret)
 
 
+def pfind2ls(xml):
+    return [x.split("<", 1)[0] for x in xml.split("<D:href>")[1:]]
+
+
+class TC(unittest.TestCase):
+    def __init__(self, *a, **ka):
+        super(TC, self).__init__(*a, **ka)
+
+    def assertStart(self, member, container, msg=None):
+        if not container.startswith(member):
+            standardMsg = "%s not found in %s" % (member, container)
+            self.fail(self._formatMessage(msg, standardMsg))
+
+
 class Cfg(Namespace):
     def __init__(self, a=None, v=None, c=None, **ka0):
         ka = {}
 
-        ex = "chpw daw dav_auth dav_inf dav_mac dav_rt e2d e2ds e2dsa e2t e2ts e2tsr e2v e2vu e2vp early_ban ed emp exp force_js getmod grid gsel hardlink ih ihead magic hardlink_only nid nih no_acode no_athumb no_clone no_cp no_dav no_db_ip no_del no_dirsz no_dupe no_lifetime no_logues no_mv no_pipe no_poll no_readme no_robots no_sb_md no_sb_lg no_scandir no_tarcmp no_thumb no_vthumb no_zip nrand nsort nw og og_no_head og_s_title ohead q rand re_dirsz rss smb srch_dbg srch_excl stats uqe vague_403 vc ver write_uplog xdev xlink xvol zs"
+        ex = "chpw daw dav_auth dav_mac dav_rt e2d e2ds e2dsa e2t e2ts e2tsr e2v e2vu e2vp early_ban ed emp exp force_js getmod grid gsel hardlink ih ihead magic hardlink_only nid nih no_acode no_athumb no_bauth no_clone no_cp no_dav no_db_ip no_del no_dirsz no_dupe no_lifetime no_logues no_mv no_pipe no_poll no_readme no_robots no_sb_md no_sb_lg no_scandir no_tarcmp no_thumb no_vthumb no_zip nrand nsort nw og og_no_head og_s_title ohead q rand re_dirsz rss smb srch_dbg srch_excl stats uqe vague_403 vc ver write_uplog xdev xlink xvol zs"
         ka.update(**{k: False for k in ex.split()})
 
-        ex = "dedup dotpart dotsrch hook_v no_dhash no_fastboot no_fpool no_htp no_rescan no_sendfile no_ses no_snap no_up_list no_voldump re_dhash plain_ip"
+        ex = "dav_inf dedup dotpart dotsrch hook_v no_dhash no_fastboot no_fpool no_htp no_rescan no_sendfile no_ses no_snap no_up_list no_voldump re_dhash plain_ip"
         ka.update(**{k: True for k in ex.split()})
 
         ex = "ah_cli ah_gen css_browser hist ipu js_browser js_other mime mimes no_forget no_hash no_idx nonsus_urls og_tpl og_ua"