Merge observer firmware and webconfig updates
# Conflicts: # MQTT_INTERNALS.md # build.sh # docs/cli_commands.md # examples/simple_repeater/MyMesh.cpp # examples/simple_repeater/MyMesh.h # examples/simple_repeater/UITask.cpp # examples/simple_room_server/MyMesh.cpp # examples/simple_room_server/MyMesh.h # examples/simple_room_server/UITask.cpp # examples/simple_sensor/SensorMesh.cpp # platformio.ini # scripts/generate_webconfig_html.py # src/MeshCore.h # src/helpers/CommonCLI.cpp # src/helpers/CommonCLI.h # src/helpers/CommonCLI_Observer.cpp # src/helpers/ESP32Board.h # src/helpers/MQTTMessageBuilder.cpp # src/helpers/MQTTPresets.h # src/helpers/NRF52Board.cpp # src/helpers/NRF52Board.h # src/helpers/bridges/MQTTBridge.cpp # src/helpers/bridges/MQTTBridge.h # src/helpers/esp32/WebConfigServer.cpp # src/helpers/esp32/WebConfigServer.h # src/helpers/radiolib/RadioLibWrappers.cpp # src/helpers/radiolib/RadioLibWrappers.h # variants/lilygo_tbeam_SX1262/platformio.ini # variants/lilygo_tbeam_SX1276/platformio.ini # variants/lilygo_tlora_v2_1/platformio.ini # webui/index.html
This commit is contained in:
mikecarper committed
commit
c1caa5ad81
109 files changed
+14844
-2297
No files matched your search
@@ -0,0 +1,162 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Require ArduinoJson 7.4.3 in checked-in PlatformIO configurations.
|
||||
|
||||
The checker scans the root ``platformio.ini`` plus every
|
||||
``variants/**/platformio.ini``. It checks every explicit ArduinoJson
|
||||
declaration and, without a fragile fixed count, requires one in the native
|
||||
test environment and each MQTT bridge environment.
|
||||
|
||||
Run ``python3 scripts/check_arduinojson_pin.py`` for the repository and add
|
||||
``--self-test`` to exercise accepted, unpinned, wrong, and missing pins in a
|
||||
temporary fixture.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
import re
|
||||
import sys
|
||||
from tempfile import TemporaryDirectory
|
||||
|
||||
|
||||
VERSION = "7.4.3"
|
||||
PACKAGE = "bblanchon/ArduinoJson"
|
||||
PACKAGE_PATTERN = re.compile(r"\bbblanchon\s*/\s*ArduinoJson\b", re.IGNORECASE)
|
||||
PIN_PATTERN = re.compile(r"@\s*7\.4\.3\Z")
|
||||
SECTION_PATTERN = re.compile(r"^\s*\[([^]]+)\]\s*$")
|
||||
|
||||
|
||||
def active(line):
|
||||
"""Remove PlatformIO whole-line and inline comments."""
|
||||
|
||||
if line.lstrip().startswith((";", "#")):
|
||||
return ""
|
||||
comments = [index for index in (line.find(";"), line.find("#")) if index >= 0]
|
||||
return line[: min(comments)] if comments else line
|
||||
|
||||
|
||||
def configuration_paths(root):
|
||||
root_config = root / "platformio.ini"
|
||||
if not root_config.is_file():
|
||||
raise FileNotFoundError(f"root PlatformIO configuration is missing: {root_config}")
|
||||
variants = root / "variants"
|
||||
return [root_config] + sorted(variants.rglob("platformio.ini") if variants.is_dir() else [])
|
||||
|
||||
|
||||
def sections(lines):
|
||||
"""Yield ``(name, header_line, lines)`` while preserving source locations."""
|
||||
|
||||
name, header_line, content = "<global>", 1, []
|
||||
for number, line in enumerate(lines, 1):
|
||||
match = SECTION_PATTERN.fullmatch(active(line).strip())
|
||||
if match:
|
||||
yield name, header_line, content
|
||||
name, header_line, content = match.group(1), number, []
|
||||
else:
|
||||
content.append((number, line))
|
||||
yield name, header_line, content
|
||||
|
||||
|
||||
def expects_pin(config, root, section, lines):
|
||||
if config == root / "platformio.ini":
|
||||
return section == "env:native"
|
||||
if not section.startswith("env:"):
|
||||
return False
|
||||
text = "\n".join(active(line) for _, line in lines)
|
||||
return "WITH_MQTT_BRIDGE" in text or "elims/PsychicMqttClient" in text
|
||||
|
||||
|
||||
def check(root):
|
||||
"""Return (configs, declarations, required environments, error messages)."""
|
||||
|
||||
root = root.resolve()
|
||||
configs, declarations, required, errors = configuration_paths(root), 0, 0, []
|
||||
for config in configs:
|
||||
relative = config.relative_to(root).as_posix()
|
||||
for section, header_line, lines in sections(config.read_text(encoding="utf-8").splitlines()):
|
||||
found = False
|
||||
for number, raw_line in lines:
|
||||
declaration = active(raw_line)
|
||||
match = PACKAGE_PATTERN.search(declaration)
|
||||
if not match:
|
||||
continue
|
||||
found, declarations = True, declarations + 1
|
||||
version = declaration[match.end() :].strip()
|
||||
if not version:
|
||||
errors.append(f"{relative}:{number}: {PACKAGE} is unpinned; use '@ {VERSION}'")
|
||||
elif not PIN_PATTERN.fullmatch(version):
|
||||
errors.append(
|
||||
f"{relative}:{number}: {PACKAGE} must be pinned exactly to {VERSION}; "
|
||||
f"found '{version}'"
|
||||
)
|
||||
if expects_pin(config, root, section, lines):
|
||||
required += 1
|
||||
if not found:
|
||||
errors.append(
|
||||
f"{relative}:{header_line}: missing required {PACKAGE} @ {VERSION} "
|
||||
f"in [{section}]"
|
||||
)
|
||||
return configs, declarations, required, errors
|
||||
|
||||
|
||||
def self_test():
|
||||
"""Exercise discovery and the valid, unpinned, wrong, and missing cases."""
|
||||
|
||||
with TemporaryDirectory() as directory:
|
||||
root = Path(directory)
|
||||
root_config = root / "platformio.ini"
|
||||
variant = root / "variants" / "nested" / "mqtt" / "platformio.ini"
|
||||
variant.parent.mkdir(parents=True)
|
||||
root_config.write_text("[env:native]\nbblanchon/ArduinoJson\n", encoding="utf-8")
|
||||
variant.write_text(
|
||||
"[env:wrong]\n-D WITH_MQTT_BRIDGE=1\nbblanchon/ArduinoJson @ ^7.4.3\n"
|
||||
"[env:missing]\nelims/PsychicMqttClient@^0.2.4\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
configs, _, required, errors = check(root)
|
||||
joined = "\n".join(errors)
|
||||
if len(configs) != 2 or required != 3 or len(errors) != 3 or not all(
|
||||
phrase in joined for phrase in ("unpinned", "must be pinned exactly", "missing required")
|
||||
):
|
||||
print("ArduinoJson pin checker self-test failed: invalid cases were not reported.", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
root_config.write_text("[env:native]\nbblanchon/ArduinoJson @ 7.4.3\n", encoding="utf-8")
|
||||
variant.write_text(
|
||||
"[env:one]\n-D WITH_MQTT_BRIDGE=1\nbblanchon/ArduinoJson @ 7.4.3\n"
|
||||
"[env:two]\nelims/PsychicMqttClient@^0.2.4\nbblanchon/ArduinoJson@7.4.3\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
if check(root)[3]:
|
||||
print("ArduinoJson pin checker self-test failed: valid pins were rejected.", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
print("ArduinoJson pin checker self-test passed.")
|
||||
return 0
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--project-root", type=Path, default=Path(__file__).resolve().parents[1])
|
||||
parser.add_argument("--self-test", action="store_true")
|
||||
args = parser.parse_args(argv)
|
||||
if args.self_test:
|
||||
return self_test()
|
||||
|
||||
try:
|
||||
configs, declarations, required, errors = check(args.project_root)
|
||||
except FileNotFoundError as error:
|
||||
print(f"ArduinoJson pin check could not run: {error}", file=sys.stderr)
|
||||
return 2
|
||||
if errors:
|
||||
print("ArduinoJson pin check failed:", *[f" {error}" for error in errors], sep="\n", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
print(
|
||||
f"ArduinoJson pin check passed: {declarations} declaration(s) across {len(configs)} "
|
||||
f"root/variant configuration file(s); {required} required environment section(s) covered."
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Executable
+283
@@ -0,0 +1,283 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Compare MQTT built-in preset *names* between two MQTTPresets.h files.
|
||||
|
||||
Only the first string field of each ``MQTT_PRESETS`` entry is compared (set
|
||||
equality, case-sensitive). URL, auth, CA, keepalive, and credentials are
|
||||
ignored so channel branches may diverge on config details without failing CI.
|
||||
|
||||
Usage::
|
||||
|
||||
python3 scripts/check_mqtt_preset_parity.py FILE_A FILE_B \\
|
||||
--label-a observer-firmware --label-b observer-firmware-dev
|
||||
|
||||
python3 scripts/check_mqtt_preset_parity.py --self-test
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from tempfile import TemporaryDirectory
|
||||
|
||||
|
||||
COUNT_PATTERN = re.compile(
|
||||
r"static\s+const\s+int\s+MQTT_PRESET_COUNT\s*=\s*(\d+)\s*;"
|
||||
)
|
||||
# Start of the preset table; allow optional whitespace/newlines.
|
||||
TABLE_START_PATTERN = re.compile(
|
||||
r"(?:static|extern)\s+const\s+MQTTPresetDef\s+MQTT_PRESETS\s*"
|
||||
r"\[\s*[^\]]*\]\s*=\s*\{",
|
||||
re.MULTILINE,
|
||||
)
|
||||
# First quoted string after an opening brace at entry start (after optional WS).
|
||||
ENTRY_NAME_PATTERN = re.compile(r'\{\s*"([^"]+)"\s*,')
|
||||
|
||||
|
||||
def _strip_line_comment(line: str) -> str:
|
||||
"""Remove ``//`` comments; do not treat ``://`` in URLs as comments."""
|
||||
|
||||
in_string = False
|
||||
i = 0
|
||||
while i < len(line):
|
||||
ch = line[i]
|
||||
if ch == '"' and (i == 0 or line[i - 1] != "\\"):
|
||||
in_string = not in_string
|
||||
elif not in_string and ch == "/" and i + 1 < len(line) and line[i + 1] == "/":
|
||||
# Prefer not to cut URL schemes: require that the previous char is not ':'.
|
||||
if i == 0 or line[i - 1] != ":":
|
||||
return line[:i]
|
||||
i += 1
|
||||
return line
|
||||
|
||||
|
||||
def extract_preset_block(text: str) -> str:
|
||||
"""Return the interior of the ``MQTT_PRESETS[...] = { ... };`` initializer."""
|
||||
|
||||
match = TABLE_START_PATTERN.search(text)
|
||||
if not match:
|
||||
raise ValueError("MQTT_PRESETS table not found")
|
||||
|
||||
depth = 1
|
||||
i = match.end()
|
||||
in_string = False
|
||||
while i < len(text):
|
||||
ch = text[i]
|
||||
if ch == '"' and (i == 0 or text[i - 1] != "\\"):
|
||||
in_string = not in_string
|
||||
elif not in_string:
|
||||
if ch == "{":
|
||||
depth += 1
|
||||
elif ch == "}":
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
return text[match.end() : i]
|
||||
i += 1
|
||||
raise ValueError("MQTT_PRESETS table is not closed")
|
||||
|
||||
|
||||
def parse_presets(text: str, *, source: str = "<input>") -> tuple[list[str], int | None]:
|
||||
"""Return (names in file order, MQTT_PRESET_COUNT or None if absent)."""
|
||||
|
||||
count_match = COUNT_PATTERN.search(text)
|
||||
declared_count = int(count_match.group(1)) if count_match else None
|
||||
|
||||
block = extract_preset_block(text)
|
||||
names: list[str] = []
|
||||
for raw_line in block.splitlines():
|
||||
line = _strip_line_comment(raw_line).strip()
|
||||
if not line:
|
||||
continue
|
||||
entry = ENTRY_NAME_PATTERN.match(line)
|
||||
if entry:
|
||||
names.append(entry.group(1))
|
||||
|
||||
if not names:
|
||||
raise ValueError(f"{source}: no preset entries found in MQTT_PRESETS")
|
||||
|
||||
seen: set[str] = set()
|
||||
dupes: list[str] = []
|
||||
for name in names:
|
||||
if name in seen:
|
||||
dupes.append(name)
|
||||
seen.add(name)
|
||||
if dupes:
|
||||
raise ValueError(f"{source}: duplicate preset name(s): {', '.join(sorted(set(dupes)))}")
|
||||
|
||||
if declared_count is not None and declared_count != len(names):
|
||||
raise ValueError(
|
||||
f"{source}: MQTT_PRESET_COUNT is {declared_count} but found {len(names)} preset entries"
|
||||
)
|
||||
|
||||
return names, declared_count
|
||||
|
||||
|
||||
def format_missing(label: str, names: set[str]) -> str:
|
||||
if not names:
|
||||
return f"Missing from {label}: (none)"
|
||||
return f"Missing from {label}: {', '.join(sorted(names))}"
|
||||
|
||||
|
||||
def compare(
|
||||
names_a: set[str],
|
||||
names_b: set[str],
|
||||
*,
|
||||
label_a: str,
|
||||
label_b: str,
|
||||
) -> list[str]:
|
||||
"""Return human-readable error lines if sets differ; empty list if equal."""
|
||||
|
||||
only_a = names_a - names_b
|
||||
only_b = names_b - names_a
|
||||
if not only_a and not only_b:
|
||||
return []
|
||||
return [
|
||||
format_missing(label_b, only_a), # in A but not B -> missing from B
|
||||
format_missing(label_a, only_b), # in B but not A -> missing from A
|
||||
]
|
||||
|
||||
|
||||
def load_names(path: Path) -> set[str]:
|
||||
text = path.read_text(encoding="utf-8")
|
||||
names, _ = parse_presets(text, source=str(path))
|
||||
return set(names)
|
||||
|
||||
|
||||
def self_test() -> int:
|
||||
"""Exercise equal-with-different-URLs, missing-name, and duplicate cases."""
|
||||
|
||||
header = """
|
||||
static const int MQTT_PRESET_COUNT = {count};
|
||||
static const MQTTPresetDef MQTT_PRESETS[MQTT_PRESET_COUNT] = {{
|
||||
{body}
|
||||
}};
|
||||
"""
|
||||
|
||||
with TemporaryDirectory() as directory:
|
||||
root = Path(directory)
|
||||
equal_a = root / "a.h"
|
||||
equal_b = root / "b.h"
|
||||
missing = root / "missing.h"
|
||||
dupes = root / "dupes.h"
|
||||
|
||||
equal_a.write_text(
|
||||
header.format(
|
||||
count=2,
|
||||
body=' { "alpha", "wss://a.example/mqtt", nullptr },\n'
|
||||
' { "beta", "wss://b.example/mqtt", nullptr },',
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
# Same names, different URLs/config -- must pass.
|
||||
equal_b.write_text(
|
||||
header.format(
|
||||
count=2,
|
||||
body=' { "beta", "mqtt://other:1883", nullptr },\n'
|
||||
' { "alpha", "wss://changed.example/mqtt", nullptr },',
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
if compare(load_names(equal_a), load_names(equal_b), label_a="a", label_b="b"):
|
||||
print("self-test failed: equal name sets with different URLs were rejected.", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
missing.write_text(
|
||||
header.format(
|
||||
count=1,
|
||||
body=' { "alpha", "wss://a.example/mqtt", nullptr },',
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
errs = compare(load_names(equal_a), load_names(missing), label_a="a", label_b="missing")
|
||||
if len(errs) != 2 or "beta" not in errs[0] or "(none)" not in errs[1]:
|
||||
print(f"self-test failed: missing-name case unexpected: {errs!r}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
try:
|
||||
parse_presets(
|
||||
header.format(
|
||||
count=2,
|
||||
body=' { "alpha", "wss://a.example/mqtt", nullptr },\n'
|
||||
' { "alpha", "wss://b.example/mqtt", nullptr },',
|
||||
),
|
||||
source="dupes",
|
||||
)
|
||||
print("self-test failed: duplicate names were accepted.", file=sys.stderr)
|
||||
return 1
|
||||
except ValueError as error:
|
||||
if "duplicate" not in str(error):
|
||||
print(f"self-test failed: unexpected duplicate error: {error}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
try:
|
||||
parse_presets(
|
||||
header.format(
|
||||
count=99,
|
||||
body=' { "alpha", "wss://a.example/mqtt", nullptr },',
|
||||
),
|
||||
source="count-mismatch",
|
||||
)
|
||||
print("self-test failed: count mismatch was accepted.", file=sys.stderr)
|
||||
return 1
|
||||
except ValueError as error:
|
||||
if "MQTT_PRESET_COUNT" not in str(error):
|
||||
print(f"self-test failed: unexpected count error: {error}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
# URL with :// must not be treated as a line comment.
|
||||
url_comment = root / "url.h"
|
||||
url_comment.write_text(
|
||||
header.format(
|
||||
count=1,
|
||||
body=' { "alpha", "wss://mqtt.example:443/mqtt", nullptr }, // note',
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
if load_names(url_comment) != {"alpha"}:
|
||||
print("self-test failed: URL scheme was treated as a comment.", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
# Silence unused path in fixture layout.
|
||||
dupes.write_text("// unused\n", encoding="utf-8")
|
||||
|
||||
print("MQTT preset parity checker self-test passed.")
|
||||
return 0
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("file_a", type=Path, nargs="?", help="First MQTTPresets.h (e.g. production)")
|
||||
parser.add_argument("file_b", type=Path, nargs="?", help="Second MQTTPresets.h (e.g. dev)")
|
||||
parser.add_argument("--label-a", default="file_a", help="Label for file_a in reports")
|
||||
parser.add_argument("--label-b", default="file_b", help="Label for file_b in reports")
|
||||
parser.add_argument("--self-test", action="store_true")
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
if args.self_test:
|
||||
return self_test()
|
||||
|
||||
if args.file_a is None or args.file_b is None:
|
||||
parser.error("FILE_A and FILE_B are required unless --self-test is set")
|
||||
|
||||
try:
|
||||
names_a = load_names(args.file_a)
|
||||
names_b = load_names(args.file_b)
|
||||
except (OSError, ValueError) as error:
|
||||
print(f"MQTT preset parity check could not run: {error}", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
errors = compare(names_a, names_b, label_a=args.label_a, label_b=args.label_b)
|
||||
if errors:
|
||||
print("MQTT preset name parity check failed:", *errors, sep="\n ", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
print(
|
||||
f"MQTT preset name parity check passed: {len(names_a)} preset name(s) "
|
||||
f"match between {args.label_a} and {args.label_b}."
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,508 @@
|
||||
#!/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)
|
||||
# latched at AP start, like WebConfigServer::_initial_setup
|
||||
self.initial_setup = args.setup and self.cfg["wifi"]["ssid"] == ""
|
||||
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 == "password":
|
||||
# Stored outside cfg: it must never appear in the /api/config GET. The
|
||||
# firmware overwrites the CLI's "password now: <secret>" echo, so the
|
||||
# reply carries no secret either.
|
||||
global ADMIN_PASSWORD
|
||||
ADMIN_PASSWORD = val
|
||||
return True, "OK"
|
||||
|
||||
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 {}
|
||||
|
||||
# `password` maps to the top-level CLI command rather than a setter. It
|
||||
# is accepted in both modes (LAN already required a login), but first
|
||||
# onboarding cannot finish without it.
|
||||
if "password" in setmap:
|
||||
pwd = str(setmap["password"])
|
||||
if not 0 < len(pwd) <= 15 or "\r" in pwd or "\n" in pwd:
|
||||
return self._json(400, {"error": "admin password must be 1-15 characters with no line breaks"})
|
||||
elif ST.setup_mode and ST.initial_setup and (reboot or "wifi.ssid" in setmap):
|
||||
return self._json(400, {"error": "admin password required for initial setup"})
|
||||
|
||||
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()
|
||||
Reference in new issue
Block a user