feat: add safe filename handling for backup and sync processes to prevent invalid paths

app/utils/backup.py

@@ -1,5 +1,7 @@
 import os
 import zipfile
+import hashlib
+import re
 from datetime import datetime
 from typing import List
 from app.utils.logger import logger
@@ -19,6 +21,28 @@ BACKUP_DIR = os.path.abspath(
 )
 
 
+def _safe_zip_entry_name(name: str, extension: str = ".m3u8") -> str:
+    """Return a safe zip entry filename.
+
+    Prevents zip-slip style paths and avoids problematic characters.
+    """
+
+    original = (name or "").strip()
+    base = os.path.basename(original)
+    base = re.sub(r"[\x00-\x1f\x7f]", "_", base)
+    invalid = set('<>:"/\\|?*')
+    cleaned = "".join(("_" if ch in invalid else ch) for ch in base).strip().strip(". ")
+    if not cleaned:
+        cleaned = "playlist"
+
+    cleaned = cleaned[:160].rstrip().strip(". ")
+    if cleaned != original:
+        digest = hashlib.sha1(original.encode("utf-8", errors="ignore")).hexdigest()[:8]
+        cleaned = f"{cleaned}__{digest}"
+
+    return f"{cleaned}{extension}"
+
+
 def ensure_backup_dir():
     """Ensure the backup directory exists."""
     if not os.path.exists(BACKUP_DIR):
@@ -118,7 +142,7 @@ def backup_local_playlists(local_path: str) -> str | None:
                 
                 # Get the playlist name without extension and add .m3u8 extension
                 playlist_name = os.path.splitext(entry.name)[0]
-                archive_name = f"{playlist_name}.m3u8"
+                archive_name = _safe_zip_entry_name(playlist_name)
                 
                 # Write to zip
                 zipf.writestr(archive_name, content)
@@ -217,7 +241,7 @@ def backup_cloud_playlists(library_name: str) -> str | None:
                     
                     if len(lines) > 1:  # More than just #EXTM3U
                         content = "\n".join(lines)
-                        archive_name = f"{playlist.title}.m3u8"
+                        archive_name = _safe_zip_entry_name(getattr(playlist, "title", "playlist"))
                         zipf.writestr(archive_name, content)
                         playlist_count += 1
                         

app/utils/sync_manager.py

@@ -2,6 +2,8 @@ import threading
 import asyncio
 import json
 import os
+import hashlib
+import re
 from datetime import datetime
 from app.utils.logger import logger
 from app.utils.playlist_merge import sync_all_playlists, SyncMode
@@ -139,7 +141,7 @@ class SyncManager:
                     local_result_path = os.path.join(output_dir, "outputs", "local_result.m3u8")
                     if os.path.exists(local_result_path):
                         tracks = load_local_playlist(local_result_path)
-                        dest_path = os.path.join(server_config.local_path, f"{playlist_name}.m3u8")
+                        dest_path = self._safe_local_playlist_path(server_config.local_path, playlist_name, ".m3u8")
                         # Ensure directory exists
                         os.makedirs(os.path.dirname(dest_path), exist_ok=True)
                         write_local_playlist(dest_path, tracks)
@@ -156,10 +158,10 @@ class SyncManager:
 
                 elif action == "deleted":
                     # Delete Local
-                    dest_path = os.path.join(server_config.local_path, f"{playlist_name}.m3u8")
+                    dest_path = self._safe_local_playlist_path(server_config.local_path, playlist_name, ".m3u8")
                     delete_local_playlist(dest_path)
                     # Also check for .m3u
-                    dest_path_m3u = os.path.join(server_config.local_path, f"{playlist_name}.m3u")
+                    dest_path_m3u = self._safe_local_playlist_path(server_config.local_path, playlist_name, ".m3u")
                     delete_local_playlist(dest_path_m3u)
                     
                     # Delete Remote
@@ -167,6 +169,54 @@ class SyncManager:
             except Exception as e:
                 logger.error(f"Error applying sync result for playlist {playlist_name}: {e}")
 
+    @staticmethod
+    def _safe_local_playlist_path(local_dir: str, playlist_name: str, extension: str) -> str:
+        base_dir = os.path.abspath(local_dir or "")
+        if not base_dir:
+            raise ValueError("Local playlist directory is not configured")
+
+        original = (playlist_name or "").strip()
+        # Drop any path components.
+        name = os.path.basename(original)
+        # Remove control chars.
+        name = re.sub(r"[\x00-\x1f\x7f]", "_", name)
+        # Replace path separators and Windows-invalid characters.
+        invalid = set('<>:"/\\|?*')
+        cleaned = "".join(("_" if ch in invalid else ch) for ch in name).strip().strip(". ")
+
+        windows_reserved = {
+            "CON", "PRN", "AUX", "NUL",
+            *(f"COM{i}" for i in range(1, 10)),
+            *(f"LPT{i}" for i in range(1, 10)),
+        }
+
+        needs_hash = False
+        if not cleaned:
+            cleaned = "playlist"
+            needs_hash = True
+        if cleaned.upper() in windows_reserved:
+            needs_hash = True
+        if cleaned != original:
+            needs_hash = True
+
+        cleaned = cleaned[:160].rstrip().strip(". ")
+        if not cleaned:
+            cleaned = "playlist"
+            needs_hash = True
+
+        if needs_hash:
+            digest = hashlib.sha1(original.encode("utf-8", errors="ignore")).hexdigest()[:8]
+            cleaned = f"{cleaned}__{digest}"
+
+        filename = f"{cleaned}{extension}"
+        candidate = os.path.abspath(os.path.join(base_dir, filename))
+
+        # Ensure the final path stays within base_dir.
+        if os.path.commonpath([base_dir, candidate]) != base_dir:
+            raise ValueError("Refusing to write outside local playlist directory")
+
+        return candidate
+
     def _complete_sync(self, status, error=None):
         with self._lock:
             self._last_status = status