这是一个用于 SSH 和远程运行命令的片段类,我在我的一个脚本中使用了它,如果它有效,请告诉我,这里我使用 python 标准PTY 模块作为伪终端。
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)