Max/openwrt-xiaomi_xmir-patcher
1
0
Fork 0
mirror of https://github.com/openwrt-xiaomi/xmir-patcher.git synced 2026-09-22 22:57:55 +03:00
Code Issues Packages Projects Releases Wiki Activity

gateway: Cleanup code in get_telnet, download, upload

Browse Source
This commit is contained in:
remittor
2025-04-17 07:11:37 +03:00
parent 2ae47a1236
commit fea8d7df80
1 changed files with 51 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway.py
+51 -50
View File
@@ -786,7 +786,7 @@ class Gateway():
return True return True
except Exception as e: except Exception as e:
if verbose: if verbose:
die("TELNET not responding (IP: {})".format(self.ip_addr)) die(f"TELNET not responding (IP: {self.ip_addr})")
return False return False
def get_telnet(self, verbose = 0, password = None): def get_telnet(self, verbose = 0, password = None):
@@ -794,37 +794,37 @@ class Gateway():
tn = telnetlib.Telnet(self.ip_addr, timeout=4) tn = telnetlib.Telnet(self.ip_addr, timeout=4)
except Exception as e: except Exception as e:
if verbose: if verbose:
die("TELNET not responding (IP: {})".format(self.ip_addr)) die("TELNET not responding (IP: {self.ip_addr})")
return None return None
try: try:
p_login = b'login: ' p_login = b'login: '
p_passw = b'Password: ' p_passw = b'Password: '
prompt = "{}@XiaoQiang:(.*?)#".format(self.login).encode('ascii') prompt = f"{self.login}@XiaoQiang:(.*?)#".encode('ascii')
idx, obj, output = tn.expect([p_login, prompt], timeout=2) idx, obj, output = tn.expect([p_login, prompt], timeout=2)
if idx < 0: if idx < 0:
raise Exception('') raise Exception('TELNET auth error (1)')
if idx > 0: if idx > 0:
tn.prompt = obj.group() tn.prompt = obj.group()
return tn return tn
tn.write("{}\n".format(self.login).encode('ascii')) tn.write(f"{self.login}\n".encode('ascii'))
idx, obj, output = tn.expect([p_passw, prompt], timeout=2) idx, obj, output = tn.expect([p_passw, prompt], timeout=2)
if idx < 0: if idx < 0:
raise Exception('') raise Exception('TELNET auth error (2)')
if idx > 0: if idx > 0:
tn.prompt = obj.group() tn.prompt = obj.group()
return tn return tn
if password is None: if password is None:
password = self.passw password = self.passw
tn.write("{}\n".format(password).encode('ascii')) tn.write(f"{password}\n".encode('ascii'))
idx, obj, output = tn.expect([prompt], timeout=2) idx, obj, output = tn.expect([prompt], timeout=2)
if idx < 0: if idx < 0:
raise Exception('') raise Exception('TELNET auth error (3)')
tn.prompt = obj.group() tn.prompt = obj.group()
return tn return tn
except Exception as e: except Exception as e:
#print(e) #print(e)
if verbose: if verbose:
die("Can't login to TELNET (IP: {})".format(self.ip_addr)) die(f"Can't login to TELNET (IP: {self.ip_addr})")
return None return None
def check_ftp(self, check_upload = False, timeout = None): def check_ftp(self, check_upload = False, timeout = None):
@@ -924,7 +924,7 @@ class Gateway():
ssh.set_timeout(100) ssh.set_timeout(100)
error = -4 error = -4
if die_on_error: if die_on_error:
die("SSH execute command timed out! CMD: \"{}\"".format(cmd)) die(f'SSH execute command timed out! CMD: "{cmd}"')
if timeout is not None: if timeout is not None:
ssh.set_timeout(saved_timeout) ssh.set_timeout(saved_timeout)
try: try:
@@ -946,61 +946,62 @@ class Gateway():
def download(self, fn_remote, fn_local, verbose = 1): def download(self, fn_remote, fn_local, verbose = 1):
if verbose and self.verbose: if verbose and self.verbose:
print('Download file: "{}" ....'.format(fn_remote)) print(f'Download file: "{fn_remote}" ....')
if self.use_ssh: if self.use_ssh:
ssh = self.get_ssh(self.verbose) ssh = self.get_ssh(self.verbose)
channel, fileinfo = ssh.scp_recv2(fn_remote) channel, fileinfo = ssh.scp_recv2(fn_remote)
total_size = fileinfo.st_size total_size = fileinfo.st_size
read_size = 0 read_size = 0
with open(fn_local, 'wb') as file: with open(fn_local, 'wb') as file:
while read_size < total_size: while read_size < total_size:
size, data = channel.read() size, data = channel.read()
if size > 0: if size > 0:
if read_size + len(data) > total_size: if read_size + len(data) > total_size:
file.write(data[:total_size - read_size]) file.write(data[:total_size - read_size])
else: else:
file.write(data) file.write(data)
read_size += size read_size += size
elif self.use_ftp: elif self.use_ftp:
ftp = self.get_ftp(self.verbose) ftp = self.get_ftp(self.verbose)
file = open(fn_local, 'wb') file = open(fn_local, 'wb')
ftp.retrbinary('RETR ' + fn_remote, file.write) ftp.retrbinary('RETR ' + fn_remote, file.write)
file.close() file.close()
else: else:
raise RuntimeError('FIXME') raise RuntimeError('FIXME')
return True return True
def upload(self, fn_local, fn_remote, md5chk = True, verbose = 1): def upload(self, fn_local, fn_remote, md5chk = True, verbose = 1):
if not os.path.exists(fn_local): if not os.path.exists(fn_local):
die(f'File "{fn_local}" not found.') die(f'File "{fn_local}" not found.')
if md5chk: if md5chk:
md5_local = self.get_md5_for_local_file(fn_local) md5_local = self.get_md5_for_local_file(fn_local)
file = open(fn_local, 'rb') file = open(fn_local, 'rb')
if verbose and self.verbose: if verbose and self.verbose:
print('Upload file: "{}" ....'.format(fn_local)) print(f'Upload file: "{fn_local}" ....')
if self.use_ssh: if self.use_ssh:
ssh = self.get_ssh(self.verbose) ssh = self.get_ssh(self.verbose)
finfo = os.stat(fn_local) finfo = os.stat(fn_local)
channel = ssh.scp_send64(fn_remote, finfo.st_mode & 0o777, finfo.st_size, finfo.st_mtime, finfo.st_atime) channel = ssh.scp_send64(fn_remote, finfo.st_mode & 0o777, finfo.st_size, finfo.st_mtime, finfo.st_atime)
size = 0 size = 0
for data in file: if True:
channel.write(data) for data in file:
size = size + len(data) channel.write(data)
#except ssh2.exceptions.SCPProtocolError as e: size = size + len(data)
#except ssh2.exceptions.SCPProtocolError as e:
elif self.use_ftp: elif self.use_ftp:
ftp = self.get_ftp(self.verbose) ftp = self.get_ftp(self.verbose)
ftp.storbinary('STOR ' + fn_remote, file) ftp.storbinary('STOR ' + fn_remote, file)
else: else:
raise RuntimeError('FIXME') raise RuntimeError('FIXME')
file.close() file.close()
if md5chk: if md5chk:
md5_remote = self.get_md5_for_remote_file(fn_remote) md5_remote = self.get_md5_for_remote_file(fn_remote)
if md5_remote != md5_local: if md5_remote != md5_local:
if md5chk == 2: if md5chk == 2:
die(f'File "{fn_local}" uploaded, but MD5 incorrect!') die(f'File "{fn_local}" uploaded, but MD5 incorrect!')
#if verbose: #if verbose:
print(f'ERROR: File "{fn_local}" uploaded, but MD5 incorrect!') print(f'ERROR: File "{fn_local}" uploaded, but MD5 incorrect!')
return False return False
return True return True
def get_md5_for_remote_file(self, fn_remote, timeout = 8): def get_md5_for_remote_file(self, fn_remote, timeout = 8):