fix(image): 图片跳转校验沿用已配置的非公网放行网段 (#7110)

app/application/configuration.py

@@ -429,6 +429,7 @@ class ChainRuntimeConfig:
     :param wallpaper_image_url: 静态壁纸地址，可使用前端可访问的本地路径或 URL
     :param customize_wallpaper_api_url: 自定义壁纸api地址
     :param security_image_suffixes: 允许的图片文件后缀格式
+    :param image_proxy_allowed_private_ranges: 图片代理允许访问的非公网 IP/CIDR
     :param cache_path: 运行缓存文件根目录
     :param global_image_cache_days: 全局图片缓存保留天数
     """
@@ -516,6 +517,7 @@ class ChainRuntimeConfig:
     wallpaper_image_url: Optional[str] = None
     customize_wallpaper_api_url: Optional[str] = None
     security_image_suffixes: tuple[str, ...] = ()
+    image_proxy_allowed_private_ranges: tuple[str, ...] = ()
     cache_path: Path = Path(".")
     global_image_cache_days: int = 7
 

app/application/image.py

@@ -313,6 +313,27 @@ def _image_ports_snapshot() -> tuple[ImageTransport, InternalAddressPort]:
     return transport, internal_address
 
 
+def _redirect_private_ranges() -> tuple[str, ...]:
+    """
+    读取默认重定向校验使用的非公网放行网段。
+
+    图片域名的首个请求由调用方（图片代理端点）带上
+    ``IMAGE_PROXY_ALLOWED_PRIVATE_RANGES`` 完成校验，但内部调用方（推荐海报
+    预缓存、通知配图）不传 ``redirect_validator``，重定向会退回本模块的默认
+    校验。默认校验若固定不放行任何非公网网段，就会与首个请求的判定不一致：
+    使用 fake-ip / 分流 DNS 时域名解析到保留地址，首个请求放行、跳转却被
+    自己拦掉。此处沿用同一份配置，让两次校验口径一致。
+
+    :return: 已配置的非公网 IP/CIDR 元组；运行时配置尚未装配时返回空元组，
+        保持默认拒绝
+    """
+    try:
+        snapshot = get_chain_runtime_config_snapshot()
+    except RuntimeError:
+        return ()
+    return tuple(snapshot.image_proxy_allowed_private_ranges)
+
+
 def _empty_wallpaper_provider() -> Optional[str]:
     """
     在启动组合根尚未装配壁纸来源时返回空结果。
@@ -764,7 +785,7 @@ class ImageHelper(metaclass=Singleton):
         redirect_validator: Optional[Callable[[str], bool]] = None,
     ) -> Optional[ImageResponsePort]:
         """
-        逐跳校验后读取图片；重试从原地址开始，默认只允许公网重定向。
+        逐跳校验后读取图片；重试从原地址开始，默认只允许公网与已配置网段。
         """
         response: Optional[ImageResponsePort] = None
         for _ in range(_IMAGE_FETCH_ATTEMPTS):
@@ -778,9 +799,13 @@ class ImageHelper(metaclass=Singleton):
                 target = ImageHelper._redirect_target(response, current_url)
                 if not target or redirect_count == _IMAGE_MAX_REDIRECTS:
                     return response
-                allowed = redirect_validator(target) if redirect_validator else SecurityUtils.is_safe_url(
-                    target, {target}, strict=True, block_private=True,
-                )
+                if redirect_validator:
+                    allowed = redirect_validator(target)
+                else:
+                    allowed = SecurityUtils.is_safe_url(
+                        target, {target}, strict=True, block_private=True,
+                        allowed_private_ranges=_redirect_private_ranges(),
+                    )
                 if not allowed:
                     return response
                 current_options = ImageHelper._redirect_options(current_options, current_url, target, response, cookies)
@@ -811,9 +836,13 @@ class ImageHelper(metaclass=Singleton):
                 target = ImageHelper._redirect_target(response, current_url)
                 if not target or redirect_count == _IMAGE_MAX_REDIRECTS:
                     return response
-                allowed = await redirect_validator(target) if redirect_validator else await SecurityUtils.is_safe_url_async(
-                    target, {target}, strict=True, block_private=True,
-                )
+                if redirect_validator:
+                    allowed = await redirect_validator(target)
+                else:
+                    allowed = await SecurityUtils.is_safe_url_async(
+                        target, {target}, strict=True, block_private=True,
+                        allowed_private_ranges=_redirect_private_ranges(),
+                    )
                 if not allowed:
                     return response
                 current_options = ImageHelper._redirect_options(current_options, current_url, target, response, cookies)

app/startup/composition/configuration.py

@@ -338,6 +338,9 @@ def build_chain_runtime_config(settings: Settings) -> ChainRuntimeConfig:
         wallpaper_image_url=settings.WALLPAPER_IMAGE_URL,
         customize_wallpaper_api_url=settings.CUSTOMIZE_WALLPAPER_API_URL,
         security_image_suffixes=tuple(settings.SECURITY_IMAGE_SUFFIXES),
