fix: preserve read-only hardlink attributes on Windows

README.rst

@@ -66,6 +66,12 @@ Usage
 
 This is used in `dvc`_ and `dvc-data`_.
 
+On Windows, removing a read-only local file preserves the read-only attribute
+shared by its other hardlinks. This requires the filesystem to support extended
+file disposition with the ignore-read-only flag. If that operation is unsupported
+or denied, removal raises an error instead of clearing the shared attribute.
+This also applies to read-only files encountered during directory removal.
+
 Contributing
 ------------
 

src/dvc_objects/fs/system.py

@@ -15,6 +15,70 @@
 has_link = hasattr(os, "link")
 
 
+def unlink_readonly(path: "AnyFSPath") -> None:
+    """Remove a Windows file without changing attributes shared by hardlinks.
+
+    Raise OSError if the OS or filesystem rejects extended file disposition.
+    Clearing the read-only attribute as a fallback would unprotect other links.
+    """
+    if sys.platform == "win32":
+        from ctypes import wintypes
+
+        kernel = ctypes.WinDLL("kernel32", use_last_error=True)
+        create = kernel.CreateFileW
+        create.argtypes = [
+            wintypes.LPCWSTR,
+            wintypes.DWORD,
+            wintypes.DWORD,
+            wintypes.LPVOID,
+            wintypes.DWORD,
+            wintypes.DWORD,
+            wintypes.HANDLE,
+        ]
+        create.restype = wintypes.HANDLE
+        set_info = kernel.SetFileInformationByHandle
+        set_info.argtypes = [
+            wintypes.HANDLE,
+            ctypes.c_int,
+            wintypes.LPVOID,
+            wintypes.DWORD,
+        ]
+        set_info.restype = wintypes.BOOL
+        close = kernel.CloseHandle
+        close.argtypes = [wintypes.HANDLE]
+        close.restype = wintypes.BOOL
+
+        delete_access = 0x00010000
+        share_read_write_delete = 0x7
+        open_existing = 3
+        open_reparse_point = 0x00200000
+        handle = create(
+            os.fsdecode(path),
+            delete_access,
+            share_read_write_delete,
+            None,
+            open_existing,
+            open_reparse_point,
+            None,
+        )
+        if handle == wintypes.HANDLE(-1).value:
+            raise ctypes.WinError(ctypes.get_last_error())
+        try:
+            # POSIX deletion removes this name; IGNORE_READONLY leaves the shared
+            # attribute intact. Do not fall back to chmod if the OS rejects it.
+            disposition_ex = 21
+            flags = wintypes.DWORD(0x1 | 0x2 | 0x10)
+            if not set_info(
+                handle, disposition_ex, ctypes.byref(flags), ctypes.sizeof(flags)
+            ):
+                raise ctypes.WinError(ctypes.get_last_error())
+        finally:
+            if not close(handle) and sys.exc_info()[0] is None:
+                raise ctypes.WinError(ctypes.get_last_error())
+    else:
+        raise OSError(errno.ENOTSUP, "read-only unlink is only supported on Windows")
+
+
 def hardlink(source: "AnyFSPath", link_name: "AnyFSPath") -> None:
     if not has_link:
         # some Python ports (eg: Termux) don't have os.link()

src/dvc_objects/fs/utils.py

@@ -82,6 +82,14 @@ def move(src: "AnyFSPath", dst: "AnyFSPath") -> None:
 
 def _chmod(func, p, excinfo):
     perm = os.lstat(p).st_mode
+    if (
+        os.name == "nt"
+        and func is os.unlink
+        and stat.S_ISREG(perm)
+        and not perm & stat.S_IWRITE
+    ):
+        system.unlink_readonly(p)
+        return
     perm |= stat.S_IWRITE
 
     try:

tests/fs/test_system.py

@@ -1,9 +1,54 @@
+import errno
 import os
 from os import fspath
+from types import SimpleNamespace
 
+import pytest
+
+from dvc_objects.fs import system
 from dvc_objects.fs.system import hardlink, inode, is_hardlink, is_symlink, symlink
 
 
