DECOルーターをSeleniumで「ハック」して家族の在宅をリアルタイム検知

# 通常のSelenium（レスポンスボディは取得不可）
# from selenium import webdriver

# selenium-wire（ドロップイン置き換え — レスポンスボディも参照可能）
from seleniumwire import webdriver

sudo apt-get update
sudo apt-get install -y chromium-browser chromium-chromedriver

pip install selenium-wire pycryptodome

import json, re, base64, urllib.parse, time
from seleniumwire import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from Crypto.Cipher import AES
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
CHROMEDRIVER_PATH = "/usr/bin/chromedriver"
DECO_URL = "http://192.168.xx.1/webpages/index.html"
PASSWORD = "<your-deco-password>"
SAME_THRESHOLD = 3  # 同一データが何回続いたら再ログインするか

options = Options()
options.add_argument("--headless")
options.add_argument("--no-sandbox")
options.add_argument("--disable-dev-shm-usage")
options.add_argument("--window-size=1920,1080")
options.add_argument("--force-device-scale-factor=2")
options.add_argument("--disable-gpu")
options.add_argument("--disable-software-rasterizer")
options.add_argument("--no-zygote")          # Chromiumゾンビプロセス対策
options.add_argument("--single-process")
options.add_argument("--disable-extensions")
options.add_argument("--disable-background-networking")
options.add_argument("--disable-sync")
options.add_argument("--metrics-recording-only")

def wait_for_localstorage(driver, key, timeout=10):
    """localStorageに値がセットされるまでポーリングして待機する"""
    end = time.time() + timeout
    while time.time() < end:
        val = driver.execute_script(f"return localStorage.getItem('{key}');")
        if val:
            return val
        time.sleep(0.5)
    return None
def deco_decrypt(data_base64, key_str, iv_str):
    """AES-CBCで暗号化されたBase64データを復号してJSONとして返す"""
    key = key_str.encode('utf-8')
    iv  = iv_str.encode('utf-8')
    data_bytes = base64.b64decode(data_base64)
    cipher = AES.new(key, AES.MODE_CBC, iv)
    decrypted_bytes = cipher.decrypt(data_bytes)
    # PKCS7 パディング除去
    pad_len = decrypted_bytes[-1]
    decrypted_bytes = decrypted_bytes[:-pad_len]
    return json.loads(decrypted_bytes.decode('utf-8'))
def decode_name(name_str):
    """Base64+URLエンコードされたデバイス名を可読な文字列にデコードする"""
    try:
        decoded = base64.b64decode(name_str).decode('utf-8')
        decoded = urllib.parse.unquote(decoded)
        return decoded
    except Exception:
        return name_str  # デコード失敗時は元の文字列をそのまま返す

def deco_login(service, options):
    driver = webdriver.Chrome(service=service, options=options)
    driver.get(DECO_URL)
    # パスワード入力欄が表示されるまで待機してパスワードを入力
    pw_input = WebDriverWait(driver, 10).until(
        EC.presence_of_element_located((By.CSS_SELECTOR, "input.password-text"))
    )
    pw_input.clear()
    pw_input.send_keys(PASSWORD)
    pw_input.send_keys(Keys.ENTER)
    time.sleep(5)  # ログイン完了を待機
    # AES鍵とIVをlocalStorageから取得（値がセットされるまでポーリング）
    encryptorAES = wait_for_localstorage(driver, "encryptorAES")
    if not encryptorAES:
        raise RuntimeError("encryptorAES取得失敗（ログイン未完了）")
    # 形式: k=1234567890123456&i=1234567890123456
    m = re.search(r'k=(\d+)&i=(\d+)', encryptorAES)
    key_str = m.group(1)
    iv_str  = m.group(2)
    # stokセッショントークンをインターセプトしたURLから抽出
    stok = None
    timeout = time.time() + 10
    while time.time() < timeout:
        for request in driver.requests:
            if request.response and "stok=" in request.url:
                m2 = re.search(r"stok=([^/]+)", request.url)
                if m2:
                    stok_val = m2.group(1)
                    if stok_val and stok_val != "login":
                        stok = stok_val
                        break
        if stok:
            break
        time.sleep(0.5)
    if stok is None:
        raise RuntimeError("stok が取得できませんでした")
    return driver, key_str, iv_str

