subprocess.Popen（「ssh主機@遠程cmd」）失敗

Question

我是新的python子流程。當我想使用Python subprocess.Popen.communicate來完成交互的passwd與外殼CMD「網絡廣告加入-U管理員」，它的輸出不重定向到管道，但在stdout.My代碼下：

import subprocess 

import os 
import sys 

passwd = b'123456' 
cmd = "net ads join -U administrator" 
s1 = ("%s\n" % str(passwd)).encode('utf-8') 
p1 = subprocess.Popen(cmd1,shell=True,stdin=subprocess.PIPE,stdout=subprocess.PIPE, 
            stderr=subprocess.PIPE,close_fds=True) 
out,err = p1.communicate(s1) 
print(out) 

當我運行測試時，通信不起作用，輸出仍在stdout中。

$ python ads.py 
Enter administrator's password: 

當我將cmd更改爲「ssh root @ ip ls /」時，我也遇到了問題。

我試過pdb，但問題似乎不在這裏。 有任何建議來處理這個問題？謝謝

Answer 1

這裏是一個用於遠程SSH和遠程運行命令的片段類，我在其中的一個腳本中使用過，請讓我知道是否可行，這裏我使用python標準PTY Module作爲僞終端。

import pty 
import os 
import re 

class BaseError(Exception): 
    pass 

class SSHError(BaseError): 
    pass 

class Ssh: 

    """SSH on the box""" 

    def __init__(self, ip, passwd, user, port): 
     self.ip = ip 
     self.passwd = passwd 
     self.user = user 
     self.port = port 

    def run_cmd(self, c): 
     (pid, f) = pty.fork() 
     if pid == 0: 
      os.execlp(
       "ssh", "ssh", '-o ConnectTimeout=10', 
       '-o StrictHostKeyChecking=no', 
       '-o UserKnownHostsFile=/dev/null', '-q', 
       '-p %d' % self.port, 
       self.user + '@' + self.ip, c) 
     else: 
      return (pid, f) 

    def run_scp(self, src, dst): 
     (pid, f) = pty.fork() 
     if pid == 0: 
      os.execlp(
       "scp", "scp", '-o StrictHostKeyChecking=no', 
       '-o UserKnownHostsFile=/dev/null', '-q', 
       '-P %d' % self.port, "-r", src, 
       self.user + '@' + self.ip + ':' + dst) 
     else: 
      return (pid, f) 

    def read(self, f): 
     x = "" 
     try: 
      x = os.read(f, 2024) 
     except Exception: 
      pass 
     return x 

    def sshOutputs(self, pid, f): 

     output = "" 
     readOutLoud = self.read(f) 
     m = re.search("authenticity of host", readOutLoud) 

     if m: 
      os.write(f, 'yes\n') 
      while True: 
       readOutLoud = self.read(f) 
       m = re.search("Permanently added", readOutLoud) 
       if m: 
        break 

      readOutLoud = self.read(f) 
     m = re.search("assword:", readOutLoud) 

     if m: 
      os.write(f, self.passwd + '\n') 
      tmp = self.read(f) 
      tmp += self.read(f) 
      m = re.search("Permission denied", tmp) 
      if m: 
       raise SSHError("Invalid passwd") 
      p = re.search("[pP]assword", tmp) 
      if p: 
       raise SSHError("Invalid passwd") 

      # passwd was accepted 
      readOutLoud = tmp 

     while readOutLoud and len(readOutLoud) > 0: 
      output += readOutLoud 
      readOutLoud = self.read(f) 

     os.waitpid(pid, 0) 
     os.close(f) 

     return output 

    def exe_ssh_cmd(self, c): 
     (pid, f) = self.run_cmd(c) 
     return self.sshOutputs(pid, f) 

    def exe_scp_cmd(self, src, dst): 
     (pid, f) = self.run_scp(src, dst) 
     return self.sshOutputs(pid, f)