mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-06-08 07:26:48 +00:00
fix: sign media server image proxy URLs
This commit is contained in:
jxxghp
@@ -1,11 +1,9 @@
|
||||
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 parse_qs, unquote, urljoin, urlparse
|
||||
from urllib.parse import urljoin, urlparse
|
||||
|
||||
import aiofiles
|
||||
import pillow_avif # noqa 用于自动注册AVIF支持
|
||||
@@ -32,7 +30,6 @@ from app.db.user_oper import (
|
||||
get_current_active_user_async,
|
||||
)
|
||||
from app.helper.image import ImageHelper
|
||||
from app.helper.mediaserver import MediaServerHelper
|
||||
from app.helper.message import MessageHelper
|
||||
from app.helper.progress import ProgressHelper
|
||||
from app.helper.rule import RuleHelper
|
||||
@@ -52,21 +49,6 @@ 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:
|
||||
@@ -359,95 +341,6 @@ 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,
|
||||
@@ -455,7 +348,6 @@ 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缓存和磁盘缓存
|
||||
@@ -466,17 +358,16 @@ async def fetch_image(
|
||||
if allowed_domains is None:
|
||||
allowed_domains = set(settings.SECURITY_IMAGE_DOMAINS)
|
||||
|
||||
fetch_url = SecurityUtils.strip_url_signature(url)
|
||||
# 验证URL安全性
|
||||
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()
|
||||
):
|
||||
) and not (fetch_url := SecurityUtils.verify_signed_url(url)):
|
||||
logger.warn(f"Blocked unsafe image URL: {url}")
|
||||
return None
|
||||
|
||||
content = await ImageHelper().async_fetch_image(
|
||||
url=url,
|
||||
url=fetch_url,
|
||||
proxy=proxy,
|
||||
use_cache=use_cache,
|
||||
cookies=cookies,
|
||||
@@ -491,7 +382,7 @@ async def fetch_image(
|
||||
# 返回缓存图片
|
||||
return Response(
|
||||
content=content,
|
||||
media_type=UrlUtils.get_mime_type(url, "image/jpeg"),
|
||||
media_type=UrlUtils.get_mime_type(fetch_url, "image/jpeg"),
|
||||
headers=headers,
|
||||
)
|
||||
return None
|
||||
@@ -509,13 +400,6 @@ async def proxy_img(
|
||||
"""
|
||||
图片代理，可选是否使用代理服务器，支持 HTTP 缓存
|
||||
"""
|
||||
# 媒体服务器添加图片代理支持
|
||||
hosts = [
|
||||
config.config.get("host")
|
||||
for config in MediaServerHelper().get_configs().values()
|
||||
if config and config.config and config.config.get("host")
|
||||
]
|
||||
media_server_domains = set(hosts)
|
||||
allowed_domains = set(settings.SECURITY_IMAGE_DOMAINS)
|
||||
cookies = (
|
||||
MediaServerChain().get_image_cookies(server=None, image_url=imgurl)
|
||||
@@ -529,7 +413,6 @@ async def proxy_img(
|
||||
cookies=cookies,
|
||||
if_none_match=if_none_match,
|
||||
allowed_domains=allowed_domains,
|
||||
media_server_domains=media_server_domains,
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -8,6 +8,7 @@ from app.db.mediaserver_oper import MediaServerOper
|
||||
from app.helper.service import ServiceConfigHelper
|
||||
from app.log import logger
|
||||
from app.schemas import MediaServerLibrary, MediaServerItem, MediaServerSeasonInfo, MediaServerPlayItem
|
||||
from app.utils.security import SecurityUtils
|
||||
|
||||
lock = threading.Lock()
|
||||
|
||||
@@ -17,12 +18,54 @@ class MediaServerChain(ChainBase):
|
||||
媒体服务器处理链
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def _sign_image_url(url: Optional[str]) -> Optional[str]:
|
||||
"""
|
||||
为返回前端的媒体服务器图片 URL 添加代理签名。
|
||||
"""
|
||||
return SecurityUtils.sign_url(url) if url else url
|
||||
|
||||
def _sign_library_images(
|
||||
self, libraries: Optional[List[MediaServerLibrary]]
|
||||
) -> List[MediaServerLibrary]:
|
||||
"""
|
||||
给媒体库列表中的封面和封面组添加代理签名。
|
||||
"""
|
||||
for library in libraries or []:
|
||||
if library.image:
|
||||
library.image = self._sign_image_url(library.image)
|
||||
if library.image_list:
|
||||
library.image_list = [
|
||||
self._sign_image_url(image)
|
||||
for image in library.image_list
|
||||
if image
|
||||
]
|
||||
return libraries or []
|
||||
|
||||
def _sign_play_item_images(
|
||||
self, items: Optional[List[MediaServerPlayItem]]
|
||||
) -> List[MediaServerPlayItem]:
|
||||
"""
|
||||
给媒体服务器播放条目中的图片 URL 添加代理签名。
|
||||
"""
|
||||
for item in items or []:
|
||||
if item.image:
|
||||
item.image = self._sign_image_url(item.image)
|
||||
return items or []
|
||||
|
||||
def librarys(self, server: str, username: Optional[str] = None,
|
||||
hidden: bool = False) -> List[MediaServerLibrary]:
|
||||
"""
|
||||
获取媒体服务器所有媒体库
|
||||
"""
|
||||
return self.run_module("mediaserver_librarys", server=server, username=username, hidden=hidden)
|
||||
return self._sign_library_images(
|
||||
self.run_module(
|
||||
"mediaserver_librarys",
|
||||
server=server,
|
||||
username=username,
|
||||
hidden=hidden,
|
||||
)
|
||||
)
|
||||
|
||||
def items(self, server: str, library_id: Union[str, int],
|
||||
start_index: Optional[int] = 0, limit: Optional[int] = -1) -> Generator[Any, None, None]:
|
||||
@@ -83,22 +126,46 @@ class MediaServerChain(ChainBase):
|
||||
"""
|
||||
获取媒体服务器正在播放信息
|
||||
"""
|
||||
return self.run_module("mediaserver_playing", count=count, server=server, username=username)
|
||||
return self._sign_play_item_images(
|
||||
self.run_module(
|
||||
"mediaserver_playing",
|
||||
count=count,
|
||||
server=server,
|
||||
username=username,
|
||||
)
|
||||
)
|
||||
|
||||
def latest(self, server: str, count: Optional[int] = 20,
|
||||
username: Optional[str] = None) -> List[MediaServerPlayItem]:
|
||||
"""
|
||||
获取媒体服务器最新入库条目
|
||||
"""
|
||||
return self.run_module("mediaserver_latest", count=count, server=server, username=username)
|
||||
return self._sign_play_item_images(
|
||||
self.run_module(
|
||||
"mediaserver_latest",
|
||||
count=count,
|
||||
server=server,
|
||||
username=username,
|
||||
)
|
||||
)
|
||||
|
||||
def get_latest_wallpapers(self, server: Optional[str] = None, count: Optional[int] = 10,
|
||||
remote: bool = True, username: Optional[str] = None) -> List[str]:
|
||||
"""
|
||||
获取最新最新入库条目海报作为壁纸，缓存1小时
|
||||
"""
|
||||
return self.run_module("mediaserver_latest_images", server=server, count=count,
|
||||
remote=remote, username=username)
|
||||
wallpapers = self.run_module(
|
||||
"mediaserver_latest_images",
|
||||
server=server,
|
||||
count=count,
|
||||
remote=remote,
|
||||
username=username,
|
||||
)
|
||||
return [
|
||||
self._sign_image_url(wallpaper)
|
||||
for wallpaper in wallpapers or []
|
||||
if wallpaper
|
||||
]
|
||||
|
||||
def get_latest_wallpaper(self, server: Optional[str] = None,
|
||||
remote: bool = True, username: Optional[str] = None) -> Optional[str]:
|
||||
|
||||
@@ -1,16 +1,21 @@
|
||||
import hmac
|
||||
import ipaddress
|
||||
import socket
|
||||
import time
|
||||
from hashlib import sha256
|
||||
from pathlib import Path
|
||||
from typing import List, Optional, Set, Union
|
||||
from urllib.parse import quote, urlparse
|
||||
from urllib.parse import parse_qsl, quote, urlencode, urlparse, urlunparse
|
||||
|
||||
from anyio import Path as AsyncPath
|
||||
|
||||
from app.core.config import settings
|
||||
from app.log import logger
|
||||
|
||||
|
||||
class SecurityUtils:
|
||||
_SIGNED_URL_PURPOSE = "image-proxy"
|
||||
_SIGNED_URL_EXPIRE_SECONDS = 86400
|
||||
|
||||
@staticmethod
|
||||
def is_safe_path(base_path: Path, user_path: Path,
|
||||
@@ -107,6 +112,103 @@ class SecurityUtils:
|
||||
return False
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
def _url_signature_payload(url: str, expires_at: int, purpose: str) -> bytes:
|
||||
"""
|
||||
构造 URL 签名载荷。
|
||||
|
||||
签名覆盖用途、过期时间和完整 URL，确保同一个签名不能挪用到其它
|
||||
内网地址或其它代理用途。
|
||||
"""
|
||||
return f"{purpose}\n{expires_at}\n{url}".encode("utf-8")
|
||||
|
||||
@staticmethod
|
||||
def _sign_url_payload(url: str, expires_at: int, purpose: str) -> str:
|
||||
"""
|
||||
使用 RESOURCE_SECRET_KEY 对 URL 签名载荷生成 HMAC。
|
||||
"""
|
||||
return hmac.new(
|
||||
settings.RESOURCE_SECRET_KEY.encode("utf-8"),
|
||||
SecurityUtils._url_signature_payload(url, expires_at, purpose),
|
||||
sha256,
|
||||
).hexdigest()
|
||||
|
||||
@staticmethod
|
||||
def strip_url_signature(url: str) -> str:
|
||||
"""
|
||||
移除 URL fragment 中的代理签名信息，得到真正要请求的地址。
|
||||
|
||||
图片代理签名放在 fragment 中，浏览器会把它传给 MoviePilot，但 HTTP
|
||||
客户端请求媒体服务器前不能把这些内部参数带过去。
|
||||
"""
|
||||
if not url:
|
||||
return url
|
||||
parsed_url = urlparse(url)
|
||||
return urlunparse(parsed_url._replace(fragment=""))
|
||||
|
||||
@staticmethod
|
||||
def sign_url(
|
||||
url: str,
|
||||
expires_in: int = _SIGNED_URL_EXPIRE_SECONDS,
|
||||
purpose: str = _SIGNED_URL_PURPOSE,
|
||||
) -> str:
|
||||
"""
|
||||
给服务端返回的资源 URL 添加临时签名。
|
||||
|
||||
该签名用于允许 `/system/img` 代理访问服务端已经确认过的私网图片 URL，
|
||||
避免代理端点重新依赖媒体服务器的具体路径规则。
|
||||
"""
|
||||
if not url:
|
||||
return url
|
||||
parsed_url = urlparse(url)
|
||||
if parsed_url.scheme not in {"http", "https"} or not parsed_url.netloc:
|
||||
return url
|
||||
clean_url = SecurityUtils.strip_url_signature(url)
|
||||
expires_at = int(time.time() + expires_in)
|
||||
signature = SecurityUtils._sign_url_payload(clean_url, expires_at, purpose)
|
||||
fragment = urlencode(
|
||||
{
|
||||
"mp_exp": str(expires_at),
|
||||
"mp_sig": signature,
|
||||
"mp_purpose": purpose,
|
||||
}
|
||||
)
|
||||
return urlunparse(urlparse(clean_url)._replace(fragment=fragment))
|
||||
|
||||
@staticmethod
|
||||
def verify_signed_url(
|
||||
url: str,
|
||||
purpose: str = _SIGNED_URL_PURPOSE,
|
||||
) -> Optional[str]:
|
||||
"""
|
||||
验证 URL fragment 中的代理签名，成功时返回去签名后的真实 URL。
|
||||
"""
|
||||
if not url:
|
||||
return None
|
||||
parsed_url = urlparse(url)
|
||||
if parsed_url.scheme not in {"http", "https"} or not parsed_url.netloc:
|
||||
return None
|
||||
fragment_params = dict(parse_qsl(parsed_url.fragment, keep_blank_values=True))
|
||||
expires_at = fragment_params.get("mp_exp")
|
||||
signature = fragment_params.get("mp_sig")
|
||||
signed_purpose = fragment_params.get("mp_purpose")
|
||||
if not expires_at or not signature or signed_purpose != purpose:
|
||||
return None
|
||||
try:
|
||||
expires_at_int = int(expires_at)
|
||||
except ValueError:
|
||||
return None
|
||||
if expires_at_int < int(time.time()):
|
||||
return None
|
||||
|
||||
clean_url = SecurityUtils.strip_url_signature(url)
|
||||
expected_signature = SecurityUtils._sign_url_payload(
|
||||
clean_url, expires_at_int, purpose
|
||||
)
|
||||
if not hmac.compare_digest(signature, expected_signature):
|
||||
return None
|
||||
return clean_url
|
||||
|
||||
@staticmethod
|
||||
def is_safe_url(
|
||||
url: str,
|
||||
|
||||
82
tests/test_mediaserver_image_signing.py
Normal file
82
tests/test_mediaserver_image_signing.py
Normal file
@@ -0,0 +1,82 @@
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import Mock
|
||||
|
||||
for _module_name in (
|
||||
"app.chain.mediaserver",
|
||||
"app.db.models",
|
||||
"app.db.user_oper",
|
||||
"app.helper.message",
|
||||
"app.utils.crypto",
|
||||
):
|
||||
if _module_name in sys.modules and not hasattr(
|
||||
sys.modules[_module_name], "__file__"
|
||||
):
|
||||
del sys.modules[_module_name]
|
||||
|
||||
from app.chain.mediaserver import MediaServerChain
|
||||
from app.schemas import MediaServerLibrary, MediaServerPlayItem
|
||||
from app.utils.security import SecurityUtils
|
||||
|
||||
|
||||
class MediaServerImageSigningTest(unittest.TestCase):
|
||||
@staticmethod
|
||||
def _build_chain(result):
|
||||
"""
|
||||
构造只带 run_module 的 MediaServerChain，避免单测初始化真实模块管理器。
|
||||
"""
|
||||
chain = MediaServerChain.__new__(MediaServerChain)
|
||||
chain.run_module = Mock(return_value=result)
|
||||
return chain
|
||||
|
||||
def test_librarys_signs_image_fields(self):
|
||||
"""
|
||||
媒体库接口返回前需要给 image 和 image_list 加签。
|
||||
"""
|
||||
image = "http://192.168.1.50:8096/Items/lib/Images/Primary"
|
||||
image_list = [
|
||||
"http://192.168.1.50:32400/library/metadata/1/thumb/1",
|
||||
]
|
||||
chain = self._build_chain(
|
||||
[
|
||||
MediaServerLibrary(
|
||||
id="lib",
|
||||
image=image,
|
||||
image_list=image_list,
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
result = chain.librarys(server="jellyfin")
|
||||
|
||||
self.assertEqual(SecurityUtils.verify_signed_url(result[0].image), image)
|
||||
self.assertEqual(
|
||||
SecurityUtils.verify_signed_url(result[0].image_list[0]),
|
||||
image_list[0],
|
||||
)
|
||||
|
||||
def test_latest_signs_play_item_images(self):
|
||||
"""
|
||||
最近入库接口返回前需要给条目图片加签。
|
||||
"""
|
||||
image = "http://192.168.1.50:8096/Items/item/Images/Backdrop"
|
||||
chain = self._build_chain([MediaServerPlayItem(id="item", image=image)])
|
||||
|
||||
result = chain.latest(server="jellyfin")
|
||||
|
||||
self.assertEqual(SecurityUtils.verify_signed_url(result[0].image), image)
|
||||
|
||||
def test_latest_wallpapers_signs_urls(self):
|
||||
"""
|
||||
媒体服务器壁纸 URL 返回前也需要加签。
|
||||
"""
|
||||
wallpaper = "http://192.168.1.50:8096/Items/item/Images/Backdrop"
|
||||
chain = self._build_chain([wallpaper])
|
||||
|
||||
result = chain.get_latest_wallpapers()
|
||||
|
||||
self.assertEqual(SecurityUtils.verify_signed_url(result[0]), wallpaper)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -6,6 +6,44 @@ from app.utils.security import SecurityUtils
|
||||
|
||||
|
||||
class SecurityUtilsTest(TestCase):
|
||||
def test_signed_url_roundtrip_returns_clean_url(self):
|
||||
"""
|
||||
URL 签名验证成功后返回不含签名片段的真实请求地址。
|
||||
"""
|
||||
url = "http://192.168.1.50:8096/Items/abc/Images/Primary?api_key=demo"
|
||||
|
||||
signed_url = SecurityUtils.sign_url(url)
|
||||
|
||||
self.assertIn("#mp_exp=", signed_url)
|
||||
self.assertEqual(SecurityUtils.verify_signed_url(signed_url), url)
|
||||
self.assertEqual(SecurityUtils.strip_url_signature(signed_url), url)
|
||||
|
||||
def test_signed_url_rejects_tampered_url(self):
|
||||
"""
|
||||
签名绑定完整 URL，签名后修改路径必须校验失败。
|
||||
"""
|
||||
signed_url = SecurityUtils.sign_url(
|
||||
"http://192.168.1.50:8096/Items/abc/Images/Primary"
|
||||
)
|
||||
tampered_url = signed_url.replace(
|
||||
"/Items/abc/Images/Primary",
|
||||
"/System/Info/Public",
|
||||
)
|
||||
|
||||
self.assertIsNone(SecurityUtils.verify_signed_url(tampered_url))
|
||||
|
||||
def test_signed_url_rejects_expired_signature(self):
|
||||
"""
|
||||
已过期签名不能继续放行私网图片代理请求。
|
||||
"""
|
||||
with patch("app.utils.security.time.time", return_value=1000):
|
||||
signed_url = SecurityUtils.sign_url(
|
||||
"http://192.168.1.50:8096/Items/abc/Images/Primary",
|
||||
expires_in=10,
|
||||
)
|
||||
|
||||
with patch("app.utils.security.time.time", return_value=1011):
|
||||
self.assertIsNone(SecurityUtils.verify_signed_url(signed_url))
|
||||
|
||||
def test_is_safe_url_keeps_default_allowlist_behavior(self):
|
||||
"""
|
||||
|
||||
@@ -37,7 +37,12 @@ _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(), Event=_Dummy)
|
||||
_stub_module(
|
||||
"app.core.event",
|
||||
eventmanager=_Dummy(),
|
||||
Event=_Dummy,
|
||||
EventManager=_Dummy,
|
||||
)
|
||||
_stub_module("app.core.metainfo", MetaInfo=_Dummy)
|
||||
_stub_module("app.core.module", ModuleManager=_Dummy)
|
||||
_stub_module(
|
||||
@@ -82,10 +87,12 @@ from app.api.endpoints import system as system_endpoint
|
||||
|
||||
|
||||
class NettestSecurityTest(unittest.TestCase):
|
||||
def test_fetch_image_allows_private_media_server_image_path(self):
|
||||
def test_fetch_image_allows_signed_private_url(self):
|
||||
"""
|
||||
已配置媒体服务器的内网图片接口需要继续允许代理，保证前端封面显示。
|
||||
服务端签名过的私网图片 URL 可以继续代理，保证前端封面显示。
|
||||
"""
|
||||
image_url = "http://192.168.1.50:8096/System/Info/Public"
|
||||
signed_url = system_endpoint.SecurityUtils.sign_url(image_url)
|
||||
image_helper = Mock()
|
||||
image_helper.async_fetch_image = AsyncMock(return_value=b"image-bytes")
|
||||
|
||||
@@ -96,14 +103,18 @@ class NettestSecurityTest(unittest.TestCase):
|
||||
):
|
||||
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"},
|
||||
url=signed_url,
|
||||
allowed_domains=set(),
|
||||
)
|
||||
)
|
||||
|
||||
self.assertEqual(resp.status_code, 200)
|
||||
image_helper.async_fetch_image.assert_awaited_once()
|
||||
image_helper.async_fetch_image.assert_awaited_once_with(
|
||||
url=image_url,
|
||||
proxy=None,
|
||||
use_cache=False,
|
||||
cookies=None,
|
||||
)
|
||||
|
||||
def test_fetch_image_blocks_private_allowed_url_before_request(self):
|
||||
"""
|
||||
@@ -123,39 +134,23 @@ class NettestSecurityTest(unittest.TestCase):
|
||||
|
||||
self.assertIsNone(resp)
|
||||
|
||||
def test_fetch_image_blocks_private_media_server_non_image_path(self):
|
||||
def test_fetch_image_blocks_tampered_signed_private_url(self):
|
||||
"""
|
||||
媒体服务器 host 只放行图片接口，不放行同 host 下的任意 API。
|
||||
私网签名绑定完整 URL，改动路径后不能继续代理。
|
||||
"""
|
||||
signed_url = system_endpoint.SecurityUtils.sign_url(
|
||||
"http://192.168.1.50:8096/Items/abc/Images/Primary"
|
||||
).replace("/Items/abc/Images/Primary", "/System/Info/Public")
|
||||
|
||||
class FailIfCalled:
|
||||
def __init__(self, *args, **kwargs):
|
||||
raise AssertionError("fetch_image should block non-image media server paths")
|
||||
raise AssertionError("fetch_image should block tampered signed URLs")
|
||||
|
||||
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"},
|
||||
url=signed_url,
|
||||
allowed_domains=set(),
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user