Add network monitoring, dashboard, and Home Assistant MQTT publishing
Keep /alive unchanged and serve a dashboard at / with status cards, latency graphs, an event timeline, and a Wi-Fi section. Background monitors on configurable intervals: TCP reachability, DNS resolution, public IP change detection, host reboot detection, Wi-Fi link quality, and Wi-Fi scan of networks in range via iw. The Wi-Fi interface is selected with WIFI_INTERFACE; the page lists visible interfaces and reports when none is available. Scan results stay in memory, other metrics go to SQLite on a mounted volume. Optional MQTT publishing with Home Assistant discovery groups all sensors under one device per host. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TdvJYoY8kc8bmBck89U2i6
This commit is contained in:
Milan PandurovandClaude Fable 5.1 committed
1 parent
aace768660
commit
7206e0c4db
13 files changed
+1509
-6
No files matched your search
@@ -0,0 +1,200 @@
|
||||
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"
|
||||
Reference in new issue
Block a user