快捷導(dǎo)航

Python如何實(shí)現(xiàn)SSH遠(yuǎn)程連接與文件傳輸

更新時(shí)間：2023年05月30日 15:02:51 作者：冰點(diǎn)契約丶

這篇文章主要介紹了Python如何實(shí)現(xiàn)SSH遠(yuǎn)程連接與文件傳輸問(wèn)題，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教

Python SSH遠(yuǎn)程連接與文件傳輸

from paramiko import (SSHClient, SFTPClient, AutoAddPolicy)
import argparse
class Args(argparse.ArgumentParser):
    def __init__(self, help_info: str = "remote host login  args"):
        """
        使用 python xx.py -h 查看參數(shù)傳遞幫助
        :param help_info:
        """
        super(Args, self).__init__(description=help_info)
    def __call__(self, *args, **kwargs):
        """
        :param args:
        :param kwargs:
        :return: 返回參數(shù)對(duì)象,可通過(guò) args.xxx 獲取參數(shù)
        """
        self.add_argument("--ip", help="remote host ip address")
        self.add_argument("--username", help="SSH login username", default="")
        self.add_argument("--password", help="SSH login password", default="")
        self.add_argument("--port", help="remote host port", default=22)
        return self.parse_args()
class SSH(object):
    def __init__(self, ip_address: str, username: str, password: str, port: int = 22):
        """
        :param ip_address:遠(yuǎn)程ip地址
        :param username:用戶名
        :param password:密碼
        :param port:端口號(hào),默認(rèn)22
        """
        self.ip = ip_address
        self.username = username
        self.password = password
        self.port = port
        self.__client = SSHClient()
    def connect(self) -> None:
        """
        打開(kāi)連接
        :return:None
        """
        self.__client.set_missing_host_key_policy(AutoAddPolicy())
        self.__client.connect(self.ip, self.port, self.username, self.password)
    def execute(self, command: str) -> None:
        """
        執(zhí)行命令，stderr未啟用
        :param command: windows命令
        :return: None
        """
        std_in, stdout, stderr = self.__client.exec_command(command=command)
        print(stdout.read().decode("utf-8"))
    def upload_file(self, local_file_path: str, remote_file_path: str) -> None:
        """
        打開(kāi)sftp會(huì)話，用于將本地文件上傳到遠(yuǎn)程設(shè)備
        :param local_file_path: 本地文件絕對(duì)路徑
        :param remote_file_path: 遠(yuǎn)程文件路徑:命名方式:path+filename
        :return:
        """
        sftp: SFTPClient = self.__client.open_sftp()
        try:
            sftp.put(localpath=local_file_path, remotepath=remote_file_path)
            print(f"file:{local_file_path} upload success！")
        except Exception as e:
            print(f"upload file file,please check whether the file path is correct!\nerror massage：{e} ")
    def download_file(self, remote_file_path: str, local_save_path) -> None:
        """
        打開(kāi)sftp會(huì)話，用于將遠(yuǎn)程設(shè)備文件拉取到本地
        :param remote_file_path: 遠(yuǎn)程設(shè)備絕對(duì)路徑
        :param local_save_path: 本地文件保存路徑 命名方式:file +filename 注意需要指定文件名，否則報(bào)錯(cuò)
        :return:
        """
        sftp: SFTPClient = self.__client.open_sftp()
        try:
            sftp.get(remotepath=remote_file_path, localpath=local_save_path)
            print(f"file:{remote_file_path} download success!")
        except Exception as e:
            print(f"upload file file,please check whether the file path is correct!\nerror massage：{e} ")
    def get_shell(self) -> None:
        """
        獲取shell
        :return:
        """
        while True:
            command = input(f"{self.ip}@{self.username}$:")
            if command.__eq__("quit"):
                break
            self.execute(command=command)
    def __del__(self):
        print("Disconnected!")
        self.__client.close()

Python建立ssh連接并返回shell執(zhí)行命令結(jié)果

調(diào)用paramiko模塊

