healthcheck-container/wifi.py

import os
import re
import subprocess

SYS_NET = "/sys/class/net"

_KV = re.compile(r"^\s*([^:]+?):\s*(.*)$")
_BSS = re.compile(r"^BSS ([0-9a-f]{2}(?::[0-9a-f]{2}){5})", re.IGNORECASE)


class WifiError(Exception):
    pass


def interfaces():
    try:
        names = sorted(os.listdir(SYS_NET))
    except OSError:
        return []
    result = []
    for name in names:
        path = os.path.join(SYS_NET, name)
        wireless = os.path.isdir(os.path.join(path, "wireless"))
        physical = os.path.exists(os.path.join(path, "device"))
        if wireless or physical:
            result.append({"name": name, "wireless": wireless})
    return result


def freq_to_channel(freq):
    if freq is None:
        return None
    if freq == 2484:
        return 14
    if 2412 <= freq <= 2472:
        return (freq - 2407) // 5
    if 5160 <= freq <= 5885:
        return (freq - 5000) // 5
    if 5955 <= freq <= 7115:
        return (freq - 5950) // 5
    return None


def freq_to_band(freq):
    if freq is None:
        return None
    if freq < 3000:
        return "2.4 GHz"
    if freq < 5925:
        return "5 GHz"
    return "6 GHz"


def _run(args, timeout):
    try:
        proc = subprocess.run(["iw", *args], capture_output=True, text=True, timeout=timeout)
    except FileNotFoundError:
        raise WifiError("the 'iw' tool is not installed in the container image")
    except subprocess.TimeoutExpired:
        raise WifiError(f"'iw {' '.join(args)}' timed out after {timeout}s")
    if proc.returncode != 0:
        raise WifiError(_explain(proc.stderr.strip() or proc.stdout.strip(), args))
    return proc.stdout


def _explain(stderr, args):
    if "Operation not permitted" in stderr:
        return "permission denied: scanning needs the NET_ADMIN capability (cap_add: NET_ADMIN)"
    if "Device or resource busy" in stderr:
        return "device busy: another scan is in progress, will retry on the next interval"
    if "No such device" in stderr:
        return "interface disappeared"
    if "Network is down" in stderr:
        return "interface is down"
    return f"iw {' '.join(args)} failed: {stderr}"


def link(iface):
    return parse_link(_run(["dev", iface, "link"], timeout=10))


def station(iface):
    return parse_station_dump(_run(["dev", iface, "station", "dump"], timeout=10))


def scan(iface):
    return parse_scan(_run(["dev", iface, "scan"], timeout=40))


def _first_number(value):
    return float(value.split()[0])


def parse_link(text):
    lines = text.strip().splitlines()
    if not lines or not lines[0].startswith("Connected to"):
        return {"connected": False}
    info = {"connected": True, "bssid": lines[0].split()[2].lower()}
    for line in lines[1:]:
        match = _KV.match(line)
        if not match:
            continue
        key, value = match.group(1).strip(), match.group(2).strip()
        if key == "SSID":
            info["ssid"] = value
        elif key == "freq":
            info["freq_mhz"] = int(_first_number(value))
        elif key == "signal":
            info["signal_dbm"] = int(_first_number(value))
        elif key == "tx bitrate":
            info["tx_bitrate_mbps"] = _first_number(value)
        elif key == "rx bitrate":
            info["rx_bitrate_mbps"] = _first_number(value)
    freq = info.get("freq_mhz")
    info["channel"] = freq_to_channel(freq)
    info["band"] = freq_to_band(freq)
    return info


def parse_station_dump(text):
    info = {}
    for line in text.splitlines():
        if line.startswith("Station") and info:
            break
        match = _KV.match(line)
        if not match:
            continue
        key, value = match.group(1).strip(), match.group(2).strip()
        if key == "tx retries":
            info["tx_retries"] = int(value)
        elif key == "tx failed":
            info["tx_failed"] = int(value)
        elif key == "beacon loss":
            info["beacon_loss"] = int(value)
        elif key == "connected time":
            info["connected_s"] = int(_first_number(value))
        elif key == "signal avg":
            info["signal_avg_dbm"] = int(_first_number(value))
    return info


def parse_scan(text):
    aps = []
    current = None
    for line in text.splitlines():
        match = _BSS.match(line)
        if match:
            current = {
                "bssid": match.group(1).lower(),
                "ssid": "",
                "associated": "associated" in line,
                "_rsn": False,
                "_wpa": False,
                "_sae": False,
                "_privacy": False,
            }
            aps.append(current)
            continue
        if current is None:
            continue
        stripped = line.strip()
        if stripped.startswith("freq:"):
            current["freq_mhz"] = int(_first_number(stripped[5:]))
        elif stripped.startswith("signal:"):
            current["signal_dbm"] = round(_first_number(stripped[7:]), 1)
        elif stripped.startswith("SSID:"):
            current["ssid"] = stripped[5:].strip()
        elif stripped.startswith("RSN:"):
            current["_rsn"] = True
        elif stripped.startswith("WPA:"):
            current["_wpa"] = True
        elif stripped.startswith("capability:") and "Privacy" in stripped:
            current["_privacy"] = True
        elif "Authentication suites:" in stripped and "SAE" in stripped:
            current["_sae"] = True

    for ap in aps:
        freq = ap.get("freq_mhz")
        ap["channel"] = freq_to_channel(freq)
        ap["band"] = freq_to_band(freq)
        ap["security"] = _security_label(ap)
        for key in ("_rsn", "_wpa", "_sae", "_privacy"):
            del ap[key]
    aps.sort(key=lambda ap: ap.get("signal_dbm", -1000), reverse=True)
    return aps


def _security_label(ap):
    labels = []
    if ap["_sae"]:
        labels.append("WPA3")
    if ap["_rsn"]:
        labels.append("WPA2")
    if ap["_wpa"]:
        labels.append("WPA")
    if labels:
        return "/".join(labels)
    if ap["_privacy"]:
        return "WEP"
    return "Open"