# 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.trace("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.trace("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.trace("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.trace("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.trace("path_supports_problematic chars: path=%r result=%s", path, result)
    return result