def main_loop():
    service = Service(CHROMEDRIVER_PATH)
    driver, key_str, iv_str = deco_login(service, options)
    prev_json_str = None
    same_count = 0
    last_update = time.time()
    try:
        while True:
            # selenium-wireが傍受したリクエストから最新のclient_listレスポンスを取得
            latest_req = None
            for req in reversed(driver.requests):
                if req.response and "client?form=client_list" in req.url:
                    latest_req = req
                    break
            if latest_req:
                body = latest_req.response.body.decode()
                # JSONではないレスポンスはスキップ
                if not body or not body.strip().startswith("{"):
                    driver.requests.clear()
                    time.sleep(5)
                    continue
                resp_json = json.loads(body)
                if "data" not in resp_json:
                    raise RuntimeError("セッション切れの可能性")
                # AES-CBCで復号
                decrypted_json = deco_decrypt(resp_json["data"], key_str, iv_str)
                # ---- 同一データ連続対策 ----
                current_json_str = json.dumps(decrypted_json, sort_keys=True)
                if current_json_str == prev_json_str:
                    same_count += 1
                else:
                    same_count = 0
                prev_json_str = current_json_str
                if same_count >= SAME_THRESHOLD:
                    # キャッシュが古くなっているため再ログイン
                    driver.quit()
                    driver, key_str, iv_str = deco_login(service, options)
                    prev_json_str = None
                    same_count = 0
                    continue
                # クライアントリストを処理してInfluxDBへ書き込み
                client_list = decrypted_json.get("result", {}).get("client_list", [])
                clients = []
                for client in client_list:
                    if "name" in client and client["name"]:
                        decoded_name = decode_name(client["name"])
                        client["name"] = decoded_name
                        client["owner"] = family_devices_map.get(decoded_name)
                        client["is_presence_device"] = decoded_name in family_devices_map
                    clients.append(client)
                influx.insert_connected_deco_device_data(__file__, clients)
                last_update = time.time()
            driver.requests.clear()  # 傍受済みリクエストをクリア（メモリ対策）
            time.sleep(5)
            # 5分間データ更新がなければ再ログインのためにループを抜ける
            if time.time() - last_update > 300:
                break
    finally:
        if driver:
            driver.quit()
# 外側の再ログインループ — 例外や5分タイムアウト時に自動で再実行
while True:
    try:
        main_loop()
    except Exception:
        time.sleep(5)

# family_device_status_record.py より（在宅判定ロジックの抜粋）
# ①DECO接続確認（is_deco_connected）: このプログラムが書き込んだデータを参照
# ②ping疎通確認（is_ping_reachable）: 並列でpingを実行
# ③ARP確認（is_arp_found）: ARPテーブルにIPが存在するか確認
ping_ok  = fd.get("is_ping_reachable", False)
deco_ok  = fd.get("is_deco_connected", False)
arp_ok   = fd.get("is_arp_found", False)
# 3つのうちいずれか1つでも確認できれば在宅と判定
is_connected = ping_ok or deco_ok or arp_ok

python3 /home/<username>/Projects/smarthome/core/record/lan/lan_status_record_from_deco.py

クライアントリストを監視中...
最新のクライアントリストリクエストを取得: http://192.168.xx.1/stok=XXXXXXXX/ds?...
復号されたJSON: {'error_code': 0, 'result': {'client_list': [...]}}
同一データ連続カウント: 0
[{'name': 'タロウのiPhone', 'ip': '192.168.xx.xx', ...}, ...]

# InfluxDB 2.x CLI で確認
influx query 'from(bucket: "smarthome") |> range(start: -1m) |> filter(fn: (r) => r._measurement == "client")'

# /etc/systemd/system/deco-lan-recorder.service
[Unit]
Description=DECO LAN Status Recorder
After=network.target
[Service]
ExecStart=/usr/bin/python3 /home/<username>/Projects/smarthome/core/record/lan/lan_status_record_from_deco.py
WorkingDirectory=/home/<username>/Projects/smarthome
Restart=always
RestartSec=5
User=<username>
[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable deco-lan-recorder
sudo systemctl start deco-lan-recorder
sudo systemctl status deco-lan-recorder