gh-148178: Validate remote debug offset tables on load

Treat the debug offset tables read from a target process as untrusted input
and validate them before the unwinder uses any reported sizes or offsets.

Add a shared validator in debug_offsets_validation.h and run it once when
_Py_DebugOffsets is loaded and once when AsyncioDebug is loaded. The checks
cover section sizes used for fixed local buffers and every offset that is
later dereferenced against a local buffer or local object view. This keeps
the bounds checks out of the sampling hot path while rejecting malformed
tables up front.

Also add Linux poison tests that corrupt both the main and asyncio offset
tables and assert that RemoteUnwinder fails with an invalid-offset error
instead of walking into bad accesses or misleading late failures.

Author: pablogsal

Lib/test/test_external_inspection.py

@@ -1,5 +1,6 @@
 import unittest
 import os
+import json
 import textwrap
 import contextlib
 import importlib
@@ -418,6 +419,164 @@ def _frame_to_lineno_tuple(frame):
         filename, location, funcname, opcode = frame
         return (filename, location.lineno, funcname, opcode)
 
+    @contextmanager
+    def _poisoned_debug_offsets_process(self, poison_code):
+        script = (
+            textwrap.dedent(
+                """\
+                import json
+                import os
+                import struct
+                import sys
+                import threading
+                import time
+
+                cookie = b"xdebugpy"
+                pid = os.getpid()
+
+                def find_section_address(section_name, filename_substr):
+                    ehdr_fmt = "<16sHHIQQQIHHHHHH"
+                    phdr_fmt = "<IIQQQQQQ"
+                    shdr_fmt = "<IIQQQQIIQQ"
+
+                    def search_elf_file_for_section(start_address, elf_file):
+                        with open(elf_file, "rb") as f:
+                            data = f.read()
+                        if data[:4] != b"\\x7fELF" or data[4] != 2 or data[5] != 1:
+                            raise RuntimeError(f"unsupported ELF file: {elf_file}")
+
+                        ehdr = struct.unpack_from(ehdr_fmt, data, 0)
+                        e_phoff = ehdr[5]
+                        e_shoff = ehdr[6]
+                        e_phentsize = ehdr[9]
+                        e_phnum = ehdr[10]
+                        e_shentsize = ehdr[11]
+                        e_shnum = ehdr[12]
+                        e_shstrndx = ehdr[13]
+
+                        shstr = struct.unpack_from(
+                            shdr_fmt, data, e_shoff + e_shstrndx * e_shentsize
+                        )
+                        shstrtab = data[shstr[4]:shstr[4] + shstr[5]]
+
+                        section = None
+                        for i in range(e_shnum):
+                            shdr = struct.unpack_from(
+                                shdr_fmt, data, e_shoff + i * e_shentsize
+                            )
+                            name_end = shstrtab.find(b"\\0", shdr[0])
+                            name = shstrtab[shdr[0]:name_end].decode()
+                            if name.lstrip(".") == section_name:
+                                section = shdr
+                                break
+                        if section is None:
+                            return 0
+
+                        first_load = None
+                        for i in range(e_phnum):
+                            phdr = struct.unpack_from(
+                                phdr_fmt, data, e_phoff + i * e_phentsize
+                            )
+                            if phdr[0] == 1:
+                                first_load = phdr
+                                break
+                        if first_load is None:
+                            raise RuntimeError(f"no PT_LOAD segment in {elf_file}")
+
+                        p_vaddr = first_load[3]
+                        p_align = first_load[7] or 1
+                        elf_load_addr = p_vaddr - (p_vaddr % p_align)
+                        return start_address + section[3] - elf_load_addr
+
+                    with open(f"/proc/{pid}/maps") as f:
+                        for line in f:
+                            parts = line.split(None, 5)
+                            if len(parts) < 6:
+                                continue
+                            path = parts[5].strip()
+                            if path.startswith("["):
+                                continue
+                            if filename_substr not in os.path.basename(path):
+                                continue
+                            start = int(parts[0].split("-")[0], 16)
+                            try:
+                                addr = search_elf_file_for_section(start, path)
+                            except OSError:
+                                continue
+                            if addr:
+                                return addr
+                    raise RuntimeError(f"{section_name} section not found")
+
+                def find_debug_offsets():
+                    with open(f"/proc/{pid}/maps") as f:
+                        for line in f:
+                            parts = line.split()
+                            if len(parts) < 2 or "rw" not in parts[1]:
+                                continue
+                            start, end = (int(x, 16) for x in parts[0].split("-"))
+                            if end - start > 10_000_000:
+                                continue
+                            try:
+                                fd = os.open(f"/proc/{pid}/mem", os.O_RDONLY)
+                                os.lseek(fd, start, 0)
+                                data = os.read(fd, end - start)
+                                os.close(fd)
+                            except OSError:
+                                continue
+                            off = data.find(cookie)
+                            if off == -1:
+                                continue
+                            version = struct.unpack_from("<Q", data, off + 8)[0]
+                            if ((version >> 24) & 0xFF) != sys.version_info.major:
+                                continue
+                            return start + off
+                    raise RuntimeError("debug offsets not found")
+
+                addr = find_debug_offsets()
+                """
+            )
+            + textwrap.dedent(poison_code)
+            + "\n"
+            + textwrap.dedent(
+                """\
+                print(
+                    json.dumps({"pid": pid, "native_tid": threading.get_native_id()}),
+                    flush=True,
+                )
+                time.sleep(60)
+                """
+            )
+        )
+
+        proc = subprocess.Popen(
+            [sys.executable, "-c", script],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+        )
+        try:
+            line = proc.stdout.readline()
+            if not line:
+                stderr = proc.stderr.read()
+                self.fail(
+                    "poisoned child failed to initialize: "
+                    f"{stderr.strip() or 'no stderr output'}"
+                )
+            yield proc, json.loads(line)
+        finally:
+            try:
+                proc.terminate()
+                proc.wait(timeout=SHORT_TIMEOUT)
+            except subprocess.TimeoutExpired:
+                proc.kill()
+                proc.wait(timeout=SHORT_TIMEOUT)
+            except OSError:
+                pass
+            if proc.stdout is not None:
+                proc.stdout.close()
+            if proc.stderr is not None:
+                proc.stderr.close()
+
     def _extract_coroutine_stacks_lineno_only(self, stack_trace):
         """Extract coroutine stacks with line numbers only (no column offsets).
 
@@ -3574,5 +3733,114 @@ def test_get_stats_disabled_raises(self):
             client_socket.sendall(b"done")
 
 
+@requires_remote_subprocess_debugging()
+@unittest.skipUnless(sys.platform == "linux", "requires /proc/self/mem")
+@unittest.skipIf(
+    not PROCESS_VM_READV_SUPPORTED,
+    "Test only runs on Linux with process_vm_readv support",
+)
+class TestInvalidDebugOffsets(RemoteInspectionTestBase):
+    @skip_if_not_supported
+    def test_rejects_poisoned_thread_state_size(self):
+        poison_code = textwrap.dedent(
+            """\
+            thread_state_offset = 8 + 8 + 8 + 3 * 8 + 16 * 8
+            fd = os.open(f"/proc/{pid}/mem", os.O_WRONLY)
+            os.lseek(fd, addr + thread_state_offset, 0)
+            os.write(fd, struct.pack("<Q", 0xFFFFFFFFFFFFFFFF))
+            os.close(fd)
+            """
+        )
+
+        with self._poisoned_debug_offsets_process(poison_code) as (_, info):
+            with self.assertRaisesRegex(RuntimeError, "Invalid debug offsets"):
+                RemoteUnwinder(info["pid"], all_threads=True, debug=True)
+
+    @skip_if_not_supported
+    def test_rejects_poisoned_thread_state_offset(self):
+        poison_code = textwrap.dedent(
+            """\
+            thread_state_offset = 8 + 8 + 8 + 3 * 8 + 16 * 8
+            native_thread_id_offset = thread_state_offset + 64
+            fd = os.open(f"/proc/{pid}/mem", os.O_WRONLY)
+            os.lseek(fd, addr + native_thread_id_offset, 0)
+            os.write(fd, struct.pack("<Q", 2000))
+            os.close(fd)
+            """
+        )
+
+        with self._poisoned_debug_offsets_process(poison_code) as (_, info):
+            with self.assertRaisesRegex(RuntimeError, "Invalid debug offsets"):
+                RemoteUnwinder(info["pid"], all_threads=True, debug=True)
+
+    @skip_if_not_supported
+    def test_rejects_poisoned_runtime_state_interpreters_head(self):
+        poison_code = textwrap.dedent(
+            """\
+            interpreters_head_offset = 8 + 8 + 8 + 8 + 8
+            fd = os.open(f"/proc/{pid}/mem", os.O_WRONLY)
+            os.lseek(fd, addr + interpreters_head_offset, 0)
+            os.write(fd, struct.pack("<Q", 0xFFFFFFFFFFFFFFFF))
+            os.close(fd)
+            """
+        )
+
+        with self._poisoned_debug_offsets_process(poison_code) as (_, info):
+            with self.assertRaisesRegex(RuntimeError, "Invalid debug offsets"):
+                RemoteUnwinder(info["pid"], all_threads=True, debug=True)
+
+    @skip_if_not_supported
+    def test_rejects_poisoned_thread_state_datastack_chunk(self):
+        poison_code = textwrap.dedent(
+            """\
+            thread_state_offset = 8 + 8 + 8 + 3 * 8 + 16 * 8
+            datastack_chunk_offset = thread_state_offset + 72
+            fd = os.open(f"/proc/{pid}/mem", os.O_WRONLY)
+            os.lseek(fd, addr + datastack_chunk_offset, 0)
+            os.write(fd, struct.pack("<Q", 0x10000))
+            os.close(fd)
+            """
+        )
+
+        with self._poisoned_debug_offsets_process(poison_code) as (_, info):
+            with self.assertRaisesRegex(RuntimeError, "Invalid debug offsets"):
+                RemoteUnwinder(info["pid"], all_threads=True, debug=True)
+
+    @skip_if_not_supported
+    def test_rejects_poisoned_asyncio_task_node_offset(self):
+        poison_code = textwrap.dedent(
+            """\
+            asyncio_addr = find_section_address("AsyncioDebug", "python")
+            task_node_offset = 6 * 8
+            fd = os.open(f"/proc/{pid}/mem", os.O_WRONLY)
+            os.lseek(fd, asyncio_addr + task_node_offset, 0)
+            os.write(fd, struct.pack("<Q", 0x10000))
+            os.close(fd)
+            """
+        )
+
+        with self._poisoned_debug_offsets_process(poison_code) as (_, info):
+            with self.assertRaisesRegex(RuntimeError, "Invalid AsyncioDebug offsets"):
+                RemoteUnwinder(info["pid"], debug=True)
+
+    @skip_if_not_supported
+    def test_rejects_poisoned_asyncio_running_task_offset(self):
+        poison_code = textwrap.dedent(
+            """\
+            asyncio_addr = find_section_address("AsyncioDebug", "python")
+            asyncio_thread_state_offset = 7 * 8 + 2 * 8
+            asyncio_running_task_offset = asyncio_thread_state_offset + 2 * 8
+            fd = os.open(f"/proc/{pid}/mem", os.O_WRONLY)
+            os.lseek(fd, asyncio_addr + asyncio_running_task_offset, 0)
+            os.write(fd, struct.pack("<Q", 0x10000))
+            os.close(fd)
+            """
+        )
+
+        with self._poisoned_debug_offsets_process(poison_code) as (_, info):
+            with self.assertRaisesRegex(RuntimeError, "Invalid AsyncioDebug offsets"):
+                RemoteUnwinder(info["pid"], debug=True)
+
+
 if __name__ == "__main__":
     unittest.main()

Misc/NEWS.d/next/Security/2026-04-06-13-55-00.gh-issue-148178.Rs7kLm.rst

@@ -0,0 +1,2 @@
+Hardened :mod:`!_remote_debugging` by validating remote debug offset tables
+before using them to size memory reads or interpret remote layouts.

Modules/_remote_debugging/_remote_debugging.h

@@ -417,6 +417,9 @@ extern void cached_code_metadata_destroy(void *ptr);
 /* Validation */
 extern int is_prerelease_version(uint64_t version);
 extern int validate_debug_offsets(struct _Py_DebugOffsets *debug_offsets);
+enum {
+    PY_REMOTE_DEBUG_INVALID_ASYNC_DEBUG_OFFSETS = -2,
+};
 
 /* ============================================================================
  * MEMORY READING FUNCTION DECLARATIONS

Modules/_remote_debugging/asyncio.c

@@ -6,6 +6,7 @@
  ******************************************************************************/
 
 #include "_remote_debugging.h"
+#include "debug_offsets_validation.h"
 
 /* ============================================================================
  * ASYNCIO DEBUG ADDRESS FUNCTIONS
@@ -71,8 +72,13 @@ read_async_debug(RemoteUnwinderObject *unwinder)
     int result = _Py_RemoteDebug_PagedReadRemoteMemory(&unwinder->handle, async_debug_addr, size, &unwinder->async_debug_offsets);
     if (result < 0) {
         set_exception_cause(unwinder, PyExc_RuntimeError, "Failed to read AsyncioDebug offsets");
+        return result;
     }
-    return result;
+    if (_PyRemoteDebug_ValidateAsyncDebugOffsets(&unwinder->async_debug_offsets) < 0) {
+        set_exception_cause(unwinder, PyExc_RuntimeError, "Invalid AsyncioDebug offsets");
+        return PY_REMOTE_DEBUG_INVALID_ASYNC_DEBUG_OFFSETS;
+    }
+    return 0;
 }
 
 int
@@ -85,7 +91,11 @@ ensure_async_debug_offsets(RemoteUnwinderObject *unwinder)
 
     // Try to load async debug offsets (the target process may have
     // loaded asyncio since we last checked)
-    if (read_async_debug(unwinder) < 0) {
+    int result = read_async_debug(unwinder);
+    if (result == PY_REMOTE_DEBUG_INVALID_ASYNC_DEBUG_OFFSETS) {
+        return -1;
+    }
+    if (result < 0) {
         PyErr_Clear();
         PyErr_SetString(PyExc_RuntimeError, "AsyncioDebug section not available");
         set_exception_cause(unwinder, PyExc_RuntimeError,