mirror of
https://github.com/agessaman/MeshCore.git
synced 2026-08-29 03:18:30 +00:00
agessaman
dc44311e49
Add validation for request IDs in the web configuration server to ensure they conform to the expected format. Enhance error responses for invalid or unknown request IDs, improving the robustness of request handling. Update the web UI to reflect these changes, ensuring that clients can properly handle errors related to request ID mismatches.
489 lines
20 KiB
Python
489 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""Local mock of the WebConfig portal backend, for iterating on webui/index.html
|
|
in a real browser with no firmware, no flashing, and no paid emulator account.
|
|
|
|
It serves the real webui/index.html and implements the same /api/* contract as
|
|
src/helpers/esp32/WebConfigServer.cpp — including the 202+reqid handshake, the
|
|
pending -> done result polling, aggregate-success reboot gating, secret masking
|
|
(********), and the IATA / owner-key / length validation the firmware enforces.
|
|
So the browser drives the actual portal JS (wizard, save/poll/reqid, effective
|
|
value handling, reboot overlay, stats, scan) against realistic responses.
|
|
|
|
It does NOT run the C++ handlers (that's what test/ gtest covers) or the
|
|
AsyncTCP transport — it's a frontend + contract harness.
|
|
|
|
Usage:
|
|
python3 scripts/webconfig_mock_server.py # LAN mode (login: password)
|
|
python3 scripts/webconfig_mock_server.py --setup # first-boot setup wizard
|
|
python3 scripts/webconfig_mock_server.py --port 9000 --active-slots 2
|
|
Then open http://localhost:8080/ (or the chosen port). Editing index.html and
|
|
refreshing shows changes immediately — the page is re-read per request.
|
|
|
|
Stdlib only; no pip install.
|
|
"""
|
|
|
|
import argparse
|
|
import copy
|
|
import json
|
|
import os
|
|
import re
|
|
import secrets
|
|
import threading
|
|
import time
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from urllib.parse import parse_qs, urlsplit
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
INDEX_HTML = os.path.join(HERE, "..", "webui", "index.html")
|
|
|
|
SENTINEL = "********"
|
|
ADMIN_PASSWORD = "password" # matches the default ADMIN_PASSWORD build flag
|
|
BATCH_PENDING_SECS = 0.8 # how long POST->done takes, to exercise polling
|
|
SCAN_SECS = 0.8
|
|
|
|
# Destination buffer sizes (chars, minus the NUL) — mirrors the MQTTPrefs fields
|
|
# the firmware validates in CommonCLI_Observer.cpp.
|
|
LEN_LIMITS = {
|
|
"name": 31, "wifi.ssid": 31, "wifi.pwd": 63, "mqtt.origin": 31,
|
|
"mqtt.email": 63, "mqtt.ntp": 63, "timezone": 31, "snmp.community": 23,
|
|
}
|
|
SLOT_LEN_LIMITS = {"server": 63, "username": 31, "password": 63,
|
|
"token": 47, "topic": 95, "audience": 63}
|
|
|
|
# Preset names + what the UI must collect (mirrors handlePresets()).
|
|
PRESETS = (
|
|
[(n, "none") for n in (
|
|
"analyzer-us", "analyzer-eu", "nz-analyzer", "meshmapper", "waev",
|
|
"meshomatic", "cascadiamesh", "tennmesh", "nashmesh", "ctmesh", "chimesh",
|
|
"meshat.se", "eastidahomesh", "coloradomesh", "dutchmeshcore-1",
|
|
"dutchmeshcore-2", "meshcore-ca-1", "meshcore-ca-2", "meshcore-fi",
|
|
"bostonmesh", "rflab", "ipnt.uk", "flmesh", "corecomms")]
|
|
+ [("meshrank", "token"), ("inwmesh", "userpass")]
|
|
)
|
|
|
|
SCAN_NETWORKS = [
|
|
{"ssid": "Wokwi-GUEST", "rssi": -42, "enc": False},
|
|
{"ssid": "HomeNet", "rssi": -55, "enc": True},
|
|
{"ssid": "HomeNet-5G", "rssi": -61, "enc": True},
|
|
{"ssid": "Neighbor 2.4", "rssi": -78, "enc": True},
|
|
{"ssid": "OpenGuest", "rssi": -83, "enc": False},
|
|
]
|
|
|
|
|
|
def default_config(setup_mode):
|
|
return {
|
|
"radio": {
|
|
"freq": 910.525, "bw": 62.5, "sf": 7, "cr": 5, "tx": 22, "af": 1.0,
|
|
"rxdelay": 0.0, "txdelay": 0.5, "cad": False, "rxgain": True,
|
|
"repeat": True, "flood_max": 64, "flood_max_advert": 8,
|
|
"flood_max_unscoped": 8, "loop_detect": "moderate",
|
|
"name": "MockNode", "lat": 39.7392, "lon": -104.9903,
|
|
"advert_interval": 240, "flood_advert_interval": 6,
|
|
},
|
|
"wifi": {
|
|
# setup mode = unconfigured (empty ssid -> wizard); LAN mode = joined
|
|
"ssid": "" if setup_mode else "HomeNet",
|
|
"pwd": "" if setup_mode else "secretpw", # stored raw; masked on GET
|
|
"powersave": "min",
|
|
},
|
|
"mqtt": {
|
|
"origin": "" if setup_mode else "MockNode", "iata": "" if setup_mode else "DEN",
|
|
"status": True, "packets": True, "raw": False, "tx": "advert", "rx": True,
|
|
"interval": 5, "timezone": "MST7MDT,M3.2.0,M11.1.0", "timezone_offset": -7,
|
|
"ntp": "pool.ntp.org", "owner": "", "email": "", "snmp": False,
|
|
"snmp_community": "public",
|
|
"slots": [_slot() for _ in range(6)],
|
|
},
|
|
}
|
|
|
|
|
|
def _slot():
|
|
return {"preset": "none", "server": "", "port": 8883, "username": "",
|
|
"password": "", "token": "", "topic": "", "audience": ""}
|
|
|
|
|
|
class State:
|
|
def __init__(self, args):
|
|
self.lock = threading.Lock()
|
|
self.setup_mode = args.setup
|
|
self.active_slots = args.active_slots
|
|
self.cfg = default_config(args.setup)
|
|
self.start = time.time()
|
|
self.session = None # cookie token when logged in (LAN mode)
|
|
self.batch = {"state": "idle"}
|
|
self.scan_started = None
|
|
|
|
# ---- auth -------------------------------------------------------------
|
|
def is_authed(self, headers):
|
|
if self.setup_mode:
|
|
return True # setup mode: proximity trust, no auth
|
|
if not self.session:
|
|
return False
|
|
cookie = headers.get("Cookie", "")
|
|
m = re.search(r"wcs=([0-9a-f]+)", cookie)
|
|
return bool(m and m.group(1) == self.session)
|
|
|
|
# ---- config serialization (masks secrets, like handleConfigGet) -------
|
|
def config_json(self):
|
|
c = copy.deepcopy(self.cfg)
|
|
c["wifi"]["pwd"] = SENTINEL if self.cfg["wifi"]["pwd"] else ""
|
|
for s in c["mqtt"]["slots"]:
|
|
s["password"] = SENTINEL if s["password"] else ""
|
|
s["token"] = SENTINEL if s["token"] else ""
|
|
return c
|
|
|
|
def status_json(self, authed):
|
|
return {
|
|
"mode": "setup" if self.setup_mode else "lan",
|
|
"auth": authed,
|
|
"needs_setup": self.cfg["wifi"]["ssid"] == "",
|
|
"name": self.cfg["radio"]["name"], "node_id": "a1b2c3d4e5f60718",
|
|
"fw": "v1.7.1-mock", "role": "Repeater", "board": "Heltec V3 (mock)",
|
|
"uptime_s": int(time.time() - self.start),
|
|
"runtime_slots": 6, "max_slots": 6, "active_slots": self.active_slots,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# set-command application + validation (mirrors the firmware's setters enough
|
|
# to produce realistic per-field OK / Error replies for the UI chips).
|
|
# ---------------------------------------------------------------------------
|
|
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")}
|
|
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"),
|
|
"timezone.offset": ("mqtt", "timezone_offset")}
|
|
FLOAT_KEYS = {"lat": ("radio", "lat"), "lon": ("radio", "lon"),
|
|
"af": ("radio", "af"), "rxdelay": ("radio", "rxdelay"),
|
|
"txdelay": ("radio", "txdelay")}
|
|
STR_KEYS = {"name": ("radio", "name"), "wifi.ssid": ("wifi", "ssid"),
|
|
"wifi.powersave": ("wifi", "powersave"), "loop.detect": ("radio", "loop_detect"),
|
|
"mqtt.origin": ("mqtt", "origin"), "mqtt.ntp": ("mqtt", "ntp"),
|
|
"mqtt.email": ("mqtt", "email"), "timezone": ("mqtt", "timezone"),
|
|
"snmp.community": ("mqtt", "snmp_community"), "mqtt.tx": ("mqtt", "tx")}
|
|
SECRET_STR_KEYS = {"wifi.pwd": ("wifi", "pwd")}
|
|
|
|
|
|
def _hex64(v):
|
|
return len(v) == 64 and all(c in "0123456789abcdefABCDEF" for c in v)
|
|
|
|
|
|
def apply_set(cfg, key, val):
|
|
"""Return (ok, reply) and mutate cfg. Mirrors the firmware's validation for
|
|
the fields where it matters (length, IATA, owner key, port, radio combo)."""
|
|
# length guard for the plain string fields
|
|
if key in LEN_LIMITS and len(val) > LEN_LIMITS[key]:
|
|
return False, "Error: %s too long (max %d chars)" % (key, LEN_LIMITS[key])
|
|
|
|
if key == "radio":
|
|
try:
|
|
f, bw, sf, cr = val.split(",")
|
|
f, bw, sf, cr = float(f), float(bw), int(sf), int(cr)
|
|
except ValueError:
|
|
return False, "Error, invalid radio params"
|
|
if not (150 <= f <= 2500 and 7 <= bw <= 500 and 5 <= sf <= 12 and 5 <= cr <= 8):
|
|
return False, "Error, invalid radio params"
|
|
cfg["radio"].update(freq=f, bw=bw, sf=sf, cr=cr)
|
|
return True, "OK - reboot to apply"
|
|
|
|
if key == "mqtt.iata":
|
|
if val == "":
|
|
cfg["mqtt"]["iata"] = ""
|
|
return True, "OK - IATA cleared"
|
|
if len(val) != 3 or not val.isalnum() or not val.isascii():
|
|
return False, "Error: IATA code must be exactly 3 letters/digits (e.g. DEN)"
|
|
cfg["mqtt"]["iata"] = val.upper()
|
|
return True, "OK"
|
|
|
|
if key == "mqtt.owner":
|
|
if val == "":
|
|
cfg["mqtt"]["owner"] = ""
|
|
return True, "OK - owner key cleared"
|
|
if not _hex64(val):
|
|
return False, "Error: public key must be 64 hex characters (32 bytes)"
|
|
cfg["mqtt"]["owner"] = val
|
|
return True, "OK"
|
|
|
|
m = re.match(r"^mqtt([1-6])\.(\w+)$", key)
|
|
if m:
|
|
return apply_slot_set(cfg, int(m.group(1)) - 1, m.group(2), val)
|
|
|
|
if key in BOOL_KEYS:
|
|
sec, f = BOOL_KEYS[key]
|
|
cfg[sec][f] = (val == "on")
|
|
return True, "OK"
|
|
if key in INT_KEYS:
|
|
sec, f = INT_KEYS[key]
|
|
try:
|
|
cfg[sec][f] = int(val)
|
|
except ValueError:
|
|
return False, "Error: expected a number"
|
|
return True, "OK"
|
|
if key in FLOAT_KEYS:
|
|
sec, f = FLOAT_KEYS[key]
|
|
try:
|
|
cfg[sec][f] = float(val)
|
|
except ValueError:
|
|
return False, "Error: expected a number"
|
|
return True, "OK"
|
|
if key in SECRET_STR_KEYS:
|
|
sec, f = SECRET_STR_KEYS[key]
|
|
cfg[sec][f] = val
|
|
return True, "OK"
|
|
if key in STR_KEYS:
|
|
sec, f = STR_KEYS[key]
|
|
cfg[sec][f] = val
|
|
return True, "OK"
|
|
return True, "OK" # unknown-but-allowlisted: accept (mock is lenient here)
|
|
|
|
|
|
def apply_slot_set(cfg, idx, field, val):
|
|
slot = cfg["mqtt"]["slots"][idx]
|
|
if field in SLOT_LEN_LIMITS and len(val) > SLOT_LEN_LIMITS[field]:
|
|
return False, "Error: %s too long (max %d chars)" % (field, SLOT_LEN_LIMITS[field])
|
|
if field == "port":
|
|
try:
|
|
p = int(val)
|
|
except ValueError:
|
|
return False, "Error: port must be between 1 and 65535"
|
|
if not (1 <= p <= 65535):
|
|
return False, "Error: port must be between 1 and 65535"
|
|
slot["port"] = p
|
|
return True, "OK"
|
|
if field in ("preset", "server", "username", "password", "token", "topic", "audience"):
|
|
slot[field] = val
|
|
if field == "token":
|
|
return True, "OK - slot %d token set" % (idx + 1)
|
|
return True, "OK"
|
|
return False, "Error: unknown slot field"
|
|
|
|
|
|
def is_secret_key(key):
|
|
return key == "wifi.pwd" or bool(re.match(r"^mqtt[1-6]\.(password|token)$", key))
|
|
|
|
|
|
def valid_reqid(reqid):
|
|
return isinstance(reqid, str) and bool(re.fullmatch(r"[0-9A-Fa-f]{16}", reqid))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# HTTP handler
|
|
# ---------------------------------------------------------------------------
|
|
class Handler(BaseHTTPRequestHandler):
|
|
protocol_version = "HTTP/1.1"
|
|
|
|
def log_message(self, fmt, *args): # concise one-line log
|
|
print(" %s %s" % (self.command, self.path))
|
|
|
|
# -- helpers --
|
|
def _json(self, code, obj, extra_headers=None):
|
|
body = json.dumps(obj).encode()
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.send_header("Cache-Control", "no-store")
|
|
for k, v in (extra_headers or {}):
|
|
self.send_header(k, v)
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def _read_body(self):
|
|
n = int(self.headers.get("Content-Length", 0))
|
|
return self.rfile.read(n) if n else b""
|
|
|
|
def _need_auth(self):
|
|
if not ST.is_authed(self.headers):
|
|
self._json(401, {"error": "auth"})
|
|
return True
|
|
return False
|
|
|
|
# -- GET --
|
|
def do_GET(self):
|
|
path = self.path.split("?", 1)[0]
|
|
if path == "/":
|
|
return self._serve_index()
|
|
if path == "/api/status":
|
|
return self._json(200, ST.status_json(ST.is_authed(self.headers)))
|
|
if path == "/api/presets":
|
|
return self._json(200, {"presets": [{"name": n, "needs": nd} for n, nd in PRESETS]})
|
|
if path == "/api/config":
|
|
if self._need_auth():
|
|
return
|
|
with ST.lock:
|
|
return self._json(200, ST.config_json())
|
|
if path == "/api/config/result":
|
|
if self._need_auth():
|
|
return
|
|
return self._config_result()
|
|
if path == "/api/stats":
|
|
if self._need_auth():
|
|
return
|
|
return self._json(200, self._stats())
|
|
if path == "/api/scan":
|
|
if self._need_auth():
|
|
return
|
|
return self._scan()
|
|
return self._json(404, {"error": "not found"})
|
|
|
|
# -- POST --
|
|
def do_POST(self):
|
|
path = self.path.split("?", 1)[0]
|
|
if path == "/api/login":
|
|
return self._login()
|
|
if path == "/api/logout":
|
|
ST.session = None
|
|
return self._json(200, {"ok": True}, [("Set-Cookie", "wcs=; Max-Age=0; Path=/")])
|
|
if path == "/api/config":
|
|
if self._need_auth():
|
|
return
|
|
return self._config_post()
|
|
if path == "/api/reboot":
|
|
if self._need_auth():
|
|
return
|
|
return self._json(200, {"ok": True})
|
|
if path == "/api/portal/exit":
|
|
return self._json(200, {"ok": True, "url": "http://localhost:%d/" % PORT})
|
|
return self._json(404, {"error": "not found"})
|
|
|
|
# -- endpoint impls --
|
|
def _serve_index(self):
|
|
try:
|
|
with open(INDEX_HTML, "rb") as f: # re-read each time -> live edits
|
|
html = f.read()
|
|
except OSError:
|
|
self.send_error(500, "webui/index.html not found")
|
|
return
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/html; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(html)))
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.end_headers()
|
|
self.wfile.write(html)
|
|
|
|
def _login(self):
|
|
if ST.setup_mode:
|
|
return self._json(200, {"ok": True})
|
|
try:
|
|
body = json.loads(self._read_body() or b"{}")
|
|
except ValueError:
|
|
return self._json(400, {"error": "bad request"})
|
|
if body.get("password") != ADMIN_PASSWORD:
|
|
return self._json(401, {"error": "wrong password"})
|
|
ST.session = secrets.token_hex(16)
|
|
return self._json(200, {"ok": True},
|
|
[("Set-Cookie", "wcs=%s; HttpOnly; SameSite=Lax; Path=/" % ST.session)])
|
|
|
|
def _config_post(self):
|
|
raw = self._read_body()
|
|
if len(raw) > 4096:
|
|
return self._json(413, {"error": "body too large"})
|
|
try:
|
|
body = json.loads(raw or b"{}")
|
|
except ValueError:
|
|
return self._json(400, {"error": "bad json"})
|
|
reqid = body.get("reqid", "")
|
|
if not valid_reqid(reqid):
|
|
return self._json(400, {"error": "bad reqid"})
|
|
reboot = bool(body.get("reboot", False))
|
|
setmap = body.get("set", {}) or {}
|
|
|
|
with ST.lock:
|
|
if ST.batch.get("state") != "idle" and ST.batch.get("reqid") == reqid:
|
|
return self._json(202, {
|
|
"state": ST.batch["state"], "count": len(ST.batch.get("results", [])),
|
|
"reqid": reqid,
|
|
})
|
|
if ST.batch.get("state") == "pending":
|
|
return self._json(409, {"error": "busy", "reqid": ST.batch.get("reqid", "")})
|
|
# drop unchanged secrets (sentinel), like the firmware does
|
|
entries = [(k, v) for k, v in setmap.items()
|
|
if not (is_secret_key(k) and v == SENTINEL)]
|
|
if not entries and not reboot:
|
|
return self._json(400, {"error": "no changes"})
|
|
# apply now, but expose as pending->done to exercise polling
|
|
results, all_ok = [], True
|
|
for k, v in entries:
|
|
ok, reply = apply_set(ST.cfg, k, str(v))
|
|
if not ok:
|
|
all_ok = False
|
|
results.append({"key": k, "reply": reply})
|
|
ST.batch = {"state": "pending", "reqid": reqid, "results": results,
|
|
"all_ok": all_ok, "reboot": reboot,
|
|
"done_at": time.time() + BATCH_PENDING_SECS}
|
|
return self._json(202, {"state": "pending", "count": len(entries), "reqid": reqid})
|
|
|
|
def _config_result(self):
|
|
query = parse_qs(urlsplit(self.path).query)
|
|
reqid = query.get("reqid", [""])[0]
|
|
if not valid_reqid(reqid):
|
|
return self._json(400, {"error": "bad reqid"})
|
|
with ST.lock:
|
|
b = ST.batch
|
|
if b.get("state") == "idle":
|
|
return self._json(200, {"state": "idle", "reqid": reqid})
|
|
if b.get("reqid") != reqid:
|
|
return self._json(404, {"error": "unknown request"})
|
|
if b["state"] == "pending" and time.time() < b["done_at"]:
|
|
return self._json(200, {"state": "pending", "reqid": b["reqid"]})
|
|
b["state"] = "done" # stays readable until next POST
|
|
return self._json(200, {
|
|
"state": "done", "reqid": b["reqid"], "all_ok": b["all_ok"],
|
|
"reboot": b["reboot"] and b["all_ok"], "results": b["results"],
|
|
})
|
|
|
|
def _scan(self):
|
|
rescan = "rescan=1" in self.path
|
|
now = time.time()
|
|
if rescan or ST.scan_started is None:
|
|
ST.scan_started = now
|
|
return self._json(200, {"state": "scanning"})
|
|
if now - ST.scan_started < SCAN_SECS:
|
|
return self._json(200, {"state": "scanning"})
|
|
return self._json(200, {"state": "done", "networks": SCAN_NETWORKS})
|
|
|
|
def _stats(self):
|
|
up = int(time.time() - ST.start)
|
|
slots = []
|
|
for i, s in enumerate(ST.cfg["mqtt"]["slots"]):
|
|
if s["preset"] == "none":
|
|
continue
|
|
slots.append({"n": i + 1, "name": s["preset"], "state": "ok",
|
|
"ok": 100 + up, "err": 0})
|
|
return {
|
|
"uptime_s": up, "batt_mv": 4020, "heap_free": 142000, "heap_min": 118000,
|
|
"heap_max_alloc": 96000, "noise": -98, "rssi": -71, "snr": 9.5,
|
|
"airtime_s": up // 20, "rx_airtime_s": up // 8, "recv": 512 + up,
|
|
"sent": 88 + up // 3, "rx_err": 3, "sent_flood": 40, "sent_direct": 48,
|
|
"recv_flood": 300, "recv_direct": 212, "tx_queue": 0, "mqtt_queue": 0,
|
|
"wifi_rssi": -58, "ip": "192.168.1.42", "slots": slots,
|
|
}
|
|
|
|
|
|
def main():
|
|
global ST, PORT
|
|
ap = argparse.ArgumentParser(description="Mock WebConfig portal backend")
|
|
ap.add_argument("--port", type=int, default=8080)
|
|
ap.add_argument("--setup", action="store_true", help="first-boot setup wizard mode")
|
|
ap.add_argument("--active-slots", type=int, default=5, help="server slots to expose (2 or 5)")
|
|
args = ap.parse_args()
|
|
ST, PORT = State(args), args.port
|
|
|
|
srv = ThreadingHTTPServer(("127.0.0.1", args.port), Handler)
|
|
mode = "SETUP (wizard)" if args.setup else "LAN (login: %s)" % ADMIN_PASSWORD
|
|
print("WebConfig mock backend — %s" % mode)
|
|
print(" open http://localhost:%d/ (Ctrl-C to stop)" % args.port)
|
|
try:
|
|
srv.serve_forever()
|
|
except KeyboardInterrupt:
|
|
print("\nstopped")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|