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 --- packages/meshbay-cli/src/meshbay_cli/__main__.py | 9 +- .../meshbay-cli/src/meshbay_cli/client/relay.py | 365 +++++++++++++++++++++ .../src/meshbay_cli/client/transfers.py | 14 + .../meshbay-cli/src/meshbay_cli/commands/stream.py | 135 ++++++++ packages/meshbay-cli/src/meshbay_cli/manpage.py | 40 ++- packages/meshbay-cli/tests/fake_node.py | 7 +- packages/meshbay-cli/tests/test_cli_actions.py | 28 ++ packages/meshbay-cli/tests/test_e2e_local.py | 78 +++++ packages/meshbay-cli/tests/test_stream.py | 285 ++++++++++++++++ 9 files changed, 952 insertions(+), 9 deletions(-) create mode 100644 packages/meshbay-cli/src/meshbay_cli/client/relay.py create mode 100644 packages/meshbay-cli/src/meshbay_cli/commands/stream.py create mode 100644 packages/meshbay-cli/tests/test_stream.py (limited to 'packages/meshbay-cli') diff --git a/packages/meshbay-cli/src/meshbay_cli/__main__.py b/packages/meshbay-cli/src/meshbay_cli/__main__.py index 2a71c17..f5b002a 100644 --- a/packages/meshbay-cli/src/meshbay_cli/__main__.py +++ b/packages/meshbay-cli/src/meshbay_cli/__main__.py @@ -28,6 +28,7 @@ from .commands import ( put, rm, search, + stream, trust, ) from .commands.context import Context @@ -35,8 +36,8 @@ from .errors import INTERRUPTED, USAGE, CliError ACTIONS = {"login": login, "logout": logout, "groups": groups, "invitations": invitations, "join": join, "decline": decline, "leave": leave, "ls": ls, "search": search, "get": get, - "put": put, "rm": rm, "chat": chat, "devices": devices, "trust": trust, - "browser-access": browser_access, "help": help} + "stream": stream, "put": put, "rm": rm, "chat": chat, "devices": devices, + "trust": trust, "browser-access": browser_access, "help": help} class _Parser(argparse.ArgumentParser): @@ -142,8 +143,8 @@ def main(argv: list[str] | None = None) -> int: def _utf8_output() -> None: """Names are printed as they are; a Windows console's code page cannot hold most.""" - for stream in (sys.stdout, sys.stderr): - reconfigure = getattr(stream, "reconfigure", None) + for out in (sys.stdout, sys.stderr): + reconfigure = getattr(out, "reconfigure", None) if reconfigure is not None: try: reconfigure(encoding="utf-8", errors="replace") 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.""" diff --git a/packages/meshbay-cli/src/meshbay_cli/commands/stream.py b/packages/meshbay-cli/src/meshbay_cli/commands/stream.py new file mode 100644 index 0000000..87f2b42 --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/commands/stream.py @@ -0,0 +1,135 @@ +""" +`stream`: a video or a sound file of a group, played by a media player. + +The file is served on this machine at an address printed on stdout, for VLC, +mpv or any player that reads HTTP (`client/relay.py`); `--player PROGRAM` +starts the player on it. Seeking is the player's: it asks for another byte +range. The stream runs until Ctrl+C, or until the player started with +`--player` is closed; the address works only while it runs. +""" + +import asyncio +import contextlib +import json +import os +import shutil +import subprocess +import sys + +from ..client.groups import my_groups, open_group +from ..client.paths import handle, parse, resolve_group +from ..client.relay import ChunkSource, Relay +from ..client.tree import Tree +from ..errors import NOT_FOUND, REFUSED, USAGE, CliError +from ..output import err, human_size, safe, sentence +from .context import Context + +HELP = "play a video or a sound file in VLC, mpv or another player" +PLAYABLE = ("video", "audio") + + +def add_parser(sub, parent) -> None: + p = sub.add_parser("stream", parents=[parent], help=HELP, description=sentence(HELP) + + ". The file is served on this machine at an address to open in " + "the player, until Ctrl+C. Nothing is written to disk.") + p.add_argument("path", metavar="PATH", help="group[@owner]/root/.../file") + p.add_argument("--player", metavar="PROGRAM", + help="start PROGRAM (vlc, mpv, ...) on the address; closing it stops " + "the stream") + + +def player_command(program: str) -> list[str]: + """The player to start: a name on the PATH, or a path. VLC on Windows is + rarely on the PATH, so its usual places are looked in too.""" + found = shutil.which(program) + if found is None and sys.platform == "win32" and program.lower() in ("vlc", "vlc.exe"): + for base in (os.environ.get("ProgramFiles"), os.environ.get("ProgramFiles(x86)")): + candidate = os.path.join(base or "", "VideoLAN", "VLC", "vlc.exe") + if base and os.path.isfile(candidate): + found = candidate + break + if found is None: + raise CliError(USAGE, f"{program} was not found; install it, or run without " + "--player and open the address yourself.") + return [found] + + +async def run(ctx: Context) -> int: + a = ctx.args + remote = parse(a.path) + if remote is None or not remote.parts: + raise CliError(USAGE, "Name the file to play: group/root/.../file.") + command = player_command(a.player) if a.player else None + await ctx.hub.check_version() + account = ctx.account() + session = source = relay = player = None + try: + group = resolve_group(await my_groups(account), remote.group) + session = await open_group(account, group) + entry = _the_file(Tree.from_index(await session.index()), remote, a.path) + name = entry["name"] + notes = (lambda _: None) if a.quiet else err + source = ChunkSource(session, entry, lambda: open_group(account, group), notes) + session = None # the source owns it now + relay = Relay(source, name, entry["type"]) + url = await relay.start() + + if a.json: + print(json.dumps({"url": url, "name": name, "size": int(entry["size"]), + "group": handle(group)}, ensure_ascii=False), flush=True) + else: + if not a.quiet: + err(f"Streaming {safe(name)} ({human_size(int(entry['size']))}) from " + f"{safe(handle(group))}." + + ("" if command else " Open this address in VLC or mpv:")) + print(url, flush=True) + if not a.quiet: + err("Close the player or press Ctrl+C to stop." if command + else "Ctrl+C to stop.") + + waits = [source.stopped] + if command: + player = subprocess.Popen( + command + [url], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, start_new_session=sys.platform != "win32") + # A thread, not asyncio's subprocesses: Windows' selector loop has none. + waits.append(asyncio.ensure_future(asyncio.to_thread(player.wait))) + done, _ = await asyncio.wait(waits, return_when=asyncio.FIRST_COMPLETED) + if source.stopped in done: + # The player is told too, rather than left waiting on a closed socket. + await relay.close(grace=1) + raise source.stopped.exception() + if not a.quiet and not a.json: + err("The player was closed; the stream is stopped.") + return 0 + finally: + if player is not None and player.poll() is None: + with contextlib.suppress(OSError): + player.terminate() + if relay is not None: + await relay.close() + if source is not None: + await source.close() + if session is not None: + await session.close() + await account.close() + + +def _the_file(tree: Tree, remote, text: str) -> dict: + found = tree.expand(remote.parts) + if not found and tree.file(remote.inner): + found = [remote.inner] # a literal name with brackets in it + files = [p for p in found if p and not tree.is_dir(p)] + if not files: + if found: + raise CliError(USAGE, f"{safe(text)} is a folder; name a file in it.") + raise CliError(NOT_FOUND, f"Nothing in this group matches {safe(text)}.") + if len(files) > 1: + raise CliError(USAGE, f"{safe(text)} names {len(files)} files; stream plays one.") + entry = tree.file(files[0]) + if entry.get("type") not in PLAYABLE: + raise CliError(USAGE, f"{safe(entry['name'])} is not a video or a sound file; " + "fetch it with get.") + if not tree.root_available(files[0]): + raise CliError(REFUSED, f"{safe(text)}: its folder on the node is unavailable.") + return entry diff --git a/packages/meshbay-cli/src/meshbay_cli/manpage.py b/packages/meshbay-cli/src/meshbay_cli/manpage.py index 601bd5b..c4710f3 100644 --- a/packages/meshbay-cli/src/meshbay_cli/manpage.py +++ b/packages/meshbay-cli/src/meshbay_cli/manpage.py @@ -28,7 +28,7 @@ EXIT_CODES = { errors.NODE: "No node serving the group could be reached, or the node refused " "the connection. Also: this machine is waiting for device " "approval (the code is on stderr), or the node asks for an " - "invitation's code.", + "invitation's code. Also: the node refused a file being streamed.", errors.NODE_KEY: "A node presented another key than the one pinned at first " "contact. Nothing was sent. See \\fBtrust\\fR.", errors.VERSION: "This client, or the protocol it speaks, is too old or too new " @@ -38,15 +38,16 @@ EXIT_CODES = { "group you own, revoking this machine, deleting a file you may not " "delete.", errors.INTERRUPTED: "Interrupted (Ctrl+C), which is also how chat \\-\\-follow " - "ends. A download resumes where it stopped on the next run.", + "and stream end. A download resumes where it stopped on the " + "next run.", } DESCRIPTION = r""" .B meshbay-cli is a member's view of their MeshBay groups from a terminal: sign in, join and leave groups, list them and their folders, search them all at once, download, -upload and delete files, read and post in a group's chat, and approve this -account's other devices. +upload and delete files, play a film or a song in a media player, read and +post in a group's chat, and approve this account's other devices. It talks to the hub only to sign in and to find a group's nodes; files go directly between this machine and the node, end-to-end encrypted, never through the hub. @@ -277,6 +278,34 @@ It is this machine's setting, as Browser access is the desktop application's: the hub keeps a copy for browsers to read, and that copy never turns it on here. """ +STREAMING = r""" +.B stream +plays one video or sound file of a group in a media player that reads HTTP, +such as VLC or mpv. +The file is served on this machine at an address printed on stdout, as it is +on the node: the player decodes it, so every format the player knows plays, +subtitles held in the file included. +Seeking is the player's: it asks for another part of the file, fetched from the +node then. +Nothing is written to disk. +.PP +The address works only on this machine (127.0.0.1) and only while +.B stream +runs; it carries a random part, so another program or another account of this +machine cannot guess it, and a request naming another host, as a web page +would send, is refused. +.B \-\-player +\fIPROGRAM\fR starts the player on the address (on Windows, VLC is looked for +where it installs itself too); closing it stops the stream. +Ctrl+C stops it too (status 130), and closes the player +.B stream +started. +A connection to the node that drops or stops answering is opened again; the +player sees a slower read. +If the node refuses the file, because it was removed or its disk went away, +the stream stops (status 5). +""" + FILES = r""" .TP .I $XDG_DATA_HOME/meshbay\-cli/ \fR(default \fI~/.local/share/meshbay\-cli/\fR) @@ -336,6 +365,8 @@ meshbay\-cli groups meshbay\-cli ls \-l Family/photos/2024 meshbay\-cli get \-R Family/photos/2024 \-o ~/Pictures/ meshbay\-cli get 'Family/music/*.flac' +meshbay\-cli stream \-\-player vlc 'Family/films/Night Drive.mkv' +meshbay\-cli stream \-\-player mpv "$(meshbay\-cli search \-t audio nocturne | head \-1)" meshbay\-cli put \-R ~/scans Family/documents tar c notes | meshbay\-cli put \- Family/documents/notes.tar meshbay\-cli ls \-\-json \-R / | jq '.[] | .path' @@ -444,6 +475,7 @@ def render() -> str: out += [".SH REMOTE PATHS", PATHS.strip(), ".SH INTERACTIVE MODE", INTERACTIVE.strip(), ".SH SESSION, DEVICES AND NODE KEYS", SESSION.strip(), + ".SH STREAMING", STREAMING.strip(), ".SH GROUPS: JOINING, LEAVING, DELETING, CHAT, DEVICES", MEMBERSHIP.strip(), ".SH EXIT STATUS"] for code, text in sorted(EXIT_CODES.items()): diff --git a/packages/meshbay-cli/tests/fake_node.py b/packages/meshbay-cli/tests/fake_node.py index b4e64ff..f9c552c 100644 --- a/packages/meshbay-cli/tests/fake_node.py +++ b/packages/meshbay-cli/tests/fake_node.py @@ -57,6 +57,7 @@ class FakeNode: self.fail_chunk: tuple[str, int] | None = None # (file_id, index) → refused self.refuse_upload_code = "" self.chunk_requests = 0 + self.leaseless_reads: list[tuple[str, int]] = [] self.chunk_delay = 0.0 # This machine's key here, the account's other devices, and the requests # waiting for an approval — each `{pk_ed25519, pk_x25519, code_hash}`. @@ -215,7 +216,11 @@ class FakeChannel: def _file_req(self, msg): n = self.node n.chunk_requests += 1 - if n.leases.get(msg.get("tr")) != "granted": + if "tr" not in msg: + # A read without a lease, as a player's: the node admits a dozen + # files at once per connection. + n.leaseless_reads.append((msg["file_id"], msg["chunk_index"])) + elif n.leases.get(msg.get("tr")) != "granted": return {"type": "error", "detail": "no lease", "code": "lease_not_granted"} if n.fail_chunk == (msg["file_id"], msg["chunk_index"]): return {"type": "error", "detail": "The drive stopped answering"} diff --git a/packages/meshbay-cli/tests/test_cli_actions.py b/packages/meshbay-cli/tests/test_cli_actions.py index 240ac67..bcbe0c4 100644 --- a/packages/meshbay-cli/tests/test_cli_actions.py +++ b/packages/meshbay-cli/tests/test_cli_actions.py @@ -8,6 +8,8 @@ to the parser without a run below. import io import json +import os +import sys import pytest from fake_hub import PASSPHRASE, USERNAME, FakeHub @@ -568,6 +570,32 @@ def test_chat_reads_the_last_messages_and_posts(node, monkeypatch, capsys): assert main("chat", "-0", "Family") == 1 +def test_stream_serves_a_film_to_the_player_it_starts(node, tmp_path, monkeypatch, capsys): + from meshbay_cli.commands import stream + + async def open_group(account, group): + return node.session() + monkeypatch.setattr(stream, "open_group", open_group) + film = os.urandom(1024 * 1024 + 7) + node.add("share/films", "Night Drive.mkv", film)["type"] = "video" + node.add("share", "notes.txt", b"hello") + + assert main("stream", "Family/share/notes.txt") == 1 + assert "not a video or a sound file" in capsys.readouterr().err + if sys.platform == "win32": + return # a script is not a program there + played = tmp_path / "played" + player = tmp_path / "player" + player.write_text(f"#!{sys.executable}\nimport sys, urllib.request\n" + "opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))\n" + f"open({str(played)!r}, 'wb').write(opener.open(sys.argv[1]).read())\n") + player.chmod(0o755) + assert main("stream", "--player", str(player), "Family/share/films/Night Drive.mkv") == 0 + assert played.read_bytes() == film + out = capsys.readouterr() + assert out.out.startswith("http://127.0.0.1:") and "player was closed" in out.err + + def test_every_action_is_run_here(): """Runs last in this module: every action the parser knows was run above.""" assert RUN == set(cli.ACTIONS), f"never run: {set(cli.ACTIONS) - RUN}" diff --git a/packages/meshbay-cli/tests/test_e2e_local.py b/packages/meshbay-cli/tests/test_e2e_local.py index fb251bd..c385bb6 100644 --- a/packages/meshbay-cli/tests/test_e2e_local.py +++ b/packages/meshbay-cli/tests/test_e2e_local.py @@ -1172,3 +1172,81 @@ def test_chat_follow_against_the_real_node(world): _, err = follower.communicate(timeout=20) assert follower.returncode == 130 assert "reconnecting" not in err, err + + +@pytest.mark.skipif(not (shutil.which("ffmpeg") and shutil.which("ffprobe") + and shutil.which("cvlc")), + reason="ffmpeg, ffprobe and VLC are needed") +def test_stream_a_film_to_real_players(world, tmp_path): + """A film put on the real node, then streamed: ffprobe reads it, ffmpeg seeks + near its end, a foreign Host is refused, Ctrl+C stops it; then VLC started + with --player plays it to the end and, closing, ends the stream.""" + import signal + _cli_signed_in(world, "cli-stream") + film = tmp_path / "Night Drive.mkv" + subprocess.run(["ffmpeg", "-v", "error", "-f", "lavfi", "-i", + "testsrc=duration=12:size=640x360:rate=25", "-f", "lavfi", "-i", + "sine=frequency=440:duration=12", "-vf", "noise=alls=30:allf=t", + "-c:v", "libx264", "-b:v", "3M", + "-g", "50", "-c:a", "aac", "-shortest", str(film)], check=True) + assert film.stat().st_size > 3 * 1024 * 1024 # several of the node's chunks + r = cli(world, "put", str(film), f"{_handle()}/share", home="cli-stream") + # After a test that withdrew the node's copy of the keys, this machine is a + # new device: approved as the application would approve it. + if r.returncode == 5 and "Waiting for this machine to be approved" in r.stderr: + code = re.search(r"code ([0-9A-Z]{4}-[0-9A-Z]{4})", r.stderr).group(1) + asyncio.run(world["app"].approve(code)) + r = cli(world, "put", str(film), f"{_handle()}/share", home="cli-stream") + assert r.returncode == 0, r.stderr + path = f"{_handle()}/share/Night Drive.mkv" + deadline = time.monotonic() + 60 # once the node has indexed it + while "Night Drive.mkv" not in cli(world, "ls", f"{_handle()}/share", + home="cli-stream").stdout: + assert time.monotonic() < deadline, "the film was never indexed" + time.sleep(1) + + env = {**os.environ, "MESHBAY_CLI_HOME": str(world["root"] / "cli-stream"), + "PYTHON_KEYRING_BACKEND": "keyring.backends.fail.Keyring"} + exe = os.environ.get("MESHBAY_CLI_E2E_BIN") or str(BIN / "meshbay-cli") + streamer = subprocess.Popen([exe, "--json", "stream", path], env=env, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) + try: + line = streamer.stdout.readline() + assert line, streamer.stderr.read() + shown = json.loads(line) + url = shown["url"] + assert shown["size"] == film.stat().st_size + probe = subprocess.run(["ffprobe", "-v", "error", "-show_entries", + "format=duration:stream=codec_name", "-of", "json", url], + capture_output=True, text=True, timeout=60) + assert probe.returncode == 0, probe.stderr + info = json.loads(probe.stdout) + assert abs(float(info["format"]["duration"]) - 12) < 0.5 + assert {s["codec_name"] for s in info["streams"]} == {"h264", "aac"} + frame = tmp_path / "frame.png" + seek = subprocess.run(["ffmpeg", "-v", "error", "-ss", "10", "-i", url, + "-frames:v", "1", str(frame)], + capture_output=True, text=True, timeout=60) + assert seek.returncode == 0 and frame.stat().st_size > 0, seek.stderr + port = url.split("/")[2].split(":")[1] + foreign = httpx.get(url, headers={"Host": f"rebound.example:{port}"}, timeout=10, + trust_env=False) + assert foreign.status_code == 403 + finally: + streamer.send_signal(signal.SIGINT) + _, err = streamer.communicate(timeout=20) + assert streamer.returncode == 130, err + with pytest.raises(httpx.HTTPError): + httpx.get(url, timeout=5, trust_env=False) + + player = tmp_path / "vlc-to-the-end" + log = tmp_path / "vlc.log" + player.write_text("#!/bin/sh\nexec cvlc --play-and-exit --intf dummy --vout dummy " + f"--aout dummy \"$1\" >{log} 2>&1\n") + player.chmod(0o755) + started = time.monotonic() + r = subprocess.run([exe, "stream", "--player", str(player), path], env=env, + capture_output=True, text=True, timeout=120) + assert r.returncode == 0, r.stderr + log.read_text() + assert "The player was closed" in r.stderr + assert time.monotonic() - started > 10, "VLC played the film, in real time" diff --git a/packages/meshbay-cli/tests/test_stream.py b/packages/meshbay-cli/tests/test_stream.py new file mode 100644 index 0000000..46dc630 --- /dev/null +++ b/packages/meshbay-cli/tests/test_stream.py @@ -0,0 +1,285 @@ +""" +`stream`: a file served on this machine to a media player — byte ranges from +the node's chunks, read without a lease, only to a request that names this +address and carries the token; the connection opened again when it dies, and +the stream stopped by the node's refusal, by Ctrl+C or by closing the player. +""" + +import asyncio +import json +import os +import secrets +import stat +import sys +from urllib.parse import urlsplit + +import pytest +from fake_hub import PASSPHRASE, USERNAME, FakeHub +from fake_node import FakeNode +from meshbay_cli import __main__ as cli +from meshbay_cli.client import relay +from meshbay_cli.client.relay import byte_range, content_type +from meshbay_cli.commands import context, stream +from meshbay_cli.errors import NODE, NOT_FOUND, USAGE, CliError + +MiB = 1024 * 1024 +FILM = secrets.token_bytes(3 * MiB + 12345) + + +@pytest.fixture +def world(monkeypatch, capsys): + hub, node = FakeHub(), FakeNode() + monkeypatch.setattr(context, "Hub", lambda url: hub) + sessions = [] + + async def open_group(account, group, **_): + sessions.append(node.session()) + return sessions[-1] + monkeypatch.setattr(stream, "open_group", open_group) + monkeypatch.setattr(relay, "PING_EVERY", 0.05) + monkeypatch.setattr(relay, "RECONNECT_WAITS", (0.01,)) + node.add("share/films", "Holiday.mkv", FILM)["type"] = "video" + node.add("share/docs", "notes.txt", b"hello") + assert cli.main(["--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE, + "login"]) == 0 + capsys.readouterr() + + def start(*argv): + args = cli.build_parser().parse_args(list(argv)) + return asyncio.ensure_future(stream.run(context.Context(args))) + return node, sessions, start + + +def node_film(node) -> dict: + return next(e for e in node.files.values() if e["name"] == "Holiday.mkv") + + +async def _until(cond, timeout=5.0): + deadline = asyncio.get_running_loop().time() + timeout + while not cond(): + assert asyncio.get_running_loop().time() < deadline, "timed out" + await asyncio.sleep(0.02) + + +async def _url(task, capsys, said: list | None = None) -> str: + """The address printed on stdout; what was said on stderr meanwhile into `said`.""" + out = [] + + def printed(): + assert not task.done(), task.exception() + got = capsys.readouterr() + out.extend(got.out.split()) + if said is not None: + said.append(got.err) + return out + await _until(printed) + return out[0] + + +async def http(url: str, *, method="GET", headers=None, host=None, path=None, + keep=False, reader_writer=None): + """One request; (status, headers, body). With `keep`, the connection is returned + for another request.""" + u = urlsplit(url) + reader, writer = reader_writer or await asyncio.open_connection(u.hostname, u.port) + lines = [f"{method} {path or u.path} HTTP/1.1", f"Host: {host or u.netloc}", + *(f"{k}: {v}" for k, v in (headers or {}).items())] + if not keep: + lines.append("Connection: close") + writer.write(("\r\n".join(lines) + "\r\n\r\n").encode()) + await writer.drain() + head = (await reader.readuntil(b"\r\n\r\n")).decode("latin-1").split("\r\n") + status = int(head[0].split(" ")[1]) + fields = {k.strip().lower(): v.strip() for k, _, v in + (line.partition(":") for line in head[1:] if line)} + body = b"" + if method != "HEAD": + body = await reader.readexactly(int(fields.get("content-length", 0))) + if keep: + return status, fields, body, (reader, writer) + writer.close() + return status, fields, body + + +# ── Ranges and types ───────────────────────────────────────────────────────── + +def test_a_range_is_read_as_players_send_it(): + assert byte_range(None, 1000) is None + assert byte_range("bytes=0-", 1000) == (0, 999) + assert byte_range("bytes=10-19", 1000) == (10, 19) + assert byte_range("bytes=990-5000", 1000) == (990, 999) + assert byte_range("bytes=-100", 1000) == (900, 999) + assert byte_range("bytes=-5000", 1000) == (0, 999) + assert byte_range("bytes=0-1,5-9", 1000) is None # several: the whole file + assert byte_range("items=0-9", 1000) is None + for bad in ("bytes=1000-", "bytes=20-10", "bytes=x-", "bytes=-0"): + with pytest.raises(ValueError): + byte_range(bad, 1000) + + +def test_a_file_is_served_as_audio_or_video_and_never_as_a_page(): + assert content_type("Film.MKV", "video") == "video/x-matroska" + assert content_type("a.flac", "audio") == "audio/flac" + assert content_type("trap.html", "video") == "video/octet-stream" + assert content_type("trap.svg", "audio") == "audio/octet-stream" + + +# ── Serving ────────────────────────────────────────────────────────────────── + +async def test_the_player_gets_the_file_and_any_range_of_it(world, capsys): + node, sessions, start = world + task = start("stream", "Family/share/films/Holiday.mkv") + try: + said = [] + url = await _url(task, capsys, said) + assert url.startswith("http://127.0.0.1:") and url.endswith("/Holiday.mkv") + assert "Ctrl+C to stop" in "".join(said) + + status, fields, body = await http(url) + assert status == 200 and body == FILM + assert fields["content-type"] == "video/x-matroska" + assert fields["accept-ranges"] == "bytes" + assert fields["x-content-type-options"] == "nosniff" + + # Across a chunk boundary, as a seek lands. + a, b = MiB - 100, 2 * MiB + 50 + status, fields, body = await http(url, headers={"Range": f"bytes={a}-{b}"}) + assert status == 206 and body == FILM[a:b + 1] + assert fields["content-range"] == f"bytes {a}-{b}/{len(FILM)}" + + # The end of the file, as a player probes for an index there; then the + # same connection again. + status, _, body, conn = await http(url, headers={"Range": "bytes=-64"}, keep=True) + assert status == 206 and body == FILM[-64:] + status, _, body, conn = await http(url, headers={"Range": "bytes=5-9"}, keep=True, + reader_writer=conn) + assert status == 206 and body == FILM[5:10] + conn[1].close() + + status, fields, body = await http(url, method="HEAD") + assert status == 200 and fields["content-length"] == str(len(FILM)) and body == b"" + status, fields, _ = await http(url, headers={"Range": f"bytes={len(FILM)}-"}) + assert status == 416 and fields["content-range"] == f"bytes */{len(FILM)}" + + # Read as a player reads, without a lease; chunks kept are not asked twice. + assert node.leases == {} and node.leaseless_reads + assert sorted(node.leaseless_reads) == [(node_film(node)["id"], i) for i in range(4)] + assert len(sessions) == 1 + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +async def test_only_this_address_with_its_token_is_answered(world, capsys): + node, _, start = world + task = start("stream", "Family/share/films/Holiday.mkv") + try: + url = await _url(task, capsys) + port = urlsplit(url).port + # A page that rebinds its own name to 127.0.0.1 sends its own Host. + status, _, body = await http(url, host=f"evil.example:{port}") + assert status == 403 and body == b"" + status, _, body = await http(url, path="/not-the-token/Holiday.mkv") + assert status == 404 and body == b"" + status, _, _ = await http(url, method="POST") + assert status == 405 + assert node.leaseless_reads == [], "nothing refused reached the node" + status, _, body = await http(url, host=f"localhost:{port}", + headers={"Range": "bytes=0-3"}) + assert status == 206 and body == FILM[:4] + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +async def test_a_dead_connection_is_opened_again(world, capsys): + node, sessions, start = world + task = start("stream", "Family/share/films/Holiday.mkv") + try: + url = await _url(task, capsys) + assert (await http(url, headers={"Range": "bytes=0-9"}))[2] == FILM[:10] + + sessions[-1].ch.closed.set() # the link goes down + await _until(lambda: len(sessions) == 2) + a = 3 * MiB + assert (await http(url, headers={"Range": f"bytes={a}-"}))[2] == FILM[a:] + + node.unresponsive = True # it stops answering pings + await _until(lambda: len(sessions) == 3) + node.unresponsive = False + assert "reconnecting" in capsys.readouterr().err + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +async def test_a_refusal_by_the_node_stops_the_stream(world, capsys): + node, _, start = world + node.fail_chunk = (node_film(node)["id"], 2) + task = start("stream", "Family/share/films/Holiday.mkv") + url = await _url(task, capsys) + status, _, _ = await http(url, headers={"Range": f"bytes={2 * MiB}-"}) + assert status == 502 + with pytest.raises(CliError) as e: + await asyncio.wait_for(task, 5) + assert e.value.code == NODE and "drive stopped answering" in e.value.message + + +async def test_what_cannot_be_streamed_is_said(world, capsys): + _, _, start = world + for argv, code, says in [ + (["Family/share/docs/notes.txt"], USAGE, "not a video or a sound file"), + (["Family/share/films"], USAGE, "is a folder"), + (["Family/share/films/Nope.mkv"], NOT_FOUND, "Nothing"), + (["Family"], USAGE, "Name the file"), + (["--player", "no-such-player-here", "Family/share/films/Holiday.mkv"], USAGE, + "was not found"), + ]: + with pytest.raises(CliError) as e: + await start("stream", *argv) + assert e.value.code == code and says in e.value.message, argv + + +# ── --player ───────────────────────────────────────────────────────────────── + +def _player(tmp_path, body: str) -> str: + script = tmp_path / "player" + script.write_text(f"#!{sys.executable}\nimport os, sys, time, urllib.request\n{body}\n") + script.chmod(script.stat().st_mode | stat.S_IXUSR) + return str(script) + + +@pytest.mark.skipif(sys.platform == "win32", reason="a script as a program") +async def test_closing_the_player_stops_the_stream(world, capsys, tmp_path): + out = tmp_path / "played" + player = _player(tmp_path, ( + "opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))\n" + f"open({str(out)!r}, 'wb').write(opener.open(sys.argv[1]).read())")) + _, _, start = world + code = await asyncio.wait_for( + start("stream", "--player", player, "Family/share/films/Holiday.mkv"), 10) + assert code == 0 and out.read_bytes() == FILM + assert "player was closed" in capsys.readouterr().err + + +@pytest.mark.skipif(sys.platform == "win32", reason="a script as a program") +async def test_stopping_the_stream_closes_the_player_it_started(world, capsys, tmp_path): + pid = tmp_path / "pid" + player = _player(tmp_path, f"open({str(pid)!r}, 'w').write(str(os.getpid()))\n" + "time.sleep(60)") + _, _, start = world + task = start("--json", "stream", "--player", player, "Family/share/films/Holiday.mkv") + await _until(lambda: pid.exists() and pid.read_text()) + shown = json.loads(capsys.readouterr().out) + assert shown["name"] == "Holiday.mkv" and shown["size"] == len(FILM) + task.cancel() # Ctrl+C + await asyncio.gather(task, return_exceptions=True) + + def gone(): + try: + os.kill(int(pid.read_text()), 0) + except ProcessLookupError: + return True + with open(f"/proc/{pid.read_text()}/stat") as f: + return f.read().split()[2] == "Z" # ended, not yet reaped + await _until(gone) -- cgit v1.2.3