【Python学习实验室】Python实现路由器的重启和查看实时流量

0.问题引入

最近因为要使用一个可以插卡上网的4G路由器,但是不知道怎么回事,这个路由器总是在一个随机的时刻无法访问互联网(SIM卡是没有问题的),就是那种在网上刷着视频听着音乐看着直播突然就没有互联网的感觉,就非常非常的让人不舒服。

我也找了大量的资料和方法,比如修改配置参数文件,关闭路由器自带的WiFi功能都基本上是无法解决。自始至终,只有一个方法可以重新让路由器访问互联网:重启他

重启路由器的方法也有两种,一种是通过硬件手段,断电再插电,实现硬件上的重新启动,这个方法就是有点麻烦,况且如果是自动实现的话也是需要成本的,我也考虑过通过WiFi IOT设备进行路由器DC12V的自动通断,但是我不会啊。

于是只能用另外一种方法,就是从路由器的后台管理的后台进行重启。

Python实现路由器的重启和查看实时流量 路由器登陆界面
Python实现路由器的重启和查看实时流量路由器的登陆后台

就是进入路由器的后台界面,找到重启的按钮,点击他,路由器就会自动重启。

DlztFP.png
路由器的重启设置界面

每次都进入后台去输入东西肯定是不现实的东西。于是乎,现在的问题就是如何去控制路由器的重启功能。

这个文章就带你了解如何使用Python来进行路由器的重启和查看实时网速,在查看后面的内容同时也别忘了在各大平台订阅关注我以获取第一时间的更新!

1.尝试登录到路由器

凡是要登陆到路由器,肯定就要输入密码,因为进入路由器后台管理界面的第一项就是。

使用浏览器自带的工具可能会不清晰,所以我并不会使用,至于为什么不愿使用Chrome自带的抓包工具,我会在这个博客文章对应的视频里面讲到,我这里使用的抓包软件是BurpSuite v2.1,感觉还是挺好用的。

(题外话:最近一段时间,习惯常用英语交流,所以可能中文的语序就会看起来很奇怪,但是也可以理解写了什么,我会尽可能的优化好的~!)

D1ZnYQ.png
D1ZmFg.png
软件里面的代理监听设置(http)
D1ZZTS.png
Windows系统自带的自定义代理页面

设置代理的方法我这里就不再过多叙述,这里就简单的两个图片说明下,在软件里面开启本地的8080端口的http代理,没有其他的过多的设置,因为其网站本身也十分的简单,并不需要下面的那些高级的设置。然后就是Windows系统自带的自定义代理页面,在网络-代理选项里面,在里面填写上本地的代理IP也就是127.0.0.1和代理的端口8080,同时也别忘了把下面的192.168.*.*这个代理的选项给删除掉,因为受控制的路由器的IP地址是在这个范围里面的。

综上所述,你的软件就已经连接上互联网了。

接下来就是直接使用浏览器进行模拟的登录操作:

第一步就是进入到页面并且尝试输入一次密码

抓包拦截要求发送请求东西是:

D3yvZR.png
POST /cgi-bin/login HTTP/1.1
Host: 192.168.0.1
Content-Length: 39
Accept: */*
X-Requested-With: XMLHttpRequest
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.198 Safari/537.36
Content-Type: application/x-www-form-urlencoded; charset=UTF-8
Origin: http://192.168.0.1
Referer: http://192.168.0.1/computer/login.html
Accept-Encoding: gzip, deflate
Accept-Language: en-US,en;q=0.9,zh-TW;q=0.8,zh;q=0.7,ko-KR;q=0.6,ko;q=0.5,ja-JP;q=0.4,ja;q=0.3,zh-CN;q=0.2
Connection: close

{"username":"admin","password":"admin"}

这里面{"username":"admin","password":"admin"}就是发送的内容,客户端也就是网页就向服务器发送这个数据以达成登录,这个时候我们可以使用Python中的requests库中的post功能来实现这个。

这里值得注意的一点是,虽然这里直接显示的是以字典形式发送的,但是实际上是发送的这一段的字符串,所以你不能直接发送字典形式的data数据上去。

import requests
if __name__ == '__main__':
    data = '{"username":"admin","password":"admin"}'
    requests.post("http://192.168.0.1/cgi-bin/login",data=data)

其实不然,通过这个很简单的代码,你就可以实现登录到服务器了。

我们再查看这个post任务的响应内容

{"errCode":0,"errMsg":"OK","configDone":false,"errCode":0,"errMsg":"OK","configDone":false}

返回的是一个json内容,但是是字符串的。

如果要检测是否是成功的实现登录到路由器了,可以直接判断是否存在"errCode":0,,如果存在,那么是登录成功了的,如果不存在,则没有登录成功。代码如下:

import requests


def trylogin(ip, password):
    data = '{"username":"admin","password":"' + password + '"}'
    try:
        login = requests.post('http://' + ip + '/cgi-bin/login', data=data)
        return login.text
    except:
        return "error"


if __name__ == '__main__':
    ip = "192.168.0.1"
    password = "admin"
    if '"errCode":0,' in trylogin(ip,password):
        print("登录成功")
    else:
        print("登陆失败")

如果路由器设置正常,那么会输出显示“登陆成功”

在这里我还特地加了检测IP地址和密码选项,方便来进行IP地址和密码的更改

2.重启路由器

研究重启路由器的步骤和如何去登录路由器的步骤大致是相同的,但是我这里把抓取包的过程给换成了拦截包,以便于我们更好的去获取拦截的内容。

DlztFP.png
路由器的重启设置界面
D3yXL9.png
BurpSuite拦截请求功能

打开路由器的重启设置界面,这里有一个重启设备的按钮,在WEB里面点下他,将会自动的进行路由器的重启工作。

首先打开BurpSuite的拦截请求功能,然后按下WEB里面的重启设备按钮,我们将得到这样的请求内容。

D3yxd1.png
软件拦截到的请求
POST /cgi-bin/mbox-config?method=SET&section=system_reboot HTTP/1.1
Host: 192.168.0.1
Content-Length: 2
Accept: application/json, text/javascript, */*; q=0.01
X-Requested-With: XMLHttpRequest
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.198 Safari/537.36
Content-Type: appliation/json
Origin: http://192.168.0.1
Referer: http://192.168.0.1/computer/reboot.html
Accept-Encoding: gzip, deflate
Accept-Language: en-US,en;q=0.9,zh-TW;q=0.8,zh;q=0.7,ko-KR;q=0.6,ko;q=0.5,ja-JP;q=0.4,ja;q=0.3,zh-CN;q=0.2
Cookie: COMFAST_SESSIONID=c0a800db-ffffffc05627ffffff8c3fffffffbb-238e1f29
Connection: close

