Consolidate parallel deletion, RTLO prevention, and .env auto-fix from PRs #146, #149, #163, #164, #166

.jules/sentinel.md

@@ -0,0 +1,11 @@
+## 2026-02-09 - RTLO/Bidi Spoofing in Folder Names
+
+**Vulnerability:** Input validation for folder names allowed Unicode Bidi control characters (e.g., `\u202e`), enabling Homograph/Spoofing attacks (RTLO - Right-To-Left Override).
+
+**Example Attack:** A folder name like `"safe\u202eexe.pdf"` would render as `"safepdf.exe"` in some terminals and UIs, potentially misleading users about file types or content.
+
+**Learning:** Standard "printable" checks (`isprintable()`) do not block Bidi control characters, which can manipulate text direction and visual presentation.
+
+**Prevention:** Explicitly block all known Bidi control characters (U+202A-U+202E, U+2066-U+2069, U+200E-U+200F) in user-visible strings. Also block path separators (/, \) to prevent confusion.
+
+**Implementation:** Pre-compiled character sets at module level for performance, tested comprehensively for all 11 blocked Bidi characters.

main.py

@@ -104,39 +104,57 @@
 
 def check_env_permissions(env_path: str = ".env") -> None:
     """
-    Check .env file permissions and warn if readable by others.
+    Check .env file permissions and auto-fix if readable by others.
+
+    Security: Automatically sets permissions to 600 (owner read/write only)
+    if the file is world-readable. This prevents other users on the system
+    from stealing secrets stored in .env files.
 
     Args:
         env_path: Path to the .env file to check (default: ".env")
     """
     if not os.path.exists(env_path):
         return
 
+    # Windows doesn't have Unix permissions
+    if os.name == "nt":
+        # Just warn on Windows, can't auto-fix
+        sys.stderr.write(
+            f"{Colors.WARNING}⚠️  Security Warning: "
+            f"Please ensure {env_path} is only readable by you.{Colors.ENDC}\n"
+        )
+        return
+
     try:
         file_stat = os.stat(env_path)
         # Check if group or others have any permission
         if file_stat.st_mode & (stat.S_IRWXG | stat.S_IRWXO):
-            platform_hint = (
-                "Please secure your .env file so it is only readable by " "the owner."
-            )
-            if os.name != "nt":
-                platform_hint += " For example: 'chmod 600 .env'."
             perms = format(stat.S_IMODE(file_stat.st_mode), "03o")
-            sys.stderr.write(
-                f"{Colors.WARNING}⚠️  Security Warning: .env file is "
-                f"readable by others ({perms})! {platform_hint}"
-                f"{Colors.ENDC}\n"
-            )
-    except Exception as error:
+
+            # Auto-fix: Set to 600 (owner read/write only)
+            try:
+                os.chmod(env_path, 0o600)
+                sys.stderr.write(
+                    f"{Colors.GREEN}✓ Fixed {env_path} permissions "
+                    f"(was {perms}, now set to 600){Colors.ENDC}\n"
+                )
+            except OSError as fix_error:
+                # Auto-fix failed, show warning with instructions
+                sys.stderr.write(
+                    f"{Colors.WARNING}⚠️  Security Warning: {env_path} is "
+                    f"readable by others ({perms})! Auto-fix failed: {fix_error}. "
+                    f"Please run: chmod 600 {env_path}{Colors.ENDC}\n"
+                )
+    except OSError as error:
+        # More specific exception type as suggested by bot review
         exception_type = type(error).__name__
         sys.stderr.write(
-            f"{Colors.WARNING}⚠️  Security Warning: Could not check .env "
+            f"{Colors.WARNING}⚠️  Security Warning: Could not check {env_path} "
             f"permissions ({exception_type}: {error}){Colors.ENDC}\n"
         )
 
 
-# SECURITY: Check .env permissions (after Colors is defined for NO_COLOR support)
-check_env_permissions()
+# SECURITY: Check .env permissions will be called in main() to avoid side effects at import time
 log = logging.getLogger("control-d-sync")
 
 # --------------------------------------------------------------------------- #
@@ -149,6 +167,32 @@
 PROFILE_ID_PATTERN = re.compile(r"^[a-zA-Z0-9_-]+$")
 RULE_PATTERN = re.compile(r"^[a-zA-Z0-9.\-_:*\/]+$")
 
