Fix support for rr

.github/workflows/validate.yml

@@ -11,9 +11,9 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v3
-    - uses: actions/setup-python@v3
+    - uses: actions/setup-python@v5.0.0
       with:
-        python-version: "3.11"
+        python-version: "3.8"
     - uses: pre-commit/action@v3.0.0
 
   docs_link_check:
@@ -23,9 +23,9 @@ jobs:
       contents: read
     steps:
       - name: checkout
-        uses: actions/checkout@v2
+        uses: actions/checkout@v4
       - name: Check links
-        uses: lycheeverse/lychee-action@v1.4.1
+        uses: lycheeverse/lychee-action@v1.9.1
         env:
           GITHUB_TOKEN: ${{secrets.GITHUB_TOKEN}}
         with:

gef.py

@@ -682,7 +682,7 @@ def is_executable(self) -> bool:
     @property
     def size(self) -> int:
         if self.page_end is None or self.page_start is None:
-            return -1
+            raise AttributeError
         return self.page_end - self.page_start
 
     @property
@@ -691,16 +691,17 @@ def realpath(self) -> str:
         return self.path if gef.session.remote is None else f"/tmp/gef/{gef.session.remote:d}/{self.path}"
 
     def __str__(self) -> str:
-        return (f"Section(page_start={self.page_start:#x}, page_end={self.page_end:#x}, "
-                f"permissions={self.permission!s})")
+        return (f"Section(start={self.page_start:#x}, end={self.page_end:#x}, "
+                f"perm={self.permission!s})")
+
+    def __repr__(self) -> str:
+        return str(self)
 
     def __eq__(self, other: "Section") -> bool:
         return other and \
             self.page_start == other.page_start and \
-            self.page_end == other.page_end and \
-            self.offset == other.offset and \
+            self.size == other.size and \
             self.permission == other.permission and \
-            self.inode == other.inode and \
             self.path == other.path
 
 
@@ -6041,15 +6042,16 @@ def do_invoke(self, _: List[str], **kwargs: Any) -> None:
         # calls `is_remote_debug` which checks if `remote_initializing` is True or `.remote` is None
         # This prevents some spurious errors being thrown during startup
         gef.session.remote_initializing = True
-        gef.session.remote = GefRemoteSessionManager(args.host, args.port, args.pid, qemu_binary)
+        session = GefRemoteSessionManager(args.host, args.port, args.pid, qemu_binary)
 
-        dbg(f"[remote] initializing remote session with {gef.session.remote.target} under {gef.session.remote.root}")
-        if not gef.session.remote.connect(args.pid):
-            raise EnvironmentError(f"Cannot connect to remote target {gef.session.remote.target}")
-        if not gef.session.remote.setup():
-            raise EnvironmentError(f"Failed to create a proper environment for {gef.session.remote.target}")
+        dbg(f"[remote] initializing remote session with {session.target} under {session.root}")
+        if not session.connect(args.pid) or not session.setup():
+            gef.session.remote = None
+            gef.session.remote_initializing = False
+            raise EnvironmentError("Failed to setup remote target")
 
         gef.session.remote_initializing = False
+        gef.session.remote = session
         reset_all_caches()
         gdb.execute("context")
         return
@@ -6060,7 +6062,7 @@ class SkipiCommand(GenericCommand):
     """Skip N instruction(s) execution"""
 
     _cmdline_ = "skipi"
