This repository has no description
tools.vodafone-station-exporter: init
author
lassulus
committer
lassulus
date
(Oct 18, 2026, 10:33 AM CEST)
commit
ce307ded
ce307ded54c44c8436efb8bbf35cd271e65190fb
parent
fde2ec74
fde2ec7491d59daf11699cfb0a54b675bad58616
+323
+22
tools/vodafone-station-exporter/flake-module.nix
+22
tools/vodafone-station-exporter/flake-module.nix
···
1
+
{ ... }:
2
+
{
3
+
perSystem =
4
+
{ pkgs, ... }:
5
+
{
6
+
# Prometheus exporter for the ARRIS TG3442DE "Vodafone Station":
7
+
# DOCSIS channel power/SNR/lock and router status from its web UI.
8
+
packages.vodafone-station-exporter =
9
+
(pkgs.writeShellApplication {
10
+
name = "vodafone-station-exporter";
11
+
text = ''
12
+
exec ${
13
+
pkgs.python3.withPackages (p: [
14
+
p.requests
15
+
p.cryptography
16
+
])
17
+
}/bin/python3 ${./vodafone_station_exporter.py} "$@"
18
+
'';
19
+
}).overrideAttrs
20
+
{ passthru.usage = builtins.readFile ./usage.kdl; };
21
+
};
22
+
}
+16
tools/vodafone-station-exporter/usage.kdl
+16
tools/vodafone-station-exporter/usage.kdl
···
1
+
bin "vodafone-station-exporter"
2
+
about "Prometheus exporter for the ARRIS TG3442DE Vodafone Station: logs into the web UI per scrape, exports DOCSIS channel and router status"
3
+
4
+
flag "--router" help="Router base URL (default: http://192.168.0.1)" {
5
+
arg "<url>"
6
+
}
7
+
flag "--user" help="Web UI user (default: admin)" {
8
+
arg "<name>"
9
+
}
10
+
flag "--password-file" help="File holding the web UI password" required=#true {
11
+
arg "<file>"
12
+
}
13
+
flag "--listen" help="host:port serving /metrics (default: 127.0.0.1:9764)" {
14
+
arg "<host:port>"
15
+
}
16
+
flag "--once" help="Print the metrics once and exit"
+285
tools/vodafone-station-exporter/vodafone_station_exporter.py
+285
tools/vodafone-station-exporter/vodafone_station_exporter.py
···
1
+
"""Prometheus exporter for the ARRIS TG3442DE "Vodafone Station" cable router.
2
+
3
+
Logs into the web UI on every scrape (the router keeps one admin session and
4
+
times it out after 10 minutes idle, so a held session would lock the web UI
5
+
for humans), reads DOCSIS channel status and the status page, and logs out.
6
+
"""
7
+
8
+
import argparse
9
+
import hashlib
10
+
import json
11
+
import logging
12
+
import re
13
+
import threading
14
+
import time
15
+
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
16
+
17
+
import requests
18
+
from cryptography.hazmat.primitives.ciphers.aead import AESCCM
19
+
20
+
log = logging.getLogger("vodafone-station-exporter")
21
+
22
+
TIMEOUT = 15
23
+
24
+
25
+
class LoginError(Exception):
26
+
pass
27
+
28
+
29
+
def 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
+
36
+
def 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
+
41
+
class 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
+
100
+
def esc(v: str) -> str:
101
+
return str(v).replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")
102
+
103
+
104
+
class 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
+
122
+
def 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
+
127
+
def 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
+
178
+
def 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
+
241
+
def 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
+
284
+
if __name__ == "__main__":
285
+
main()