+# Parallel processing configuration
+DELETE_WORKERS = 3  # Conservative for DELETE operations due to rate limits
+
+# Security: Dangerous characters for folder names
+# XSS and HTML injection characters
+_DANGEROUS_FOLDER_CHARS = set("<>\"'`")
+# Path separators (prevent confusion and directory traversal attempts)
+_DANGEROUS_FOLDER_CHARS.update(["/", "\\"])
+
+# Security: Unicode Bidi control characters (prevent RTLO/homograph attacks)
+# These characters can be used to mislead users about file extensions or content
+# See: https://en.wikipedia.org/wiki/Right-to-left_override
+_BIDI_CONTROL_CHARS = {
+    "\u202a",  # LEFT-TO-RIGHT EMBEDDING (LRE)
+    "\u202b",  # RIGHT-TO-LEFT EMBEDDING (RLE)
+    "\u202c",  # POP DIRECTIONAL FORMATTING (PDF)
+    "\u202d",  # LEFT-TO-RIGHT OVERRIDE (LRO)
+    "\u202e",  # RIGHT-TO-LEFT OVERRIDE (RLO) - primary attack vector
+    "\u2066",  # LEFT-TO-RIGHT ISOLATE (LRI)
+    "\u2067",  # RIGHT-TO-LEFT ISOLATE (RLI)
+    "\u2068",  # FIRST STRONG ISOLATE (FSI)
+    "\u2069",  # POP DIRECTIONAL ISOLATE (PDI)
+    "\u200e",  # LEFT-TO-RIGHT MARK (LRM) - defense in depth
+    "\u200f",  # RIGHT-TO-LEFT MARK (RLM) - defense in depth
+}
+
 # Pre-compiled patterns for log sanitization
 _BASIC_AUTH_PATTERN = re.compile(r"://[^/@]+@")
 _SENSITIVE_PARAM_PATTERN = re.compile(
@@ -445,7 +489,7 @@
             if not PROFILE_ID_PATTERN.match(profile_id):
                 log.error("Invalid profile ID format (contains unsafe characters)")
             elif len(profile_id) > 64:
                 log.error("Invalid profile ID length (max 64 chars)")
         return False
     return True
 
@@ -468,16 +512,20 @@
 
 def is_valid_folder_name(name: str) -> bool:
     """
-    Validates folder name to prevent XSS and ensure printability.
-    Allowed: Anything printable except < > " ' `
+    Validates folder name to prevent XSS, path traversal, and homograph attacks.
+
+    Blocks:
+    - XSS/HTML injection characters: < > " ' `
+    - Path separators: / \\
+    - Unicode Bidi control characters (RTLO spoofing)
+    - Empty or whitespace-only names
+    - Non-printable characters
     """
     if not name or not name.strip() or not name.isprintable():
         return False
 
-    # Block XSS and HTML injection characters
-    # Allow: ( ) [ ] { } for folder names (e.g. "Work (Private)")
-    dangerous_chars = set("<>\"'`")
-    if any(c in dangerous_chars for c in name):
+    # Check for dangerous characters (pre-compiled at module level for performance)
+    if any(c in _DANGEROUS_FOLDER_CHARS or c in _BIDI_CONTROL_CHARS for c in name):
         return False
 
     return True
@@ -516,7 +564,7 @@
         )
         return False
 
     return True
 
 
 def _api_get(client: httpx.Client, url: str) -> httpx.Response:
@@ -631,7 +679,7 @@
     except httpx.HTTPStatusError as e:
         code = e.response.status_code
         if code == 401:
             log.critical(
                 f"{Colors.FAIL}❌ Authentication Failed: The API Token is invalid.{Colors.ENDC}"
             )
             log.critical(
@@ -687,7 +735,7 @@
         except Exception as e:
             # We log error but don't stop the whole process;
             # individual folder failure shouldn't crash the sync
             log.warning(f"Error fetching rules for folder {folder_id}: {e}")
             return []
 
     try:
@@ -724,7 +772,7 @@
                 except Exception as e:
                     folder_id = future_to_folder[future]
                     log.warning(f"Failed to fetch rules for folder ID {folder_id}: {e}")
 
         log.info(f"Total existing rules across all folders: {len(all_rules)}")
         return all_rules
     except Exception as e:
@@ -1183,16 +1231,40 @@
             existing_folders = list_existing_folders(client, profile_id)
             if not no_delete:
                 deletion_occurred = False
+
+                # Identify folders to delete
+                folders_to_delete = []
                 for folder_data in folder_data_list:
                     name = folder_data["group"]["group"].strip()
                     if name in existing_folders:
-                        # Optimization: Maintain local state of folders to avoid re-fetching
-                        # delete_folder returns True on success
-                        if delete_folder(
-                            client, profile_id, name, existing_folders[name]
-                        ):
-                            del existing_folders[name]
-                            deletion_occurred = True
+                        folders_to_delete.append((name, existing_folders[name]))
+
+                if folders_to_delete:
+                    # Parallel delete to speed up the "clean slate" phase
+                    # Using DELETE_WORKERS (3) for balance between speed and rate limits
+                    with concurrent.futures.ThreadPoolExecutor(
+                        max_workers=DELETE_WORKERS
+                    ) as delete_executor:
+                        future_to_name = {
+                            delete_executor.submit(
+                                delete_folder, client, profile_id, name, folder_id
+                            ): name
+                            for name, folder_id in folders_to_delete
+                        }
+
+                        for future in concurrent.futures.as_completed(future_to_name):
+                            name = future_to_name[future]
+                            try:
+                                if future.result():
+                                    del existing_folders[name]
+                                    deletion_occurred = True
+                            except Exception as exc:
+                                # Sanitize both name and exception to prevent log injection
+                                log.error(
+                                    "Failed to delete folder %s: %s",
+                                    sanitize_for_log(name),
+                                    sanitize_for_log(exc),
+                                )
 
                 # CRITICAL FIX: Increased wait time for massive folders to clear
                 if deletion_occurred:
@@ -1264,6 +1336,9 @@
 
 
 def main():
+    # SECURITY: Check .env permissions (after Colors is defined for NO_COLOR support)
+    check_env_permissions()
+
     global TOKEN
     args = parse_args()
     profiles_arg = (

tests/test_env_permissions.py

@@ -0,0 +1,191 @@
+"""Tests for .env file permissions checking and auto-fix functionality."""
+
+import os
+import stat
+from unittest.mock import MagicMock, patch
+
+
+# Set TOKEN before importing main to avoid issues with load_dotenv()
+os.environ.setdefault("TOKEN", "test-token-123")
+os.environ.setdefault("NO_COLOR", "1")
+
+
+def test_env_permissions_auto_fix_success():
+    """Test successful auto-fix of insecure .env permissions."""
+    # Import here to avoid side effects during test collection
+    from main import check_env_permissions
+
+    # Set up POSIX environment
+    with patch("os.name", "posix"), \
+         patch("os.path.exists", return_value=True):
+
+        # Mock file with insecure permissions (644 = world-readable)
+        mock_stat_result = MagicMock()
+        mock_stat_result.st_mode = stat.S_IFREG | 0o644
+
+        with patch("os.stat", return_value=mock_stat_result):
+            # Mock chmod to succeed
+            chmod_calls = []
+
+            def mock_chmod(path, mode):
+                chmod_calls.append((path, mode))
+
+            with patch("os.chmod", side_effect=mock_chmod):
+                # Mock stderr
+                mock_stderr = MagicMock()
+                with patch("sys.stderr", mock_stderr):
+                    check_env_permissions()
+
+                    # Verify chmod was called with 600
+                    assert len(chmod_calls) == 1
+                    assert chmod_calls[0] == (".env", 0o600)
+
+                    # Verify success message was written
+                    mock_stderr.write.assert_called()
+                    output = "".join(call.args[0] for call in mock_stderr.write.call_args_list)
+                    assert "Fixed .env permissions" in output
+                    assert "644" in output
+                    assert "600" in output
+
+
+def test_env_permissions_auto_fix_failure():
+    """Test warning when auto-fix fails."""
+    # Import here to avoid side effects during test collection
+    from main import check_env_permissions
+
+    with patch("os.name", "posix"), \
+         patch("os.path.exists", return_value=True):
+
+        # Mock file with insecure permissions
+        mock_stat_result = MagicMock()
+        mock_stat_result.st_mode = stat.S_IFREG | 0o666
+
+        with patch("os.stat", return_value=mock_stat_result):
+            # Mock chmod to fail
+            with patch("os.chmod", side_effect=OSError("Permission denied")):
+                # Mock stderr
+                mock_stderr = MagicMock()
+                with patch("sys.stderr", mock_stderr):
+                    check_env_permissions()
+
+                    # Verify warning was written
+                    mock_stderr.write.assert_called()
+                    output = "".join(call.args[0] for call in mock_stderr.write.call_args_list)
+                    assert "Security Warning" in output
+                    assert "Auto-fix failed" in output
+                    assert "chmod 600 .env" in output
+
+
+def test_env_permissions_windows_warning():
+    """Test that Windows shows a warning (no auto-fix)."""
+    # Import here to avoid side effects during test collection
+    from main import check_env_permissions
+
+    with patch("os.name", "nt"), \
+         patch("os.path.exists", return_value=True):
+
+        # Mock stderr
+        mock_stderr = MagicMock()
+        with patch("sys.stderr", mock_stderr):
+            check_env_permissions()
+
+            # Verify warning was written (not a fix message)
+            mock_stderr.write.assert_called_once()
+            output = mock_stderr.write.call_args[0][0]
+            assert "Security Warning" in output
+            assert "Please ensure .env is only readable by you" in output
+
+
+def test_env_permissions_secure_file_no_output():
+    """Test that secure permissions don't trigger any output."""
+    # Import here to avoid side effects during test collection
+    from main import check_env_permissions
+
+    with patch("os.name", "posix"), \
+         patch("os.path.exists", return_value=True):
+
+        # Mock file with secure permissions (600)
+        mock_stat_result = MagicMock()
+        mock_stat_result.st_mode = stat.S_IFREG | 0o600
+
+        with patch("os.stat", return_value=mock_stat_result):
+            # Mock stderr
+            mock_stderr = MagicMock()
+            with patch("sys.stderr", mock_stderr):
+                check_env_permissions()
+
+                # Verify no output
+                mock_stderr.write.assert_not_called()
+
+
+def test_env_permissions_file_not_exists():
+    """Test that missing .env file is silently ignored."""
+    # Import here to avoid side effects during test collection
+    from main import check_env_permissions
+
+    with patch("os.path.exists", return_value=False):
+        # Mock stderr
+        mock_stderr = MagicMock()
+        with patch("sys.stderr", mock_stderr):
+            check_env_permissions()
+
+            # Verify no output
+            mock_stderr.write.assert_not_called()
+
+
+def test_env_permissions_stat_error():
+    """Test handling of os.stat errors."""
+    # Import here to avoid side effects during test collection
+    from main import check_env_permissions
+
+    with patch("os.name", "posix"), \
+         patch("os.path.exists", return_value=True), \
+         patch("os.stat", side_effect=OSError("Access denied")):
+
+        # Mock stderr
+        mock_stderr = MagicMock()
+        with patch("sys.stderr", mock_stderr):
+            check_env_permissions()
+
+            # Verify error message
+            mock_stderr.write.assert_called()
+            output = "".join(call.args[0] for call in mock_stderr.write.call_args_list)
+            assert "Could not check .env permissions" in output
+            assert "OSError" in output
+
+
+def test_env_permissions_respects_custom_path():
+    """Test that custom .env path is respected."""
+    # Import here to avoid side effects during test collection
+    from main import check_env_permissions
+
+    with patch("os.name", "posix"), \
+         patch("os.path.exists", return_value=True):
+
+        # Mock file with insecure permissions
+        mock_stat_result = MagicMock()
+        mock_stat_result.st_mode = stat.S_IFREG | 0o644
+
+        stat_calls = []
+
+        def mock_stat(path, **kwargs):
+            stat_calls.append(path)
+            return mock_stat_result
+
+        with patch("os.stat", side_effect=mock_stat):
+            chmod_calls = []
+
+            def mock_chmod(path, mode):
+                chmod_calls.append((path, mode))
+
+            with patch("os.chmod", side_effect=mock_chmod):
+                # Mock stderr
+                mock_stderr = MagicMock()
+                with patch("sys.stderr", mock_stderr):
+                    check_env_permissions("/custom/.env")
+
+                    # Verify custom path was used
+                    assert "/custom/.env" in stat_calls
+                    assert len(chmod_calls) == 1
+                    assert chmod_calls[0][0] == "/custom/.env"
+