当前位置:网站首页>telnet+ftp to control and upgrade the device
telnet+ftp to control and upgrade the device
2022-08-09 13:43:00 【Hong Dayu】
import telnetlib
import time
from ftplib import FTP
import serial
import configparser
import signal
import os
import tarfile
class TelnetClient():
def __init__(self, host_ip, username, password):
self.tn = telnetlib.Telnet()
self.host_ip = host_ip
self.username = username
self.password = password
def login_host(self):
try:
self.tn.open(self.host_ip, port=23)
except:
print('%s网络连接失败' % self.host_ip)
return False
self.tn.read_until(b'login: ', timeout=10)
self.tn.write(self.username.encode('ascii') + b'\n')
self.tn.read_until(b'Password: ', timeout=10)
self.tn.write(self.password.encode('ascii') + b'\n')
command_result = self.tn.read_very_eager().decode('ascii')
if 'Login incorrect' not in command_result:
print('%s登录成功' % self.host_ip)
return True
else:
print('%s登录失败,用户名或密码错误' % self.host_ip)
return False
def execute_some_command(self, command,mode="nowait"):
self.tn.write(command.encode('ascii')+b'\n')
time.sleep(5)
print("\n***************************************************\n")
if mode == "nowait":
command_result = self.tn.read_very_eager().decode('ascii')
elif mode == "wait":
command_result = self.tn.read_until("DONE".encode('ascii')).decode('ascii')
print('命令执行结果:\n%s' % command_result)
print("\n***************************************************\n")
def __del__(self):
self.tn.write(b"exit\n")
def TelnetExecCmd(tn: TelnetClient, cmd: str,mode="nowait") -> bool:
if tn.login_host():
tn.execute_some_command(cmd,mode)
return True
else:
return False
class FtpClient():
def __init__(self, ip: str, username: str, password: str) -> None:
self.ip = ip
self.user = username
self.password = password
self.ftp = FTP()
def connect(self):
try:
self.ftp.set_debuglevel(2)
self.ftp.connect(self.ip, port=21)
self.ftp.login(self.user, self.password)
print(self.ftp.getwelcome())
except:
print("ftp connect {} error ...".format(self.ip))
def put(self, filename: str):
try:
fp = open(filename, "rb")
self.ftp.storbinary("STOR "+filename, fp)
except:
print("ftp upload {} error".format(filename))
def get(self, filename: str):
try:
fp = open(filename, "wb")
self.ftp.retrbinary("RETR "+filename, fp.write)
except:
print("ftp get {} error".format(filename))
def __del__(self):
self.ftp.set_debuglevel(0)
self.ftp.quit()
class SerialClient:
def __init__(self, port: str, bps: int) -> None:
try:
self.port = port
self.bps = bps
self.ser = serial.Serial(port, bps)
except:
print("Open Serial Error {}".format(port))
def write(self, cmd: str) -> bool:
if self.ser.isOpen():
print("Open Serial {}".format(self.port))
cmd = cmd+" \n"
self.ser.write(cmd.encode("utf8"))
return True
else:
return False
#如果终端有响应一直读取数据
def read(self):
if self.ser.isOpen():
while self.ser.readable():
line = self.ser.readline(4096)
print(line.decode('ascii'))
def __del__(self):
self.ser.close()
# AU DEVICE: 192.168.1.2
# RRU DEVICE: 192.168.98.98
def dos2unix(dos: str, unix: str):
content = ''
outsize = 0
with open(dos, 'rb') as infile:
content = infile.read()
with open(unix, 'wb') as output:
for line in content.splitlines():
outsize += len(line) + 1
output.write(line + b'\n')
print("Done. Stripped %s bytes." % (len(content)-outsize))
def Exit(signum, frame):
exit()
def CompressFile(TarFile:str,File:str,module:str):
tar = tarfile.open(TarFile,module)
for root,dir,files in os.walk(File):
for file in files:
fullpath = os.path.join(root,file)
tar.add(fullpath)
tar.close()
if __name__ == '__main__':
signal.signal(signal.SIGINT, Exit)
Parser = configparser.ConfigParser()
config = "config.ini"
if os.access(config,os.F_OK) == False:
print("Error:Can't find upgrade.ini, Please check your ini config")
exit()
Parser.read(config)
Parser.sections()
IP = Parser["BaseConfig"]["ServerIpAddr"]
Scripts = Parser["BaseConfig"]["UpgradeScript"]
UserName = Parser["UserConfig"]["UserName"]
PassWord = Parser["UserConfig"]["UserPassword"]
Filename = Parser["UserConfig"]["UpgradeFilename"]
CompressedMode = Parser["BaseConfig"]["CompressedMode"]
dos2unix(Scripts,Scripts)
if os.access(Filename+".tar",os.F_OK):
os.remove(Filename+".tar")
CompressFile(Filename+".tar",Filename,CompressedMode)
ftp = FtpClient(IP,UserName,PassWord)
ftp.connect()
ftp.put(Scripts)
ftp.put(Filename+".tar")
os.remove(Filename+".tar")
exec_upgrade = TelnetClient(IP,UserName,PassWord)
CMD = "chmod +rwx "+ "/mnt/flash/"+Scripts+" && /mnt/flash/"+Scripts
TelnetExecCmd(exec_upgrade,CMD,"wait")
TelnetExecCmd(exec_upgrade,"rm /mnt/flash/"+Scripts+" ; rm -rf /run/"+Filename+"*")
os.system("pause")
边栏推荐
- Flutter入门进阶之旅(五)Image Widget
- Extract EventBus encapsulation to base class using annotations
- 第六届“强网杯”全国网络安全挑战赛
- Rust from entry to proficient 04 - data types
- The new features of ABP 6.0.0 - rc. 1
- ERP不规范,同事两行泪 (转载非原创)
- JVM常用监控工具解释以及使用
- 联通网管协议框图
- Flutter入门进阶之旅(六)Layout Widget
- FFmpeg compiles and installs on win10 (configure libx264)
猜你喜欢

