Python SSH自动化：/dev/tty子进程通信问题及解决需求

需求 #

• 完成SSH认证连接
• 将控制权移交用户以进行交互式会话
• 干净终止进程，避免残留进程

当前问题 #

SSH直接读取/dev/tty而非stdin，导致现有脚本存在以下问题：

• 父子进程通信失败：密码提示出现时无法自动填入密码造成卡顿；发送SIGINT时陷入密码与OTP提示的无限循环
• 认证流程卡顿或无法正确发送凭证

约束条件 #

• 仅允许使用Python标准库，不可使用pexpect、paramiko等第三方库
• 运行于共享堡垒机，无法安装外部工具

现有实现代码 #

import os
import pty
import subprocess
import time

# 需自行定义SSH_COMMAND、PASSWORD、generate_otp()
SSH_COMMAND = "ssh user@bastion-host"
PASSWORD = "your-password"

def generate_otp():
    # 替换为实际的OTP生成逻辑
    return "123456"

def spawn_with_pty():
    master_fd, slave_fd = pty.openpty()
    proc = subprocess.Popen(SSH_COMMAND,
                            stdin=slave_fd,
                            stdout=slave_fd,
                            stderr=slave_fd,
                            text=True,
                            close_fds=True)
    os.close(slave_fd)
    return proc, master_fd


def write_to_master(master_fd, text):
    os.write(master_fd, text.encode())


def main():
    try:
        process, master_fd = spawn_with_pty()
        time.sleep(2)
        write_to_master(master_fd, PASSWORD + '\n')
        time.sleep(2)
        otp = generate_otp()
        write_to_master(master_fd, otp + '\n')
        process.wait()
    except KeyboardInterrupt:
        print("Connection closed by user.")
        os.close(master_fd)
        process.terminate()
    except Exception as e:
        print(f"SOMETHING WENT WRONG: {str(e)}")
        
if __name__ == "__main__":
    main()

解决方案 #

核心问题分析 #

现有脚本用固定time.sleep()等待提示，无法精准匹配认证时机，导致凭证发送时机错误；同时未处理终端控制权移交，也未正确处理信号传递与资源清理。

改进实现代码 #

import os
import pty
import subprocess
import select
import signal
import termios

# 配置项
SSH_COMMAND = "ssh user@bastion-host"
PASSWORD = "your-password"

def generate_otp():
    # 替换为实际OTP生成逻辑（比如读取TOTP密钥）
    return "123456"

def handle_terminal(master_fd):
    # 同步终端属性，保证交互式会话一致性
    old_attr = termios.tcgetattr(0)
    try:
        pty_attr = termios.tcgetattr(master_fd)
        termios.tcsetattr(0, termios.TCSANOW, pty_attr)
    except termios.error:
        pass

    try:
        while True:
            # 监听用户输入和PTY输出
            rlist, _, _ = select.select([0, master_fd], [], [])
            if 0 in rlist:
                # 将用户输入转发到PTY
                data = os.read(0, 1024)
                os.write(master_fd, data)
            if master_fd in rlist:
                # 将PTY输出转发到终端
                data = os.read(master_fd, 1024)
                if not data:
                    break
                os.write(1, data)
    finally:
        # 恢复原终端属性
        termios.tcsetattr(0, termios.TCSANOW, old_attr)

def main():
    proc = None
    master_fd = None
    try:
        master_fd, slave_fd = pty.openpty()
        proc = subprocess.Popen(SSH_COMMAND,
                                stdin=slave_fd,
                                stdout=slave_fd,
                                stderr=slave_fd,
                                close_fds=True)
        os.close(slave_fd)

        # 动态匹配认证提示，发送凭证
        authenticated = False
        password_sent = False
        otp_sent = False

        while not authenticated:
            rlist, _, _ = select.select([master_fd], [], [], 1)
            if master_fd in rlist:
                output = os.read(master_fd, 1024).decode('utf-8', errors='ignore')
                os.write(1, output.encode())  # 实时回显认证过程

                # 匹配密码提示（根据实际堡垒机提示调整字符串）
                if "password:" in output.lower() and not password_sent:
                    os.write(master_fd, (PASSWORD + '\n').encode())
                    password_sent = True
                # 匹配OTP提示（根据实际堡垒机提示调整字符串）
                elif ("otp:" in output.lower() or "verification code:" in output.lower()) and not otp_sent:
                    otp = generate_otp()
                    os.write(master_fd, (otp + '\n').encode())
                    otp_sent = True
                # 判断是否进入交互式会话（根据堡垒机提示符调整）
                elif "$" in output or "#" in output:
                    authenticated = True

        # 移交终端控制权给用户
        handle_terminal(master_fd)

        proc.wait()
    except KeyboardInterrupt:
        print("\nConnection closed by user.")
    except Exception as e:
        print(f"Error: {str(e)}")
    finally:
        # 强制清理资源，避免残留进程
        if master_fd:
            try:
                os.close(master_fd)
            except:
                pass
        if proc and proc.poll() is None:
            proc.terminate()
            proc.wait()

if __name__ == "__main__":
    main()

改进点说明 #

1. 动态匹配提示：用select监听PTY输出，精准匹配密码/OTP提示，替代固定延迟，避免凭证发送时机错误
2. 终端控制权移交：通过termios同步终端属性，让用户获得完整的交互式会话体验
3. 资源与信号处理：在finally块统一清理文件描述符、终止进程，彻底避免僵尸进程残留
4. 实时反馈：将PTY输出实时回显到终端，用户可直观看到认证流程状态

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