Debian Linux下Python用pty模块与交互式Bash交互时进程停止问题修复

问题背景 #

在Debian Linux环境下，需要启动一个交互式Bash会话并传递命令，实现命令间的变量持久化，但现有代码中Bash在初始化阶段就突然停止，无法正常交互。要求不借助外部包解决，以加深相关技术理解。

失败尝试代码 #

# file: Shell.py
import os
import pty
import subprocess
import time


class Shell:
    def __init__(self):
        # Start a persistent interactive Bash session
        self.master, self.slave = pty.openpty()
        self.process = subprocess.Popen(
            ["bash", "-i"], stdin=self.slave, stdout=self.master, stderr=self.master
        )

    def execute(self, command: str) -> str:
        time.sleep(2)  # Wait for the Bash prompt to appear

        # Send command to Bash
        os.write(self.master, (command + "\n").encode())

        # Capture output until the Bash prompt appears
        output = []
        while True:
            data = os.read(self.master, 1024).decode()
            output.append(data)
            if data.endswith("$ ") or data.endswith("# "):
                break
        return "".join(output).strip()

    def close(self):
        # Terminate the Bash process
        self.process.terminate()
        os.close(self.master)
        os.close(self.slave)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.close()


# Example usage
with Shell() as shell:
    print(shell.execute("echo Hello, World!"))
    print(shell.execute("export TEST_VAR='Persistent Variable'"))
    print(shell.execute("echo $TEST_VAR"))
    time.sleep(10)  # Wait enough time for the persistent variable to be available

运行异常行为 #

$ python3 Shell.py 

[1]+  Stopped                 python3 Shell.py

补充诊断信息 #

执行脚本后，通过以下命令查看进程状态：

$ jobs -l
[1]+ 15437 Stopped (tty input)     python3 Shell.py
$ kill -CONT 15437
$ jobs
[1]+  Running                 python3 Shell.py &
$ ps --tty $(tty)
    PID TTY          TIME CMD
   9103 pts/2    00:00:00 bash
  15437 pts/2    00:00:00 python3
  15438 pts/2    00:00:00 bash
  15501 pts/2    00:00:00 ps

可见Python进程及其启动的Bash进程，与调用脚本的父Bash实例共用同一个TTY，这是导致进程暂停的核心原因。

问题根源 #

当使用pty.openpty()创建伪终端时，父Python进程的标准输入输出仍绑定到原TTY。交互式Bash会尝试读取TTY输入，触发终端的SIGTTIN信号，导致Python进程被操作系统暂停。

修复方案代码 #

# file: Shell.py
import os
import pty
import subprocess
import select


class Shell:
    def __init__(self):
        # 创建伪终端
        self.master, self.slave = pty.openpty()
        
        # 重定向Python进程的标准输入输出到/dev/null，隔绝原TTY干扰
        with open(os.devnull, 'r+') as devnull:
            os.dup2(devnull.fileno(), 0)
            os.dup2(devnull.fileno(), 1)
            os.dup2(devnull.fileno(), 2)
        
        # 启动交互式Bash，强制设置固定格式提示符，避免环境差异
        self.process = subprocess.Popen(
            ["bash", "-i"],
            stdin=self.slave,
            stdout=self.master,
            stderr=self.master,
            env={**os.environ, "PS1": "$ "}
        )
        
        # 读取初始启动输出，直到出现提示符
        self._read_until_prompt()

    def _read_until_prompt(self) -> str:
        """读取输出直到出现指定提示符，返回提示符之前的内容"""
        output = []
        while True:
            # 使用select等待数据可读，避免无限阻塞
            ready, _, _ = select.select([self.master], [], [], 5)
            if not ready:
                raise TimeoutError("等待Bash响应超时")
            
            data = os.read(self.master, 1024).decode('utf-8', errors='replace')
            output.append(data)
            if "$ " in data:
                full_output = "".join(output)
                return full_output[:full_output.rfind("$ ")]

    def execute(self, command: str) -> str:
        # 发送命令并追加换行符
        os.write(self.master, (command + "\n").encode())
        
        # 读取命令执行结果直到提示符出现
        return self._read_until_prompt().strip()

    def close(self):
        # 安全终止Bash进程并关闭伪终端
        self.process.terminate()
        try:
            self.process.wait(timeout=2)
        except subprocess.TimeoutExpired:
            self.process.kill()
        os.close(self.master)
        os.close(self.slave)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.close()


# 示例使用
if __name__ == "__main__":
    with Shell() as shell:
        print(shell.execute("echo Hello, World!"))
        shell.execute("export TEST_VAR='Persistent Variable'")
        print(shell.execute("echo $TEST_VAR"))

修复说明 #

1. 隔绝原TTY干扰：将Python进程的标准IO重定向到/dev/null，让子Bash完全通过伪终端交互，避免触发终端信号导致进程暂停。
2. 动态等待输出：用select替代固定time.sleep()，实时等待伪终端数据，提升可靠性并添加超时处理。
3. 固定提示符：强制设置PS1为$ ，确保输出捕获逻辑能稳定识别提示符，避免不同环境下的格式差异。
4. 精准截取结果：仅返回命令执行的有效输出，自动过滤提示符和启动信息。

内容的提问来源于stack exchange，提问作者suuudooo