Use target remote for remote connections

tests/api/gef_memory.py

@@ -12,6 +12,7 @@
     ARCH,
     debug_target,
     gdbserver_session,
+    get_random_port,
     qemuuser_session,
     GDBSERVER_DEFAULT_HOST,
 )
@@ -121,48 +122,35 @@ def test_func_parse_maps_local_procfs(self):
     @pytest.mark.slow
     def test_func_parse_maps_remote_gdbserver(self):
         gef, gdb = self._gef, self._gdb
-        # When in a gef-remote session `parse_gdb_info_proc_maps` should work to
+        # When in a remote session `parse_gdb_info_proc_maps` should work to
         # query the memory maps
-        while True:
-            port = random.randint(1025, 65535)
-            if port != self._port:
-                break
+        port = get_random_port()
 
         with pytest.raises(Exception):
-            gdb.execute(f"gef-remote {GDBSERVER_DEFAULT_HOST} {port}")
+            gdb.execute(f"target remote :{port}")
 
         with gdbserver_session(port=port) as _:
-            gdb.execute(f"gef-remote {GDBSERVER_DEFAULT_HOST} {port}")
+            gdb.execute(f"target remote :{port}")
             sections = gef.memory.maps
             assert len(sections) > 0
 
     @pytest.mark.skipif(ARCH not in ("x86_64",), reason=f"Skipped for {ARCH}")
     def test_func_parse_maps_remote_qemu(self):
         gdb, gef = self._gdb, self._gef
-        # When in a gef-remote qemu-user session `parse_gdb_info_proc_maps`
-        # should work to query the memory maps
-        while True:
-            port = random.randint(1025, 65535)
-            if port != self._port:
-                break
+        port = get_random_port()
 
         with qemuuser_session(port=port) as _:
-            cmd = f"gef-remote --qemu-user --qemu-binary {self._target} {GDBSERVER_DEFAULT_HOST} {port}"
+            cmd = f"target remote :{port}"
             gdb.execute(cmd)
             sections = gef.memory.maps
             assert len(sections) > 0
 
     def test_func_parse_maps_realpath(self):
         gef, gdb = self._gef, self._gdb
-        # When in a gef-remote session `parse_gdb_info_proc_maps` should work to
-        # query the memory maps
-        while True:
-            port = random.randint(1025, 65535)
-            if port != self._port:
-                break
+        port = get_random_port()
 
         with gdbserver_session(port=port) as _:
-            gdb.execute(f"gef-remote {GDBSERVER_DEFAULT_HOST} {port}")
+            gdb.execute(f"target remote :{port}")
             gdb.execute("b main")
             gdb.execute("continue")
             sections = gef.memory.maps

tests/api/gef_remote.py

