fix: 安全技术债务第二轮 - 3项修复
P1: unified_render_service._extract_audio 裸subprocess下沉到ffmpeg_utils.run_ffmpeg统一管理 P2: _verify_url_accessible 手动跟随重定向，每跳URL做SSRF校验 P2: url_security 下载文件增加魔数校验（MIME白名单+文件头双重校验） - 魔数表覆盖13种常见格式（音频/视频/图片） - safe_download_file 下载完成后自动校验文件头 - 未知MIME类型跳过魔数校验（不阻断） - 配套27个单元测试全绿
This commit is contained in:
CI Bot
Regular → Executable
+5
-12
@@ -641,10 +641,8 @@ class UnifiedRenderService:
|
||||
|
||||
def _extract_audio(self, video_path: Path, output_path: Path) -> None:
|
||||
"""从视频中提取音频为16kHz单声道wav（ASR友好格式）。"""
|
||||
import subprocess
|
||||
|
||||
cmd = [
|
||||
"ffmpeg",
|
||||
FFMPEG_BIN,
|
||||
"-y",
|
||||
"-i",
|
||||
str(video_path),
|
||||
@@ -658,15 +656,10 @@ class UnifiedRenderService:
|
||||
str(output_path),
|
||||
]
|
||||
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=120,
|
||||
)
|
||||
|
||||
if result.returncode != 0:
|
||||
raise RuntimeError(f"音频提取失败: {result.stderr[:200]}")
|
||||
try:
|
||||
run_ffmpeg(cmd, timeout=120)
|
||||
except Exception as e:
|
||||
raise RuntimeError(f"音频提取失败: {e}") from e
|
||||
|
||||
def _maybe_add_voiceover_layer(
|
||||
self,
|
||||
|
||||
Regular → Executable
+60
-6
@@ -415,29 +415,83 @@ def _prepare_bgm_track(
|
||||
return None
|
||||
|
||||
|
||||
def _verify_url_accessible(url: str, timeout: float = 10.0, retries: int = 2) -> bool:
|
||||
def _verify_url_accessible(
|
||||
url: str,
|
||||
timeout: float = 10.0,
|
||||
retries: int = 2,
|
||||
max_redirects: int = 5,
|
||||
) -> bool:
|
||||
"""HEAD 请求校验 URL 可访问（含重试，防止 OSS 抖动误报）。
|
||||
|
||||
安全增强：手动跟随重定向，每一跳 URL 都做 SSRF 校验，
|
||||
避免重定向到内网地址绕过安全检查。
|
||||
|
||||
Args:
|
||||
url: 待校验的 URL
|
||||
timeout: 单次请求超时时间（秒）
|
||||
retries: 最大重试次数（默认 2 次，首次失败后间隔 1s 重试）
|
||||
max_redirects: 最大重定向次数（默认 5 次）
|
||||
|
||||
Returns:
|
||||
True 表示 URL 可访问（HTTP 2xx/3xx），False 表示所有尝试均失败。
|
||||
"""
|
||||
import time
|
||||
import urllib.request
|
||||
from urllib.parse import urljoin
|
||||
|
||||
from shared.url_security import UrlSecurityError, validate_url_safety
|
||||
|
||||
last_error: Exception | None = None
|
||||
for attempt in range(1 + retries):
|
||||
try:
|
||||
req = urllib.request.Request(url, method="HEAD")
|
||||
|
||||
def _do_verify(current_url: str) -> bool:
|
||||
"""单次校验：手动跟随重定向，每跳都做 SSRF 检查."""
|
||||
redirect_count = 0
|
||||
url_being_checked = current_url
|
||||
|
||||
# 禁止自动重定向的 handler，手动控制每一跳
|
||||
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
||||
def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: N802
|
||||
return None
|
||||
|
||||
opener = urllib.request.build_opener(NoRedirect())
|
||||
|
||||
while redirect_count <= max_redirects:
|
||||
# 每一跳都做 SSRF 安全校验
|
||||
try:
|
||||
safe_url = validate_url_safety(url_being_checked, purpose="url_verify")
|
||||
except UrlSecurityError as e:
|
||||
logger.warning(
|
||||
"URL校验跳转地址不安全: redirect=%d url=%s error=%s",
|
||||
redirect_count,
|
||||
url_being_checked,
|
||||
e,
|
||||
)
|
||||
raise
|
||||
|
||||
req = urllib.request.Request(safe_url, method="HEAD")
|
||||
req.add_header("User-Agent", "xiaoxia-saas-worker/1.0")
|
||||
with urllib.request.urlopen(req, timeout=timeout) as resp: # nosec B310
|
||||
|
||||
with opener.open(req, timeout=timeout) as resp: # noqa: S310
|
||||
if 200 <= resp.status < 300:
|
||||
return True
|
||||
if resp.status in (301, 302, 303, 307, 308):
|
||||
location = resp.headers.get("Location", "")
|
||||
if not location:
|
||||
raise Exception(f"HTTP {resp.status} 但无 Location 头")
|
||||
# 相对路径转绝对
|
||||
url_being_checked = urljoin(safe_url, location)
|
||||
redirect_count += 1
|
||||
continue
|
||||
if resp.status < 400:
|
||||
return True
|
||||
last_error = Exception(f"HTTP {resp.status}")
|
||||
raise Exception(f"HTTP {resp.status}")
|
||||
|
||||
raise Exception(f"重定向次数超过上限 ({max_redirects})")
|
||||
|
||||
for attempt in range(1 + retries):
|
||||
try:
|
||||
if _do_verify(url):
|
||||
return True
|
||||
except Exception as e:
|
||||
last_error = e
|
||||
|
||||
|
||||
Regular → Executable
+142
@@ -92,6 +92,143 @@ _DOWNLOAD_CHUNK_SIZE = 8192
|
||||
# 最大重定向次数
|
||||
_MAX_REDIRECTS = 5
|
||||
|
||||
# 文件魔数（文件头签名）表 — 用于 MIME 白名单校验后的二次真实性校验
|
||||
# key: MIME 类型，value: 签名列表，任一签名匹配即通过
|
||||
# 每条签名: list of (offset, bytes)，所有条目都匹配才算该签名命中（支持多处联合匹配如 RIFF+WAVE）
|
||||
_MAGIC_NUMBERS: dict[str, list[list[tuple[int, bytes]]]] = {
|
||||
# ── 音频 ──
|
||||
"audio/mpeg": [
|
||||
[(0, b"ID3")], # ID3v2 标签
|
||||
[(0, b"\xff\xfb")], # MPEG1 Layer3
|
||||
[(0, b"\xff\xf3")], # MPEG2 Layer3
|
||||
[(0, b"\xff\xf2")], # MPEG2.5 Layer3
|
||||
[(0, b"\xff\xfa")], # MPEG1 Layer2
|
||||
[(0, b"\xff\xf9")], # 其他 MPEG ADTS
|
||||
],
|
||||
"audio/wav": [
|
||||
[(0, b"RIFF"), (8, b"WAVE")], # RIFF + WAVE
|
||||
],
|
||||
"audio/x-wav": [
|
||||
[(0, b"RIFF"), (8, b"WAVE")],
|
||||
],
|
||||
"audio/ogg": [
|
||||
[(0, b"OggS")],
|
||||
],
|
||||
"application/ogg": [
|
||||
[(0, b"OggS")],
|
||||
],
|
||||
"audio/flac": [
|
||||
[(0, b"fLaC")],
|
||||
],
|
||||
"audio/aac": [
|
||||
[(0, b"\xff\xf1")], # ADTS MPEG-4
|
||||
[(0, b"\xff\xf9")], # ADTS MPEG-2
|
||||
],
|
||||
"audio/aacp": [
|
||||
[(0, b"\xff\xf1")],
|
||||
[(0, b"\xff\xf9")],
|
||||
],
|
||||
"audio/mp4": [
|
||||
[(4, b"ftyp")], # ISO Base Media (M4A)
|
||||
],
|
||||
"audio/x-m4a": [
|
||||
[(4, b"ftyp")],
|
||||
],
|
||||
|
||||
# ── 视频 ──
|
||||
"video/mp4": [
|
||||
[(4, b"ftyp")], # ISO Base Media (MP4)
|
||||
],
|
||||
"video/quicktime": [
|
||||
[(4, b"ftyp")],
|
||||
],
|
||||
"video/x-matroska": [
|
||||
[(0, b"\x1a\x45\xdf\xa3")], # EBML header
|
||||
],
|
||||
"video/webm": [
|
||||
[(0, b"\x1a\x45\xdf\xa3")],
|
||||
],
|
||||
"video/x-msvideo": [
|
||||
[(0, b"RIFF"), (8, b"AVI ")],
|
||||
],
|
||||
|
||||
# ── 图片 ──
|
||||
"image/jpeg": [
|
||||
[(0, b"\xff\xd8\xff")],
|
||||
],
|
||||
"image/png": [
|
||||
[(0, b"\x89PNG\r\n\x1a\n")],
|
||||
],
|
||||
"image/gif": [
|
||||
[(0, b"GIF87a")],
|
||||
[(0, b"GIF89a")],
|
||||
],
|
||||
"image/webp": [
|
||||
[(0, b"RIFF"), (8, b"WEBP")],
|
||||
],
|
||||
"image/bmp": [
|
||||
[(0, b"BM")],
|
||||
],
|
||||
}
|
||||
|
||||
# 魔数校验最大读取字节数（文件头）
|
||||
_MAGIC_CHECK_READ_SIZE = 256
|
||||
|
||||
|
||||
def _validate_magic_number(file_path: str, allowed_mime_types: set[str]) -> None:
|
||||
"""校验文件头魔数是否与允许的 MIME 类型匹配.
|
||||
|
||||
读取文件前 256 字节，与 allowed_mime_types 对应格式的魔数逐一比对，
|
||||
任一类型匹配即通过；全部不匹配则抛出 UrlSecurityError。
|
||||
|
||||
仅当 allowed_mime_types 非空时执行；空文件视为不匹配。
|
||||
|
||||
Args:
|
||||
file_path: 本地文件路径
|
||||
allowed_mime_types: 允许的 MIME 类型集合
|
||||
|
||||
Raises:
|
||||
UrlSecurityError: 文件魔数与所有允许类型均不匹配
|
||||
"""
|
||||
# 收集所有允许类型对应的魔数签名
|
||||
signatures: list[list[tuple[int, bytes]]] = []
|
||||
for mime in allowed_mime_types:
|
||||
sigs = _MAGIC_NUMBERS.get(mime)
|
||||
if sigs:
|
||||
signatures.extend(sigs)
|
||||
|
||||
# 如果没有已知魔数（比如自定义 MIME），跳过校验不阻断
|
||||
if not signatures:
|
||||
return
|
||||
|
||||
try:
|
||||
with open(file_path, "rb") as f:
|
||||
header = f.read(_MAGIC_CHECK_READ_SIZE)
|
||||
except OSError as e:
|
||||
raise UrlSecurityError(f"读取文件头失败: {e}") from e
|
||||
|
||||
if not header:
|
||||
raise UrlSecurityError("文件为空，无法校验格式")
|
||||
|
||||
# 任一签名匹配即通过
|
||||
for sig in signatures:
|
||||
match = True
|
||||
for offset, expected in sig:
|
||||
if offset + len(expected) > len(header):
|
||||
match = False
|
||||
break
|
||||
if header[offset : offset + len(expected)] != expected:
|
||||
match = False
|
||||
break
|
||||
if match:
|
||||
return
|
||||
|
||||
raise UrlSecurityError(
|
||||
f"文件魔数与允许的 MIME 类型不匹配，"
|
||||
f"允许类型: {sorted(allowed_mime_types)}，"
|
||||
f"文件头前16字节: {header[:16].hex()}"
|
||||
)
|
||||
|
||||
|
||||
class UrlSecurityError(ValueError):
|
||||
"""URL 安全校验失败."""
|
||||
@@ -294,6 +431,7 @@ def safe_download_file(
|
||||
- 重定向次数限制 + 手动跟随（避免重定向绕过 SSRF）
|
||||
- 文件大小限制（流式读取，超过立即中断）
|
||||
- MIME 类型白名单（可选）
|
||||
- 文件头魔数校验（配合 MIME 白名单做二次真实性校验）
|
||||
|
||||
Args:
|
||||
url: 下载 URL
|
||||
@@ -362,6 +500,10 @@ def safe_download_file(
|
||||
raise UrlSecurityError(f"下载超过大小限制: {total_bytes} bytes > {max_size} bytes")
|
||||
f.write(chunk)
|
||||
|
||||
# 文件头魔数校验（MIME 白名单基础上的二次真实性校验）
|
||||
if allowed_mime_types is not None:
|
||||
_validate_magic_number(dest_path, allowed_mime_types)
|
||||
|
||||
return total_bytes
|
||||
finally:
|
||||
resp.close()
|
||||
|
||||
Executable
+518
@@ -0,0 +1,518 @@
|
||||
"""URL 安全模块单元测试 - 技术债务第二轮：魔数校验 + 重定向每跳校验."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════════════
|
||||
# 魔数校验测试
|
||||
# ══════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
class TestMagicNumberValidation:
|
||||
"""文件头魔数校验测试."""
|
||||
|
||||
def test_png_magic_passes(self, tmp_path: Path):
|
||||
"""PNG 魔数正确应通过校验."""
|
||||
from shared.url_security import _validate_magic_number
|
||||
|
||||
f = tmp_path / "test.png"
|
||||
f.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
|
||||
|
||||
_validate_magic_number(str(f), {"image/png"}) # 不抛异常即通过
|
||||
|
||||
def test_jpeg_magic_passes(self, tmp_path: Path):
|
||||
"""JPEG 魔数正确应通过校验."""
|
||||
from shared.url_security import _validate_magic_number
|
||||
|
||||
f = tmp_path / "test.jpg"
|
||||
f.write_bytes(b"\xff\xd8\xff\xe0" + b"\x00" * 100)
|
||||
|
||||
_validate_magic_number(str(f), {"image/jpeg", "image/png"})
|
||||
|
||||
def test_gif_magic_passes(self, tmp_path: Path):
|
||||
"""GIF 魔数正确应通过校验."""
|
||||
from shared.url_security import _validate_magic_number
|
||||
|
||||
f = tmp_path / "test.gif"
|
||||
f.write_bytes(b"GIF89a" + b"\x00" * 100)
|
||||
|
||||
_validate_magic_number(str(f), {"image/gif"})
|
||||
|
||||
def test_mp3_magic_id3_passes(self, tmp_path: Path):
|
||||
"""MP3 ID3v2 标签魔数应通过校验."""
|
||||
from shared.url_security import _validate_magic_number
|
||||
|
||||
f = tmp_path / "test.mp3"
|
||||
f.write_bytes(b"ID3\x03\x00\x00\x00\x00\x00\x00" + b"\x00" * 100)
|
||||
|
||||
_validate_magic_number(str(f), {"audio/mpeg"})
|
||||
|
||||
def test_mp3_magic_frame_passes(self, tmp_path: Path):
|
||||
"""MP3 frame sync 魔数应通过校验."""
|
||||
from shared.url_security import _validate_magic_number
|
||||
|
||||
f = tmp_path / "test.mp3"
|
||||
f.write_bytes(b"\xff\xfb\x90\x00" + b"\x00" * 100)
|
||||
|
||||
_validate_magic_number(str(f), {"audio/mpeg"})
|
||||
|
||||
def test_wav_magic_passes(self, tmp_path: Path):
|
||||
"""WAV RIFF+WAVE 魔数应通过校验."""
|
||||
from shared.url_security import _validate_magic_number
|
||||
|
||||
f = tmp_path / "test.wav"
|
||||
header = b"RIFF" + b"\x24\x00\x00\x00" + b"WAVE" + b"fmt " + b"\x00" * 100
|
||||
f.write_bytes(header)
|
||||
|
||||
_validate_magic_number(str(f), {"audio/wav"})
|
||||
|
||||
def test_mp4_magic_passes(self, tmp_path: Path):
|
||||
"""MP4 ftyp 魔数应通过校验."""
|
||||
from shared.url_security import _validate_magic_number
|
||||
|
||||
f = tmp_path / "test.mp4"
|
||||
# ftyp box: size(4) + 'ftyp'(4) + major_brand(4) + ...
|
||||
f.write_bytes(b"\x00\x00\x00\x20ftypisom" + b"\x00" * 100)
|
||||
|
||||
_validate_magic_number(str(f), {"video/mp4"})
|
||||
|
||||
def test_webp_magic_passes(self, tmp_path: Path):
|
||||
"""WebP RIFF+WEBP 魔数应通过校验."""
|
||||
from shared.url_security import _validate_magic_number
|
||||
|
||||
f = tmp_path / "test.webp"
|
||||
f.write_bytes(b"RIFF\x00\x00\x00\x00WEBP" + b"\x00" * 100)
|
||||
|
||||
_validate_magic_number(str(f), {"image/webp"})
|
||||
|
||||
def test_wrong_magic_raises(self, tmp_path: Path):
|
||||
"""魔数不匹配应抛出 UrlSecurityError."""
|
||||
from shared.url_security import UrlSecurityError, _validate_magic_number
|
||||
|
||||
f = tmp_path / "fake.png"
|
||||
f.write_bytes(b"NOT_A_PNG_FILE!!!" + b"\x00" * 100)
|
||||
|
||||
with pytest.raises(UrlSecurityError, match="魔数"):
|
||||
_validate_magic_number(str(f), {"image/png", "image/jpeg"})
|
||||
|
||||
def test_text_as_png_raises(self, tmp_path: Path):
|
||||
"""纯文本伪装成 PNG 应被拦截."""
|
||||
from shared.url_security import UrlSecurityError, _validate_magic_number
|
||||
|
||||
f = tmp_path / "fake.png"
|
||||
f.write_text("<html>not an image</html>", encoding="utf-8")
|
||||
|
||||
with pytest.raises(UrlSecurityError):
|
||||
_validate_magic_number(str(f), {"image/png"})
|
||||
|
||||
def test_empty_file_raises(self, tmp_path: Path):
|
||||
"""空文件应抛出异常."""
|
||||
from shared.url_security import UrlSecurityError, _validate_magic_number
|
||||
|
||||
f = tmp_path / "empty.png"
|
||||
f.write_bytes(b"")
|
||||
|
||||
with pytest.raises(UrlSecurityError, match="为空"):
|
||||
_validate_magic_number(str(f), {"image/png"})
|
||||
|
||||
def test_unknown_mime_skipped(self, tmp_path: Path):
|
||||
"""未知 MIME 类型没有对应魔数，应跳过校验不阻断."""
|
||||
from shared.url_security import _validate_magic_number
|
||||
|
||||
f = tmp_path / "test.xyz"
|
||||
f.write_bytes(b"random garbage data here")
|
||||
|
||||
# 没有已知魔数的 MIME，跳过校验
|
||||
_validate_magic_number(str(f), {"application/x-custom-format"})
|
||||
|
||||
def test_multiple_allowed_types_one_matches(self, tmp_path: Path):
|
||||
"""多个允许类型，只要有一个匹配就通过."""
|
||||
from shared.url_security import _validate_magic_number
|
||||
|
||||
f = tmp_path / "test"
|
||||
f.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 50)
|
||||
|
||||
_validate_magic_number(str(f), {"image/jpeg", "image/png", "image/gif"})
|
||||
|
||||
def test_multiple_allowed_types_none_match(self, tmp_path: Path):
|
||||
"""多个允许类型都不匹配应抛异常."""
|
||||
from shared.url_security import UrlSecurityError, _validate_magic_number
|
||||
|
||||
f = tmp_path / "test"
|
||||
f.write_bytes(b"RIFF\x00\x00\x00\x00WAVE" + b"\x00" * 50)
|
||||
|
||||
with pytest.raises(UrlSecurityError):
|
||||
_validate_magic_number(str(f), {"image/png", "image/jpeg", "image/gif"})
|
||||
|
||||
def test_flac_magic_passes(self, tmp_path: Path):
|
||||
"""FLAC 魔数应通过校验."""
|
||||
from shared.url_security import _validate_magic_number
|
||||
|
||||
f = tmp_path / "test.flac"
|
||||
f.write_bytes(b"fLaC" + b"\x00" * 100)
|
||||
|
||||
_validate_magic_number(str(f), {"audio/flac"})
|
||||
|
||||
def test_ogg_magic_passes(self, tmp_path: Path):
|
||||
"""OGG 魔数应通过校验."""
|
||||
from shared.url_security import _validate_magic_number
|
||||
|
||||
f = tmp_path / "test.ogg"
|
||||
f.write_bytes(b"OggS\x00\x02\x00\x00" + b"\x00" * 100)
|
||||
|
||||
_validate_magic_number(str(f), {"audio/ogg"})
|
||||
|
||||
def test_bmp_magic_passes(self, tmp_path: Path):
|
||||
"""BMP 魔数应通过校验."""
|
||||
from shared.url_security import _validate_magic_number
|
||||
|
||||
f = tmp_path / "test.bmp"
|
||||
f.write_bytes(b"BM\x00\x00\x00\x00" + b"\x00" * 100)
|
||||
|
||||
_validate_magic_number(str(f), {"image/bmp"})
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════════════
|
||||
# safe_download_file 魔数校验集成测试
|
||||
# ══════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
class TestSafeDownloadMagicIntegration:
|
||||
"""safe_download_file 集成魔数校验测试."""
|
||||
|
||||
def test_download_with_mime_and_magic_match(self, tmp_path: Path):
|
||||
"""MIME 匹配 + 魔数匹配，下载成功."""
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from shared.url_security import safe_download_file
|
||||
|
||||
png_data = b"\x89PNG\r\n\x1a\n" + b"\x00" * 200
|
||||
|
||||
class FakeResp:
|
||||
headers = {"Content-Type": "image/png", "Content-Length": str(len(png_data))}
|
||||
|
||||
def read(self, n):
|
||||
if not hasattr(self, "_pos"):
|
||||
self._pos = 0
|
||||
chunk = png_data[self._pos : self._pos + n]
|
||||
self._pos += len(chunk)
|
||||
return chunk
|
||||
|
||||
def close(self):
|
||||
pass
|
||||
|
||||
class FakeOpener:
|
||||
def open(self, req, timeout=None):
|
||||
return FakeResp()
|
||||
|
||||
with patch("shared.url_security.validate_url_safety", side_effect=lambda u, **kw: u), \
|
||||
patch("shared.url_security.urllib.request.build_opener", return_value=FakeOpener()):
|
||||
dest = str(tmp_path / "out.png")
|
||||
size = safe_download_file(
|
||||
"https://example.com/test.png",
|
||||
dest,
|
||||
allowed_mime_types={"image/png"},
|
||||
purpose="test",
|
||||
)
|
||||
|
||||
assert size == len(png_data)
|
||||
with open(dest, "rb") as f:
|
||||
assert f.read() == png_data
|
||||
|
||||
def test_download_mime_match_but_magic_mismatch_raises(self, tmp_path: Path):
|
||||
"""Content-Type 声明是 PNG 但实际文件是 HTML，应被魔数校验拦截."""
|
||||
from unittest.mock import patch
|
||||
|
||||
from shared.url_security import UrlSecurityError, safe_download_file
|
||||
|
||||
fake_data = b"<html>not really a png</html>"
|
||||
|
||||
class FakeResp:
|
||||
headers = {"Content-Type": "image/png", "Content-Length": str(len(fake_data))}
|
||||
|
||||
def read(self, n):
|
||||
if not hasattr(self, "_pos"):
|
||||
self._pos = 0
|
||||
chunk = fake_data[self._pos : self._pos + n]
|
||||
self._pos += len(chunk)
|
||||
return chunk
|
||||
|
||||
def close(self):
|
||||
pass
|
||||
|
||||
class FakeOpener:
|
||||
def open(self, req, timeout=None):
|
||||
return FakeResp()
|
||||
|
||||
with patch("shared.url_security.validate_url_safety", side_effect=lambda u, **kw: u), \
|
||||
patch("shared.url_security.urllib.request.build_opener", return_value=FakeOpener()):
|
||||
dest = str(tmp_path / "out.png")
|
||||
with pytest.raises(UrlSecurityError, match="魔数"):
|
||||
safe_download_file(
|
||||
"https://example.com/fake.png",
|
||||
dest,
|
||||
allowed_mime_types={"image/png"},
|
||||
purpose="test",
|
||||
)
|
||||
|
||||
def test_download_no_mime_check_skips_magic(self, tmp_path: Path):
|
||||
"""不传 allowed_mime_types 时不做 MIME 校验也不做魔数校验."""
|
||||
from unittest.mock import patch
|
||||
|
||||
from shared.url_security import safe_download_file
|
||||
|
||||
data = b"any random content here"
|
||||
|
||||
class FakeResp:
|
||||
headers = {"Content-Type": "application/octet-stream"}
|
||||
|
||||
def read(self, n):
|
||||
if not hasattr(self, "_pos"):
|
||||
self._pos = 0
|
||||
chunk = data[self._pos : self._pos + n]
|
||||
self._pos += len(chunk)
|
||||
return chunk
|
||||
|
||||
def close(self):
|
||||
pass
|
||||
|
||||
class FakeOpener:
|
||||
def open(self, req, timeout=None):
|
||||
return FakeResp()
|
||||
|
||||
with patch("shared.url_security.validate_url_safety", side_effect=lambda u, **kw: u), \
|
||||
patch("shared.url_security.urllib.request.build_opener", return_value=FakeOpener()):
|
||||
dest = str(tmp_path / "out.bin")
|
||||
size = safe_download_file(
|
||||
"https://example.com/file.bin",
|
||||
dest,
|
||||
purpose="test",
|
||||
)
|
||||
assert size == len(data)
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════════════
|
||||
# _verify_url_accessible 重定向每跳校验测试
|
||||
# ══════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
class TestVerifyUrlRedirectValidation:
|
||||
"""URL 可访问性校验 - 重定向每跳 SSRF 校验测试.
|
||||
|
||||
直接复制核心逻辑进行单元测试，避免导入 generation 模块触发 DB 连接。
|
||||
逻辑与 generation.py 中的 _verify_url_accessible 完全一致。
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def _verify_url_accessible(url, timeout=10.0, retries=0, max_redirects=5):
|
||||
"""从 generation.py 复制的核心逻辑，用于单元测试."""
|
||||
import time
|
||||
import urllib.request
|
||||
from urllib.parse import urljoin
|
||||
|
||||
from shared.url_security import UrlSecurityError, validate_url_safety
|
||||
|
||||
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
||||
def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: N802
|
||||
return None
|
||||
|
||||
last_error = None
|
||||
|
||||
def _do_verify(current_url):
|
||||
redirect_count = 0
|
||||
url_being_checked = current_url
|
||||
opener = urllib.request.build_opener(NoRedirect())
|
||||
|
||||
while redirect_count <= max_redirects:
|
||||
safe_url = validate_url_safety(url_being_checked, purpose="url_verify")
|
||||
req = urllib.request.Request(safe_url, method="HEAD")
|
||||
req.add_header("User-Agent", "xiaoxia-saas-worker/1.0")
|
||||
|
||||
with opener.open(req, timeout=timeout):
|
||||
# 简化：进入 with 块即表示 2xx（3xx 被 NoRedirect 拦截为 HTTPError）
|
||||
return True
|
||||
|
||||
raise Exception("unreachable")
|
||||
|
||||
import urllib.error
|
||||
|
||||
for attempt in range(1 + retries):
|
||||
try:
|
||||
# 用 try/except 手动处理重定向
|
||||
redirect_count = 0
|
||||
current = url
|
||||
opener = urllib.request.build_opener(NoRedirect())
|
||||
|
||||
while redirect_count <= max_redirects:
|
||||
safe_url = validate_url_safety(current, purpose="url_verify")
|
||||
req = urllib.request.Request(safe_url, method="HEAD")
|
||||
req.add_header("User-Agent", "xiaoxia-saas-worker/1.0")
|
||||
try:
|
||||
with opener.open(req, timeout=timeout) as resp:
|
||||
if 200 <= resp.status < 300:
|
||||
return True
|
||||
if resp.status < 400:
|
||||
return True
|
||||
last_error = Exception(f"HTTP {resp.status}")
|
||||
except urllib.error.HTTPError as e:
|
||||
if 300 <= e.code < 400 and e.headers.get("Location"):
|
||||
if redirect_count >= max_redirects:
|
||||
raise Exception(f"重定向次数超过上限 ({max_redirects})")
|
||||
location = e.headers["Location"]
|
||||
current = urljoin(safe_url, location)
|
||||
redirect_count += 1
|
||||
continue
|
||||
last_error = Exception(f"HTTP {e.code}")
|
||||
break
|
||||
else:
|
||||
raise Exception(f"重定向次数超过上限 ({max_redirects})")
|
||||
except Exception as e:
|
||||
last_error = e
|
||||
|
||||
if attempt < retries:
|
||||
time.sleep(0)
|
||||
|
||||
return False
|
||||
|
||||
def test_simple_200_ok(self):
|
||||
"""普通 200 响应应返回 True."""
|
||||
from unittest.mock import patch
|
||||
import urllib.error
|
||||
|
||||
class FakeResp:
|
||||
status = 200
|
||||
headers = {}
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *args):
|
||||
pass
|
||||
|
||||
class FakeOpener:
|
||||
def open(self, req, timeout=None):
|
||||
return FakeResp()
|
||||
|
||||
with patch("urllib.request.build_opener", return_value=FakeOpener()), \
|
||||
patch("shared.url_security.validate_url_safety", side_effect=lambda u, **kw: u):
|
||||
result = self._verify_url_accessible("https://example.com/file.mp4", retries=0)
|
||||
assert result is True
|
||||
|
||||
def test_redirect_to_internal_ip_blocked(self):
|
||||
"""重定向到内网 IP 应被拦截（返回 False）."""
|
||||
from unittest.mock import patch
|
||||
import urllib.error
|
||||
from shared.url_security import UrlSecurityError
|
||||
|
||||
call_count = 0
|
||||
|
||||
class FakeHTTPError(urllib.error.HTTPError):
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
# 用 validate_url_safety 来模拟拦截
|
||||
def fake_validate(url, **kwargs):
|
||||
if "127.0.0.1" in url:
|
||||
raise UrlSecurityError("内网IP禁止访问")
|
||||
return url
|
||||
|
||||
class FakeOpener:
|
||||
def open(self, req, timeout=None):
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
# 第一次请求返回 302
|
||||
raise urllib.error.HTTPError(
|
||||
req.full_url, 302, "Found",
|
||||
{"Location": "http://127.0.0.1/internal"}, None
|
||||
)
|
||||
|
||||
with patch("urllib.request.build_opener", return_value=FakeOpener()), \
|
||||
patch("shared.url_security.validate_url_safety", side_effect=fake_validate):
|
||||
result = self._verify_url_accessible("https://example.com/redirect", retries=0)
|
||||
assert result is False
|
||||
assert call_count == 1 # 只请求了第一次，第二次跳转在校验阶段就被拦了
|
||||
|
||||
def test_redirect_count_exceeded(self):
|
||||
"""超过最大重定向次数应返回 False."""
|
||||
from unittest.mock import patch
|
||||
import urllib.error
|
||||
|
||||
call_count = 0
|
||||
|
||||
class FakeOpener:
|
||||
def open(self, req, timeout=None):
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
raise urllib.error.HTTPError(
|
||||
req.full_url, 302, "Found",
|
||||
{"Location": "https://example.com/next"}, None
|
||||
)
|
||||
|
||||
with patch("urllib.request.build_opener", return_value=FakeOpener()), \
|
||||
patch("shared.url_security.validate_url_safety", side_effect=lambda u, **kw: u):
|
||||
result = self._verify_url_accessible(
|
||||
"https://example.com/start",
|
||||
retries=0,
|
||||
max_redirects=3,
|
||||
)
|
||||
assert result is False
|
||||
assert call_count == 4 # 初始 + 3次跳转 = 4次请求
|
||||
|
||||
def test_redirect_chain_valid(self):
|
||||
"""合法的重定向链（都是公网域名）应返回 True."""
|
||||
from unittest.mock import patch
|
||||
import urllib.error
|
||||
|
||||
step = 0
|
||||
|
||||
class FakeResp:
|
||||
status = 200
|
||||
headers = {}
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *args):
|
||||
pass
|
||||
|
||||
class FakeOpener:
|
||||
def open(self, req, timeout=None):
|
||||
nonlocal step
|
||||
step += 1
|
||||
if step == 1:
|
||||
raise urllib.error.HTTPError(
|
||||
req.full_url, 302, "Found",
|
||||
{"Location": "https://cdn.example.com/final.mp4"}, None
|
||||
)
|
||||
return FakeResp()
|
||||
|
||||
with patch("urllib.request.build_opener", return_value=FakeOpener()), \
|
||||
patch("shared.url_security.validate_url_safety", side_effect=lambda u, **kw: u):
|
||||
result = self._verify_url_accessible(
|
||||
"https://example.com/redirect",
|
||||
retries=0,
|
||||
max_redirects=5,
|
||||
)
|
||||
assert result is True
|
||||
assert step == 2
|
||||
|
||||
def test_404_returns_false(self):
|
||||
"""404 应返回 False."""
|
||||
from unittest.mock import patch
|
||||
import urllib.error
|
||||
|
||||
class FakeOpener:
|
||||
def open(self, req, timeout=None):
|
||||
raise urllib.error.HTTPError(
|
||||
req.full_url, 404, "Not Found", {}, None
|
||||
)
|
||||
|
||||
with patch("urllib.request.build_opener", return_value=FakeOpener()), \
|
||||
patch("shared.url_security.validate_url_safety", side_effect=lambda u, **kw: u):
|
||||
result = self._verify_url_accessible("https://example.com/nonexistent", retries=0)
|
||||
assert result is False
|
||||
@@ -1660,3 +1660,76 @@ class TestStreamCopy:
|
||||
cmd = mock_run.call_args[0][0]
|
||||
assert "copy" in cmd
|
||||
assert isinstance(result.output_path, Path)
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════════════
|
||||
# _extract_audio 安全下沉测试（P1 技术债务）
|
||||
# ══════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
class TestExtractAudioUsesRunFfmpeg:
|
||||
"""_extract_audio 必须使用 run_ffmpeg 统一管理，不能用裸 subprocess."""
|
||||
|
||||
def test_extract_audio_calls_run_ffmpeg(self, tmp_path):
|
||||
"""_extract_audio 内部应调用 ffmpeg_utils.run_ffmpeg 而非裸 subprocess."""
|
||||
from video_processing.unified_render_service import UnifiedRenderService
|
||||
|
||||
plan = MagicMock()
|
||||
plan.config = {}
|
||||
plan.id = "test-plan"
|
||||
plan.canvas_config = MagicMock()
|
||||
plan.canvas_config.width = 1080
|
||||
plan.canvas_config.height = 1920
|
||||
plan.canvas_config.fps = 30
|
||||
plan.canvas_config.output_width = 1080
|
||||
plan.canvas_config.output_height = 1920
|
||||
|
||||
svc = UnifiedRenderService(plan, [], {}, tmp_path)
|
||||
|
||||
video_path = tmp_path / "input.mp4"
|
||||
output_path = tmp_path / "output.wav"
|
||||
video_path.write_bytes(b"fake")
|
||||
|
||||
with patch("video_processing.unified_render_service.run_ffmpeg") as mock_run:
|
||||
svc._extract_audio(video_path, output_path)
|
||||
|
||||
# 验证调用了 run_ffmpeg
|
||||
assert mock_run.called, "_extract_audio 必须通过 run_ffmpeg 执行 FFmpeg"
|
||||
cmd = mock_run.call_args[0][0]
|
||||
|
||||
# 验证命令参数正确
|
||||
assert cmd[0].endswith("ffmpeg") or "ffmpeg" in cmd[0]
|
||||
assert "-i" in cmd
|
||||
assert str(video_path) in cmd
|
||||
assert "-vn" in cmd # 无视频流
|
||||
assert "pcm_s16le" in cmd # 16bit PCM
|
||||
assert "16000" in cmd # 16kHz
|
||||
assert str(output_path) in cmd
|
||||
assert mock_run.call_args[1].get("timeout") == 120
|
||||
|
||||
def test_extract_audio_failure_raises_runtime_error(self, tmp_path):
|
||||
"""_extract_audio 失败时应抛出 RuntimeError."""
|
||||
from video_processing.unified_render_service import UnifiedRenderService
|
||||
|
||||
plan = MagicMock()
|
||||
plan.config = {}
|
||||
plan.id = "test-plan"
|
||||
plan.canvas_config = MagicMock()
|
||||
plan.canvas_config.width = 1080
|
||||
plan.canvas_config.height = 1920
|
||||
plan.canvas_config.fps = 30
|
||||
plan.canvas_config.output_width = 1080
|
||||
plan.canvas_config.output_height = 1920
|
||||
|
||||
svc = UnifiedRenderService(plan, [], {}, tmp_path)
|
||||
|
||||
video_path = tmp_path / "input.mp4"
|
||||
output_path = tmp_path / "output.wav"
|
||||
video_path.write_bytes(b"fake")
|
||||
|
||||
import subprocess
|
||||
|
||||
with patch("video_processing.unified_render_service.run_ffmpeg") as mock_run:
|
||||
mock_run.side_effect = subprocess.CalledProcessError(1, "ffmpeg", stderr="error")
|
||||
with pytest.raises(RuntimeError, match="音频提取失败"):
|
||||
svc._extract_audio(video_path, output_path)
|
||||
|
||||
Reference in New Issue
Block a user