diff --git a/apps/api/app/api/routes/generation_preview.py b/apps/api/app/api/routes/generation_preview.py index 12270525c..2867b084c 100755 --- a/apps/api/app/api/routes/generation_preview.py +++ b/apps/api/app/api/routes/generation_preview.py @@ -213,7 +213,7 @@ def get_preview_generation_task( # 权限校验:任务必须属于当前用户 task_user_id = getattr(task, "created_by_user_id", "") or "" - if task_user_id and task_user_id != authenticated_user.user.id: + if not task_user_id or task_user_id != authenticated_user.user.id: raise HTTPException(status_code=403, detail="无权访问该任务") # 校验是否为预览任务 diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index d697368f6..e2eb61897 100755 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -1056,8 +1056,9 @@ def _render_video( # 确保输出分辨率配置存在 # 优先级:用户指定 > 模板配置 > 默认 1280x720 # 预览模式:强制 854x480 + 低码率 - plan_cfg = virtual_plan.config or {} - export_cfg = plan_cfg.get("export", {}) or {} + # 注意:必须拷贝字典,避免预览模式修改污染源对象(模板配置) + plan_cfg = dict(virtual_plan.config or {}) + export_cfg = dict(plan_cfg.get("export", {}) or {}) if is_preview: # 预览模式强制 480p + 低码率 export_cfg["resolution"] = "854x480" diff --git a/tests/unit/test_generation_preview.py b/tests/unit/test_generation_preview.py index 06b7f839a..ca05fca75 100755 --- a/tests/unit/test_generation_preview.py +++ b/tests/unit/test_generation_preview.py @@ -945,6 +945,26 @@ class TestGetPreviewRoute: assert resp.status == "pending" assert resp.is_preview is True + def test_empty_user_id_denied(self): + """任务 created_by_user_id 为空 → 403(防止越权)""" + repo = MagicMock() + vid_repo = MagicMock() + task = _make_task() + task.created_by_user_id = "" # 空字符串 + + from fastapi import HTTPException + + with patch("app.api.routes.generation_preview.GetGenerationTaskUseCase") as MockUC: + MockUC.return_value.execute.return_value = task + with pytest.raises(HTTPException) as exc_info: + get_preview_generation_task( + task_id="preview_task_001", + authenticated_user=_make_user(), + generation_task_repository=repo, + generated_video_repository=vid_repo, + ) + assert exc_info.value.status_code == 403 + def test_get_completed_task_with_videos(self): """查询 completed 状态任务,返回视频列表""" repo = MagicMock() @@ -980,11 +1000,12 @@ class TestWorkerPreviewResolution: def test_preview_mode_forces_480p(self): """预览模式强制 854x480 + 1M 码率""" - # 模拟 worker 中 export_cfg 逻辑 + # 模拟 worker 中 export_cfg 逻辑(与实际代码一致,使用 dict 拷贝) is_preview = True resolution = "1920x1080" # 用户指定的分辨率应被忽略 - plan_cfg = {"export": {"resolution": "1280x720", "bitrate": "5M"}} - export_cfg = plan_cfg.get("export", {}) or {} + original_config = {"export": {"resolution": "1280x720", "bitrate": "5M"}} + plan_cfg = dict(original_config) + export_cfg = dict(plan_cfg.get("export", {}) or {}) if is_preview: export_cfg["resolution"] = "854x480" @@ -992,15 +1013,19 @@ class TestWorkerPreviewResolution: elif resolution: export_cfg["resolution"] = resolution + plan_cfg["export"] = export_cfg + assert export_cfg["resolution"] == "854x480" assert export_cfg["bitrate"] == "1M" + # 验证原始配置未被污染 + assert original_config["export"]["resolution"] == "1280x720" def test_non_preview_uses_user_resolution(self): """非预览模式使用用户指定分辨率""" is_preview = False resolution = "1920x1080" - plan_cfg = {"export": {"resolution": "1280x720"}} - export_cfg = plan_cfg.get("export", {}) or {} + plan_cfg = dict({"export": {"resolution": "1280x720"}}) + export_cfg = dict(plan_cfg.get("export", {}) or {}) if is_preview: export_cfg["resolution"] = "854x480" @@ -1014,8 +1039,8 @@ class TestWorkerPreviewResolution: """非预览模式且用户未指定分辨率,使用模板配置""" is_preview = False resolution = "" - plan_cfg = {"export": {"resolution": "1280x720"}} - export_cfg = plan_cfg.get("export", {}) or {} + plan_cfg = dict({"export": {"resolution": "1280x720"}}) + export_cfg = dict(plan_cfg.get("export", {}) or {}) if is_preview: export_cfg["resolution"] = "854x480" @@ -1030,8 +1055,8 @@ class TestWorkerPreviewResolution: def test_preview_mode_empty_export_cfg(self): """预览模式且模板无 export 配置""" is_preview = True - plan_cfg = {} - export_cfg = plan_cfg.get("export", {}) or {} + plan_cfg = dict({}) + export_cfg = dict(plan_cfg.get("export", {}) or {}) if is_preview: export_cfg["resolution"] = "854x480"