diff options
Diffstat (limited to 'packages/meshbay-node')
| -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 |
2 files changed, 195 insertions, 0 deletions
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() |