通过伪 TTY 的 Python SSH Clear

Question

我编写了以下 python 模块来处理程序中的 ssh 连接:

#!/usr/bin/env python
from vxpty import VX_PTY

class SSHError(Exception):
  def __init__(self, msg):
    self.msg = msg
  def __str__(self):
    return repr(self.msg)

class SSHShell:
  def __init__(self, host, port, user, password):
    self.host = host
    self.port = port
    self.user = user
    self.password = password
    self.authenticated = False
  def authenticate(self):
    self.tty = VX_PTY(['/usr/bin/ssh', 'ssh', '-p'+str(self.port), self.user+'@'+self.host])
    resp = self.tty.read()
    if "authenticity of host" in resp:
      self.tty.println('yes')
      while 1:
        resp = self.tty.read()
        if "added" in resp:
          break
      resp = self.tty.read()
    if "assword:" in resp:
      self.tty.println(self.password)
      tmp_resp = self.tty.read()
      tmp_resp += self.tty.read()
      if "denied" in tmp_resp or "assword:" in tmp_resp:
        raise(SSHError("Authentication failed"))
      else:
        self.authenticated = True
        self.tty.println("PS1=''")
    return self.authenticated
  def execute(self, os_cmd):
    self.tty.println(os_cmd)
    resp_buf = self.tty.read().replace(os_cmd+'\r\n', '')
    return resp_buf

它使用我之前编写的 pty 模块:

#!/usr/bin/env python
import os,pty

class PTYError(Exception):
  def __init__(self, msg):
    self.msg = msg
  def __str__(self):
    return repr(self.msg)

class VX_PTY:
  def __init__(self, execlp_args):
    self.execlp_args = execlp_args
    self.pty_execlp(execlp_args)
  def pty_execlp(self, execlp_args):
    (self.pid, self.f) = pty.fork()
    if self.pid==0:
      os.execlp(*execlp_args)
    elif self.pid<0:
      raise(PTYError("Failed to fork pty"))
  def read(self):
    data = None
    try:
      data = os.read(self.f, 1024)
    except Exception:
      raise(PTYError("Read failed"))
    return data
  def write(self, data):
    try:
      os.write(self.f, data)
    except Exception:
      raise(PTYError("Write failed"))
  def fsync(self):
    os.fsync(self.f)
  def seek_end(self):
    os.lseek(self.f, os.SEEK_END, os.SEEK_CUR)
  def println(self, ln):
    self.write(ln+'\n')

但是，每当我调用 execute() 方法时，我最终都会读取第一行的输出:

>>> import SSH;shell=SSH.SSHShell('localhost',22,'735tesla','notmypassword');shell.authenticate()
True
>>> shell.execute('whoami')
"\x1b[?1034hLaptop:~ 735Tesla$ PS1=''\r\n"
>>>

然后第二次调用 read() 我得到输出:

>>> shell.tty.read()
'whoami\r\n735Tesla\r\n'
>>> 

从输出中删除 whoami\r\n 不是问题，但是有没有办法清除输出，这样我就不必使用第一个命令调用 read 两次？

Answer 1

我认为你的问题比你意识到的更深刻。幸运的是，解决这个问题比您想象的要容易。

您似乎想要的是 os.read 在一次调用中返回 shell 必须发送给您的全部内容。这不是你能要求的。根据多种因素，包括但不限于 shell 的实现、网络带宽和延迟以及 PTY(您的和远程主机的)的行为，您将在每次调用 read 可以是所有内容，也可以是单个字符。

如果您只想接收命令的输出，您应该用唯一的标记将其括起来，并且不用担心弄乱 PS1。我的意思是，您需要让 shell 在命令执行之前输出一个唯一的字符串，并在命令执行之后输出另一个字符串。然后，您的 tty.read 方法应该返回它在这两个标记字符串之间找到的所有文本。让 shell 输出这些唯一字符串的最简单方法就是使用 echo 命令。

对于多行命令，您必须将命令包装在 shell 函数中，并在执行该函数之前和之后回显标记。

简单的实现如下:

def execute(self, cmd):
    if '\n' in cmd:
        self.pty.println(
            '__cmd_func__(){\n%s\n' % cmd +
            '}; echo __"cmd_start"__; __cmd_func__; echo __"cmd_end"__; unset -f __cmd_func__'
        )
    else:
        self.pty.println('echo __"cmd_start"__; %s; echo __"cmd_end"__' % cmd)

    resp = ''
    while not '__cmd_start__\r\n' in resp:
        resp += self.pty.read()

    resp = resp[resp.find('__cmd_start__\r\n') + 15:] # 15 == len('__cmd_start__\r\n')

    while not '_cmd_end__' in resp:
        resp += self.pty.read()

    return resp[:resp.find('__cmd_end__')]