@@ -0,0 +1,108 @@
+"""
+`target remote/extended-remote` test module.
+"""
+
+
+from tests.base import RemoteGefUnitTestGeneric
+from tests.utils import debug_target, gdbserver_session, gdbserver_multi_session, get_random_port, qemuuser_session
+
+
+class GefRemoteApi(RemoteGefUnitTestGeneric):
+
+    def setUp(self) -> None:
+        self._target = debug_target("default")
+        return super().setUp()
+
+    def test_gef_remote_test_gdbserver(self):
+        """Test `gdbserver file`"""
+        _gdb = self._gdb
+        _gef = self._gef
+        _root = self._conn.root
+        port = get_random_port()
+
+        with gdbserver_session(port=port):
+            assert not _root.eval("is_target_remote()")
+            assert not _root.eval("is_target_remote_or_extended()")
+            assert not _root.eval("is_running_in_gdbserver()")
+            assert not _root.eval("is_running_in_rr()")
+            assert not _root.eval("is_running_in_qemu()")
+
+            _gdb.execute(f"target remote :{port}")
+
+            assert _root.eval("is_target_remote()")
+            assert _root.eval("is_target_remote_or_extended()")
+            assert _root.eval("is_running_in_gdbserver()")
+            assert _root.eval("is_running_in_gdbserver()")
+
+            assert not _root.eval("is_target_extended_remote()")
+            assert not _root.eval("is_running_in_qemu()")
+            assert not _root.eval("is_running_in_qemu_system()")
+            assert not _root.eval("is_running_in_qemu_user()")
+            assert not _root.eval("is_running_in_rr()")
+
+            assert hasattr(_gef.session, "remote")
+            assert "GDBSERVER" in str(_gef.session.remote)
+            assert "GDBSERVER_MULTI" not in str(_gef.session.remote)
+
+    def test_gef_remote_test_gdbserver_multi(self):
+        """Test `gdbserver --multi file`"""
+        _gdb = self._gdb
+        _gef = self._gef
+        _root = self._conn.root
+        port = get_random_port()
+
+        with gdbserver_multi_session(port=port):
+            assert not _root.eval("is_target_remote()")
+            assert not _root.eval("is_target_remote_or_extended()")
+            assert not _root.eval("is_running_in_gdbserver()")
+            assert not _root.eval("is_running_in_rr()")
+            assert not _root.eval("is_running_in_qemu()")
+
+            _gdb.execute(f"target extended-remote :{port}")
+            _gdb.execute(f"set remote exec-file {self._target}")
+            _gdb.execute(f"file {self._target}")
+            _gdb.execute(f"start {self._target}")
+
+            assert _root.eval("is_target_remote_or_extended()")
+            assert _root.eval("is_target_extended_remote()")
+            assert _root.eval("is_running_in_gdbserver()")
+            assert not _root.eval("is_target_remote()")
+
+            assert not _root.eval("is_running_in_qemu()")
+            assert not _root.eval("is_running_in_qemu_system()")
+            assert not _root.eval("is_running_in_qemu_user()")
+            assert not _root.eval("is_running_in_rr()")
+
+            assert hasattr(_gef.session, "remote")
+            assert "GDBSERVER_MULTI" in str(_gef.session.remote)
+
+    def test_gef_remote_test_qemuuser(self):
+        """Test `qemu-user -g`"""
+        _gdb = self._gdb
+        _gef = self._gef
+        _root = self._conn.root
+        port = get_random_port()
+
+        with qemuuser_session(port=port):
+            assert not _root.eval("is_target_remote()")
+            assert not _root.eval("is_target_remote_or_extended()")
+            assert not _root.eval("is_running_in_gdbserver()")
+
+            _gdb.execute(f"target remote :{port}")
+
+            assert _root.eval("is_target_remote()")
+            assert _root.eval("is_target_remote_or_extended()")
+            assert _root.eval("is_running_in_qemu()")
+            assert _root.eval("is_running_in_qemu_user()")
+
+            assert not _root.eval("is_target_extended_remote()")
+            assert not _root.eval("is_running_in_qemu_system()")
+            assert not _root.eval("is_running_in_gdbserver()")
+            assert not _root.eval("is_running_in_rr()")
+
+            assert hasattr(_gef.session, "remote")
+            assert "QEMU_USER" in str(_gef.session.remote)
+
+    # TODO add tests for
+    #  - [ ] qemu-system
+    #  - [ ] rr

tests/api/gef_session.py

@@ -14,6 +14,7 @@
     ARCH,
     debug_target,
     gdbserver_session,
+    get_random_port,
     qemuuser_session,
     GDBSERVER_DEFAULT_HOST,
 )
@@ -63,12 +64,11 @@ def test_root_dir_local(self):
     def test_root_dir_remote(self):
         gdb = self._gdb
         gdb.execute("start")
-
         expected = os.stat("/")
-        host = GDBSERVER_DEFAULT_HOST
-        port = random.randint(1025, 65535)
+        port = get_random_port()
+
         with gdbserver_session(port=port):