+@pytest.mark.parametrize("platform", ["linux", "darwin"])
+def test_unlink_readonly_unsupported_platform(tmp_path, mocker, platform):
+    path = tmp_path / "file"
+    path.write_bytes(b"keep this file")
+    mocker.patch.object(system, "sys", SimpleNamespace(platform=platform))
+
+    with pytest.raises(OSError, match="only supported on Windows") as error:
+        system.unlink_readonly(path)
+
+    assert error.value.errno == errno.ENOTSUP
+    assert path.read_bytes() == b"keep this file"
+
+
+@pytest.mark.skipif(os.name != "nt", reason="Windows deletion API")
+@pytest.mark.parametrize("failure", ["open", "disposition", "close", "both"])
+def test_unlink_readonly_api_failure(mocker, failure):
+    kernel = mocker.Mock()
+    kernel.CreateFileW.return_value = (
+        system.ctypes.c_void_p(-1).value if failure == "open" else 123
+    )
+    kernel.SetFileInformationByHandle.return_value = failure not in (
+        "disposition",
+        "both",
+    )
+    kernel.CloseHandle.return_value = failure not in ("close", "both")
+    mocker.patch.object(system.ctypes, "WinDLL", return_value=kernel)
+    last_error = mocker.patch.object(
+        system.ctypes, "get_last_error", side_effect=[5, 6]
+    )
+    with pytest.raises(PermissionError) as error:
+        system.unlink_readonly("readonly-file")
+    assert error.value.errno == errno.EACCES
+    last_error.assert_called_once_with()
+    if failure == "open":
+        kernel.SetFileInformationByHandle.assert_not_called()
+        kernel.CloseHandle.assert_not_called()
+    else:
+        kernel.CloseHandle.assert_called_once_with(123)
+
+
 def test_inode(tmp_path):
     file = tmp_path / "foo"
     file.write_text("foo content", encoding="utf8")

tests/fs/test_utils.py

@@ -1,11 +1,79 @@
+import errno
 import filecmp
 import os
 import re
+import stat
 from os import fspath
 
+import pytest
+
 from dvc_objects.fs import utils
 
 
+@pytest.mark.skipif(
+    os.name != "nt", reason="Windows shares attributes across hardlinks"
+)
+@pytest.mark.parametrize("directory", [False, True])
+def test_remove_preserves_readonly_hardlinks(tmp_path, directory):
+    cache = tmp_path / "cache"
+    cache.write_text("original", encoding="utf8")
+    folder = tmp_path / "workspace"
+    folder.mkdir()
+    link = folder / "file"
+    os.link(cache, link)
+    cache.chmod(stat.S_IREAD)
+    try:
+        utils.remove(folder if directory else link)
+        assert not link.exists()
+        assert cache.stat().st_file_attributes & stat.FILE_ATTRIBUTE_READONLY
+        assert cache.read_text(encoding="utf8") == "original"
+        with pytest.raises(PermissionError):
+            cache.write_text("modified", encoding="utf8")
+    finally:
+        cache.chmod(stat.S_IREAD | stat.S_IWRITE)
+
+
+@pytest.mark.skipif(os.name != "nt", reason="Windows read-only deletion")
+@pytest.mark.parametrize("directory", [False, True])
+@pytest.mark.parametrize(
+    "error",
+    [PermissionError("deletion denied"), OSError(errno.ENOTSUP, "unsupported")],
+)
+def test_remove_readonly_failure_preserves_protection(
+    tmp_path, mocker, directory, error
+):
+    cache = tmp_path / "cache"
+    cache.write_text("original", encoding="utf8")
+    folder = tmp_path / "workspace"
+    folder.mkdir()
+    link = folder / "link"
+    os.link(cache, link)
+    cache.chmod(stat.S_IREAD)
+    mocker.patch.object(utils.system, "unlink_readonly", side_effect=error)
+    try:
+        with pytest.raises(OSError, match=re.escape(str(error))) as raised:
+            utils.remove(folder if directory else link)
+        assert raised.value is error
+        assert link.exists()
+        assert cache.stat().st_file_attributes & stat.FILE_ATTRIBUTE_READONLY
+        assert cache.read_text(encoding="utf8") == "original"
+    finally:
+        cache.chmod(stat.S_IREAD | stat.S_IWRITE)
+
+
+@pytest.mark.skipif(os.name != "nt", reason="Windows read-only deletion")
+def test_remove_readonly_file(tmp_path):
+    path = tmp_path / "file"
+    path.write_text("original", encoding="utf8")
+    path.chmod(stat.S_IREAD)
+    try:
+        utils.remove(path)
+        assert not path.exists()
+    finally:
+        if path.exists():
+            path.chmod(stat.S_IREAD | stat.S_IWRITE)
+
+
 def test_tmp_fname():
     file = os.path.join("path", "to", "file")