So after some research on why sys.exit was used to call main I think we can remove that and just call main instead.
So have integrated to 2 lots of python code , I know the ping test part of it works ok but I can test it once the reboot code is added because I don't have all the modules installed, or the router to talk to.
So here's the code I would test it at the command line first using the command , were filename is what ever you saved the code as.
Code: Select all
#!/usr/bin/python3
#""" reboots the Huawei 618 router (and potentially others with similar firmware) """
import subprocess
import xml.etree.ElementTree as ET
import sys
import uuid
import hashlib
import hmac
from time import sleep
from binascii import hexlify
import requests
from config import ROUTER, USER, PASSWORD
#**************************************** code that does router reset **************************
def generate_nonce():
""" generate random clientside nonce """
return uuid.uuid4().hex + uuid.uuid4().hex
def setup_session(client, server):
""" gets the url from the server ignoring the respone, just to get session cookie set up """
url = "http://%s/" % server
response = client.get(url)
response.raise_for_status()
# will have to debug this one as without delay here it was throwing a buffering exception on one of the machines
sleep(1)
def get_server_token(client, server):
""" retrieves server token """
url = "http://%s/api/webserver/token" % server
token_response = client.get(url).text
root = ET.fromstring(token_response)
return root.findall('./token')[0].text
def get_client_proof(clientnonce, servernonce, password, salt, iterations):
""" calculates server client proof (part of the SCRAM algorithm) """
msg = "%s,%s,%s" % (clientnonce, servernonce, servernonce)
salted_pass = hashlib.pbkdf2_hmac(
'sha256', password, bytearray.fromhex(salt), iterations)
client_key = hmac.new(b'Client Key', msg=salted_pass,
digestmod=hashlib.sha256)
stored_key = hashlib.sha256()
stored_key.update(client_key.digest())
signature = hmac.new(msg.encode('utf_8'),
msg=stored_key.digest(), digestmod=hashlib.sha256)
client_key_digest = client_key.digest()
signature_digest = signature.digest()
client_proof = bytearray()
i = 0
while i < client_key.digest_size:
client_proof.append(client_key_digest[i] ^ signature_digest[i])
i = i + 1
return hexlify(client_proof)
def login(client, server, user, password):
""" logs in to the router using SCRAM method of authentication """
setup_session(client, server)
token = get_server_token(client, server)
url = "http://%s/api/user/challenge_login" % server
request = ET.Element('request')
username = ET.SubElement(request, 'username')
username.text = user
clientnonce = generate_nonce()
firstnonce = ET.SubElement(request, 'firstnonce')
firstnonce.text = clientnonce
mode = ET.SubElement(request, 'mode')
mode.text = '1'
headers = {'Content-type': 'text/html',
'__RequestVerificationToken': token[32:]}
response = client.post(url, data=ET.tostring(
request, encoding='utf8', method='xml'), headers=headers)
scram_data = ET.fromstring(response.text)
servernonce = scram_data.findall('./servernonce')[0].text
salt = scram_data.findall('./salt')[0].text
iterations = int(scram_data.findall('./iterations')[0].text)
verification_token = response.headers['__RequestVerificationToken']
login_request = ET.Element('request')
clientproof = ET.SubElement(login_request, 'clientproof')
clientproof.text = get_client_proof(
clientnonce, servernonce, password, salt, iterations).decode('UTF-8')
finalnonce = ET.SubElement(login_request, 'finalnonce')
finalnonce.text = servernonce
headers = {'Content-type': 'application/x-www-form-urlencoded; charset=UTF-8',
'__RequestVerificationToken': verification_token}
url = "http://%s/api/user/authentication_login" % server
result = client.post(url, data=ET.tostring(
login_request, encoding='utf8', method='xml'), headers=headers)
verification_token = result.headers['__RequestVerificationTokenone']
return verification_token
def reboot(client, server, user, password):
""" reboots the router :) """
verification_token = login(client, server, user, password)
url = "http://%s/api/device/control" % server
headers = {'Content-type': 'application/x-www-form-urlencoded; charset=UTF-8',
'__RequestVerificationToken': verification_token}
client.post(
url, data='<?xml version:"1.0" encoding="UTF-8"?><request><Control>1</Control></request>', headers=headers)
def main():
""" main method """
client = requests.Session()
reboot(client, ROUTER, USER, PASSWORD)
#**************************************** code that does the ping testing ***************
SITES = ["google.com", "raspberrypi.org" , "no-ip.com", "grc.com", "youtube.com"]
DELAY_BETWEEN_PINGS = 1 # delay in seconds
DELAY_BETWEEN_TESTS = 120 # delay in seconds I would suggest 120-180 seconds or more
DELAY_AFTER_ROUTER_REBOOT = 180 # delay in seconds probably needs to be 180-240 seconds
# issue Linux ping command to determine internet connection status
def ping(site):
cmd = "/bin/ping -c 1 " + site
try:
output = subprocess.check_output(cmd, stderr=subprocess.STDOUT, shell=True)
except subprocess.CalledProcessError:
print (site,"bad") # can be removed
return 0
else:
print (site,"good") # can be removed
return 1
def ping_sites(site_list, wait_time, times):
successful_pings = 0
attempted_pings = times * len(site_list)
for t in range(0, times):
for s in site_list:
successful_pings += ping(s)
sleep(wait_time)
return successful_pings / float(attempted_pings) # return percentage successful
while True:
success = ping_sites(SITES, DELAY_BETWEEN_PINGS, 2)
print (success)
if success <= 0.8:
print ("Internet is DOWN")
main()
sleep(DELAY_AFTER_ROUTER_REBOOT)
else:
print ("Internet is OK")
sleep(DELAY_BETWEEN_TESTS)
now if it works at the command line it might still not work from cron , because cron is a different environment and it may not be able in import your router settings from the config file, in which case it may be simpler to add the settings to the program, and dispense with the config file.
So if it fails to run with some error please post the error back here in its entirety. I would say we have a 50/50 chance of it working first try
