"""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()