From 0abe0a8066e66abecb912c54cabf372a64fa1629 Mon Sep 17 00:00:00 2001 From: Elijah Hauber <21344568+ethauber@users.noreply.github.com> Date: Sun, 13 Jul 2025 21:33:01 -0500 Subject: [PATCH] Add tests for issue #2621 (#3148) * Add tests for issue #2621 https://github.com/pwndbg/pwndbg/issues/2621#issuecomment-2595162380 * Update to use xuntil, use threading event, and reduce time for timeout * fix f-string mistake --- .../gdb/tests/test_command_hijack_fd.py | 301 ++++++++++++++++++ 1 file changed, 301 insertions(+) create mode 100644 tests/library/gdb/tests/test_command_hijack_fd.py diff --git a/tests/library/gdb/tests/test_command_hijack_fd.py b/tests/library/gdb/tests/test_command_hijack_fd.py new file mode 100644 index 000000000..39fbc480f --- /dev/null +++ b/tests/library/gdb/tests/test_command_hijack_fd.py @@ -0,0 +1,301 @@ +from __future__ import annotations + +import os +import socket +import tempfile +import threading +import time + +import gdb +import pytest + +import tests + +REFERENCE_BINARY = tests.get_binary("reference-binary.out") +USE_FDS_BINARY = tests.get_binary("use-fds.out") + + +class TCPServerThread(threading.Thread): + def __init__(self, *, ip: str, port: int): + super().__init__(daemon=True) + self.sock = socket.socket( + socket.AF_INET6 if ":" in ip else socket.AF_INET, socket.SOCK_STREAM + ) + try: + self.sock.bind((ip, port)) + except OSError: + pytest.skip(f"Could not bind to {ip}:{port}.") + self.port = self.sock.getsockname()[1] + self.sock.listen(1) + self.stop_event = threading.Event() + + def stop(self): + self.stop_event.set() + self.sock.close() + + def run(self): + try: + # Accept one conn and wait for stop event + conn, addr = self.sock.accept() + while not self.stop_event.is_set(): + time.sleep(0.1) + except OSError: + pass # Socket closed + + +class UDPServerThread(threading.Thread): + def __init__(self, *, ip: str, port: int): + super().__init__(daemon=True) + self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + self.sock.bind((ip, port)) + except OSError: + pytest.skip(f"Could not bind UDP to {ip}:{port}.") + self.port = self.sock.getsockname()[1] + self.received_data = None + + def stop(self): + self.sock.close() + + def run(self): + try: + # Wait for data with timeout + self.sock.settimeout(1) + data, addr = self.sock.recvfrom(1024) + self.received_data = data + except socket.timeout: + pass + except OSError: + pass + + +def test_hijack_fd_file_redirection(start_binary): + """ + Test hijack_fd command with file redirection + """ + start_binary(REFERENCE_BINARY) + + # Create a temporary file for testing + with tempfile.NamedTemporaryFile(mode="w", delete=False) as temp_file: + temp_file.write("test content for hijack_fd") + temp_file_path = temp_file.name + + try: + # Run until break_here + gdb.execute("xuntil &break_here") + + # Hijack stdout (fd 1) to point to our temporary file + result = gdb.execute(f"hijack-fd 1 {temp_file_path}", to_string=True) + assert "Operation succeeded" in result + + # Actually write to the hijacked file descriptor to validate it works + gdb.execute('call write(1, "hello\\n", 6)') + + # Check the file content to verify the write went to our file + with open(temp_file_path, "r") as f: + content = f.read() + assert "hello" in content + + # Verify the file descriptor change using procinfo + final_procinfo = gdb.execute("procinfo", to_string=True) + + # Check that fd[1] now points to our temporary file + assert temp_file_path in final_procinfo + + finally: + # Clean up temporary file + if os.path.exists(temp_file_path): + os.unlink(temp_file_path) + + +@pytest.mark.parametrize("ip_connect", ["127.0.0.1", "::1"]) +def test_hijack_fd_socket_redirection(start_binary, ip_connect): + """ + Test hijack_fd command with TCP socket redirection (IPv4 and IPv6) + """ + # Start TCP server + server = TCPServerThread(ip=ip_connect, port=0) + server.start() + + try: + start_binary(REFERENCE_BINARY) + + gdb.execute("xuntil break_here") + + # Use URI style for IPv6, plain for IPv4 + if ":" in ip_connect: + uri = f"tcp://[{ip_connect}]:{server.port}" + else: + uri = f"{ip_connect}:{server.port}" + + result = gdb.execute(f"hijack-fd 2 {uri}", to_string=True) + assert "Operation succeeded" in result + + # Verify the file descriptor change using procinfo + final_procinfo = gdb.execute("procinfo", to_string=True) + + # Check that fd[2] now shows a socket connection + if ":" in ip_connect: + assert f"[{ip_connect}]:{server.port}" in final_procinfo + else: + assert f"{ip_connect}:{server.port}" in final_procinfo + + finally: + server.stop() + server.join(timeout=2) # Wait for thread to finish + + +def test_hijack_fd_udp_socket_redirection(start_binary): + """ + Test hijack_fd command with UDP socket redirection + """ + # Start UDP server + server = UDPServerThread(ip="127.0.0.1", port=0) + server.start() + + try: + start_binary(REFERENCE_BINARY) + + gdb.execute("xuntil break_here") + + # Hijack stderr (fd 2) to point to our UDP socket + result = gdb.execute(f"hijack-fd 2 udp://127.0.0.1:{server.port}", to_string=True) + assert "Operation succeeded" in result + + # Verify the file descriptor change using procinfo + final_procinfo = gdb.execute("procinfo", to_string=True) + + # Check that fd[2] is now a socket (UDP sockets may not show IP:port in procinfo) + # The important thing is that it's a socket, not the original stderr + assert "socket:" in final_procinfo + + # Verify that fd[2] specifically is not pointing to /dev/pts/ (original stderr) + # We need to check that the line containing fd[2] doesn't contain /dev/pts/ + lines = final_procinfo.split("\n") + fd2_line = None + for line in lines: + if line.strip().startswith("fd[2]"): + fd2_line = line + break + + assert fd2_line is not None, "fd[2] not found in procinfo output" + assert "/dev/pts/" not in fd2_line, f"fd[2] still points to /dev/pts/: {fd2_line}" + + finally: + server.stop() + server.join(timeout=2) # Wait for thread to finish + + +def test_hijack_fd_invalid_fd(start_binary): + """ + Test hijack_fd command with invalid file descriptor + """ + start_binary(REFERENCE_BINARY) + + # Run until break_here + gdb.execute("xuntil break_here") + + # Try to hijack an invalid file descriptor (negative number) + result = gdb.execute("hijack-fd -1 /dev/null", to_string=True) + # The command should execute but may not work as expected + # We verify it doesn't crash and returns some response + assert result is not None + + +def test_hijack_fd_nonexistent_file(start_binary): + """ + Test hijack_fd command with nonexistent file + """ + start_binary(REFERENCE_BINARY) + + gdb.execute("xuntil break_here") + + # Use a path that can be created without root permissions + test_file_path = "/tmp/nonexistent_test_file" + + # Try to hijack to a nonexistent file + result = gdb.execute(f"hijack-fd 1 {test_file_path}", to_string=True) + # The command should succeed in creating the file descriptor, + # even if the file doesn't exist initially + assert "Operation succeeded" in result + + # Verify the file was actually created + assert os.path.exists(test_file_path) + + # Clean up + if os.path.exists(test_file_path): + os.unlink(test_file_path) + + +def test_hijack_fd_invalid_socket_address(start_binary): + """ + Test hijack_fd command with invalid socket address + """ + start_binary(REFERENCE_BINARY) + + gdb.execute("xuntil break_here") + + # Try to hijack to an invalid socket address + result = gdb.execute("hijack-fd 2 invalid://address:port", to_string=True) + # The command should handle the error gracefully + # We verify it doesn't crash the debugger and returns some response + assert result is not None + + +def test_hijack_fd_before_binary_start(): + """ + Test hijack_fd command before binary is started + """ + # Try to use hijack_fd before starting any binary + result = gdb.execute("hijack-fd 1 /dev/null", to_string=True) + assert "The program is not being run" in result + + +def test_hijack_fd_help(): + """ + Test hijack_fd command help + """ + result = gdb.execute("hijack-fd --help", to_string=True) + assert "usage: hijack-fd" in result + assert "Replace a file descriptor" in result + assert "fdnum" in result + assert "newfile" in result + + +def test_hijack_fd_with_use_fds_binary(start_binary): + """ + Test hijack_fd command with the use-fds binary which opens a file + """ + start_binary(USE_FDS_BINARY) + + # Run until main + gdb.execute("start") + + # Stop after the open() call + gdb.execute("nextcall") + gdb.execute("nextcall") + + # Get the file descriptor number + fd_var = gdb.newest_frame().read_var("fd") + fd_num = int(fd_var) + + # Create a temporary file for testing + with tempfile.NamedTemporaryFile(mode="w", delete=False) as temp_file: + temp_file.write("hijacked content") + temp_file_path = temp_file.name + + try: + # Hijack the opened file descriptor to our temporary file + result = gdb.execute(f"hijack-fd {fd_num} {temp_file_path}", to_string=True) + assert "Operation succeeded" in result + + # Verify the file descriptor change using procinfo + final_procinfo = gdb.execute("procinfo", to_string=True) + + # Check that the hijacked fd now points to our temporary file + assert temp_file_path in final_procinfo + + finally: + # Clean up temporary file + if os.path.exists(temp_file_path): + os.unlink(temp_file_path)