如何终止通过asyncssh以root身份启动的命令进程？

问题重现 #

测试代码如下：

import asyncio
import asyncssh

async def main() -> None:
    async with asyncssh.connect("localhost", username="root", request_pty="force") as conn:
        print("start process, now try to terminate it by pressing CTRL-C")
        async with conn.create_process("sleep 20") as process:
            await process.wait()

asyncio.run(main())

现象：

• 启动脚本后，通过watch -n 1 pgrep -a sleep可观测到sleep进程正常创建
• 按CTRL-C终止脚本时，第一次操作会导致程序挂起，第二次触发KeyboardInterrupt异常，但远程的sleep进程仍未终止
• 以普通用户身份运行脚本（移除username="root"）时一切正常，直接执行ssh root@localhost -t sleep 20也能正常终止进程

原因分析 #

核心问题出在root用户的PTY信号转发逻辑上：

• 普通用户环境下，PTY的终端配置（如tty的intr键映射）会自动将本地的CTRL-C转化为远程的SIGINT信号传递给进程
• 而root用户的PTY默认配置可能屏蔽了这种自动信号转发，同时AsyncSSH的create_process方法默认不会主动将本地信号同步到远程进程
• 直接使用ssh -t命令时，OpenSSH会自动处理PTY的信号转发流程，但AsyncSSH需要手动显式配置或触发信号传递

解决方案 #

方案一：手动捕获中断信号并发送给远程进程 #

修改代码捕获KeyboardInterrupt异常，主动调用process.send_signal()向远程进程发送SIGINT：

import asyncio
import asyncssh
import signal

async def main() -> None:
    async with asyncssh.connect("localhost", username="root", request_pty="force") as conn:
        print("start process, now try to terminate it by pressing CTRL-C")
        async with conn.create_process("sleep 20") as process:
            try:
                await process.wait()
            except KeyboardInterrupt:
                # 发送SIGINT信号终止远程进程
                process.send_signal(signal.SIGINT)
                await process.wait()

asyncio.run(main())

方案二：配置PTY终端属性，让远程自动响应中断信号 #

在创建连接时指定PTY的终端属性，将intr键映射为CTRL-C（ASCII码3），确保远程PTY能识别中断指令：

import asyncio
import asyncssh
import termios

async def main() -> None:
    # 获取本地终端属性并修改中断键配置
    term_attrs = termios.tcgetattr(0)
    term_attrs[6][termios.VINTR] = 3  # 将VINTR设置为CTRL-C对应的ASCII码
    
    async with asyncssh.connect(
        "localhost", 
        username="root", 
        request_pty="force",
        term_attrs=term_attrs
    ) as conn:
        print("start process, now try to terminate it by pressing CTRL-C")
        async with conn.create_process("sleep 20") as process:
            await process.wait()

asyncio.run(main())

可选尝试：调整PTY请求参数 #

可尝试将request_pty="force"改为request_pty=True，让AsyncSSH自动适配root用户的PTY配置，但这种方式兼容性依赖环境，优先推荐前两种方案。

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