diff --git a/docs/management_reports.md b/docs/management_reports.md index 5cf96aa0f..999e2343b 100644 --- a/docs/management_reports.md +++ b/docs/management_reports.md @@ -32,13 +32,17 @@ Use a long randomly generated password. There is no password getter. Firmware stores the derived 32-byte key, not the plaintext password. That stored key is equivalent authority to decrypt reports and must also be protected. Passwords entered into terminal programs may still be recorded by those programs. +The recognized firmware password command wipes its value from the input buffer +on success and error, including unavailable management or allocation failure. `data.tx` is the single shared route for management reports, telemetry history, and future scheduled data producers. Its fresh-install defaults are `path=direct` (zero hops) and `region=auto`; configuring it never enables a producer. Paths use `1:`, `2:` or `3:` followed by complete hop hashes without separators, or -the comma-separated form accepted by `set outpath`. `none` removes the path. +the comma-separated form accepted by `set outpath`. Empty hops, including a +trailing comma, are rejected. `none` removes the path. `get/set mgmt.path` remain compatibility aliases for `get/set data.tx path`. +Alias expansion that exceeds the command buffer is rejected, not truncated. `region=auto` uses the radio's configured default region when it is usable; otherwise it resolves the unique deepest flood-enabled entry in the region @@ -59,7 +63,10 @@ shared path. Flood accepts 21-90 days and requires a resolvable shared region. At least one route must remain active while reporting is enabled. The legacy `set mgmt.interval N` shorthand enables both, setting direct to `N` and flood to `max(21,N)`. No transmission is sent merely by configuring a password. Initial -reporting waits for the configured schedules. Reports have deterministic +reporting waits for the configured schedules. Invalid settings leave saved +configuration, countdowns, and any report pages already in progress unchanged. +Day counts must be unsigned decimal integers; signs and trailing text are rejected. +Reports have deterministic per-radio/per-sequence jitter of up to an hour; pages are spaced at least a minute apart. Watchdog events appear in the next scheduled snapshot, not as immediate radio alerts. Multiple events between snapshots can replace the @@ -84,6 +91,12 @@ writes stop reporting; `get mgmt` shows `FAULT(no TX)` until storage is repaired and the radio restarted. `/management` and the shared `/data_tx` are versioned, CRC-protected, and replaced transactionally. +On upgrade, an existing saved `data.tx` route takes precedence over the legacy +management-only path. Otherwise that path must be saved successfully to +`/data_tx` before reporting resumes. A failed migration leaves the old +`/management` file intact for retry after storage is repaired and the radio +restarted. + Weekly history is collected once a minute while enabled. Hour-bucket extrema cover 7 days to 7 days + 1 hour (conservative boundary bucket). The first report after enable/reboot is marked partial where appropriate. Since-report extrema @@ -126,8 +139,11 @@ python tools/management/report.py --mqtt capture.jsonl The password is prompted, not supplied as a process argument. Input may be JSONL or a JSON array of MQTT messages. Alternatively omit `--mqtt` for a JSON array -of canonical payload hex strings. `--match-admin FULL_PUBLIC_KEY` in canonical -payload mode matches a known administrator against the encrypted fingerprints. +of canonical payload hex strings. `--match-admin FULL_PUBLIC_KEY` matches a known +administrator against the encrypted fingerprints in either input mode. MQTT +captures match each radio's fingerprints separately. Invalid files or keys are +rejected before the password prompt; an unavailable password input gives a +command-line error instead of a traceback. This is an offline capture decoder, not a broker subscriber or downlink service. For a browser-local decoder that also decrypts and authenticates ACL entries, @@ -237,7 +253,10 @@ six entries per page and up to six pages, with no USB block. Its absent USB status is decoded as unavailable, not as all switches off. Each private entry is a per-radio 12-byte keyed fingerprint followed by flags: -bit 0 administrator, bit 1 trusted OTA signer. Duplicate entries combine flags. +bit 0 administrator, bit 1 trusted OTA signer. The firmware combines permissions +for the same fingerprint before paging, so every transmitted entry is unique. +Decoders reject authenticated duplicate fingerprints within or across pages; +repeated identical packet copies from multiple uplinks are still deduplicated. An oversized ACL fails closed rather than silently truncating. The reported allowlist includes all current full administrators and all four possible trusted OTA signing keys; region/filter managers and ordinary clients are excluded. diff --git a/test/test_management_report.py b/test/test_management_report.py index 5da506477..8c7803ea8 100644 --- a/test/test_management_report.py +++ b/test/test_management_report.py @@ -1,11 +1,17 @@ """Production crypto vs RFC vector + independent Python decoder, no AES mocks.""" from pathlib import Path +import contextlib import importlib.util +import io +import json import os import shutil import subprocess +import sys import tempfile import unittest +from unittest import mock +from test_replay_reset_integration import extract_braced ROOT = Path(__file__).resolve().parents[1] spec = importlib.util.spec_from_file_location("management", ROOT / "tools/management/report.py") @@ -90,6 +96,154 @@ class ManagementTests(unittest.TestCase): ciphertext, tag = cipher.encrypt_and_digest(private) return bytes(header) + ciphertext + tag + def _run_cli(self, path, *options): + output, errors = io.StringIO(), io.StringIO() + status = 0 + with mock.patch.object(sys, "argv", ["report.py", str(path), *options]), \ + mock.patch.object(report.getpass, "getpass", return_value="management test password"), \ + contextlib.redirect_stdout(output), contextlib.redirect_stderr(errors): + try: + report.main() + except SystemExit as error: + status = error.code + return status, output.getvalue(), errors.getvalue() + + @classmethod + def _acl_pages(cls, entries, current=True): + from Crypto.Cipher import AES + base = cls.empty_event_pages[0][:111] if current else cls.legacy_pages[0][:83] + per_page = 4 if current else 6 + count = max(1, (len(entries) + per_page - 1) // per_page) + pages = [] + for index in range(count): + header = bytearray(base) + block = entries[index * per_page:(index + 1) * per_page] + header[78:83] = bytes((index, count, len(entries), index * per_page, len(block))) + key = report.derive(report.password_key("management test password"), + "MeshCore-MGR1-SIV", header[4:20]) + cipher = AES.new(key, AES.MODE_SIV); cipher.update(header) + ciphertext, tag = cipher.encrypt_and_digest(b"".join(token + bytes((flags,)) for token, flags in block)) + pages.append(bytes(header) + ciphertext + tag) + return pages + + def test_authenticated_duplicate_acl_fingerprints_are_rejected(self): + password = "management test password" + for current in (False, True): + duplicate = self._acl_pages([(bytes(12), 1), (bytes(12), 2)], current) + with self.assertRaisesRegex(ValueError, "[Dd]uplicate"): + report.decode_page(duplicate[0], password) + with self.assertRaisesRegex(ValueError, "[Dd]uplicate"): + report.decode_report(duplicate, password) + per_page = 4 if current else 6 + unique = [(index.to_bytes(12, "little"), 3) for index in range(per_page)] + duplicate = self._acl_pages([*unique, (unique[0][0], 2)], current) + for page in duplicate: + report.decode_page(page, password) # No duplicates within either page. + with self.assertRaisesRegex(ValueError, "[Dd]uplicate"): + report.decode_report(duplicate, password) + messages = [] + for page in duplicate: + padded = page + bytes(3 + ((len(page) - 3 + 15) // 16) * 16 - len(page)) + messages.append(dict(type="PACKET", direction="rx", raw=(b"\x1a\x00" + padded).hex())) + with self.assertRaisesRegex(ValueError, "[Dd]uplicate"): + report.mqtt_reports(messages, password) + valid = self._acl_pages([*unique, (per_page.to_bytes(12, "little"), 3)], current) + decoded = report.decode_report(valid, password) + self.assertEqual(len(decoded["acl"]), per_page + 1) + self.assertTrue(all(entry["admin"] and entry["ota_signer"] for entry in decoded["acl"])) + + def test_cli_matches_administrator_for_each_mqtt_report(self): + from Crypto.Cipher import AES + administrator = bytes(range(32)) + messages = [] + for radio in (bytes(range(16)), bytes(range(16, 32))): + header = bytearray(self.empty_event_pages[0][:111]) + header[4:20] = radio + header[78:83] = bytes((0, 1, 1, 0, 1)) + key = report.derive(report.password_key("management test password"), + "MeshCore-MGR1-SIV", radio) + private = report.fingerprint("management test password", radio, administrator) + b"\x03" + cipher = AES.new(key, AES.MODE_SIV); cipher.update(header) + ciphertext, tag = cipher.encrypt_and_digest(private) + page = bytes(header) + ciphertext + tag + padded = page + bytes(3 + ((len(page) - 3 + 15) // 16) * 16 - len(page)) + messages.append(dict(type="PACKET", direction="rx", raw=(b"\x1a\x00" + padded).hex())) + path = Path(self.work.name) / "cli-mqtt.json" + for jsonl in (False, True): + path.write_text("\n".join(json.dumps(message) for message in messages) + if jsonl else json.dumps(messages), encoding="ascii") + status, output, errors = self._run_cli(path, "--mqtt", "--match-admin", administrator.hex()) + self.assertEqual((status, errors), (0, "")) + decoded = json.loads(output) + self.assertEqual(len(decoded), 2) + for item in decoded: + self.assertEqual(item.get("matching_acl"), item["acl"]) + self.assertEqual(len(item["matching_acl"]), 1) + self.assertTrue(item["matching_acl"][0]["admin"] and item["matching_acl"][0]["ota_signer"]) + status, output, errors = self._run_cli(path, "--mqtt", "--match-admin", "ff" * 32) + self.assertEqual((status, errors), (0, "")) + self.assertEqual([item.get("matching_acl") for item in json.loads(output)], [[], []]) + + def test_cli_rejects_invalid_match_key_even_with_no_mqtt_reports(self): + path = Path(self.work.name) / "cli-empty.json" + path.write_text("[]", encoding="ascii") + for candidate in ("not-hex", "00", ""): + status, output, errors = self._run_cli(path, "--mqtt", "--match-admin", candidate) + self.assertEqual(status, 2) + self.assertEqual(output, "") + self.assertIn("error:", errors) + + def test_cli_file_read_errors_are_reported_without_tracebacks(self): + for path in (Path(self.work.name) / "missing.json", Path(self.work.name)): + status, output, errors = self._run_cli(path) + self.assertEqual(status, 2) + self.assertEqual(output, "") + self.assertIn("error:", errors) + self.assertNotIn("Traceback", errors) + + def test_cli_closed_stdin_reports_input_errors_without_tracebacks(self): + path = Path(self.work.name) / "cli-no-password.json" + path.write_text(json.dumps([page.hex() for page in self.pages]), encoding="ascii") + for arguments, expected in (([str(path)], "password input unavailable"), + ([str(path.parent / "absent.json")], "No such file"), + ([str(path), "--match-admin", "00"], "administrator key")): + result = subprocess.run([sys.executable, "-B", str(ROOT / "tools/management/report.py"), + *arguments], input="", capture_output=True, text=True) + self.assertEqual(result.returncode, 2) + self.assertEqual(result.stdout, "") + self.assertIn(expected, result.stderr) + self.assertNotIn("Traceback", result.stderr) + if expected != "password input unavailable": + self.assertNotIn("Management password:", result.stderr) + + def test_cli_canonical_match_and_input_errors(self): + path = Path(self.work.name) / "cli-pages.json" + path.write_text(json.dumps([page.hex() for page in self.pages]), encoding="ascii") + status, output, errors = self._run_cli(path, "--match-admin", "00" * 32) + self.assertEqual((status, errors), (0, "")) + self.assertEqual(json.loads(output)["matching_acl"], []) + for malformed in ("{", '["not hex"]'): + path.write_text(malformed, encoding="ascii") + status, output, errors = self._run_cli(path) + self.assertEqual(status, 2) + self.assertEqual(output, "") + self.assertIn("error:", errors) + + def test_cli_rejects_excessive_json_nesting_before_password_prompt(self): + path = Path(self.work.name) / "cli-deep.json" + path.write_text("[" * 10000 + "0" + "]" * 10000, encoding="ascii") + for options in ((), ("--mqtt",)): + with mock.patch.object(report.getpass, "getpass") as password_prompt: + with mock.patch.object(sys, "argv", ["report.py", str(path), *options]), \ + contextlib.redirect_stdout(io.StringIO()), \ + contextlib.redirect_stderr(io.StringIO()) as errors: + with self.assertRaises(SystemExit) as failure: + report.main() + self.assertEqual(failure.exception.code, 2) + self.assertIn("nested too deeply", errors.getvalue()) + self.assertNotIn("Traceback", errors.getvalue()) + password_prompt.assert_not_called() + def test_interoperable_full_report(self): decoded = report.decode_report(list(reversed(self.pages)), "management test password") self.assertEqual(decoded["sequence"], 42) @@ -242,6 +396,29 @@ process.stdin.on('end', async () => { def test_runtime_storage_scheduling_rollover_and_failures(self): subprocess.run([str(self.runtime)], check=True) + def test_actual_legacy_route_adoption_persistence_and_precedence(self): + source = (ROOT / "src/helpers/CommonCLI_Management.cpp").read_text(encoding="utf-8") + state = source[source.index("static constexpr char DATA_ROUTE_FILE[]"): + source.index("static uint32_t readRoute32(")] + generated = ("namespace mesh {\n" + state + + extract_braced(source, "static uint8_t nibble(") + "\n" + + extract_braced(source, "static char* trimDataRoute(") + "\n" + + extract_braced(source, "static bool parseDataPath(") + "\n" + + extract_braced(source, "static void writeRoute32(") + "\n" + + extract_braced(source, "static bool dataRouteSave(") + "\n}\n" + + extract_braced(source, "bool CommonCLI::adoptLegacyDataTxPath(") + "\n") + (Path(self.work.name) / "production_route.inc").write_text(generated, encoding="ascii") + executable = Path(self.work.name) / "route-test" + sanitizers = [] if os.name == "nt" else ["-DHOST_BUILD", "-fsanitize=address,undefined", + "-fno-sanitize-recover=all", "-fno-omit-frame-pointer"] + compiled = subprocess.run([shutil.which("g++") or "g++", "-std=c++17", "-O1", "-g", + "-Wall", "-Wextra", "-DRP2040_PLATFORM", *sanitizers, + "-I", self.work.name, + str(ROOT / "test/fixtures/management/route_test.cpp"), + "-o", str(executable)], capture_output=True, text=True) + self.assertEqual(compiled.returncode, 0, compiled.stderr) + subprocess.run([str(executable)], check=True) + def test_tampering_and_wrong_password(self): for position in (4, 20, 28, 74, 76, 83, 85, 86, 90, 94, *range(98, 111), 111, -1): changed = bytearray(self.pages[0]); changed[position] ^= 1 diff --git a/tools/management/report.py b/tools/management/report.py index dc2e83a2d..321a05d12 100644 --- a/tools/management/report.py +++ b/tools/management/report.py @@ -121,11 +121,15 @@ def decode_page(payload, password): cipher.update(payload[:header]) private = cipher.decrypt_and_verify(payload[header:-TAG], payload[-TAG:]) entries = [] + seen = set() for i in range(count): token = private[i * ENTRY:i * ENTRY + 12] flags = private[i * ENTRY + 12] if not flags or flags & ~3: raise ValueError("invalid ACL role flags") + if token in seen: + raise ValueError("duplicate ACL fingerprint") + seen.add(token) entries.append(dict(fingerprint=token.hex(), admin=bool(flags & 1), ota_signer=bool(flags & 2))) return entries @@ -210,6 +214,9 @@ def decode_report(payloads, password): if any(p[:78] != first[:78] or p[79:81] != first[79:81] or p[HEADER:header] != first[HEADER:header] for p, _ in decoded): raise ValueError("mixed report snapshots") + acl = [entry for _, entries in decoded for entry in entries] + if len({entry["fingerprint"] for entry in acl}) != len(acl): + raise ValueError("duplicate ACL fingerprint") fields = struct.unpack_from("