基于Nodered的曲线监控集客AP网络速度

guitengyue · 发表于昨天 22:54

家里埋线有点小问题，导致部分AP定期会跳到100Mbps的网速，时不时去查看又嫌麻烦，于是有了本教程。。。

直接用nodered的http request去调用一直不成功，于是尝试用ubuntu中去识别，直接成功，感谢chatgpt的帮助

一、ubuntu直接安装nodered （不要使用docker等）

二、创建 python脚本，存放到目录/home/xxx/jike/ap_status.py，修改对应的ap的ip即可，密码是默认的admin

2.1 适用于7系列的集客AP （小米AX5，1800，N3000等）

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import time
import hashlib
import requests


BASE_URL = "http://192.168.0.107"
USERNAME = "root"
PASSWORD = "admin"  # 网页登录明文密码


def md5(s: str) -> str:
    return hashlib.md5(s.encode("utf-8")).hexdigest()


def main():
    s = requests.Session()

    # 1️⃣ 先访问首页拿初始 sysauth（跟你浏览器访问 http://192.168.0.107/?t=... 一样）
    t = int(time.time() * 1000)
    resp = s.get(f"{BASE_URL}/", params={"t": t}, timeout=5)
    resp.raise_for_status()

    # 2️⃣ 调用 /cgi-bin/api/admin/getrandom 获取 random
    randtime = int(time.time() * 1000)
    resp = s.get(f"{BASE_URL}/cgi-bin/api/admin/getrandom", params={"randtime": randtime}, timeout=5)
    resp.raise_for_status()
    data = resp.json()

    if data.get("ret") != 1:
        raise RuntimeError(f"getrandom ret != 1: {data}")

    random = data["random"]

    # 3️⃣ 按正确算法生成加密密码：MD5(random + 明文密码)
    hashed_pwd = md5(random + PASSWORD)

    # 4️⃣ POST /cgi-bin/api/admin/sysauth 登录
    login_url = f"{BASE_URL}/cgi-bin/api/admin/sysauth"
    payload = {
        "username": USERNAME,
        "password": hashed_pwd,
    }

    resp = s.post(login_url, data=payload, timeout=5)
    resp.raise_for_status()
    login_result = resp.json()

    if login_result.get("ret") != 1:
        raise RuntimeError(f"登录失败: {login_result}")

    # 5️⃣ 登录成功后，用同一个 session 获取 overview 数据
    randtime2 = int(time.time() * 1000)
    status_url = f"{BASE_URL}/cgi-bin/api/admin/status/overview"
    resp = s.get(status_url, params={"status": 1, "randtime": randtime2}, timeout=5)
    resp.raise_for_status()

    try:
        status_data = resp.json()
        
        # 提取 ipaddr 和 speed
        ipaddr = status_data.get('lan', {}).get('ipaddr', 'Unknown')
        speed = status_data.get('lan', {}).get('speed', 'Unknown')
        
        # 打印只需要的两条信息
        print(f"[+] IP 地址: {ipaddr}")
        print(f"[+] 连接速度: {speed}")

    except ValueError:
        # 如果不是 JSON 格式的数据，输出原始响应
        print("[+] overview 原始响应文本:")
        print(resp.text)


if __name__ == "__main__":
    main()

2.2 适用于8系列版本N3000，主要是https的调用

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import time
import hashlib
import requests
import urllib3

# 关掉因为 verify=False 产生的告警
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

BASE_URL = "https://192.168.0.131"   # 🔴 这里改成 https
USERNAME = "root"
PASSWORD = "admin"


def md5(s: str) -> str:
    return hashlib.md5(s.encode("utf-8")).hexdigest()


def main():
    s = requests.Session()
    s.verify = False   # 🔴 所有请求都不校验证书

    # 1️⃣ 访问首页拿初始 cookie
    t = int(time.time() * 1000)
    resp = s.get(f"{BASE_URL}/", params={"t": t}, timeout=5)
    resp.raise_for_status()

    # 2️⃣ /getrandom
    randtime = int(time.time() * 1000)
    resp = s.get(f"{BASE_URL}/cgi-bin/api/admin/getrandom",
                 params={"randtime": randtime},
                 timeout=5)
    resp.raise_for_status()
    data = resp.json()
    if data.get("ret") != 1:
        raise RuntimeError(f"getrandom ret != 1: {data}")
    random = data["random"]

    # 3️⃣ 生成密码：MD5(random + 明文)
    hashed_pwd = md5(random + PASSWORD)

    # 4️⃣ 登录
    login_url = f"{BASE_URL}/cgi-bin/api/admin/sysauth"
    payload = {"username": USERNAME, "password": hashed_pwd}
    resp = s.post(login_url, data=payload, timeout=5)
    resp.raise_for_status()
    login_result = resp.json()
    if login_result.get("ret") != 1:
        raise RuntimeError(f"登录失败: {login_result}")

    # 5️⃣ 获取 overview
    randtime2 = int(time.time() * 1000)
    status_url = f"{BASE_URL}/cgi-bin/api/admin/status/overview"
    resp = s.get(status_url,
                 params={"status": 1, "randtime": randtime2},
                 timeout=5)
    resp.raise_for_status()
    status_data = resp.json()

    ipaddr = status_data.get("lan", {}).get("ipaddr", "Unknown")
    speed = status_data.get("lan", {}).get("speed", "Unknown")
    print(f"[+] ipaddr: {ipaddr} [+] speed: {speed}")