-            gdb.execute(f"gef-remote {host} {port}")
+            gdb.execute(f"target remote :{port}")
             result = self._conn.root.eval("os.stat(gef.session.root)")
             assert (expected.st_dev == result.st_dev) and (
                 expected.st_ino == result.st_ino
@@ -77,11 +77,8 @@ def test_root_dir_remote(self):
     @pytest.mark.skipif(ARCH not in ("x86_64",), reason=f"Skipped for {ARCH}")
     def test_root_dir_qemu(self):
         gdb, gef = self._gdb, self._gef
+        port = get_random_port()
 
-        host = GDBSERVER_DEFAULT_HOST
-        port = random.randint(1025, 65535)
         with qemuuser_session(port=port):
-            gdb.execute(
-                f"gef-remote --qemu-user --qemu-binary {self._target} {host} {port}"
-            )
+            gdb.execute(f"target remote :{port}")
             assert re.search(r"\/proc\/[0-9]+/root", str(gef.session.root))

tests/base.py

@@ -10,7 +10,7 @@
 
 import rpyc
 
-from .utils import debug_target
+from .utils import debug_target, get_random_port
 
 COVERAGE_DIR = os.getenv("COVERAGE_DIR", "")
 GEF_PATH = pathlib.Path(os.getenv("GEF_PATH", "gef.py")).absolute()
@@ -58,7 +58,7 @@ def __setup(self):
         #
         # Select a random tcp port for rpyc
         #
-        self._port = random.randint(1025, 65535)
+        self._port = get_random_port()
         self._commands = ""
 
         if COVERAGE_DIR:

tests/commands/gef_remote.py

@@ -11,6 +11,7 @@
     ARCH,
     debug_target,
     gdbserver_session,
+    get_random_port,
     qemuuser_session,
     GDBSERVER_DEFAULT_HOST,
 )
@@ -26,50 +27,40 @@ def setUp(self) -> None:
     def test_cmd_gef_remote_gdbserver(self):
         gdb = self._gdb
         gef = self._gef
-        root = self._conn.root
+        port = get_random_port()
         gdbserver_mode = "GDBSERVER"
-        while True:
-            port = random.randint(1025, 65535)
-            if port != self._port:
-                break
 
         with gdbserver_session(port=port):
-            gdb.execute(f"gef-remote {GDBSERVER_DEFAULT_HOST} {port}")
-            res: str = root.eval("str(gef.session.remote)")
-            assert res.startswith(f"RemoteSession(target='{GDBSERVER_DEFAULT_HOST}:{port}', local='/tmp/")
-            assert res.endswith(f"pid={gef.session.pid}, mode={gdbserver_mode})")
+            gdb.execute(f"target remote {GDBSERVER_DEFAULT_HOST}:{port}")
+            res: str = str(gef.session.remote)
+            assert res.startswith(f"RemoteSession(target='{GDBSERVER_DEFAULT_HOST}:{port}', local='/")
+            assert res.endswith(f"mode={gdbserver_mode}, pid={gef.session.pid})")
 
     @pytest.mark.slow
     @pytest.mark.skipif(ARCH not in ("x86_64",), reason=f"Skipped for {ARCH}")
     def test_cmd_gef_remote_qemu_user(self):
         gdb = self._gdb
         gef = self._gef
-        root = self._conn.root
-        qemu_mode = "QEMU"
-        while True:
-            port = random.randint(1025, 65535)
-            if port != self._port:
-                break
+        qemu_mode = "QEMU_USER"
+        port = get_random_port()
 
         with qemuuser_session(port=port):
-            cmd = f"gef-remote --qemu-user --qemu-binary {self._target} {GDBSERVER_DEFAULT_HOST} {port}"
+            cmd = f"target remote {GDBSERVER_DEFAULT_HOST}:{port}"
             gdb.execute(cmd)
-            res = root.eval("str(gef.session.remote)")
-            assert res.startswith(f"RemoteSession(target='{GDBSERVER_DEFAULT_HOST}:{port}', local='/tmp/")
-            assert res.endswith(f"pid={gef.session.pid}, mode={qemu_mode})")
+            res = str(gef.session.remote)
+            assert res.startswith(f"RemoteSession(target='{GDBSERVER_DEFAULT_HOST}:{port}', local='/")
+            assert res.endswith(f"mode={qemu_mode})")
 
     def test_cmd_target_remote(self):
         gdb = self._gdb
         gef = self._gef
-        root = self._conn.root
         gdbserver_mode = "GDBSERVER"