+        image_proxy_allowed_private_ranges=tuple(
+            settings.IMAGE_PROXY_ALLOWED_PRIVATE_RANGES
+        ),
         cache_path=settings.CACHE_PATH,
         global_image_cache_days=settings.GLOBAL_IMAGE_CACHE_DAYS,
     )

tests/test_system_image_proxy.py

@@ -22,6 +22,7 @@ from app.adapters.network import http as http_adapter
 from app.adapters.system.host import SystemUtils
 from app.api.endpoints import system as system_endpoint
 from app.application import image as image_service
+from app.application.configuration import ChainRuntimeConfig
 from app.application.image import (
     ImageHelper,
     ImageResponsePort,
@@ -470,6 +471,92 @@ async def test_redirects_are_bounded_and_every_hop_is_checked(isolated_image_cac
     assert validator.await_count == 5
 
 
+class _RedirectThenImageTransport:
+    """
+    首个请求返回跨来源重定向，其后返回图片内容并记录请求地址。
+    """
+
+    def __init__(self, location: str, content: bytes) -> None:
+        """
+        保存重定向目标与最终图片内容。
+
+        :param location: 首个请求返回的 Location 头
+        :param content: 跟随重定向后返回的图片字节
+        """
+        self.location = location
+        self.content = content
+        self.sync_urls: list[str] = []
+        self.async_urls: list[str] = []
+
+    def get(self, url: str, *, options: Mapping[str, Any]) -> Optional[ImageResponsePort]:
+        """
+        记录同步请求地址，首个请求返回重定向，其余返回图片。
+
+        :param url: 本次请求地址
+        :param options: 传输参数，用于确认重定向未被自动跟随
+        :return: 构造的图片响应契约
+        """
+        self.sync_urls.append(url)
+        if len(self.sync_urls) == 1:
+            return SimpleNamespace(status_code=307, content=b"", headers={"location": self.location})
+        return SimpleNamespace(status_code=200, content=self.content, headers={})
+
+    async def async_get(self, url: str, *, options: Mapping[str, Any]) -> Optional[ImageResponsePort]:
+        """
+        记录异步请求地址，首个请求返回重定向，其余返回图片。
+
+        :param url: 本次请求地址
+        :param options: 传输参数，用于确认重定向未被自动跟随
+        :return: 构造的图片响应契约
+        """
+        self.async_urls.append(url)
+        if len(self.async_urls) == 1:
+            return SimpleNamespace(status_code=307, content=b"", headers={"location": self.location})
+        return SimpleNamespace(status_code=200, content=self.content, headers={})
+
+
+@pytest.mark.parametrize(("ranges", "expected"), [([], False), (["198.18.0.0/15"], True)])
+@pytest.mark.parametrize("mode", ["sync", "async"])
+def test_default_image_redirect_honours_configured_private_ranges(
+    monkeypatch: pytest.MonkeyPatch,
+    isolated_image_cache: ImageHelper,
+    mode: str,
+    ranges: list[str],
+    expected: bool,
+) -> None:
+    """
+    内部调用方不传 redirect_validator 时，默认重定向校验沿用图片代理配置的
+    非公网放行网段；未配置时仍按默认拒绝非公网解析结果。
+    """
+    content = _image_bytes("PNG")
+    transport = _RedirectThenImageTransport("https://cdn.example/poster.png", content)
+    configure_image_ports(transport=transport, internal_address=_FakeInternalAddress())
+    monkeypatch.setattr(
+        image_service,
+        "get_chain_runtime_config_snapshot",
+        lambda: ChainRuntimeConfig(
+            media_extensions=(),
+            image_proxy_allowed_private_ranges=tuple(ranges),
+        ),
+    )
+    monkeypatch.setattr(
+        system_endpoint.SecurityUtils, "_hostname_addresses", lambda _: [ipaddress.ip_address("198.18.0.2")],
+    )
+    monkeypatch.setattr(
+        system_endpoint.SecurityUtils, "_hostname_addresses_async",
+        AsyncMock(return_value=[ipaddress.ip_address("198.18.0.2")]),
+    )
+    url = "https://image.tmdb.org/redirect"
+    if mode == "sync":
+        result = isolated_image_cache.fetch_image(url, use_cache=False)
+        requested = transport.sync_urls
+    else:
+        result = asyncio.run(isolated_image_cache.async_fetch_image(url, use_cache=False))
+        requested = transport.async_urls
+    assert (result == content) is expected
+    assert requested == ([url] if not expected else [url, "https://cdn.example/poster.png"])
+
+
 def test_bangumi_image_proxy_domain_is_added_to_allowlist() -> None:
     """
     启用 Bangumi 图片代理时，应允许其配置主机通过通用图片端点。