refactor

radicale/pathutils.py

@@ -0,0 +1,217 @@
+# This file is part of Radicale Server - Calendar Server
+# Copyright © 2014 Jean-Marc Martins
+# Copyright © 2012-2017 Guillaume Ayoub
+# Copyright © 2017-2018 Unrud <unrud@outlook.com>
+#
+# This library is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Radicale.  If not, see <http://www.gnu.org/licenses/>.
+
+import os
+import posixpath
+import threading
+from contextlib import contextmanager
+
+if os.name == "nt":
+    import ctypes
+    import ctypes.wintypes
+    import msvcrt
+
+    LOCKFILE_EXCLUSIVE_LOCK = 2
+    if ctypes.sizeof(ctypes.c_void_p) == 4:
+        ULONG_PTR = ctypes.c_uint32
+    else:
+        ULONG_PTR = ctypes.c_uint64
+
+    class Overlapped(ctypes.Structure):
+        _fields_ = [
+            ("internal", ULONG_PTR),
+            ("internal_high", ULONG_PTR),
+            ("offset", ctypes.wintypes.DWORD),
+            ("offset_high", ctypes.wintypes.DWORD),
+            ("h_event", ctypes.wintypes.HANDLE)]
+
+    lock_file_ex = ctypes.windll.kernel32.LockFileEx
+    lock_file_ex.argtypes = [
+        ctypes.wintypes.HANDLE,
+        ctypes.wintypes.DWORD,
+        ctypes.wintypes.DWORD,
+        ctypes.wintypes.DWORD,
+        ctypes.wintypes.DWORD,
+        ctypes.POINTER(Overlapped)]
+    lock_file_ex.restype = ctypes.wintypes.BOOL
+    unlock_file_ex = ctypes.windll.kernel32.UnlockFileEx
+    unlock_file_ex.argtypes = [
+        ctypes.wintypes.HANDLE,
+        ctypes.wintypes.DWORD,
+        ctypes.wintypes.DWORD,
+        ctypes.wintypes.DWORD,
+        ctypes.POINTER(Overlapped)]
+    unlock_file_ex.restype = ctypes.wintypes.BOOL
+elif os.name == "posix":
+    import fcntl
+
+
+class RwLock:
+    """A readers-Writer lock that locks a file."""
+
+    def __init__(self, path):
+        self._path = path
+        self._readers = 0
+        self._writer = False
+        self._lock = threading.Lock()
+
+    @property
+    def locked(self):
+        with self._lock:
+            if self._readers > 0:
+                return "r"
+            if self._writer:
+                return "w"
+            return ""
+
+    @contextmanager
+    def acquire(self, mode):
+        if mode not in "rw":
+            raise ValueError("Invalid mode: %r" % mode)
+        with open(self._path, "w+") as lock_file:
+            if os.name == "nt":
+                handle = msvcrt.get_osfhandle(lock_file.fileno())
+                flags = LOCKFILE_EXCLUSIVE_LOCK if mode == "w" else 0
+                overlapped = Overlapped()
+                if not lock_file_ex(handle, flags, 0, 1, 0, overlapped):
+                    raise RuntimeError("Locking the storage failed: %s" %
+                                       ctypes.FormatError())
+            elif os.name == "posix":
+                _cmd = fcntl.LOCK_EX if mode == "w" else fcntl.LOCK_SH
+                try:
+                    fcntl.flock(lock_file.fileno(), _cmd)
+                except OSError as e:
+                    raise RuntimeError("Locking the storage failed: %s" %
+                                       e) from e
+            else:
+                raise RuntimeError("Locking the storage failed: "
+                                   "Unsupported operating system")
+            with self._lock:
+                if self._writer or mode == "w" and self._readers != 0:
+                    raise RuntimeError("Locking the storage failed: "
+                                       "Guarantees failed")
+                if mode == "r":
+                    self._readers += 1
+                else:
+                    self._writer = True
+            try:
+                yield
+            finally:
+                with self._lock:
+                    if mode == "r":
+                        self._readers -= 1
+                    self._writer = False
+
+
+def fsync(fd):
+    if os.name == "posix" and hasattr(fcntl, "F_FULLFSYNC"):
+        fcntl.fcntl(fd, fcntl.F_FULLFSYNC)
+    else:
+        os.fsync(fd)
+
+
+def sanitize_path(path):
+    """Make path absolute with leading slash to prevent access to other data.
+
+    Preserve potential trailing slash.
+
+    """
+    trailing_slash = "/" if path.endswith("/") else ""
+    path = posixpath.normpath(path)
+    new_path = "/"
+    for part in path.split("/"):
+        if not is_safe_path_component(part):
+            continue
+        new_path = posixpath.join(new_path, part)
+    trailing_slash = "" if new_path.endswith("/") else trailing_slash
+    return new_path + trailing_slash
+
+
+def is_safe_path_component(path):
+    """Check if path is a single component of a path.
+
+    Check that the path is safe to join too.
+
+    """
+    return path and "/" not in path and path not in (".", "..")
+
+
+def is_safe_filesystem_path_component(path):
+    """Check if path is a single component of a local and posix filesystem
+       path.
+
+    Check that the path is safe to join too.
+
+    """
+    return (
+        path and not os.path.splitdrive(path)[0] and
+        not os.path.split(path)[0] and path not in (os.curdir, os.pardir) and
+        not path.startswith(".") and not path.endswith("~") and
+        is_safe_path_component(path))
+
+
+def path_to_filesystem(root, *paths):
+    """Convert path to a local filesystem path relative to base_folder.
+
+    `root` must be a secure filesystem path, it will be prepend to the path.
+
+    Conversion of `paths` is done in a secure manner, or raises ``ValueError``.
+
+    """
+    paths = [sanitize_path(path).strip("/") for path in paths]
+    safe_path = root
+    for path in paths:
+        if not path:
+            continue
+        for part in path.split("/"):
+            if not is_safe_filesystem_path_component(part):
+                raise UnsafePathError(part)
+            safe_path_parent = safe_path
+            safe_path = os.path.join(safe_path, part)
+            # Check for conflicting files (e.g. case-insensitive file systems
+            # or short names on Windows file systems)
+            if (os.path.lexists(safe_path) and
+                    part not in (e.name for e in
+                                 os.scandir(safe_path_parent))):
+                raise CollidingPathError(part)
+    return safe_path
+
+
+class UnsafePathError(ValueError):
+    def __init__(self, path):
+        message = "Can't translate name safely to filesystem: %r" % path
+        super().__init__(message)
+
+
+class CollidingPathError(ValueError):
+    def __init__(self, path):
+        message = "File name collision: %r" % path
+        super().__init__(message)
+
+
+def name_from_path(path, collection):
+    """Return Radicale item name from ``path``."""
+    path = path.strip("/") + "/"
+    start = collection.path + "/"
+    if not path.startswith(start):
+        raise ValueError("%r doesn't start with %r" % (path, start))
+    name = path[len(start):][:-1]
+    if name and not is_safe_path_component(name):
+        raise ValueError("%r is not a component in collection %r" %
+                         (name, collection.path))
+    return name