Intranet penetration tool ngrok usage tutorial

Go 事,如何成为一个Gopher ,并在7天找到 Go 语言相关工作,第1篇

链表噩梦之一?5000多字带你弄清它的来龙去脉

Flutter introduction advanced trip (5) Image Widget

#WeArePlay | 与更多开发者一起,探索新世界

novel research

保存Simulink仿真模型为图片或者PDF的方法

Flutter入门进阶之旅(三)Text Widgets

注释、关键字、标识符的区别你知道吗?

Flutter Getting Started and Advanced Tour (8) Button Widget
随机推荐
透明tune proxy
ERP不规范,同事两行泪 (转载非原创)
造自己的芯,让谷歌买单!谷歌再度开源 180nm 工艺的芯片
World's 4th mad scientist dies on his 103rd birthday
从NPU-SLAM-EDA技术分析
批量读取word docx文件指定表格内容,保存在excel文件中
面试题精选:神奇的斐波那契数列
MySQL principle and optimization of Group By optimization techniques
自定义VIEW实现应用内消息提醒上下轮播
NFS 特别注意权限的问题
Say goodbye to the AI era of hand looms
JVM之配置介绍(一)
Yocto 可以下载的第三方库
How to reduce the size of desktop icons after the computer is reinstalled
注:检测到当前使用的ADB不是HBuilder内置或自定义ADB:PID为:9544进程名称为:adb.exe 路径为:c:\users\administrator\appdata\local\and
十进制数字→十六进制字符
Report: The number of students who want to learn AI has increased by 200%, and there are not enough teachers
陈强教授《机器学习及R应用》课程 第十八章作业
Flutter入门进阶之旅(八)Button Widget
MySQL5.6到8.0的账号迁移