不到200行程式碼，使用python寫一個超簡單的跳板機工具
在日常工作中，少不了需要上伺服器執行一些操作或者命令，即使在現在雲時代，但是一旦出現問題，還是需要上機排查，所以寫了一個小小的跳板機工具
在寫這個小工具之前，我就在想一定要非常簡單，足夠簡單，所以這裡用了不到 200 行程式碼，寫了一個，可以說是非常簡單。
1. 使用模組介紹
- pexpect
就這一個就完了！
簡單介紹一下這個模組：
pexpect 是 expect 的一個 python 實現，用來進行人機互動，比如執行某一段程式時要求輸入使用者名稱和密碼，或者是 yes/no 這種情況時，使用 pexpect 便可以通過捕獲這種關鍵字進行輸入一些必要資訊來繼續操作程式。
pexpect 的使用範圍很廣，可以用來實現與 ssh、ftp 、telnet 等程式的自動互動
1.1 使用方式
pexpect 的使用基本上分為三個步驟:
- 首先用 spawn 來執行一條命令或者一段程式
- 然後 expect 捕獲關鍵字
- 捕獲到指定關鍵字後，執行 send 指令傳送必要的內容來繼續操作程式
1.1.1 spawn類
spawn是 pexpect 的一個主要類，用來執行一個程式，並返回一個操作這個程式的控制代碼，後面的一系列操作都可以通過這個控制代碼來完成，下面是它的建構函式的定義：
class spawn(command, args=[], timeout=30, maxread=2000, searchwindowsize=None, logfile=None, cwd=None,env=None, ignore_sighup=False, echo=True, preexec_fn=None, encoding=None, codec_errors='strict', dimensions=None, use_poll=False)
command是任意的命令
child = pexpect.spawn('/usr/bin/ftp')
child = pexpect.spawn('/usr/bin/ssh [email protected]')
child = pexpect.spawn('ls -latr /tmp')
但是當包含一些特殊字元時(>, |, or *)，就必須啟動一個 shell 來執行了，例如：
child = pexpect.spawn('/bin/bash -c "ls -l | grep LOG > logs.txt"')
child.expect(pexpect.EOF)
也可以這樣寫，定一個變數，將要執行的命令通過這個變數來接收
shell_cmd = 'ls -l | grep LOG > logs.txt'
child = pexpect.spawn('/bin/bash', ['-c', shell_cmd])
child.expect(pexpect.EOF)
args=[]執行程式時傳入所需的引數
child = pexpect.spawn('/usr/bin/ftp', [])
child = pexpect.spawn('/usr/bin/ssh', ['[email protected]'])
child = pexpect.spawn('ls', ['-latr', '/tmp'])
-
timeout=30設定超時時間 -
maxread=2000pexpect 從終端控制檯一次讀取的最大位元組數 -
searchwindowsize匹配緩衝區字串的位置，預設是從開始位置匹配
不過有時候需要列印下執行的結果，也就是將輸出列印到標準輸出裡，方式如下：
import pexpect
import sys
child = pexpect.spawn("df -h", encoding='utf-8')
child.logfile = sys.stdout
child.expect(pexpect.EOF)
必須要加上encoding='utf-8'，否則會報TypeError: write() argument must be str, not bytes
當然，也可以使用logfile這個引數，例如：
child = pexpect.spawn("df -h", logfile=sys.stdout, encoding='utf-8')
child.expect(pexpect.EOF)
1.1.2 expect方法
expect最後會返回 0 表示匹配到了所需的關鍵字，如果定義了一個關鍵字列表的話，就會返回一個數字表示匹配到了列表中第幾個關鍵字，從 0 開始計算，也就是該關鍵字所在的索引號
expect(pattern, timeout=-1, searchwindowsize=-1, async_=False, **kw)
pattern 可以是 StringType, EOF，正則表示式，或者這些型別的列表
如果當pattern是一個列表時，且列表裡有多個可被匹配到的值時，最後只能返回第一個被匹配到的索引，例如：
child = pexpect.spawn("echo 'hello world'", logfile=sys.stdout, encoding='utf-8')
index = child.expect(['hahaha', 'hello', 'hello world'])
print(index) # 結果為1，'hello'的索引編號
注意： 這裡一定要知道, 是拿
pattern裡的內容來匹配spawn中的關鍵字
1.1.3 send方法
send就是傳送關鍵字給程式，這裡舉一個示例，寫一個簡單的 shell 指令碼,這個指令碼會接收一個 username 的引數，並把該變數的值寫道一個檔案裡
# test.sh
#!/bin/bash
read -p "Input your name:" username
echo "$username" > name.txt
然後通過 expect 進行捕獲關鍵字，最後使用 send 傳送關鍵字
child = pexpect.spawn("sh test.sh", logfile=sys.stdout, encoding='utf-8')
index = child.expect("Input your name")
if index == 0:
child.send("dogfei") # 執行完還要再敲一次回車
上面執行完之後，還要再敲下回車，如果想不用敲回車，可以使用sendline
結果當前目錄下會有一個 name.txt 檔案，並且裡面的內容就是剛傳進去的變數的值
ok, 說了這麼多，基本上對 pexpect 這個模組有了一個瞭解，那麼接下來就直奔主題！
實現遠端 ssh 連線到目標主機
既然是跳板機，那肯定是通過執行ssh來遠端登入，那麼就要有伺服器地址、使用者名稱、密碼、埠等資訊，然後遠端登入的時候，一般會有幾種情況，如下：
- 第一次 ssh 連線時，一般會提示：
Are you sure you want to continue connecting (yes/no)，要求輸入yes/no - 如果 ssh 連線時，ssh 的埠不正確，一般會提示：
Connection refused - 如果網路問題，一般會連線超時
- 如果之前 ssh 連線過，那再次連線時，就會提示：
password:，要求輸入密碼 - 輸入密碼且密碼正確，一般會提示：
Last login - 密碼錯誤，一般會提示：
Permission denied, please try again
知道了，這些，那麼我們在寫的時候，就比較容易了。
import pexpect
import os
def run_cmd(cmd, patterns):
child = pexpect.spawn(cmd, encoding='utf-8')
child.setwinsize(lines, columns)
index = child.expect(patterns, timeout=10)
return [index, child]
該函式返回一個列表，分別是捕獲關鍵字的索引編號，操作程式的一個控制代碼，然後下面是針對遠端 ssh 時的各種情況，用到了一個遞迴，這樣避免使用巢狀while迴圈
def sshclient(host, user, port, passwd):
ssh_newkey = "continue"
ssh_passwd = "assword:"
ssh_confirm = "yes"
ssh_refuse = "Connection refused"
ssh_login = "Last login:"
ssh_repeat_passwd = "Permission denied, please try again"
ssh_noroutetohost = "No route to host"
ssh_conntimeout = "Connection timed out"
# 遠端ssh時的完整命令
ssh_cmd = "ssh {u}@{h} -p {p}".format(u=user, h=host, p=port)
# 初始化一個控制代碼，並獲取索引號
index, child = run_cmd(ssh_cmd, [
ssh_newkey,
ssh_passwd,
ssh_refuse,
ssh_login,
ssh_noroutetohost,
ssh_conntimeout,
pexpect.EOF,
pexpect.TIMEOUT])
try:
if index == 0:
child.sendline(ssh_confirm)
# 一般第一次ssh時，會讓你輸入yes/no之類的，所以匹配到這個的時候，就做一次遞迴
return sshclient(host, user, port, passwd)
elif index == 1:
print("Begin Load Password...")
child.sendline(passwd)
result = child.expect([
ssh_repeat_passwd,
ssh_login,
])
if result == 1:
print("{} login success (-_-)".format(host))
child.interact()
return
elif result == 0:
# 說明密碼錯誤，需要重新輸入密碼，並進行遞迴
passwd = input('Passwd: ').strip()
return sshclient(host, user, port, passwd)
elif index == 2:
print("Connect refused, Pls check ssh port.")
return
elif index == 3:
print("Login success")
child.interact()
return
elif index == 4:
print("The host %s connected faild: No route to host" % host)
return
elif index == 5:
print("The host %s connected faild: Connection timeout" % host)
return
elif index == 6:
print("Abnormal exit")
return
elif index == 7:
print("Timeout for connect host %s, pls check network" % host)
return
return
except Exception as e:
raise e
到這裡，我們就能使用這個程式來進行遠端操作了，拿本機做下測試：
if __name__ == "__main__":
sshclient('127.0.0.1','dogfei',22, '123456')
提示如下：
$ python3 test_jp.py
Begin Load Password...
127.0.0.1 login success (-_-)
Sun Jun 20 19:43:32 2021 from 127.0.0.1
但這裡僅實現了能遠端連線到遠端主機，可是跳板機的話，是有很多臺機器，並且有很多型別，每臺機器都有自己的編號，所以我們要實現這樣的功能，淦！
簡易跳板機的實現
既然有很多主機，這些主機又劃分了很多型別，也就是標籤之類的，還有可能每臺主機的密碼都不一樣，又或者相同型別的機器密碼一樣，又或者不能使用 root 登入等等等等，所以我們一定要做一個既簡單又靈活的，機器密碼不一致和使用者名稱不一致這個可以通過定義一個完整的登入命令來解決，而對主機劃分型別、設定密碼（預設密碼），使用者名稱這些資訊，就要通過一個非常靈活的資料庫表結構來實現了，這裡我通過一個本地的配置檔案來解決，方式如下：
global:
user: root
port: 22
passwd: 123456
jumpserver:
- name: k8s
hostList:
- 192.168.1.1
- 192.168.1.2
- 192.168.1.3
該配置檔案有一個全域性的配置，global 下的為全域性配置，如果jumpserver部分中沒有指定，則會拿全域性配置，如果所有的機器使用者名稱、密碼、埠都一樣，那麼就可以這樣寫.
如果使用者名稱，密碼不一樣，那就可以這樣寫：
global:
user: root
port: 22
passwd: 123456
jumpserver:
- name: k8s
hostList:
- 192.168.1.1
- 192.168.1.2
- 192.168.1.3
- name: mysql
hostList:
- host: 192.168.1.4
user: dogfei
- host: 192.168.1.5
user: db
passwd: 111111
經過這樣設計以後，那麼我們的程式碼就變成了這個樣子：
import yaml
def parseYaml(yamlfile, parse_list=None):
if parse_list is None:
parse_list = []
with open(yamlfile, 'r', encoding='utf-8') as fr:
yaml_to_dict = yaml.safe_load(fr)
global_user = yaml_to_dict['global']['user']
global_passwd = yaml_to_dict['global']['passwd']
global_port = int(yaml_to_dict['global']['port'])
for detail in yaml_to_dict['jumpserver']:
tag = detail['name']
get_hostList = detail['hostList']
if isinstance(get_hostList[0], dict):
for ssh in get_hostList:
sshDetail = {
'tag': tag,
'host': ssh['host'],
'user': ssh['user'] if 'user' in ssh else global_user,
'port': int(ssh['port']) if 'port' in ssh else global_port,
'passwd': ssh['passwd'] if 'passwd' in ssh else global_passwd
}
parse_list.append(sshDetail)
elif isinstance(get_hostList[0], str):
for h in get_hostList:
sshDetail = {
'tag': tag,
'host': h,
'user': global_user,
'port': global_port,
'passwd': global_passwd
}
parse_list.append(sshDetail)
return parse_list
if __name__ == '__main__':
print(parseYaml('ip.yaml'))
最終返回的是一個列表，列表裡的元素是一個字典，如下所示：
[
{'tag': 'k8s', 'host': '192.168.1.1', 'user': 'root', 'port': 22, 'passwd': 123456},
{'tag': 'k8s', 'host': '192.168.1.2', 'user': 'root', 'port': 22, 'passwd': 123456},
{'tag': 'k8s', 'host': '192.168.1.3', 'user': 'root', 'port': 22, 'passwd': 123456},
{'tag': 'mysql', 'host': '192.168.1.4', 'user': 'dogfei', 'port': 22, 'passwd': 123456},
{'tag': 'mysql', 'host': '192.168.1.5', 'user': 'db', 'port': 22, 'passwd': 111111}
]
拿到這一堆資料字後，就可以進行美化了，如下所示：
def list_info(originList):
try:
print("******\033[1;30;43mIP資訊如下，請選擇對應的編號進行登陸\033[0m******\n")
print("\033[0;32m{:<5}\033[0m{:<19}{}".format("編號", "IP地址", "標籤"))
sshList = []
sshDict = {}
for info in originList:
id = originList.index(info) + 1
host = info['host']
tag = info['tag']
user = info['user']
port = int(info['port'])
passwd = info['passwd']
sshDict[id] = [
host, user, passwd, port
]
print("{:<5}{:<22}{}".format(id, host, tag))
return sshDict
except Exception as e:
raise e
這段程式碼的執行結果如圖所示：
然後下面就是一堆迴圈，主要是實現列印主機資訊、退出、進入主機等功能，如下所示：
def login(yamlfile):
try:
print("\033[1;30;47m{:^50}\033[0m\n".format("簡易跳板機"))
outer_flag = False
while not outer_flag:
print("\033[5;35;46m{:<}\033[0m\n".format("請選擇"))
print("\033[0;32m輸入 'p/P' 列印所有主機資訊\033[0m")
print("\033[0;31m輸入 'q/quit' 退出\033[0m\n")
input_x = input(">>>>>: ").strip().lower()
if input_x == 'p':
os.system("clear")
ip_info = list_info(yamlfile)
print("\n")
print("\033[0;32m輸入 '編號' 進入對應的伺服器\033[0m")
print("\033[0;32m輸入 'q' 退出\033[0m")
print("\033[0;32m輸入 'b' 返回\033[0m")
inner_flag = False
while not inner_flag:
act = input("\033[0;32m>>>>>: \033[0m").strip().lower()
if act.isdigit():
ip_id = int(act)
if ip_id in ip_info.keys():
host = ip_info[ip_id][0]
user = ip_info[ip_id][1]
passwd = str(ip_info[ip_id][2])
port = int(ip_info[ip_id][3])
sshclient(
host=host,
user=user,
port=port,
passwd=passwd
)
inner_flag = True
else:
print("\033[0;31m編號不存在，請重新輸入，退出請輸入任意字元\033[0m")
continue
else:
if act == 'q' or act == 'quit':
print("\033[0;31m告辭 !!!\033[0m")
inner_flag = True
outer_flag = True
elif act == 'b' or act == 'back':
inner_flag = True
elif input_x == 'q' or input_x == 'quit':
print("\033[0;31m告辭 !!!\033[0m")
outer_flag = True
else:
print("\033[0;31m請輸入指定的內容 !!!\033[0m")
continue
except Exception as e:
raise e
然後下面看下效果圖：
下面是動態演示：
總共加上空行，一共 185 行，真是太實用了啊！
完整程式碼可去我的個人部落格或者關注公眾號進行檢視：https://www.dogfei.cn
歡迎各位朋友關注我的公眾號，來一起學習進步哦