Block SIGTTOU only while lending the terminal to a pipeline

SIGTTOU was blocked on the main thread for the whole life of a terminal
pipeline. A signal mask survives fork and exec, so every child started
meanwhile, whether a shell producer or a subprocess run by command code,
kept SIGTTOU blocked for life and could change terminal modes from the
background. The consumer now owns the terminal only during a lend, so
block SIGTTOU for the lend alone, and spawn a shell producer, which
starts inside one, with it unblocked.

Author: tleonhardt

cmd2/cmd2.py

@@ -3621,13 +3621,6 @@ def _redirect_output(self, statement: Statement) -> utils.RedirectionSavedState:
                 # exit must unblock a producer writing to a full pipe immediately.
                 subproc_stdin.close()
                 if terminal_fd is not None:
-                    import signal
-
-                    # The producer may still write diagnostics while the consumer owns
-                    # the terminal. Block SIGTTOU after Popen so the child retains normal
-                    # job-control behavior, and restore our mask with the handoff.
-                    previous_mask = signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGTTOU})
-                    terminal_stack.callback(signal.pthread_sigmask, signal.SIG_SETMASK, previous_mask)
                     cmd_pipe_proc_reader = utils.ProcReader(proc, self.stdout, sys.stderr, terminal_fd=terminal_fd)
                     terminal_stack.enter_context(cmd_pipe_proc_reader.manage_terminal())
 
@@ -5304,14 +5297,16 @@ def do_shell(self, args: argparse.Namespace) -> None:
                 terminal_stack.enter_context(pipeline.lend_terminal())
             while True:
                 try:
-                    # For any stream that is a StdSim, we will use a pipe so we can capture its output
-                    proc = subprocess.Popen(  # noqa: S602
-                        expanded_command,
-                        stdout=subprocess.PIPE if isinstance(self.stdout, utils.StdSim) else self.stdout,  # type: ignore[unreachable]
-                        stderr=subprocess.PIPE if isinstance(sys.stderr, utils.StdSim) else sys.stderr,
-                        shell=True,
-                        **kwargs,
-                    )
+                    # For any stream that is a StdSim, we will use a pipe so we can capture its output.
+                    # A command joining the pipeline is spawned inside the lend, which blocks SIGTTOU.
+                    with utils.unblocked_sigttou() if "process_group" in kwargs else contextlib.nullcontext():
+                        proc = subprocess.Popen(  # noqa: S602
+                            expanded_command,
+                            stdout=subprocess.PIPE if isinstance(self.stdout, utils.StdSim) else self.stdout,  # type: ignore[unreachable]
+                            stderr=subprocess.PIPE if isinstance(sys.stderr, utils.StdSim) else sys.stderr,
+                            shell=True,
+                            **kwargs,
+                        )
                     break
                 except PermissionError:
                     # The pipeline exited before the command could join its group.

cmd2/utils.py

@@ -536,6 +536,22 @@ def write(self, b: bytes) -> None:
                 self.std_sim_instance.flush()
 
 
+@contextlib.contextmanager
+def unblocked_sigttou() -> Iterator[None]:
+    """Let a child started inside :meth:`ProcReader.lend_terminal` keep normal job control.
+
+    The lend blocks SIGTTOU for its thread, and a child inherits that mask for life. Spawning
+    touches no terminal, so unblocking it for the spawn alone cannot stop this thread.
+    """
+    import signal
+
+    previous_mask = signal.pthread_sigmask(signal.SIG_UNBLOCK, {signal.SIGTTOU})
+    try:
+        yield
+    finally:
+        signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask)
+
+
 class ProcReader:
     """Used to capture stdout and stderr from a Popen process if any of those were set to subprocess.PIPE.
 
@@ -685,25 +701,35 @@ def lend_terminal(self) -> Iterator[None]:
         reads through input(), getpass(), or third-party libraries. Lending during
         writes lets an interactive consumer drain a full pipe without deadlocking.
         """
+        import signal
+
         terminal_fd = self._terminal_fd
         if terminal_fd is None or self._proc.returncode is not None:
             yield
             return
-        with self._terminal_lock:
-            try:
-                self._set_foreground_group(terminal_fd, self._proc.pid)
-            except OSError as error:
-                # The group can disappear before the watcher has reaped its leader.
-                if error.errno not in (errno.ESRCH, errno.EINVAL):
-                    raise
-            self._terminal_available.set()
+        # While the consumer owns the terminal, a signal handler run on this thread may still
+        # write diagnostics to it. Block SIGTTOU for the lend only: a signal mask survives fork
+        # and exec, so blocking it for the whole pipeline would leak into every child the
+        # command starts. A child started during a lend must unblock it; see unblocked_sigttou().
+        previous_mask = signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGTTOU})
         try:
-            yield
-        finally:
             with self._terminal_lock:
