Validate management CLI captures and authenticated ACL snapshots

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.

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

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("<IIIII", first, 20)
     sequence, timestamp, firmware, bootloader, target = fields
     valid = int.from_bytes(first[76:78], "little")
@@ -233,7 +240,7 @@ def decode_report(payloads, password):
                   temperature_source="MCU" if valid & 64 else "unknown",
                   usb_logging=_usb_status(first),
                   usb_watchdog_last=_watchdog_event(first),
-                  acl=[entry for _, entries in decoded for entry in entries])
+                  acl=acl)
     return result
 
 
@@ -243,20 +250,34 @@ def main():
     parser.add_argument("--match-admin", help="full public key to match against private ACL fingerprints")
     parser.add_argument("--mqtt", action="store_true", help="input is JSON array or JSONL of saved MQTT uplinks")
     args = parser.parse_args()
-    password = getpass.getpass("Management password: ")
     try:
-        text = args.pages.read_text()
+        administrator = None
+        if args.match_admin is not None:
+            administrator = bytes.fromhex(args.match_admin)
+            if len(administrator) != 32:
+                raise ValueError("administrator key must be 32 bytes")
+        text = args.pages.read_text(encoding="utf-8")
         if args.mqtt:
             messages = json.loads(text) if text.lstrip().startswith("[") else [json.loads(line) for line in text.splitlines() if line.strip()]
-            print(json.dumps(mqtt_reports(messages, password), indent=2))
-            return
-        report = decode_report([bytes.fromhex(p) for p in json.loads(text)], password)
-        if args.match_admin:
-            match = fingerprint(password, bytes.fromhex(report["radio_id"]), bytes.fromhex(args.match_admin)).hex()
-            report["matching_acl"] = [entry for entry in report["acl"] if entry["fingerprint"] == match]
-    except (ValueError, TypeError, KeyError) as exc:
+        else:
+            payloads = [bytes.fromhex(p) for p in json.loads(text)]
+        password = getpass.getpass("Management password: ")
+        if args.mqtt:
+            reports = mqtt_reports(messages, password)
+        else:
+            reports = [decode_report(payloads, password)]
+        if administrator is not None:
+            for report in reports:
+                match = fingerprint(password, bytes.fromhex(report["radio_id"]), administrator).hex()
+                report["matching_acl"] = [entry for entry in report["acl"] if entry["fingerprint"] == match]
+        output = json.dumps(reports if args.mqtt else reports[0], indent=2)
+    except EOFError:
+        parser.error("management password input unavailable; run in an interactive terminal")
+    except RecursionError:
+        parser.error("input JSON is nested too deeply")
+    except (OSError, ValueError, TypeError, KeyError) as exc:
         parser.error(str(exc))
-    print(json.dumps(report, indent=2))
+    print(output)
 
 
 if __name__ == "__main__":