fix fs permission under windows

b2sdk/_internal/scan/folder.py

@@ -108,6 +108,20 @@ def join_b2_path(relative_dir_path: str | Path, file_name: str):
         return relative_dir_path + '/' + file_name
 
 
+if sys.platform == 'win32':
+
+    def _file_read_access(path):
+        try:
+            with open(path, 'rb', buffering=0):
+                return True
+        except PermissionError:
+            return False
+else:
+
+    def _file_read_access(path):
+        return os.access(path, os.R_OK)
+
+
 class LocalFolder(AbstractFolder):
     """
     Folder interface to a directory on the local machine.
@@ -261,7 +275,9 @@ def _walk_relative_paths(
             try:
                 is_dir = local_path.is_dir()
             except PermissionError:  # `chmod -x dir` can trigger this
-                if reporter is not None:
+                if reporter is not None and not policies_manager.should_exclude_local_directory(
+                    str(relative_file_path)
+                ):
                     reporter.local_permission_error(str(local_path))
                 continue
 
@@ -292,7 +308,7 @@ def _walk_relative_paths(
                 if policies_manager.should_exclude_local_path(local_scan_path):
                     continue  # Skip excluded files
 
-                if not os.access(local_path, os.R_OK):
+                if not _file_read_access(local_path):
                     if reporter is not None:
                         reporter.local_permission_error(str(local_path))
                         continue

b2sdk/_internal/scan/report.py

@@ -10,6 +10,7 @@
 from __future__ import annotations
 
 import logging
+import re
 import threading
 import time
 from dataclasses import dataclass
@@ -20,6 +21,21 @@
 
 logger = logging.getLogger(__name__)
 
+_REMOVE_EXTENDED_PATH_PREFIX = re.compile(r'\\\\\?\\')
+
+
+def _safe_path_print(path: str) -> str:
+    """
+    Print a path, escaping control characters if necessary.
+
+    Windows extended path prefix is removed from the path before printing for better readability.
+    Since Windows 10 the prefix is not needed.
+
+    :param path: a path to print
+    :return: a path that can be printed
+    """
+    return escape_control_chars(_REMOVE_EXTENDED_PATH_PREFIX.sub('', path))
+
 
 @dataclass
 class ProgressReport:
@@ -180,7 +196,7 @@ def local_access_error(self, path: str) -> None:
         :param path: file path
         """
         self.warnings.append(
-            f'WARNING: {escape_control_chars(path)} could not be accessed (broken symlink?)'
+            f'WARNING: {_safe_path_print(path)} could not be accessed (broken symlink?)'
         )
 
     def local_permission_error(self, path: str) -> None:
@@ -190,7 +206,7 @@ def local_permission_error(self, path: str) -> None:
         :param path: file path
         """
         self.warnings.append(
-            f'WARNING: {escape_control_chars(path)} could not be accessed (no permissions to read?)'
+            f'WARNING: {_safe_path_print(path)} could not be accessed (no permissions to read?)'
         )
 
     def symlink_skipped(self, path: str) -> None:
@@ -203,7 +219,7 @@ def circular_symlink_skipped(self, path: str) -> None:
         :param path: file path
         """
         self.warnings.append(
-            f'WARNING: {escape_control_chars(path)} is a circular symlink, which was already visited. Skipping.'
+            f'WARNING: {_safe_path_print(path)} is a circular symlink, which was already visited. Skipping.'
         )
 
     def invalid_name(self, path: str, error: str) -> None:
