fix: SSRF 白名单加固 — 修复 endpoint 解析绕过 + 补充 IPv6 内网拦截

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

@@ -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 域名（<bucket>.<endpoint>）
+        # 允许：自家 OSS bucket 域名（<bucket>.<endpoint>）或 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)

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