从 asyncio 子进程获取实时输出

Question

我正在尝试使用 Python asyncio 子进程来启动交互式 SSH 会话并自动输入密码。实际用例并不重要，但它有助于说明我的问题。这是我的代码：

    proc = await asyncio.create_subprocess_exec(
        'ssh', 'user@127.0.0.1',
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.STDOUT,
        stdin=asyncio.subprocess.PIPE,
    )

    # This loop could be replaced by async for, I imagine
    while True:
        buf = await proc.stdout.read()
        if not buf:
            break
        print(f'stdout: { buf }')

我希望它能像 asyncio 流一样工作，我可以在其中创建两个任务/子例程/future，一个用于监听StreamReader（在本例中由 给出proc.stdout），另一个用于写入 StreamWriter ( proc.stdin)。

但是，它并没有按预期工作。ssh 命令的前几行输出直接打印到终端，直到出现密码提示（或主机密钥提示，视情况而定）并等待手动输入。我希望能够读取前几行，检查它是否要求输入密码或主机提示，并相应地写入 StreamReader。

它运行该行的唯一一次print(f'stdout: { buf }')是在我按 Enter 键后，当它打印时，显然，“stderr：b'主机密钥验证失败。\r\n'”。

我也尝试了推荐的proc.communicate()，它不像使用 StreamReader/Writer 那么简洁，但它有同样的问题：等待手动输入时执行冻结。

这实际上应该如何运作？如果这不是我想象的那样，为什么不呢？有没有什么方法可以实现这一点，而无需诉诸线程中的某种繁忙循环？

PS：我解释使用 ssh 只是为了清楚起见。我最终使用 plink 来实现我想要的功能，但我想了解如何使用 python 来运行任意命令。

Answer 1

如果其他人来到这里寻求更通用的问题答案，请参阅以下示例：

import asyncio

async def _read_stream(stream, cb):  
    while True:
        line = await stream.readline()
        if line:
            cb(line)
        else:
            break

async def _stream_subprocess(cmd, stdout_cb, stderr_cb):  
    process = await asyncio.create_subprocess_exec(*cmd,
            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)

    await asyncio.gather(
        _read_stream(process.stdout, stdout_cb),
        _read_stream(process.stderr, stderr_cb)
    )
    return await process.wait()


def execute(cmd, stdout_cb, stderr_cb):  
    loop = asyncio.get_event_loop()
    rc = loop.run_until_complete(
        _stream_subprocess(
            cmd,
            stdout_cb,
            stderr_cb,
    ))
    loop.close()
    return rc

if __name__ == '__main__':  
    print(execute(
        ["bash", "-c", "echo stdout && sleep 1 && echo stderr 1>&2 && sleep 1 && echo done"],
        lambda x: print("STDOUT: %s" % x),
        lambda x: print("STDERR: %s" % x),
    ))

Answer 2

这不是 asyncio 特有的问题。该ssh进程不与stdin和stdout流交互，而是直接访问TTY 设备，以确保密码输入得到适当的保护。

您可以通过三个选项来解决此问题：

• 不要使用ssh，而是使用其他一些 SSH 客户端，一种不需要 TTY 控制的客户端。对于 asyncio，您可以使用asyncssh库。该库直接实现 SSH 协议，因此不需要单独的进程，并且它直接接受用户名和密码凭据。

• 为 SSH提供一个伪 tty来与之通信，这是由 Python 程序控制的。该pexpect库提供了一个高级 API 来为您执行此操作，并可用于完全控制命令ssh。

• 设置一个可供 ssh 使用的替代密码提示器。如果没有 TTY，程序可以通过环境ssh变量让其他程序SSH_ASKPASS处理密码输入。大多数版本对于何时接受都ssh非常挑剔，但是，您也需要进行设置，使用命令行开关并使用该命令在新会话中运行，与任何 TTY 断开连接。SSH_ASKPASSDISPLAY-nssh setsidssh

  我之前在回答有关and的问题时描述了如何使用SSH_ASKPASSasyncio 。gitssh

阻力最小的路径是使用 pexpect，因为它原生支持 asyncio（任何接受的方法都async_=True可以用作协程）：

import pexpect

proc = pexpect.spawn('ssh user@127.0.0.1')
await child.expect('password:', timeout=120, async_=True) 
child.sendline(password_for_user)