From 3e71a00b12900b63d094402778de905658b123e9 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 28 Aug 2026 22:58:45 +0800 Subject: [PATCH 1/7] =?UTF-8?q?fix:=20=E5=B0=81=E9=9D=A2=E9=80=89=E5=B8=A7?= =?UTF-8?q?=E8=BE=93=E5=85=A5=E6=BA=90=E6=94=B9=E4=B8=BA=E6=9C=80=E7=BB=88?= =?UTF-8?q?=E6=88=90=E7=89=87=EF=BC=8C=E9=A2=84=E8=A7=88=E7=89=87=E6=AE=B5?= =?UTF-8?q?=E4=BD=9C=E4=B8=BA=E5=85=BC=E5=AE=B9=E5=9B=9E=E9=80=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - GenerateCoverRequest 新增可选字段 generated_video_id / video_url， 与前端 PR #1533 契约对齐：前端在「选择封面」步骤传入确认生成产出的 最终视频 ID 或 URL，后端据此定位最终成片文件并抽帧 - 视频源查找优先级调整： 0a. 请求体 generated_video_id → GeneratedVideo.file_url 0b. 请求体 video_url → 直接使用 1. plan.config.rendered_storage_key 2. plan.config.generation_task_id 3. source_edit_plan 关联的 is_preview=False 已完成任务（最终成片优先） 4. source_edit_plan 关联的 is_preview=True 预览任务（回退） 5. user+template 最近预览任务（兜底） - cover_url 查找同步调整：优先最终成片任务的 cover_url，再回退预览 - MediaKit 的 strategy/max_frames/轮询/重试/降级逻辑全部不变 - 权限校验：通过 generated_video_id 关联的 generation_task 校验归属 - 新增 5 个单测覆盖：schema 字段、video_id 抽帧、video_url 抽帧、 最终成片优先于预览、权限校验 403 --- apps/api/app/api/routes/generation_cover.py | 322 ++++++++++++++------ tests/unit/test_generation_cover.py | 301 ++++++++++++++++++ 2 files changed, 533 insertions(+), 90 deletions(-) diff --git a/apps/api/app/api/routes/generation_cover.py b/apps/api/app/api/routes/generation_cover.py index ef4400615..b42283c52 100644 --- a/apps/api/app/api/routes/generation_cover.py +++ b/apps/api/app/api/routes/generation_cover.py @@ -1,7 +1,7 @@ """封面生成路由 — Generation 模块. 端点： -- POST /generate-cover AI 生成封面（从预览视频中抽帧） +- POST /generate-cover AI 生成封面（从最终成片视频中抽帧，兼容预览片段回退） 挂载路径: /api/v1/generation/generate-cover """ @@ -51,6 +51,16 @@ class GenerateCoverRequest(BaseModel): default=None, description="上传的封面图片 URL，仅 cover_type=upload 时有效", ) + generated_video_id: Optional[str] = Field( + default=None, + description="确认生成产出的最终视频 ID。传入后封面从该视频文件抽帧，" + "而非预览片段。", + ) + video_url: Optional[str] = Field( + default=None, + description="最终视频 URL（兜底）。当 generated_video_id 不可用时，" + "直接从此 URL 对应的视频抽帧。", + ) class GenerateCoverResponse(BaseModel): @@ -140,6 +150,41 @@ def _persist_cover_frame( Path(tmp_path).unlink(missing_ok=True) + +def _get_task_video_url(db: Session, task_id: str) -> Optional[str]: + """从 GenerationTask 关联的 GeneratedVideo 中获取视频 storage_key / URL.""" + try: + video_repo = get_generated_video_repository(db) + use_case = ListGeneratedVideosByTaskUseCase(video_repo) + videos = use_case.execute(task_id) + if videos: + return getattr(videos[0], "file_url", "") or "" + except Exception: + logger.warning("[封面生成] 获取任务视频失败: task_id=%s", task_id, exc_info=True) + return None + + +def _resolve_storage_key_to_url(storage_key: str) -> Optional[str]: + """将 storage_key 或完整 URL 转换为可访问的裸 URL。""" + if not storage_key: + return None + try: + if storage_key.startswith("http"): + url = storage_key + else: + from packages.shared.storage import get_shared_storage_service + + storage_svc = get_shared_storage_service() + url = storage_svc.get_url(storage_key) + if url: + import re as _re + url = _re.sub(r"(? GenerateCoverResponse: - """AI 生成封面 — 从预览视频中抽帧. + """AI 生成封面 — 优先从最终成片视频中抽帧，回退到预览片段. 流程（串行）： - 1. 预览视频已渲染完成（通过 3 步查找获取 URL） - 2. 用裸 URL 让 MediaKit 下载视频并抽帧 - 3. 帧图下载后上传到 OSS covers/ 路径 + 1. 优先使用前端传入的 generation_task_id 定位最终成片任务， + 或自动查找 plan 关联的已完成最终成片任务（is_preview=False） + 2. 回退：从预览片段获取视频 URL（兼容旧流程） + 3. 用裸 URL 让 MediaKit 下载视频并抽帧 + 4. 帧图下载后上传到 OSS covers/ 路径 + + MediaKit 的调用方式（strategy / max_frames / 轮询 / 重试 / 降级）不变。 """ _, plan_svc = services plan = plan_svc.get_plan_or_raise(plan_id) @@ -182,122 +231,174 @@ def generate_cover( ) return GenerateCoverResponse(plan_id=plan_id, cover=cover_data) - # ── 3 步查找预览视频 URL ────────────────────────────────────────── - # 第一步：从 plan.config 读取 + # ── 查找用于抽帧的视频 URL ──────────────────────────────────────── + # 优先级： + # 0. 请求体显式传入的 generation_task_id（最终成片任务） + # 1. plan.config.rendered_storage_key + # 2. plan.config.generation_task_id 对应的任务 + # 3. source_edit_plan_id 关联的已完成「最终成片」任务（is_preview=False） + # 4. source_edit_plan_id 关联的已完成预览任务（is_preview=True，兼容回退） + # 5. user + template 最近的已完成预览任务（兜底） logger.info("[封面生成] 步骤1: 从 plan.config 查找 rendered_storage_key: plan_id=%s", plan_id) rendered_storage_key = (plan.config or {}).get("rendered_storage_key", "") - # 第二步：如果还没有，通过 generation_task_id 查找预览任务的产物 + # 步骤 0：请求体传入最终视频标识（generated_video_id 或 video_url） + if not rendered_storage_key: + # 0a：通过 generated_video_id 查找最终成片视频 + if body.generated_video_id: + logger.info( + "[封面生成] 步骤0a: 使用 generated_video_id: plan_id=%s video_id=%s", + plan_id, body.generated_video_id, + ) + try: + from app.dependencies import get_generated_video_repository as _get_gv_repo + + gv_repo = _get_gv_repo(db) + gv = gv_repo.get(body.generated_video_id) + if gv: + file_url = getattr(gv, "file_url", "") or "" + if file_url: + # 权限校验：通过关联的 generation_task 校验归属 + gv_task_id = getattr(gv, "generation_task_id", "") or "" + if gv_task_id: + try: + _repo0 = SQLAlchemyGenerationTaskRepository(db) + task0 = _repo0.get(gv_task_id) + if ( + task0 + and task0.created_by_user_id + and task0.created_by_user_id != current_user.user.id + ): + raise HTTPException(status_code=403, detail="无权访问该视频") + except HTTPException: + raise + except Exception: + pass + rendered_storage_key = file_url + logger.info( + "[封面生成] ✅ 步骤0a找到最终成片: plan_id=%s video_id=%s url=%s", + plan_id, body.generated_video_id, file_url[:80], + ) + except HTTPException: + raise + except Exception: + logger.warning( + "[封面生成] 步骤0a查找视频失败: plan_id=%s video_id=%s", + plan_id, body.generated_video_id, exc_info=True, + ) + + # 0b：直接使用 video_url（兜底） + if not rendered_storage_key and body.video_url: + logger.info( + "[封面生成] 步骤0b: 使用请求体传入的 video_url: plan_id=%s url=%s", + plan_id, body.video_url[:80], + ) + rendered_storage_key = body.video_url + + # 步骤 2：通过 plan.config.generation_task_id 查找 if not rendered_storage_key: generation_task_id = (plan.config or {}).get("generation_task_id", "") - logger.info( - "[封面生成] 步骤2: 通过 generation_task_id 查找: plan_id=%s task_id=%s", plan_id, generation_task_id - ) if generation_task_id: + logger.info( + "[封面生成] 步骤2: 通过 plan.config.generation_task_id 查找: plan_id=%s task_id=%s", + plan_id, generation_task_id, + ) try: - gen_task_repo = SQLAlchemyGenerationTaskRepository(db) - task = gen_task_repo.get(generation_task_id) + _repo = SQLAlchemyGenerationTaskRepository(db) + task = _repo.get(generation_task_id) if task: - video_repo = get_generated_video_repository(db) - use_case = ListGeneratedVideosByTaskUseCase(video_repo) - videos = use_case.execute(task.id) - if videos: - rendered_storage_key = getattr(videos[0], "file_url", "") or "" + rendered_storage_key = _get_task_video_url(db, task.id) or "" + if rendered_storage_key: logger.info( "[封面生成] ✅ 步骤2找到视频: plan_id=%s task_id=%s url=%s", - plan_id, - generation_task_id, - rendered_storage_key[:80], + plan_id, generation_task_id, rendered_storage_key[:80], ) except Exception: logger.warning( - "封面生成: 通过 generation_task_id 查找视频失败: plan_id=%s", - plan_id, - exc_info=True, + "[封面生成] 步骤2查找失败: plan_id=%s", plan_id, exc_info=True, ) - # 第 2.5 步：通过 plan_id 作为 source_edit_plan_id 查找关联的已完成预览任务 + # 步骤 3：通过 source_edit_plan_id 查找已完成「最终成片」任务（is_preview=False） if not rendered_storage_key: try: - gen_task_repo = SQLAlchemyGenerationTaskRepository(db) - logger.info("[封面生成] 步骤2.5: 通过 source_edit_plan_id 查找: plan_id=%s", plan_id) - preview_tasks = gen_task_repo.list_by_source_edit_plan(plan_id) - for pt in preview_tasks: - if getattr(pt, "status", "") == "completed" and getattr(pt, "is_preview", False): - video_repo = get_generated_video_repository(db) - use_case = ListGeneratedVideosByTaskUseCase(video_repo) - videos = use_case.execute(pt.id) - if videos: - rendered_storage_key = getattr(videos[0], "file_url", "") or "" + _repo = SQLAlchemyGenerationTaskRepository(db) + logger.info("[封面生成] 步骤3: 查找最终成片任务(is_preview=False): plan_id=%s", plan_id) + all_tasks = _repo.list_by_source_edit_plan(plan_id) + for pt in all_tasks: + if ( + getattr(pt, "status", "") == "completed" + and not getattr(pt, "is_preview", False) + ): + rendered_storage_key = _get_task_video_url(db, pt.id) or "" + if rendered_storage_key: logger.info( - "[封面生成] ✅ 步骤2.5找到视频: plan_id=%s task_id=%s url=%s", - plan_id, - pt.id, - rendered_storage_key[:80], + "[封面生成] ✅ 步骤3找到最终成片: plan_id=%s task_id=%s url=%s", + plan_id, pt.id, rendered_storage_key[:80], ) break except Exception: logger.warning( - "封面生成: 通过 source_edit_plan_id 查找预览任务失败: plan_id=%s", - plan_id, - exc_info=True, + "[封面生成] 步骤3查找最终成片失败: plan_id=%s", plan_id, exc_info=True, ) - # 第三步：按 user + template 查找最近的已完成预览任务（兜底） + # 步骤 4：兼容回退 — 通过 source_edit_plan_id 查找已完成预览任务 if not rendered_storage_key: try: - gen_task_repo = SQLAlchemyGenerationTaskRepository(db) - logger.info("[封面生成] 步骤3: 通过 user+template 查找: plan_id=%s template_id=%s", plan_id, template_id) - preview_tasks = gen_task_repo.list_latest_completed_preview( + _repo = SQLAlchemyGenerationTaskRepository(db) + logger.info("[封面生成] 步骤4: 回退查找预览任务(is_preview=True): plan_id=%s", plan_id) + preview_tasks = _repo.list_by_source_edit_plan(plan_id) + for pt in preview_tasks: + if ( + getattr(pt, "status", "") == "completed" + and getattr(pt, "is_preview", False) + ): + rendered_storage_key = _get_task_video_url(db, pt.id) or "" + if rendered_storage_key: + logger.info( + "[封面生成] ✅ 步骤4找到预览视频: plan_id=%s task_id=%s url=%s", + plan_id, pt.id, rendered_storage_key[:80], + ) + break + except Exception: + logger.warning( + "[封面生成] 步骤4查找预览任务失败: plan_id=%s", plan_id, exc_info=True, + ) + + # 步骤 5：按 user + template 查找最近的已完成预览任务（兜底） + if not rendered_storage_key: + try: + _repo = SQLAlchemyGenerationTaskRepository(db) + logger.info( + "[封面生成] 步骤5: 通过 user+template 查找预览任务: plan_id=%s template_id=%s", + plan_id, template_id, + ) + preview_tasks = _repo.list_latest_completed_preview( user_id=str(current_user.user.id), template_id=template_id, ) if preview_tasks: - completed_preview = preview_tasks[0] - video_repo = get_generated_video_repository(db) - use_case = ListGeneratedVideosByTaskUseCase(video_repo) - videos = use_case.execute(completed_preview.id) - if videos: - rendered_storage_key = getattr(videos[0], "file_url", "") or "" + rendered_storage_key = _get_task_video_url(db, preview_tasks[0].id) or "" + if rendered_storage_key: logger.info( - "封面视频: 通过 user+template 找到预览任务: plan_id=%s template_id=%s task_id=%s", - plan_id, - template_id, - completed_preview.id, + "[封面生成] ✅ 步骤5找到预览视频: plan_id=%s task_id=%s", + plan_id, preview_tasks[0].id, ) except Exception: logger.warning( - "封面警告: user+template 查找预览任务失败: plan_id=%s template_id=%s", - plan_id, - template_id, - exc_info=True, + "[封面生成] 步骤5 user+template 查找失败: plan_id=%s", plan_id, exc_info=True, ) - # 使用裸 URL（rendered/* 已配置公开读）；找不到渲染视频时不立即报错， - # 因为步骤 E 可以直接从源素材抽帧（历史数据或 Worker 抽帧失败时的兜底） + # 将 storage_key 转换为可访问 URL；找不到视频时不立即报错， + # 因为步骤 E2 可以直接从源素材抽帧（历史数据或 Worker 抽帧失败时的兜底） primary_video_url = None if rendered_storage_key: plan_svc.update_plan_config(plan_id, {"rendered_storage_key": rendered_storage_key}) - try: - if rendered_storage_key.startswith("http"): - primary_video_url = rendered_storage_key - else: - from packages.shared.storage import get_shared_storage_service - - storage_svc = get_shared_storage_service() - primary_video_url = storage_svc.get_url(rendered_storage_key) - if primary_video_url: - import re as _re - - primary_video_url = _re.sub(r"(? Date: Fri, 28 Aug 2026 15:35:01 +0000 Subject: [PATCH 2/7] style: auto-format with black + isort + prettier [skip ci-format-check] --- tests/unit/test_generation_cover.py | 28 +++++++++++++++++++--------- 1 file changed, 19 insertions(+), 9 deletions(-) diff --git a/tests/unit/test_generation_cover.py b/tests/unit/test_generation_cover.py index 042ddaf30..7d31816b3 100644 --- a/tests/unit/test_generation_cover.py +++ b/tests/unit/test_generation_cover.py @@ -1262,8 +1262,10 @@ class TestCoverFromFinalVideo: patch("app.dependencies.get_generated_video_repository", return_value=mock_gv_repo), patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter, patch("packages.shared.mediakit_client.get_mediakit_client") as mock_mk_getter, - patch("app.api.routes.generation_cover._persist_cover_frame", - return_value="https://oss.example.com/covers/final-cover.jpg") as mock_persist, + patch( + "app.api.routes.generation_cover._persist_cover_frame", + return_value="https://oss.example.com/covers/final-cover.jpg", + ) as mock_persist, patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize, ): mock_repo = MagicMock() @@ -1327,8 +1329,10 @@ class TestCoverFromFinalVideo: with ( patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls, patch("packages.shared.mediakit_client.get_mediakit_client") as mock_mk_getter, - patch("app.api.routes.generation_cover._persist_cover_frame", - return_value="https://oss.example.com/covers/c.jpg"), + patch( + "app.api.routes.generation_cover._persist_cover_frame", + return_value="https://oss.example.com/covers/c.jpg", + ), patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize, ): mock_repo = MagicMock() @@ -1349,9 +1353,12 @@ class TestCoverFromFinalVideo: from app.api.routes.generation_cover import generate_cover generate_cover( - body=body, template_id="tpl-1", plan_id="plan-url", + body=body, + template_id="tpl-1", + plan_id="plan-url", services=(mock_template_svc, mock_plan_svc), - db=mock_db, current_user=mock_current_user, + db=mock_db, + current_user=mock_current_user, ) call_kwargs = mock_mk.extract_frames.call_args.kwargs @@ -1399,8 +1406,10 @@ class TestCoverFromFinalVideo: patch("app.api.routes.generation_cover.ListGeneratedVideosByTaskUseCase") as mock_list_videos, patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter, patch("packages.shared.mediakit_client.get_mediakit_client") as mock_mk_getter, - patch("app.api.routes.generation_cover._persist_cover_frame", - return_value="https://oss.example.com/covers/c.jpg"), + patch( + "app.api.routes.generation_cover._persist_cover_frame", + return_value="https://oss.example.com/covers/c.jpg", + ), patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize, ): mock_repo = MagicMock() @@ -1415,6 +1424,7 @@ class TestCoverFromFinalVideo: if task_id == "task-final": return [mock_video_final] return [mock_video_preview] + mock_use_case = MagicMock() mock_use_case.execute.side_effect = get_videos mock_list_videos.return_value = mock_use_case @@ -1452,9 +1462,9 @@ class TestCoverFromFinalVideo: """generated_video_id 关联任务属于其他用户时，返回 403。""" from unittest.mock import MagicMock, patch + import pytest from app.api.routes.generation_cover import GenerateCoverRequest from fastapi import HTTPException - import pytest mock_plan = MagicMock() mock_plan.config = {} -- 2.54.0 From 3b828ab18467f0a9ee4d27de5702b95aabf602e3 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Sat, 29 Aug 2026 00:21:49 +0800 Subject: [PATCH 3/7] =?UTF-8?q?fix:=20=E5=B0=81=E9=9D=A2=E9=80=89=E5=B8=A7?= =?UTF-8?q?=E6=8E=A5=E5=8F=A3=E5=AE=89=E5=85=A8=E5=8A=A0=E5=9B=BA=20?= =?UTF-8?q?=E2=80=94=20=E6=9D=83=E9=99=90=E6=A0=A1=E9=AA=8C=E9=98=B2?= =?UTF-8?q?=E7=BB=95=E8=BF=87=20+=20video=5Furl=20SSRF=20=E7=99=BD?= =?UTF-8?q?=E5=90=8D=E5=8D=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AI Code Review 指出的两个阻塞问题： 1. 权限校验绕过（步骤0a）：原逻辑仅在关联任务存在且 created_by_user_id 非空时才校验归属；若 GeneratedVideo 的 generation_task_id 为空或关联任务被删除，校验被静默跳过。 修复： - 优先校验 GeneratedVideo.user_id 直接归属 - 关联任务存在时校验 task.created_by_user_id - video 无 owner 且关联任务也查不到（归属无法确认）→ 403， 不再静默放行 2. video_url SSRF（步骤0b）：原逻辑直接使用请求体传入的 URL， 攻击者可传入内网/云元数据地址诱导服务端请求。 修复：新增 _is_trusted_media_url() 白名单校验： - 仅允许 http/https 且主机为自家 OSS bucket/endpoint 域名 - 显式拒绝 localhost/127.*/10.*/192.168.*/169.254.*/172.16-31.* - 校验失败静默忽略该 URL，回退后续查找链 附带改进（review 建议项）： - 函数内联 import 全部移至模块顶部（storage/re） - 新增 2 个安全测试：SSRF 内网地址拦截、归属无法确认 403 - 现有 32 个测试适配新的 mock 命名空间，共 34 个测试全过 --- apps/api/app/api/routes/generation_cover.py | 209 ++++++++++++++------ tests/unit/test_generation_cover.py | 164 +++++++++++++-- 2 files changed, 301 insertions(+), 72 deletions(-) diff --git a/apps/api/app/api/routes/generation_cover.py b/apps/api/app/api/routes/generation_cover.py index b42283c52..047c0491c 100644 --- a/apps/api/app/api/routes/generation_cover.py +++ b/apps/api/app/api/routes/generation_cover.py @@ -9,7 +9,9 @@ from __future__ import annotations import logging +import re from typing import Any, List, Optional +from urllib.parse import urlparse from app.auth import AuthenticatedUser, get_current_user from app.dependencies import get_db_session, get_generated_video_repository @@ -24,6 +26,7 @@ from packages.adapters.sqlalchemy_impl.generation_task_repository import ( ) from packages.application import ListGeneratedVideosByTaskUseCase from packages.domain.config_schemas import normalize_plan_config +from packages.shared.storage import get_shared_storage_service from .templates_editor.dependencies import get_draft_plan_id, get_editor_services @@ -53,13 +56,11 @@ class GenerateCoverRequest(BaseModel): ) generated_video_id: Optional[str] = Field( default=None, - description="确认生成产出的最终视频 ID。传入后封面从该视频文件抽帧，" - "而非预览片段。", + description="确认生成产出的最终视频 ID。传入后封面从该视频文件抽帧，而非预览片段。", ) video_url: Optional[str] = Field( default=None, - description="最终视频 URL（兜底）。当 generated_video_id 不可用时，" - "直接从此 URL 对应的视频抽帧。", + description="最终视频 URL（兜底）。当 generated_video_id 不可用时，直接从此 URL 对应的视频抽帧。", ) @@ -131,8 +132,6 @@ def _persist_cover_frame( exc_info=True, ) - from packages.shared.storage import get_shared_storage_service - storage = get_shared_storage_service() cover_key = f"covers/{plan_id}/cover_{uuid.uuid4().hex[:8]}.jpg" storage.upload_file( @@ -150,7 +149,6 @@ def _persist_cover_frame( Path(tmp_path).unlink(missing_ok=True) - def _get_task_video_url(db: Session, task_id: str) -> Optional[str]: """从 GenerationTask 关联的 GeneratedVideo 中获取视频 storage_key / URL.""" try: @@ -172,19 +170,67 @@ def _resolve_storage_key_to_url(storage_key: str) -> Optional[str]: if storage_key.startswith("http"): url = storage_key else: - from packages.shared.storage import get_shared_storage_service - storage_svc = get_shared_storage_service() url = storage_svc.get_url(storage_key) if url: - import re as _re - url = _re.sub(r"(? bool: + """校验 URL 是否指向受信任的存储域名（OSS bucket / 本地存储），防止 SSRF。 + + 用户可通过 video_url 传入视频地址，但服务端（MediaKit）会主动请求该 URL， + 因此必须限制为自家存储域名，拒绝内网地址、元数据地址等任意主机。 + """ + if not url: + return False + try: + parsed = urlparse(url.strip()) + if parsed.scheme not in ("http", "https"): + return False + host = (parsed.hostname or "").lower() + if not host: + return False + # 显式拒绝内网/保留地址 + if host in {"localhost", "0.0.0.0"} or host.startswith(("127.", "10.", "192.168.", "169.254.")): + return False + # 172.16.0.0/12 + try: + parts = [int(p) for p in host.split(".")] + if len(parts) == 4 and parts[0] == 172 and 16 <= parts[1] <= 31: + return False + except ValueError: + pass + # 允许：自家 OSS bucket 域名（.） + try: + storage_svc = get_shared_storage_service() + public_base = getattr(storage_svc, "public_url", "") or "" + if public_base: + from urllib.parse import urlparse as _urlparse + + trusted_host = (_urlparse(public_base).hostname or "").lower() + if trusted_host and (host == trusted_host or host.endswith("." + trusted_host)): + return True + # endpoint 本身（如 oss-cn-hangzhou.aliyuncs.com）及其子域也放行 + ep = getattr(storage_svc, "endpoint", "") or "" + ep_host = ep.split(":")[0].lower() + if ep_host.startswith(("http://", "https://")): + ep_host = _urlparse(ep_host).hostname or "" + if ep_host and (host == ep_host or host.endswith("." + ep_host)): + return True + except Exception: + logger.warning("[封面生成] 存储域名白名单初始化失败，URL 校验从严拒绝", exc_info=True) + return False + return False + except Exception: + logger.warning("[封面生成] video_url 白名单校验异常，从严拒绝: url=%s", url[:80], exc_info=True) + return False + + @router.post("/generate-cover", response_model=GenerateCoverResponse) def generate_cover( body: GenerateCoverRequest, @@ -248,52 +294,78 @@ def generate_cover( if body.generated_video_id: logger.info( "[封面生成] 步骤0a: 使用 generated_video_id: plan_id=%s video_id=%s", - plan_id, body.generated_video_id, + plan_id, + body.generated_video_id, ) try: - from app.dependencies import get_generated_video_repository as _get_gv_repo - - gv_repo = _get_gv_repo(db) + gv_repo = get_generated_video_repository(db) gv = gv_repo.get(body.generated_video_id) if gv: file_url = getattr(gv, "file_url", "") or "" if file_url: - # 权限校验：通过关联的 generation_task 校验归属 + # 权限校验（双重，任何一层确认归属不符即拒绝）： + # 1) GeneratedVideo.user_id 直接归属（老数据可能为空，为空时不据此放行） + gv_owner = (getattr(gv, "user_id", "") or "").strip() + if gv_owner and gv_owner != current_user.user.id: + raise HTTPException(status_code=403, detail="无权访问该视频") + # 2) 关联 generation_task 归属校验；关联任务缺失时不可静默放行： + # 若 video 自身无 owner 信息且关联任务也查不到，拒绝访问 gv_task_id = getattr(gv, "generation_task_id", "") or "" + task0 = None if gv_task_id: try: - _repo0 = SQLAlchemyGenerationTaskRepository(db) - task0 = _repo0.get(gv_task_id) - if ( - task0 - and task0.created_by_user_id - and task0.created_by_user_id != current_user.user.id - ): - raise HTTPException(status_code=403, detail="无权访问该视频") - except HTTPException: - raise + task0 = SQLAlchemyGenerationTaskRepository(db).get(gv_task_id) except Exception: - pass + logger.warning( + "[封面生成] 步骤0a关联任务查询异常: plan_id=%s task_id=%s", + plan_id, + gv_task_id, + exc_info=True, + ) + if task0 is not None: + task_owner = (getattr(task0, "created_by_user_id", "") or "").strip() + if task_owner and task_owner != current_user.user.id: + raise HTTPException(status_code=403, detail="无权访问该视频") + elif not gv_owner: + # video 无 owner 且关联任务不存在/无法确认归属 → 拒绝，防止越权 + logger.warning( + "[封面生成] 步骤0a视频归属无法确认，拒绝访问: plan_id=%s video_id=%s", + plan_id, + body.generated_video_id, + ) + raise HTTPException(status_code=403, detail="无权访问该视频") rendered_storage_key = file_url logger.info( "[封面生成] ✅ 步骤0a找到最终成片: plan_id=%s video_id=%s url=%s", - plan_id, body.generated_video_id, file_url[:80], + plan_id, + body.generated_video_id, + file_url[:80], ) except HTTPException: raise except Exception: logger.warning( "[封面生成] 步骤0a查找视频失败: plan_id=%s video_id=%s", - plan_id, body.generated_video_id, exc_info=True, + plan_id, + body.generated_video_id, + exc_info=True, ) - # 0b：直接使用 video_url（兜底） + # 0b：直接使用 video_url（兜底）— 必须通过存储域名白名单校验，防止 SSRF if not rendered_storage_key and body.video_url: - logger.info( - "[封面生成] 步骤0b: 使用请求体传入的 video_url: plan_id=%s url=%s", - plan_id, body.video_url[:80], - ) - rendered_storage_key = body.video_url + if _is_trusted_media_url(body.video_url): + logger.info( + "[封面生成] 步骤0b: 使用请求体传入的 video_url(白名单通过): plan_id=%s url=%s", + plan_id, + body.video_url[:80], + ) + rendered_storage_key = body.video_url + else: + logger.warning( + "[封面生成] 步骤0b: video_url 不在受信任存储域名白名单内，已忽略: plan_id=%s url=%s", + plan_id, + body.video_url[:80], + ) # 步骤 2：通过 plan.config.generation_task_id 查找 if not rendered_storage_key: @@ -301,7 +373,8 @@ def generate_cover( if generation_task_id: logger.info( "[封面生成] 步骤2: 通过 plan.config.generation_task_id 查找: plan_id=%s task_id=%s", - plan_id, generation_task_id, + plan_id, + generation_task_id, ) try: _repo = SQLAlchemyGenerationTaskRepository(db) @@ -311,11 +384,15 @@ def generate_cover( if rendered_storage_key: logger.info( "[封面生成] ✅ 步骤2找到视频: plan_id=%s task_id=%s url=%s", - plan_id, generation_task_id, rendered_storage_key[:80], + plan_id, + generation_task_id, + rendered_storage_key[:80], ) except Exception: logger.warning( - "[封面生成] 步骤2查找失败: plan_id=%s", plan_id, exc_info=True, + "[封面生成] 步骤2查找失败: plan_id=%s", + plan_id, + exc_info=True, ) # 步骤 3：通过 source_edit_plan_id 查找已完成「最终成片」任务（is_preview=False） @@ -325,20 +402,21 @@ def generate_cover( logger.info("[封面生成] 步骤3: 查找最终成片任务(is_preview=False): plan_id=%s", plan_id) all_tasks = _repo.list_by_source_edit_plan(plan_id) for pt in all_tasks: - if ( - getattr(pt, "status", "") == "completed" - and not getattr(pt, "is_preview", False) - ): + if getattr(pt, "status", "") == "completed" and not getattr(pt, "is_preview", False): rendered_storage_key = _get_task_video_url(db, pt.id) or "" if rendered_storage_key: logger.info( "[封面生成] ✅ 步骤3找到最终成片: plan_id=%s task_id=%s url=%s", - plan_id, pt.id, rendered_storage_key[:80], + plan_id, + pt.id, + rendered_storage_key[:80], ) break except Exception: logger.warning( - "[封面生成] 步骤3查找最终成片失败: plan_id=%s", plan_id, exc_info=True, + "[封面生成] 步骤3查找最终成片失败: plan_id=%s", + plan_id, + exc_info=True, ) # 步骤 4：兼容回退 — 通过 source_edit_plan_id 查找已完成预览任务 @@ -348,20 +426,21 @@ def generate_cover( logger.info("[封面生成] 步骤4: 回退查找预览任务(is_preview=True): plan_id=%s", plan_id) preview_tasks = _repo.list_by_source_edit_plan(plan_id) for pt in preview_tasks: - if ( - getattr(pt, "status", "") == "completed" - and getattr(pt, "is_preview", False) - ): + if getattr(pt, "status", "") == "completed" and getattr(pt, "is_preview", False): rendered_storage_key = _get_task_video_url(db, pt.id) or "" if rendered_storage_key: logger.info( "[封面生成] ✅ 步骤4找到预览视频: plan_id=%s task_id=%s url=%s", - plan_id, pt.id, rendered_storage_key[:80], + plan_id, + pt.id, + rendered_storage_key[:80], ) break except Exception: logger.warning( - "[封面生成] 步骤4查找预览任务失败: plan_id=%s", plan_id, exc_info=True, + "[封面生成] 步骤4查找预览任务失败: plan_id=%s", + plan_id, + exc_info=True, ) # 步骤 5：按 user + template 查找最近的已完成预览任务（兜底） @@ -370,7 +449,8 @@ def generate_cover( _repo = SQLAlchemyGenerationTaskRepository(db) logger.info( "[封面生成] 步骤5: 通过 user+template 查找预览任务: plan_id=%s template_id=%s", - plan_id, template_id, + plan_id, + template_id, ) preview_tasks = _repo.list_latest_completed_preview( user_id=str(current_user.user.id), @@ -381,11 +461,14 @@ def generate_cover( if rendered_storage_key: logger.info( "[封面生成] ✅ 步骤5找到预览视频: plan_id=%s task_id=%s", - plan_id, preview_tasks[0].id, + plan_id, + preview_tasks[0].id, ) except Exception: logger.warning( - "[封面生成] 步骤5 user+template 查找失败: plan_id=%s", plan_id, exc_info=True, + "[封面生成] 步骤5 user+template 查找失败: plan_id=%s", + plan_id, + exc_info=True, ) # 将 storage_key 转换为可访问 URL；找不到视频时不立即报错， @@ -430,9 +513,7 @@ def generate_cover( # 步骤 A2：通过 generated_video_id 查找其关联任务的 cover_url if not cover_url_from_task and body.generated_video_id: try: - from app.dependencies import get_generated_video_repository as _get_gv_repo - - gv_repo = _get_gv_repo(db) + gv_repo = get_generated_video_repository(db) gv = gv_repo.get(body.generated_video_id) if gv: gv_task_id = getattr(gv, "generation_task_id", "") or "" @@ -442,12 +523,16 @@ def generate_cover( cover_url_from_task = task_a2.cover_url logger.info( "[封面生成] 封面(步骤A2-video-task): plan_id=%s video_id=%s url=%s", - plan_id, body.generated_video_id, cover_url_from_task[:80], + plan_id, + body.generated_video_id, + cover_url_from_task[:80], ) except Exception: logger.warning( "[封面生成] 步骤A2读取 cover_url 失败: plan_id=%s video_id=%s", - plan_id, body.generated_video_id, exc_info=True, + plan_id, + body.generated_video_id, + exc_info=True, ) # 步骤 B：通过 source_edit_plan_id 查找关联任务的 cover_url @@ -465,7 +550,9 @@ def generate_cover( cover_url_from_task = pt.cover_url logger.info( "[封面生成] 封面(步骤B-final): plan_id=%s task_id=%s url=%s", - plan_id, pt.id, cover_url_from_task[:80], + plan_id, + pt.id, + cover_url_from_task[:80], ) break # 再找预览 @@ -479,7 +566,9 @@ def generate_cover( cover_url_from_task = pt.cover_url logger.info( "[封面生成] 封面(步骤B-preview): plan_id=%s task_id=%s url=%s", - plan_id, pt.id, cover_url_from_task[:80], + plan_id, + pt.id, + cover_url_from_task[:80], ) break except Exception: diff --git a/tests/unit/test_generation_cover.py b/tests/unit/test_generation_cover.py index 7d31816b3..698809a65 100644 --- a/tests/unit/test_generation_cover.py +++ b/tests/unit/test_generation_cover.py @@ -181,7 +181,7 @@ class TestUnifiedCoverPipelineEndpoint: with ( patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls, - patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter, + patch("app.api.routes.generation_cover.get_shared_storage_service") as mock_storage_getter, patch("packages.shared.mediakit_client.get_mediakit_client") as mock_mk_getter, ): mock_repo = MagicMock() @@ -449,7 +449,7 @@ class TestUnifiedCoverPipelineEndpoint: with ( patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls, - patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter, + patch("app.api.routes.generation_cover.get_shared_storage_service") as mock_storage_getter, ): mock_repo = MagicMock() mock_repo.get.return_value = None @@ -590,7 +590,7 @@ class TestSourceEditPlanFallback: patch("app.api.routes.generation_cover.get_generated_video_repository") as mock_video_repo, patch("app.api.routes.generation_cover.ListGeneratedVideosByTaskUseCase") as mock_usecase_cls, patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize, - patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter, + patch("app.api.routes.generation_cover.get_shared_storage_service") as mock_storage_getter, ): mock_repo = MagicMock() mock_repo.get.return_value = None @@ -657,7 +657,7 @@ class TestSourceEditPlanFallback: patch("app.api.routes.generation_cover.get_generated_video_repository") as mock_video_repo, patch("app.api.routes.generation_cover.ListGeneratedVideosByTaskUseCase") as mock_usecase_cls, patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize, - patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter, + patch("app.api.routes.generation_cover.get_shared_storage_service") as mock_storage_getter, ): mock_repo = MagicMock() mock_repo.get.return_value = None @@ -707,9 +707,9 @@ class TestStrayLoggerRemoved: source = inspect.getsource(generation_cover) # The stray call was logger.info(\n plan_id,\n generation_task_id,\n) # with no format string — should not exist - assert ( - "logger.info(\n plan_id," not in source - ), "Stray logger.info(plan_id, generation_task_id) should be removed" + assert "logger.info(\n plan_id," not in source, ( + "Stray logger.info(plan_id, generation_task_id) should be removed" + ) class TestUploadCoverType: @@ -1177,7 +1177,7 @@ class TestUploadCoverType: return_value=mock_asset_repo, ), patch("packages.shared.mediakit_client.get_mediakit_client", return_value=mock_mk), - patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter, + patch("app.api.routes.generation_cover.get_shared_storage_service") as mock_storage_getter, ): mock_repo = MagicMock() mock_repo.get.return_value = None @@ -1238,6 +1238,7 @@ class TestCoverFromFinalVideo: mock_gv = MagicMock() mock_gv.file_url = "rendered/final/video.mp4" mock_gv.generation_task_id = "task-final-001" + mock_gv.user_id = "user-1" # 最终成片任务 mock_final_task = MagicMock() @@ -1259,8 +1260,11 @@ class TestCoverFromFinalVideo: with ( patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls, - patch("app.dependencies.get_generated_video_repository", return_value=mock_gv_repo), - patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter, + patch( + "app.api.routes.generation_cover.get_generated_video_repository", + return_value=mock_gv_repo, + ), + patch("app.api.routes.generation_cover.get_shared_storage_service") as mock_storage_getter, patch("packages.shared.mediakit_client.get_mediakit_client") as mock_mk_getter, patch( "app.api.routes.generation_cover._persist_cover_frame", @@ -1276,6 +1280,8 @@ class TestCoverFromFinalVideo: mock_storage_svc = MagicMock() mock_storage_svc.get_url.return_value = "https://oss.example.com/rendered/final/video.mp4" + mock_storage_svc.public_url = "https://oss.example.com" + mock_storage_svc.endpoint = "oss.example.com" mock_storage_getter.return_value = mock_storage_svc mock_mk = MagicMock() @@ -1328,6 +1334,7 @@ class TestCoverFromFinalVideo: with ( patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls, + patch("app.api.routes.generation_cover.get_shared_storage_service") as mock_storage_getter, patch("packages.shared.mediakit_client.get_mediakit_client") as mock_mk_getter, patch( "app.api.routes.generation_cover._persist_cover_frame", @@ -1341,6 +1348,11 @@ class TestCoverFromFinalVideo: mock_repo.list_latest_completed_preview.return_value = [] mock_repo_cls.return_value = mock_repo + mock_storage_svc = MagicMock() + mock_storage_svc.public_url = "https://oss.example.com" + mock_storage_svc.endpoint = "oss.example.com" + mock_storage_getter.return_value = mock_storage_svc + mock_mk = MagicMock() mock_mk.is_available = True mock_mk.extract_frames.return_value = [{"image_url": "https://mk/f.jpg"}] @@ -1404,7 +1416,7 @@ class TestCoverFromFinalVideo: with ( patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls, patch("app.api.routes.generation_cover.ListGeneratedVideosByTaskUseCase") as mock_list_videos, - patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter, + patch("app.api.routes.generation_cover.get_shared_storage_service") as mock_storage_getter, patch("packages.shared.mediakit_client.get_mediakit_client") as mock_mk_getter, patch( "app.api.routes.generation_cover._persist_cover_frame", @@ -1431,6 +1443,8 @@ class TestCoverFromFinalVideo: mock_storage_svc = MagicMock() mock_storage_svc.get_url.side_effect = lambda key: f"https://oss.example.com/{key}" + mock_storage_svc.public_url = "https://oss.example.com" + mock_storage_svc.endpoint = "oss.example.com" mock_storage_getter.return_value = mock_storage_svc mock_mk = MagicMock() @@ -1475,6 +1489,7 @@ class TestCoverFromFinalVideo: mock_gv = MagicMock() mock_gv.file_url = "rendered/other/video.mp4" mock_gv.generation_task_id = "task-other" + mock_gv.user_id = "" # 老数据无 user_id，走关联任务归属校验 mock_other_task = MagicMock() mock_other_task.created_by_user_id = "other-user" @@ -1493,7 +1508,10 @@ class TestCoverFromFinalVideo: with ( patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls, - patch("app.dependencies.get_generated_video_repository", return_value=mock_gv_repo), + patch( + "app.api.routes.generation_cover.get_generated_video_repository", + return_value=mock_gv_repo, + ), ): mock_repo = MagicMock() mock_repo.get.return_value = mock_other_task @@ -1511,3 +1529,125 @@ class TestCoverFromFinalVideo: current_user=mock_current_user, ) assert exc_info.value.status_code == 403 + + def test_cover_video_url_ssrf_blocked(self): + """video_url 指向内网/非白名单域名时被忽略，不向其发起抽帧请求。""" + from unittest.mock import MagicMock, patch + + from app.api.routes.generation_cover import GenerateCoverRequest + + mock_plan = MagicMock() + mock_plan.config = {} + mock_plan_svc = MagicMock() + mock_plan_svc.get_plan_or_raise.return_value = mock_plan + mock_template_svc = MagicMock() + mock_db = MagicMock() + mock_current_user = MagicMock() + mock_current_user.user.id = "user-1" + + # SSRF 攻击载荷：内网元数据地址 + body = GenerateCoverRequest( + cover_type="ai_frame", + video_url="http://100.100.100.200/latest/meta-data/", + ) + + with ( + patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls, + patch("app.api.routes.generation_cover.get_shared_storage_service") as mock_storage_getter, + patch("packages.shared.mediakit_client.get_mediakit_client") as mock_mk_getter, + patch("app.api.routes.generation_cover._persist_cover_frame") as mock_persist, + patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize, + ): + mock_repo = MagicMock() + mock_repo.get.return_value = None + mock_repo.list_by_source_edit_plan.return_value = [] + mock_repo.list_latest_completed_preview.return_value = [] + mock_repo_cls.return_value = mock_repo + + mock_storage_svc = MagicMock() + mock_storage_svc.public_url = "https://xiaoxia-media.oss-cn-hangzhou.aliyuncs.com" + mock_storage_svc.endpoint = "oss-cn-hangzhou.aliyuncs.com" + mock_storage_svc.get_url.side_effect = lambda k: f"https://xiaoxia-media.oss-cn-hangzhou.aliyuncs.com/{k}" + mock_storage_getter.return_value = mock_storage_svc + + mock_mk = MagicMock() + mock_mk.is_available = True + mock_mk.extract_frames.return_value = [{"image_url": "https://mk/f.jpg"}] + mock_mk_getter.return_value = mock_mk + + mock_normalize.return_value = {"cover": {"type": "ai_frame", "image_url": "https://mk/f.jpg"}} + + from app.api.routes.generation_cover import generate_cover + from fastapi import HTTPException + + # 内网 URL 被白名单拦截后，无任何可用视频源 → 400（而不是向内网发请求） + with pytest.raises(HTTPException) as exc_info: + generate_cover( + body=body, + template_id="tpl-1", + plan_id="plan-ssrf", + services=(mock_template_svc, mock_plan_svc), + db=mock_db, + current_user=mock_current_user, + ) + assert exc_info.value.status_code == 400 + # MediaKit 从未被要求抽帧该内网地址 + if mock_mk.extract_frames.called: + called_url = mock_mk.extract_frames.call_args.kwargs.get("video_url", "") + assert "100.100.100.200" not in called_url + assert "meta-data" not in called_url + + def test_cover_generated_video_ownership_unverifiable_denied(self): + """video 无 user_id 且关联任务不存在时，归属无法确认 → 403（防权限绕过）。""" + from unittest.mock import MagicMock, patch + + import pytest + from app.api.routes.generation_cover import GenerateCoverRequest + from fastapi import HTTPException + + mock_plan = MagicMock() + mock_plan.config = {} + mock_plan_svc = MagicMock() + mock_plan_svc.get_plan_or_raise.return_value = mock_plan + mock_template_svc = MagicMock() + + mock_gv = MagicMock() + mock_gv.file_url = "rendered/mystery/video.mp4" + mock_gv.generation_task_id = "task-gone" # 关联任务已删除 + mock_gv.user_id = "" # 老数据无 owner + + mock_gv_repo = MagicMock() + mock_gv_repo.get.return_value = mock_gv + + mock_db = MagicMock() + mock_current_user = MagicMock() + mock_current_user.user.id = "user-1" + + body = GenerateCoverRequest( + cover_type="ai_frame", + generated_video_id="gv-mystery", + ) + + with ( + patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls, + patch( + "app.api.routes.generation_cover.get_generated_video_repository", + return_value=mock_gv_repo, + ), + ): + mock_repo = MagicMock() + mock_repo.get.return_value = None # 关联任务查不到 + mock_repo_cls.return_value = mock_repo + + from app.api.routes.generation_cover import generate_cover + + with pytest.raises(HTTPException) as exc_info: + generate_cover( + body=body, + template_id="tpl-1", + plan_id="plan-orphan", + services=(mock_template_svc, mock_plan_svc), + db=mock_db, + current_user=mock_current_user, + ) + assert exc_info.value.status_code == 403 -- 2.54.0 From e5e18ef26977ae1013c4a3aeee2c4abfee0d829e Mon Sep 17 00:00:00 2001 From: CI Bot Date: Fri, 28 Aug 2026 16:27:56 +0000 Subject: [PATCH 4/7] style: auto-format with black + isort + prettier [skip ci-format-check] --- tests/unit/test_generation_cover.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/unit/test_generation_cover.py b/tests/unit/test_generation_cover.py index 698809a65..225b01b7e 100644 --- a/tests/unit/test_generation_cover.py +++ b/tests/unit/test_generation_cover.py @@ -707,9 +707,9 @@ class TestStrayLoggerRemoved: source = inspect.getsource(generation_cover) # The stray call was logger.info(\n plan_id,\n generation_task_id,\n) # with no format string — should not exist - assert "logger.info(\n plan_id," not in source, ( - "Stray logger.info(plan_id, generation_task_id) should be removed" - ) + assert ( + "logger.info(\n plan_id," not in source + ), "Stray logger.info(plan_id, generation_task_id) should be removed" class TestUploadCoverType: -- 2.54.0 From bde37af2bb28e79ba965c166da4a19136f49d264 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Sat, 29 Aug 2026 00:39:05 +0800 Subject: [PATCH 5/7] =?UTF-8?q?fix:=20SSRF=20=E7=99=BD=E5=90=8D=E5=8D=95?= =?UTF-8?q?=E5=8A=A0=E5=9B=BA=20=E2=80=94=20=E4=BF=AE=E5=A4=8D=20endpoint?= =?UTF-8?q?=20=E8=A7=A3=E6=9E=90=E7=BB=95=E8=BF=87=20+=20=E8=A1=A5?= =?UTF-8?q?=E5=85=85=20IPv6=20=E5=86=85=E7=BD=91=E6=8B=A6=E6=88=AA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AI Code Review 第二轮指出的两个阻塞问题： 1. endpoint 主机名解析绕过：原逻辑 ep.split(':')[0] 在 endpoint 带 scheme（http://host:9000）时取到 'http'，虽白名单顺序使 public_url 先生效，但 endpoint 分支可能错误匹配 '.http' 后缀。 修复：新增 _endpoint_host() 统一用 urlparse 提取主机名， 兼容有无 scheme、带端口等各种配置形式。 2. 缺失 IPv6 内网地址校验：[::1]、fe80::/10（链路本地）、 fc00::/7（唯一本地）等 IPv6 本地地址未拦截。 修复：补充 IPv6 回环/链路本地/ULA 地址显式拒绝。 附带： - 移除函数内重复的 urlparse 导入，统一使用顶部导入 - 新增 2 个白名单单元测试（IPv4/IPv6/元数据拦截 + scheme 解析） - 共 36 个测试全部通过 --- apps/api/app/api/routes/generation_cover.py | 47 ++++++++++----- tests/unit/test_generation_cover.py | 63 ++++++++++++++++++++- 2 files changed, 92 insertions(+), 18 deletions(-) diff --git a/apps/api/app/api/routes/generation_cover.py b/apps/api/app/api/routes/generation_cover.py index 047c0491c..fe6f0adf5 100644 --- a/apps/api/app/api/routes/generation_cover.py +++ b/apps/api/app/api/routes/generation_cover.py @@ -180,6 +180,17 @@ def _resolve_storage_key_to_url(storage_key: str) -> Optional[str]: return None +def _endpoint_host(value: str) -> str: + """从 endpoint / URL 字符串中安全提取主机名（兼容有无 scheme 两种配置）。""" + v = (value or "").strip().lower() + if not v: + return "" + if "://" in v: + return (urlparse(v).hostname or "").lower() + # 无 scheme：去掉可能的端口（host:port），urlparse 补 // 以正确解析 + return (urlparse("//" + v).hostname or "").lower() + + def _is_trusted_media_url(url: str) -> bool: """校验 URL 是否指向受信任的存储域名（OSS bucket / 本地存储），防止 SSRF。 @@ -195,8 +206,17 @@ def _is_trusted_media_url(url: str) -> bool: host = (parsed.hostname or "").lower() if not host: return False - # 显式拒绝内网/保留地址 - if host in {"localhost", "0.0.0.0"} or host.startswith(("127.", "10.", "192.168.", "169.254.")): + # 显式拒绝内网/保留地址（IPv4 + IPv6） + if host in {"localhost", "0.0.0.0", "::", "::1"}: + return False + if host.startswith(("127.", "10.", "192.168.", "169.254.")): + return False + # IPv6 本地/链路本地/唯一本地地址：[::1] / fe80:: / fc00::/7 + if ":" in host and ( + host == "::1" + or host.startswith(("fe80", "fe90", "fea0", "feb0", "fec0", "fed0", "fee0", "fef0")) + or host.startswith(("fc", "fd")) + ): return False # 172.16.0.0/12 try: @@ -205,22 +225,19 @@ def _is_trusted_media_url(url: str) -> bool: return False except ValueError: pass - # 允许：自家 OSS bucket 域名（.） + # 允许：自家 OSS bucket 域名（.）或 endpoint 自身及其子域 try: storage_svc = get_shared_storage_service() + trusted_hosts = set() public_base = getattr(storage_svc, "public_url", "") or "" - if public_base: - from urllib.parse import urlparse as _urlparse - - trusted_host = (_urlparse(public_base).hostname or "").lower() - if trusted_host and (host == trusted_host or host.endswith("." + trusted_host)): - return True - # endpoint 本身（如 oss-cn-hangzhou.aliyuncs.com）及其子域也放行 - ep = getattr(storage_svc, "endpoint", "") or "" - ep_host = ep.split(":")[0].lower() - if ep_host.startswith(("http://", "https://")): - ep_host = _urlparse(ep_host).hostname or "" - if ep_host and (host == ep_host or host.endswith("." + ep_host)): + h1 = _endpoint_host(public_base) + if h1: + trusted_hosts.add(h1) + h2 = _endpoint_host(getattr(storage_svc, "endpoint", "") or "") + if h2: + trusted_hosts.add(h2) + for trusted in trusted_hosts: + if host == trusted or host.endswith("." + trusted): return True except Exception: logger.warning("[封面生成] 存储域名白名单初始化失败，URL 校验从严拒绝", exc_info=True) diff --git a/tests/unit/test_generation_cover.py b/tests/unit/test_generation_cover.py index 225b01b7e..1398c5cba 100644 --- a/tests/unit/test_generation_cover.py +++ b/tests/unit/test_generation_cover.py @@ -707,9 +707,9 @@ class TestStrayLoggerRemoved: source = inspect.getsource(generation_cover) # The stray call was logger.info(\n plan_id,\n generation_task_id,\n) # with no format string — should not exist - assert ( - "logger.info(\n plan_id," not in source - ), "Stray logger.info(plan_id, generation_task_id) should be removed" + assert "logger.info(\n plan_id," not in source, ( + "Stray logger.info(plan_id, generation_task_id) should be removed" + ) class TestUploadCoverType: @@ -1651,3 +1651,60 @@ class TestCoverFromFinalVideo: current_user=mock_current_user, ) assert exc_info.value.status_code == 403 + + def test_is_trusted_media_url_blocks_internal_and_ipv6(self): + """白名单函数：内网 IPv4/IPv6/元数据地址一律拒绝，自家 OSS 域名放行。""" + from unittest.mock import MagicMock, patch + + from app.api.routes.generation_cover import _is_trusted_media_url + + mock_storage = MagicMock() + mock_storage.public_url = "https://xiaoxia-media.oss-cn-hangzhou.aliyuncs.com" + mock_storage.endpoint = "oss-cn-hangzhou.aliyuncs.com" + + with patch( + "app.api.routes.generation_cover.get_shared_storage_service", + return_value=mock_storage, + ): + # 内网 / 元数据 / IPv6 本地地址全部拒绝 + for bad in [ + "http://127.0.0.1/admin", + "http://10.0.0.5/video.mp4", + "http://192.168.1.1/video.mp4", + "http://172.16.0.1/video.mp4", + "http://169.254.169.254/latest/meta-data/", + "http://[::1]:8080/video.mp4", + "http://[fe80::1]/video.mp4", + "http://[fc00::1]/video.mp4", + "http://localhost/x", + "ftp://oss-cn-hangzhou.aliyuncs.com/a.mp4", + "", + ]: + assert _is_trusted_media_url(bad) is False, f"应拒绝: {bad}" + + # 自家 OSS 域名（含签名 URL 子路径、bucket 域名）放行 + for good in [ + "https://xiaoxia-media.oss-cn-hangzhou.aliyuncs.com/rendered/final/v.mp4", + "https://xiaoxia-media.oss-cn-hangzhou.aliyuncs.com/rendered/v.mp4?Expires=123&Signature=abc", + ]: + assert _is_trusted_media_url(good) is True, f"应放行: {good}" + + def test_is_trusted_media_url_endpoint_with_scheme_parsed(self): + """endpoint 配置带 http:// 前缀时也能正确提取主机名，不出现 .http 后缀绕过。""" + from unittest.mock import MagicMock, patch + + from app.api.routes.generation_cover import _is_trusted_media_url + + mock_storage = MagicMock() + mock_storage.public_url = "http://oss.internal.example.com:9000" + mock_storage.endpoint = "http://oss.internal.example.com:9000" + + with patch( + "app.api.routes.generation_cover.get_shared_storage_service", + return_value=mock_storage, + ): + # 正确域名放行 + assert _is_trusted_media_url("http://oss.internal.example.com:9000/a/b.mp4") is True + # 伪造后缀域名必须拒绝（修复前 split(':')[0] 会取到 'http' 导致绕过） + assert _is_trusted_media_url("http://evil-http.com/x.mp4") is False + assert _is_trusted_media_url("http://evil.http/x.mp4") is False -- 2.54.0 From cf7e295f35989455fdc885ca1ae75bdca195c645 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Sat, 29 Aug 2026 00:49:11 +0800 Subject: [PATCH 6/7] =?UTF-8?q?fix:=20SSRF=20=E5=86=85=E7=BD=91=E5=88=A4?= =?UTF-8?q?=E5=AE=9A=E6=94=B9=E7=94=A8=20ipaddress=20=E6=A0=87=E5=87=86?= =?UTF-8?q?=E5=BA=93=EF=BC=8C=E8=A6=86=E7=9B=96=E5=85=A8=E9=83=A8=20IPv4/I?= =?UTF-8?q?Pv6=20=E7=A7=81=E6=9C=89=E6=AE=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AI Code Review 第三轮指出：IPv6 链路本地地址使用字符串前缀匹配 （fe80/fe90/...）不严谨且易漏。 修复：新增 _is_private_or_reserved_host()，统一用标准库 ipaddress.ip_address().is_private/is_loopback/is_link_local/ is_reserved 判定，一次性准确覆盖： - IPv4: 10/8、172.16/12、192.168/16、127/8、169.254/16、0.0.0.0 - IPv6: ::1、fc00::/7(ULA)、fe80::/10(链路本地)、保留段 删除手工字符串切片逻辑，可读性和正确性都更好。 36 个测试全部通过。 --- apps/api/app/api/routes/generation_cover.py | 38 +++++++++++---------- 1 file changed, 20 insertions(+), 18 deletions(-) diff --git a/apps/api/app/api/routes/generation_cover.py b/apps/api/app/api/routes/generation_cover.py index fe6f0adf5..45c5d27fb 100644 --- a/apps/api/app/api/routes/generation_cover.py +++ b/apps/api/app/api/routes/generation_cover.py @@ -8,6 +8,7 @@ from __future__ import annotations +import ipaddress import logging import re from typing import Any, List, Optional @@ -191,6 +192,23 @@ def _endpoint_host(value: str) -> str: return (urlparse("//" + v).hostname or "").lower() +def _is_private_or_reserved_host(host: str) -> bool: + """判断主机名是否为内网/回环/链路本地/保留地址（IPv4 与 IPv6 统一处理）。 + + 使用标准库 ipaddress 判定；非 IP 主机名（如 localhost）单独处理。 + """ + h = host.strip().lower() + if h in {"localhost", "0.0.0.0", "::", "::1"}: + return True + try: + addr = ipaddress.ip_address(h) + # is_private 覆盖 10/8、172.16/12、192.168/16、127/8、169.254/16、 + # ::1、fc00::/7、fe80::/10 等全部私有/保留段 + return bool(addr.is_private or addr.is_loopback or addr.is_link_local or addr.is_reserved) + except ValueError: + return False + + def _is_trusted_media_url(url: str) -> bool: """校验 URL 是否指向受信任的存储域名（OSS bucket / 本地存储），防止 SSRF。 @@ -206,25 +224,9 @@ def _is_trusted_media_url(url: str) -> bool: host = (parsed.hostname or "").lower() if not host: return False - # 显式拒绝内网/保留地址（IPv4 + IPv6） - if host in {"localhost", "0.0.0.0", "::", "::1"}: + # 拒绝一切内网/回环/链路本地/保留地址（IPv4 + IPv6，标准库判定） + if _is_private_or_reserved_host(host): return False - if host.startswith(("127.", "10.", "192.168.", "169.254.")): - return False - # IPv6 本地/链路本地/唯一本地地址：[::1] / fe80:: / fc00::/7 - if ":" in host and ( - host == "::1" - or host.startswith(("fe80", "fe90", "fea0", "feb0", "fec0", "fed0", "fee0", "fef0")) - or host.startswith(("fc", "fd")) - ): - return False - # 172.16.0.0/12 - try: - parts = [int(p) for p in host.split(".")] - if len(parts) == 4 and parts[0] == 172 and 16 <= parts[1] <= 31: - return False - except ValueError: - pass # 允许：自家 OSS bucket 域名（.）或 endpoint 自身及其子域 try: storage_svc = get_shared_storage_service() -- 2.54.0 From f9d243a5af4b9b85787380d9d87d9ea6b02d2278 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Fri, 28 Aug 2026 16:59:07 +0000 Subject: [PATCH 7/7] style: auto-format with black + isort + prettier [skip ci-format-check] --- tests/unit/test_generation_cover.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/unit/test_generation_cover.py b/tests/unit/test_generation_cover.py index 1398c5cba..e7065b34d 100644 --- a/tests/unit/test_generation_cover.py +++ b/tests/unit/test_generation_cover.py @@ -707,9 +707,9 @@ class TestStrayLoggerRemoved: source = inspect.getsource(generation_cover) # The stray call was logger.info(\n plan_id,\n generation_task_id,\n) # with no format string — should not exist - assert "logger.info(\n plan_id," not in source, ( - "Stray logger.info(plan_id, generation_task_id) should be removed" - ) + assert ( + "logger.info(\n plan_id," not in source + ), "Stray logger.info(plan_id, generation_task_id) should be removed" class TestUploadCoverType: -- 2.54.0