paramiko
1、安装
pip3 install paramiko
二、使用
SSHClient
用于连接远程服务器并执行基本命令
基于用户名密码连接:
import paramiko # 创建SSH对象 ssh = paramiko.SSHClient() # 允许连接不在know_hosts文件中的主机 ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # 连接服务器 ssh.connect(hostname=‘c1.salt.com‘, port=22, username=‘wupeiqi‘, password=‘123‘) # 执行命令 stdin, stdout, stderr = ssh.exec_command(‘ls‘) # 获取命令结果 result = stdout.read() # 关闭连接 ssh.close()
#SSHClient 封装 Transport import paramiko transport = paramiko.Transport((‘hostname‘, 22)) transport.connect(username=‘wupeiqi‘, password=‘123‘) ssh = paramiko.SSHClient() ssh._transport = transport stdin, stdout, stderr = ssh.exec_command(‘df‘) print stdout.read() transport.close()
基于公钥密钥连接:
import paramiko private_key = paramiko.RSAKey.from_private_key_file(‘/home/auto/.ssh/id_rsa‘) # 创建SSH对象 ssh = paramiko.SSHClient() # 允许连接不在know_hosts文件中的主机 ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # 连接服务器 ssh.connect(hostname=‘c1.salt.com‘, port=22, username=‘wupeiqi‘, key=private_key) # 执行命令 stdin, stdout, stderr = ssh.exec_command(‘df‘) # 获取命令结果 result = stdout.read() # 关闭连接 ssh.close()
#SSHClient 封装 Transport import paramiko private_key = paramiko.RSAKey.from_private_key_file(‘/home/auto/.ssh/id_rsa‘) transport = paramiko.Transport((‘hostname‘, 22)) transport.connect(username=‘wupeiqi‘, pkey=private_key) ssh = paramiko.SSHClient() ssh._transport = transport stdin, stdout, stderr = ssh.exec_command(‘df‘) transport.close()
SFTPClient
用于连接远程服务器并执行上传下载
基于用户名密码上传下载:
import paramiko transport = paramiko.Transport((‘hostname‘,22)) transport.connect(username=‘wupeiqi‘,password=‘123‘) sftp = paramiko.SFTPClient.from_transport(transport) # 将location.py 上传至服务器 /tmp/test.py sftp.put(‘/tmp/location.py‘, ‘/tmp/test.py‘) # 将remove_path 下载到本地 local_path sftp.get(‘remove_path‘, ‘local_path‘) transport.close()
基于公钥密钥上传下载:
import paramiko private_key = paramiko.RSAKey.from_private_key_file(‘/home/auto/.ssh/id_rsa‘) transport = paramiko.Transport((‘hostname‘, 22)) transport.connect(username=‘wupeiqi‘, pkey=private_key ) sftp = paramiko.SFTPClient.from_transport(transport) # 将location.py 上传至服务器 /tmp/test.py sftp.put(‘/tmp/location.py‘, ‘/tmp/test.py‘) # 将remove_path 下载到本地 local_path sftp.get(‘remove_path‘, ‘local_path‘) transport.close()
#!/usr/bin/env python # -*- coding:utf-8 -*- import paramiko import uuid class SSHConnection(object): def __init__(self, host=‘172.16.103.191‘, port=22, username=‘wupeiqi‘,pwd=‘123‘): self.host = host self.port = port self.username = username self.pwd = pwd self.__k = None def create_file(self): file_name = str(uuid.uuid4()) with open(file_name,‘w‘) as f: f.write(‘sb‘) return file_name def run(self): self.connect() self.upload(‘/home/wupeiqi/tttttttttttt.py‘) self.rename(‘/home/wupeiqi/tttttttttttt.py‘, ‘/home/wupeiqi/ooooooooo.py) self.close() def connect(self): transport = paramiko.Transport((self.host,self.port)) transport.connect(username=self.username,password=self.pwd) self.__transport = transport def close(self): self.__transport.close() def upload(self,target_path): # 连接,上传 file_name = self.create_file() sftp = paramiko.SFTPClient.from_transport(self.__transport) # 将location.py 上传至服务器 /tmp/test.py sftp.put(file_name, target_path) def rename(self, old_path, new_path): ssh = paramiko.SSHClient() ssh._transport = self.__transport # 执行命令 cmd = "mv %s %s" % (old_path, new_path,) stdin, stdout, stderr = ssh.exec_command(cmd) # 获取命令结果 result = stdout.read() def cmd(self, command): ssh = paramiko.SSHClient() ssh._transport = self.__transport # 执行命令 stdin, stdout, stderr = ssh.exec_command(command) # 获取命令结果 result = stdout.read() return result ha = SSHConnection() ha.run()
# 对于更多限制命令,需要在系统中设置 /etc/sudoers Defaults requiretty Defaults:cmdb !requiretty
import paramiko import uuid class SSHConnection(object): def __init__(self, host=‘192.168.11.61‘, port=22, username=‘alex‘,pwd=‘alex3714‘): self.host = host self.port = port self.username = username self.pwd = pwd self.__k = None def run(self): self.connect() pass self.close() def connect(self): transport = paramiko.Transport((self.host,self.port)) transport.connect(username=self.username,password=self.pwd) self.__transport = transport def close(self): self.__transport.close() def cmd(self, command): ssh = paramiko.SSHClient() ssh._transport = self.__transport # 执行命令 stdin, stdout, stderr = ssh.exec_command(command) # 获取命令结果 result = stdout.read() return result def upload(self,local_path, target_path): # 连接,上传 sftp = paramiko.SFTPClient.from_transport(self.__transport) # 将location.py 上传至服务器 /tmp/test.py sftp.put(local_path, target_path) ssh = SSHConnection() ssh.connect() r1 = ssh.cmd(‘df‘) ssh.upload(‘s2.py‘, "/home/alex/s7.py") ssh.close()
堡垒机
import paramiko import sys import os import socket import getpass from paramiko.py3compat import u # windows does not have termios... try: import termios import tty has_termios = True except ImportError: has_termios = False def interactive_shell(chan): if has_termios: posix_shell(chan) else: windows_shell(chan) def posix_shell(chan): import select oldtty = termios.tcgetattr(sys.stdin) try: tty.setraw(sys.stdin.fileno()) tty.setcbreak(sys.stdin.fileno()) chan.settimeout(0.0) log = open(‘handle.log‘, ‘a+‘, encoding=‘utf-8‘) flag = False temp_list = [] while True: r, w, e = select.select([chan, sys.stdin], [], []) if chan in r: try: x = u(chan.recv(1024)) if len(x) == 0: sys.stdout.write(‘\r\n*** EOF\r\n‘) break if flag: if x.startswith(‘\r\n‘): pass else: temp_list.append(x) flag = False sys.stdout.write(x) sys.stdout.flush() except socket.timeout: pass if sys.stdin in r: x = sys.stdin.read(1) import json if len(x) == 0: break if x == ‘\t‘: flag = True else: temp_list.append(x) if x == ‘\r‘: log.write(‘‘.join(temp_list)) log.flush() temp_list.clear() chan.send(x) finally: termios.tcsetattr(sys.stdin, termios.TCSADRAIN, oldtty) def windows_shell(chan): import threading sys.stdout.write("Line-buffered terminal emulation. Press F6 or ^Z to send EOF.\r\n\r\n") def writeall(sock): while True: data = sock.recv(256) if not data: sys.stdout.write(‘\r\n*** EOF ***\r\n\r\n‘) sys.stdout.flush() break sys.stdout.write(data) sys.stdout.flush() writer = threading.Thread(target=writeall, args=(chan,)) writer.start() try: while True: d = sys.stdin.read(1) if not d: break chan.send(d) except EOFError: # user hit ^Z or F6 pass def run(): default_username = getpass.getuser() username = input(‘Username [%s]: ‘ % default_username) if len(username) == 0: username = default_username hostname = input(‘Hostname: ‘) if len(hostname) == 0: print(‘*** Hostname required.‘) sys.exit(1) tran = paramiko.Transport((hostname, 22,)) tran.start_client() default_auth = "p" auth = input(‘Auth by (p)assword or (r)sa key[%s] ‘ % default_auth) if len(auth) == 0: auth = default_auth if auth == ‘r‘: default_path = os.path.join(os.environ[‘HOME‘], ‘.ssh‘, ‘id_rsa‘) path = input(‘RSA key [%s]: ‘ % default_path) if len(path) == 0: path = default_path try: key = paramiko.RSAKey.from_private_key_file(path) except paramiko.PasswordRequiredException: password = getpass.getpass(‘RSA key password: ‘) key = paramiko.RSAKey.from_private_key_file(path, password) tran.auth_publickey(username, key) else: pw = getpass.getpass(‘Password for %[email protected]%s: ‘ % (username, hostname)) tran.auth_password(username, pw) # 打开一个通道 chan = tran.open_session() # 获取一个终端 chan.get_pty() # 激活器 chan.invoke_shell() interactive_shell(chan) chan.close() tran.close() if __name__ == ‘__main__‘: run()
时间: 2024-10-13 17:09:22