diff options
| -rw-r--r-- | CLAUDE.md | 3 | ||||
| -rw-r--r-- | devel-phases-next.md | 93 | ||||
| -rw-r--r-- | packages/meshbay-common/src/meshbay_common/protocol.py | 4 | ||||
| -rw-r--r-- | packages/meshbay-hub/src/meshbay_hub/static/app.js | 177 | ||||
| -rw-r--r-- | packages/meshbay-hub/src/meshbay_hub/static/i18n.js | 2 | ||||
| -rw-r--r-- | packages/meshbay-hub/src/meshbay_hub/static/style.css | 25 | ||||
| -rw-r--r-- | packages/meshbay-hub/src/meshbay_hub/static/transport.js | 22 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 133 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_webrtc_transport.py | 62 |
9 files changed, 467 insertions, 54 deletions
@@ -185,6 +185,9 @@ SFR residential Fedora 44 → meshbay.org OVH VPS: | GEK wrap AES (Python) | `meshbay_common.crypto` | Phase 10b.2 — `wrap_gek_aes()` / `unwrap_gek_aes()` | | IndexedDB cache (browser) | `static/app.js` | Phase 10b.5 — group index caching | | Cross-group search (browser) | `static/app.js` | Phase 10b.6 — SearchPage, client-side | +| MSE video streaming (node) | `meshbay_node.transport.webrtc_server` | Phase 10c — ffmpeg fMP4 remux + encrypted segments | +| MSE video streaming (browser) | `static/app.js` | Phase 10c — MediaSource + SourceBuffer progressive playback | +| Video codec detection | `meshbay_node.transport.webrtc_server` | Phase 10c — `_probe_video()` ffprobe + MSE codec strings | | Demo scripts | — | `QE/demo-v1/*.py`, `QE/demo-v2/*.py`, `QE/demo-v3/*.py` (not versioned) | ## meshbay.org server (état cible) diff --git a/devel-phases-next.md b/devel-phases-next.md index 3e49f19..e423d83 100644 --- a/devel-phases-next.md +++ b/devel-phases-next.md @@ -1,6 +1,6 @@ # MeshBay — Next Implementation Phases -> Base: Phases 1–10b complete (except 10.9 → Phase 13). 166 tests. Web SPA + admin panel + self-service UI live on meshbay.org. +> Base: Phases 1–10c complete (except 10.9 → Phase 13). 167 tests. Web SPA + admin panel + self-service UI + MSE video streaming live on meshbay.org. > Architecture reference: docs/meshbay-draft-v4.md > First security review: first-review.md (2026-08-10) @@ -189,7 +189,7 @@ ICE/STUN handles all tested NAT types automatically. path-based dedup (remove old entry before adding new). **Known remaining items for future phases:** -- True video streaming (MSE or Service Worker) — currently downloads full file first +- ~~True video streaming (MSE or Service Worker)~~ → Phase 10c (2026-08-11) - Multiple shared directories per node (UI + config) - Multi-node per user support @@ -405,6 +405,95 @@ Results link back to the group page. Accessible from sidebar. --- +## Phase 10c — MSE video streaming (real-time playback) + +Pending commit — 167 tests. + +**Objective:** replace the download-then-play video player with real-time +MSE (MediaSource Extensions) streaming. Playback starts within seconds +instead of waiting for the full file download. + +### Architecture + +``` +Browser Node + │ │ + ├── stream_req {file_id} ──────►│ + │ ├── ffprobe → codec info + │◄──── stream_init {codec,dur} ──┤ + │ ├── ffmpeg -c copy → fMP4 pipe + │◄──── stream_data {seg 0, ct} ──┤ (256 KB encrypted segments) + │◄──── stream_data {seg 1, ct} ──┤ + │ ... │ + │◄──── stream_end ───────────────┤ + │ │ + MediaSource → SourceBuffer │ + ├── appendBuffer(decrypted) │ + ├── video.play() after ~2-3s │ +``` + +**Key design decisions:** + +1. **Node-side remux via ffmpeg** — `ffmpeg -c copy -movflags frag_keyframe+empty_moov+default_base_moof -f mp4 pipe:1` remuxes any video format (MP4, MKV, AVI, WebM, MOV) into fragmented MP4 (fMP4) that MSE can consume. No transcoding — just remuxing. Near-zero CPU overhead. + +2. **Codec detection via ffprobe** — the node probes the video to determine the exact codec string for MSE SourceBuffer creation (e.g., `avc1.640028,mp4a.40.2` for H.264 High@4.0 + AAC-LC). This ensures the browser creates the correct decoder. + +3. **Same encryption model** — each 256 KB fMP4 segment is encrypted with AES-256-GCM using the same key derivation as file downloads (GEK + file_hash + segment_index → HKDF → chunk_key). E2E encryption is maintained. + +4. **Progressive SourceBuffer append** — the browser creates a MediaSource, opens a SourceBuffer with the probed codec, and appends decrypted segments as they arrive. SourceBuffer handles partial MP4 boxes internally. Playback starts after ~2-3 segments (~512 KB buffered). + +### Supported codecs + +| Codec | MSE string | Browser support | +|---|---|---| +| H.264 (AVC) | `avc1.PPCCLL` | Chrome, Firefox, Safari, Edge | +| H.265 (HEVC) | `hev1.1.6.L93.B0` | Safari, Chrome (partial) | +| VP9 | `vp09.00.10.08` | Chrome, Firefox | +| AV1 | `av01.0.01M.08` | Chrome, Firefox | +| AAC | `mp4a.40.2` | All | +| MP3 | `mp4a.6b` | All | +| Opus | `opus` | Chrome, Firefox | +| AC-3 | `ac-3` | Safari, Chrome | + +### New MNP message types + +| Type | Direction | Description | +|---|---|---| +| `stream_req` | client → node | Request MSE video stream for file_id | +| `stream_init` | node → client | Codec string + duration (probed via ffprobe) | +| `stream_data` | node → client | Encrypted fMP4 segment (256 KB, AES-GCM) | +| `stream_end` | node → client | End of stream signal | + +### Milestones + +| # | Component | Status | +|---|---|---| +| 10c.1 | MNP protocol: STREAM_REQUEST/INIT/DATA/END message types | ✅ | +| 10c.2 | Node: ffprobe codec detection + MSE codec string derivation | ✅ | +| 10c.3 | Node: ffmpeg fMP4 remux + encrypted segment streaming | ✅ | +| 10c.4 | Transport: event-based stream message dispatch | ✅ | +| 10c.5 | Browser: MSE VideoPlayer (MediaSource + SourceBuffer) | ✅ | +| 10c.6 | Tests: stream_request error handling | ✅ | + +### File changes + +**Modified:** +- `packages/meshbay-common/src/meshbay_common/protocol.py` — STREAM_REQUEST/INIT/DATA/END +- `packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py` — `_probe_video()`, `_stream_video()` handler +- `packages/meshbay-hub/src/meshbay_hub/static/transport.js` — `requestStream()`, stream event handlers +- `packages/meshbay-hub/src/meshbay_hub/static/app.js` — MSE-based VideoPlayer component +- `packages/meshbay-hub/src/meshbay_hub/static/style.css` — streaming progress bar +- `packages/meshbay-hub/src/meshbay_hub/static/i18n.js` — buffering/MSE error strings +- `packages/meshbay-node/tests/test_webrtc_transport.py` — stream_request error test + +### Known limitations (future work) + +- No seeking beyond buffered range (user must wait for data to arrive) +- No adaptive bitrate (single quality stream) +- Requires ffmpeg/ffprobe on the node (already a dependency for the old STREAM_SEGMENT handler) + +--- + ## Phase 11 — Android client MVP **Objective:** Android app for account creation, group browsing, file download, diff --git a/packages/meshbay-common/src/meshbay_common/protocol.py b/packages/meshbay-common/src/meshbay_common/protocol.py index 4095215..bf83906 100644 --- a/packages/meshbay-common/src/meshbay_common/protocol.py +++ b/packages/meshbay-common/src/meshbay_common/protocol.py @@ -35,6 +35,10 @@ class MNP: FILE_UPLOAD_ACK = "file_upload_ack" # node acknowledges chunk receipt FILE_DELETE = "file_delete" # client requests file deletion FILE_DELETE_ACK = "file_delete_ack" # node confirms deletion + STREAM_REQUEST = "stream_req" # client requests MSE video stream + STREAM_INIT = "stream_init" # node sends codec info + signals stream start + STREAM_DATA = "stream_data" # node sends encrypted fMP4 segment + STREAM_END = "stream_end" # node signals end of stream EPHEMERAL_STREAM = "ephemeral_stream" # reserved — mobile live push diff --git a/packages/meshbay-hub/src/meshbay_hub/static/app.js b/packages/meshbay-hub/src/meshbay_hub/static/app.js index 34c2517..cb3e05b 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/app.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/app.js @@ -1468,72 +1468,140 @@ function ChatPanel({ transportRef, username, entries, gekRef, onRefreshIndex }) `; } -// ── Video Player ──────────────────────────────────────────────────────── +// ── Video Player (MSE streaming) ──────────────────────────────────────── -const VIDEO_MIMES = { - '.mp4': 'video/mp4', '.webm': 'video/webm', '.mkv': 'video/x-matroska', - '.avi': 'video/x-msvideo', '.mov': 'video/quicktime', '.m4v': 'video/mp4', - '.flv': 'video/x-flv', '.wmv': 'video/x-ms-wmv', -}; - -function videoMime(name) { - const dot = name.lastIndexOf('.'); - if (dot < 0) return 'video/mp4'; - return VIDEO_MIMES[name.slice(dot).toLowerCase()] || 'video/mp4'; +function _mseSupported(codec) { + if (!window.MediaSource) return false; + const mime = `video/mp4; codecs="${codec}"`; + return MediaSource.isTypeSupported(mime); } function VideoPlayer({ entry, transportRef, gekRef, onClose }) { const [phase, setPhase] = useState('loading'); const [progress, setProgress] = useState(0); const [error, setError] = useState(''); + const [buffered, setBuffered] = useState(0); const videoRef = useRef(null); + const msRef = useRef(null); + const sbRef = useRef(null); const blobUrlRef = useRef(null); + const queueRef = useRef([]); + const appendingRef = useRef(false); + const endedRef = useRef(false); + + const flushQueue = useCallback(() => { + const sb = sbRef.current; + if (!sb || appendingRef.current || sb.updating) return; + if (queueRef.current.length === 0) { + if (endedRef.current && msRef.current?.readyState === 'open') { + try { msRef.current.endOfStream(); } catch {} + } + return; + } + appendingRef.current = true; + const chunk = queueRef.current.shift(); + try { + sb.appendBuffer(chunk); + } catch (e) { + appendingRef.current = false; + console.error('[MSE] appendBuffer error:', e); + } + }, []); useEffect(() => { let cancelled = false; + const transport = transportRef.current; + if (!transport || !transport.connected) { + setError(t('video.err_transport')); + setPhase('error'); + return; + } - const load = async () => { - const transport = transportRef.current; - if (!transport || !transport.connected) { - setError(t('video.err_transport')); - setPhase('error'); - return; + const startStream = async () => { + if (!gekRef.current && window.MeshBayCrypto) { + const gekB64 = await transport.fetchGEK(); + gekRef.current = await window.MeshBayCrypto.importGEK(gekB64); } - try { - if (!gekRef.current && window.MeshBayCrypto) { - const gekB64 = await transport.fetchGEK(); - gekRef.current = await window.MeshBayCrypto.importGEK(gekB64); - } - - const totalChunks = Math.ceil(entry.size / CHUNK_SIZE); - let downloaded = 0; - const chunks = await pipelinedDownload( - transport, gekRef.current, entry.id, totalChunks, - (bytes) => { downloaded += bytes; setProgress(downloaded / entry.size); }, - ); + let streamCodec = null; + let totalBytes = 0; + transport.onStreamInit = (msg) => { if (cancelled) return; + streamCodec = msg.codec; + const mime = `video/mp4; codecs="${streamCodec}"`; - const blob = new Blob(chunks, { type: videoMime(entry.name) }); - const url = URL.createObjectURL(blob); - blobUrlRef.current = url; - setPhase('ready'); - } catch (err) { - if (!cancelled) { - setError(err.message); + if (!window.MediaSource || !MediaSource.isTypeSupported(mime)) { + setError(t('video.err_mse', { codec: streamCodec })); setPhase('error'); + return; } - } + + const ms = new MediaSource(); + msRef.current = ms; + const url = URL.createObjectURL(ms); + blobUrlRef.current = url; + + ms.addEventListener('sourceopen', () => { + if (cancelled) return; + const sb = ms.addSourceBuffer(mime); + sbRef.current = sb; + sb.addEventListener('updateend', () => { + appendingRef.current = false; + if (videoRef.current) { + setBuffered(videoRef.current.buffered.length > 0 + ? videoRef.current.buffered.end(0) : 0); + } + flushQueue(); + }); + setPhase('streaming'); + flushQueue(); + }); + + if (videoRef.current) { + videoRef.current.src = url; + } + }; + + transport.onStreamData = async (msg) => { + if (cancelled) return; + try { + const plaintext = await window.MeshBayCrypto.decryptChunkBin( + gekRef.current, entry.id, msg.segment_index, msg.nonce, msg.ct); + totalBytes += plaintext.byteLength; + setProgress(entry.size > 0 ? totalBytes / entry.size : 0); + queueRef.current.push(plaintext); + flushQueue(); + } catch (e) { + console.error('[MSE] decrypt error:', e); + } + }; + + transport.onStreamEnd = () => { + if (cancelled) return; + endedRef.current = true; + flushQueue(); + }; + + transport.requestStream(entry.id); }; - load(); - return () => { cancelled = true; }; - }, [entry]); + startStream().catch(err => { + if (!cancelled) { setError(err.message); setPhase('error'); } + }); + + return () => { + cancelled = true; + if (transport) { + transport.onStreamInit = null; + transport.onStreamData = null; + transport.onStreamEnd = null; + } + }; + }, [entry, flushQueue]); useEffect(() => { - if (phase === 'ready' && videoRef.current && blobUrlRef.current) { - videoRef.current.src = blobUrlRef.current; + if (phase === 'streaming' && videoRef.current) { videoRef.current.play().catch(() => {}); } }, [phase]); @@ -1558,29 +1626,34 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose }) { if (e.target.classList.contains('video-overlay')) onClose(); }}> <div class="video-top-bar"> - <span class="video-title">${entry.name}</span> + <span class="video-title">${entry.name} (${formatSize(entry.size)})</span> <button class="video-close" onClick=${onClose} title="${t('video.close')}">✕</button> </div> ${phase === 'loading' && html` <div class="video-loading"> - <div class="video-loading-label">${t('video.loading', { name: entry.name })}</div> - <div class="video-progress-bar"> - <div class="video-progress-fill" - style="width:${Math.round(progress * 100)}%"></div> - </div> - <div class="video-progress-text"> - ${formatSize(Math.round(progress * entry.size))} / ${formatSize(entry.size)} - </div> + <div class="video-loading-label">${t('video.buffering')}</div> </div> `} - ${phase === 'ready' && html` + ${(phase === 'streaming' || phase === 'loading') && html` <div class="video-container"> <video ref=${videoRef} controls autoplay /> </div> `} + ${phase === 'streaming' && progress < 1 && html` + <div class="video-stream-bar"> + <div class="video-progress-bar small"> + <div class="video-progress-fill" + style="width:${Math.round(progress * 100)}%"></div> + </div> + <span class="video-stream-label"> + ${formatSize(Math.round(progress * entry.size))} / ${formatSize(entry.size)} + </span> + </div> + `} + ${phase === 'error' && html` <div class="video-error">${error}</div> `} diff --git a/packages/meshbay-hub/src/meshbay_hub/static/i18n.js b/packages/meshbay-hub/src/meshbay_hub/static/i18n.js index 8ea4600..2a91407 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/i18n.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/i18n.js @@ -108,8 +108,10 @@ const en = { // Video player 'video.loading': 'Loading {name}...', + 'video.buffering': 'Buffering...', 'video.close': 'Close (Esc)', 'video.err_transport': 'Transport not connected', + 'video.err_mse': 'Codec not supported for streaming: {codec}', // Settings 'settings.title': 'Settings', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/style.css b/packages/meshbay-hub/src/meshbay_hub/static/style.css index 542c672..faf8b0f 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/style.css +++ b/packages/meshbay-hub/src/meshbay_hub/static/style.css @@ -909,6 +909,31 @@ button:disabled { opacity: 0.5; cursor: not-allowed; } max-width: 400px; } +.video-stream-bar { + position: absolute; + bottom: 60px; + left: 50%; + transform: translateX(-50%); + display: flex; + align-items: center; + gap: 8px; + background: rgba(0, 0, 0, 0.6); + padding: 4px 12px; + border-radius: 12px; + z-index: 10; +} + +.video-progress-bar.small { + width: 120px; + height: 4px; +} + +.video-stream-label { + font-size: 0.7em; + color: rgba(255, 255, 255, 0.7); + white-space: nowrap; +} + .play-btn { background: none; border: 1px solid var(--border); diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js index 011c33e..1dc1ded 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js @@ -27,11 +27,17 @@ class MeshBayTransport { this._recvBuf = new Uint8Array(0); this._connected = false; this._onChat = null; + this._onStreamInit = null; + this._onStreamData = null; + this._onStreamEnd = null; } get connected() { return this._connected; } set onChat(fn) { this._onChat = fn; } + set onStreamInit(fn) { this._onStreamInit = fn; } + set onStreamData(fn) { this._onStreamData = fn; } + set onStreamEnd(fn) { this._onStreamEnd = fn; } async connect(nodeId, jwtToken, groupId) { this._pc = new RTCPeerConnection({ @@ -186,6 +192,10 @@ class MeshBayTransport { return msg; } + requestStream(fileId) { + this._send({ type: 'stream_req', v: '0.1', file_id: fileId }); + } + async uploadChunk(filename, chunkIndex, totalChunks, data) { const msg = await this._sendAndWait({ type: 'file_upload', @@ -259,6 +269,18 @@ class MeshBayTransport { this._onChat(msg); return; } + if (msg.type === 'stream_init' && this._onStreamInit) { + this._onStreamInit(msg); + return; + } + if (msg.type === 'stream_data' && this._onStreamData) { + this._onStreamData(msg); + return; + } + if (msg.type === 'stream_end' && this._onStreamEnd) { + this._onStreamEnd(msg); + return; + } const oldest = this._pending.entries().next(); if (!oldest.done) { diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index 1d10aa8..ab95550 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -46,6 +46,58 @@ CHUNK_SIZE = 1024 * 1024 MAX_MSG = 64 * 1024 * 1024 +STREAM_SEGMENT_SIZE = 256 * 1024 + +_H264_PROFILES = {"Baseline": "42", "Main": "4d", "High": "64", "High 10": "6e"} + + +async def _probe_video(path: str) -> tuple[str | None, float]: + """Probe video file with ffprobe, return (MSE codec string, duration).""" + import json as _json + proc = await asyncio.create_subprocess_exec( + "ffprobe", "-v", "error", + "-show_entries", "stream=codec_name,profile,level,codec_type", + "-show_entries", "format=duration", + "-of", "json", path, + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + ) + stdout, _ = await proc.communicate() + info = _json.loads(stdout) + duration = float(info.get("format", {}).get("duration", 0)) + + v_codec = a_codec = "" + for s in info.get("streams", []): + if s.get("codec_type") == "video" and not v_codec: + cn = s.get("codec_name", "") + if cn == "h264": + p = _H264_PROFILES.get(s.get("profile", "High"), "64") + lvl = int(s.get("level", 40)) + v_codec = f"avc1.{p}00{lvl:02x}" + elif cn == "hevc": + v_codec = "hev1.1.6.L93.B0" + elif cn == "vp9": + v_codec = "vp09.00.10.08" + elif cn == "av1": + v_codec = "av01.0.01M.08" + elif s.get("codec_type") == "audio" and not a_codec: + cn = s.get("codec_name", "") + if cn == "aac": + a_codec = "mp4a.40.2" + elif cn in ("mp3", "mp2"): + a_codec = "mp4a.6b" + elif cn == "opus": + a_codec = "opus" + elif cn == "ac3": + a_codec = "ac-3" + elif cn == "flac": + a_codec = "flac" + + if not v_codec: + return None, duration + codec = f"{v_codec},{a_codec}" if a_codec else v_codec + return codec, duration + + def _pack(obj: dict) -> bytes: data = msgpack.packb(obj, use_bin_type=True) return struct.pack(">I", len(data)) + data @@ -118,6 +170,8 @@ class WebRTCPeerSession: self._do_file_upload(msg) elif mtype == MNP.FILE_DELETE: self._do_file_delete(msg) + elif mtype == MNP.STREAM_REQUEST: + asyncio.ensure_future(self._stream_video(msg)) else: log.warning("Unknown MNP message type on DataChannel: %s", mtype) except Exception as e: @@ -417,6 +471,85 @@ class WebRTCPeerSession: "file_id": file_id, }) + async def _stream_video(self, msg: dict) -> None: + """Stream a video file as fMP4 segments via MSE-compatible output.""" + ctx = self._group_ctx() + file_id = msg.get("file_id", "") + entry = ctx["index"].get_entry(file_id) + if not entry: + self._send({"type": "error", "detail": "File not found"}) + return + + file_path = ctx["shared_root"] / entry.path / entry.name + if not file_path.exists(): + self._send({"type": "error", "detail": "File not on disk"}) + return + + gek = ctx.get("gek") + file_hash = bytes.fromhex(entry.id) + + try: + codec_str, duration = await _probe_video(str(file_path)) + except Exception as e: + self._send({"type": "error", "detail": f"Probe failed: {e}"}) + return + + if not codec_str: + self._send({"type": "error", "detail": "Unsupported video codec"}) + return + + proc = await asyncio.create_subprocess_exec( + "ffmpeg", "-hide_banner", "-loglevel", "error", + "-i", str(file_path), + "-c", "copy", + "-movflags", "frag_keyframe+empty_moov+default_base_moof", + "-f", "mp4", "pipe:1", + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + ) + + self._send({ + "type": MNP.STREAM_INIT, + "v": MNP_VERSION, + "file_id": file_id, + "codec": codec_str, + "duration": duration, + }) + + index = 0 + try: + while True: + data = await proc.stdout.read(STREAM_SEGMENT_SIZE) + if not data: + break + ckey = chunk_key_aes(gek, file_hash, index) + nonce, ct = encrypt_chunk_aes(ckey, data) + self._send({ + "type": MNP.STREAM_DATA, + "v": MNP_VERSION, + "file_id": file_id, + "segment_index": index, + "nonce": nonce, + "ct": ct, + "plaintext_size": len(data), + }) + index += 1 + await asyncio.sleep(0) + except Exception as e: + log.error("Stream error: %s", e) + finally: + try: + proc.kill() + except ProcessLookupError: + pass + await proc.wait() + + self._send({ + "type": MNP.STREAM_END, + "v": MNP_VERSION, + "file_id": file_id, + }) + log.info("Streamed %s: %d segments", entry.name, index) + def _send(self, obj: dict) -> None: if self._channel and self._channel.readyState == "open": self._channel.send(_pack(obj)) diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py index d3847ff..b6664e8 100644 --- a/packages/meshbay-node/tests/test_webrtc_transport.py +++ b/packages/meshbay-node/tests/test_webrtc_transport.py @@ -725,3 +725,65 @@ async def test_webrtc_stream_segment_missing_file(sk_node, sk_hub, gek, shared_d await browser_pc.close() await transport.close_all() + + +@pytest.mark.asyncio +async def test_webrtc_stream_request_missing_file(sk_node, sk_hub, gek, shared_dir): + """WebRTC DataChannel: stream_request for non-existent file returns error.""" + hub_pk_pem = _hub_pk_pem(sk_hub) + indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + await indexer.initial_scan() + + transport = WebRTCTransport( + sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, + shared_root=shared_dir, index=indexer.index, + stun_servers=[], + ) + + browser_pc = RTCPeerConnection() + received = asyncio.Queue() + buf = bytearray() + + channel = browser_pc.createDataChannel("mnp") + + @channel.on("open") + def on_open(): + channel.send(_pack({ + "type": MNP.HANDSHAKE, "v": MNP_VERSION, + "token": _make_jwt(sk_hub), + })) + + @channel.on("message") + def on_msg(message): + if isinstance(message, str): + message = message.encode() + buf.extend(message) + while len(buf) >= 4: + length = struct.unpack(">I", buf[:4])[0] + if len(buf) < 4 + length: + break + msg_bytes = bytes(buf[4:4 + length]) + del buf[:4 + length] + received.put_nowait(msgpack.unpackb(msg_bytes, raw=False)) + + offer = await browser_pc.createOffer() + await browser_pc.setLocalDescription(offer) + answer_sdp, _ = await transport.handle_offer( + browser_pc.localDescription.sdp, "peer-mse") + await browser_pc.setRemoteDescription( + RTCSessionDescription(sdp=answer_sdp, type="answer")) + + ack = await asyncio.wait_for(received.get(), timeout=5.0) + assert ack["type"] == MNP.HANDSHAKE_ACK + + channel.send(_pack({ + "type": MNP.STREAM_REQUEST, "v": MNP_VERSION, + "file_id": "nonexistent-file-id", + })) + + msg = await asyncio.wait_for(received.get(), timeout=5.0) + assert msg["type"] == "error" + assert "not found" in msg["detail"].lower() + + await browser_pc.close() + await transport.close_all() |