fix: #274

app.py

@@ -8,7 +8,6 @@ import logging.handlers
 import time
 from pathlib import Path
 import os
-import atexit
 import sys
 import redis
 import socket
@@ -17,10 +16,15 @@ from urllib.parse import urlparse
 # Run DB migrations as early as possible, before importing any routers that may touch DBs
 try:
     from routes.migrations import run_migrations_if_needed
+
     run_migrations_if_needed()
-	logging.getLogger(__name__).info("Database migrations executed (if needed) early in startup.")
+    logging.getLogger(__name__).info(
+        "Database migrations executed (if needed) early in startup."
+    )
 except Exception as e:
-	logging.getLogger(__name__).error(f"Database migration step failed early in startup: {e}", exc_info=True)
+    logging.getLogger(__name__).error(
+        f"Database migration step failed early in startup: {e}", exc_info=True
+    )
 
 # Import route routers (to be created)
 from routes.auth.credentials import router as credentials_router
@@ -44,7 +48,6 @@ from routes.auth import AUTH_ENABLED
 from routes.auth.middleware import AuthMiddleware
 
 # Import and initialize routes (this will start the watch manager)
-import routes
 
 
 # Configure application-wide logging
@@ -151,7 +154,9 @@ async def lifespan(app: FastAPI):
 
     # Check Redis connection
     if not check_redis_connection():
-		logging.error("Failed to connect to Redis. Please ensure Redis is running and accessible.")
+        logging.error(
+            "Failed to connect to Redis. Please ensure Redis is running and accessible."
+        )
         # Don't exit, but warn - some functionality may not work
 
     # Start Celery workers
@@ -177,7 +182,7 @@ def create_app():
         description="Music download service API",
         version="3.0.0",
         lifespan=lifespan,
-		redirect_slashes=True  # Enable automatic trailing slash redirects
+        redirect_slashes=True,  # Enable automatic trailing slash redirects
     )
 
     # Set up CORS
@@ -202,13 +207,16 @@ def create_app():
     # Include SSO router if available
     try:
         from routes.auth.sso import router as sso_router
+
         app.include_router(sso_router, prefix="/api/auth", tags=["sso"])
         logging.info("SSO functionality enabled")
     except ImportError as e:
         logging.warning(f"SSO functionality not available: {e}")
     app.include_router(config_router, prefix="/api/config", tags=["config"])
     app.include_router(search_router, prefix="/api/search", tags=["search"])
-	app.include_router(credentials_router, prefix="/api/credentials", tags=["credentials"])
+    app.include_router(
+        credentials_router, prefix="/api/credentials", tags=["credentials"]
+    )
     app.include_router(album_router, prefix="/api/album", tags=["album"])
     app.include_router(track_router, prefix="/api/track", tags=["track"])
     app.include_router(playlist_router, prefix="/api/playlist", tags=["playlist"])
@@ -245,19 +253,69 @@ def create_app():
         # Serve React App - catch-all route for SPA (but not for API routes)
         @app.get("/{full_path:path}")
         async def serve_react_app(full_path: str):
-			"""Serve React app with fallback to index.html for SPA routing"""
+            """Serve React app with fallback to index.html for SPA routing. Prevent directory traversal."""
             static_dir = "spotizerr-ui/dist"
+            static_dir_path = Path(static_dir).resolve()
+            index_path = static_dir_path / "index.html"
+            allowed_exts = {
+                ".html",
+                ".js",
+                ".css",
+                ".map",
+                ".png",
+                ".jpg",
+                ".jpeg",
+                ".svg",
+                ".webp",
+                ".ico",
+                ".json",
+                ".txt",
+                ".woff",
+                ".woff2",
+                ".ttf",
+                ".eot",
+                ".mp3",
+                ".ogg",
+                ".mp4",
+                ".webm",
+            }
 
             # Don't serve React app for API routes (more specific check)
             if full_path.startswith("api") or full_path.startswith("api/"):
                 raise HTTPException(status_code=404, detail="API endpoint not found")
 
