From c68288e897aa640ca85fae5151f6ec9429e568c2 Mon Sep 17 00:00:00 2001 From: LS Date: Wed, 23 Sep 2026 16:34:25 +0800 Subject: [PATCH] =?UTF-8?q?feat(security):=20=E5=9B=BE=E7=89=87=E4=BB=A3?= =?UTF-8?q?=E7=90=86=E8=87=AA=E5=8A=A8=E4=BF=A1=E4=BB=BB=E5=B7=B2=E5=90=AF?= =?UTF-8?q?=E7=94=A8=E7=9A=84=E5=AA=92=E4=BD=93=E6=9C=8D=E5=8A=A1=E5=99=A8?= =?UTF-8?q?=E4=B8=BB=E6=9C=BA=20(#6769)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 在局域网部署 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 --- app/api/endpoints/system.py | 20 +++++++++ app/application/security/url.py | 18 +++++++++ tests/test_system_image_proxy.py | 69 ++++++++++++++++++++++++++++++++ 3 files changed, 107 insertions(+) diff --git a/app/api/endpoints/system.py b/app/api/endpoints/system.py index c86934cf2..23fd3ce82 100644 --- a/app/api/endpoints/system.py +++ b/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 diff --git a/app/application/security/url.py b/app/application/security/url.py index 53203bcc6..afa7dc7f7 100644 --- a/app/application/security/url.py +++ b/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, diff --git a/tests/test_system_image_proxy.py b/tests/test_system_image_proxy.py index d0ab85ba6..5835ae146 100644 --- a/tests/test_system_image_proxy.py +++ b/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"), [