152 lines
5.7 KiB
Python
152 lines
5.7 KiB
Python
"""Parse and safely download image payloads from BOXIM private messages."""
|
|
|
|
import ipaddress
|
|
import json
|
|
import os
|
|
import socket
|
|
from dataclasses import dataclass
|
|
from pathlib import PurePosixPath
|
|
from urllib.parse import unquote, urljoin, urlsplit
|
|
|
|
import httpx
|
|
|
|
|
|
MAX_REDIRECTS = 3
|
|
|
|
|
|
class BoxIMImageError(RuntimeError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DownloadedBoxIMImage:
|
|
content: bytes
|
|
filename: str
|
|
mime_type: str
|
|
source_url: str
|
|
|
|
|
|
def parse_boxim_image_url(content: str, *, base_url: str = "") -> str:
|
|
try:
|
|
payload = json.loads(content or "")
|
|
except (TypeError, ValueError) as exc:
|
|
raise BoxIMImageError("BOXIM 图片消息格式无效") from exc
|
|
if not isinstance(payload, dict):
|
|
raise BoxIMImageError("BOXIM 图片消息格式无效")
|
|
|
|
value = payload.get("originUrl") or payload.get("thumbUrl") or payload.get("url")
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise BoxIMImageError("BOXIM 图片消息缺少图片地址")
|
|
value = value.strip()
|
|
if value.startswith("/"):
|
|
if not base_url:
|
|
raise BoxIMImageError("BOXIM 图片地址不完整")
|
|
value = urljoin(f"{base_url.rstrip('/')}/", value)
|
|
return value
|
|
|
|
|
|
def _configured_hosts(name: str) -> set[str]:
|
|
return {
|
|
value.strip().lower().rstrip(".")
|
|
for value in os.getenv(name, "").split(",")
|
|
if value.strip()
|
|
}
|
|
|
|
|
|
def _host_matches(host: str, configured: set[str]) -> bool:
|
|
return any(host == value or host.endswith(f".{value}") for value in configured)
|
|
|
|
|
|
def _resolved_addresses(host: str, port: int) -> set[ipaddress.IPv4Address | ipaddress.IPv6Address]:
|
|
try:
|
|
return {
|
|
ipaddress.ip_address(item[4][0])
|
|
for item in socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
|
|
}
|
|
except (OSError, ValueError) as exc:
|
|
raise BoxIMImageError("BOXIM 图片地址无法解析") from exc
|
|
|
|
|
|
def _is_safe_remote_url(url: str) -> None:
|
|
parsed = urlsplit(url)
|
|
scheme = parsed.scheme.lower()
|
|
allow_http = os.getenv("BOXIM_IMAGE_ALLOW_HTTP", "").lower() in {"1", "true", "yes"}
|
|
if scheme not in ({"https", "http"} if allow_http else {"https"}):
|
|
raise BoxIMImageError("BOXIM 图片地址必须使用 HTTPS")
|
|
if parsed.username or parsed.password or not parsed.hostname:
|
|
raise BoxIMImageError("BOXIM 图片地址无效")
|
|
|
|
host = parsed.hostname.lower().rstrip(".")
|
|
allowed_hosts = _configured_hosts("BOXIM_IMAGE_ALLOWED_HOSTS")
|
|
if allowed_hosts and not _host_matches(host, allowed_hosts):
|
|
raise BoxIMImageError("BOXIM 图片地址不在允许的域名范围内")
|
|
|
|
private_hosts = _configured_hosts("BOXIM_IMAGE_PRIVATE_HOSTS")
|
|
try:
|
|
addresses = {ipaddress.ip_address(host)}
|
|
except ValueError:
|
|
addresses = _resolved_addresses(host, parsed.port or (443 if scheme == "https" else 80))
|
|
if not addresses:
|
|
raise BoxIMImageError("BOXIM 图片地址无法解析")
|
|
if _host_matches(host, private_hosts):
|
|
return
|
|
if any(not address.is_global for address in addresses):
|
|
raise BoxIMImageError("BOXIM 图片地址指向受限网络")
|
|
|
|
|
|
def _filename_from_url(url: str) -> str:
|
|
value = unquote(PurePosixPath(urlsplit(url).path).name).strip()
|
|
value = value.replace("\x00", "")
|
|
return (value or "boxim-image")[:255]
|
|
|
|
|
|
def download_boxim_image(
|
|
content: str,
|
|
*,
|
|
base_url: str = "",
|
|
transport: httpx.BaseTransport | None = None,
|
|
) -> DownloadedBoxIMImage:
|
|
"""Download one BOXIM image without redirects or oversized responses escaping checks."""
|
|
url = parse_boxim_image_url(content, base_url=base_url)
|
|
max_bytes = max(1024, int(os.getenv("CHAT_IMAGE_MAX_BYTES", str(8 * 1024 * 1024))))
|
|
timeout = max(1.0, min(float(os.getenv("BOXIM_IMAGE_TIMEOUT_SECONDS", "15")), 60.0))
|
|
|
|
with httpx.Client(
|
|
timeout=timeout,
|
|
follow_redirects=False,
|
|
trust_env=False,
|
|
transport=transport,
|
|
) as client:
|
|
for _ in range(MAX_REDIRECTS + 1):
|
|
_is_safe_remote_url(url)
|
|
try:
|
|
with client.stream("GET", url, headers={"Accept": "image/*"}) as response:
|
|
if response.status_code in {301, 302, 303, 307, 308}:
|
|
location = response.headers.get("location", "").strip()
|
|
if not location:
|
|
raise BoxIMImageError("BOXIM 图片跳转地址无效")
|
|
url = urljoin(url, location)
|
|
continue
|
|
response.raise_for_status()
|
|
raw_length = response.headers.get("content-length", "")
|
|
if raw_length.isdigit() and int(raw_length) > max_bytes:
|
|
raise BoxIMImageError("BOXIM 图片超过大小限制")
|
|
chunks = bytearray()
|
|
for chunk in response.iter_bytes():
|
|
chunks.extend(chunk)
|
|
if len(chunks) > max_bytes:
|
|
raise BoxIMImageError("BOXIM 图片超过大小限制")
|
|
if not chunks:
|
|
raise BoxIMImageError("BOXIM 图片内容为空")
|
|
return DownloadedBoxIMImage(
|
|
content=bytes(chunks),
|
|
filename=_filename_from_url(url),
|
|
mime_type=response.headers.get("content-type", "").split(";", 1)[0][:100],
|
|
source_url=url,
|
|
)
|
|
except BoxIMImageError:
|
|
raise
|
|
except (httpx.HTTPError, OSError) as exc:
|
|
raise BoxIMImageError("BOXIM 图片下载失败") from exc
|
|
raise BoxIMImageError("BOXIM 图片跳转次数过多")
|