diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport/quic_server.py')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/quic_server.py | 76 |
1 files changed, 0 insertions, 76 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py index 284b488..e34153c 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py @@ -22,7 +22,6 @@ import base64 import logging import os import struct -import subprocess import uuid from pathlib import Path from typing import Any, Callable @@ -54,7 +53,6 @@ from meshbay_common.groupbox import PURPOSE_ACK, seal from meshbay_common.protocol import MNP, file_chunk_wire from meshbay_node.indexer import GroupIndex from meshbay_node.transport.wire import index_sync_message -from meshbay_node import platform log = logging.getLogger(__name__) @@ -62,16 +60,6 @@ CHUNK_SIZE = 1024 * 1024 MAX_MSG = 64 * 1024 * 1024 ALPN = ["meshbay-mnp"] -# ffmpeg is spawned per STREAM_SEGMENT request, and `_extract_segment` runs -# `subprocess.run` synchronously — so without a bound, an authenticated peer can -# both fork-bomb the node and block its event loop for up to 30 s per request -# (finding M2c). Extraction now runs in a thread and passes through this -# semaphore. Small on purpose: the QUIC path has no shipping client yet, this is -# parity work with the WebRTC transcode cap. -_MAX_CONCURRENT_SEGMENTS = 4 -_segment_sem = asyncio.Semaphore(_MAX_CONCURRENT_SEGMENTS) - - class Denylist: """ Denylist for revoked users, groups and invalidated JWTs. @@ -247,8 +235,6 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._do_index_sync_sync(stream_id) elif mtype == MNP.FILE_REQUEST: self._do_file_request_sync(stream_id, msg) - elif mtype == MNP.STREAM_SEGMENT: - self._spawn(self._do_stream_segment(stream_id, msg)) elif mtype == MNP.CHAT_MESSAGE: self._do_chat_message_sync(stream_id, msg) elif mtype == MNP.PING: @@ -435,49 +421,6 @@ class _MNPServerProtocol(QuicConnectionProtocol): ctx["gek"], file_path, chunk_index, file_hash, entry.id) self._send(stream_id, chunk_data) - async def _do_stream_segment(self, stream_id: int, msg: dict) -> None: - """ - Extract and serve one segment via ffmpeg — off the event loop and behind - a concurrency bound, so one request can neither stall the whole node nor - fork-bomb it (finding M2c). The WebRTC path has had both since Phase 11.5. - """ - try: - ctx = self._group_ctx() - file_id = msg["file_id"] - segment_index = msg["segment_index"] - segment_duration = msg.get("segment_duration", 4) - - entry = ctx["index"].get_entry(file_id) - if not entry: - self._send(stream_id, {"type": "error", "detail": "File not found"}) - return - - file_path = entry_abs_path(ctx["roots"], entry) - if not file_path.exists(): - self._send(stream_id, {"type": "error", "detail": "File not on disk"}) - return - - start_time = segment_index * segment_duration - loop = asyncio.get_event_loop() - async with _segment_sem: - segment_data = await loop.run_in_executor( - None, _extract_segment, file_path, start_time, segment_duration) - if segment_data is None: - self._send(stream_id, {"type": "error", "detail": "Segment extraction failed"}) - return - - self._send(stream_id, { - "type": MNP.STREAM_SEGMENT, - "v": MNP_VERSION, - "file_id": file_id, - "segment_index": segment_index, - "data_b64": base64.b64encode(segment_data).decode(), - "size": len(segment_data), - }) - except Exception as e: - log.error("stream_segment: %s", e) - self._send(stream_id, {"type": "error", "detail": "Segment extraction failed"}) - def _do_chat_message_sync(self, stream_id: int, msg: dict) -> None: """ Store a chat message and broadcast it to the rest of THIS group. @@ -548,25 +491,6 @@ def _read_and_encrypt( return file_chunk_wire(gek, plaintext, chunk_index, file_hash, file_id) -def _extract_segment(file_path: Path, start_time: float, duration: float) -> bytes | None: - """Extract one HLS segment via ffmpeg. Returns MPEG-TS bytes or None on failure.""" - try: - result = subprocess.run( - [platform.ffmpeg_cmd(), "-hide_banner", "-loglevel", "error", - "-ss", str(start_time), - "-i", str(file_path), - "-t", str(duration), - "-c:v", "copy", "-c:a", "copy", - "-f", "mpegts", "pipe:1"], - capture_output=True, timeout=30, - ) - if result.returncode == 0 and result.stdout: - return result.stdout - return None - except Exception: - return None - - # ── QuicChunkServer ──────────────────────────────────────────────────────────── class QuicChunkServer: |