Merge branch 'master' of https://github.com/mar10/wsgidav

sample_wsgidav.yaml

@@ -127,6 +127,16 @@ provider_mapping:
         kwargs:
             path: '/path/to/share3'
             another_arg: 42
+    # Example: 
+    #     make sure that a `/favicon.ico` URL is resolved, even if a `*.html`
+    #     or `*.txt` resource file was opened using the DirBrowser
+    # '/':
+    #     class: 'wsgidav.fs_dav_provider.FilesystemProvider'
+    #     kwargs:
+    #         root_folder: 'tests/fixtures/share'
+    #         # readonly: true
+    #         shadow:
+    #             '/favicon.ico': 'file_path/to/favicon.ico'
 
 
 # ==============================================================================
@@ -179,6 +189,7 @@ pam_dc:
     encoding: 'utf-8'
     resetcreds: true
 
+
 # ----------------------------------------------------------------------------
 # CORS
 # (Requires `wsgidav.mw.cors.Cors`, which is enabled by default.)
@@ -242,6 +253,7 @@ mutable_live_props:
     # preserve-timestamp flags in a mounted DAV share (may be RFC4918 incompliant)
     - '{DAV:}getlastmodified'
 
+
 # ----------------------------------------------------------------------------
 # Lock Manager Storage
 #
@@ -263,7 +275,6 @@ mutable_live_props:
 lock_storage: true
 
 
-
 # ==============================================================================
 # DEBUGGING
 

tests/test_util.py

@@ -6,6 +6,7 @@
 
 import logging
 import logging.handlers
+import sys
 import unittest
 from io import StringIO
 
@@ -121,7 +122,10 @@ def testBasics(self):
         assert parse_if_match_header(' W/"abc" , def ') == ["abc", "def"]
 
         self.assertRaises(ValueError, fix_path, "a/b", "/root/x")
-        assert fix_path("a/b", "/root/x", must_exist=False) == "/root/x/a/b"
+        if sys.platform == "win32":
+            assert fix_path("a/b", "/root/x", must_exist=False) == r"C:\root\x\a\b"
+        else:
+            assert fix_path("a/b", "/root/x", must_exist=False) == "/root/x/a/b"
         assert fix_path("/a/b", "/root/x", must_exist=False) == "/a/b"
 
         headers = [("foo", "bar"), ("baz", "qux")]

wsgidav/fs_dav_provider.py

@@ -355,7 +355,7 @@ def set_last_modified(self, dest_path, time_stamp, *, dry_run):
 # FilesystemProvider
 # ========================================================================
 class FilesystemProvider(DAVProvider):
-    def __init__(self, root_folder, *, readonly=False):
+    def __init__(self, root_folder, *, readonly=False, shadow=None):
         # and resolve relative to config file
         # root_folder = os.path.expandvars(os.xpath.expanduser(root_folder))
         root_folder = os.path.abspath(root_folder)
@@ -366,6 +366,10 @@ def __init__(self, root_folder, *, readonly=False):
 
         self.root_folder_path = root_folder
         self.readonly = readonly
+        if shadow:
+            self.shadow = {k.lower(): v for k, v in shadow.items()}
+        else:
+            self.shadow = {}
 
     def __repr__(self):
         rw = "Read-Write"
@@ -375,6 +379,28 @@ def __repr__(self):
             self.__class__.__name__, self.root_folder_path, rw
         )
 
+    def _resolve_shadow_path(self, path, environ, file_path):
+        """File not found: See if there is a shadow configured."""
+        shadow = self.shadow.get(path.lower())
+        # _logger.info(f"Shadow {path} -> {shadow} {self.shadow}")
+        if not shadow:
+            return False, file_path
+
+        err = None
+        method = environ["REQUEST_METHOD"].upper()
+        if method not in ("GET", "HEAD", "OPTIONS"):
+            err = f"Shadow {path} -> {shadow}: ignored for method {method!r}."
+        elif os.path.exists(file_path):
+            err = f"Shadow {path} -> {shadow}: ignored for existing resource {file_path!r}."
+        elif not os.path.exists(shadow):
+            err = f"Shadow {path} -> {shadow}: does not exist."
+
+        if err:
+            _logger.warning(err)
+            return False, file_path
+        _logger.info(f"Shadow {path} -> {shadow}")
+        return True, shadow
+
     def _loc_to_file_path(self, path, environ=None):
         """Convert resource path to a unicode absolute file path.
         Optional environ argument may be useful e.g. in relation to per-user
@@ -387,7 +413,8 @@ def _loc_to_file_path(self, path, environ=None):
 
         path_parts = path.strip("/").split("/")
         file_path = os.path.abspath(os.path.join(root_path, *path_parts))
-        if not file_path.startswith(root_path):
+        is_shadow, file_path = self._resolve_shadow_path(path, environ, file_path)
+        if not file_path.startswith(root_path) and not is_shadow:
             raise RuntimeError(
                 "Security exception: tried to access file outside root: {}".format(
                     file_path

wsgidav/request_server.py

@@ -1570,7 +1570,7 @@ def _send_resource(self, environ, start_response, is_head_method):
             self._fail(
                 HTTP_FORBIDDEN,
                 "Directory browsing is not enabled."
-                "(to enable it put WsgiDavDirBrowser into the middleware_stack "
+                "(to enable it add WsgiDavDirBrowser to the middleware_stack "
                 "option and set dir_browser.enabled = True option.)",
             )
 

wsgidav/wsgidav_app.py

@@ -332,7 +332,7 @@ def add_provider(self, share, provider, *, readonly=False):
                 # Syntax:
                 #   <mount_path>: {"root": <path>, "redaonly": <bool>}
                 provider = FilesystemProvider(
-                    util.fix_path(provider["root"]),
+                    util.fix_path(provider["root"], self.config),
                     readonly=bool(provider.get("readonly", False)),
                 )
             else:
@@ -341,12 +341,12 @@ def add_provider(self, share, provider, *, readonly=False):
                 )
         elif type(provider) in (list, tuple):
             raise ValueError(
-                "Provider {}: tuple/list syntax is no longer supported".format(provider)
+                f"Provider {provider}: tuple/list syntax is no longer supported"
             )
             # provider = FilesystemProvider(provider[0], provider[1])
 
         if not isinstance(provider, DAVProvider):
-            raise ValueError("Invalid provider {}".format(provider))
+            raise ValueError(f"Invalid provider {provider}")
 
         provider.set_share_path(share)
         if self.mount_path:
@@ -414,7 +414,7 @@ def __call__(self, environ, start_response):
 
         # GC issue 22: Pylons sends root as u'/'
         if not util.is_str(path):
-            _logger.warning("Got non-native PATH_INFO: {!r}".format(path))
+            _logger.warning(f"Got non-native PATH_INFO: {path!r}")
             # path = path.encode("utf8")
             path = util.to_str(path)
 
@@ -476,7 +476,7 @@ def _start_response_wrapper(status, response_headers, exc_info=None):
             headerDict = {}
             for header, value in response_headers:
                 if header.lower() in headerDict:
-                    _logger.error("Duplicate header in response: {}".format(header))
+                    _logger.error(f"Duplicate header in response: {header}")
                 headerDict[header.lower()] = value
 
             # Check if we should close the connection after this request.