From d713ea54c16785fe4e423109ed84452040d147cd Mon Sep 17 00:00:00 2001 From: jxxghp Date: Mon, 25 May 2026 12:25:57 +0800 Subject: [PATCH] fix: allow media server image proxy paths --- app/api/endpoints/system.py | 119 ++++++++++++++++++++++++++++++++++- tests/test_system_nettest.py | 65 ++++++++++++++++++- 2 files changed, 179 insertions(+), 5 deletions(-) diff --git a/app/api/endpoints/system.py b/app/api/endpoints/system.py index 3ce1b3f8..0a1d9268 100644 --- a/app/api/endpoints/system.py +++ b/app/api/endpoints/system.py @@ -1,9 +1,11 @@ import asyncio import json +import posixpath +import re from collections import deque from datetime import datetime from typing import Any, Optional, Union, Annotated -from urllib.parse import urljoin, urlparse +from urllib.parse import parse_qs, unquote, urljoin, urlparse import aiofiles import pillow_avif # noqa 用于自动注册AVIF支持 @@ -50,6 +52,21 @@ from version import APP_VERSION router = APIRouter() _NETTEST_REDIRECT_STATUS_CODES = {301, 302, 303, 307, 308} +_MEDIA_SERVER_IMAGE_PATH_PATTERNS = ( + re.compile( + r"^/(?:emby/)?Items/[^/]+/Images/" + r"(?:Primary|Art|Backdrop|Banner|Logo|Thumb|Disc|Box|Screenshot|Menu|Chapter)" + r"(?:/[^/]+)?$", + re.IGNORECASE, + ), + re.compile( + r"^/library/metadata/[^/]+/" + r"(?:thumb|art|banner|poster|clearlogo|clearart|background)" + r"(?:/[^/]+)?$", + re.IGNORECASE, + ), + re.compile(r"^/api/v1/sys/img/.+", re.IGNORECASE), +) def _match_nettest_prefix(url: str, prefix: str) -> bool: @@ -342,6 +359,95 @@ async def _close_nettest_response(response: Any) -> None: logger.debug(f"关闭网络测试响应失败: {err}") +def _normalize_proxy_image_path(path: str) -> str: + """ + 归一化代理图片路径，用于识别媒体服务器图片接口。 + + URL path 可能包含编码后的特殊字符，这里先解码再规范化路径，避免 + `%2e%2e` 或重复斜杠绕过后续的媒体图片路径判断。 + """ + decoded_path = unquote(path or "/") + normalized_path = posixpath.normpath(decoded_path) + if not normalized_path.startswith("/"): + normalized_path = f"/{normalized_path}" + return normalized_path + + +def _is_known_media_server_image_path(path: str) -> bool: + """ + 判断路径是否属于已知媒体服务器图片读取接口。 + + 这里仅覆盖 MoviePilot 自身会返回给前端的封面、背景图和图片流接口， + 不允许媒体服务器同 host 下的任意 API 路径继续通过图片代理访问。 + """ + normalized_path = _normalize_proxy_image_path(path) + return any( + pattern.match(normalized_path) + for pattern in _MEDIA_SERVER_IMAGE_PATH_PATTERNS + ) + + +def _is_plex_transcode_image_url(url: str) -> bool: + """ + 校验 Plex 图片转码接口只转码 Plex 自身 metadata 图片路径。 + + Plex 的 posterUrl/artUrl 可能使用 `/photo/:/transcode` 包装真实图片路径， + 因此需要额外检查 query 里的 `url` 仍然是 metadata 图片路径，而不是 + 任意可被 Plex 代取的地址。 + """ + parsed_url = urlparse(url) + if _normalize_proxy_image_path(parsed_url.path) != "/photo/:/transcode": + return False + source_path = parse_qs(parsed_url.query).get("url", [None])[0] + if not source_path: + return False + source_url = urlparse(source_path) + if source_url.scheme or source_url.netloc: + return False + return _is_known_media_server_image_path(source_path) + + +def _is_ugreen_image_stream_url(url: str) -> bool: + """ + 校验绿联本机图片流接口只代理官方 scraper 图片。 + + 绿联本地图片需要带加密鉴权头，目前模块只会把 scraper.ugnas.com 的签名图 + 转成 getImaStream，本检查避免用户把该接口改造成任意远程 URL 中转。 + """ + parsed_url = urlparse(url) + if _normalize_proxy_image_path(parsed_url.path) != "/ugreen/v2/video/getImaStream": + return False + source_url = parse_qs(parsed_url.query).get("name", [None])[0] + if not source_url: + return False + parsed_source = urlparse(source_url) + return ( + parsed_source.scheme in {"http", "https"} + and parsed_source.netloc.lower() == "scraper.ugnas.com" + ) + + +def _is_allowed_media_server_image_url( + url: str, + media_server_domains: set[str], +) -> bool: + """ + 判断内网媒体服务器 URL 是否可作为图片代理目标。 + + 私有地址默认仍然禁止；只有 URL host 精确命中已配置媒体服务器，并且路径是 + 已知图片接口时才允许访问，用于兼容前端媒体库和最近入库图片展示。 + """ + if not media_server_domains: + return False + if not SecurityUtils.is_safe_url(url, media_server_domains, strict=True): + return False + return ( + _is_known_media_server_image_path(urlparse(url).path) + or _is_plex_transcode_image_url(url) + or _is_ugreen_image_stream_url(url) + ) + + async def fetch_image( url: str, proxy: Optional[bool] = None, @@ -349,6 +455,7 @@ async def fetch_image( if_none_match: Optional[str] = None, cookies: Optional[str | dict] = None, allowed_domains: Optional[set[str]] = None, + media_server_domains: Optional[set[str]] = None, ) -> Optional[Response]: """ 处理图片缓存逻辑，支持HTTP缓存和磁盘缓存 @@ -360,7 +467,11 @@ async def fetch_image( allowed_domains = set(settings.SECURITY_IMAGE_DOMAINS) # 验证URL安全性 - if not SecurityUtils.is_safe_url(url, allowed_domains, block_private=True): + if not SecurityUtils.is_safe_url( + url, allowed_domains, block_private=True + ) and not _is_allowed_media_server_image_url( + url, media_server_domains or set() + ): logger.warn(f"Blocked unsafe image URL: {url}") return None @@ -404,7 +515,8 @@ async def proxy_img( for config in MediaServerHelper().get_configs().values() if config and config.config and config.config.get("host") ] - allowed_domains = set(settings.SECURITY_IMAGE_DOMAINS) | set(hosts) + media_server_domains = set(hosts) + allowed_domains = set(settings.SECURITY_IMAGE_DOMAINS) cookies = ( MediaServerChain().get_image_cookies(server=None, image_url=imgurl) if use_cookies @@ -417,6 +529,7 @@ async def proxy_img( cookies=cookies, if_none_match=if_none_match, allowed_domains=allowed_domains, + media_server_domains=media_server_domains, ) diff --git a/tests/test_system_nettest.py b/tests/test_system_nettest.py index b9dd01fd..5a5811a3 100644 --- a/tests/test_system_nettest.py +++ b/tests/test_system_nettest.py @@ -2,7 +2,7 @@ import asyncio import sys import unittest from types import ModuleType, SimpleNamespace -from unittest.mock import patch +from unittest.mock import AsyncMock, Mock, patch def _stub_module(name: str, **attrs): @@ -37,7 +37,7 @@ _stub_module("app.chain.media", MediaChain=_Dummy) _stub_module("app.chain.mediaserver", MediaServerChain=_Dummy) _stub_module("app.chain.search", SearchChain=_Dummy) _stub_module("app.chain.system", SystemChain=_Dummy) -_stub_module("app.core.event", eventmanager=_Dummy()) +_stub_module("app.core.event", eventmanager=_Dummy(), Event=_Dummy) _stub_module("app.core.metainfo", MetaInfo=_Dummy) _stub_module("app.core.module", ModuleManager=_Dummy) _stub_module( @@ -82,6 +82,29 @@ from app.api.endpoints import system as system_endpoint class NettestSecurityTest(unittest.TestCase): + def test_fetch_image_allows_private_media_server_image_path(self): + """ + 已配置媒体服务器的内网图片接口需要继续允许代理，保证前端封面显示。 + """ + image_helper = Mock() + image_helper.async_fetch_image = AsyncMock(return_value=b"image-bytes") + + with patch.object(system_endpoint, "ImageHelper", return_value=image_helper), patch.object( + system_endpoint.HashUtils, "md5", return_value="etag", create=True + ), patch.object( + system_endpoint.RequestUtils, "generate_cache_headers", return_value={}, create=True + ): + resp = asyncio.run( + system_endpoint.fetch_image( + url="http://192.168.1.50:8096/Items/abc/Images/Primary", + allowed_domains={"http://192.168.1.50:8096"}, + media_server_domains={"http://192.168.1.50:8096"}, + ) + ) + + self.assertEqual(resp.status_code, 200) + image_helper.async_fetch_image.assert_awaited_once() + def test_fetch_image_blocks_private_allowed_url_before_request(self): """ 图片代理即使拿到内网 allowlist 项，也必须在发起请求前拦截。 @@ -100,6 +123,44 @@ class NettestSecurityTest(unittest.TestCase): self.assertIsNone(resp) + def test_fetch_image_blocks_private_media_server_non_image_path(self): + """ + 媒体服务器 host 只放行图片接口，不放行同 host 下的任意 API。 + """ + class FailIfCalled: + def __init__(self, *args, **kwargs): + raise AssertionError("fetch_image should block non-image media server paths") + + with patch.object(system_endpoint, "ImageHelper", FailIfCalled): + resp = asyncio.run( + system_endpoint.fetch_image( + url="http://192.168.1.50:8096/System/Info/Public", + allowed_domains={"http://192.168.1.50:8096"}, + media_server_domains={"http://192.168.1.50:8096"}, + ) + ) + + self.assertIsNone(resp) + + def test_fetch_image_blocks_traversal_in_media_server_image_path(self): + """ + 编码后的路径穿越不能借媒体图片前缀绕过私网 SSRF 防护。 + """ + class FailIfCalled: + def __init__(self, *args, **kwargs): + raise AssertionError("fetch_image should block traversal image paths") + + with patch.object(system_endpoint, "ImageHelper", FailIfCalled): + resp = asyncio.run( + system_endpoint.fetch_image( + url="http://192.168.1.50:5666/api/v1/sys/img/%2e%2e/manager/user/list", + allowed_domains={"http://192.168.1.50:5666"}, + media_server_domains={"http://192.168.1.50:5666"}, + ) + ) + + self.assertIsNone(resp) + def test_nettest_targets_are_served_by_backend(self): resp = asyncio.run(system_endpoint.nettest_targets(_="token"))