feat(security): 图片代理自动信任已启用的媒体服务器主机 (#6769)

在局域网部署 Emby/Jellyfin/Plex 的用户，此前需要手动维护
SECURITY_IMAGE_DOMAINS（整表覆盖，会随版本默认值漂移）和
IMAGE_PROXY_ALLOWED_PRIVATE_RANGES，未签名的媒体服务器图片 URL
才能通过图片代理。

- is_safe_image_url_async 新增可选参数 trusted_hosts：URL 的 netloc
  （含端口）或 hostname 精确命中时直接放行，不受私网拦截限制；
  默认 None，现有调用方行为不变。
- fetch_image（覆盖 /img 与 /cache/image）从已启用媒体服务器的
  host / play_host 计算受信主机；读取配置失败时返回空集合，不影响代理。

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>

app/api/endpoints/system.py

@@ -38,6 +38,7 @@ from app.application.configuration import (
 )
 from app.application.database import get_database_governance
 from app.application.image import ImageHelper
+from app.application.mediaserver import get_mediaserver_configs
 from app.application.messaging.message import MessageHelper
 from app.application.network import get_configured_network_test_service
 from app.application.rules import RuleHelper
@@ -123,6 +124,24 @@ def _get_image_proxy_allowed_domains() -> set[str]:
     return allowed_domains
 
 
+def _get_image_proxy_trusted_hosts() -> set[str]:
+    """返回已启用媒体服务器的主机（host / play_host），图片代理可直接信任。"""
+    trusted_hosts: set[str] = set()
+    try:
+        for conf in get_mediaserver_configs():
+            for key in ("host", "play_host"):
+                host = str((conf.config or {}).get(key) or "").strip()
+                if not host:
+                    continue
+                parsed = urlsplit(host if "://" in host else f"https://{host}")
+                if parsed.scheme in {"http", "https"} and parsed.netloc:
+                    trusted_hosts.add(parsed.netloc.lower())
+    except Exception as err:  # noqa: BLE001 - 配置不可用时不影响图片代理
+        logger.debug(f"读取媒体服务器配置失败，图片代理不信任任何媒体服务器主机：{err}")
+        return set()
+    return trusted_hosts
+
+
 def _database_backup_artifact_data(artifact: Any) -> _SchemaDatabaseBackupArtifactData:
     """将内部备份制品映射为不含宿主路径的 Web DTO。"""
     return _SchemaDatabaseBackupArtifactData(
@@ -203,6 +222,7 @@ async def fetch_image(
             "IMAGE_PROXY_ALLOWED_PRIVATE_RANGES",
             [],
         ),
+        trusted_hosts=_get_image_proxy_trusted_hosts(),
     ):
         return None
 

app/application/security/url.py

@@ -825,10 +825,14 @@ class SecurityUtils:
         url: str,
         allowed_domains: Union[Set[str], List[str]],
         allowed_private_ranges: Optional[Iterable[str]] = None,
+        trusted_hosts: Optional[Iterable[str]] = None,
     ) -> bool:
         """
         判定 URL 是否可作为图片代理请求目标。
 
+        `trusted_hosts` 为管理员已配置的主机（如已启用的媒体服务器），URL 的
+        netloc（含端口）或 hostname 精确命中时直接放行，不受私网拦截限制。
+
         校验顺序：协议 + 域名 allowlist + DNS SSRF 拦截 + 非公网放行匹配；标准
         校验失败时再用 `verify_signed_url` 兜底，允许后端预签名的媒体服务器
         URL 跳过私网拦截。两者皆失败才视为拒绝。
@@ -847,6 +851,8 @@ class SecurityUtils:
         )
         if diagnosis.allowed:
             return True
+        if trusted_hosts and SecurityUtils._is_trusted_host(url, trusted_hosts):
+            return True
         if SecurityUtils.verify_signed_url(url) is not None:
             return True
         await _emit_image_proxy_block_warning(
@@ -857,6 +863,18 @@ class SecurityUtils:
         )
         return False
 
+    @staticmethod
+    def _is_trusted_host(url: str, trusted_hosts: Iterable[str]) -> bool:
+        """URL 的 netloc（含端口）或 hostname 是否精确命中受信主机。"""
+        try:
+            parsed_url = urlparse(url)
+        except ValueError:
+            return False
+        if parsed_url.scheme not in {"http", "https"} or not parsed_url.hostname:
+            return False
+        trusted = {host.lower() for host in trusted_hosts}
+        return parsed_url.netloc.lower() in trusted or parsed_url.hostname in trusted
+
     @staticmethod
     def _diagnose_resolved_addresses(
         url: str,

tests/test_system_image_proxy.py

@@ -1,5 +1,6 @@
 import asyncio
 import io
+import ipaddress
 from collections.abc import Iterator, Mapping
 from hashlib import sha256
 from types import SimpleNamespace
@@ -103,6 +104,74 @@ def test_bangumi_image_proxy_domain_is_added_to_allowlist() -> None:
         }
 
 
+def test_image_proxy_trusts_enabled_mediaserver_hosts() -> None:
+    """已启用媒体服务器的 host / play_host 应被规范化为受信主机。"""
+    confs = [
+        SimpleNamespace(config={"host": "192.168.1.10:8096", "play_host": "https://Emby.Example.com"}),
+        SimpleNamespace(config={"host": "http://jellyfin.lan"}),
+        SimpleNamespace(config=None),
+    ]
+    with patch.object(system_endpoint, "get_mediaserver_configs", return_value=confs) as getter:
+        assert system_endpoint._get_image_proxy_trusted_hosts() == {
+            "192.168.1.10:8096",
+            "emby.example.com",
+            "jellyfin.lan",
+        }
+    getter.assert_called_once_with()
+
+
+def test_image_proxy_trusted_hosts_empty_when_lookup_fails() -> None:
+    """媒体服务器配置读取失败时不信任任何主机，也不抛出异常。"""
+    with patch.object(system_endpoint, "get_mediaserver_configs", side_effect=RuntimeError("db down")):
+        assert system_endpoint._get_image_proxy_trusted_hosts() == set()
+
+
+@pytest.mark.parametrize(
+    ("url", "expected"),
+    [
+        ("http://192.168.1.10:8096/Items/1/Images/Primary", True),
+        ("http://jellyfin.lan:8096/Items/1/Images/Primary", True),
+        ("http://192.168.1.10:22/poster.jpg", False),
+        ("http://192.168.1.11:8096/Items/1/Images/Primary", False),
+    ],
+)
+def test_image_proxy_allows_lan_url_only_for_trusted_mediaserver_host(url: str, expected: bool) -> None:
+    """未配置私网网段时，仅受信媒体服务器主机的内网图片可通过代理。"""
+    trusted = {"192.168.1.10:8096", "jellyfin.lan"}
+    with patch.object(
+        system_endpoint.SecurityUtils,
+        "_hostname_addresses_async",
+        new=AsyncMock(return_value=[ipaddress.ip_address("192.168.1.10")]),
+    ), patch("app.application.security.url._emit_image_proxy_block_warning", new=AsyncMock()):
+        allowed = asyncio.run(
+            system_endpoint.SecurityUtils.is_safe_image_url_async(
+                url,
+                {"image.tmdb.org"},
+                allowed_private_ranges=[],
+                trusted_hosts=trusted,
+            )
+        )
+    assert allowed is expected
+
+
+def test_fetch_image_passes_mediaserver_trusted_hosts() -> None:
+    """fetch_image 应把媒体服务器受信主机传给图片 URL 安全校验。"""
+    image_helper = Mock()
+    image_helper.async_fetch_image_with_mime_type = AsyncMock(return_value=(_image_bytes("PNG"), "image/png"))
+    is_safe = AsyncMock(return_value=True)
+    with patch.object(system_endpoint.SecurityUtils, "is_safe_image_url_async", new=is_safe), patch.object(
+        system_endpoint, "ImageHelper", return_value=image_helper
+    ), patch.object(system_endpoint, "_get_image_proxy_trusted_hosts", return_value={"192.168.1.10:8096"}):
+        response = asyncio.run(
+            system_endpoint.fetch_image(
+                url="http://192.168.1.10:8096/Items/1/Images/Primary",
+                allowed_domains={"image.tmdb.org"},
+            )
+        )
+    assert response is not None
+    assert is_safe.await_args.kwargs["trusted_hosts"] == {"192.168.1.10:8096"}
+
+
 @pytest.mark.parametrize(
     ("image_format", "expected_mime"),
     [