{}

在这里我们可以知道,要post到的地址是/cgi-bin/mbox-config?method=SET&section=system_reboot ,传输的数据内容是{}

没错,即使内容很简单,我们也要照样子发送上去,那么这一段post请求重启的指令可以写成这样:

request.post('http://' + ip + '/cgi-bin/mbox-config?method=SET&section=system_reboot', data="{}")

如果你直接这样就尝试发送post请求要求路由器重启的话,路由器是肯定不允许的,因为你并没有登录,此时你的路由器是没有登录的状态,可能细心的人已经观察到了,两次拦截到的请求的head里面,第二次多了一层COOKIE,路由器就是凭的你这个COOKIE的密码来进行路由器的一些操作。

但是问题来了,COOKI后面的密码每一次都是变化的,那怎么办呢,如果你正在担心这个的话,肯定也是没有问题的,其实解决他的方法很简单——使用会话保持

我们可以利用

requests.session()

requests这个库里面的会话保持功能来忽视掉这个问题,在第二次发送post的时候,会使用上第一次请求后得到的回复中一些看不见的内容,比如说获取的cookie值来实现第二次的请求。

简而言之:通过会话保持功能,在第一次post登陆请求后就获取到了COOKIE值,第二次post请求的时候会自动的带上这个COOKIE值。有关于会话保持的可以去看看这个文章:requests session的作用 - 简书 (jianshu.com) 这个文章写得还是挺不错的。

关于讲解他的内容太浪费时间了,与其深入了解原理不如知道怎么去使用他,反正只要创建好一个实例化的s,然后我们通过这个s再post就行了

import requests

#实例化s
s = requests.session()

def tryreboot(ip):
    try:
        reboot = s.post('http://' + ip + '/cgi-bin/mbox-config?method=SET&section=system_reboot', data="{}")
        return reboot
    except:
        return ""


def trylogin(ip, password):
    data = '{"username":"admin","password":"' + password + '"}'
    try:
        login = s.post('http://' + ip + '/cgi-bin/login', data=data)
        return login.text
    except:
        return "error"


if __name__ == '__main__':
    ip = "192.168.0.1"
    password = "admin"
    if '"errCode":0,' in trylogin(ip,password):
        print("登录成功")
        if '{"errCode":0,"errMsg":"OK","configDone":false}' in tryreboot(ip).text:
            print("重启成功")
    else:
        print("失败")

只要路由器没有问题,通过这个代码就可以实现路由器的登录+重启。

3.查看路由器实时流量

如果你已经熟练掌握了前面两个解决方案的内容,那么第三个内容肯定是轻而易举的

因为这个路由器提供了一个可以查看路由器实时信息的接口

/cgi-bin/mbox-config?method=GET&section=system_usage

这个接口post请求后收到的就是路由器的总流量、CPU占用、内存占用这些信息

post收到的信息是这样:

{"linedata":{"tx_history":"464494522","rx_history":"2209030368"},"memory":{"usage":"57.60%"},"cpu_usage":"cpu  31990 0 48466 2091595 0 0 66041 0 0 0","cpu_usage":{"cpu_used":"7%"},"errCode":0,"errMsg":"OK","configDone":false}