-    _syntax_  = ("{_cmdline_} [LOCATION] [--n NUM_INSTRUCTIONS]"
+    _syntax_  = (f"{_cmdline_} [LOCATION] [--n NUM_INSTRUCTIONS]"
                 "\n\tLOCATION\taddress/symbol from where to skip"
                  "\t--n NUM_INSTRUCTIONS\tSkip the specified number of instructions instead of the default 1.")
 
@@ -6095,7 +6097,7 @@ class NopCommand(GenericCommand):
     aware."""
 
     _cmdline_ = "nop"
-    _syntax_  = ("{_cmdline_} [LOCATION] [--i ITEMS] [--f] [--n] [--b]"
+    _syntax_  = (f"{_cmdline_} [LOCATION] [--i ITEMS] [--f] [--n] [--b]"
                  "\n\tLOCATION\taddress/symbol to patch (by default this command replaces whole instructions)"
                  "\t--i ITEMS\tnumber of items to insert (default 1)"
                  "\t--f\tForce patch even when the selected settings could overwrite partial instructions"
@@ -10460,10 +10462,12 @@ def read_ascii_string(self, address: int) -> Optional[str]:
     def maps(self) -> List[Section]:
         if not self.__maps:
             self.__maps = self._parse_maps()
+            if not self.__maps:
+                raise RuntimeError("Failed to get memory layout")
         return self.__maps
 
     @classmethod
-    def _parse_maps(cls) -> List[Section]:
+    def _parse_maps(cls) -> Optional[List[Section]]:
         """Return the mapped memory sections. If the current arch has its maps
         method defined, then defer to that to generated maps, otherwise, try to
         figure it out from procfs, then info sections, then monitor info
@@ -10472,12 +10476,12 @@ def _parse_maps(cls) -> List[Section]:
             return list(gef.arch.maps())
 
         try:
-            return list(cls.parse_procfs_maps())
+            return list(cls.parse_gdb_info_proc_maps())
         except:
             pass
 
         try:
-            return list(cls.parse_gdb_info_sections())
+            return list(cls.parse_procfs_maps())
         except:
             pass
 
@@ -10486,7 +10490,6 @@ def _parse_maps(cls) -> List[Section]:
         except:
             pass
 
-        warn("Cannot get memory map")
         return None
 
     @staticmethod
@@ -10521,30 +10524,27 @@ def parse_procfs_maps() -> Generator[Section, None, None]:
         return
 
     @staticmethod
-    def parse_gdb_info_sections() -> Generator[Section, None, None]:
+    def parse_gdb_info_proc_maps() -> Generator[Section, None, None]:
         """Get the memory mapping from GDB's command `maintenance info sections` (limited info)."""
-        stream = StringIO(gdb.execute("maintenance info sections", to_string=True))
-
-        for line in stream:
+        lines = (gdb.execute("info proc mappings", to_string=True) or "").splitlines()
+        if len(lines) < 5:
+            # See expected format in tests/api/gef_memory.py:test_api_gef_memory_parse_info_proc_maps*
+            return
+        for line in lines[4:]:
             if not line:
                 break
 
-            try:
-                parts = [x for x in line.split()]
-                addr_start, addr_end = [int(x, 16) for x in parts[1].split("->")]
-                off = int(parts[3][:-1], 16)
-                path = parts[4]
-                perm = Permission.from_info_sections(parts[5:])
-                yield Section(page_start=addr_start,
-                              page_end=addr_end,
-                              offset=off,
-                              permission=perm,
-                              path=path)
-
-            except IndexError:
-                continue
-            except ValueError:
-                continue
+            parts = [x.strip() for x in line.split()]
+            addr_start, addr_end, offset, _ = list(map(lambda x: int(x, 16), parts[0:4]))
+            perm = Permission.from_process_maps(parts[4])
+            path = " ".join(parts[5:]) if len(parts) >= 5 else ""
+            yield Section(
+                page_start=addr_start,
+                page_end=addr_end,
+                offset=offset,
+                permission=perm,
+                path=path,
+            )
         return
 
     @staticmethod
@@ -10560,7 +10560,7 @@ def parse_monitor_info_mem() -> Generator[Section, None, None]:
                 ranges, off, perms = line.split()
                 off = int(off, 16)
                 start, end = [int(s, 16) for s in ranges.split("-")]
-            except ValueError as e:
+            except ValueError:
                 continue
 
             perm = Permission.from_monitor_info_mem(perms)
@@ -10877,6 +10877,9 @@ def reset_caches(self) -> None:
     def __str__(self) -> str:
         return f"Session({'Local' if self.remote is None else 'Remote'}, pid={self.pid or 'Not running'}, os='{self.os}')"
 
+    def __repr__(self) -> str:
+        return str(self)
+
     @property
     def auxiliary_vector(self) -> Optional[Dict[str, int]]:
         if not is_alive():
@@ -10993,6 +10996,12 @@ def root(self) -> Optional[pathlib.Path]:
 class GefRemoteSessionManager(GefSessionManager):
     """Class for managing remote sessions with GEF. It will create a temporary environment
     designed to clone the remote one."""
+
+    class RemoteMode(enum.IntEnum):
+        GDBSERVER = 0
+        QEMU = 1
+        RR = 2
+
     def __init__(self, host: str, port: int, pid: int =-1, qemu: Optional[pathlib.Path] = None) -> None:
         super().__init__()
         self.__host = host
@@ -11014,7 +11023,10 @@ def in_qemu_user(self) -> bool:
         return self.__qemu is not None
 
     def __str__(self) -> str:
-        return f"RemoteSession(target='{self.target}', local='{self.root}', pid={self.pid}, qemu_user={bool(self.in_qemu_user())})"
+        return f"RemoteSession(target='{self.target}', local='{self.root}', pid={self.pid}, mode={self.mode})"
+
+    def __repr__(self) -> str:
+        return str(self)
 
     @property
     def target(self) -> str:
@@ -11028,11 +11040,11 @@ def root(self) -> pathlib.Path:
     def file(self) -> pathlib.Path:
         """Path to the file being debugged as seen by the remote endpoint."""
         if not self._file:
-            filename = gdb.current_progspace().filename
-            if not filename:
+            progspace = gdb.current_progspace()
+            if not progspace or not progspace.filename:
                 raise RuntimeError("No session started")
-            start_idx = len("target:") if filename.startswith("target:") else 0
-            self._file = pathlib.Path(filename[start_idx:])
+            start_idx = len("target:") if progspace.filename.startswith("target:") else 0
+            self._file = pathlib.Path(progspace.filename[start_idx:])
         return self._file
 
     @property
@@ -11046,6 +11058,14 @@ def maps(self) -> pathlib.Path:
             self._maps = self.root / f"proc/{self.pid}/maps"
         return self._maps
 
+    @property
+    def mode(self) -> RemoteMode:
+        if self.in_qemu_user():
+            return GefRemoteSessionManager.RemoteMode.QEMU
+        if os.environ.get("GDB_UNDER_RR", None) == "1":
+            return GefRemoteSessionManager.RemoteMode.RR
+        return GefRemoteSessionManager.RemoteMode.GDBSERVER
+
     def sync(self, src: str, dst: Optional[str] = None) -> bool:
         """Copy the `src` into the temporary chroot. If `dst` is provided, that path will be
         used instead of `src`."""
@@ -11063,7 +11083,12 @@ def connect(self, pid: int) -> bool:
         """Connect to remote target. If in extended mode, also attach to the given PID."""
         # before anything, register our new hook to download files from the remote target
         dbg(f"[remote] Installing new objfile handlers")
-        gef_on_new_unhook(new_objfile_handler)
+        try:
+            gef_on_new_unhook(new_objfile_handler)
+        except SystemError:
+            # the default objfile handler might already have been removed, ignore failure
+            pass
+
         gef_on_new_hook(self.remote_objfile_event_handler)
 
         # then attempt to connect
@@ -11085,13 +11110,17 @@ def connect(self, pid: int) -> bool:
 
     def setup(self) -> bool:
         # setup remote adequately depending on remote or qemu mode
-        if self.in_qemu_user():
+        if self.mode == GefRemoteSessionManager.RemoteMode.QEMU:
             dbg(f"Setting up as qemu session, target={self.__qemu}")
             self.__setup_qemu()
-        else:
+        elif self.mode == GefRemoteSessionManager.RemoteMode.RR:
+            dbg(f"Setting up as rr session")
+            self.__setup_rr()
+        elif self.mode == GefRemoteSessionManager.RemoteMode.GDBSERVER:
             dbg(f"Setting up as remote session")
             self.__setup_remote()
-
+        else:
+            raise Exception
         # refresh gef to consider the binary
         reset_all_caches()
         gef.binary = Elf(self.lfile)
@@ -11155,6 +11184,14 @@ def __setup_remote(self) -> bool:
                 fd.write(f"{mem_range} rwxp 00000000 00:00 0                    {fname}\n")
         return True
 
+    def __setup_rr(self) -> bool:
+        #
+        # Simply override the local root path, the binary must exist
+        # on the host.
+        #
+        self.__local_root_path = pathlib.Path("/")
+        return True
+
     def remote_objfile_event_handler(self, evt: "gdb.NewObjFileEvent") -> None:
         dbg(f"[remote] in remote_objfile_handler({evt.new_objfile.filename if evt else 'None'}))")
         if not evt or not evt.new_objfile.filename: