如何用AsyncSSH实现交互式su -c command -操作？

我已经用Paramiko成功执行su -c whoami -，实现代码如下：

def amiroot(ssh: paramiko.client.SSHClient, root_pass: str) -> bool:
    session = ssh.get_transport().open_session()
    session.set_combine_stderr(True)
    session.get_pty()
    # print("Sending su -c whoami -")
    session.exec_command("su -c whoami -")
    stdin = session.makefile('wb', -1)
    stdout = session.makefile('rb', -1)
    while not re.search(b"[Pp]assword", session.recv(1024)):
        time.sleep(1)
    # print("Sending password")
    stdin.write(root_pass + "\n")
    stdin.flush()
    start = time.monotonic()
    while not stdout.channel.eof_received:
        time.sleep(1)
        if (time.monotonic() - start) > 10:
            # print("STDOUT timeout")
            stdout.channel.close()
            break
    # lines = list(filter(None, (lb.decode("utf-8").strip() for lb in stdout.readlines())))
    # print(lines)
    return stdout.read().decode("utf-8").strip() == "root"

该逻辑可行，但我需要针对多目标异步执行，因此改用AsyncSSH编写了如下代码：

async def amiroot(address: str, username: str, password: str, rootpass: str):
    async with asyncssh.connect(
        address,
        username=username,
        password=password,
        known_hosts=None,
        connect_timeout=10,
        login_timeout=10,
    ) as conn:
        print(f"Connected to {address}, sending su")
        async with conn.create_process("su -c whoami -", term_type="xterm") as process:
            print(f"{address}: su sent, waiting response")  ###
            while "password" not in (await process.stdout.read()):
                await asyncio.sleep(1)
            await process.stdin.write(rootpass + "\n")
            print(f"{address}: rootpass sent, waiting response")
            print(await process.stdout.read())

但程序卡在标记###的行，无法执行到“rootpass sent”步骤，输出示例如下：

Connected to 1.2.3.4, sending su
1.2.3.4: su sent, waiting response

请问如何用AsyncSSH实现类似Paramiko的su -c ... -交互式操作？

代码卡住的核心问题是await process.stdout.read()会一直阻塞直到流结束（EOF），而su在等待密码输入时并不会触发EOF，导致循环永远无法退出。另外，部分系统中su的密码提示会输出到stderr而非stdout，Paramiko里你设置了合并标准错误和输出，AsyncSSH也需要做类似配置。

修正后的代码如下：

import asyncio
import asyncssh
import re

async def amiroot(address: str, username: str, password: str, rootpass: str) -> bool:
    async with asyncssh.connect(
        address,
        username=username,
        password=password,
        known_hosts=None,
        connect_timeout=10,
        login_timeout=10,
    ) as conn:
        print(f"Connected to {address}, sending su")
        # 合并stderr到stdout，开启pty（对应Paramiko的get_pty）
        async with conn.create_process(
            "su -c whoami -",
            term_type="xterm",
            stderr=asyncssh.STDOUT
        ) as process:
            print(f"{address}: su sent, waiting response")
            received = b""
            start = asyncio.get_event_loop().time()
            # 循环读取内容，直到匹配到密码提示或超时
            while not re.search(rb"[Pp]assword", received):
                try:
                    chunk = await asyncio.wait_for(process.stdout.read(1024), timeout=1)
                    if not chunk:
                        print(f"{address}: No password prompt received")
                        return False
                    received += chunk
                except asyncio.TimeoutError:
                    pass
                if asyncio.get_event_loop().time() - start > 10:
                    print(f"{address}: Timeout waiting for password prompt")
                    return False
            
            # 发送root密码并确保数据写入完成
            await process.stdin.write(rootpass + "\n")
            await process.stdin.drain()
            print(f"{address}: rootpass sent, waiting response")
            
            # 读取执行结果
            output = b""
            start = asyncio.get_event_loop().time()
            while not process.stdout.at_eof():
                try:
                    chunk = await asyncio.wait_for(process.stdout.read(1024), timeout=1)
                    if chunk:
                        output += chunk
                except asyncio.TimeoutError:
                    pass
                if asyncio.get_event_loop().time() - start > 10:
                    print(f"{address}: Timeout waiting for command output")
                    break
            
            result = output.decode("utf-8").strip()
            print(f"{address}: Command result: {result}")
            return result == "root"

关键改动说明： #

• 使用stderr=asyncssh.STDOUT合并标准错误和输出，确保能捕获到su的密码提示（无论它输出到哪个流）
• 用process.stdout.read(1024)代替read()，每次读取固定长度内容，避免阻塞到EOF
• 添加双层超时逻辑：单次读取超时+整体等待超时，避免无限挂起
• 发送密码后调用process.stdin.drain()，确保所有数据都已写入通道，防止丢失

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