paramiko模塊

paramiko模塊提供了ssh及sft進行遠程登錄服務器執行命令和上傳下載文件的功能。這是一個第三方的軟件包，使用之前需要安裝。

import paramiko

# ssh root@ip

# 創建一個ssh對象

client = paramiko.SSHClient()

#如果第一次連接陌生的IP,自動選擇yes確認連接

client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

# 連接服務器

client.connect(hostname='111.231.215.66',port=22,username='手動屏蔽帳號',password='手動屏蔽密碼')

# 執行操作

stdin,stdout,stderr = client.exec_command('hostname')

# 獲取命令執行結果

host = stdout.read().decode('utf-8').strip()

print(host)

while True:

command = input("[root@%s]# " %host)

if command == 'exit()':

break

else:

r = client.exec_command(command)[1].read().decode('utf-8')

print(r)

# 關閉連接

client.close()

例子1：

連接文件 主機信息.txt 中的所有主機

不能連接返回連接失敗，成功連接，返回其主機名

把連接情況寫入文件。

import paramiko

client = paramiko.SSHClient()

client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

def ssh_info(ip,port=22,user='root',passwd='westos'):

try:

client.connect(hostname=ip,port=port,username=user,password=passwd)

r = client.exec_command('hostname')[1].read().decode('utf-8')

except:

print('連接失敗......')

return '連接失敗......'

else:

client.close()

print('連接成功......','\n主機名：%s' %r)

return '連接成功......'+'\n主機名：%s' %r

with open('連接狀況.txt','w') as f,open('主機信息.txt') as p:

for line in p:

line = line.strip()

ip, port, user, passwd = line.split(':')

s = ' 正在連接%s '.center(30, '+') % ip

f.write(s+'\n')

f.write(ssh_info(ip,port,user,passwd)+'\n')

例子2：

基于公鑰私鑰的批量連接

import paramiko

client = paramiko.SSHClient()

client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

def ssh_info(ip,pkey,port=22,user='root'):

try:

print(' 正在連接%s '.center(30, '+') % ip)

client.connect(hostname=ip,port=port,username=user,pkey=pkey)

r = client.exec_command('hostname')[1].read().decode('utf-8')

except:

print('連接失敗......')

else:

client.close()

print('連接成功......','\n主機名：%s' %r)

pkey = paramiko.RSAKey.from_private_key_file('id_rsa')

for i in range(254):

ip = '172.25.254.'+str(i+1)

ssh_info(ip,pkey)

例子3：

基于公鑰私鑰的上傳下載

import paramiko

private_key = paramiko.RSAKey.from_private_key_file('id_rsa')

transport = paramiko.Transport(('172.25.254.39', 22))

transport.connect(username='root',pkey=private_key)

sftp = paramiko.SFTPClient.from_transport(transport)

#上川下載

sftp.put('/tmp/kiosk', '/mnt/kiosk2')

sftp.get('/mnt/kiosk2', '/home/kiosk/Desktop/day18/kiosk')

transport.close()

paramiko綜合練習實例1

import os

import paramiko

group = [file.rstrip('.conf') for file in os.listdir('host')]

print('主機組'.center(30,'+'))

for i,g in enumerate(group):

print('\t',str(i+1),'\t',g)

choice = input('選擇操作組：')

print('主機'.center(30,'+'))

host_info_list = [] # 存儲文件中主機信息

file = 'host/'+choice+'.conf'

with open(file) as f:

for line in f:

line =line.split(':')

print('\t',line[0])

host_info_list.append(line)

def do_cmd(cmd,hostname,port=22, username='root', password='westos'):

print('連接主機 ',hostname)

try:

client.connect(hostname, port, username, password)

stdin, stdout, stderr = client.exec_command(cmd)

result = stdout.read().decode('utf-8').strip()

except:

print('連接失敗......')

else:

client.close()

print(result)

def get_put(cmd,hostname,port=22, username='root', password='westos'):

print('連接主機 ', hostname)

try:

transport = paramiko.Transport((hostname, int(port)))

transport.connect(username=username, password=password)

sftp = paramiko.SFTPClient.from_transport(transport)

except:

print('連接失敗......')

else:

# 上川下載

if cmd[0] == 'put':

sftp.put(cmd[1], cmd[2])

print('上傳成功......')

elif cmd[0] == 'get':

sftp.get(cmd[1], cmd[2])

print('下載成功......')

else:

print('暫時沒有這個命令的操作......',cmd[0])

transport.close()

while True:

cmd = input('>>>: ')

cmd = cmd.split()

if cmd == []:

continue

elif len(cmd) == 1:

client = paramiko.SSHClient()

client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

for host in host_info_list:

do_cmd(cmd[0],host[0],host[1],host[2],host[3])

else:

for host in host_info_list:

get_put(cmd,host[0],host[1],host[2],host[3])

paramiko封裝實例1：

import os

import paramiko

class SSH_host:

def __init__(self,ip,port,u,p,cmd):

self.ip = ip

self.port = port

self.username = u

self.passwd = p

self.cmd = cmd

def cmd(self):

print('連接主機 ',self.ip)

client = paramiko.SSHClient()

client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

try:

client.connect(self.ip, self.port, self.username, self.passwd)

stdin, stdout, stderr = client.exec_command(self.cmd[0])

result = stdout.read().decode('utf-8').strip()

except:

print('連接失敗......')

else:

print(result)

client.close()

def get(self):

print('連接主機 ', self.ip)

try:

transport = paramiko.Transport((self.ip, self.port))

transport.connect(self.username, self.passwd)

sftp = paramiko.SFTPClient.from_transport(transport)

except:

print('連接失敗......')

else:

sftp.get(self.cmd[1], self.cmd[2])

print('下載成功......')

transport.close()

def put(self):

print('連接主機 ', self.ip)

try:

transport = paramiko.Transport((self.ip, self.port))

transport.connect(self.username, self.passwd)

sftp = paramiko.SFTPClient.from_transport(transport)

except:

print('連接失敗......')

else:

sftp.put(self.cmd[1], self.cmd[2])

print('上傳成功......')

transport.close()

if __name__ == "__main__":

group = [file.rstrip('.conf') for file in os.listdir('host')]

print('主機組'.center(30,'+'))

for i,g in enumerate(group):

print('\t',str(i+1),'\t',g)

choice = input('選擇操作組：')

print('主機'.center(30,'+'))

host_info_list = [] # 存儲文件中主機信息

file = 'host/'+choice+'.conf'

with open(file) as f:

for line in f:

line =line.split(':')

print('\t',line[0])

host_info_list.append(line)

while True:

cmd = input('>>>: ')

cmd = cmd.split()

for host in host_info_list:

link = SSH_host(host[0], int(host[1]), host[2], host[3], cmd)

if cmd == []:

continue

else:

if hasattr(link, cmd[0]):

print(cmd[0])

總結

以上是生活随笔為你收集整理的python paramiko sftp_Python学习—paramiko模块实现简单的ssh与sftp的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。