-			# If it's a file that exists, serve it
+            # Reject null bytes early
-			if full_path and os.path.exists(os.path.join(static_dir, full_path)):
+            if "\x00" in full_path:
-				return FileResponse(os.path.join(static_dir, full_path))
+                return FileResponse(str(index_path))
+
+            # Sanitize path: normalize backslashes and strip URL schemes
+            sanitized = full_path.replace("\\", "/").lstrip("/")
+            if sanitized.startswith("http://") or sanitized.startswith("https://"):
+                return FileResponse(str(index_path))
+
+            # Resolve requested path safely and ensure it stays within static_dir
+            try:
+                requested_path = (static_dir_path / sanitized).resolve()
+            except Exception:
+                requested_path = index_path
+
+            # If traversal attempted or non-file within static dir, fall back to index.html for SPA routing
+            if not str(requested_path).startswith(str(static_dir_path)):
+                return FileResponse(str(index_path))
+
+            # Disallow hidden files (starting with dot) and enforce safe extensions
+            if requested_path.is_file():
+                name = requested_path.name
+                if name.startswith("."):
+                    return FileResponse(str(index_path))
+                suffix = requested_path.suffix.lower()
+                if suffix in allowed_exts:
+                    return FileResponse(str(requested_path))
+                # Not an allowed asset; fall back to SPA index
+                return FileResponse(str(index_path))
             else:
                 # Fallback to index.html for SPA routing
-				return FileResponse(os.path.join(static_dir, "index.html"))
+                return FileResponse(str(index_path))
     else:
         logging.warning("React app build directory not found at spotizerr-ui/dist")
 
@@ -285,10 +343,4 @@ if __name__ == "__main__":
     except ValueError:
         port = 7171
 
-	uvicorn.run(
+    uvicorn.run(app, host=host, port=port, log_level="info", access_log=True)
-		app,
-		host=host,
-		port=port,
-		log_level="info",
-		access_log=True
-	)

routes/core/history.py

@@ -1,9 +1,8 @@
-from fastapi import APIRouter, HTTPException, Request, Depends
+from fastapi import APIRouter, Request, Depends
 from fastapi.responses import JSONResponse
-import json
-import traceback
 import logging
 from routes.utils.history_manager import history_manager
+from typing import Any, Dict
 
 # Import authentication dependencies
 from routes.auth.middleware import require_auth_from_state, User
@@ -15,7 +14,9 @@ router = APIRouter()
 
 @router.get("/")
 @router.get("")
-async def get_history(request: Request, current_user: User = Depends(require_auth_from_state)):
+async def get_history(
+    request: Request, current_user: User = Depends(require_auth_from_state)
+):
     """
     Retrieve download history with optional filtering and pagination.
 
@@ -36,8 +37,10 @@ async def get_history(request: Request, current_user: User = Depends(require_aut
         valid_types = ["track", "album", "playlist"]
         if download_type and download_type not in valid_types:
             return JSONResponse(
-                content={"error": f"Invalid download_type. Must be one of: {valid_types}"},
+                content={
-                status_code=400
+                    "error": f"Invalid download_type. Must be one of: {valid_types}"
+                },
+                status_code=400,
             )
 
         # Validate status if provided
@@ -45,54 +48,50 @@ async def get_history(request: Request, current_user: User = Depends(require_aut
         if status and status not in valid_statuses:
             return JSONResponse(
                 content={"error": f"Invalid status. Must be one of: {valid_statuses}"},
-                status_code=400
+                status_code=400,
             )
 
         # Get history from manager
         history = history_manager.get_download_history(
-            limit=limit,
+            limit=limit, offset=offset, download_type=download_type, status=status
-            offset=offset,
-            download_type=download_type,
-            status=status
         )
 
         # Add pagination info
-        response_data = {
+        response_data: Dict[str, Any] = {
             "downloads": history,
             "pagination": {
                 "limit": limit,
                 "offset": offset,
-                "returned_count": len(history)
+                "returned_count": len(history),
-            }
+            },
         }
 
+        filters: Dict[str, Any] = {}
         if download_type:
-            response_data["filters"] = {"download_type": download_type}
+            filters["download_type"] = download_type
         if status:
-            if "filters" not in response_data:
+            filters["status"] = status
-                response_data["filters"] = {}
+        if filters:
-            response_data["filters"]["status"] = status
+            response_data["filters"] = filters
 
-        return JSONResponse(
+        return JSONResponse(content=response_data, status_code=200)
-            content=response_data, 
-            status_code=200
-        )
 
     except ValueError as e:
         return JSONResponse(
-            content={"error": f"Invalid parameter value: {str(e)}"},
+            content={"error": f"Invalid parameter value: {str(e)}"}, status_code=400
-            status_code=400
         )
     except Exception as e:
         logger.error(f"Error retrieving download history: {e}", exc_info=True)
         return JSONResponse(
             content={"error": "Failed to retrieve download history", "details": str(e)},
-            status_code=500
+            status_code=500,
         )
 
 
 @router.get("/{task_id}")
-async def get_download_by_task_id(task_id: str, current_user: User = Depends(require_auth_from_state)):
+async def get_download_by_task_id(
+    task_id: str, current_user: User = Depends(require_auth_from_state)
+):
     """
     Retrieve specific download history by task ID.
 
