gh-130168: pyrepl can messup and poll is not thread safe

Lib/_pyrepl/unix_console.py

@@ -28,6 +28,7 @@
 import signal
 import struct
 import termios
+import threading
 import time
 import types
 import platform
@@ -161,6 +162,9 @@ def __init__(
 
  self.pollob = poll()
  self.pollob.register(self.input_fd, select.POLLIN)
+ self._poll_lock = threading.RLock()
+ self._polling_thread: threading.Thread | None = None
+ self.__svtermstate = None
  self.terminfo = terminfo.TermInfo(term or None)
  self.term = term
  self.is_apple_terminal = (
@@ -341,10 +345,12 @@ def prepare(self):
  """
  Prepare the console for input/output operations.
  """
+ # gh-130168: prevents signal handlers from overwriting the original state
+ if self.__svtermstate is None:
+ self.__svtermstate = tcgetattr(self.input_fd)
  self.__buffer = []
 
- self.__svtermstate = tcgetattr(self.input_fd)
- raw = self.__svtermstate.copy()
+ raw = tcgetattr(self.input_fd).copy()
  raw.iflag &= ~(termios.INPCK | termios.ISTRIP | termios.IXON)
  raw.oflag &= ~(termios.OPOST)
  raw.cflag &= ~(termios.CSIZE | termios.PARENB)
@@ -384,7 +390,15 @@ def restore(self):
  self.__disable_bracketed_paste()
  self.__maybe_write_code(self._rmkx)
  self.flushoutput()
- self.__input_fd_set(self.__svtermstate)
+ try:
+ if self.__svtermstate is not None:
+ tcsetattr(self.input_fd, termios.TCSADRAIN, self.__svtermstate)
+ self.__input_fd_set(self.__svtermstate)
+ # Reset the state for the next prepare() call.
+ self.__svtermstate = None
+ except termios.error as e:
+ if e.args[0] != errno.EIO:
+ raise
 
  if self.is_apple_terminal:
  os.write(self.output_fd, b"\033[?7h")
@@ -435,10 +449,20 @@ def wait(self, timeout: float | None = None) -> bool:
  """
  Wait for events on the console.
  """
- return (
- not self.event_queue.empty()
- or bool(self.pollob.poll(timeout))
- )
+ if not self.event_queue.empty():
+ return True
+ current_thread = threading.current_thread()
+ if self._polling_thread is current_thread:
+ # Forbid re-entrant calls and use the old REPL error message.
+ raise RuntimeError("can't re-enter readline")
+ if not self._poll_lock.acquire(blocking=False):
+ return False
+ try:
+ self._polling_thread = current_thread
+ return bool(self.pollob.poll(timeout))
+ finally:
+ self._polling_thread = None
+ self._poll_lock.release()
 
  def set_cursor_vis(self, visible):
  """
@@ -804,7 +828,10 @@ def __tputs(self, fmt, prog=delayprog):
  # using .get() means that things will blow up
  # only if the bps is actually needed (which I'm
  # betting is pretty unlkely)
- bps = ratedict.get(self.__svtermstate.ospeed)
+ if self.__svtermstate is not None:
+ bps = ratedict.get(self.__svtermstate.ospeed)
+ else:
+ bps = None
  while True:
  m = prog.search(fmt)
  if not m:

Lib/test/test_pyrepl/test_unix_console.py

@@ -4,6 +4,7 @@
 import signal
 import subprocess
 import sys
+import threading
 import unittest
 from functools import partial
 from test.support import os_helper, force_not_colorized_test_class
@@ -308,6 +309,18 @@ def test_getheightwidth_with_invalid_environ(self, _os_write):
  os.environ = []
  self.assertIsInstance(console.getheightwidth(), tuple)
 
+ def test_wait_reentry_protection(self, _os_write):
+ # gh-130168: Test signal handler re-entry protection
+ console = UnixConsole(term="xterm")
+ console.prepare()
+
+ console._polling_thread = threading.current_thread()
+ with self.assertRaisesRegex(RuntimeError, "can't re-enter readline"):
+ console.wait(timeout=0)
+
+ console._polling_thread = None
+ console.restore()
+
  @unittest.skipUnless(sys.platform == "darwin", "requires macOS")
  def test_restore_with_invalid_environ_on_macos(self, _os_write):
  # gh-128636 for macOS

Misc/NEWS.d/next/Library/2025-09-07-19-47-35.gh-issue-130168.qvtlOZ.rst

@@ -0,0 +1,2 @@
+Make sure that the new REPL does not get scrambled when a signal handler
+uses interactive input.