warehack.ing/mcbebop
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

io: ARStream2 video, read-only FTP areas, allow-listed debug shell

Browse Source 
Ports bebop-2's working video.py, which was verified against the live
aircraft. Firmware 4.7.1 serves no RTSP, so there is no URL and no RTSP
path here: we describe our own receiving port in an SDP and let ffmpeg
bind it before VideoEnable goes out, because RTP is connectionless and
packets that arrive before the sink is listening are gone.

Three changes on top of the port. StreamSession no longer owns the drone
link, since the tools layer holds a long-lived session; it takes an async
sender callable instead, which also keeps media/ loadable while arsdk/
and protocol/ are still being written. The context manager is async for
the same reason, with the blocking subprocess waits moved off the event
loop. And downscale() shrinks a frame before it reaches a model, because
a full 856x480 is most of a context window spent on pixels nobody asked
for.

FTP exposes media (21), flightplans (61) and logs (21, scoped to the
Debug tree). Port 51 is deliberately absent: it serves /update as root,
it is how firmware is pushed, it has no read use case, and the drone's
Wi-Fi is open. Fetches stream to capture_dir under a size cap so a 1080p
recording cannot be pulled into a tool result.

The shell is an allow-list of eleven read-only command names rather than
a deny-list, because the login is `exec /bin/sh -l` with no password and
deny-lists on shells leak. Arguments carrying shell metacharacters are
refused before the socket opens. Output is bracketed between two echoed
nonce markers rather than trimmed by prompt pattern, since telnetd's pty
echoes our input with the prompt glued to the front and sends all of it
before anything runs.
This commit is contained in:
rsp2k
2026-10-02 00:12:18 -06:00
parent 5732befe82
commit d38b215348
6 changed files with 1780 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_ftp.py
+242
View File
@@ -0,0 +1,242 @@
"""FTP: the area-to-port map, path containment, and the size cap.
ftplib is replaced with a fake; nothing opens a socket.
"""
import ftplib
from typing import ClassVar
import pytest
from mcbebop.files import ftp
# --- the area map ------------------------------------------------------------
def test_the_three_read_only_areas_map_to_the_ports_found_on_the_aircraft():
assert {a.name: a.port for a in ftp.areas()} == {"media": 21, "flightplans": 61, "logs": 21}
def test_the_firmware_write_port_is_not_reachable_through_the_api():
# Port 51 serves /update as root. It is how firmware is pushed, it has no
# read use case, and the drone's Wi-Fi is open. Exposing it would hand the
# aircraft's weakest point to anything that can call a tool.
assert ftp.FIRMWARE_WRITE_PORT == 51
assert 51 not in {a.port for a in ftp.areas()}
for name in ("update", "firmware", "51"):
with pytest.raises(ftp.FtpError):
ftp.resolve_area(name)
def test_an_unknown_area_says_what_is_available():
with pytest.raises(ftp.FtpError, match="flightplans"):
ftp.resolve_area("nope")
def test_logs_is_the_media_ftpd_narrowed_to_the_blackbox_tree():
logs = ftp.resolve_area("logs")
media = ftp.resolve_area("media")
assert logs.port == media.port == 21
assert logs.prefix == "internal_000/Debug/current"
assert media.prefix == ""
def test_every_area_describes_itself():
assert all(a.describe for a in ftp.areas())
# --- path containment --------------------------------------------------------
def test_a_logs_path_stays_inside_the_debug_tree():
area = ftp.resolve_area("logs")
assert ftp._safe_path(area, "") == "internal_000/Debug/current"
assert ftp._safe_path(area, "boot.log") == "internal_000/Debug/current/boot.log"
def test_dot_dot_cannot_climb_out_of_an_area():
area = ftp.resolve_area("logs")
for path in ("../../..", "a/../../b", ".."):
with pytest.raises(ftp.FtpError, match=r"\.\."):
ftp._safe_path(area, path)
def test_an_absolute_path_is_refused_rather_than_silently_rebased():
with pytest.raises(ftp.FtpError, match="relative"):
ftp._safe_path(ftp.resolve_area("media"), "/data/ftp/internal_000")
def test_an_empty_media_path_lists_the_area_root():
assert ftp._safe_path(ftp.resolve_area("media"), "") == "."
def test_redundant_separators_and_dots_collapse():
area = ftp.resolve_area("media")
assert ftp._safe_path(area, "internal_000//./media/") == "internal_000/media"
# --- listing -----------------------------------------------------------------
UNIX_LINE = "-rw-r--r-- 1 root root 1048576 Jan 1 00:00 flightPlan.mavlink"
DIR_LINE = "drwxr-xr-x 2 root root 4096 Jan 1 00:00 internal_000"
def test_a_unix_list_line_parses():
entry = ftp._parse_line(UNIX_LINE)
assert entry.name == "flightPlan.mavlink"
assert entry.size == 1048576
assert entry.is_dir is False
def test_a_directory_line_is_marked_as_one():
assert ftp._parse_line(DIR_LINE).is_dir is True
def test_an_unrecognised_line_still_yields_a_name():
entry = ftp._parse_line("weird-output.txt")
assert entry.name == "weird-output.txt"
assert entry.size is None
assert entry.raw == "weird-output.txt"
# --- a fake ftpd -------------------------------------------------------------
class FakeFTP:
"""Just enough ftplib surface, recording what was asked of it.
`FakeFTP.config` is what a test uses to describe the ftpd it wants, since
ftplib.FTP is constructed inside the module under test.
"""
instances: ClassVar[list["FakeFTP"]] = []
config: ClassVar[dict] = {}
def __init__(self, timeout=None):
self.timeout = timeout
self.connected_to = None
self.logged_in = False
self.commands: list[str] = []
self.quit_called = False
self.listing = FakeFTP.config.get("listing", [DIR_LINE, UNIX_LINE])
self.content = FakeFTP.config.get("content", b"x" * 64)
self.declared_size = FakeFTP.config.get("declared_size")
self.size_raises = FakeFTP.config.get("size_raises", False)
FakeFTP.instances.append(self)
def connect(self, host, port):
self.connected_to = (host, port)
def login(self, *a):
self.logged_in = True
def retrlines(self, cmd, callback):
self.commands.append(cmd)
for line in self.listing:
callback(line)
def retrbinary(self, cmd, callback):
self.commands.append(cmd)
for i in range(0, len(self.content), 16):
callback(self.content[i : i + 16])
def voidcmd(self, cmd):
self.commands.append(cmd)
def size(self, path):
if self.size_raises:
raise ftplib.error_perm("550 SIZE not understood")
return self.declared_size
def quit(self):
self.quit_called = True
def close(self):
pass
@pytest.fixture
def fake_ftp(monkeypatch):
FakeFTP.instances = []
FakeFTP.config = {}
monkeypatch.setattr(ftp.ftplib, "FTP", FakeFTP)
return FakeFTP
def test_list_dir_connects_to_the_area_port_and_logs_in_anonymously(fake_ftp):
entries = ftp.list_dir("flightplans", host="192.168.42.1")
conn = fake_ftp.instances[0]
assert conn.connected_to == ("192.168.42.1", 61)
assert conn.logged_in
assert conn.timeout == ftp.DEFAULT_TIMEOUT
assert [e.name for e in entries] == ["internal_000", "flightPlan.mavlink"]
assert conn.quit_called
def test_list_dir_scopes_the_logs_area_in_the_command_it_sends(fake_ftp):
ftp.list_dir("logs", host="h")
assert fake_ftp.instances[0].commands[0] == "LIST internal_000/Debug/current"
def test_a_refused_connection_explains_the_access_point(fake_ftp, monkeypatch):
def refuse(self, host, port):
raise OSError("connection refused")
monkeypatch.setattr(FakeFTP, "connect", refuse)
with pytest.raises(ftp.FtpError, match="access point"):
ftp.list_dir("media", host="h")
# --- fetching ----------------------------------------------------------------
def test_fetch_streams_to_capture_dir_and_returns_the_path(fake_ftp, tmp_path):
dest = ftp.fetch("flightplans", "flightPlan.mavlink", host="h", capture_dir=tmp_path)
assert dest == tmp_path / "flightplans" / "flightPlan.mavlink"
assert dest.read_bytes() == b"x" * 64
assert "RETR flightPlan.mavlink" in fake_ftp.instances[0].commands
def test_fetch_asks_for_the_size_in_binary_mode_first(fake_ftp, tmp_path):
# busybox will not answer SIZE in ASCII mode, so TYPE I has to come first.
ftp.fetch("media", "internal_000/a.jpg", host="h", capture_dir=tmp_path)
cmds = fake_ftp.instances[0].commands
assert cmds[0] == "TYPE I"
def test_a_file_the_drone_declares_as_oversized_is_never_transferred(fake_ftp, tmp_path):
fake_ftp.config["declared_size"] = 50_000_000
with pytest.raises(ftp.TooLarge, match="cap"):
ftp.fetch("media", "internal_000/big.mp4", host="h", capture_dir=tmp_path)
assert not any(c.startswith("RETR") for c in fake_ftp.instances[0].commands)
assert not (tmp_path / "media" / "big.mp4").exists()
def test_a_cap_passed_mid_transfer_aborts_and_removes_the_partial_file(fake_ftp, tmp_path):
# The second line of defence: an ftpd that will not answer SIZE still
# cannot stream a video into memory, because the callback counts bytes.
fake_ftp.config.update(content=b"y" * 4096, size_raises=True)
with pytest.raises(ftp.TooLarge, match="mid-transfer"):
ftp.fetch("media", "internal_000/big.mp4", host="h", capture_dir=tmp_path, max_bytes=100)
assert not (tmp_path / "media" / "big.mp4").exists()
def test_fetch_refuses_a_directory(fake_ftp, tmp_path):
with pytest.raises(ftp.FtpError, match="file path"):
ftp.fetch("media", "", host="h", capture_dir=tmp_path)
def test_fetch_cannot_be_talked_out_of_its_area(fake_ftp, tmp_path):
with pytest.raises(ftp.FtpError):
ftp.fetch("logs", "../../internal_000/video.mp4", host="h", capture_dir=tmp_path)
def test_a_server_error_during_fetch_leaves_no_stub_file(fake_ftp, tmp_path, monkeypatch):
def fail(self, cmd, callback):
raise ftplib.error_perm("550 No such file")
monkeypatch.setattr(FakeFTP, "retrbinary", fail)
with pytest.raises(ftp.FtpError, match="Could not fetch"):
ftp.fetch("media", "internal_000/ghost.jpg", host="h", capture_dir=tmp_path)
assert not (tmp_path / "media" / "ghost.jpg").exists()
tests/test_shell.py
+378
View File
@@ -0,0 +1,378 @@
"""Shell: the allow-list, metacharacter refusal, and the telnet handshake.
The socket is a fake. The point of most of these tests is that nothing
reaches it: validation happens before a byte is sent.
"""
import socket
import pytest
from mcbebop.files import shell
# --- the allow-list ----------------------------------------------------------
def test_only_read_only_commands_are_allowed():
assert shell.allowed_commands() == [
"bcmwl",
"cat",
"df",
"dmesg",
"getprop",
"head",
"ls",
"mount",
"ps",
"tail",
"uptime",
]
@pytest.mark.parametrize(
"name",
[
"rm",
"sh",
"bash",
"dd",
"mv",
"cp",
"chmod",
"reboot",
"telnetd",
"echo",
"eval",
"python",
"CAT",
"ls ",
],
)
def test_a_command_outside_the_allow_list_is_refused(name):
# Deny-lists on a root shell leak; there is always another spelling. So
# the only names that work are the ones explicitly listed.
with pytest.raises(shell.NotAllowed, match="not one of"):
shell.validate(name)
def test_the_refusal_lists_what_is_available():
with pytest.raises(shell.NotAllowed, match="getprop"):
shell.validate("rm")
@pytest.mark.parametrize("meta", [";", "|", "&", "$", "`", ">", "<", "\n", "\r", "\\", '"', "'", "(", ")"])
def test_an_argument_carrying_a_shell_metacharacter_is_refused(meta):
with pytest.raises(shell.NotAllowed, match="metacharacter"):
shell.validate("cat", [f"/proc/version{meta}"])
@pytest.mark.parametrize(
"arg",
[
"/etc/passwd; rm -rf /",
"$(reboot)",
"`id`",
"/dev/null > /bin/sh",
"a | sh",
"x && reboot",
"a\necho hi",
"\\;",
],
)
def test_the_classic_injections_never_get_past_validate(arg):
with pytest.raises(shell.NotAllowed):
shell.validate("cat", [arg])
def test_an_empty_argument_is_refused():
with pytest.raises(shell.NotAllowed, match="Empty"):
shell.validate("ls", [""])
def test_flags_and_ordinary_paths_are_fine():
assert shell.validate("ls", ["-l", "/data/ftp"]) == "ls -l /data/ftp"
assert shell.validate("getprop", ["ro.parrot.build.version"]) == "getprop ro.parrot.build.version"
assert shell.validate("tail", ["-n", "40", "/var/log/messages"]) == "tail -n 40 /var/log/messages"
assert shell.validate("uptime") == "uptime"
def test_validation_happens_before_the_socket_opens(monkeypatch):
def explode(*a, **kw):
raise AssertionError("must not connect for a refused command")
monkeypatch.setattr(shell.socket, "create_connection", explode)
with pytest.raises(shell.NotAllowed):
shell.run("rm", ["-rf", "/"], host="192.168.42.1")
with pytest.raises(shell.NotAllowed):
shell.run("cat", ["/etc/passwd; reboot"], host="192.168.42.1")
# --- a fake telnetd ----------------------------------------------------------
class FakeSocket:
"""Replays a scripted server side and records what we sent."""
def __init__(self, chunks):
self.chunks = list(chunks)
self.sent = bytearray()
self.closed = False
self.timeout = None
def settimeout(self, t):
self.timeout = t
def sendall(self, data):
self.sent += data
def recv(self, n):
if not self.chunks:
return b""
chunk = self.chunks.pop(0)
if chunk is TimeoutError:
raise TimeoutError("timed out")
return chunk
def close(self):
self.closed = True
@pytest.fixture
def fake_telnet():
"""Somewhere for a scripted connect() to record what it saw."""
return {}
def _markers(sock: FakeSocket) -> tuple[str, str]:
"""Dig the two markers out of what the module sent, so a fake can echo them."""
lines = sock.sent.decode().strip().splitlines()
return lines[0].split()[-1], lines[-1].split()[-1]
def _pty(holder, *, output=b"", banner=b"", trailing=b"", negotiate=b""):
"""A telnetd that behaves like a real one: pty echo first, output after.
The echo of all three input lines arrives in one burst *before* anything
runs, so a reader that stopped at the first sight of the end marker would
return nothing at all. That is the case this fake exists to reproduce.
"""
def connect(address, timeout=None):
holder["address"] = address
holder["timeout"] = timeout
class Responder(FakeSocket):
def recv(self, n):
if not self.sent or self.chunks == ["done"]:
return b""
begin, end = _markers(self)
echo = self.sent.decode()
self.chunks = ["done"]
return (
negotiate
+ banner
# the pty echoing our input, prompt glued to the front
+ b"".join(f"/ # {ln}\n".encode() for ln in echo.splitlines())
+ f"{begin}\n".encode()
+ (output + b"\n" if output else b"")
+ f"/ # {end}\n".encode()
+ trailing
)
sock = Responder([])
holder["sock"] = sock
return sock
return connect
def test_a_command_runs_and_its_output_comes_back(monkeypatch, fake_telnet):
monkeypatch.setattr(shell.socket, "create_connection", _pty(fake_telnet, output=b"Linux version 3.4.11"))
result = shell.run("cat", ["/proc/version"], host="192.168.42.1")
assert result.command == "cat /proc/version"
assert result.stdout == "Linux version 3.4.11"
assert fake_telnet["address"] == ("192.168.42.1", 23)
assert fake_telnet["sock"].closed
def test_the_pty_echo_of_our_own_input_never_reaches_the_caller(monkeypatch, fake_telnet):
monkeypatch.setattr(shell.socket, "create_connection", _pty(fake_telnet, output=b"3.4.11"))
out = shell.run("cat", ["/proc/version"], host="h").stdout
assert out == "3.4.11"
assert "cat /proc/version" not in out
assert "MCBEBOP" not in out
def test_a_login_banner_before_the_markers_is_dropped(monkeypatch, fake_telnet):
monkeypatch.setattr(
shell.socket,
"create_connection",
_pty(fake_telnet, banner=b"BusyBox v1.20.2 built-in shell\n\n", output=b"4.7.1"),
)
assert shell.run("getprop", ["ro.parrot.build.version"], host="h").stdout == "4.7.1"
def test_anything_after_the_end_marker_is_discarded(monkeypatch, fake_telnet):
monkeypatch.setattr(
shell.socket, "create_connection", _pty(fake_telnet, output=b"real output", trailing=b"/ # \nnoise\n")
)
assert shell.run("uptime", host="h").stdout == "real output"
def test_multi_line_output_keeps_its_blank_lines_and_indentation(monkeypatch, fake_telnet):
monkeypatch.setattr(shell.socket, "create_connection", _pty(fake_telnet, output=b"one\n\n three"))
assert shell.run("dmesg", host="h").stdout == "one\n\n three"
def test_a_command_with_no_output_returns_an_empty_string(monkeypatch, fake_telnet):
monkeypatch.setattr(shell.socket, "create_connection", _pty(fake_telnet))
assert shell.run("ls", ["/nonexistent-but-quiet"], host="h").stdout == ""
def test_the_command_we_send_is_one_simple_command(monkeypatch, fake_telnet):
monkeypatch.setattr(shell.socket, "create_connection", _pty(fake_telnet, output=b"ok"))
shell.run("ls", ["-l", "/data/ftp"], host="h")
lines = fake_telnet["sock"].sent.decode().strip().splitlines()
# Three lines: open marker, the command, close marker. stderr is merged so
# a failure explains itself; nothing else is added.
assert len(lines) == 3
assert lines[1] == "ls -l /data/ftp 2>&1"
assert lines[0].startswith("echo __MCBEBOP_") and lines[2].startswith("echo __MCBEBOP_")
def test_a_closed_port_says_how_to_open_it(monkeypatch):
def refuse(address, timeout=None):
raise ConnectionRefusedError("connection refused")
monkeypatch.setattr(shell.socket, "create_connection", refuse)
with pytest.raises(shell.ShellUnavailable, match="four times"):
shell.run("uptime", host="192.168.42.1")
def test_a_silent_shell_times_out_rather_than_hanging(monkeypatch, fake_telnet):
def connect(address, timeout=None):
sock = FakeSocket([TimeoutError])
fake_telnet["sock"] = sock
return sock
monkeypatch.setattr(shell.socket, "create_connection", connect)
with pytest.raises(shell.ShellError, match="went quiet"):
shell.run("dmesg", host="h", timeout=0.01)
def test_a_shell_that_exits_early_returns_what_arrived(monkeypatch, fake_telnet):
def connect(address, timeout=None):
sock = FakeSocket([b"partial output\n", b""])
fake_telnet["sock"] = sock
return sock
monkeypatch.setattr(shell.socket, "create_connection", connect)
assert shell.run("dmesg", host="h").stdout == "partial output"
def test_a_flood_of_output_is_capped(monkeypatch, fake_telnet):
def connect(address, timeout=None):
sock = FakeSocket([b"z" * 4096] * 100)
fake_telnet["sock"] = sock
return sock
monkeypatch.setattr(shell.socket, "create_connection", connect)
with pytest.raises(shell.ShellError, match="cap"):
shell.run("cat", ["/dev/urandom"], host="h", max_bytes=1024)
# --- telnet negotiation ------------------------------------------------------
IAC, DO, DONT, WILL, WONT, SB, SE = 255, 253, 254, 251, 252, 250, 240
ECHO_OPT, SGA = 1, 3
def test_every_option_the_server_offers_is_declined():
sock = FakeSocket([])
data = shell._answer_negotiation(sock, bytes([IAC, DO, ECHO_OPT, IAC, WILL, SGA]) + b"hello")
assert data == b"hello" # control bytes never reach the caller
assert bytes(sock.sent) == bytes([IAC, WONT, ECHO_OPT, IAC, DONT, SGA])
def test_a_subnegotiation_block_is_skipped_whole():
sock = FakeSocket([])
payload = bytes([IAC, SB, 24, 0, 65, 66, IAC, SE]) + b"after"
assert shell._answer_negotiation(sock, payload) == b"after"
assert bytes(sock.sent) == b""
def test_an_escaped_literal_ff_survives():
sock = FakeSocket([])
assert shell._answer_negotiation(sock, b"a" + bytes([IAC, IAC]) + b"b") == b"a\xffb"
def test_a_truncated_sequence_does_not_leak_a_stray_control_byte():
sock = FakeSocket([])
assert shell._answer_negotiation(sock, b"ok" + bytes([IAC, DO])) == b"ok"
assert shell._answer_negotiation(sock, b"ok" + bytes([IAC])) == b"ok"
def test_negotiation_is_stripped_out_of_real_output(monkeypatch, fake_telnet):
monkeypatch.setattr(
shell.socket,
"create_connection",
_pty(fake_telnet, negotiate=bytes([IAC, DO, ECHO_OPT]), output=b"3.4.11"),
)
assert shell.run("cat", ["/proc/version"], host="h").stdout == "3.4.11"
def test_the_module_never_imports_telnetlib():
# Removed in Python 3.13, which is why the protocol lives here.
with pytest.raises(ImportError):
__import__("telnetlib")
assert "telnetlib" not in dir(shell)
def test_we_use_the_stdlib_socket_module_directly():
assert shell.socket is socket
# --- bracketing a realistic busybox transcript -------------------------------
def test_a_busybox_transcript_is_reduced_to_the_output():
begin, end = "__MCBEBOP_dead_B__", "__MCBEBOP_dead_E__"
transcript = (
"BusyBox v1.20.2 built-in shell\n"
"\n"
f"/ # echo {begin}\n"
"/ # getprop ro.parrot.build.version 2>&1\n"
f"/ # echo {end}\n"
f"{begin}\n"
"4.7.1\n"
f"/ # {end}\n"
"/ # "
)
assert shell._between(transcript, begin, end) == "4.7.1"
def test_output_that_happens_to_look_like_a_prompt_is_kept():
begin, end = "__B_B__", "__B_E__"
assert shell._between(f"{begin}\ndata\n#\nmore\n{end}\n", begin, end) == "data\n#\nmore"
def test_without_a_begin_marker_everything_up_to_the_end_is_kept():
# An echo-less shell that somehow skipped the opening marker should still
# yield output rather than nothing.
assert shell._between("4.7.1\n__E__\n", "__B__", "__E__") == "4.7.1"
def test_a_marker_line_is_told_apart_from_its_own_echo():
assert shell._is_marker_line("/ # __M__", "__M__")
assert shell._is_marker_line("__M__", "__M__")
assert not shell._is_marker_line("echo __M__", "__M__")
assert not shell._is_marker_line("/ # echo __M__", "__M__")
def test_the_read_sentinel_waits_for_a_complete_line():
# A marker still arriving must not stop the read, or the last chunk of
# real output is lost.
assert not shell._saw_marker_line(b"out\n__M_", b"__M__")
assert not shell._saw_marker_line(b"echo __M__\n", b"__M__")
assert shell._saw_marker_line(b"out\n__M__\n", b"__M__")
tests/test_video.py
+357
View File
@@ -0,0 +1,357 @@
"""Video: SDP text, ffmpeg argv, the downscale helper, and the start ordering.
Nothing here spawns ffmpeg or touches a drone.
"""
import io
import subprocess
from pathlib import Path
import pytest
from PIL import Image
from mcbebop.media import video
def test_sdp_describes_our_own_port_not_the_drone():
# c= must be the wildcard: naming the drone makes ffmpeg wait to be sent
# to, which is backwards. The port is ours, the one we named in the
# handshake, and ffmpeg binds it.
text = video.sdp_text()
assert "c=IN IP4 0.0.0.0" in text
assert f"m=video {video.STREAM_PORT} RTP/AVP 96" in text
assert "a=rtpmap:96 H264/90000" in text
assert text.startswith("v=0\n")
assert "rtsp" not in text.lower()
def test_sdp_lines_are_in_order_and_complete():
keys = [line.split("=", 1)[0] for line in video.sdp_text().strip().splitlines()]
assert keys == ["v", "o", "s", "c", "t", "m", "a"]
def test_sdp_port_is_overridable():
assert "m=video 60000 RTP/AVP 96" in video.sdp_text(port=60000)
def test_write_sdp_creates_parents(tmp_path):
out = video.write_sdp(tmp_path / "nested" / "bebop.sdp")
assert out.read_text() == video.sdp_text()
@pytest.fixture
def sdp(tmp_path):
return tmp_path / "bebop.sdp"
def test_every_sink_whitelists_rtp_and_udp(sdp, tmp_path):
# ffmpeg refuses to open rtp/udp referenced from a file-based SDP unless
# they are explicitly whitelisted, and the failure reads like a bad file.
sinks = [
video.view(sdp),
video.record(sdp, tmp_path / "out.mp4"),
video.snapshot(sdp, tmp_path / "out.png"),
]
for sink in sinks:
argv = sink.argv
assert "-protocol_whitelist" in argv
assert argv[argv.index("-protocol_whitelist") + 1] == "file,rtp,udp"
def test_record_copies_the_stream_rather_than_re_encoding(sdp, tmp_path):
argv = video.record(sdp, tmp_path / "out.mp4").argv
assert argv[argv.index("-c") + 1] == "copy"
assert "-t" not in argv
assert argv[-1] == str(tmp_path / "out.mp4")
def test_record_with_a_duration_passes_t(sdp, tmp_path):
argv = video.record(sdp, tmp_path / "out.mp4", seconds=12.5).argv
assert argv[argv.index("-t") + 1] == "12.5"
def test_snapshot_skips_the_settle_period_and_asks_for_one_image(sdp, tmp_path):
# -ss discards the frames that reference an SPS/PPS we have not seen yet
# ("non-existing PPS 0") and lets auto-exposure settle. -update is what
# lets a single filename work at all; image2 otherwise demands %03d.
argv = video.snapshot(sdp, tmp_path / "f.png", settle_seconds=3).argv
assert argv[argv.index("-ss") + 1] == "3"
assert argv[argv.index("-frames:v") + 1] == "1"
assert argv[argv.index("-update") + 1] == "1"
# -ss has to come after -i here: we are discarding received frames, not
# seeking in a file.
assert argv.index("-i") < argv.index("-ss")
def test_snapshot_settle_default_is_not_zero(sdp, tmp_path):
argv = video.snapshot(sdp, tmp_path / "f.png").argv
assert float(argv[argv.index("-ss") + 1]) > 0
def test_view_uses_ffplay_and_low_latency_flags(sdp):
sink = video.view(sdp, title="hello")
assert sink.tool == "ffplay"
assert sink.argv[0] == "ffplay"
assert "-nobuffer" not in sink.argv # it is a value, not a flag
assert sink.argv[sink.argv.index("-fflags") + 1] == "nobuffer"
assert sink.argv[sink.argv.index("-window_title") + 1] == "hello"
def test_no_sink_opens_a_stream_url(sdp, tmp_path):
# Firmware 4.7.1 refuses port 554 entirely, so an rtsp:// URL anywhere
# here would be a regression to what pyparrot does. Match on the scheme
# rather than the bare word: pytest's own tmp_path is named after the test.
sinks = (video.view(sdp), video.record(sdp, tmp_path / "a.mp4"), video.snapshot(sdp, tmp_path / "a.png"))
for sink in sinks:
assert not any(part.startswith(("rtsp://", "rtmp://", "http://")) for part in sink.argv)
assert not any(":554" in part for part in sink.argv)
# --- downscale ---------------------------------------------------------------
def _png(width: int, height: int) -> bytes:
buf = io.BytesIO()
Image.new("RGB", (width, height), (40, 90, 140)).save(buf, format="PNG")
return buf.getvalue()
def _jpeg(width: int, height: int) -> bytes:
buf = io.BytesIO()
Image.new("RGB", (width, height), (40, 90, 140)).save(buf, format="JPEG")
return buf.getvalue()
def test_downscale_shrinks_and_keeps_aspect_ratio():
out = video.downscale(_png(856, 480), 640)
with Image.open(io.BytesIO(out)) as im:
assert im.width == 640
assert im.height == round(480 * 640 / 856)
assert im.format == "PNG"
def test_downscale_leaves_a_narrow_image_byte_for_byte():
data = _png(320, 180)
assert video.downscale(data, 640) is data
def test_downscale_at_exactly_max_width_is_a_passthrough():
data = _png(640, 360)
assert video.downscale(data, 640) is data
def test_downscale_preserves_jpeg_format():
out = video.downscale(_jpeg(856, 480), 200)
with Image.open(io.BytesIO(out)) as im:
assert im.format == "JPEG"
assert im.width == 200
def test_downscale_handles_rgba_into_jpeg():
buf = io.BytesIO()
Image.new("RGBA", (856, 480), (1, 2, 3, 255)).save(buf, format="PNG")
out = video.downscale(buf.getvalue(), 100)
with Image.open(io.BytesIO(out)) as im:
assert im.width == 100
def test_downscale_actually_saves_bytes():
big = _png(856, 480)
assert len(video.downscale(big, 160)) < len(big)
def test_downscale_rejects_a_nonsense_width():
with pytest.raises(ValueError):
video.downscale(_png(856, 480), 0)
# --- StreamSession -----------------------------------------------------------
class FakeProc:
def __init__(self, argv):
self.argv = argv
self.terminated = False
self.killed = False
self._returncode = None
def poll(self):
return self._returncode
def terminate(self):
self.terminated = True
self._returncode = 0
def kill(self):
self.killed = True
self._returncode = -9
def wait(self, timeout=None):
return 0
@pytest.fixture
def harness(monkeypatch, tmp_path):
"""Record the order of (subprocess start, command send) events."""
events: list[tuple] = []
procs: list[FakeProc] = []
def fake_popen(argv, *a, **kw):
events.append(("popen", argv))
proc = FakeProc(argv)
procs.append(proc)
return proc
monkeypatch.setattr(video.subprocess, "Popen", fake_popen)
monkeypatch.setattr(video.shutil, "which", lambda tool: f"/usr/bin/{tool}")
async def send(name, args):
events.append(("send", name, args))
return {"ok": True}
return events, procs, send
async def test_the_receiver_binds_before_the_stream_is_enabled(harness, tmp_path):
# RTP is connectionless. Enable the stream first and the opening packets,
# which carry the SPS/PPS, hit a closed port and are gone.
events, _, send = harness
sink = video.snapshot(tmp_path / "s.sdp", tmp_path / "f.png")
async with video.StreamSession(sink, send):
pass
assert events[0][0] == "popen"
assert events[1] == ("send", video.VIDEO_ENABLE, {"enable": 1})
async def test_the_stream_is_disabled_and_the_sink_stopped_on_exit(harness, tmp_path):
events, procs, send = harness
sink = video.snapshot(tmp_path / "s.sdp", tmp_path / "f.png")
async with video.StreamSession(sink, send):
pass
assert ("send", video.VIDEO_ENABLE, {"enable": 0}) in events
assert procs[0].terminated
async def test_exposure_is_set_then_restored(harness, tmp_path):
events, _, send = harness
sink = video.snapshot(tmp_path / "s.sdp", tmp_path / "f.png")
state = {video.EXPOSURE_STATE_KEY: 0.25}
async with video.StreamSession(
sink, send, exposure=1.0, read_state=lambda keys: {k: state[k] for k in keys}
):
pass
sends = [e for e in events if e[0] == "send"]
assert sends[0] == ("send", video.EXPOSURE, {"value": 1.0})
assert sends[-1] == ("send", video.EXPOSURE, {"value": 0.25})
async def test_exposure_outside_the_aircraft_range_is_refused(harness, tmp_path):
_, _, send = harness
with pytest.raises(ValueError):
await video.set_exposure(send, 3.0)
async def test_a_refused_videoenable_stops_the_subprocess(harness, tmp_path):
_, procs, _ = harness
sink = video.snapshot(tmp_path / "s.sdp", tmp_path / "f.png")
async def refusing(name, args):
raise RuntimeError("the drone did not acknowledge VideoEnable")
with pytest.raises(RuntimeError):
async with video.StreamSession(sink, refusing):
pass
assert procs and procs[0].terminated
async def test_a_missing_ffmpeg_is_reported_before_anything_starts(monkeypatch, tmp_path):
monkeypatch.setattr(video.shutil, "which", lambda tool: None)
started = []
monkeypatch.setattr(video.subprocess, "Popen", lambda *a, **kw: started.append(a))
async def send(name, args):
raise AssertionError("must not reach the drone")
sink = video.snapshot(tmp_path / "s.sdp", tmp_path / "f.png")
with pytest.raises(video.VideoUnavailable):
async with video.StreamSession(sink, send):
pass
assert not started
async def test_a_sink_that_outlives_sigterm_is_killed(harness, monkeypatch, tmp_path):
_, procs, send = harness
sink = video.view(tmp_path / "s.sdp")
class Stubborn(FakeProc):
def terminate(self):
self.terminated = True # but stays running
def wait(self, timeout=None):
if timeout is not None:
raise subprocess.TimeoutExpired("ffplay", timeout)
return 0
def popen(argv, *a, **kw):
procs.append(Stubborn(argv))
return procs[-1]
monkeypatch.setattr(video.subprocess, "Popen", popen)
async with video.StreamSession(sink, send):
pass
assert procs[-1].killed
async def test_wait_returns_none_while_the_sink_is_still_running(harness, tmp_path):
_, _, send = harness
sink = video.view(tmp_path / "s.sdp")
session = video.StreamSession(sink, send)
async with session:
session.proc.wait = lambda timeout=None: (_ for _ in ()).throw(
subprocess.TimeoutExpired("ffplay", timeout or 0)
)
assert await session.wait(0.01) is None
def test_the_module_does_not_import_the_parallel_streams():
# media/ must stay loadable while arsdk/ and protocol/ are still being
# written; it talks to them through a callable the caller supplies.
source = Path(video.__file__).read_text()
assert "import mcbebop.arsdk" not in source
assert "from mcbebop.arsdk" not in source
assert "from mcbebop.protocol" not in source
async def test_a_failed_start_puts_the_exposure_back(harness, tmp_path):
# __aexit__ never runs for a failed __aenter__, so the restore has to
# happen on the way out of __aenter__ or the next snapshot inherits it.
events, procs, _ = harness
sink = video.snapshot(tmp_path / "s.sdp", tmp_path / "f.png")
async def send(name, args):
events.append(("send", name, args))
if name == video.VIDEO_ENABLE:
raise RuntimeError("the drone did not acknowledge VideoEnable")
with pytest.raises(RuntimeError):
async with video.StreamSession(
sink, send, exposure=1.0, read_state=lambda keys: {video.EXPOSURE_STATE_KEY: -0.5}
):
pass
sends = [e for e in events if e[0] == "send"]
assert sends[0] == ("send", video.EXPOSURE, {"value": 1.0})
assert sends[-1] == ("send", video.EXPOSURE, {"value": -0.5})
assert procs[0].terminated
async def test_a_rejected_exposure_does_not_trip_the_restore_path(harness, tmp_path):
# The exposure was never changed, so there is nothing to put back and no
# NameError on the way out either.
events, procs, send = harness
sink = video.snapshot(tmp_path / "s.sdp", tmp_path / "f.png")
with pytest.raises(ValueError):
async with video.StreamSession(sink, send, exposure=99.0):
pass
assert not [e for e in events if e[0] == "send"]
assert procs[0].terminated