paramiko是一個(gè)用于做遠(yuǎn)程控制的模塊，使用該模塊可以對(duì)遠(yuǎn)程服務(wù)器進(jìn)行命令或文件操作。

安裝

使用pip可以直接安裝

pip3 install paramiko #python3

代碼

import os
import sys
import paramiko
# 創(chuàng)建SSH對(duì)象
ssh = paramiko.SSHClient()
# 把要連接的機(jī)器添加到known_hosts文件中
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 輸入?yún)?shù)并進(jìn)行判斷
if len(sys.argv) == 4:
    ip = sys.argv[1]
    uname = sys.argv[2]
    passwd = sys.argv[3] 
else:
	#若用戶沒(méi)有輸入命令行參數(shù)，則提示用戶
	print("Invalid amount of arguments.")
	print("example:python3 ssh.py <ip> <uname> <passwd>")
	sys.exit()
# 連接服務(wù)器
# 用戶名密碼
ssh.connect(hostname=ip, port=22, username=uname, password=passwd)
#ssh.connect(hostname='xxx.xxx.xx.xx', port=22, username='xxx', password='xxx')
cmd = 'cd /;ls -l;ifconfig'
# cmd = 'ls -l;ifconfig'       #多個(gè)命令用;隔開(kāi)
stdin, stdout, stderr = ssh.exec_command(cmd)
result = stdout.read()
if not result:
	result = stderr.read()
ssh.close()
print(result.decode())

關(guān)于linux中stdin, stdout, stderr三個(gè)參數(shù)的說(shuō)明

在Linux下，當(dāng)一個(gè)用戶進(jìn)程被創(chuàng)建的時(shí)候，系統(tǒng)會(huì)自動(dòng)為該進(jìn)程創(chuàng)建三個(gè)數(shù)據(jù)流，stdin, stdout 和 stderr

三個(gè)數(shù)據(jù)流默認(rèn)是表現(xiàn)在用戶終端上的

執(zhí)行一個(gè)shell命令行時(shí)通常會(huì)自動(dòng)打開(kāi)三個(gè)標(biāo)準(zhǔn)文件：

標(biāo)準(zhǔn)輸入文件（stdin），通常對(duì)應(yīng)終端的鍵盤；
標(biāo)準(zhǔn)輸出文件（stdout）和標(biāo)準(zhǔn)錯(cuò)誤輸出文件（stderr），這兩個(gè)文件都對(duì)應(yīng)終端的屏幕。

進(jìn)程將從標(biāo)準(zhǔn)輸入文件中得到輸入數(shù)據(jù)，將正常輸出數(shù)據(jù)輸出到標(biāo)準(zhǔn)輸出文件，而將錯(cuò)誤信息送到標(biāo)準(zhǔn)錯(cuò)誤文件中。

證書登錄

import os
import sys
import time
import paramiko
# 創(chuàng)建SSH對(duì)象
ssh = paramiko.SSHClient()
pkey = paramiko.RSAKey.from_private_key_file('/**/**') #私鑰證書路徑
# 把要連接的機(jī)器添加到known_hosts文件中
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
if len(sys.argv) == 3:
    ip = sys.argv[1]
    uname = sys.argv[2]
    #passwd = sys.argv[3]
else:
	#若用戶沒(méi)有輸入命令行參數(shù)，則提示用戶
	print("Invalid amount of arguments.")
	print("example:python3 ssh.py <ip> <uname> <passwd>")
	sys.exit()
# 連接服務(wù)器
# 私鑰證書登錄
ssh.connect(hostname=ip, port=22, username=uname, pkey=pkey)
cmd = 'cd /;ls -l;ifconfig'
# cmd = 'ls -l;ifconfig'       #多個(gè)命令用;隔開(kāi)
stdin, stdout, stderr = ssh.exec_command(cmd)
time.sleep(5)#增加更多時(shí)間來(lái)處理命令
result = stdout.read()
if not result:
	result = stderr.read()
ssh.close()
print(result.decode())