From b72b02f55b589b313cbb7b883db46224ef060492 Mon Sep 17 00:00:00 2001 From: agessaman Date: Fri, 7 Aug 2026 22:52:15 -0700 Subject: [PATCH] fix(webconfig): make the mock answer the whole CLI surface MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `get radio.fem.rxgain` returned "unknown config key" from the mock, which reads as the terminal offering a command that does not exist. It does exist: CommonCLI implements get and set for it, gated at runtime by Board::canControlLoRaFemLna() rather than compiled out, so the command is present in every build and the board answers for itself — "Error: unsupported" where there is no front-end module. Auditing the whole table found 31 of 70 config keys unanswered, all the ones no portal form drives: alert.*, bridge.*, owner.info, path.hash.mode, dutycycle and the rest. Plus 14 verbs (gps, powersaving, sensor, region, clock sync) with no handler at all. They now live in a "cli" section of the mock config, typed through the existing lookup tables and stripped from /api/config, which does not carry them. Two real bugs behind that: - the `set` path gated on whether a key was *readable*, so write-only and computed keys (prv.key, dutycycle, radio.fem.rxgain) were rejected as unknown. apply_set now owns that decision alone. - apply_set accepted anything it did not recognise and replied OK. That leniency is what let the gap hide: a CLI `set` on an unknown key looked like it worked. It is strict now — verified against every key in WC_ALLOWED_SET_KEYS so the form batch is unaffected. Also mqtt.neighbors / mqtt.neighbors.interval, which the MQTT tab binds but the mock's config never carried, so that toggle could not round-trip. webconfig_cli_audit.py keeps the two honest: it drives every command the autocomplete table offers through /api/cli and fails on anything unanswered. 119 commands, all answered. --- scripts/webconfig_cli_audit.py | 157 +++++++++++++++++++++++++++++++ scripts/webconfig_mock_server.py | 122 ++++++++++++++++++++++-- 2 files changed, 273 insertions(+), 6 deletions(-) create mode 100644 scripts/webconfig_cli_audit.py diff --git a/scripts/webconfig_cli_audit.py b/scripts/webconfig_cli_audit.py new file mode 100644 index 00000000..ca88ff6d --- /dev/null +++ b/scripts/webconfig_cli_audit.py @@ -0,0 +1,157 @@ +#!/usr/bin/env python3 +"""Check the portal terminal's command table against the mock backend. + +Autocomplete in webui/index.html carries its own list of commands. Nothing ties +that list to what a node actually answers, so it can quietly drift into offering +commands that do not exist — or, more often here, the mock can lag the table and +make a perfectly real command look broken. + +This drives every command the table offers through /api/cli and reports the ones +that come back an error, so the two stay honest about each other. + + python3 scripts/webconfig_mock_server.py --port 8137 & + python3 scripts/webconfig_cli_audit.py + +Exits non-zero if anything fails that is not in EXPECTED_FAILURES. Stdlib only. +""" + +import json +import os +import re +import secrets +import sys +import time +import urllib.error +import urllib.request + +BASE = os.environ.get("WEBCONFIG_MOCK", "http://localhost:8137") +HERE = os.path.dirname(os.path.abspath(__file__)) +INDEX_HTML = os.path.join(HERE, "..", "webui", "index.html") + +# Errors that are the correct answer, not a gap. +EXPECTED_FAILURES = { + # Runtime-gated on the real device by Board::canControlLoRaFemLna(); the + # command exists in every build and the board answers for itself. The mock + # board is a Heltec V3, which has no front-end module. + "get radio.fem.rxgain": "unsupported", + "set radio.fem.rxgain on": "unsupported", + # Guarded by the firmware the same way when no alert PSK is configured. + "alert test": "not configured", +} + +# Commands that change the node out from under the audit. +SKIP = {"reboot", "clkreboot", "poweroff", "shutdown", "erase", "start ota", + "stop webconfig", "ota update", "start webconfig", "start webconfig ap"} + + +def table(): + """The commands autocomplete offers, read straight out of the page.""" + html = open(INDEX_HTML, encoding="utf-8").read() + + def section(start, end): + return html[html.index(start):html.index(end)] + + verbs = re.findall(r'\["([^"]+)","', section("var CLI_VERBS=", "var CLI_KEYS=")) + keys = re.findall(r'\["([^"]+)","(?:[^"\\]|\\.)*",(\d)', + section("var CLI_KEYS=", "var CLI_SLOT=")) + fields = re.findall(r'\["(\w+)","', section("var CLI_SLOT=", "var CLI_TYPES=")) + + gets = ["get " + k for k, mode in keys if mode != "2"] + gets += ["get mqtt%d.%s" % (n, f) for n in (1, 3) for f in fields] + # Verbs taking an argument need a value the node will accept; those are + # covered by the round-trip probes below rather than guessed at here. + plain = [v for v in verbs if not v.endswith(" ") and v not in SKIP] + return gets + plain + + +# set -> get pairs, checking a value survives the round trip. +ROUND_TRIPS = [ + ("set radio.watchdog 30", "get radio.watchdog", "30"), + ("set dutycycle 25", "get dutycycle", "25.0"), + ("set alert.mqtt on", "get alert.mqtt", "on"), + ("set bridge.source tx", "get bridge.source", "tx"), + ("set mqtt.neighbors on", "get mqtt.neighbors", "on"), + ("set path.hash.mode 2", "get path.hash.mode", "2"), + ("set mqtt.iata den", "get mqtt.iata", "DEN"), + ("set guest.password hunter2", "get guest.password", "********"), +] + + +class Client: + def __init__(self, base): + self.base = base + r = self._open("/api/login", b'{"password":"password"}') + self.cookie = r.headers["Set-Cookie"].split(";")[0] + + def _open(self, path, data=None): + headers = {"Content-Type": "application/json"} + if getattr(self, "cookie", None): + headers["Cookie"] = self.cookie + return urllib.request.urlopen(urllib.request.Request( + self.base + path, data=data, headers=headers, + method="POST" if data is not None else "GET")) + + def run(self, cmds, chunk=40): + out = [] + for i in range(0, len(cmds), chunk): + out += self._sequence(cmds[i:i + chunk]) + return out + + def _sequence(self, cmds): + reqid = secrets.token_hex(8) + body = json.dumps({"reqid": reqid, "cmds": cmds}).encode() + for _ in range(200): # the executor frees itself in time + try: + self._open("/api/cli", body) + break + except urllib.error.HTTPError as e: + if e.code != 409: + raise + time.sleep(0.5) + while True: + r = json.load(self._open("/api/cli/result?reqid=" + reqid)) + if r["state"] == "done": + return r["results"] + time.sleep(0.05) + + +def main(): + try: + cli = Client(BASE) + except OSError as e: + sys.exit("cannot reach the mock at %s (%s)\n" + "start it with: python3 scripts/webconfig_mock_server.py --port 8137" % (BASE, e)) + + failures = [] + + cmds = table() + unexpected = [] + for res in cli.run(cmds): + if res["ok"]: + continue + want = EXPECTED_FAILURES.get(res["cmd"]) + if want and want in res["reply"]: + continue + unexpected.append((res["cmd"], res["reply"])) + print("commands offered by autocomplete : %d" % len(cmds)) + print("answered : %d" % (len(cmds) - len(unexpected))) + for cmd, reply in unexpected: + print(" FAIL %-30s %s" % (cmd, reply)) + failures += unexpected + + results = cli.run([c for probe in ROUND_TRIPS for c in probe[:2]]) + print("\nround-trips : %d" % len(ROUND_TRIPS)) + for i, (setc, getc, want) in enumerate(ROUND_TRIPS): + setr, getr = results[i * 2], results[i * 2 + 1] + if setr["ok"] and getr["reply"] == want: + continue + print(" FAIL %-30s got %r, wanted %r (set: %s)" + % (getc, getr["reply"], want, setr["reply"])) + failures.append((getc, getr["reply"])) + + print("\n%s" % ("FAILED: %d" % len(failures) if failures else "all clear")) + return 1 if failures else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/webconfig_mock_server.py b/scripts/webconfig_mock_server.py index e1b73d2e..4623f9e5 100644 --- a/scripts/webconfig_mock_server.py +++ b/scripts/webconfig_mock_server.py @@ -113,8 +113,23 @@ def default_config(setup_mode): "interval": 5, "timezone": "MST7MDT,M3.2.0,M11.1.0", "timezone_offset": -7, "ntp": "pool.ntp.org", "owner": "", "email": "", "snmp": False, "snmp_community": "public", + "neighbors": False, "neighbors_interval": 24, "slots": [_slot() for _ in range(6)], }, + # Settings the CLI reaches but no portal form does, so they are absent + # from /api/config (see config_json) and live only here. Without them + # the terminal answers "unknown config key" for perfectly real commands. + "cli": { + "radio.watchdog": 0, "int.thresh": 0, "agc.reset.interval": 0, + "direct.txdelay": 0.0, "multi.acks": 0, "allow.read.only": False, + "path.hash.mode": 0, "owner.info": "", "guest.password": "", + "adc.multiplier": 1.0, + "alert": False, "alert.psk": "", "alert.hashtag": "", + "alert.region": "", "alert.interval": 15, + "alert.mqtt": False, "alert.wifi": False, + "bridge.enabled": False, "bridge.source": "rx", "bridge.baud": 115200, + "bridge.delay": 0, "bridge.channel": 0, "bridge.secret": "", + }, } @@ -151,6 +166,7 @@ class State: # ---- config serialization (masks secrets, like handleConfigGet) ------- def config_json(self): c = copy.deepcopy(self.cfg) + c.pop("cli") # CLI-only settings: not part of this contract c["wifi"]["pwd"] = SENTINEL if self.cfg["wifi"]["pwd"] else "" for s in c["mqtt"]["slots"]: s["password"] = SENTINEL if s["password"] else "" @@ -176,13 +192,15 @@ class State: BOOL_KEYS = {"cad": ("radio", "cad"), "radio.rxgain": ("radio", "rxgain"), "repeat": ("radio", "repeat"), "mqtt.status": ("mqtt", "status"), "mqtt.packets": ("mqtt", "packets"), "mqtt.raw": ("mqtt", "raw"), - "mqtt.rx": ("mqtt", "rx"), "snmp": ("mqtt", "snmp")} + "mqtt.rx": ("mqtt", "rx"), "snmp": ("mqtt", "snmp"), + "mqtt.neighbors": ("mqtt", "neighbors")} INT_KEYS = {"tx": ("radio", "tx"), "flood.max": ("radio", "flood_max"), "flood.max.advert": ("radio", "flood_max_advert"), "flood.max.unscoped": ("radio", "flood_max_unscoped"), "advert.interval": ("radio", "advert_interval"), "flood.advert.interval": ("radio", "flood_advert_interval"), "mqtt.interval": ("mqtt", "interval"), + "mqtt.neighbors.interval": ("mqtt", "neighbors_interval"), "timezone.offset": ("mqtt", "timezone_offset")} FLOAT_KEYS = {"lat": ("radio", "lat"), "lon": ("radio", "lon"), "af": ("radio", "af"), "rxdelay": ("radio", "rxdelay"), @@ -194,6 +212,16 @@ STR_KEYS = {"name": ("radio", "name"), "wifi.ssid": ("wifi", "ssid"), "snmp.community": ("mqtt", "snmp_community"), "mqtt.tx": ("mqtt", "tx")} SECRET_STR_KEYS = {"wifi.pwd": ("wifi", "pwd")} +# The CLI-only settings, typed the same way so apply_set/cli_read_key reach them +# through the existing lookups rather than a parallel code path. +for _k, _v in default_config(False)["cli"].items(): + _t = {bool: BOOL_KEYS, int: INT_KEYS, float: FLOAT_KEYS, str: STR_KEYS}[type(_v)] + _t[_k] = ("cli", _k) +SECRET_STR_KEYS.update({k: ("cli", k) for k in + ("guest.password", "alert.psk", "bridge.secret")}) +for _k in SECRET_STR_KEYS: + STR_KEYS.pop(_k, None) + def _hex64(v): return len(v) == 64 and all(c in "0123456789abcdefABCDEF" for c in v) @@ -214,6 +242,19 @@ def apply_set(cfg, key, val): ADMIN_PASSWORD = val return True, "OK" + if key == "radio.fem.rxgain": + return False, "Error: unsupported" # no FEM on the mock board, see GETTERS + + if key == "dutycycle": + try: + dc = float(val) + except ValueError: + return False, "Error: expected a number" + if not 0 < dc <= 100: + return False, "Error, must be 1-100" + cfg["radio"]["af"] = 100.0 / dc - 1 # the CLI stores it as airtime_factor + return True, "OK" + if key in ("freq", "bw", "sf", "cr"): # single-component radio setters, reachable from the CLI but not from # the form batch (which always sends the whole `radio` combo) @@ -243,6 +284,12 @@ def apply_set(cfg, key, val): cfg["mqtt"]["iata"] = val.upper() return True, "OK" + if key == "prv.key": + # write-only by design: the identity goes in, nothing reads it back + if not _hex64(val): + return False, "Error: private key must be 64 hex characters" + return True, "OK - identity restored, reboot to apply" + if key == "mqtt.owner": if val == "": cfg["mqtt"]["owner"] = "" @@ -282,7 +329,10 @@ def apply_set(cfg, key, val): sec, f = STR_KEYS[key] cfg[sec][f] = val return True, "OK" - return True, "OK" # unknown-but-allowlisted: accept (mock is lenient here) + # Strict fallthrough: this function is the single authority on what can be + # set, for the batch and the CLI alike. Accepting unknown keys here once hid + # the fact that the CLI could not reach `dutycycle` or `radio.fem.rxgain`. + return False, "Error: unknown config key '%s'" % key # Payload-type names accepted alongside the decimal form. Mirrors @@ -360,7 +410,9 @@ def apply_slot_set(cfg, idx, field, val): def is_secret_key(key): - return key == "wifi.pwd" or bool(re.match(r"^mqtt[1-6]\.(password|token)$", key)) + # The serial console prints these back; the portal is reachable over the + # LAN, so it masks them in `get` replies the way /api/config already does. + return key in SECRET_STR_KEYS or bool(re.match(r"^mqtt[1-6]\.(password|token)$", key)) # --------------------------------------------------------------------------- @@ -389,6 +441,18 @@ GETTERS = { "mqtt.presets": lambda c: "\n".join( "%2d. %s%s" % (i + 1, n, "" if nd == "none" else " (needs %s)" % nd) for i, (n, nd) in enumerate(PRESETS)), + "role": lambda c: "Repeater", + # not its own pref: the CLI derives it from airtime_factor both ways + "dutycycle": lambda c: "%.1f" % (100.0 / (c["radio"]["af"] + 1)), + "mqtt.config.valid": lambda c: ( + "yes" if any(s["preset"] != "none" for s in c["mqtt"]["slots"]) else "no - no slot configured"), + "mqtt.ntp.diag": lambda c: "last sync: 42s ago via %s (offset +0.011s)" % (c["mqtt"]["ntp"] or "none"), + "mqtt.stats": lambda c: ("published: %d\ndropped: 0\nqueue: 0/24\nreconnects: 1" + % (100 + int(time.time() - ST.start))), + # Runtime-gated on the real device (Board::canControlLoRaFemLna), not + # compiled out — the command exists everywhere and the board answers for + # itself. The mock board is a Heltec V3, which has no FEM. + "radio.fem.rxgain": lambda c: None, } @@ -405,7 +469,8 @@ def cli_mqtt_status(cfg): def cli_get(cfg, key): if key in GETTERS: - return True, GETTERS[key](cfg) + val = GETTERS[key](cfg) + return (True, val) if val is not None else (False, "Error: unsupported") if is_secret_key(key): # The serial console prints these; the portal is reachable over the LAN, # so it masks them the same way /api/config does. @@ -464,6 +529,50 @@ def run_cli(cfg, line): if cmd == "neighbors": return True, ("d4e5f60718 -71 dBm snr 9.5 2m ago\n" "1122334455 -94 dBm snr 2.0 14m ago") + if cmd == "clock sync": + return True, "OK - clock set: %s UTC" % time.strftime("%H:%M - %d/%m/%Y", time.gmtime()) + if cmd == "region": + return True, "US915" + if cmd == "sensor list": + return True, "0: battery (mV)\n1: temperature (C)\n2: humidity (%)" + if cmd.startswith("sensor get "): + return True, "> 22.4" + if cmd.startswith("sensor set "): + return True, "OK" + if cmd.startswith("gps advert "): + mode = cmd[11:] + if mode not in ("none", "share", "prefs"): + return False, "Error, must be none, share or prefs" + return True, "OK - advert position: %s" % mode + if cmd in ("gps on", "gps off"): + return True, "OK - GPS %s" % cmd[4:] + if cmd == "gps sync": + return True, "OK - clock and location set from GPS" + if cmd == "gps setloc": + return True, "OK - lat/lon set from the current fix" + if cmd == "gps": + return True, "GPS: no fix (0 satellites)" + if cmd in ("powersaving on", "powersaving off"): + return True, "OK - power saving %s" % cmd[12:] + if cmd == "powersaving": + return True, "off" + if cmd.startswith("alert test"): + if not ST.cfg["cli"]["alert.psk"]: + return False, "Error: alert channel not configured (set alert.psk or set alert.hashtag)" + return True, "OK - test alert sent" + if cmd.startswith("ota "): + return True, ("v1.7.2 available (current v1.7.1-mock)" if cmd == "ota check" + else "OK - downloading v1.7.2, will reboot when flashed") + if cmd.startswith("start webconfig"): + return True, "OK - already running (you are using it)" + if cmd == "stop webconfig": + return True, "OK - portal stopping" + if cmd == "start ota": + return True, "OK - upload AP raised at 192.168.4.1" + if cmd.startswith("neighbor.remove "): + return (True, "OK") if _hex64(cmd[16:]) else (False, "ERR: bad pubkey") + if cmd.startswith("tempradio "): + return True, "OK - temporary radio params applied (not saved)" if cmd == "clear stats": return True, "OK - stats cleared" if cmd.startswith("stats-"): @@ -485,8 +594,9 @@ def run_cli(cfg, line): key, _, val = rest.partition(" ") if not key: return False, "Error: set what?" - if cli_read_key(cfg, key) is None and not re.match(r"^mqtt[1-6]\.", key): - return False, "Error: unknown config key '%s'" % key + # apply_set owns the "is this settable" decision; gating on whether the + # key is *readable* rejected write-only and computed ones (`dutycycle`, + # `prv.key`, `radio.fem.rxgain`). return apply_set(cfg, key, val.strip()) return False, "Error: unknown command '%s'" % cmd[:40]