-                self._terminal_available.clear()
-                if os.tcgetpgrp(terminal_fd) == self._proc.pid:
-                    self._set_foreground_group(terminal_fd, self._original_group)
+                try:
+                    self._set_foreground_group(terminal_fd, self._proc.pid)
+                except OSError as error:
+                    # The group can disappear before the watcher has reaped its leader.
+                    if error.errno not in (errno.ESRCH, errno.EINVAL):
+                        raise
+                self._terminal_available.set()
+            try:
+                yield
+            finally:
+                with self._terminal_lock:
+                    self._terminal_available.clear()
+                    if os.tcgetpgrp(terminal_fd) == self._proc.pid:
+                        self._set_foreground_group(terminal_fd, self._original_group)
+        finally:
+            signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask)
 
     def _wait_for_job(self, terminal_fd: int) -> None:
         """Reap a foreground pipeline and relay its stops to the outer shell's job.

tests/test_cmd2.py

@@ -963,10 +963,9 @@ def start_pipe(*args, **kwargs):
         reader.wait_for_exit.assert_called_once_with(0.2)
         reader.wait.assert_called_once_with()
         process.wait.assert_not_called()
-        assert sigmask.call_args_list == [
-            mock.call(signal.SIG_BLOCK, {signal.SIGTTOU}),
-            mock.call(signal.SIG_SETMASK, set()),
-        ]
+        # SIGTTOU is blocked only inside ProcReader's lends. Blocking it for the whole
+        # pipeline would leak the mask into every child the command starts.
+        sigmask.assert_not_called()
     else:
         process.wait.assert_called_once()
     assert popen.call_args.kwargs["stdin"].closed

tests/test_pipeline_job_control.py

@@ -642,6 +642,85 @@ def wait_until(predicate):
         process.wait(timeout=5)
 
 
+@pytest.mark.parametrize("producer", ["command", "shell"])
+def test_pipeline_children_inherit_an_ordinary_signal_mask(tmp_path, producer) -> None:
+    """Processes started during a terminal pipeline must not inherit a blocked SIGTTOU.
+
+    cmd2 blocks SIGTTOU for itself while it lends the terminal. A signal mask survives fork
+    and exec, so a child spawned with it blocked -- a shell producer, or a subprocess run by
+    command code -- would keep it for life, and change terminal modes from the background
+    where it should be stopped.
+    """
+    import pty
+
+    shell = shutil.which("bash")
+    if shell is None:
+        pytest.skip("requires an interactive bash shell")
+    probe = tmp_path / "probe.py"
+    outcome = tmp_path / "outcome"
+    probe.write_text(
+        "import pathlib, signal\n"
+        "blocked = signal.SIGTTOU in signal.pthread_sigmask(signal.SIG_BLOCK, [])\n"
+        f"pathlib.Path({str(outcome)!r}).write_text(repr(blocked))\n",
+        encoding="utf-8",
+    )
+    application = tmp_path / "application.py"
+    application.write_text(
+        "import subprocess, sys\n"
+        "from cmd2 import Cmd\n"
+        "class App(Cmd):\n"
+        "    def do_probe(self, _):\n"
+        "        self.poutput('probing')\n"
+        f"        subprocess.run([sys.executable, {str(probe)!r}], check=True)\n"
+        "app = App()\n"
+        "app.prompt = 'TEST> '\n"
+        "app.cmdloop()\n",
+        encoding="utf-8",
+    )
+    master, slave = pty.openpty()
+    bootstrap = (
+        "import os, fcntl, termios; os.setsid(); "
+        "fcntl.ioctl(0, termios.TIOCSCTTY, 0); "
+        "os.execv(os.environ['TEST_SHELL'], ['bash', '--noprofile', '--norc', '-i'])"
+    )
+    env = dict(os.environ, TERM="xterm-256color", PS1="OUTER> ", TEST_SHELL=shell, SHELL=shell)
+    env["PYTHONPATH"] = str(Path(__file__).resolve().parents[1])
+    process = subprocess.Popen([sys.executable, "-c", bootstrap], stdin=slave, stdout=slave, stderr=slave, env=env)
+    os.close(slave)
+    decoder = codecs.getincrementaldecoder("utf-8")("replace")
+    transcript = ""
+
+    def wait_until(predicate):
+        nonlocal transcript
+        deadline = time.monotonic() + 10
+        while time.monotonic() < deadline:
+            if select.select([master], [], [], 0.05)[0]:
+                data = decoder.decode(os.read(master, 65536))
+                transcript += data
+                if "\x1b[6n" in data:
+                    # Answer prompt-toolkit's cursor-position request as a terminal would.
+                    os.write(master, b"\x1b[1;1R")
+            if predicate():
+                return
+        pytest.fail(f"terminal condition timed out:\n{transcript}\n{describe_processes(process.pid, master)}")
+
+    command = "probe" if producer == "command" else f"shell {shlex.quote(sys.executable)} {shlex.quote(str(probe))}"
+    try:
+        wait_until(lambda: "OUTER> " in transcript)
+        os.write(master, f"{shlex.quote(sys.executable)} {shlex.quote(str(application))}\n".encode())
+        wait_until(lambda: "TEST>" in transcript)
+        start = len(transcript)
+        os.write(master, f"{command} | cat\n".encode())
+        wait_until(lambda: outcome.exists() and outcome.read_text() != "" and "TEST>" in transcript[start:])
+        assert outcome.read_text() == "False"
+        os.write(master, b"quit\n")
+        wait_until(lambda: os.tcgetpgrp(master) == process.pid)
+    finally:
+        os.close(master)
+        process.kill()
+        process.wait(timeout=5)
+
+
 @pytest.mark.parametrize("suspend", [False, True])
 def test_shell_producer_keeps_the_terminal_after_its_consumer_exits(tmp_path, suspend) -> None:
     """A shell producer that outlives its consumer still reads the terminal.