mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-08-04 16:38:49 +00:00
fix(wecom): sanitize fallback media filename
This commit is contained in:
parent
98507ae4fe
commit
be5af019b9
@ -32,10 +32,12 @@ _SAFE_NAME_RE = re.compile(r"[^\w.\-()\[\]()【】\u4e00-\u9fff]+", re.UNICO
|
||||
|
||||
def _sanitize_filename(name: str, fallback: str = "unnamed") -> str:
|
||||
"""Sanitize filename to avoid traversal and problematic chars."""
|
||||
name = (name or "").strip()
|
||||
name = Path(name).name
|
||||
name = _SAFE_NAME_RE.sub("_", name).strip("._ ")
|
||||
return name or fallback
|
||||
def _clean(value: str) -> str:
|
||||
value = (value or "").strip()
|
||||
value = Path(value).name
|
||||
return _SAFE_NAME_RE.sub("_", value).strip("._ ")
|
||||
|
||||
return _clean(name) or _clean(fallback) or "unnamed"
|
||||
|
||||
|
||||
_IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp"}
|
||||
|
||||
@ -99,8 +99,10 @@ def test_sanitize_filename_empty_input() -> None:
|
||||
def test_sanitize_filename_empty_or_dots_fallback() -> None:
|
||||
assert _sanitize_filename("...") == "unnamed"
|
||||
assert _sanitize_filename("..", fallback="fallback.txt") == "fallback.txt"
|
||||
assert _sanitize_filename("...", fallback="../../outside.txt") == "outside.txt"
|
||||
assert _sanitize_filename("") == "unnamed"
|
||||
|
||||
|
||||
def test_guess_wecom_media_type_image() -> None:
|
||||
for ext in (".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp"):
|
||||
assert _guess_wecom_media_type(f"photo{ext}") == "image"
|
||||
@ -149,6 +151,27 @@ async def test_download_and_save_success() -> None:
|
||||
os.unlink(path)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_download_and_save_sanitizes_sdk_fallback(tmp_path: Path) -> None:
|
||||
"""An unsafe SDK filename cannot escape the channel media directory."""
|
||||
channel = WecomChannel(WecomConfig(bot_id="b", secret="s", allow_from=["*"]), MessageBus())
|
||||
client = _FakeWeComClient()
|
||||
client.download_file.return_value = (b"payload", "../../outside.txt")
|
||||
channel._client = client
|
||||
|
||||
with patch("nanobot.channels.wecom.runtime.get_media_dir", return_value=tmp_path):
|
||||
path = await channel._download_and_save_media(
|
||||
"https://example.com/file",
|
||||
"aes_key",
|
||||
"file",
|
||||
"...",
|
||||
)
|
||||
|
||||
assert path is not None
|
||||
assert Path(path) == tmp_path / "outside.txt"
|
||||
assert Path(path).read_bytes() == b"payload"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_download_and_save_oversized_rejected() -> None:
|
||||
"""Data exceeding 200MB is rejected → returns None."""
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user