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