@@ -213,7 +229,7 @@ def invalid_name(self, path: str, error: str) -> None:
         :param path: file path
         """
         self.warnings.append(
-            f'WARNING: {escape_control_chars(path)} path contains invalid name ({error}). Skipping.'
+            f'WARNING: {_safe_path_print(path)} path contains invalid name ({error}). Skipping.'
         )
 
 

pyproject.toml

@@ -30,8 +30,6 @@ classifiers = [
     "Programming Language :: Python :: 3.12",
 ]
 
-[project.optional-dependencies]
-
 [project.urls]
 Homepage = "https://github.com/Backblaze/b2-sdk-python"
 
@@ -174,6 +172,7 @@ test = [
     # remove `and platform_python_implementation!='PyPy'` after dropping Python 3.7 support as that
     # will allow us to update pydantic to a version which installs properly under PyPy
     "pydantic>=2.0.1,<3; python_version>='3.8' and platform_python_implementation!='PyPy'",
+    "pywin32>=306; sys_platform == \"win32\"",
 ]
 release = [
     "towncrier==23.11.0; python_version>='3.8'",

test/unit/conftest.py

@@ -15,6 +15,13 @@
 from glob import glob
 from pathlib import Path
 
+try:
+    import ntsecuritycon
+    import win32api
+    import win32security
+except ImportError:
+    ntsecuritycon = win32api = win32security = None
+
 import pytest
 
 pytest.register_assert_rewrite('test.unit')
@@ -195,26 +202,84 @@ def file_info():
     return {'key': 'value'}
 
 
+class PermTool:
+    def allow_access(self, path):
+        pass
+
+    def deny_access(self, path):
+        pass
+
+
+class UnixPermTool(PermTool):
+    def allow_access(self, path):
+        path.chmod(0o700)
+
+    def deny_access(self, path):
+        path.chmod(0o000)
+
+
+class WindowsPermTool(PermTool):
+    def __init__(self):
+        self.user_sid = win32security.GetTokenInformation(
+            win32security.OpenProcessToken(win32api.GetCurrentProcess(), win32security.TOKEN_QUERY),
+            win32security.TokenUser
+        )[0]
+
+    def allow_access(self, path):
+        dacl = win32security.ACL()
+        dacl.AddAccessAllowedAce(
+            win32security.ACL_REVISION, ntsecuritycon.FILE_ALL_ACCESS, self.user_sid
+        )
+
+        security_desc = win32security.GetFileSecurity(
+            str(path), win32security.DACL_SECURITY_INFORMATION
+        )
+        security_desc.SetSecurityDescriptorDacl(1, dacl, 0)
+        win32security.SetFileSecurity(
+            str(path), win32security.DACL_SECURITY_INFORMATION, security_desc
+        )
+
+    def deny_access(self, path):
+        dacl = win32security.ACL()
+        dacl.AddAccessDeniedAce(
+            win32security.ACL_REVISION, ntsecuritycon.FILE_ALL_ACCESS, self.user_sid
+        )
+
+        security_desc = win32security.GetFileSecurity(
+            str(path), win32security.DACL_SECURITY_INFORMATION
+        )
+        security_desc.SetSecurityDescriptorDacl(1, dacl, 0)
+        win32security.SetFileSecurity(
+            str(path), win32security.DACL_SECURITY_INFORMATION, security_desc
+        )
+
+
 @pytest.fixture
-def tmp_path_permission_cleanup(tmp_path):
+def fs_perm_tool(tmp_path):
     """
     Ensure tmp_path is delete-able after the test.
 
     Important for the tests that mess with filesystem permissions.
     """
-    yield
+    if os.name == 'nt':
+        if win32api is None:
+            pytest.skip('pywin32 is required to run this test')
+        perm_tool = WindowsPermTool()
+    else:
+        perm_tool = UnixPermTool()
+    yield perm_tool
 
     try:
         shutil.rmtree(tmp_path)
     except OSError:
-        tmp_path.chmod(0o700)
+        perm_tool.allow_access(tmp_path)
 
         for root, dirs, files in os.walk(tmp_path, topdown=True):
             for name in dirs:
-                (Path(root) / name).chmod(0o700)
+                perm_tool.allow_access(Path(root) / name)
             for name in files:
                 file_path = Path(root) / name
-                file_path.chmod(0o600)
+                perm_tool.allow_access(file_path)
                 file_path.unlink()
 
         for root, dirs, files in os.walk(tmp_path, topdown=False):

test/unit/scan/test_folder_traversal.py

@@ -11,8 +11,10 @@
 
 import codecs
 import os
+import pathlib
 import platform
 import re
+import shlex
 import sys
 from unittest.mock import MagicMock, patch
 
@@ -686,7 +688,7 @@ def test_excluded_no_access_check(self, tmp_path):
         platform.system() == 'Windows',
         reason="Unix-only filesystem permissions are tested",
     )
-    def test_dir_without_exec_permission(self, tmp_path, tmp_path_permission_cleanup):
+    def test_dir_without_exec_permission(self, tmp_path, fs_perm_tool):
         """Test that a excluded directory/file without permissions emits warnings."""
         no_perm_dir = tmp_path / "no_perm_dir"
         no_perm_dir.mkdir()
@@ -705,13 +707,13 @@ def test_dir_without_exec_permission(self, tmp_path, tmp_path_permission_cleanup
 
         # Check that no access warnings are issued for the excluded directory/file
         assert set(reporter.warnings) == {
-            f'WARNING: {tmp_path}/no_perm_dir/file.txt could not be accessed (no permissions to read?)',
-            f'WARNING: {tmp_path}/no_perm_dir/file2.txt could not be accessed (no permissions to read?)',
+            f'WARNING: {tmp_path/"no_perm_dir/file.txt"} could not be accessed (no permissions to read?)',
+            f'WARNING: {tmp_path/"no_perm_dir/file2.txt"} could not be accessed (no permissions to read?)',
         }
 
         reporter.close()
 
-    def test_without_permissions(self, tmp_path, tmp_path_permission_cleanup):
+    def test_without_permissions(self, tmp_path, fs_perm_tool):
         """Test that a excluded directory/file without permissions emits warnings."""
         no_perm_dir = tmp_path / "no_perm_dir"
         no_perm_dir.mkdir()
@@ -723,28 +725,32 @@ def test_without_permissions(self, tmp_path, tmp_path_permission_cleanup):
         (included_dir / "included_file.txt").touch()
 
         # Modify directory permissions to simulate lack of access
-        (included_dir / "no_perm_file.txt").chmod(0o000)
-        no_perm_dir.chmod(0o000)
+        fs_perm_tool.deny_access(included_dir / "no_perm_file.txt")
+        fs_perm_tool.deny_access(no_perm_dir)
 
         scan_policy = ScanPoliciesManager()
         reporter = ProgressReport(sys.stdout, False)
 
         folder = LocalFolder(str(tmp_path))
         local_paths = folder.all_files(reporter=reporter, policies_manager=scan_policy)
-        absolute_paths = [path.absolute_path for path in local_paths]
+        absolute_paths = [pathlib.Path(path.absolute_path) for path in local_paths]
 
         # Check that only included_dir/included_file.txt was return
-        assert any('included_file.txt' in path for path in absolute_paths)
+        assert {path.name for path in absolute_paths} == {"included_file.txt"}
+
+        def s(p):
+            # shlex.quote works differently depending if its on windows or unix
+            return shlex.quote(str(p))
 
         # Check that no access warnings are issued for the excluded directory/file
         assert set(reporter.warnings) == {
-            f'WARNING: {tmp_path}/no_perm_dir could not be accessed (no permissions to read?)',
-            f'WARNING: {tmp_path}/included_dir/no_perm_file.txt could not be accessed (no permissions to read?)'
+            f'WARNING: {s(tmp_path / "no_perm_dir")} could not be accessed (no permissions to read?)',
+            f'WARNING: {s(tmp_path / "included_dir/no_perm_file.txt")} could not be accessed (no permissions to read?)'
         }
 
         reporter.close()
 
-    def test_excluded_without_permissions(self, tmp_path, tmp_path_permission_cleanup):
+    def test_excluded_without_permissions(self, tmp_path, fs_perm_tool):
         """Test that a excluded directory/file without permissions is not processed and no warning is issued."""
         no_perm_dir = tmp_path / "no_perm_dir"
         no_perm_dir.mkdir()
@@ -756,8 +762,8 @@ def test_excluded_without_permissions(self, tmp_path, tmp_path_permission_cleanu
         (included_dir / "included_file.txt").touch()
 
         # Modify directory permissions to simulate lack of access
-        (included_dir / "no_perm_file.txt").chmod(0o000)
-        no_perm_dir.chmod(0o000)
+        fs_perm_tool.deny_access(included_dir / "no_perm_file.txt")
+        fs_perm_tool.deny_access(no_perm_dir)
 
         scan_policy = ScanPoliciesManager(
             exclude_dir_regexes=[r"no_perm_dir$"], exclude_file_regexes=[r'.*no_perm_file.txt']
@@ -768,10 +774,6 @@ def test_excluded_without_permissions(self, tmp_path, tmp_path_permission_cleanu
         local_paths = folder.all_files(reporter=reporter, policies_manager=scan_policy)
         absolute_paths = [path.absolute_path for path in local_paths]
 
-        # Restore directory permissions to clean up
-        (included_dir / "no_perm_file.txt").chmod(0o755)
-        no_perm_dir.chmod(0o755)
-
         # Check that only included_dir/included_file.txt was return
         assert any('included_file.txt' in path for path in absolute_paths)