[toc]
In daily work, it is necessary to perform some operations or commands on the server, even in the cloud era, but once there is a problem, you still need to go to the machine to check, so I wrote a small jump machine tool
Before I wrote this little tool, I thought it had to be really simple, simple enough, so in less than 200 lines of code, I wrote one, which is pretty simple.
1. Introduction to usage modules
- pexpect
This one is over!
A brief introduction to this module: Pexpect is a Python implementation of Expect, used for human-computer interaction, such as when a program requires a username and password, or yes/no, to capture such keywords and enter the necessary information to continue operating the program.
Pexpect has a wide range of uses, enabling automatic interactions with PROGRAMS like SSH, FTP, and Telnet
1.1 Usage Mode
The use of Pexpect is basically divided into three steps:
- Start with spawn to execute a command or program
- Then Expect catches the keywords
- After the specified keyword is captured, the send instruction is executed to send the necessary content to continue the program
1.1.1 spawnclass
Spawn is the main pexpect class that executes a program and returns a handle to that program that can be used to perform all subsequent operations. Here is the definition of its constructor:
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)
Copy the codecommandIt’s an arbitrary command
child = pexpect.spawn('/usr/bin/ftp')
child = pexpect.spawn('/usr/bin/ssh [email protected]')
child = pexpect.spawn('ls -latr /tmp')
Copy the codeBut when contains some special characters (> |, or *), you must start a shell to perform, such as:
child = pexpect.spawn('/bin/bash -c "ls -l | grep LOG > logs.txt"')
child.expect(pexpect.EOF)
Copy the codeYou can also write it this way, specifying a variable through which to receive the command to be executed
shell_cmd = 'ls -l | grep LOG > logs.txt'
child = pexpect.spawn('/bin/bash'['-c', shell_cmd])
child.expect(pexpect.EOF)
Copy the codeargs=[]Pass in the required parameters when executing the program
child = pexpect.spawn('/usr/bin/ftp', [])
child = pexpect.spawn('/usr/bin/ssh'['[email protected]'])
child = pexpect.spawn('ls'['-latr'.'/tmp'])
Copy the code-
Timeout =30 Sets the timeout period
-
Maxread =2000 The maximum number of bytes pexpect reads from the terminal console at one time
-
Searchwindowsize matches the position of the buffer string, starting from the default position
However, it is sometimes necessary to print the result of execution, that is, to print the output to standard output, as follows:
import pexpect
import sys
child = pexpect.spawn("df -h", encoding='utf-8')
child.logfile = sys.stdout
child.expect(pexpect.EOF)
Copy the codeTypeError: write() argument must be STR, not bytes TypeError: write() argument must be STR, not bytes TypeError: write() argument must be STR, not bytes
child = pexpect.spawn("df -h", logfile=sys.stdout, encoding='utf-8')
child.expect(pexpect.EOF)
Copy the code1.1.2 expectmethods
Expect eventually returns 0 to indicate that the desired keyword was matched, or, if a keyword list is defined, a number indicating the number of keywords in the list that were matched, starting at 0, the index number of the keyword
expect(pattern, timeout=-1, searchwindowsize=-1, async_=False, **kw)
Copy the codePattern can be StringType, EOF, regular expression, or a list of these types. If Pattern is a list with multiple matching values, only the first matching index is returned, for example:
child = pexpect.spawn("echo 'hello world'", logfile=sys.stdout, encoding='utf-8')
index = child.expect(['hahaha'.'hello'.'hello world'])
print(index) The result is 1, the index number of 'hello'
Copy the codeNote: it is important to know that the content of pattern is used to match the spawn keyword
1.1.3 sendmethods
Send means to send keywords to a program. For example, write a simple shell script that accepts a username argument and writes the value of the variable to a file
# test.sh
#! /bin/bash
read -p "Input your name:" username
echo "$username" > name.txt
Copy the codeThen, expect is used to capture the keywords, and send is used to send the keywords
child = pexpect.spawn("sh test.sh", logfile=sys.stdout, encoding='utf-8')
index = child.expect("Input your name")
if index == 0:
child.send("dogfei") After executing, press Enter again
Copy the codeIf you don’t want to hit return, you can use sendline. The result is a name.txt file in the current directory that contains the value of the variable you just passed in
Ok, so much said, basically have an understanding of the pexpect module, then go straight to the subject!
The remote SSH connection to the target host is implemented
SSH server address, user name, password, port, and other information is required for remote login. There are several situations as follows:
- During the first SSH connection, the following message is displayed:
Are you sure you want to continue connecting (yes/no), require inputyes/no - If the SSH port is incorrect during SSH connection, the following information is displayed:
Connection refused - If there is a network problem, the connection usually times out
- If you have connected to SSH before, you will be prompted to connect again:
password:, requires a password - If the password is correct, the following message is displayed:
Last login - If the password is incorrect, the following message is displayed:
Permission denied, please try again
Knowing that, then it’s easier when we’re writing.
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]
Copy the codeThis function returns a list of the capture key’s index number, a handle to the operator, and then the following recursion is used for various cases of remote SSH to avoid using nested while loops
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 complete command
ssh_cmd = "ssh {u}@{h} -p {p}".format(u=user, h=host, p=port)
Initialize a handle and get the index number
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)
The first SSH will ask you to type yes/no or something like that, so when this is matched, do a recursion
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:
The password is incorrect and needs to be re-entered
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
Copy the codeAt this point, we can use this program to remote operation, take the machine to do a test:
if __name__ == "__main__":
sshclient('127.0.0.1'.'dogfei'.22.'123456')
Copy the codeHere are the tips:
$ python3 test_jp.py Begin Load Password... 127.0.0.1 login Success (-_-) Sun Jun 20 19:43:322021 from 127.0.0.1Copy the codeBut here only can achieve remote connection to the remote host, but the jump machine, there are many machines, and many types, each machine has its own number, so we want to achieve such a function, look!
The realization of simple jump board machine
Since there are many hosts, these hosts are divided into many types, that is, tags and so on, and there may be different passwords for each host, or the same password for the same type of machine, or can not use root login, etc., so we must make a simple and flexible, Machines do not match passwords do not match the user name that can be defined a complete login command to solve, and for the host partition type, set the password (the default password), the user name that information, by a very flexible database table structure to implement, here I addressed by a local configuration file, as follows:
global:
user: root
port: 22
passwd: 123456
jumpserver:
- name: k8s
hostList:
- 192.1681.1.
- 192.1681.2.
- 192.1681.3.
Copy the codeThis configuration file has a global configuration. The global configuration under Global is global. If not specified in the JumpServer section, the global configuration is taken.
If the username and password are different, you can write:
global:
user: root
port: 22
passwd: 123456
jumpserver:
- name: k8s
hostList:
- 192.1681.1.
- 192.1681.2.
- 192.1681.3.
- name: mysql
hostList:
- host: 192.1681.4.
user: dogfei
- host: 192.1681.. 5
user: db
passwd: 111111
Copy the codeAfter this design, our code looks like this:
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'))
Copy the codeThe result is a list with a dictionary element, as follows:
[{'tag': 'k8s'.'host': '192.168.1.1'.'user': 'root'.'port': 22.'passwd': 123456},
{'tag': 'k8s'.'host': '192.168.1.2 instead.'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}]Copy the codeOnce you’ve got this bunch of data words, you can beautify them as follows:
def list_info(originList) :
try:
print(033 "* * * * * * \ [1; 30; 43 mip information as follows, please select the corresponding number to land \ [0 033 m * * * * * * \ n")
print("\033[0;32m{:<5}\033[0m{:<19}{}".format("Number"."IP address"."Label"))
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
Copy the codeThe result of this code is shown below:
Then the following is a bunch of loops, mainly to print the host information, exit, enter the host and other functions, as follows:
def login(yamlfile): try: Print ("\033[1;30;47m{:^50}\033[0m\n". Format (" simple jumper ")) outer_flag = False while not outer_flag: print("\033[1;30;47m{:^50}\033[0m\n". Print (" \ [5, 35, 033, 46 m <} {: 033 [0 m \ n \ ". The format (" select ")) print (" \ [0; 32 m 033 input 'p/p' print all host information \ [0 m ") print (" 033\033 [0; 31 m input 'q/quit' exit 033 [0 m \ \ n ") input_x = input (" > > > > > : "). The strip (). The lower () if input_x = = 'p' : OS. System ("clear") ip_info = list_info(yamlfile) print("\n") print("\033[0;32m \033[0m") print("\033[0; \033[0m") print("\033[0; \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) continue else: if act == 'q' or act == 'quit': print("\033[0;31m) continue else: if act == 'q' or act == 'quit': Print ("\033[0;31m leave!!\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 leave!!\033[0m") outer_flag = True else: Print ("\033[0;31m \!!\033[0m") continue except Exception as e: raise eCopy the codeThen take a look at the effect below:
Here is a dynamic demo:
Total plus blank lines, a total of 185 lines, is really very practical ah!
The full code can be viewed on my personal blog or on my public account: www.dogfei.cn