这里的{"linedata":{"tx_history":"464494522","rx_history":"2209030368"}目前路由器的上行和下行流量,单位是Byte。

我们可以在这里写一个while循环,每间隔1s就去循环一次就去post路由器现在的流量的情况来计算实时流量

while True:
    result = s.post('http://' + ip + '/cgi-bin/mbox-config?method=GET&section=system_usage', data="{}")

除非地球爆炸,这个循环是不会停止。为了尽量保证准确性,我们还可以导入time库里面的sleep来休眠一秒,通过间隔一秒的记录值来进行计算速度:

另外说一下,这里有更好的函数来进行速度的计算,但是我这里选择了更简单的更容易理解的一个方法,这个方法计算出来的值是有误差的。

import requests
import json
import time
#实例化s
s = requests.session()

def trylogin(ip, password):
    data = '{"username":"admin","password":"' + password + '"}'
    try:
        login = s.post('http://' + ip + '/cgi-bin/login', data=data)
        return login.text
    except:
        return "error"

if __name__ == '__main__':
    ip = "192.168.0.1"
    password = "admin"
    if '"errCode":0,' in trylogin(ip,password):
        print("登录成功")
        result_his = s.post('http://' + ip + '/cgi-bin/mbox-config?method=GET&section=system_usage', data="{}")
        tx_his = int(json.loads(result_his.text)["linedata"]["tx_history"]) - 1
        rx_his = int(json.loads(result_his.text)["linedata"]["rx_history"]) - 1
        count = 0
        while True:
            result = s.post('http://' + ip + '/cgi-bin/mbox-config?method=GET&section=system_usage', data="{}")
            num = json.loads(result.text)
            txnum = int(num["linedata"]["tx_history"])
            rxnum = int(num["linedata"]["rx_history"])
            vtxnum = str(txnum - tx_his)
            vrxnum = str(rxnum - rx_his)
            tx_his = txnum
            rx_his = rxnum
            print("\r" +
                  "{: <10}".format("上行:{}".format(vtxnum)) +
                  "{: >10}".format("下行:{}".format(vrxnum)),
                  end="")
            time.sleep(1)
    else:
        print("失败")

这里获取到的网速的内容跟我直接使用json进行处理,更方便进行提取,而不是使用字符串截取,有关于json的使用的话这里就不再讲了,因为比较基础。

输出就是这样的

D32uJx.png

当然直接输出的内容单位是Byte,我们可以再适当包装一下。

import requests
import json
import time
#实例化s
s = requests.session()

def trylogin(ip, password):
    data = '{"username":"admin","password":"' + password + '"}'
    try:
        login = s.post('http://' + ip + '/cgi-bin/login', data=data)
        return login.text
    except:
        return "error"

if __name__ == '__main__':
    ip = "192.168.0.1"
    password = "admin"
    if '"errCode":0,' in trylogin(ip,password):
        print("登录成功")
        result_his = s.post('http://' + ip + '/cgi-bin/mbox-config?method=GET&section=system_usage', data="{}")
        tx_his = int(json.loads(result_his.text)["linedata"]["tx_history"]) - 1
        rx_his = int(json.loads(result_his.text)["linedata"]["rx_history"]) - 1
        count = 0
        while True:
            result = s.post('http://' + ip + '/cgi-bin/mbox-config?method=GET&section=system_usage', data="{}")
            num = json.loads(result.text)
            txnum = int(num["linedata"]["tx_history"])
            rxnum = int(num["linedata"]["rx_history"])

            # 上传速度格式化
            if txnum - tx_his <= 1024 ** 2:
                vtxnum = str(round((txnum - tx_his) / 1024, 2)) + "KB/s"
            elif txnum - tx_his > 1024 ** 2:
                vtxnum = str(round((txnum - tx_his) / 1024 ** 2, 2)) + "MB/s"
            else:
                vtxnum = str(txnum - tx_his)

            # 下载速度格式化
            if rxnum - rx_his <= 1024 ** 2:
                vrxnum = str(round((rxnum - rx_his) / 1024, 2)) + "KB/s"
            elif rxnum - rx_his > 1024 ** 2:
                vrxnum = str(round((rxnum - rx_his) / 1024 ** 2, 2)) + "MB/s"
            else:
                vrxnum = str(rxnum - rx_his)

            tx_his = txnum
            rx_his = rxnum
            print("\r" +
                  "{: <17}".format("上行:{}".format(vtxnum)) +
                  "{: >17}".format("下行:{}".format(vrxnum)),
                  end="")
            time.sleep(1)
    else:
        print("失败")

在显示的效果就是这样

D32KW6.png

基本上的任务完成了,过几天我会更新更多的内容,代码质量也会加强。

在此特别感谢Google Adsence对我的信任,没有他的支持,网站的建设不能到现在的400天~