summaryrefslogtreecommitdiffstats
path: root/packages/meshbay-node/src/meshbay_node
diff options
context:
space:
mode:
authorChristophe Besson <cbesson@gmail.com>2026-09-07 17:45:54 +0200
committerChristophe Besson <cbesson@gmail.com>2026-09-07 17:45:54 +0200
commit77dd077491aea50e71e21e0d17555a2f91cf818b (patch)
tree87b88674b3210541f413b4992750dcb2807b8f12 /packages/meshbay-node/src/meshbay_node
parentd1f998b42137465b610667439527917a00030b4d (diff)
downloadmeshbay-77dd077491aea50e71e21e0d17555a2f91cf818b.tar.gz
refactor(mnp)!: remove stream_seg, the last unencrypted content message
`stream_seg` answered with an MPEG-TS segment as base64 with no encryption at all — the one message on the content plane that never went through a GEK-derived key. Live on both transports, answering any authenticated member. It predates `stream_data`, which does the same job properly (`chunk_ciphertext`, keyed per segment, AES-256-GCM) and has since Phase 12. Its only browser caller, `fetchStreamSegment`, was defined and never once invoked — a plaintext media endpoint with no client. Removed rather than repaired. Gone with it: `_extract_segment` and the ffmpeg semaphore in quic_server, the `fetch_stream_segment` QUIC client method, and `_b64decode` in transport.js, which had no other caller. The H6 regression test lived on this handler — it pinned `_do_stream_segment_async` to a coroutine so `subprocess.run` could not stall the event loop for thirty seconds per request. It is replaced by the property that outlives the handler: no transport carries media outside an AEAD, asserted on `stream_seg` and `data_b64` across all three transport modules. The half of H6 that survives — the live streaming path still spawns ffmpeg — keeps its own test. BREAKING CHANGE: `stream_seg` is no longer answered on either transport. No shipping client sends it. Recorded as part of MNP 2.0, whose other half — the sealed upload — carries the version bump. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01AsoWC3GmhNdwVFomW3QjH3
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node')
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/quic_client.py19
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/quic_server.py76
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py67
3 files changed, 0 insertions, 162 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_client.py b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py
index b22b8df..af87b70 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/quic_client.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py
@@ -317,22 +317,3 @@ class QuicChunkClient:
# substitutes it would otherwise choose which key we decrypt with.
return file_chunk_plaintext(
self._gek, msg, file_hash=bytes.fromhex(file_id))
-
- async def fetch_stream_segment(
- self, file_id: str, segment_index: int, segment_duration: int = 4,
- ) -> bytes:
- """Fetch one HLS segment (MPEG-TS bytes) over QUIC."""
- sid = self._new_stream()
- self._proto._send(sid, {
- "type": MNP.STREAM_SEGMENT,
- "v": MNP_VERSION,
- "file_id": file_id,
- "segment_index": segment_index,
- "segment_duration": segment_duration,
- })
- msg = await self._proto._recv(sid, timeout=30.0)
-
- if msg.get("type") == "error":
- raise LookupError(msg.get("detail", "Unknown error"))
-
- return base64.b64decode(msg["data_b64"])
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 360b9ac..29dfef2 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
from pathlib import Path
from typing import Any, Callable
@@ -53,7 +52,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__)
@@ -61,16 +59,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.
@@ -243,8 +231,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:
@@ -431,49 +417,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.
@@ -542,25 +485,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:
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 aaf3f81..f8b6c03 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -479,8 +479,6 @@ class WebRTCPeerSession:
# Chunks are matched by file and index on the client, so
# answering out of order is safe.
self._spawn(self._do_file_request(msg))
- elif mtype == MNP.STREAM_SEGMENT:
- self._do_stream_segment(msg)
elif mtype == MNP.CHAT_MESSAGE:
self._do_chat_message(msg)
elif mtype == MNP.CHAT_HISTORY:
@@ -3801,71 +3799,6 @@ class WebRTCPeerSession:
"director": director,
}
- def _do_stream_segment(self, msg: dict) -> None:
- self._spawn(self._do_stream_segment_async(msg))
-
- async def _do_stream_segment_async(self, msg: dict) -> None:
- """
- Legacy HLS segment extraction (superseded by stream_req/MSE).
-
- Finding H6: this ran subprocess.run(..., timeout=30) directly inside the
- event loop, so a single request stalled the whole daemon — every peer,
- every group — for up to thirty seconds. Now async and under the same
- transcode semaphore as _stream_video.
- """
- 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({"type": "error", "detail": "File not found"})
- return
-
- file_path = entry_abs_path(ctx["roots"], entry)
- if not file_path.exists():
- self._send({"type": "error", "detail": "File not on disk"})
- return
-
- sem = self._transcode_semaphore()
-
- try:
- async with sem:
- proc = await asyncio.create_subprocess_exec(
- platform.ffmpeg_cmd(), "-hide_banner", "-loglevel", "error",
- "-ss", str(segment_index * segment_duration),
- "-i", str(file_path),
- "-t", str(segment_duration),
- "-c:v", "copy", "-c:a", "copy",
- "-f", "mpegts", "pipe:1",
- stdout=asyncio.subprocess.PIPE,
- stderr=asyncio.subprocess.DEVNULL,
- )
- try:
- stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=30)
- except asyncio.TimeoutError:
- proc.kill()
- await proc.wait()
- self._send({"type": "error", "detail": "Segment extraction timed out"})
- return
- if proc.returncode != 0 or not stdout:
- self._send({"type": "error", "detail": "Segment extraction failed"})
- return
- segment_data = stdout
- except Exception:
- self._send({"type": "error", "detail": "Segment extraction failed"})
- return
-
- self._send({
- "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),
- })
-
def _do_chat_message(self, msg: dict) -> None:
# Per-group store — see _peer_registry() and finding H1. Reading chat_store
# off the shared transport context sent every group's messages to the first