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