From a0be6bfcac07a256949082e935e1be06753ab872 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Thu, 15 Oct 2026 10:17:29 +0200 Subject: feat(cli): stream — a film or a song played in a media player MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `meshbay-cli stream PATH` serves one video or sound file over HTTP on 127.0.0.1 for VLC, mpv or any player that reads HTTP; `--player PROGRAM` starts the player, and closing it stops the stream, as Ctrl+C does. The file goes as the node stores it: no remux, no transcode, no stream slot on the node. Byte ranges are answered from leaseless file_req chunks decrypted in memory, eight ahead and the last 24 kept. Loopback only, a random token in the address, a foreign Host refused (DNS rebinding), an audio or video type with nosniff. The connection is pinged and reopened when it dies; a refusal by the node stops the stream (exit 5). Co-Authored-By: Claude Opus 5.5 --- .../meshbay-cli/src/meshbay_cli/client/relay.py | 365 +++++++++++++++++++++ .../src/meshbay_cli/client/transfers.py | 14 + 2 files changed, 379 insertions(+) create mode 100644 packages/meshbay-cli/src/meshbay_cli/client/relay.py (limited to 'packages/meshbay-cli/src/meshbay_cli/client') diff --git a/packages/meshbay-cli/src/meshbay_cli/client/relay.py b/packages/meshbay-cli/src/meshbay_cli/client/relay.py new file mode 100644 index 0000000..931913d --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/client/relay.py @@ -0,0 +1,365 @@ +""" +`stream`: one file of a group served to a media player over HTTP on this machine. + +The player asks for byte ranges; each is answered from the node's 1 MiB chunks, +fetched and decrypted here and never written to disk. A few chunks are asked +for ahead of what the player has read, and the last few are kept, so a player +probing the end of a file and coming back costs no second fetch. The node sends +the file as it is — no remux, no transcode: the player decodes it. + +Decrypted content on a local port is what has to be guarded. The server listens +on 127.0.0.1 only; the address carries a random token, so another program or +another account of this machine cannot find the stream; and a request whose +`Host` is not that address is refused, so a web page that rebinds a name of +its own to 127.0.0.1 cannot read it. A file is served under an audio or video +type only, with `nosniff`, never under a type a browser would run. + +The connection to the node is watched — a ping every half minute, as +`chat --follow` does — and opened again when it dies; the player only sees a +slower read. +""" + +import asyncio +import contextlib +import hmac +import logging +import mimetypes +import posixpath +import secrets +from collections import OrderedDict +from collections.abc import Awaitable, Callable +from urllib.parse import quote, unquote, urlsplit + +from ..errors import NODE, CliError +from .session import GroupSession, NodeRefusal +from .transfers import DOWNLOAD_CHUNK, read_chunk + +log = logging.getLogger(__name__) + +AHEAD = 8 # chunks asked for ahead of the one a player reads +KEPT = 24 # chunks kept once read +PING_EVERY, PING_TIMEOUT = 30, 10 +RECONNECT_WAITS = (2, 5, 10, 30) +# A chunk that fails this many times on connections that opened is not a link +# going down: the stream stops rather than reconnect in a loop. +CHUNK_ATTEMPTS = 5 +HEADER_LIMIT = 16 * 1024 +IDLE_TIMEOUT = 120 # a kept-alive connection that says nothing is closed + +# What `mimetypes` does not know everywhere, or knows differently on Windows. +TYPES = { + ".mkv": "video/x-matroska", ".mk3d": "video/x-matroska", ".webm": "video/webm", + ".mp4": "video/mp4", ".m4v": "video/mp4", ".mov": "video/quicktime", + ".avi": "video/x-msvideo", ".ts": "video/mp2t", ".m2ts": "video/mp2t", + ".mpg": "video/mpeg", ".mpeg": "video/mpeg", ".wmv": "video/x-ms-wmv", + ".flv": "video/x-flv", ".ogv": "video/ogg", ".3gp": "video/3gpp", + ".mp3": "audio/mpeg", ".flac": "audio/flac", ".ogg": "audio/ogg", + ".oga": "audio/ogg", ".opus": "audio/ogg", ".m4a": "audio/mp4", ".aac": "audio/aac", + ".wav": "audio/wav", ".wma": "audio/x-ms-wma", ".aiff": "audio/aiff", + ".alac": "audio/mp4", ".mka": "audio/x-matroska", +} + + +def content_type(name: str, kind: str) -> str: + ext = posixpath.splitext(name.lower())[1] + guessed = TYPES.get(ext) or mimetypes.guess_type("x" + ext)[0] or "" + if guessed.startswith(("audio/", "video/")): + return guessed + return "audio/octet-stream" if kind == "audio" else "video/octet-stream" + + +def byte_range(header: str | None, size: int) -> tuple[int, int] | None: + """ + The one range `header` asks for, both ends included — None for the whole + file. Raises ValueError for a range that cannot be satisfied. Several ranges + at once are answered with the whole file, which HTTP allows and no player + asks for. + """ + if not header or not header.strip().lower().startswith("bytes="): + return None + spec = header.split("=", 1)[1].strip() + if "," in spec: + return None + first, sep, last = spec.partition("-") + if not sep: + return None + first, last = first.strip(), last.strip() + try: + if not first: # the last N bytes + n = int(last) + if n <= 0 or size == 0: + raise ValueError + return max(0, size - n), size - 1 + start = int(first) + end = int(last) if last else size - 1 + except ValueError: + raise ValueError(header) from None + if start < 0 or start >= size or end < start: + raise ValueError(header) + return start, min(end, size - 1) + + +class RelayStopped(CliError): + """The node refused part of the file: the stream cannot go on.""" + + +class ChunkSource: + """ + One file's chunks, from one connection to its group's node at a time. + + `opener` opens another connection when this one dies. A refusal by the node + is not retried: it stops the stream (`stopped`). + """ + + def __init__(self, session: GroupSession, entry: dict, + opener: Callable[[], Awaitable[GroupSession]], + note: Callable[[str], None] = lambda _: None): + self.session: GroupSession | None = session + self.entry = entry + self.opener = opener + self.note = note + self.chunks = max(1, -(-int(entry["size"]) // DOWNLOAD_CHUNK)) + self.stopped: asyncio.Future = asyncio.get_running_loop().create_future() + self._kept: OrderedDict[int, bytes] = OrderedDict() + self._inflight: dict[int, asyncio.Task] = {} + self._reconnecting: asyncio.Task | None = None + self._watch = asyncio.ensure_future(self._watchdog()) + + async def chunk(self, i: int, *, ahead_to: int | None = None) -> bytes: + """Chunk `i`, asking for the next ones up to `ahead_to` (included) too.""" + last = min(self.chunks - 1, i + AHEAD, self.chunks - 1 if ahead_to is None + else ahead_to) + for j in range(i, last + 1): + if j not in self._kept and j not in self._inflight: + self._inflight[j] = task = asyncio.ensure_future(self._fetch(j)) + task.add_done_callback(lambda t, j=j: self._landed(j, t)) + if i in self._kept: + self._kept.move_to_end(i) + return self._kept[i] + return await asyncio.shield(self._inflight[i]) + + def _landed(self, i: int, task: asyncio.Task) -> None: + self._inflight.pop(i, None) + if task.cancelled() or task.exception() is not None: + return + self._kept[i] = task.result() + while len(self._kept) > KEPT: + self._kept.popitem(last=False) + + def _stop(self, error: CliError) -> RelayStopped: + if not isinstance(error, RelayStopped): + error = RelayStopped(error.code, error.message) + if not self.stopped.done(): + self.stopped.set_exception(error) + self.stopped.exception() # retrieved: whoever waits sees it + return error + + async def _fetch(self, i: int) -> bytes: + for _ in range(CHUNK_ATTEMPTS): + session = await self._live() + try: + return await read_chunk(session, self.entry, i) + except NodeRefusal as e: + raise self._stop(CliError(NODE, f"The node refused this file: {e.detail}")) \ + from e + except CliError as e: + failure = e + self._lost(session) + raise self._stop(CliError(NODE, f"Chunk {i} could not be read: {failure.message}")) + + async def _live(self) -> GroupSession: + while self.session is None: + if self._reconnecting is None: + self._reconnecting = asyncio.ensure_future(self._reconnect()) + await asyncio.shield(self._reconnecting) + return self.session + + def _lost(self, session: GroupSession) -> None: + """`session` failed; the next chunk asked for opens another.""" + if self.session is session: + self.session = None + self.note("meshbay-cli: the connection to the node was lost; reconnecting...") + asyncio.ensure_future(_quietly(session.close())) + + async def _reconnect(self) -> None: + attempt = 0 + try: + while True: + try: + self.session = await self.opener() + self.note("meshbay-cli: reconnected.") + return + except CliError as e: + if e.code != NODE: + raise self._stop(e) from e + await asyncio.sleep(RECONNECT_WAITS[min(attempt, len(RECONNECT_WAITS) - 1)]) + attempt += 1 + finally: + self._reconnecting = None + + async def _watchdog(self) -> None: + """A paused player asks for nothing; a dead link is found anyway, so the + next read does not wait a chunk's whole timeout to learn it.""" + while True: + try: + session = await self._live() + except CliError: + return # the stream stopped; `stopped` says why + closed = asyncio.ensure_future(session.ch.closed.wait()) + try: + await asyncio.wait_for(asyncio.shield(closed), PING_EVERY) + except TimeoutError: + try: + await session.ch.request({"type": "ping", "token": "stream"}, "pong", + timeout=PING_TIMEOUT) + continue + except CliError: + pass + finally: + closed.cancel() + self._lost(session) + + async def close(self) -> None: + self._watch.cancel() + for task in [*self._inflight.values(), self._reconnecting]: + if task is not None: + task.cancel() + if self.session is not None: + await _quietly(self.session.close()) + self.session = None + + +async def _quietly(aw) -> None: + with contextlib.suppress(Exception): + await aw + + +class Relay: + """The HTTP side: `start()` returns the address to give the player.""" + + def __init__(self, source: ChunkSource, name: str, kind: str): + self.source = source + self.name = name + self.size = int(source.entry["size"]) + self.type = content_type(name, kind) + self.token = secrets.token_urlsafe(24) + self.url = "" + self._hosts: set[str] = set() + self._server: asyncio.Server | None = None + self._clients: set[asyncio.Task] = set() + + async def start(self, port: int = 0) -> str: + self._server = await asyncio.start_server(self._client, "127.0.0.1", port) + port = self._server.sockets[0].getsockname()[1] + self._hosts = {f"127.0.0.1:{port}", f"localhost:{port}"} + self.url = f"http://127.0.0.1:{port}/{self.token}/{quote(self.name)}" + return self.url + + async def close(self, grace: float = 0) -> None: + """Stop listening; then, after `grace` seconds for the answers being + written — a refusal's 502 — close every connection.""" + if self._server is not None: + self._server.close() + if grace and self._clients: + await asyncio.wait(list(self._clients), timeout=grace) + for task in list(self._clients): + task.cancel() + if self._clients: + await asyncio.gather(*self._clients, return_exceptions=True) + + # ── One player connection ──────────────────────────────────────────────── + + async def _client(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter): + task = asyncio.current_task() + self._clients.add(task) + try: + while await self._one_request(reader, writer): + pass + except (ConnectionError, asyncio.IncompleteReadError, asyncio.LimitOverrunError, + TimeoutError, CliError): + pass # a stopped stream: `stopped` says why + finally: + self._clients.discard(task) + writer.close() + with contextlib.suppress(Exception): + await writer.wait_closed() + + async def _one_request(self, reader, writer) -> bool: + """Answer one request; False when the connection is to be closed.""" + head = await asyncio.wait_for(reader.readuntil(b"\r\n\r\n"), IDLE_TIMEOUT) + if len(head) > HEADER_LIMIT: + return False + lines = head.decode("latin-1").split("\r\n") + try: + method, target, version = lines[0].split(" ") + except ValueError: + await self._status(writer, 400, "Bad Request") + return False + headers = {} + for line in lines[1:]: + name, sep, value = line.partition(":") + if sep: + headers[name.strip().lower()] = value.strip() + keep = (version == "HTTP/1.1" and headers.get("connection", "").lower() != "close") + + # Anti-rebinding first: a foreign Host is refused before the token is + # even looked at. + if headers.get("host", "").lower() not in self._hosts: + await self._status(writer, 403, "Forbidden") + return False + path = unquote(urlsplit(target).path) + token = path.lstrip("/").split("/", 1)[0] + if not hmac.compare_digest(token.encode(), self.token.encode()): + await self._status(writer, 404, "Not Found") + return False + if method not in ("GET", "HEAD"): + await self._status(writer, 405, "Method Not Allowed", extra={"Allow": "GET, HEAD"}) + return False + + try: + wanted = byte_range(headers.get("range"), self.size) + except ValueError: + await self._status(writer, 416, "Range Not Satisfiable", + extra={"Content-Range": f"bytes */{self.size}"}) + return keep + start, end = wanted if wanted else (0, self.size - 1) + length = max(0, end - start + 1) + fields = {"Content-Type": self.type, "Content-Length": str(length), + "Accept-Ranges": "bytes"} + if wanted: + fields["Content-Range"] = f"bytes {start}-{end}/{self.size}" + if method == "HEAD" or length == 0: + self._head(writer, 206 if wanted else 200, fields, keep) + await writer.drain() + return keep + + first, last = start // DOWNLOAD_CHUNK, end // DOWNLOAD_CHUNK + try: + data = await self.source.chunk(first, ahead_to=last) + except CliError: + await self._status(writer, 502, "Bad Gateway") + raise + self._head(writer, 206 if wanted else 200, fields, keep) + for i in range(first, last + 1): + if i != first: + data = await self.source.chunk(i, ahead_to=last) + base = i * DOWNLOAD_CHUNK + writer.write(data[max(0, start - base):end + 1 - base]) + await writer.drain() + return keep + + def _head(self, writer, status: int, fields: dict, keep: bool) -> None: + reason = {200: "OK", 206: "Partial Content"}[status] + fields = {**fields, "Cache-Control": "no-store", "X-Content-Type-Options": "nosniff", + "Connection": "keep-alive" if keep else "close"} + writer.write((f"HTTP/1.1 {status} {reason}\r\n" + + "".join(f"{k}: {v}\r\n" for k, v in fields.items()) + + "\r\n").encode("latin-1")) + + async def _status(self, writer, status: int, reason: str, *, extra=None) -> None: + fields = {"Content-Length": "0", "Cache-Control": "no-store", **(extra or {})} + writer.write((f"HTTP/1.1 {status} {reason}\r\n" + + "".join(f"{k}: {v}\r\n" for k, v in fields.items()) + + "\r\n").encode("latin-1")) + with contextlib.suppress(ConnectionError): + await writer.drain() diff --git a/packages/meshbay-cli/src/meshbay_cli/client/transfers.py b/packages/meshbay-cli/src/meshbay_cli/client/transfers.py index 3f23c85..2cbf4cd 100644 --- a/packages/meshbay-cli/src/meshbay_cli/client/transfers.py +++ b/packages/meshbay-cli/src/meshbay_cli/client/transfers.py @@ -268,6 +268,20 @@ async def _fetch_chunks(session: GroupSession, lease: Lease, file_id: str, file_ return written +async def read_chunk(session: GroupSession, entry: dict, i: int) -> bytes: + """ + Chunk `i` of a file, decrypted, asked for without a lease: what the + application's players do, and what the node bounds (a dozen files at once + per connection). A refusal raises `NodeRefusal`; a connection that failed, + another `CliError`. + """ + msg = await session.ch.request( + {"type": "file_req", "v": MNP_VERSION, "file_id": entry["id"], "chunk_index": i}, + "file_chunk", timeout=CHUNK_TIMEOUT) + return decrypt_chunk_aes(chunk_key_aes(session.gek, bytes.fromhex(entry["id"]), i), + msg["nonce"], msg["ct"]) + + def mark_of_the_web(path: Path) -> None: """ZoneId 3, as a browser marks a download, so SmartScreen and Protected View apply when it is opened. FAT and exFAT have no alternate streams: nothing to do.""" -- cgit v1.2.3