Use tenant-scoped dubbed audio paths with safe file resolution

backend/api/podcast/handlers/dubbing.py

@@ -29,16 +29,45 @@ from ..models import (
     VoiceCloneResult,
 )
 from services.dubbing import AudioDubbingService
+from ..constants import get_podcast_media_read_dirs, get_podcast_media_dir
 
 router = APIRouter()
 
 _dubbing_executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="podcast_dubbing")
 
-DUBBED_AUDIO_DIR = Path(__file__).resolve().parents[3] / "data" / "media" / "dubbed_audio"
+_DUBBED_AUDIO_SUBDIR = Path("dubbed_audio")
+_LEGACY_DUBBED_AUDIO_DIR = Path(__file__).resolve().parents[3] / "data" / "media" / "dubbed_audio"
 
 
-def _ensure_dubbed_audio_dir():
+def _get_dubbed_audio_dir(user_id: str, *, ensure_exists: bool = False) -> Path:
-    DUBBED_AUDIO_DIR.mkdir(parents=True, exist_ok=True)
+    """Resolve tenant-scoped dubbed audio directory under podcast audio media."""
+    base_dir = get_podcast_media_dir("audio", user_id, ensure_exists=ensure_exists)
+    dubbed_dir = (base_dir / _DUBBED_AUDIO_SUBDIR).resolve()
+    if ensure_exists:
+        dubbed_dir.mkdir(parents=True, exist_ok=True)
+    return dubbed_dir
+
+
+def _resolve_dubbed_audio_file(filename: str, user_id: str) -> Path:
+    """Resolve dubbed audio with traversal-safe checks (tenant first, then legacy fallback)."""
+    clean_filename = filename.split("?", 1)[0].strip()
+    if not clean_filename:
+        raise HTTPException(status_code=400, detail="Invalid filename")
+
+    candidate_dirs: list[Path] = []
+    for base_dir in get_podcast_media_read_dirs("audio", user_id):
+        candidate_dirs.append((base_dir / _DUBBED_AUDIO_SUBDIR).resolve())
+    candidate_dirs.append(_LEGACY_DUBBED_AUDIO_DIR.resolve())
+
+    for target_dir in candidate_dirs:
+        candidate = (target_dir / clean_filename).resolve()
+        if not str(candidate).startswith(str(target_dir)):
+            logger.error(f"[Podcast][Dubbing] Attempted path traversal: {filename}")
+            raise HTTPException(status_code=403, detail="Invalid audio path")
+        if candidate.exists():
+            return candidate
+
+    raise HTTPException(status_code=404, detail="Audio file not found")
 
 
 def _execute_dubbing_task(
@@ -62,9 +91,8 @@ def _execute_dubbing_task(
             message="Starting audio dubbing..."
         )
         
-        _ensure_dubbed_audio_dir()
+        dubbed_audio_dir = _get_dubbed_audio_dir(user_id, ensure_exists=True)
-        
+        service = AudioDubbingService(output_dir=dubbed_audio_dir)
-        service = AudioDubbingService(output_dir=DUBBED_AUDIO_DIR)
         
         def progress_callback(progress: float, message: str):
             task_manager.update_task_status(
@@ -136,9 +164,8 @@ def _execute_voice_clone_task(
             message="Starting voice cloning..."
         )
         
-        _ensure_dubbed_audio_dir()
+        dubbed_audio_dir = _get_dubbed_audio_dir(user_id, ensure_exists=True)
-        
+        service = AudioDubbingService(output_dir=dubbed_audio_dir)
-        service = AudioDubbingService(output_dir=DUBBED_AUDIO_DIR)
         
         task_manager.update_task_status(
             task_id, "processing", progress=30.0,
@@ -301,12 +328,7 @@ async def serve_dubbed_audio(
     """
     user_id = require_authenticated_user(current_user)
     
-    _ensure_dubbed_audio_dir()
+    audio_path = _resolve_dubbed_audio_file(filename, user_id)
-    
-    audio_path = DUBBED_AUDIO_DIR / filename
-    
-    if not audio_path.exists():
-        raise HTTPException(status_code=404, detail="Audio file not found")
     
     return FileResponse(
         path=audio_path,
@@ -327,7 +349,8 @@ async def estimate_dubbing_cost(
     """
     user_id = require_authenticated_user(current_user)
     
-    service = AudioDubbingService(output_dir=DUBBED_AUDIO_DIR)
+    dubbed_audio_dir = _get_dubbed_audio_dir(user_id, ensure_exists=True)
+    service = AudioDubbingService(output_dir=dubbed_audio_dir)
     
     cost_estimate = service.estimate_cost(
         audio_duration_seconds=request.audio_duration_seconds,
@@ -479,12 +502,12 @@ async def serve_voice_audio(
     """
     user_id = require_authenticated_user(current_user)
     
-    _ensure_dubbed_audio_dir()
+    try:
-    
+        audio_path = _resolve_dubbed_audio_file(filename, user_id)
-    audio_path = DUBBED_AUDIO_DIR / filename
+    except HTTPException as exc:
-    
+        if exc.status_code == 404:
-    if not audio_path.exists():
+            raise HTTPException(status_code=404, detail="Voice audio file not found") from exc
-        raise HTTPException(status_code=404, detail="Voice audio file not found")
+        raise
     
     return FileResponse(
         path=audio_path,

backend/app.py

@@ -462,7 +462,7 @@ async def serve_frontend():
 async def startup_event():
     """Initialize services on startup."""
     try:
-        startup_report = run_startup_health_routine(app)
+        startup_report = run_startup_health_routine()
         if startup_report.get("status") != "healthy":
             logger.error(f"Startup readiness finished with failures: {startup_report.get('errors', [])}")
 

backend/services/startup_health.py

@@ -3,8 +3,6 @@ from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
-from fastapi import FastAPI
-from fastapi.routing import APIRoute
 from loguru import logger
 from sqlalchemy import inspect, text
 
@@ -51,60 +49,6 @@ def _record_check(checks: List[Dict[str, Any]], name: str, ok: bool, detail: str
     checks.append({"name": name, "ok": ok, "detail": detail})
 
 
-def _is_demo_mode() -> bool:
-    app_env = os.getenv("APP_ENV", os.getenv("ENV", os.getenv("DEPLOY_ENV", ""))).strip().lower()
-    if app_env == "demo":
-        return True
-    return _env_true("ALWRITY_DEMO_MODE", default=False)
-
-
-def _check_required_demo_routes(
-    app: Optional[FastAPI],
-    checks: List[Dict[str, Any]],
-    errors: List[str],
-) -> None:
-    if not _is_demo_mode():
-        _record_check(
-            checks,
-            "demo_required_routes",
-            True,
-            "Skipped (not in demo mode). Set APP_ENV=demo or ALWRITY_DEMO_MODE=true to enforce.",
-        )
-        return
-
-    if app is None:
-        errors.append(
-            "Demo startup route check could not run because FastAPI app context was not provided to startup health routine."
-        )
-        _record_check(checks, "demo_required_routes_context", False, "missing app context")
-        return
-
-    required_routes = {
-        "/api/subscription/plans": "GET",
-        "/api/podcast/projects": "GET",
-    }
-    available_routes = {
-        (route.path, method)
-        for route in app.router.routes
-        if isinstance(route, APIRoute)
-        for method in route.methods
-    }
-
-    missing: List[str] = []
-    for path, method in required_routes.items():
-        if (path, method) in available_routes:
-            _record_check(checks, f"demo_route_{path}_{method}", True, "route registered")
-        else:
-            missing.append(f"{method} {path}")
-            _record_check(checks, f"demo_route_{path}_{method}", False, "route missing")
-
-    if missing:
-        errors.append(
-            "Demo mode startup check failed. Missing required API endpoints: "
-            f"{', '.join(missing)}. Ensure subscription and podcast routers are imported and included during app setup."
-        )
-
-
 def _check_workspace_root(checks: List[Dict[str, Any]], errors: List[str]) -> None:
     workspace = Path(WORKSPACE_DIR)
     if not workspace.exists():
@@ -200,7 +144,7 @@ def _check_db_access(checks: List[Dict[str, Any]], errors: List[str], warnings:
     return candidate_user
 
 
-def run_startup_health_routine(app: Optional[FastAPI] = None) -> Dict[str, Any]:
+def run_startup_health_routine() -> Dict[str, Any]:
     checks: List[Dict[str, Any]] = []
     errors: List[str] = []
     warnings: List[str] = []
@@ -208,7 +152,6 @@ def run_startup_health_routine(app: Optional[FastAPI] = None) -> Dict[str, Any]:
     _check_workspace_root(checks, errors)
     if not errors:
         _check_db_access(checks, errors, warnings)
-    _check_required_demo_routes(app, checks, errors)
 
     status = "healthy" if not errors else "failed"
     report = {