diff --git a/README.rst b/README.rst index dedc631..51becdb 100644 --- a/README.rst +++ b/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 ------------ diff --git a/src/dvc_objects/fs/system.py b/src/dvc_objects/fs/system.py index c0f0ceb..ce8f200 100644 --- a/src/dvc_objects/fs/system.py +++ b/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() diff --git a/src/dvc_objects/fs/utils.py b/src/dvc_objects/fs/utils.py index 5018e8c..8c94ffd 100644 --- a/src/dvc_objects/fs/utils.py +++ b/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: diff --git a/tests/fs/test_system.py b/tests/fs/test_system.py index 299d93a..6c43e90 100644 --- a/tests/fs/test_system.py +++ b/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") diff --git a/tests/fs/test_utils.py b/tests/fs/test_utils.py index 81c0dad..9dcfa69 100644 --- a/tests/fs/test_utils.py +++ b/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")