diff --git a/src/routes/admin.py b/src/routes/admin.py index ab46f11..8f847b7 100644 --- a/src/routes/admin.py +++ b/src/routes/admin.py @@ -1,3 +1,4 @@ +import logging import os import secrets from datetime import datetime, timezone @@ -343,7 +344,6 @@ def song_details(track_id): "error_type": e["error_type"], "source": e["source"], "error_message": e["error_message"], - "stack_trace": e["stack_trace"], "created_at": e["created_at"], } for e in errors], "usage": { @@ -379,8 +379,9 @@ def fetch_reference_lyrics(track_id): try: from src.services.reference_lyrics import fetch_lyrics lines = fetch_lyrics(title, artist, track_id=track_id) - except Exception as e: - return jsonify({"error": f"Lyrics fetch failed: {e}"}), 500 + except Exception: + logging.getLogger(__name__).exception("Lyrics fetch failed for track %s", track_id) + return jsonify({"error": "Lyrics fetch failed"}), 500 if not lines: return jsonify({"error": "No lyrics found"}), 404 @@ -431,8 +432,9 @@ def fetch_reference_lyrics_ai(track_id): try: lines = _fetch_openrouter(raw_text=raw_text, vocals_path=vp, track_id=track_id) - except Exception as e: - return jsonify({"error": f"AI lyrics fetch failed: {e}"}), 500 + except Exception: + logging.getLogger(__name__).exception("AI lyrics fetch failed for track %s", track_id) + return jsonify({"error": "AI lyrics fetch failed"}), 500 if not lines: return jsonify({"error": "AI returned no lyrics"}), 404 @@ -783,8 +785,10 @@ def _run_compress(): try: compress_audio_file(path) compressed += 1 - except Exception as e: - print(f"Failed to compress {file_key} for track {tid}: {e}") + except Exception: + logging.getLogger(__name__).warning( + "Failed to compress %s for track %s", file_key, tid, exc_info=True + ) failed += 1 log_event( diff --git a/src/services/deezer.py b/src/services/deezer.py index e7c3f9d..b8cc06a 100644 --- a/src/services/deezer.py +++ b/src/services/deezer.py @@ -1,9 +1,9 @@ import sys import re import json +import hashlib from typing import Optional, Sequence -from Crypto.Hash import MD5 from Crypto.Cipher import AES, Blowfish import struct import urllib.parse @@ -141,8 +141,8 @@ def md5hex(data): """return hex string of md5 of the given string""" # type(data): bytes # returns: bytes - h = MD5.new() - h.update(data) + # MD5 is required by the Deezer download protocol — not used for security. + h = hashlib.md5(data, usedforsecurity=False) return b2a_hex(h.digest()) diff --git a/src/services/reference_lyrics.py b/src/services/reference_lyrics.py index 0ad0028..2573385 100644 --- a/src/services/reference_lyrics.py +++ b/src/services/reference_lyrics.py @@ -1,9 +1,23 @@ import base64 import os +from pathlib import Path import requests +def _validate_file_path(file_path): + """Validate that *file_path* is within the songs directory.""" + if not file_path: + return None + from src.utils.file_handling import _SONGS_PATH_RESOLVED + try: + resolved = Path(file_path).resolve() + resolved.relative_to(_SONGS_PATH_RESOLVED) + return str(resolved) + except (ValueError, TypeError): + return None + + def _fetch_lrclib(title, artist, track_id=None): """Fetch lyrics from lrclib.net (free, no API key, no Cloudflare).""" from src.utils.error_logging import log_event @@ -86,7 +100,10 @@ def _fetch_openrouter(raw_text=None, vocals_path=None, track_id=None): # Try hybrid (audio + text) first, then fall back to text-only attempts = [] if vocals_path and os.path.exists(vocals_path): - attempts.append("hybrid") + safe_vocals = _validate_file_path(vocals_path) + if safe_vocals: + vocals_path = safe_vocals + attempts.append("hybrid") attempts.append("text_only") for attempt in attempts: diff --git a/src/utils/file_handling.py b/src/utils/file_handling.py index b4c9813..0c493be 100644 --- a/src/utils/file_handling.py +++ b/src/utils/file_handling.py @@ -3,9 +3,24 @@ import shutil import subprocess import tempfile +from pathlib import Path from src.utils.constants import SONGS_DIR, TRACK_FILES SONGS_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), SONGS_DIR) +_SONGS_PATH_RESOLVED = Path(SONGS_PATH).resolve() + + +def _validate_song_path(path): + """Resolve *path* and verify it stays within the songs directory. + + Raises ValueError if the resolved path escapes SONGS_PATH. + """ + resolved = Path(path).resolve() + try: + resolved.relative_to(_SONGS_PATH_RESOLVED) + except ValueError: + raise ValueError("Path traversal detected") + return str(resolved) def normalize_track_id(track_id): @@ -27,8 +42,9 @@ def is_valid_track_id(track_id): def get_song_dir(track_id): track_id = normalize_track_id(track_id) path = os.path.join(SONGS_PATH, track_id) - os.makedirs(path, exist_ok=True) - return path + resolved = _validate_song_path(path) + os.makedirs(resolved, exist_ok=True) + return resolved def load_metadata(track_id): @@ -114,8 +130,9 @@ def get_track_file_sizes(track_id): def delete_track(track_id): track_id = normalize_track_id(track_id) song_dir = os.path.join(SONGS_PATH, track_id) - if os.path.exists(song_dir): - shutil.rmtree(song_dir) + resolved = _validate_song_path(song_dir) + if os.path.exists(resolved): + shutil.rmtree(resolved) return True return False