This repository has no description
10 kB
285 lines
1"""Prometheus exporter for the ARRIS TG3442DE "Vodafone Station" cable router.
2
3Logs into the web UI on every scrape (the router keeps one admin session and
4times it out after 10 minutes idle, so a held session would lock the web UI
5for humans), reads DOCSIS channel status and the status page, and logs out.
6"""
7
8import argparse
9import hashlib
10import json
11import logging
12import re
13import threading
14import time
15from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
16
17import requests
18from cryptography.hazmat.primitives.ciphers.aead import AESCCM
19
20log = logging.getLogger("vodafone-station-exporter")
21
22TIMEOUT = 15
23
24
25class LoginError(Exception):
26 pass
27
28
29def js_var(text: str, name: str) -> str | None:
30 m = re.search(r"\b%s\s*=\s*'([^']*)'" % re.escape(name), text) or re.search(
31 r'\b%s\s*=\s*"([^"]*)"' % re.escape(name), text
32 )
33 return m.group(1) if m else None
34
35
36def js_json(text: str, name: str) -> list:
37 m = re.search(r"\b%s\s*=\s*(\[.*?\]);" % re.escape(name), text, re.S)
38 return json.loads(m.group(1)) if m else []
39
40
41class Router:
42 def __init__(self, url: str, user: str, password: str) -> None:
43 self.url = url.rstrip("/")
44 self.user = user
45 self.password = password
46
47 def fetch(self) -> dict:
48 """One login, the data pages, logout."""
49 s = requests.Session()
50 index = s.get(self.url + "/", timeout=TIMEOUT).text
51 iv_hex, salt_hex = js_var(index, "myIv"), js_var(index, "mySalt")
52 session_id = js_var(index, "currentSessionId")
53 if not (iv_hex and salt_hex and session_id):
54 raise LoginError("login page lacks myIv/mySalt/currentSessionId")
55 iv = bytes.fromhex(iv_hex)
56 # sjcl.misc.pbkdf2 defaults: HMAC-SHA256, here 1000 rounds, 128 bit
57 key = hashlib.pbkdf2_hmac(
58 "sha256", self.password.encode(), bytes.fromhex(salt_hex), 1000, 16
59 )
60 ccm = AESCCM(key, tag_length=16)
61 payload = json.dumps({"Password": self.password, "Nonce": session_id})
62 resp = s.post(
63 self.url + "/php/ajaxSet_Password.php",
64 json={
65 "EncryptData": ccm.encrypt(iv, payload.encode(), b"loginPassword").hex(),
66 "Name": self.user,
67 "AuthData": "loginPassword",
68 },
69 timeout=TIMEOUT,
70 ).json()
71 status = resp.get("p_status", "")
72 # "Match" (user) / "AdminMatch" (admin), like the web UI checks
73 if "Match" not in status:
74 raise LoginError(f"login refused: {resp}")
75 nonce = ccm.decrypt(iv, bytes.fromhex(resp["encryptData"]), b"nonce").decode()
76 headers = {
77 "csrfNonce": nonce,
78 "Referer": self.url + "/",
79 "X-Requested-With": "XMLHttpRequest",
80 }
81 try:
82 session = s.post(
83 self.url + "/php/ajaxSet_Session.php", headers=headers, timeout=TIMEOUT
84 ).text
85 if '"LoginStatus":"yes"' not in session:
86 raise LoginError(f"session not started: {session.strip()}")
87 pages = {}
88 for page in ("status_docsis_data", "overview_data", "status_status_data"):
89 r = s.get(f"{self.url}/php/{page}.php", headers=headers, timeout=TIMEOUT)
90 r.raise_for_status()
91 pages[page] = r.text
92 finally:
93 try:
94 s.post(self.url + "/php/logout.php", headers=headers, timeout=TIMEOUT)
95 except requests.RequestException as e:
96 log.warning("logout failed: %s", e)
97 return pages
98
99
100def esc(v: str) -> str:
101 return str(v).replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")
102
103
104class Metrics:
105 """Gauges in Prometheus text format, samples grouped by family."""
106
107 def __init__(self) -> None:
108 self.families: dict[str, tuple[str, list[str]]] = {}
109
110 def add(self, name: str, help_: str, value: float, **labels: str) -> None:
111 _, samples = self.families.setdefault(name, (help_, []))
112 lbl = ",".join(f'{k}="{esc(v)}"' for k, v in labels.items())
113 samples.append(f"{name}{{{lbl}}} {value}" if lbl else f"{name} {value}")
114
115 def text(self) -> str:
116 out = []
117 for name, (help_, samples) in self.families.items():
118 out += [f"# HELP {name} {help_}", f"# TYPE {name} gauge", *samples]
119 return "\n".join(out) + "\n"
120
121
122def number(v) -> float | None:
123 m = re.match(r"\s*(-?[0-9.]+)", str(v))
124 return float(m.group(1)) if m else None
125
126
127def channel_metrics(m: Metrics, direction: str, channels: list) -> None:
128 counts: dict[str, int] = {}
129 for ch in channels:
130 labels = {
131 "direction": direction,
132 "channel_id": str(ch.get("ChannelID", "")),
133 "channel_type": ch.get("ChannelType", ""),
134 "frequency_mhz": str(ch.get("Frequency", "")),
135 }
136 counts[labels["channel_type"]] = counts.get(labels["channel_type"], 0) + 1
137 # "6.4/66.4": dBmV/dBµV
138 power = number(str(ch.get("PowerLevel", "")).split("/")[0])
139 if power is not None:
140 m.add(
141 "vodafone_station_docsis_power_dbmv",
142 "Channel power level in dBmV",
143 power,
144 **labels,
145 )
146 snr = number(ch.get("SNRLevel", ""))
147 if snr is not None:
148 m.add(
149 "vodafone_station_docsis_snr_db",
150 "Downstream channel SNR (MER) in dB",
151 snr,
152 **labels,
153 )
154 qam = number(re.sub(r"[^0-9]", "", str(ch.get("Modulation", ""))) or "x")
155 if qam is not None:
156 m.add(
157 "vodafone_station_docsis_modulation_qam",
158 "QAM order of the channel (64, 256, 4096, ...)",
159 qam,
160 **labels,
161 )
162 m.add(
163 "vodafone_station_docsis_locked",
164 "1 if the channel is locked (OFDMA upstream: OPERATE)",
165 1 if ch.get("LockStatus") in ("Locked", "OPERATE") else 0,
166 **labels,
167 )
168 for ctype, n in counts.items():
169 m.add(
170 "vodafone_station_docsis_channels",
171 "Number of bonded channels",
172 n,
173 direction=direction,
174 channel_type=ctype,
175 )
176
177
178def collect(router: Router) -> str:
179 m = Metrics()
180 start = time.monotonic()
181 try:
182 pages = router.fetch()
183 except (requests.RequestException, LoginError, ValueError, KeyError) as e:
184 log.error("scrape failed: %s", e)
185 m.add("vodafone_station_up", "1 if the router could be scraped", 0)
186 m.add(
187 "vodafone_station_scrape_duration_seconds",
188 "Time the router scrape took",
189 time.monotonic() - start,
190 )
191 return m.text()
192
193 docsis = pages["status_docsis_data"]
194 overview = pages["overview_data"]
195 status = pages["status_status_data"]
196
197 m.add("vodafone_station_up", "1 if the router could be scraped", 1)
198 m.add(
199 "vodafone_station_scrape_duration_seconds",
200 "Time the router scrape took",
201 time.monotonic() - start,
202 )
203 m.add(
204 "vodafone_station_info",
205 "Router firmware and hardware",
206 1,
207 firmware=js_var(status, "js_FWVersion") or "",
208 hardware=js_var(status, "js_HWTypeVersion") or "",
209 )
210 conn = js_var(overview, "_ga.modemConnectionStatus") or ""
211 m.add(
212 "vodafone_station_docsis_online",
213 "1 if the cable modem reports DOCSIS Online",
214 1 if conn == "DOCSIS Online" else 0,
215 status=conn,
216 )
217 m.add(
218 "vodafone_station_cm_operational",
219 "1 if the cable modem reports itself operational",
220 1 if js_var(overview, "js_isCmOperational") == "1" else 0,
221 )
222 # "29,17,53": days,hours,minutes
223 uptime = js_var(status, "js_UptimeSinceReboot") or ""
224 parts = [int(p) for p in re.findall(r"\d+", uptime)]
225 if len(parts) == 3:
226 d, h, mi = parts
227 m.add(
228 "vodafone_station_uptime_seconds",
229 "Time since the router booted (minute resolution)",
230 ((d * 24 + h) * 60 + mi) * 60,
231 )
232 hosts = js_var(overview, "js_hostNums")
233 if hosts is not None and hosts.isdigit():
234 m.add("vodafone_station_hosts", "Hosts attached to the router", int(hosts))
235
236 channel_metrics(m, "downstream", js_json(docsis, "json_dsData"))
237 channel_metrics(m, "upstream", js_json(docsis, "json_usData"))
238 return m.text()
239
240
241def main() -> None:
242 p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
243 p.add_argument("--router", default="http://192.168.0.1")
244 p.add_argument("--user", default="admin")
245 p.add_argument("--password-file", required=True)
246 p.add_argument("--listen", default="127.0.0.1:9764")
247 p.add_argument("--once", action="store_true", help="print metrics once and exit")
248 args = p.parse_args()
249 logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
250
251 with open(args.password_file) as f:
252 password = f.read().strip()
253 router = Router(args.router, args.user, password)
254
255 if args.once:
256 print(collect(router), end="")
257 return
258
259 # The router has one session; concurrent scrapes would log each other out.
260 lock = threading.Lock()
261
262 class Handler(BaseHTTPRequestHandler):
263 def do_GET(self) -> None:
264 if self.path != "/metrics":
265 self.send_error(404)
266 return
267 with lock:
268 body = collect(router).encode()
269 self.send_response(200)
270 self.send_header("Content-Type", "text/plain; version=0.0.4")
271 self.send_header("Content-Length", str(len(body)))
272 self.end_headers()
273 self.wfile.write(body)
274
275 def log_message(self, fmt, *a) -> None:
276 pass
277
278 host, _, port = args.listen.rpartition(":")
279 server = ThreadingHTTPServer((host.strip("[]"), int(port)), Handler)
280 log.info("listening on %s", args.listen)
281 server.serve_forever()
282
283
284if __name__ == "__main__":
285 main()