Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 11 additions & 7 deletions src/routes/admin.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import logging
import os
import secrets
from datetime import datetime, timezone
Expand Down Expand Up @@ -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": {
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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(
Expand Down
6 changes: 3 additions & 3 deletions src/services/deezer.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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())


Expand Down
19 changes: 18 additions & 1 deletion src/services/reference_lyrics.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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:
Expand Down
25 changes: 21 additions & 4 deletions src/utils/file_handling.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()

Check failure

Code scanning / CodeQL

Uncontrolled data used in path expression High

This path depends on a
user-provided value
.
This path depends on a
user-provided value
.
This path depends on a
user-provided value
.
This path depends on a
user-provided value
.
This path depends on a
user-provided value
.
This path depends on a
user-provided value
.
This path depends on a
user-provided value
.
This path depends on a
user-provided value
.
This path depends on a
user-provided value
.
This path depends on a
user-provided value
.
This path depends on a
user-provided value
.
try:
resolved.relative_to(_SONGS_PATH_RESOLVED)
except ValueError:
raise ValueError("Path traversal detected")
return str(resolved)


def normalize_track_id(track_id):
Expand All @@ -27,8 +42,9 @@
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):
Expand Down Expand Up @@ -114,8 +130,9 @@
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

Expand Down
Loading