528 lines
17 KiB
Python
528 lines
17 KiB
Python
# This file is part of Radicale - CalDAV and CardDAV server
|
|
# Copyright © 2014 Jean-Marc Martins
|
|
# Copyright © 2012-2017 Guillaume Ayoub
|
|
# Copyright © 2017-2022 Unrud <unrud@outlook.com>
|
|
# Copyright © 2025-2026 Peter Bieringer <pb@bieringer.de>
|
|
#
|
|
# 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/>.
|
|
|
|
"""
|
|
Helper functions for working with the file system.
|
|
|
|
"""
|
|
|
|
import errno
|
|
import os
|
|
import pathlib
|
|
import posixpath
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
from tempfile import TemporaryDirectory
|
|
from typing import Iterator, Type, Union
|
|
|
|
from radicale import storage, types, utils
|
|
from radicale.log import logger
|
|
|
|
if sys.platform == "win32":
|
|
import ctypes
|
|
import ctypes.wintypes
|
|
import msvcrt
|
|
|
|
LOCKFILE_EXCLUSIVE_LOCK: int = 2
|
|
ULONG_PTR: Union[Type[ctypes.c_uint32], Type[ctypes.c_uint64]]
|
|
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)]
|
|
|
|
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
|
|
lock_file_ex = 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 = 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
|
|
else:
|
|
import fcntl
|
|
|
|
if sys.platform == "linux":
|
|
import ctypes
|
|
|
|
RENAME_EXCHANGE: int = 2
|
|
renameat2 = None
|
|
try:
|
|
renameat2 = ctypes.CDLL(None, use_errno=True).renameat2
|
|
except AttributeError:
|
|
pass
|
|
else:
|
|
renameat2.argtypes = [
|
|
ctypes.c_int, ctypes.c_char_p,
|
|
ctypes.c_int, ctypes.c_char_p,
|
|
ctypes.c_uint]
|
|
renameat2.restype = ctypes.c_int
|
|
|
|
if sys.platform == "darwin":
|
|
# Definition missing in PyPy
|
|
F_FULLFSYNC: int = getattr(fcntl, "F_FULLFSYNC", 51)
|
|
|
|
|
|
class RwLock:
|
|
"""A readers-Writer lock that locks a file."""
|
|
|
|
_path: str
|
|
_readers: int
|
|
_writer: bool
|
|
_lock: threading.Lock
|
|
|
|
def __init__(self, path: str) -> None:
|
|
self._path = path
|
|
self._readers = 0
|
|
self._writer = False
|
|
self._lock = threading.Lock()
|
|
|
|
@property
|
|
def locked(self) -> str:
|
|
with self._lock:
|
|
if self._readers > 0:
|
|
return "r"
|
|
if self._writer:
|
|
return "w"
|
|
return ""
|
|
|
|
@types.contextmanager
|
|
def acquire(self, mode: str) -> Iterator[None]:
|
|
if mode not in "rw":
|
|
raise ValueError("Invalid mode: %r" % mode)
|
|
with open(self._path, "w+") as lock_file:
|
|
if sys.platform == "win32":
|
|
handle = msvcrt.get_osfhandle(lock_file.fileno())
|
|
flags = LOCKFILE_EXCLUSIVE_LOCK if mode == "w" else 0
|
|
overlapped = Overlapped()
|
|
try:
|
|
if not lock_file_ex(handle, flags, 0, 1, 0, overlapped):
|
|
raise ctypes.WinError()
|
|
except OSError as e:
|
|
raise RuntimeError("Locking the storage failed: %s" % e
|
|
) from e
|
|
else:
|
|
_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
|
|
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 rename_exchange(src: str, dst: str) -> None:
|
|
"""Exchange the files or directories `src` and `dst`.
|
|
|
|
Both `src` and `dst` must exist but may be of different types.
|
|
|
|
On Linux with renameat2 the operation is atomic.
|
|
On other platforms it's not atomic.
|
|
|
|
"""
|
|
src_dir, src_base = os.path.split(src)
|
|
dst_dir, dst_base = os.path.split(dst)
|
|
src_dir = src_dir or os.curdir
|
|
dst_dir = dst_dir or os.curdir
|
|
if not src_base or not dst_base:
|
|
raise ValueError("Invalid arguments: %r -> %r" % (src, dst))
|
|
if sys.platform == "linux" and renameat2:
|
|
src_base_bytes = os.fsencode(src_base)
|
|
dst_base_bytes = os.fsencode(dst_base)
|
|
src_dir_fd = os.open(src_dir, 0)
|
|
try:
|
|
dst_dir_fd = os.open(dst_dir, 0)
|
|
try:
|
|
if renameat2(src_dir_fd, src_base_bytes,
|
|
dst_dir_fd, dst_base_bytes,
|
|
RENAME_EXCHANGE) == 0:
|
|
return
|
|
errno_ = ctypes.get_errno()
|
|
# Fallback if RENAME_EXCHANGE not supported by filesystem
|
|
if errno_ != errno.EINVAL:
|
|
raise OSError(errno_, os.strerror(errno_))
|
|
finally:
|
|
os.close(dst_dir_fd)
|
|
finally:
|
|
os.close(src_dir_fd)
|
|
with TemporaryDirectory(prefix=".Radicale.tmp-", dir=src_dir
|
|
) as tmp_dir:
|
|
os.rename(dst, os.path.join(tmp_dir, "interim"))
|
|
os.rename(src, dst)
|
|
os.rename(os.path.join(tmp_dir, "interim"), src)
|
|
|
|
|
|
def fsync(fd: int) -> None:
|
|
if sys.platform == "darwin":
|
|
try:
|
|
fcntl.fcntl(fd, F_FULLFSYNC)
|
|
return
|
|
except OSError as e:
|
|
# Fallback if F_FULLFSYNC not supported by filesystem
|
|
if e.errno != errno.EINVAL:
|
|
raise
|
|
os.fsync(fd)
|
|
|
|
|
|
def strip_path(path: str) -> str:
|
|
assert sanitize_path(path) == path
|
|
return path.strip("/")
|
|
|
|
|
|
def unstrip_path(stripped_path: str, trailing_slash: bool = False) -> str:
|
|
assert strip_path(sanitize_path(stripped_path)) == stripped_path
|
|
assert stripped_path or trailing_slash
|
|
path = "/%s" % stripped_path
|
|
if trailing_slash and not path.endswith("/"):
|
|
path += "/"
|
|
return path
|
|
|
|
|
|
def sanitize_path(path: str) -> str:
|
|
"""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: str) -> bool:
|
|
"""Check if path is a single component of a path.
|
|
|
|
Check that the path is safe to join too.
|
|
|
|
"""
|
|
return bool(path) and "/" not in path and path not in (".", "..")
|
|
|
|
|
|
def is_safe_filesystem_path_component(path: str) -> bool:
|
|
"""Check if path is a single component of a local and posix filesystem
|
|
path.
|
|
|
|
Check that the path is safe to join too.
|
|
|
|
"""
|
|
return (
|
|
bool(path) and not os.path.splitdrive(path)[0] and
|
|
(sys.platform != "win32" or ":" not in path) and # Block NTFS-ADS
|
|
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: str, sane_path: str, path_is_collision_free: bool = False) -> str:
|
|
"""Convert `sane_path` to a local filesystem path relative to `root`.
|
|
|
|
`root` must be a secure filesystem path, it will be prepend to the path.
|
|
|
|
`sane_path` must be a sanitized path without leading or trailing ``/``.
|
|
|
|
`path_is_collision_free` is a toggle whether it was earlier detected as collision-free
|
|
|
|
Conversion of `sane_path` is done in a secure manner,
|
|
or raises ``ValueError``.
|
|
|
|
"""
|
|
# logger.trace("path_to_filesystem root=%r sane_path=%r path_is_collision_free=%s", root, sane_path, path_is_collision_free)
|
|
assert sane_path == strip_path(sanitize_path(sane_path))
|
|
safe_path = root
|
|
parts = sane_path.split("/") if sane_path else []
|
|
for part in parts:
|
|
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 not path_is_collision_free:
|
|
if sys.platform == "win32" and False: # temporary for testing
|
|
# logger.trace("path_to_filesystem check (win32): %r", part)
|
|
# if (os.path.lexists(safe_path) and not os.path.realpath(safe_path).endswith(part)) and not os.path.islink(safe_path):
|
|
if (os.path.lexists(safe_path) and not os.path.realpath(safe_path).endswith(part)):
|
|
raise CollidingPathError(part)
|
|
else:
|
|
# logger.trace("path_to_filesystem check (!win32): %r", part)
|
|
if os.path.lexists(safe_path):
|
|
with os.scandir(safe_path_parent) as entries:
|
|
if part not in (e.name for e in entries):
|
|
raise CollidingPathError(part)
|
|
else:
|
|
# logger.trace("path_to_filesystem check (skipped): %r", part)
|
|
pass
|
|
return safe_path
|
|
|
|
|
|
def parent_path(path: str) -> str:
|
|
return unstrip_path(posixpath.dirname(strip_path(path)), True)
|
|
|
|
|
|
class UnsafePathError(ValueError):
|
|
|
|
def __init__(self, path: str) -> None:
|
|
super().__init__("Can't translate name safely to filesystem: %r" %
|
|
path)
|
|
|
|
|
|
class CollidingPathError(ValueError):
|
|
|
|
def __init__(self, path: str) -> None:
|
|
super().__init__("File name collision: %r" % path)
|
|
|
|
|
|
def name_from_path(path: str, collection: "storage.BaseCollection") -> str:
|
|
"""Return Radicale item name from ``path``."""
|
|
assert sanitize_path(path) == path
|
|
start = unstrip_path(collection.path, True)
|
|
if not (path + "/").startswith(start):
|
|
raise ValueError("%r doesn't start with %r" % (path, start))
|
|
name = path[len(start):]
|
|
if name and not is_safe_path_component(name):
|
|
raise ValueError("%r is not a component in collection %r" %
|
|
(name, collection.path))
|
|
return name
|
|
|
|
|
|
def path_permissions(path):
|
|
path = pathlib.Path(path)
|
|
|
|
try:
|
|
uid = utils.unknown_if_empty(path.stat().st_uid)
|
|
except (KeyError, NotImplementedError):
|
|
uid = "UNKNOWN"
|
|
|
|
try:
|
|
gid = utils.unknown_if_empty(path.stat().st_gid)
|
|
except (KeyError, NotImplementedError):
|
|
gid = "UNKNOWN"
|
|
|
|
try:
|
|
mode = utils.unknown_if_empty("%o" % path.stat().st_mode)
|
|
except (KeyError, NotImplementedError):
|
|
mode = "UNKNOWN"
|
|
|
|
try:
|
|
owner = utils.unknown_if_empty(path.owner())
|
|
except (KeyError, NotImplementedError):
|
|
owner = "UNKNOWN"
|
|
|
|
try:
|
|
group = utils.unknown_if_empty(path.group())
|
|
except (KeyError, NotImplementedError):
|
|
group = "UNKNOWN"
|
|
|
|
return [owner, uid, group, gid, mode]
|
|
|
|
|
|
def path_permissions_as_string(path):
|
|
pp = path_permissions(path)
|
|
s = "path=%r owner=%s(%s) group=%s(%s) mode=%s" % (path, pp[0], pp[1], pp[2], pp[3], pp[4])
|
|
return s
|
|
|
|
|
|
def file_check_size(path: str, limit: int):
|
|
if os.path.isfile(path):
|
|
size = os.stat(path).st_size
|
|
if size > limit:
|
|
logger.warning("file skipped because size exceeds limit %s > %s: %r", utils.format_unit(size, binary=True), utils.format_unit(limit, binary=True), path)
|
|
return False
|
|
return True
|
|
|
|
|
|
def path_supports_symlink(path):
|
|
"""Check whether path supports symlink."""
|
|
if not os.path.isdir(path):
|
|
raise ValueError("%r is not a path" % (path))
|
|
result = True
|
|
test_dir1 = tempfile.mkdtemp(dir=path)
|
|
test_dir2 = tempfile.mkdtemp(dir=path)
|
|
os.rmdir(test_dir2)
|
|
try:
|
|
os.symlink(test_dir1, test_dir2)
|
|
except PermissionError:
|
|
result = False
|
|
else:
|
|
# cleanup
|
|
os.remove(test_dir2)
|
|
finally:
|
|
# cleanup
|
|
os.rmdir(test_dir1)
|
|
return result
|
|
|
|
|
|
def path_is_collision_free_case_sensitive(path):
|
|
"""Check whether path supports case sensitive entries."""
|
|
if not os.path.isdir(path):
|
|
raise ValueError("%r is not a path" % (path))
|
|
base_dir = tempfile.mkdtemp(dir=path)
|
|
test_dir = "TESTDIR"
|
|
test_dir_uc = os.path.join(base_dir, test_dir.upper())
|
|
test_dir_lc = os.path.join(base_dir, test_dir.lower())
|
|
os.mkdir(test_dir_uc)
|
|
result = True
|
|
try:
|
|
os.mkdir(test_dir_lc)
|
|
except FileExistsError:
|
|
result = False
|
|
else:
|
|
# cleanup
|
|
os.rmdir(test_dir_lc)
|
|
# cleanup
|
|
os.rmdir(test_dir_uc)
|
|
os.rmdir(base_dir)
|
|
logger.debug("path_is_collision_free (case-sensitive): path=%r result=%s", path, result)
|
|
return result
|
|
|
|
|
|
def path_is_collision_free_no_short_filename(path):
|
|
"""Check whether path supports short-filename collision-free entries."""
|
|
if not os.path.isdir(path):
|
|
raise ValueError("%r is not a path" % (path))
|
|
base_dir = tempfile.mkdtemp(dir=path)
|
|
test_dir = "TESTDIRLONG"
|
|
test_dir_long = os.path.join(base_dir, test_dir)
|
|
test_dir_short = os.path.join(base_dir, test_dir[:6] + "~1")
|
|
os.mkdir(test_dir_long)
|
|
result = True
|
|
try:
|
|
os.mkdir(test_dir_short)
|
|
except FileExistsError:
|
|
result = False
|
|
else:
|
|
# cleanup
|
|
os.rmdir(test_dir_short)
|
|
# cleanup
|
|
os.rmdir(test_dir_long)
|
|
os.rmdir(base_dir)
|
|
logger.debug("path_is_collision_free (no short-filename): path=%r result=%s", path, result)
|
|
return result
|
|
|
|
|
|
def path_supports_unicode(path):
|
|
"""Check whether path supports unicode."""
|
|
if not os.path.isdir(path):
|
|
raise ValueError("%r is not a path" % (path))
|
|
base_dir = tempfile.mkdtemp(dir=path)
|
|
part = "TEST😀"
|
|
test_dir = os.path.join(base_dir, part)
|
|
result = True
|
|
try:
|
|
os.mkdir(test_dir)
|
|
except OSError:
|
|
result = False
|
|
else:
|
|
with os.scandir(base_dir) as entries:
|
|
if part not in (e.name for e in entries):
|
|
result = False
|
|
# cleanup
|
|
os.rmdir(test_dir)
|
|
# cleanup
|
|
os.rmdir(base_dir)
|
|
logger.debug("path_supports_unicode: path=%r result=%s", path, result)
|
|
return result
|
|
|
|
|
|
def path_supports_trailing_whitespace(path):
|
|
"""Check whether path supports trailing whitespace."""
|
|
if not os.path.isdir(path):
|
|
raise ValueError("%r is not a path" % (path))
|
|
base_dir = tempfile.mkdtemp(dir=path)
|
|
part = "TEST "
|
|
test_dir = os.path.join(base_dir, part)
|
|
result = True
|
|
try:
|
|
os.mkdir(test_dir)
|
|
except OSError:
|
|
result = False
|
|
else:
|
|
with os.scandir(base_dir) as entries:
|
|
if part not in (e.name for e in entries):
|
|
result = False
|
|
# cleanup
|
|
os.rmdir(test_dir)
|
|
# cleanup
|
|
os.rmdir(base_dir)
|
|
logger.debug("path_supports_trailing_whitespace: path=%r result=%s", path, result)
|
|
return result
|
|
|
|
|
|
def path_supports_problematic_chars(path):
|
|
"""Check whether path supports problematic chars."""
|
|
if not os.path.isdir(path):
|
|
raise ValueError("%r is not a path" % (path))
|
|
base_dir = tempfile.mkdtemp(dir=path)
|
|
result = True
|
|
for char in ['*', '?']:
|
|
part = "TES" + char + "T"
|
|
test_dir = os.path.join(base_dir, part)
|
|
try:
|
|
os.mkdir(test_dir)
|
|
except OSError:
|
|
result = False
|
|
else:
|
|
with os.scandir(base_dir) as entries:
|
|
if part not in (e.name for e in entries):
|
|
result = False
|
|
# cleanup
|
|
os.rmdir(test_dir)
|
|
# cleanup
|
|
os.rmdir(base_dir)
|
|
logger.debug("path_supports_problematic chars: path=%r result=%s", path, result)
|
|
return result
|