Home Assistant を使っていたら、UPnP の機能でルータのパケット情報を取得しているらしい。ではどうやって取得するのか？

SSDPでUPnPデバイスを探す

Gemini に「UPnPの機能で Buffalo のルータの通信速度などの情報を調べるには？」と聞いたら、「SSDP（Simple Service Discovery Protocol）を使用してネットワーク上のUPnP対応デバイスを検索します。これは、マルチキャストアドレス239.255.255.250のUDPポート1900にM-SEARCHリクエストを送信することで行います。」とのアドバイスがもらえた。

「取得するプログラムを提案して」と聞いたら、以下の Python コードを教えてくれた。

import socket
import time

# SSDPマルチキャストアドレスとポート
SSDP_ADDR = '239.255.255.250'
SSDP_PORT = 1900

# M-SEARCHリクエストの作成
# ST: Service Type、ここではルーター（WANIPConnection）を指定
# MX: Maximum wait time in seconds
msearch_request = '\r\n'.join([
    'M-SEARCH * HTTP/1.1',
    'HOST: {}:{}'.format(SSDP_ADDR, SSDP_PORT),
    'MAN: "ssdp:discover"',
    'ST: urn:schemas-upnp-org:service:WANIPConnection:1',
    'MX: 3',
    '',
    ''
])

# ソケットの作成
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.settimeout(5)  # 応答を待機する時間（秒）

print("M-SEARCHリクエストを送信中...")

try:
    # リクエストの送信
    sock.sendto(msearch_request.encode('utf-8'), (SSDP_ADDR, SSDP_PORT))

    while True:
        try:
            # 応答の受信
            data, addr = sock.recvfrom(1024)
            print("-" * 20)
            print("デバイスを発見しました！")
            print("IPアドレス:", addr[0])
            print("ポート:", addr[1])
            print("\n応答内容:")
            print(data.decode('utf-8'))

        except socket.timeout:
            print("-" * 20)
            print("タイムアウトしました。すべての応答を受信したか、デバイスが見つかりませんでした。")
            break

except Exception as e:
    print("エラーが発生しました:", e)

finally:
    sock.close()

SSDPの返答から詳細取得

実行したら、以下の情報が得られた。これより、/getedesc.xml などで 49355 ポートに問い合わせすればいいみたい。

M-SEARCHリクエストを送信中...
--------------------
デバイスを発見しました！
IPアドレス: 192.168.xx.1
ポート: 1900

応答内容:
HTTP/1.1 200 OK
CACHE-CONTROL: max-age=1800
DATE: Tue, 09 Sep 2025 13:57:21 GMT
EXT:
LOCATION: http://192.168.xx.1:49355/gatedesc.xml
SERVER: Linux/3.14.77 UPnP/1.0 Intel_SDK_for_UPnP_devices/1.2
ST: urn:schemas-upnp-org:service:WANIPConnection:1
USN: (略)

• http://192.168.xx.1:49355/gatedesc.xml
• http://192.168.xx.1:49355/gateinfoSCPD.xml – 情報量少ない
• http://192.168.xx.1:49355/gateicfgSCPD.xml
• http://192.168.xx.1:49355/gateconnSCPD.xml

送受信パケット量を取得

いろいろと試す中、送受信パケットを取得する lua スクリプトを Gemini に作らせたら、下記のようになった。

-- socket.http ライブラリをロードします。
-- 'luasocket'パッケージが必要です。
local http = require("socket.http")
local ltn12 = require("ltn12")

-- UPnPサービス情報
local url = "http://192.168.xx.1:49355/upnp/control/WANCommonIFC1"

-- リクエスト情報を格納するテーブル
local requests = {
    {
        action = "GetTotalBytesReceived",
        tag = "<NewTotalBytesReceived>",
        label = "受信"
    },
    {
        action = "GetTotalBytesSent",
        tag = "<NewTotalBytesSent>",
        label = "送信"
    }
}

-- データを取得する関数
local function get_upnp_data(req_info)
    local soap_data = string.format([[<?xml version="1.0"?>
<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/" s:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/">
  <s:Body>
    <u:%s xmlns:u="urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1">
    </u:%s>
  </s:Body>
</s:Envelope>]], req_info.action, req_info.action)

    local headers = {
        ["Content-Type"] = "text/xml; charset=\"utf-8\"",
        ["SOAPAction"] = '"urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1#' .. req_info.action .. '"'
    }

    print(req_info.label .. "データを取得中...")

    local response_body = {}
    local res, status_code = http.request{
        url = url,
        method = "POST",
        headers = headers,
        source = ltn12.source.string(soap_data),
        sink = ltn12.sink.table(response_body),
        timeout = 10
    }

    if status_code and (status_code == 200 or status_code == "200 OK") then
        local response_text = table.concat(response_body)

        local start_tag = req_info.tag
        local end_tag = string.gsub(start_tag, "<", "</")
        
        local _, _, bytes_string = string.find(response_text, start_tag .. "(.*)" .. end_tag)

        if bytes_string then
            local total_bytes = tonumber(bytes_string)
            local megabytes = total_bytes / 1024 / 1024
            
            print(string.format("合計%sバイト数: %d", req_info.label, total_bytes))
            print(string.format("合計%sメガバイト数: %.2f MB", req_info.label, megabytes))
        else
            print(req_info.label .. "データが見つかりませんでした。")
            print("応答内容:\n" .. response_text)
        end
    else
        print(req_info.label .. "データの取得に失敗しました。")
        print("ステータスコード: " .. (status_code or "不明"))
        print("応答内容:\n" .. table.concat(response_body))
    end
    print("----------------------------------------")
end

-- 各リクエストを実行
for _, req_info in ipairs(requests) do
    get_upnp_data(req_info)
end

SSDPとそのあとの処理をまとめた lua スクリプトは以下となった。

• airstation.lua

munin でのパケット流量観測に組み込んでみた。

Gemini中心のコーディング

でも、今回のプラグイン作成は、Gemini で作らせた。オレがやったのは生成されたプログラムをテストして、プログラム作ってとか、Lua で作ってとか、データ抽出では厳密なXML解析せず正規表現で作って…とか、munin のプラグインに直してとか、命令するばかりで、エディタで悩みながらコード書きはしなかった。エディタ使ったのは、コードの確認・コメント加筆・余計な動作検証の print を消す程度。