From 5b30f61e7c7093db23e4b6f34b95b7602610db59 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=81ukasz=20Langa?= Date: Mon, 15 Sep 2025 12:53:55 +0200 Subject: [PATCH 1/2] gh-134466: Don't run when termios is inaccessible Without the ability to set required capabilities, the REPL cannot function properly (syntax highlighting and multiline editing can't work). We refuse to work in this degraded state. --- Lib/_pyrepl/fancy_termios.py | 44 +++++++++++++++++++-------------- Lib/_pyrepl/unix_console.py | 48 +++++++++++++++++++++++++++--------- 2 files changed, 62 insertions(+), 30 deletions(-) diff --git a/Lib/_pyrepl/fancy_termios.py b/Lib/_pyrepl/fancy_termios.py index 0468b9a2670267..8d5bd183f21339 100644 --- a/Lib/_pyrepl/fancy_termios.py +++ b/Lib/_pyrepl/fancy_termios.py @@ -20,19 +20,25 @@ import termios +TYPE_CHECKING = False + +if TYPE_CHECKING: + from typing import cast +else: + cast = lambda typ, val: val + + class TermState: - def __init__(self, tuples): - ( - self.iflag, - self.oflag, - self.cflag, - self.lflag, - self.ispeed, - self.ospeed, - self.cc, - ) = tuples + def __init__(self, attrs: list[int | list[bytes]]) -> None: + self.iflag = cast(int, attrs[0]) + self.oflag = cast(int, attrs[1]) + self.cflag = cast(int, attrs[2]) + self.lflag = cast(int, attrs[3]) + self.ispeed = cast(int, attrs[4]) + self.ospeed = cast(int, attrs[5]) + self.cc = cast(list[bytes], attrs[6]) - def as_list(self): + def as_list(self) -> list[int | list[bytes]]: return [ self.iflag, self.oflag, @@ -45,32 +51,32 @@ def as_list(self): self.cc[:], ] - def copy(self): + def copy(self) -> "TermState": return self.__class__(self.as_list()) -def tcgetattr(fd): +def tcgetattr(fd: int) -> TermState: return TermState(termios.tcgetattr(fd)) -def tcsetattr(fd, when, attrs): +def tcsetattr(fd: int, when: int, attrs: TermState) -> None: termios.tcsetattr(fd, when, attrs.as_list()) class Term(TermState): TS__init__ = TermState.__init__ - def __init__(self, fd=0): + def __init__(self, fd: int = 0) -> None: self.TS__init__(termios.tcgetattr(fd)) self.fd = fd - self.stack = [] + self.stack: list[list[int | list[bytes]]] = [] - def save(self): + def save(self) -> None: self.stack.append(self.as_list()) - def set(self, when=termios.TCSANOW): + def set(self, when: int = termios.TCSANOW) -> None: termios.tcsetattr(self.fd, when, self.as_list()) - def restore(self): + def restore(self) -> None: self.TS__init__(self.stack.pop()) self.set() diff --git a/Lib/_pyrepl/unix_console.py b/Lib/_pyrepl/unix_console.py index a7e49923191c07..abc1bbc306a2de 100644 --- a/Lib/_pyrepl/unix_console.py +++ b/Lib/_pyrepl/unix_console.py @@ -35,7 +35,7 @@ from . import terminfo from .console import Console, Event -from .fancy_termios import tcgetattr, tcsetattr +from .fancy_termios import tcgetattr, tcsetattr, TermState from .trace import trace from .unix_eventqueue import EventQueue from .utils import wlen @@ -51,16 +51,19 @@ # types if TYPE_CHECKING: - from typing import IO, Literal, overload + from typing import AbstractSet, IO, Literal, overload, cast else: overload = lambda func: None + cast = lambda typ, val: val class InvalidTerminal(RuntimeError): - pass + def __init__(self, message: str) -> None: + super().__init__(errno.EIO, message) _error = (termios.error, InvalidTerminal) +_error_codes_to_ignore: frozenset[int] = frozenset([errno.EIO, errno.EPERM]) SIGWINCH_EVENT = "repaint" @@ -125,12 +128,13 @@ def __init__(self): def register(self, fd, flag): self.fd = fd + # note: The 'timeout' argument is received as *milliseconds* def poll(self, timeout: float | None = None) -> list[int]: if timeout is None: r, w, e = select.select([self.fd], [], []) else: - r, w, e = select.select([self.fd], [], [], timeout/1000) + r, w, e = select.select([self.fd], [], [], timeout / 1000) return r poll = MinimalPoll # type: ignore[assignment] @@ -160,8 +164,15 @@ def __init__( self.terminfo = terminfo.TermInfo(term or None) self.term = term + try: + self.__input_fd_set(tcgetattr(self.input_fd), ignore=frozenset()) + except _error as e: + raise RuntimeError(f"termios failure ({e.args[1]})") + @overload - def _my_getstr(cap: str, optional: Literal[False] = False) -> bytes: ... + def _my_getstr( + cap: str, optional: Literal[False] = False + ) -> bytes: ... @overload def _my_getstr(cap: str, optional: bool) -> bytes | None: ... @@ -201,7 +212,9 @@ def _my_getstr(cap: str, optional: bool = False) -> bytes | None: self.__setup_movement() - self.event_queue = EventQueue(self.input_fd, self.encoding, self.terminfo) + self.event_queue = EventQueue( + self.input_fd, self.encoding, self.terminfo + ) self.cursor_visible = 1 signal.signal(signal.SIGCONT, self._sigcont_handler) @@ -213,7 +226,6 @@ def _sigcont_handler(self, signum, frame): def __read(self, n: int) -> bytes: return os.read(self.input_fd, n) - def change_encoding(self, encoding: str) -> None: """ Change the encoding used for I/O operations. @@ -325,6 +337,8 @@ def prepare(self): """ Prepare the console for input/output operations. """ + self.__buffer = [] + self.__svtermstate = tcgetattr(self.input_fd) raw = self.__svtermstate.copy() raw.iflag &= ~(termios.INPCK | termios.ISTRIP | termios.IXON) @@ -336,7 +350,7 @@ def prepare(self): raw.lflag |= termios.ISIG raw.cc[termios.VMIN] = 1 raw.cc[termios.VTIME] = 0 - tcsetattr(self.input_fd, termios.TCSADRAIN, raw) + self.__input_fd_set(raw) # In macOS terminal we need to deactivate line wrap via ANSI escape code if platform.system() == "Darwin" and os.getenv("TERM_PROGRAM") == "Apple_Terminal": @@ -345,8 +359,6 @@ def prepare(self): self.screen = [] self.height, self.width = self.getheightwidth() - self.__buffer = [] - self.posxy = 0, 0 self.__gone_tall = 0 self.__move = self.__move_short @@ -368,7 +380,7 @@ def restore(self): self.__disable_bracketed_paste() self.__maybe_write_code(self._rmkx) self.flushoutput() - tcsetattr(self.input_fd, termios.TCSADRAIN, self.__svtermstate) + self.__input_fd_set(self.__svtermstate) if platform.system() == "Darwin" and os.getenv("TERM_PROGRAM") == "Apple_Terminal": os.write(self.output_fd, b"\033[?7h") @@ -803,3 +815,17 @@ def __tputs(self, fmt, prog=delayprog): os.write(self.output_fd, self._pad * nchars) else: time.sleep(float(delay) / 1000.0) + + def __input_fd_set( + self, + state: TermState, + ignore: AbstractSet[int] = _error_codes_to_ignore, + ) -> bool: + try: + tcsetattr(self.input_fd, termios.TCSADRAIN, state) + except termios.error as te: + if te.args[0] not in ignore: + raise + return False + else: + return True From 66dc5de022559e4f88352d8ee7ee90ff286277da Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=81ukasz=20Langa?= Date: Mon, 15 Sep 2025 14:05:41 +0200 Subject: [PATCH 2/2] Add Blurb --- .../2025-09-15-14-04-56.gh-issue-134466.yR4fYW.rst | 2 ++ 1 file changed, 2 insertions(+) create mode 100644 Misc/NEWS.d/next/Core_and_Builtins/2025-09-15-14-04-56.gh-issue-134466.yR4fYW.rst diff --git a/Misc/NEWS.d/next/Core_and_Builtins/2025-09-15-14-04-56.gh-issue-134466.yR4fYW.rst b/Misc/NEWS.d/next/Core_and_Builtins/2025-09-15-14-04-56.gh-issue-134466.yR4fYW.rst new file mode 100644 index 00000000000000..4fae7e0d6f6326 --- /dev/null +++ b/Misc/NEWS.d/next/Core_and_Builtins/2025-09-15-14-04-56.gh-issue-134466.yR4fYW.rst @@ -0,0 +1,2 @@ +Don't run PyREPL in a degraded environment where setting termios attributes +is not allowed.