From bde37af2bb28e79ba965c166da4a19136f49d264 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Sat, 29 Aug 2026 00:39:05 +0800 Subject: [PATCH] =?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