使用Python自動(dòng)執(zhí)行SSH和SCP任務(wù)

我最近幫助了一個(gè)朋友在100多個(gè)設(shè)備上刷新軟件。對(duì)于每個(gè)設(shè)備,他都知道MAC地址,并具有通過ssh登錄的相應(yīng)密碼。他將此信息存儲(chǔ)在Excel工作表中。他手動(dòng)閃爍的過程是

  1. 根據(jù)設(shè)備的MAC地址找出設(shè)備的IP地址。
  2. 使用IP地址和密碼SSH進(jìn)入設(shè)備。
  3. 通過SCP上載軟件安裝程序。
  4. 執(zhí)行一些Shell命令以安裝軟件。

在一臺(tái)設(shè)備上手動(dòng)執(zhí)行此操作就可以了,對(duì)于兩臺(tái)設(shè)備,您開始考慮wtf,三臺(tái)設(shè)備之后,您就會(huì)想到WTF!在軟件中,減少WTF始終是一個(gè)目標(biāo),因此我想也許我可以通過使用漂亮的Python腳本自動(dòng)化他的過程來為他提供幫助。在本文中,我想向您展示最終的結(jié)果。

查找給定了MAC地址的IP地址

我們需要解決的第一件事是找出我們本地網(wǎng)絡(luò)內(nèi)部設(shè)備的IP地址。為此,您的PC和設(shè)備都必須連接到同一網(wǎng)絡(luò)。給定一個(gè)IP地址,您可以使用python模塊scapy(請(qǐng)注意,我在本文的先前版本中使用了其他軟件包。但是,這僅在我已經(jīng)連接到或?qū)α硪粋€(gè)設(shè)備執(zhí)行ping操作時(shí)才有效)哪個(gè)MAC地址使用IP。在代碼中,這看起來像

from scapy.layers.l2 import arping
def get_mac_and_ip(suffix: int) -> Tuple[str, str]:
    ip = f"192.168.1.{suffix}"
    ans, _ = arping(ip, timeout=.2, verbose=False)
    for s, r in ans.res:
        st="%19s,Ether.src% %ARP.psrc%"
        return tuple(r.sprintf(st).strip().split())

    return ("DONT", "CARE")
mac_to_ip = dict(get_mac_and_ip(suffix) for suffix in range(1, 256))

因此,我們只測(cè)試每個(gè)IP地址并存儲(chǔ)回送MAC地址的IP地址。使用scapy時(shí),您需要以root用戶身份執(zhí)行腳本。太酷了,現(xiàn)在我們知道所有MAC地址到IP地址的連接。下一個(gè)問題剩下三個(gè)。

建立SSH連接

下一個(gè)問題是,我們?nèi)绾问褂肞ython建立SSH連接。有一個(gè)名為paramiko的簡(jiǎn)潔庫(kù),使您可以非常簡(jiǎn)單地完成此操作。從根本上講,您需要要連接的IP地址,用戶名和密碼。代碼看起來像

import paramiko
# To allow connection without having to accept the host connection
# We need to use this policy. I've just pulled it here due to 
# mediums way of displaying code ...
policy = paramiko.client.AutoAddPolicy
# Iterate over all our found mac, ip connections and SSH into
# the clients
for mac, ip in mac_to_ip.items():
    # Assume that to be given
    pwd = mac_to_pwd[mac]
with paramiko.SSHClient() as client:
        client.set_missing_host_key_policy(policy)
        try:
            client.connect(ip, username='root', password=pwd)
        except paramiko.ssh_exception.NoValidConnectionsError:
            print("Connection failed")

現(xiàn)在,我們可以使用連接的客戶端執(zhí)行Shell命令,您將很快看到它。

通過SCP上傳數(shù)據(jù)

在執(zhí)行命令之前,我們需要將安裝程序上載到我們的設(shè)備。為此,我們利用庫(kù)SCP,該庫(kù)與paramiko完美配合。在下面的示例中,假設(shè)我們已經(jīng)使用paramiko SSHClient建立了連接。

from scp import SCPClient 
import sys
def progress4(f, size, sent, p):
    prog = sent / size * 100.
    sys.stdout.write(f"({p[0]}:{p[1]}) {f}\'s progress: {prog}\r")
local_path = "/home/me/installer.tar.gz"
remote_path = "/home/root/inst/"
with SCPClient(client.get_transport(), progress4=progress4) as scp:
    scp.put(local_path, remote_path=remote_path)

progress4功能打印出的上傳進(jìn)度。對(duì)我來說,這有助于估算我仍然需要等待多長(zhǎng)時(shí)間而又不會(huì)變得急躁,這可能會(huì)導(dǎo)致另一個(gè)WTF :)?,F(xiàn)在我們快到了,只剩一步了。

通過SSH執(zhí)行Shell命令

現(xiàn)在,由于我們具有正在運(yùn)行的ssh連接并上傳了安裝程序,因此我們只需解壓縮它并運(yùn)行它。這可以通過我們可以使用SSH客戶端發(fā)送的shell命令輕松完成。在代碼中,這看起來像

def exec_blocking(cmd, client):
    print(f"Executing {cmd}")
    _, stdout_, stderr_ = client.exec_command(cmd)
    status = stdout_.channel.recv_exit_status()
    print(f"STATUS {status}")
    for line in stdout_.readlines():
        print(line)
    if status != 0:
        errors = "\n".join(list(stderr_.readlines))
        raise Exception(f"{cmd} failed with {errors}")
# Unpack the installer 
exec_blocking("tar xzvf /home/root/inst/installer.xzvf", client)
# Run the installer
exec_blocking("./installer/install.sh", client)

該命令將通過exec_command發(fā)送到設(shè)備。這是一個(gè)立即返回的非阻塞函數(shù)。要等到命令完成執(zhí)行,您需要調(diào)用recv_exit_status()。這樣,您可以確保命令執(zhí)行成功與否。

結(jié)論

將所有代碼段組合到一個(gè)腳本中,使我們可以使用單個(gè)命令在列表中所有設(shè)備上刷新/執(zhí)行命令。這為我們節(jié)省了很多時(shí)間和WTF :)

希望您學(xué)到了一些新知識(shí)。感謝您的關(guān)注,并隨時(shí)與我聯(lián)系以提出問題,意見或建議。

翻譯自:https://medium.com/@simon.hawe/save-time-by-automating-ssh-and-scp-tasks-with-python-e149de606c7b

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成,瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明:文章內(nèi)容(如有圖片或視頻亦包括在內(nèi))由作者上傳并發(fā)布,文章內(nèi)容僅代表作者本人觀點(diǎn),簡(jiǎn)書系信息發(fā)布平臺(tái),僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容