Grayscale/nanobot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(webui): support image uploads in composer and message bubbles

Browse Source
This commit is contained in:
Xubin Ren
2026-04-23 00:07:27 +08:00
committed by Xubin Ren
parent c1e7aa5504
commit 61a28c2c0a
39 changed files with 3670 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/channels/test_websocket_envelope_media.py
+416
View File
@@ -0,0 +1,416 @@
"""Tests for WS envelope media handling (client image upload path).
Exercises ``WebSocketChannel._dispatch_envelope`` for the ``message`` branch:
decoding base64 data URLs, rejecting malformed / oversized / non-whitelisted
payloads, preserving backward compatibility with media-less frames, and
forwarding saved paths to ``_handle_message``.
"""
from __future__ import annotations
import base64
import json
from pathlib import Path
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from nanobot.channels.websocket import (
WebSocketChannel,
_extract_data_url_mime,
)
def _tiny_png_data_url() -> str:
"""A 1-pixel PNG prefixed as a data URL — just enough for magic-bytes sniffing."""
# 1x1 transparent PNG
png = (
b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00"
b"\x00\x01\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\rIDATx"
b"\x9cc\xf8\xcf\xc0\x00\x00\x00\x03\x00\x01\x00\x18\xdd\x8d\xb4\x00"
b"\x00\x00\x00IEND\xaeB`\x82"
)
return f"data:image/png;base64,{base64.b64encode(png).decode()}"
def _data_url(mime: str, payload: bytes) -> str:
return f"data:{mime};base64,{base64.b64encode(payload).decode()}"
def _make_channel() -> WebSocketChannel:
bus = MagicMock()
bus.publish_inbound = AsyncMock()
channel = WebSocketChannel(
{"enabled": True, "allowFrom": ["*"], "websocketRequiresToken": False},
bus,
)
channel._handle_message = AsyncMock() # type: ignore[method-assign]
return channel
# -- Pure helpers --------------------------------------------------------------
@pytest.mark.parametrize(
("url", "expected"),
[
("data:image/png;base64,AAAA", "image/png"),
("data:image/jpeg;base64,AAAA", "image/jpeg"),
("data:IMAGE/PNG;base64,AAAA", "image/png"),
("data:image/svg+xml;base64,AAAA", "image/svg+xml"),
("data:text/plain;base64,AAAA", "text/plain"),
("http://evil.example/x.png", None),
("data:image/png,AAAA", None), # missing `;base64`
("", None),
(None, None),
],
)
def test_extract_data_url_mime(url: Any, expected: str | None) -> None:
assert _extract_data_url_mime(url) == expected
# -- max_message_bytes bump ----------------------------------------------------
def test_max_message_bytes_default_supports_multi_image_frame() -> None:
"""Default 36 MB must comfortably hold 4 × 6 MB base64-encoded images."""
from nanobot.channels.websocket import WebSocketConfig
default = WebSocketConfig().max_message_bytes
# 4 images × 6 MB × 1.37 base64 overhead ≈ 33 MB
assert default >= 33 * 1024 * 1024
# Upper bound 40 MB matches plan
with pytest.raises(Exception):
WebSocketConfig(max_message_bytes=41_943_040 + 1)
# -- _dispatch_envelope message branch + media --------------------------------
@pytest.mark.asyncio
async def test_message_without_media_backward_compatible() -> None:
"""Existing clients that don't send ``media`` keep working unchanged."""
channel = _make_channel()
mock_conn = AsyncMock()
envelope = {"type": "message", "chat_id": "abc123", "content": "hello"}
await channel._dispatch_envelope(mock_conn, "client-1", envelope)
channel._handle_message.assert_awaited_once()
call = channel._handle_message.call_args
assert call.kwargs["chat_id"] == "abc123"
assert call.kwargs["content"] == "hello"
# When no media, we pass ``media=None`` so downstream treats it as absent.
assert call.kwargs["media"] is None
@pytest.mark.asyncio
async def test_message_with_single_image_forwards_saved_path(tmp_path) -> None:
channel = _make_channel()
mock_conn = AsyncMock()
envelope = {
"type": "message",
"chat_id": "abc123",
"content": "look at this",
"media": [{"data_url": _tiny_png_data_url(), "name": "shot.png"}],
}
with patch(
"nanobot.channels.websocket.get_media_dir", return_value=tmp_path
):
await channel._dispatch_envelope(mock_conn, "client-1", envelope)
channel._handle_message.assert_awaited_once()
paths = channel._handle_message.call_args.kwargs["media"]
assert isinstance(paths, list) and len(paths) == 1
saved = Path(paths[0])
assert saved.exists()
assert saved.suffix == ".png"
assert saved.is_relative_to(tmp_path)
@pytest.mark.asyncio
async def test_message_with_multiple_images(tmp_path) -> None:
channel = _make_channel()
mock_conn = AsyncMock()
envelope = {
"type": "message",
"chat_id": "abc123",
"content": "a couple",
"media": [
{"data_url": _tiny_png_data_url()},
{"data_url": _tiny_png_data_url()},
{"data_url": _tiny_png_data_url()},
],
}
with patch(
"nanobot.channels.websocket.get_media_dir", return_value=tmp_path
):
await channel._dispatch_envelope(mock_conn, "client-1", envelope)
paths = channel._handle_message.call_args.kwargs["media"]
assert len(paths) == 3
# Saved filenames must be unique.
assert len({Path(p).name for p in paths}) == 3
@pytest.mark.asyncio
async def test_image_only_message_allows_empty_text(tmp_path) -> None:
"""When media is attached, empty text is acceptable."""
channel = _make_channel()
mock_conn = AsyncMock()
envelope = {
"type": "message",
"chat_id": "abc123",
"content": "",
"media": [{"data_url": _tiny_png_data_url()}],
}
with patch(
"nanobot.channels.websocket.get_media_dir", return_value=tmp_path
):
await channel._dispatch_envelope(mock_conn, "client-1", envelope)
channel._handle_message.assert_awaited_once()
# Error event NOT sent.
mock_conn.send.assert_not_awaited()
@pytest.mark.asyncio
async def test_message_rejected_when_more_than_four_images(tmp_path) -> None:
channel = _make_channel()
mock_conn = AsyncMock()
envelope = {
"type": "message",
"chat_id": "abc123",
"content": "hi",
"media": [{"data_url": _tiny_png_data_url()}] * 5,
}
with patch(
"nanobot.channels.websocket.get_media_dir", return_value=tmp_path
):
await channel._dispatch_envelope(mock_conn, "client-1", envelope)
channel._handle_message.assert_not_awaited()
mock_conn.send.assert_awaited_once()
err = json.loads(mock_conn.send.call_args[0][0])
assert err["event"] == "error"
assert err["detail"] == "image_rejected"
assert err["reason"] == "too_many_images"
@pytest.mark.asyncio
async def test_message_rejected_on_oversize_payload(tmp_path) -> None:
channel = _make_channel()
mock_conn = AsyncMock()
oversized = b"x" * (9 * 1024 * 1024) # > 8 MB WS limit
envelope = {
"type": "message",
"chat_id": "abc123",
"content": "big",
"media": [{"data_url": _data_url("image/png", oversized)}],
}
with patch(
"nanobot.channels.websocket.get_media_dir", return_value=tmp_path
):
await channel._dispatch_envelope(mock_conn, "client-1", envelope)
channel._handle_message.assert_not_awaited()
err = json.loads(mock_conn.send.call_args[0][0])
assert err["detail"] == "image_rejected"
assert err["reason"] == "size"
@pytest.mark.asyncio
async def test_message_rejected_on_non_image_mime(tmp_path) -> None:
channel = _make_channel()
mock_conn = AsyncMock()
envelope = {
"type": "message",
"chat_id": "abc123",
"content": "pdf?",
"media": [{"data_url": _data_url("application/pdf", b"%PDF-1.4")}],
}
with patch(
"nanobot.channels.websocket.get_media_dir", return_value=tmp_path
):
await channel._dispatch_envelope(mock_conn, "client-1", envelope)
channel._handle_message.assert_not_awaited()
err = json.loads(mock_conn.send.call_args[0][0])
assert err["detail"] == "image_rejected"
assert err["reason"] == "mime"
@pytest.mark.asyncio
async def test_message_rejected_on_svg_mime(tmp_path) -> None:
"""SVG is explicitly rejected — XSS surface inside embedded scripts."""
channel = _make_channel()
mock_conn = AsyncMock()
envelope = {
"type": "message",
"chat_id": "abc123",
"content": "svg",
"media": [{"data_url": _data_url("image/svg+xml", b"<svg/>")}],
}
with patch(
"nanobot.channels.websocket.get_media_dir", return_value=tmp_path
):
await channel._dispatch_envelope(mock_conn, "client-1", envelope)
channel._handle_message.assert_not_awaited()
err = json.loads(mock_conn.send.call_args[0][0])
assert err["reason"] == "mime"
@pytest.mark.asyncio
async def test_message_rejected_on_malformed_data_url(tmp_path) -> None:
channel = _make_channel()
mock_conn = AsyncMock()
envelope = {
"type": "message",
"chat_id": "abc123",
"content": "nope",
"media": [{"data_url": "http://evil.example/image.png"}],
}
with patch(
"nanobot.channels.websocket.get_media_dir", return_value=tmp_path
):
await channel._dispatch_envelope(mock_conn, "client-1", envelope)
channel._handle_message.assert_not_awaited()
err = json.loads(mock_conn.send.call_args[0][0])
assert err["reason"] == "decode"
@pytest.mark.asyncio
async def test_message_rejected_on_broken_base64(tmp_path) -> None:
channel = _make_channel()
mock_conn = AsyncMock()
envelope = {
"type": "message",
"chat_id": "abc123",
"content": "nope",
"media": [{"data_url": "data:image/png;base64,not-valid-base64!!!"}],
}
with patch(
"nanobot.channels.websocket.get_media_dir", return_value=tmp_path
):
await channel._dispatch_envelope(mock_conn, "client-1", envelope)
channel._handle_message.assert_not_awaited()
err = json.loads(mock_conn.send.call_args[0][0])
assert err["reason"] == "decode"
@pytest.mark.asyncio
async def test_message_rejected_when_media_item_shape_wrong(tmp_path) -> None:
channel = _make_channel()
mock_conn = AsyncMock()
envelope = {
"type": "message",
"chat_id": "abc123",
"content": "huh",
# Not a dict — plain string at the top level.
"media": ["data:image/png;base64,XXXX"],
}
with patch(
"nanobot.channels.websocket.get_media_dir", return_value=tmp_path
):
await channel._dispatch_envelope(mock_conn, "client-1", envelope)
channel._handle_message.assert_not_awaited()
err = json.loads(mock_conn.send.call_args[0][0])
assert err["reason"] == "malformed"
@pytest.mark.asyncio
async def test_message_rejected_when_media_field_is_not_list() -> None:
channel = _make_channel()
mock_conn = AsyncMock()
envelope = {
"type": "message",
"chat_id": "abc123",
"content": "huh",
"media": "not-a-list",
}
await channel._dispatch_envelope(mock_conn, "client-1", envelope)
channel._handle_message.assert_not_awaited()
err = json.loads(mock_conn.send.call_args[0][0])
assert err["detail"] == "image_rejected"
assert err["reason"] == "malformed"
@pytest.mark.asyncio
async def test_failed_media_does_not_partially_persist(tmp_path) -> None:
"""If the second image is invalid, the first must not be forwarded."""
channel = _make_channel()
mock_conn = AsyncMock()
envelope = {
"type": "message",
"chat_id": "abc123",
"content": "mixed",
"media": [
{"data_url": _tiny_png_data_url()},
{"data_url": _data_url("application/pdf", b"%PDF-1.4")},
],
}
with patch(
"nanobot.channels.websocket.get_media_dir", return_value=tmp_path
):
await channel._dispatch_envelope(mock_conn, "client-1", envelope)
channel._handle_message.assert_not_awaited()
# The first image was saved to disk (we don't roll it back — the caller
# is expected to not reference it) but the agent never sees the paths.
# That's the important invariant: no partial publish.
err = json.loads(mock_conn.send.call_args[0][0])
assert err["reason"] == "mime"
@pytest.mark.asyncio
async def test_rejects_empty_text_without_media() -> None:
"""When no media is attached, whitespace-only content is still rejected
(matches the existing behavior for backward compat)."""
channel = _make_channel()
mock_conn = AsyncMock()
envelope = {
"type": "message",
"chat_id": "abc123",
"content": " ",
}
await channel._dispatch_envelope(mock_conn, "client-1", envelope)
channel._handle_message.assert_not_awaited()
err = json.loads(mock_conn.send.call_args[0][0])
assert err["detail"] == "missing content"
@pytest.mark.asyncio
async def test_non_string_content_still_rejected() -> None:
channel = _make_channel()
mock_conn = AsyncMock()
envelope = {
"type": "message",
"chat_id": "abc123",
"content": 42,
}
await channel._dispatch_envelope(mock_conn, "client-1", envelope)
channel._handle_message.assert_not_awaited()
err = json.loads(mock_conn.send.call_args[0][0])
assert err["detail"] == "missing content"
tests/channels/test_websocket_media_route.py
+375
View File
@@ -0,0 +1,375 @@
"""Tests for the signed ``/api/media/<sig>/<payload>`` route and its replay
integration on ``/api/sessions/<key>/messages``.
The route is the return path for images attached to persisted user turns:
:meth:`WebSocketChannel._sign_media_path` mints URLs during session reads,
and :meth:`WebSocketChannel._handle_media_fetch` serves the bytes back.
These tests cover the two halves end-to-end plus the adversarial edges
(bad signatures, ``..`` traversal, non-existent files, non-image types).
"""
from __future__ import annotations
import asyncio
import functools
import hashlib
import hmac
from pathlib import Path
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
import pytest
from nanobot.channels.websocket import (
WebSocketChannel,
_b64url_decode,
_b64url_encode,
)
from nanobot.session.manager import Session, SessionManager
# PNG magic bytes + a couple of sentinel bytes so we can verify byte-for-byte
# round-trip of the served payload. Stays under mimetype + size limits.
_PNG_BYTES = (
b"\x89PNG\r\n\x1a\n"
b"\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01"
b"\x08\x06\x00\x00\x00\x1f\x15\xc4\x89"
b"\x00\x00\x00\nIDATx\x9cc\x00\x00\x00\x02\x00\x01"
b"\x00\x00\x00\x00IEND\xaeB`\x82"
)
def _ch(
bus: Any,
*,
session_manager: SessionManager | None = None,
port: int,
) -> WebSocketChannel:
return WebSocketChannel(
{
"enabled": True,
"allowFrom": ["*"],
"host": "127.0.0.1",
"port": port,
"path": "/",
"websocketRequiresToken": False,
},
bus,
session_manager=session_manager,
)
@pytest.fixture()
def bus() -> MagicMock:
b = MagicMock()
b.publish_inbound = AsyncMock()
return b
async def _http_get(
url: str, headers: dict[str, str] | None = None
) -> httpx.Response:
return await asyncio.to_thread(
functools.partial(httpx.get, url, headers=headers or {}, timeout=5.0)
)
# ---------------------------------------------------------------------------
# _sign_media_path: the URL minter
# ---------------------------------------------------------------------------
def test_sign_media_path_rejects_paths_outside_media_root(
bus: MagicMock, tmp_path: Path
) -> None:
"""Paths that resolve outside ``get_media_dir()`` must not be signed.
This is the single most important invariant of the whole scheme:
if the minter ever signed an arbitrary path, the HMAC would legitimise
it for the fetch handler and we'd hand out a disk-read primitive.
"""
outside = tmp_path / "secrets" / "cred.txt"
outside.parent.mkdir()
outside.write_text("nope")
media = tmp_path / "media"
media.mkdir()
channel = _ch(bus, port=0)
with patch("nanobot.channels.websocket.get_media_dir", return_value=media):
assert channel._sign_media_path(outside) is None
# Traversal via the media root is also rejected — the resolve() step
# normalises ``..`` out before the relative_to check.
assert channel._sign_media_path(media / ".." / "secrets" / "cred.txt") is None
def test_sign_media_path_round_trips_via_hmac(
bus: MagicMock, tmp_path: Path
) -> None:
"""The signature embeds exactly ``HMAC-SHA256(secret, payload)[:16]``."""
media = tmp_path / "media"
media.mkdir()
(media / "a.png").write_bytes(_PNG_BYTES)
channel = _ch(bus, port=0)
with patch("nanobot.channels.websocket.get_media_dir", return_value=media):
url = channel._sign_media_path(media / "a.png")
assert url is not None
assert url.startswith("/api/media/")
sig, payload = url[len("/api/media/"):].split("/", 1)
expected = hmac.new(
channel._media_secret, payload.encode("ascii"), hashlib.sha256
).digest()[:16]
assert _b64url_decode(sig) == expected
# The payload decodes back to the *relative* path — no absolute-path leaks.
assert _b64url_decode(payload).decode() == "a.png"
# ---------------------------------------------------------------------------
# /api/media/<sig>/<payload>: the serving handler
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_media_route_serves_signed_file(
bus: MagicMock, tmp_path: Path
) -> None:
"""Valid signature + existing file => 200 with correct bytes + MIME."""
media = tmp_path / "media"
media.mkdir()
target = media / "round-trip.png"
target.write_bytes(_PNG_BYTES)
channel = _ch(bus, port=29920)
with patch("nanobot.channels.websocket.get_media_dir", return_value=media):
url_path = channel._sign_media_path(target)
assert url_path is not None
server_task = asyncio.create_task(channel.start())
await asyncio.sleep(0.3)
try:
resp = await _http_get(f"http://127.0.0.1:29920{url_path}")
finally:
await channel.stop()
await server_task
assert resp.status_code == 200
assert resp.content == _PNG_BYTES
assert resp.headers["content-type"].startswith("image/png")
# Immutable cache header lets the browser skip round-trips on replay.
assert "immutable" in resp.headers.get("cache-control", "")
@pytest.mark.asyncio
async def test_media_route_rejects_bad_signature(
bus: MagicMock, tmp_path: Path
) -> None:
"""A payload re-signed with a different secret must 401.
Protects against a restart: old URLs baked into a stale tab become
un-forgeable once ``_media_secret`` regenerates.
"""
media = tmp_path / "media"
media.mkdir()
(media / "f.png").write_bytes(_PNG_BYTES)
channel = _ch(bus, port=29921)
with patch("nanobot.channels.websocket.get_media_dir", return_value=media):
good = channel._sign_media_path(media / "f.png")
assert good is not None
_, payload = good[len("/api/media/"):].split("/", 1)
# Forge a sig with a *different* secret.
forged_mac = hmac.new(
b"\x00" * 32, payload.encode("ascii"), hashlib.sha256
).digest()[:16]
forged = f"/api/media/{_b64url_encode(forged_mac)}/{payload}"
server_task = asyncio.create_task(channel.start())
await asyncio.sleep(0.3)
try:
resp = await _http_get(f"http://127.0.0.1:29921{forged}")
finally:
await channel.stop()
await server_task
assert resp.status_code == 401
@pytest.mark.asyncio
async def test_media_route_rejects_path_traversal_payload(
bus: MagicMock, tmp_path: Path
) -> None:
"""Even a validly-signed ``..`` payload must not escape the media root.
The signer never *emits* such payloads, but an attacker who somehow
obtained the secret (or the channel was misconfigured) must still be
stopped by the resolve()+relative_to() guard in the serving path.
"""
media = tmp_path / "media"
media.mkdir()
secret_file = tmp_path / "secret.txt"
secret_file.write_text("classified")
channel = _ch(bus, port=29922)
# Hand-craft a traversal payload the legit signer would refuse to mint.
payload = _b64url_encode(b"../secret.txt")
mac = hmac.new(
channel._media_secret, payload.encode("ascii"), hashlib.sha256
).digest()[:16]
url = f"/api/media/{_b64url_encode(mac)}/{payload}"
with patch("nanobot.channels.websocket.get_media_dir", return_value=media):
server_task = asyncio.create_task(channel.start())
await asyncio.sleep(0.3)
try:
resp = await _http_get(f"http://127.0.0.1:29922{url}")
finally:
await channel.stop()
await server_task
assert resp.status_code == 404
assert b"classified" not in resp.content
@pytest.mark.asyncio
async def test_media_route_404s_missing_file(
bus: MagicMock, tmp_path: Path
) -> None:
"""A signed URL for a file that no longer exists degrades to 404 so the
client can fall back to the placeholder tile instead of breaking."""
media = tmp_path / "media"
media.mkdir()
target = media / "gone.png"
target.write_bytes(_PNG_BYTES)
channel = _ch(bus, port=29923)
with patch("nanobot.channels.websocket.get_media_dir", return_value=media):
url_path = channel._sign_media_path(target)
assert url_path is not None
target.unlink() # the file vanishes between signing and fetching
server_task = asyncio.create_task(channel.start())
await asyncio.sleep(0.3)
try:
resp = await _http_get(f"http://127.0.0.1:29923{url_path}")
finally:
await channel.stop()
await server_task
assert resp.status_code == 404
@pytest.mark.asyncio
async def test_media_route_degrades_non_image_to_octet_stream(
bus: MagicMock, tmp_path: Path
) -> None:
"""A non-image extension must not be served as its native MIME.
Defence-in-depth: if media_dir ever contained (say) an HTML file, we
do not want the browser to render it as HTML via the signed route.
"""
media = tmp_path / "media"
media.mkdir()
(media / "scary.html").write_bytes(b"<script>alert(1)</script>")
channel = _ch(bus, port=29924)
with patch("nanobot.channels.websocket.get_media_dir", return_value=media):
payload = _b64url_encode(b"scary.html")
mac = hmac.new(
channel._media_secret, payload.encode("ascii"), hashlib.sha256
).digest()[:16]
url = f"/api/media/{_b64url_encode(mac)}/{payload}"
server_task = asyncio.create_task(channel.start())
await asyncio.sleep(0.3)
try:
resp = await _http_get(f"http://127.0.0.1:29924{url}")
finally:
await channel.stop()
await server_task
assert resp.status_code == 200
assert resp.headers["content-type"].startswith("application/octet-stream")
# ---------------------------------------------------------------------------
# /api/sessions/<key>/messages: media_urls hydration on session read
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_session_messages_exposes_signed_media_urls(
bus: MagicMock, tmp_path: Path
) -> None:
"""The read path must map persisted ``media`` paths onto signed URLs
and strip the raw path — the client never learns the server's layout."""
media = tmp_path / "media"
media.mkdir()
img = media / "u.png"
img.write_bytes(_PNG_BYTES)
sm = SessionManager(tmp_path / "ws_state")
sess = Session(key="websocket:media-hydrate")
sess.add_message("user", "look at this", media=[str(img)])
sess.add_message("assistant", "nice")
sm.save(sess)
channel = _ch(bus, session_manager=sm, port=29925)
with patch("nanobot.channels.websocket.get_media_dir", return_value=media):
server_task = asyncio.create_task(channel.start())
await asyncio.sleep(0.3)
try:
boot = await _http_get("http://127.0.0.1:29925/webui/bootstrap")
token = boot.json()["token"]
auth = {"Authorization": f"Bearer {token}"}
resp = await _http_get(
"http://127.0.0.1:29925/api/sessions/websocket:media-hydrate/messages",
headers=auth,
)
body = resp.json()
# The signed URL round-trips end-to-end: fetching it yields the same bytes.
user_msg = next(m for m in body["messages"] if m["role"] == "user")
urls = user_msg["media_urls"]
assert isinstance(urls, list) and len(urls) == 1
assert urls[0]["name"] == "u.png"
assert urls[0]["url"].startswith("/api/media/")
# Raw paths must not leak to the wire.
assert "media" not in user_msg
# And the URL actually works.
fetched = await _http_get(f"http://127.0.0.1:29925{urls[0]['url']}")
assert fetched.status_code == 200
assert fetched.content == _PNG_BYTES
finally:
await channel.stop()
await server_task
@pytest.mark.asyncio
async def test_session_messages_skips_vanished_media(
bus: MagicMock, tmp_path: Path
) -> None:
"""Paths that no longer resolve inside the media root produce no URL —
the message is still delivered, just without the preview."""
media = tmp_path / "media"
media.mkdir()
sm = SessionManager(tmp_path / "ws_state")
sess = Session(key="websocket:vanished")
sess.add_message("user", "missing pic", media=[str(media / "absent.png")])
sm.save(sess)
channel = _ch(bus, session_manager=sm, port=29926)
with patch("nanobot.channels.websocket.get_media_dir", return_value=media):
server_task = asyncio.create_task(channel.start())
await asyncio.sleep(0.3)
try:
boot = await _http_get("http://127.0.0.1:29926/webui/bootstrap")
token = boot.json()["token"]
resp = await _http_get(
"http://127.0.0.1:29926/api/sessions/websocket:vanished/messages",
headers={"Authorization": f"Bearer {token}"},
)
user_msg = next(m for m in resp.json()["messages"] if m["role"] == "user")
# absent.png lives inside the media root so it *does* get a signed
# URL (we don't stat the file at signing time — that would slow
# the listing). Fetching the URL is where the 404 surfaces.
urls = user_msg.get("media_urls") or []
assert len(urls) == 1
fetched = await _http_get(f"http://127.0.0.1:29926{urls[0]['url']}")
assert fetched.status_code == 404
assert "media" not in user_msg
finally:
await channel.stop()
await server_task