Subprocess Popen执行SSH命令异常：含空格命令被截断

问题根源分析 #

你的代码存在两个核心问题，导致含空格的命令被远程shell错误解析：

1. 命令发送冗余空格：execute_ssh_cmd中发送命令后额外添加了" \n"，结合初始化时发送的空命令，可能让远程shell进入异常输入状态，将命令参数识别为独立命令。
2. 输出过滤逻辑误判：if command.strip() in line.strip(): continue会跳过所有包含命令片段的行，若远程shell回显命令时出现拆分，会导致错误过滤，甚至让你误以为命令被截断。

从错误输出[': No such file or directory']可推断：远程shell实际将-l当成了独立命令执行（而非ls的参数），说明完整命令未被正确识别。

修正后的代码 #

1. 核心类修正 #

import threading
import subprocess
import queue

class ReadOutputThread(threading.Thread):
    def __init__(self, file_obj, output_queue):
        super().__init__()
        self.file_obj = file_obj
        self.output_queue = output_queue
        self.daemon = True

    def run(self):
        # 仅移除换行符，保留原始行格式，便于排查错误细节
        for line in iter(self.file_obj.readline, ''):
            self.output_queue.put(line.rstrip('\n'))
        self.file_obj.close()

class YourClass:
    def __init__(self, username: str, ssh_key: str, hostname: str="1.1.1.1"):
        self.ssh_session = subprocess.Popen(
            ['ssh', '-T', '-i', ssh_key, '-Y', f'{username}@{hostname}'],
            stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
            universal_newlines=True, shell=False
        )
        self.output_queue = queue.Queue()
        self.stdout_thread = ReadOutputThread(self.ssh_session.stdout, self.output_queue)
        self.stderr_thread = ReadOutputThread(self.ssh_session.stderr, self.output_queue)
        self.stdout_thread.start()
        self.stderr_thread.start()
        # 用明确的测试命令替代空命令，确认连接状态
        self.execute_ssh_cmd("echo 'SSH连接已建立'")

    def execute_ssh_cmd(self, command):
        try:
            print(f"Executing command: {command}")

            # 直接发送命令加换行，移除冗余空格
            self.ssh_session.stdin.write(f"{command}\n")
            self.ssh_session.stdin.flush()

            output_lines = []
            # 添加唯一标记命令，精准判断命令执行完成
            marker = f"CMD_DONE_{id(command)}"
            self.ssh_session.stdin.write(f"echo {marker}\n")
            self.ssh_session.stdin.flush()

            while True:
                try:
                    line = self.output_queue.get(timeout=2)
                    if line == marker:
                        # 收到标记，停止读取
                        break
                    # 仅跳过与命令完全一致的回显行，避免误过滤
                    if line.strip() == command.strip():
                        continue
                    output_lines.append(line)
                except queue.Empty:
                    print("等待命令输出超时")
                    break

            return output_lines
        except Exception as e:
            print("SSH会话写入错误:", e)
            return None

2. 主代码使用示例 #

ssh_key_path = "/path/to/your/ssh/key"
hostname = "1.1.1.1"
username = 'user'

your_instance = YourClass(username, ssh_key_path, hostname)
output1 = your_instance.execute_ssh_cmd('ls -l')
print("Final output 1:", output1)

output2 = your_instance.execute_ssh_cmd("pwd")
print("Final output 2:", output2)

关键改进点 #

• 移除冗余空格：命令直接以\n结尾，确保远程shell接收到完整命令行。
• 保留原始输出格式：仅移除换行符，保留其他空格，便于查看完整错误信息。
• 添加执行完成标记：通过唯一的echo命令标记执行结束，避免超时过早终止读取。
• 精准过滤回显：仅跳过与命令完全一致的回显行，防止误过滤输出中包含命令文本的内容。

更可靠的替代方案 #

如果允许使用第三方库，推荐使用paramiko（专业SSH客户端库），原生支持持久连接和命令执行，无需手动处理线程与缓冲区：

import paramiko

class SSHSession:
    def __init__(self, username, ssh_key_path, hostname="1.1.1.1"):
        self.client = paramiko.SSHClient()
        self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        self.client.connect(hostname, username=username, key_filename=ssh_key_path)
        self.shell = self.client.invoke_shell()
        # 等待shell初始化完成
        while not self.shell.recv_ready():
            pass

    def execute_cmd(self, command):
        self.shell.send(f"{command}\n")
        output = []
        while True:
            while self.shell.recv_ready():
                data = self.shell.recv(1024).decode('utf-8')
                output.append(data)
            if self.shell.recv_stderr_ready():
                data = self.shell.recv_stderr(1024).decode('utf-8')
                output.append(data)
            # 根据shell提示符判断执行完成（可根据实际环境调整）
            if "$ " in ''.join(output) or "# " in ''.join(output):
                break
        # 过滤回显与提示符
        full_output = ''.join(output).split(f"{command}\n")[-1].rstrip("$ # \n")
        return full_output.split('\n')

# 使用示例
ssh_session = SSHSession(username, ssh_key_path, hostname)
print(ssh_session.execute_cmd('ls -l'))
print(ssh_session.execute_cmd('pwd'))

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