fix: 封面选帧接口安全加固 — 权限校验防绕过 + video_url SSRF 白名单

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

@@ -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"(?<!:)//", "/", url)
+            url = re.sub(r"(?<!:)//", "/", url)
         return url
     except Exception as e:
         logger.warning("[封面生成] storage_key 转 URL 失败: key=%s err=%s", storage_key, e)
         return None
 
 
+def _is_trusted_media_url(url: str) -> 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 域名（<bucket>.<endpoint>）
+        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:

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