如何在 python 中使用 paramiko 发送控制信号？

Question

我有一个类似这样的代码片段：

ssh = paramiko.SSHClient()
ssh.load_system_host_keys()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(ip,port=Port, username=usr,password=Psw)
stdin, stdout, stderr= ssh.exec_command("watch -n1 ps")
print stdout.read(),stderr.read()

这里的问题是我必须运行 watch 或任何无限运行ning 命令 10 秒，然后我应该发送 SIGINT(Ctrl + c ) 并打印状态。

我该怎么做？

Answer 1

传输信号的唯一方法是通过 terminal，来自 ssh man:

  -t   Force pseudo-tty allocation.  This can be used to execute
       arbitrary screen-based programs on a remote machine, which can be
       very useful, e.g. when implementing menu services.  Multiple -t
       options force tty allocation, even if ssh has no local tty.

对于 paramiko 检查：http://docs.paramiko.org/en/1.16/api/channel.html

get_pty(*args, **kwds)
Request a pseudo-terminal from the server. This is usually used right after creating a client channel, to ask the server to provide
some basic terminal semantics for a shell invoked with invoke_shell. It isn’t necessary (or desirable) to call this method if you’re going to exectue a single command with exec_command.

Answer 2

解决此问题的一种方法是打开您自己的会话、伪终端，然后以非阻塞方式读取，使用 recv_ready() 知道何时读取。 10 秒后，您发送 ^C (0x03) 以终止运行ning 进程，然后关闭会话。由于您无论如何都要关闭会话，因此发送 ^C 是可选的，但如果您想让会话保持活动状态并多次执行运行命令，它可能会有用。

import paramiko
import time
import sys

ssh = paramiko.SSHClient()
ssh.load_system_host_keys()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(ip, port=Port, username=usr,password=Psw)

transport = ssh.get_transport()
session = transport.open_session()
session.setblocking(0) # Set to non-blocking mode
session.get_pty()
session.invoke_shell()

# Send command
session.send('watch -n1 ps\n')

# Loop for 10 seconds
start = time.time()    
while time.time() - start < 10:
  if session.recv_ready():
    data = session.recv(512)

    sys.stdout.write(data)
    sys.stdout.flush() # Flushing is important!

  time.sleep(0.001) # Yield CPU so we don't take up 100% usage...

# After 10 seconds, send ^C and then close
session.send('\x03')
session.close()
print

Answer 3

从 OpenSSH 7.9 开始，支持 signal requests，因此使用私有 API 可以发送 custom-tailored 信号消息。注：支持是在2018-10-19添加的，大约在提问后三年。

import os
import sys
import time
from signal import Signals

import paramiko
from paramiko.common import cMSG_CHANNEL_REQUEST

ssh_client = paramiko.SSHClient()
ssh_client.load_system_host_keys()
ssh_client.connect(
    hostname=sys.argv[1],
    port=sys.argv[2],
    username=sys.argv[3],
    key_filename=str(sys.argv[4])
)

stdin, stdout, stderr = ssh_client.exec_command(
    command=f"tail -f {os.getenv('HOME')}/test.txt",
    bufsize=0,
    get_pty=False
)
channel = stdin.channel
time.sleep(0.5)

message = paramiko.Message()
message.add_byte(cMSG_CHANNEL_REQUEST)
message.add_int(stdin.channel.remote_chanid)
message.add_string("signal")
message.add_boolean(False)
message.add_string(Signals.SIGTERM.name[3:])
channel.transport._send_user_message(message)  # angry pylint noises

while not channel.exit_status_ready():
    time.sleep(0.1)

print(f"{channel.exit_status=}")
print("stdout:")
print(stdout.read())
print()
print("channel.stderr:")
print(stderr.read())

输出：

> python3 test_ssh_sigterm.py localhost 2222 $USER keyfile
channel.exit_status=-1
stdout:
b'some string\nsome other string\n'

channel.stderr:
b''

与resize_pty()来源比较

    def resize_pty(self, width=80, height=24, width_pixels=0, height_pixels=0):
        """
        Docstring was there.
        """
        m = Message()
        m.add_byte(cMSG_CHANNEL_REQUEST)
        m.add_int(self.remote_chanid)
        m.add_string("window-change")
        m.add_boolean(False)
        m.add_int(width)
        m.add_int(height)
        m.add_int(width_pixels)
        m.add_int(height_pixels)
        self.transport._send_user_message(m)

如果您有疑问，为什么代码中没有 send_singal，好吧，现在是 PR from 2019 (not mine). It's on Next feature release 里程碑，希望今年会出现。

如何在 python 中使用 paramiko 发送控制信号？

How to send control signals using paramiko in python?

python

paramiko