if __name__ == "__main__":
    main()

2.3 适用于7.2版本固定N3000 （当时就这个一直出错，又重新抓了包，改了脚本，适配了一下）

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import time
import hashlib
import requests

BASE_URL = "http://192.168.1.105"
USERNAME = "root"
PASSWORD = "admin"  # 网页登录明文密码


def md5(s: str) -> str:
    return hashlib.md5(s.encode("utf-8")).hexdigest()


def main():
    s = requests.Session()

    # 1️⃣ 访问首页，拿初始 cookie
    t = int(time.time() * 1000)
    print(f"[+] GET /?t={t}  获取初始 cookie")
    resp = s.get(f"{BASE_URL}/", params={"t": t}, timeout=5)
    resp.raise_for_status()
    print("[+] 初始 cookies:", s.cookies.get_dict())

    # 2️⃣ /getrandom 获取 random
    randtime = int(time.time() * 1000)
    print(f"[+] GET /cgi-bin/api/admin/getrandom?randtime={randtime}")
    resp = s.get(f"{BASE_URL}/cgi-bin/api/admin/getrandom",
                 params={"randtime": randtime},
                 timeout=5)
    resp.raise_for_status()
    data = resp.json()
    print("[+] getrandom 返回:", data)

    if data.get("ret") != 1:
        raise RuntimeError(f"getrandom ret != 1: {data}")

    random = data["random"]
    print("[+] 明文密码:", PASSWORD)
    print("[+] random   :", random)

    # 3️⃣ 生成密码：MD5(random + 明文密码)
    hashed_pwd = md5(random + PASSWORD)
    print("[+] 加密后密码（应和浏览器抓包的一样）:", hashed_pwd)

    # 4️⃣ 登录 /sysauth
    login_url = f"{BASE_URL}/cgi-bin/api/admin/sysauth"
    payload = {
        "username": USERNAME,
        "password": hashed_pwd,
    }

    print(f"[+] POST {login_url} 尝试登录")
    resp = s.post(login_url, data=payload, timeout=5)
    resp.raise_for_status()
    login_result = resp.json()
    print("[+] 登录返回:", login_result)
    print("[+] 登录后 cookies:", s.cookies.get_dict())

    if login_result.get("ret") != 1:
        raise RuntimeError(f"登录失败: {login_result}")

    # 虽然返回了 token，但从抓包看 overview 根本没用它
    token = login_result.get("token")
    print("[+] 登录 token:", token)

    # 5️⃣ 模仿浏览器请求 /status/overview：
    #    - GET 方法
    #    - URL 参数只有 status / randtime
    #    - Cookie 里带 sysauth（Session 自动处理）
    #    - Header 额外再带一个 sysauth: <same value>
    randtime2 = int(time.time() * 1000)
    status_url = f"{BASE_URL}/cgi-bin/api/admin/status/overview"

    # 从当前 session 的 cookie 里拿 sysauth
    sysauth_val = s.cookies.get("sysauth", "")
    print("[+] 用于 overview 的 sysauth:", sysauth_val)

    params = {
        "status": 1,
        "randtime": randtime2,
    }

    headers = {
        # 浏览器请求里确实有这个自定义头
        "sysauth": sysauth_val,
        # 其它头一般不强制，但你想更像浏览器可以加：
        # "Accept": "application/json, text/plain, */*",
        # "User-Agent": "Mozilla/5.0 ...",
        # "Referer": f"{BASE_URL}/?t={t}",
    }

    print(f"[+] GET {status_url}  带参数: {params}  头: {headers}")
    resp = s.get(status_url, params=params, headers=headers, timeout=5)
    resp.raise_for_status()

    try:
        status_data = resp.json()
        print("[+] overview JSON 数据:")
        print(status_data)

        ipaddr = status_data.get("lan", {}).get("ipaddr", "Unknown")
        speed = status_data.get("lan", {}).get("speed", "Unknown")
        print(f"[+] ipaddr: {ipaddr} [+] speed: {speed}")

    except ValueError:
        print("[+] overview 原始响应文本:")
        print(resp.text)


if __name__ == "__main__":
    main()

三、nodered中用exec，创建工作流，命令是python3 /home/XXX/jike/ap_status.py，然后第一个输出节点，连function，提取当前速度

var output = msg.payload;

// 使用正则表达式提取 ipaddr 后面的 IP 地址
var ipaddrMatch = output.match(/speed:\s*([\d\.]+)/);

// 提取到的 ipaddr 值，如果没有匹配到则为 null
var ipaddr = ipaddrMatch ? ipaddrMatch[1] : null;

// 将提取到的 ipaddr 设置为 msg.payload，以便后续处理
msg.payload = ipaddr;

return msg;

四、继续跟上mqtt，主题内填写：homeassistant/sensor/jikeapspeed1
五、homeassistant中添加代码：

mqtt:
  sensor:

# 192.168.0.198
    - name: "AP-girl-3l"
      state_topic: "homeassistant/sensor/jikeapspeed1"
      unit_of_measurement: "M"
      value_template: "{{ value | int }}"

六、最终效果如下

七、nodered中添加监控，回传企业微信通知，大功告成！

		自动登录	找回密码
密码			立即注册

[教程系列] 基于Nodered的曲线监控集客AP网络速度

浏览过的版块