@@ -105,24 +104,25 @@ async def get_download_by_task_id(task_id: str, current_user: User = Depends(req
         if not download:
             return JSONResponse(
                 content={"error": f"Download with task ID '{task_id}' not found"},
-                status_code=404
+                status_code=404,
             )
 
-        return JSONResponse(
+        return JSONResponse(content=download, status_code=200)
-            content=download, 
-            status_code=200
-        )
 
     except Exception as e:
-        logger.error(f"Error retrieving download for task {task_id}: {e}", exc_info=True)
+        logger.error(
+            f"Error retrieving download for task {task_id}: {e}", exc_info=True
+        )
         return JSONResponse(
             content={"error": "Failed to retrieve download", "details": str(e)},
-            status_code=500
+            status_code=500,
         )
 
 
 @router.get("/{task_id}/children")
-async def get_download_children(task_id: str, current_user: User = Depends(require_auth_from_state)):
+async def get_download_children(
+    task_id: str, current_user: User = Depends(require_auth_from_state)
+):
     """
     Retrieve children tracks for an album or playlist download.
 
@@ -136,14 +136,14 @@ async def get_download_children(task_id: str, current_user: User = Depends(requi
         if not download:
             return JSONResponse(
                 content={"error": f"Download with task ID '{task_id}' not found"},
-                status_code=404
+                status_code=404,
             )
 
         children_table = download.get("children_table")
         if not children_table:
             return JSONResponse(
                 content={"error": f"Download '{task_id}' has no children tracks"},
-                status_code=404
+                status_code=404,
             )
 
         # Get children tracks
@@ -155,19 +155,21 @@ async def get_download_children(task_id: str, current_user: User = Depends(requi
             "title": download.get("title"),
             "children_table": children_table,
             "tracks": children,
-            "track_count": len(children)
+            "track_count": len(children),
         }
 
-        return JSONResponse(
+        return JSONResponse(content=response_data, status_code=200)
-            content=response_data, 
-            status_code=200
-        )
 
     except Exception as e:
-        logger.error(f"Error retrieving children for task {task_id}: {e}", exc_info=True)
+        logger.error(
+            f"Error retrieving children for task {task_id}: {e}", exc_info=True
+        )
         return JSONResponse(
-            content={"error": "Failed to retrieve download children", "details": str(e)},
+            content={
-            status_code=500
+                "error": "Failed to retrieve download children",
+                "details": str(e),
+            },
+            status_code=500,
         )
 
 
@@ -179,21 +181,23 @@ async def get_download_stats(current_user: User = Depends(require_auth_from_stat
     try:
         stats = history_manager.get_download_stats()
 
-        return JSONResponse(
+        return JSONResponse(content=stats, status_code=200)
-            content=stats, 
-            status_code=200
-        )
 
     except Exception as e:
         logger.error(f"Error retrieving download stats: {e}", exc_info=True)
         return JSONResponse(
-            content={"error": "Failed to retrieve download statistics", "details": str(e)},
+            content={
-            status_code=500
+                "error": "Failed to retrieve download statistics",
+                "details": str(e),
+            },
+            status_code=500,
         )
 
 
 @router.get("/search")
-async def search_history(request: Request, current_user: User = Depends(require_auth_from_state)):
+async def search_history(
+    request: Request, current_user: User = Depends(require_auth_from_state)
+):
     """
     Search download history by title or artist.
 
@@ -206,7 +210,7 @@ async def search_history(request: Request, current_user: User = Depends(require_
         if not query:
             return JSONResponse(
                 content={"error": "Missing required parameter: q (search query)"},
-                status_code=400
+                status_code=400,
             )
 
         limit = min(int(request.query_params.get("limit", 50)), 200)  # Cap at 200
@@ -218,29 +222,27 @@ async def search_history(request: Request, current_user: User = Depends(require_
             "query": query,
             "results": results,
             "result_count": len(results),
-            "limit": limit
+            "limit": limit,
         }
 
-        return JSONResponse(
+        return JSONResponse(content=response_data, status_code=200)
-            content=response_data, 
-            status_code=200
-        )
 
     except ValueError as e:
         return JSONResponse(
-            content={"error": f"Invalid parameter value: {str(e)}"},
+            content={"error": f"Invalid parameter value: {str(e)}"}, status_code=400
-            status_code=400
         )
     except Exception as e:
         logger.error(f"Error searching download history: {e}", exc_info=True)
         return JSONResponse(
             content={"error": "Failed to search download history", "details": str(e)},
-            status_code=500
+            status_code=500,
         )
 
 
 @router.get("/recent")
-async def get_recent_downloads(request: Request, current_user: User = Depends(require_auth_from_state)):
+async def get_recent_downloads(
+    request: Request, current_user: User = Depends(require_auth_from_state)
+):
     """
     Get most recent downloads.
 
@@ -252,32 +254,26 @@ async def get_recent_downloads(request: Request, current_user: User = Depends(re
 
         recent = history_manager.get_recent_downloads(limit)
 
-        response_data = {
+        response_data = {"downloads": recent, "count": len(recent), "limit": limit}
-            "downloads": recent,
-            "count": len(recent),
-            "limit": limit
-        }
 
-        return JSONResponse(
+        return JSONResponse(content=response_data, status_code=200)
-            content=response_data, 
-            status_code=200
-        )
 
     except ValueError as e:
         return JSONResponse(
-            content={"error": f"Invalid parameter value: {str(e)}"},
+            content={"error": f"Invalid parameter value: {str(e)}"}, status_code=400
-            status_code=400
         )
     except Exception as e:
         logger.error(f"Error retrieving recent downloads: {e}", exc_info=True)
         return JSONResponse(
             content={"error": "Failed to retrieve recent downloads", "details": str(e)},
-            status_code=500
+            status_code=500,
         )
 
 
 @router.get("/failed")
-async def get_failed_downloads(request: Request, current_user: User = Depends(require_auth_from_state)):
+async def get_failed_downloads(
+    request: Request, current_user: User = Depends(require_auth_from_state)
+):
     """
     Get failed downloads.
 
@@ -289,32 +285,26 @@ async def get_failed_downloads(request: Request, current_user: User = Depends(re
 
         failed = history_manager.get_failed_downloads(limit)
 
-        response_data = {
+        response_data = {"downloads": failed, "count": len(failed), "limit": limit}
-            "downloads": failed,
-            "count": len(failed),
-            "limit": limit
-        }
 
-        return JSONResponse(
+        return JSONResponse(content=response_data, status_code=200)
-            content=response_data, 
-            status_code=200
-        )
 
     except ValueError as e:
         return JSONResponse(
-            content={"error": f"Invalid parameter value: {str(e)}"},
+            content={"error": f"Invalid parameter value: {str(e)}"}, status_code=400
-            status_code=400
         )
     except Exception as e:
         logger.error(f"Error retrieving failed downloads: {e}", exc_info=True)
         return JSONResponse(
             content={"error": "Failed to retrieve failed downloads", "details": str(e)},
-            status_code=500
+            status_code=500,
         )
 
 
 @router.post("/cleanup")
-async def cleanup_old_history(request: Request, current_user: User = Depends(require_auth_from_state)):
+async def cleanup_old_history(
+    request: Request, current_user: User = Depends(require_auth_from_state)
+):
     """
     Clean up old download history.
 
@@ -322,31 +312,32 @@ async def cleanup_old_history(request: Request, current_user: User = Depends(req
     - days_old: Number of days old to keep (default: 30)
     """
     try:
-        data = await request.json() if request.headers.get("content-type") == "application/json" else {}
+        data = (
+            await request.json()
+            if request.headers.get("content-type") == "application/json"
+            else {}
+        )
         days_old = data.get("days_old", 30)
 
         if not isinstance(days_old, int) or days_old <= 0:
             return JSONResponse(
                 content={"error": "days_old must be a positive integer"},
-                status_code=400
+                status_code=400,
             )
 
         deleted_count = history_manager.clear_old_history(days_old)
 
         response_data = {
-            "message": f"Successfully cleaned up old download history",
+            "message": "Successfully cleaned up old download history",
             "deleted_records": deleted_count,
-            "days_old": days_old
+            "days_old": days_old,
         }
 
-        return JSONResponse(
+        return JSONResponse(content=response_data, status_code=200)
-            content=response_data, 
-            status_code=200
-        )
 
     except Exception as e:
         logger.error(f"Error cleaning up old history: {e}", exc_info=True)
         return JSONResponse(
             content={"error": "Failed to cleanup old history", "details": str(e)},
-            status_code=500
+            status_code=500,
         )