Only allow KeyboardInterrupt at specific times

Because our goal is to proxy interrupt signals to the remote process,
it's important that we only accept those signals at moments when we're
ready to handle them and send them on. In particular, we're prepared to
handle them when reading user input for a prompt, and when waiting for
the server to send a new prompt after our last command. We do not want
to accept them at other times, like while we're in the middle of
printing output to the screen, as otherwise a

    while True: "Hello"

executed at the PDB REPL can't reliably be Ctrl-C'd: it's more likely
that the signal will arrive while the client is printing than while it's
waiting for the next line from the server, and if so nothing will be
prepared to handle the `KeyboardInterrupt` and the client will exit.

Author: godlygeek

Lib/pdb.py

@@ -3009,12 +3009,53 @@ def readline_completion(self, completer):
         finally:
             readline.set_completer(old_completer)
 
+    if not hasattr(signal, "pthread_sigmask"):
+        # On Windows, we must drop signals arriving while we're ignoring them.
+        @contextmanager
+        def _block_sigint(self):
+            old_handler = signal.signal(signal.SIGINT, signal.SIG_IGN)
+            try:
+                yield
+            finally:
+                signal.signal(signal.SIGINT, old_handler)
+
+        @contextmanager
+        def _handle_sigint(self, handler):
+            old_handler = signal.signal(signal.SIGINT, handler)
+            try:
+                yield
+            finally:
+                signal.signal(signal.SIGINT, old_handler)
+    else:
+        # On Unix, we can save them to be processed later.
+        @contextmanager
+        def _block_sigint(self):
+            signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGINT})
+            try:
+                yield
+            finally:
+                signal.pthread_sigmask(signal.SIG_UNBLOCK, {signal.SIGINT})
+
+        @contextmanager
+        def _handle_sigint(self, handler):
+            old_handler = signal.signal(signal.SIGINT, handler)
+            try:
+                signal.pthread_sigmask(signal.SIG_UNBLOCK, {signal.SIGINT})
+                yield
+            finally:
+                signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGINT})
+                signal.signal(signal.SIGINT, old_handler)
+
     def cmdloop(self):
-        with self.readline_completion(self.complete):
+        with (
+            self._block_sigint(),
+            self.readline_completion(self.complete),
+        ):
             while not self.write_failed:
                 try:
-                    if not (payload_bytes := self.sockfile.readline()):
-                        break
+                    with self._handle_sigint(signal.default_int_handler):
+                        if not (payload_bytes := self.sockfile.readline()):
+                            break
                 except KeyboardInterrupt:
                     self.send_interrupt()
                     continue
@@ -3061,7 +3102,8 @@ def process_payload(self, payload):
     def prompt_for_reply(self, prompt):
         while True:
             try:
-                payload = {"reply": self.read_command(prompt)}
+                with self._handle_sigint(signal.default_int_handler):
+                    payload = {"reply": self.read_command(prompt)}
             except EOFError:
                 payload = {"signal": "EOF"}
             except KeyboardInterrupt:
@@ -3101,7 +3143,8 @@ def complete(self, text, state):
             if self.write_failed:
                 return None
 
-            payload = self.sockfile.readline()
+            with self._handle_sigint(signal.default_int_handler):
+                payload = self.sockfile.readline()
             if not payload:
                 return None