paramiko或者pexpect登录到设备并打开shell执行交互式命令。完整示例python自动化运维框架封装paramiko和pexpect-其它文档类资源-CSDN下载
class PMDeivce(object):
def __init__(self, ip, username, password, logfile='', timeout = 200):
self.ip = ip
self.username = username
self.password = password
self.shell = None
self.logfile = ''
#print("new to device: %s OKn"%self.ip)
self.def_prompt = ''
self.logfile = ''
if not self.logfile:
t=datetime.now().strftime("%Y-%m-%d-%H%M%S")
tmp_dir = os.environ['TEMP'] if os.name == 'nt' else os.environ['TMP']
self.logfile= os.path.join(tmp_dir, 'sshclient_' + re.sub('.','_', self.ip) + '_' + t + '.log')
print("logfile:%sn"%self.logfile)
paramiko.util.log_to_file(self.logfile)
self.timeout = timeout
self.wait_factor = 1
def connect_to_device(self, connection='SSH', debug=False): #set default prompt def read_device(self, prompt=None, alt_prompt='--More--', timeout=90, debug=False, verbose=False, wait_factor = None, strip_prompt=True, strip_command=True): wait_factor = wait_factor if wait_factor else self.wait_factor def login_to_cli(self, username='superman', password='talent', prompt='# ', debug=False,verbose=False, timeout=160): self.shell.send(password+'n') def execute_cmd(self, cmd): #多条命令之间以逗号隔开 def ssh_close(self): def sftp_connect(self): def sftp_close(self): def get_localfile(self, local_dir): def sfpt_putfile(self, local_dir, remote_dir): def __del__(self): class Device(): def connect_to_device(self, connection='SSH', debug = False): ''' self.connection = connection result_str = "" #connect to device try: if i == 0: else: def escape_ansi(self, line): debug_log("login to CLI as username: {},password: {}, prompt: {}<<< self.child.sendline(password) ''' if prompt: if debug: self.child.sendline(command) #result_str = self.child.before while True: #result_str += self.child.before + self.child.after return result_str
'''
connect to targtet host with username and password
'''
if not self.shell:
try:
#print(f"login to {self.ip} with username: {self.username}, password: {self.password}: n")
self.ssh = paramiko.SSHClient()
self.ssh.load_system_host_keys()
self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
self.ssh.connect(hostname=self.ip, port=22, username=self.username, password=self.password,timeout=60,allow_agent=False,look_for_keys=False)
#break
#self.sshclient = self.ssh
#self.ssh.get_pty()
self.shell = self.ssh.invoke_shell()
self.shell.setblocking(False)
self.shell.settimeout(self.timeout)
ticks = 0
resp = ''
while True:
recv_flag = self.shell.recv_ready()
if recv_flag:
resp = strip_ansi_codes(self.shell.recv(1024).decode('utf-8')).lstrip();
break
else:
time.sleep(self.wait_factor)
ticks += self.wait_factor
if ticks > self.timeout:
return None
time.sleep(1)
resp = strip_ansi_codes(self.shell.recv(1024).decode('utf-8')).lstrip() if not resp else resp
debug_log("connected to shell of device: %s OK, server respond: n%s<<<<
self.def_prompt = resp
#self.execute_command_on_device(command='date',debug=True)
#self._clear(debug=debug)
return self.shell
except Exception as e:
print ('Failed to connect to shell {} with username:{}, password: {}:n{}<<
return None
def _clear(self,debug=False):
try:
self.shell.send('n')
self.read_device(debug=True)
except (PipeTimeout, socket.timeout):
pass
'''
'''
results = ''
ticks = 0
#print("inside read_device ():got prompt value: %sn"%prompt)
prompt = self.def_prompt if not prompt else prompt
#print("inside read_device ():got prompt value: %sn"%prompt)
if alt_prompt:
prompt = "({}|{})".format(prompt, alt_prompt)
while True:
recv_flag = self.shell.recv_ready()
if recv_flag:
#debug_log("paramiko shell receive flag is ready:n",location=sys._getframe().f_code.co_name,debug=debug)
ret = strip_ansi_codes(self.shell.recv(65535).decode('utf-8'))
#strip prompt: --More--
results += re.sub(alt_prompt, '', ret)
debug_log("read from device: n%s<<<<
self.shell.send(' n')
continue
elif re.search(prompt, ret):
break
else:
#wait
debug_log("paramiko shell receive flag is NOT ready:nWAITING...",debug=debug)
time.sleep(wait_factor)
ticks += wait_factor
if ticks > timeout:
break
return results
debug_log("login to CLI as username: {},password: {}, prompt: {}<<<
#self.child.sendline(command + 'n')
self.read_device(prompt='assword: ', timeout=timeout, debug=debug, verbose=verbose)
ret = self.read_device(prompt='# ', timeout=timeout, debug=debug, verbose=verbose).lstrip()
debug_log("Done login to CLI as username: {},password: {}, prompt: {}<<<
#set default prompt
self.def_prompt = ret
return ret
#end of: def login_to_cli():
def execute_command_on_device(self, command, prompt=None, alt_prompt='--More--', delay=None, timeout=160, wait_factor = None, debug=False, verbose=False, strip_prompt=True, strip_command=True):
'''
'''
wait_factor = wait_factor if wait_factor else self.wait_factor
#print("inside execute_command_on_device(): got default prompt value: %sn"%self.def_prompt)
prompt = self.def_prompt if not prompt else prompt
debug_log("inside execute_command_on_device(): got prompt value: %s, cmd: %sn"%(prompt,command),location=sys._getframe().f_code.co_name,debug=debug)
self.shell.send(command+'n')
if delay:
time.sleep(int(delay))
#print("send command: %s to device: %s OKn"%(command,self.ip))
#debug_log("send command: %s to device: %s OKn"%(command,self.ip), location=sys._getframe().f_code.co_name,debug=debug)
results = self.read_device(debug=debug, verbose=verbose, alt_prompt='--More--', timeout=timeout)
if strip_prompt:
return re.sub(prompt,'', results) # 最后一行是linux的SP1输入提示符，没用丢弃
else:
return results
#execute_command_on_device()
stdin, stdout, stderr = self.ssh.exec_command(cmd)
res, err = stdout.read(), stderr.read()
result = res if res else err
return result.decode()
self.shell.close()
self.ssh.close()
'''
sftp
'''
try:
self.t = paramiko.Transport((self.ip, 22))
self.t.connect(username=self.username, password=self.password)
self.sftp = paramiko.SFTPClient.from_transport(self.t)
except Exception as e:
print (self.ip, e)
self.close()
allfiles = list()
for filepath, dirs, files in os.walk(local_dir, topdown=True):
for file in files:
filename = os.path.join(filepath, file)
allfiles.append(filename)
return allfiles
try:
if remote_dir[-1] != '/':
remote_dir = remote_dir + "/"
all_files = self.get_localfile(local_dir)
for file in all_files:
file1 = os.path.basename(file)
remote_filename = os.path.join(remote_dir, file1)
print (remote_filename)
try:
self.sftp.stat(remote_path)
except:
print ("远程服务器无文件夹，请先创建...")
self.sftp.put(file, remote_filename)
except:
print(traceback.format_exc())
self.shell.close()
self.ssh.close()
def __init__(self, ip, username, password, connection='SSH', timeout=180):
'''
connect: SSH|TELNET
'''
self.ip = ip
self.username = username
self.password = password
self.timeout = timeout if timeout else 120
self.logfile = ''
'''
connection: SSH|TELNET
print("Connecting to: {} with username: {}, password: {} via {}".format(self.ip, self.username, self.password, connection))
if self.connection == 'SSH':
os.system('rm -rf /root/.ssh/known_hosts')
login_str = 'ssh %s@%s' % (self.username,self.ip)
elif self.connection == 'TELNET':
login_str = 'telnet %s' % (self.ip)
child = pexpect.spawn(login_str, timeout=self.timeout)
if debug:
t=datetime.now().strftime("%Y-%m-%d-%H%M%S")
tmp_dir = os.environ['TEMP'] if os.name == 'nt' else os.environ['TMP']
self.logfile= os.path.join(tmp_dir, 'Exp_' + re.sub('.','_', self.ip) + '_' + t + '.log')
child.logfile = open(self.logfile,'w')
print("logfile:%sn"%self.logfile)
i = child.expect(['[Pp]assword:', 'continue connecting (yes/no)?','login:'],timeout = 600)
if i == 0 :
child.sendline(self.password)
elif i == 1:
child.sendline('yes')
child.expect('[Pp]assword: ')
child.sendline(self.password)
elif i == 2 :
child.sendline(self.username)
child.expect('[Pp]assword: ')
child.sendline(self.password)
child.setwinsize(24,500)
i= child.expect(['[$#>]\s+', pexpect.EOF, pexpect.TIMEOUT])
#print("got default prompt id: %d, %s<<
print("not got prompt, id: %d, %s<<
except:
print("login device: {} failed with: {}/{}, due to TIMEOUT or EOF".format(self.ip, self.username,self.password))
print("{}, {}n".format(child.before, child.after))
child = None
finally:
self.child = child
return child
#?2004hroot@localhost:~#
ansi_escape = re.compile(r'x1b[?dddddh')
return ansi_escape.sub('', line)
def set_default_prompt(self, prompt):
self.def_prompt = prompt
return self.def_prompt
def login_to_cli(self, username='superman', password='talent', prompt='# ', debug=False, timeout=60):
#self.child.sendline(command + 'n')
self.child.expect('assword: ', timeout=timeout)
j = self.child.expect(['# ', pexpect.EOF, pexpect.TIMEOUT], timeout=timeout)
if j == 0:
debug_log("login and get cli prompt: %s<<<
return True
else:
debug_log("failed to get cli prompt: %s<<<
def execute_command_on_device(self, command, prompt='', timeout=180, debug=False, verbose=False):
'''
execute_command_on_device: execute command on device
prompt = [prompt, pexpect.EOF, pexpect.TIMEOUT,'--More--']
else:
prompt = [self.def_prompt, pexpect.EOF, pexpect.TIMEOUT,'--More--']
print('execute_command_on_device: command: {}, prompt:{}<<
if self.child:
t = 0
flag = 0
#self.child.sendline(command + 'n')
j = self.child.expect(prompt, timeout=timeout)
result_str = ''
if debug and verbose:
print("execute_command_on_device() command: {},id: {}n{}<<<
if t == timeout:
flag = 1
if j == 0:
result_str += self.child.before + self.child.after
break
elif j == 0 and flag == 1:
print("There is no result in seconds " + str(t))
result_str += self.child.before + self.child.after
#print("command: {}, response:n{}<<<
elif j == 3:
result_str += self.child.before
t += 1
time.sleep(1)
self.child.send(' ')
j = self.child.expect(prompt, timeout=timeout)
continue
elif j == 1:
result_str = 'command running error:session closed!!!'
break
elif j == 2:
result_str = 'command running error:timeout!!!'
break
def __del__(self):
pass
