如何使用paramiko保持ssh会话未过期？

2 回答

• 2

  您可以使用paramiko实现交互式shell，这样在远程shell上执行命令后通道不会关闭 .

  import paramiko
  import re
  
  
  class ShellHandler:
  
      def __init__(self, host, user, psw):
          self.ssh = paramiko.SSHClient()
          self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
          self.ssh.connect(host, username=user, password=psw, port=22)
  
          channel = self.ssh.invoke_shell()
          self.stdin = channel.makefile('wb')
          self.stdout = channel.makefile('r')
  
      def __del__(self):
          self.ssh.close()
  
      @staticmethod
      def _print_exec_out(cmd, out_buf, err_buf, exit_status):
          print('command executed: {}'.format(cmd))
          print('STDOUT:')
          for line in out_buf:
              print(line, end="")
          print('end of STDOUT')
          print('STDERR:')
          for line in err_buf:
              print(line, end="")
          print('end of STDERR')
          print('finished with exit status: {}'.format(exit_status))
          print('------------------------------------')
          pass
  
      def execute(self, cmd):
          """
  
          :param cmd: the command to be executed on the remote computer
          :examples:  execute('ls')
                      execute('finger')
                      execute('cd folder_name')
          """
          cmd = cmd.strip('\n')
          self.stdin.write(cmd + '\n')
          finish = 'end of stdOUT buffer. finished with exit status'
          echo_cmd = 'echo {} $?'.format(finish)
          self.stdin.write(echo_cmd + '\n')
          shin = self.stdin
          self.stdin.flush()
  
          shout = []
          sherr = []
          exit_status = 0
          for line in self.stdout:
              if str(line).startswith(cmd) or str(line).startswith(echo_cmd):
                  # up for now filled with shell junk from stdin
                  shout = []
              elif str(line).startswith(finish):
                  # our finish command ends with the exit status
                  exit_status = int(str(line).rsplit(maxsplit=1)[1])
                  if exit_status:
                      # stderr is combined with stdout.
                      # thus, swap sherr with shout in a case of failure.
                      sherr = shout
                      shout = []
                  break
              else:
                  # get rid of 'coloring and formatting' special characters
                  shout.append(re.compile(r'(\x9B|\x1B\[)[0-?]*[ -/]*[@-~]').sub('', line).
                               replace('\b', '').replace('\r', ''))
  
          # first and last lines of shout/sherr contain a prompt
          if shout and echo_cmd in shout[-1]:
              shout.pop()
          if shout and cmd in shout[0]:
              shout.pop(0)
          if sherr and echo_cmd in sherr[-1]:
              sherr.pop()
          if sherr and cmd in sherr[0]:
              sherr.pop(0)
  
          self._print_exec_out(cmd=cmd, out_buf=shout, err_buf=sherr, exit_status=exit_status)
          return shin, shout, sherr
  

  回复于 2024-04-26T14:37:50+08:00
• 3

  我看到你在连接调用中使用了 timeout 参数：

  ssh.connect(host, 22, user, passwd, timeout=3)
  

  从文档：

  timeout（float） - TCP连接的可选超时（以秒为单位）

  在我的一个脚本中，我只是这样做：

  ssh = paramiko.SSHClient()
  ssh.connect(host, username=settings.user)
  

  这使得连接保持打开状态直到我打电话

  ssh.close()
  

  回复于 2024-04-26T14:37:50+08:00