feat(cli): stream — a film or a song played in a media player

`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 <noreply@anthropic.com>

4 files changed, 397 insertions, 1 deletions

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)