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/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 +++++++++++++++++++++++++ 4 files changed, 397 insertions(+), 1 deletion(-) create mode 100644 packages/meshbay-cli/tests/test_stream.py (limited to 'packages/meshbay-cli/tests') 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