-        while True:
-            port = random.randint(1025, 65535)
-            if port != self._port:
-                break
+        port = get_random_port()
 
         with gdbserver_session(port=port) as _:
             gdb.execute(f"target remote {GDBSERVER_DEFAULT_HOST}:{port}")
-            res: str = root.eval("str(gef.session.remote)")
-            assert res.startswith("RemoteSession(target=':0', local='/tmp/")
-            assert res.endswith(f"pid={gef.session.pid}, mode={gdbserver_mode})")
+            res: str = str(gef.session.remote)
+            assert res.startswith(
+                f"RemoteSession(target='{GDBSERVER_DEFAULT_HOST}:{port}', local='/"
+            )
+            assert res.endswith(f"mode={gdbserver_mode}, pid={gef.session.pid})")

tests/regressions/gdbserver_connection.py

@@ -10,8 +10,8 @@ def test_can_establish_connection_to_gdbserver_again_after_disconnect(self):
         gdb = self._gdb
 
         with gdbserver_session(port=5001) as _, gdbserver_session(port=5002) as _:
-            gdb.execute("gef-remote 127.0.0.1 5001")
+            gdb.execute("target remote :5001")
             gdb.execute("detach")
 
-            gdb.execute("gef-remote 127.0.0.1 5002")
+            gdb.execute("target remote :5002")
             gdb.execute("continue")

tests/utils.py

@@ -12,6 +12,7 @@
 import subprocess
 import tempfile
 import time
+import random
 
 from typing import Iterable, List, Optional, Union
 from urllib.request import urlopen
@@ -41,7 +42,8 @@ def which(program: str) -> pathlib.Path:
 STRIP_ANSI_DEFAULT = True
 GDBSERVER_DEFAULT_HOST = "localhost"
 GDBSERVER_DEFAULT_PORT = 1234
-GDBSERVER_BINARY = which("gdbserver")
+GDBSERVER_BINARY: pathlib.Path = which("gdbserver")
+GDBSERVER_STARTUP_DELAY_SEC : float = 0.5
 assert GDBSERVER_BINARY.exists()
 
 QEMU_USER_X64_BINARY = which("qemu-x86_64")
@@ -112,6 +114,13 @@ def start_gdbserver(
     logging.debug(f"Starting {cmd}")
     return subprocess.Popen(cmd)
 
+def start_gdbserver_multi(
+    host: str = GDBSERVER_DEFAULT_HOST,
+    port: int = GDBSERVER_DEFAULT_PORT,
+) -> subprocess.Popen:
+    cmd = [GDBSERVER_BINARY, "--multi", f"{host}:{port}"]
+    logging.debug(f"Starting {cmd}")
+    return subprocess.Popen(cmd)
 
 def stop_gdbserver(gdbserver: subprocess.Popen) -> None:
     """Stop the gdbserver and wait until it is terminated if it was
@@ -133,11 +142,22 @@ def gdbserver_session(
 ):
     sess = start_gdbserver(exe, host, port)
     try:
-        time.sleep(1)  # forced delay to allow gdbserver to start listening
+        time.sleep(GDBSERVER_STARTUP_DELAY_SEC)
         yield sess
     finally:
         stop_gdbserver(sess)
 
+@contextlib.contextmanager
+def gdbserver_multi_session(
+    port: int = GDBSERVER_DEFAULT_PORT,
+    host: str = GDBSERVER_DEFAULT_HOST,
+):
+    sess = start_gdbserver_multi(host, port)
+    try:
+        time.sleep(GDBSERVER_STARTUP_DELAY_SEC)
+        yield sess
+    finally:
+        stop_gdbserver(sess)
 
 def start_qemuuser(
     exe: Union[str, pathlib.Path] = debug_target("default"),
@@ -301,3 +321,14 @@ def p32(x: int) -> bytes:
 
 def p64(x: int) -> bytes:
     return struct.pack("<Q", x)
+
+
+__available_ports = list()
+def get_random_port() -> int:
+    global __available_ports
+    if len(__available_ports) < 2:
+        __available_ports = list( range(1024, 65535) )
+    idx = random.choice(range(len(__available_ports)))
+    port = __available_ports[idx]
+    __available_ports.pop(idx)
+    return port