Harden USB logging and persist watchdog events in management reports

tools/management/report.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Decode and authenticate complete MGR1 management reports.
+"""Decode and authenticate complete MGR1/MGR2/MGR3 management reports.
 
 Input is a JSON list of raw payload hex strings, one entry for each page (not
 MeshCore packet headers/paths). A relay/collector can obtain these from RX logs
@@ -14,6 +14,66 @@ import struct
 from pathlib import Path
 
 HEADER, ENTRY, PER_PAGE, TAG, MAX_KEYS = 83, 13, 6, 16, 36
+MGR2_HEADER, MGR2_PER_PAGE = 98, 5
+CURRENT_HEADER, CURRENT_PER_PAGE = 111, 4
+
+
+def _layout(payload):
+    formats = {b"MGR1": (HEADER, PER_PAGE), b"MGR2": (MGR2_HEADER, MGR2_PER_PAGE),
+               b"MGR3": (CURRENT_HEADER, CURRENT_PER_PAGE)}
+    result = formats.get(payload[:4])
+    if result is None or len(payload) < result[0] + TAG:
+        raise ValueError("invalid management payload")
+    return result
+
+
+def _page_bounds(payload):
+    header, per_page = _layout(payload)
+    if header >= MGR2_HEADER and (int.from_bytes(payload[83:85], "little") & 0xf800 or payload[85] & 0xc0):
+        raise ValueError("reserved USB status bits must be zero")
+    if header == CURRENT_HEADER:
+        _watchdog_event(payload)
+    page, pages, total, first, count = payload[78:83]
+    expected = max(1, (total + per_page - 1) // per_page)
+    if (total > MAX_KEYS or pages != expected or page >= pages or
+            first != page * per_page or first > total or
+            count != min(per_page, total - first)):
+        raise ValueError("invalid management page bounds")
+    return header, header + count * ENTRY + TAG
+
+
+def _usb_status(payload):
+    if payload[:4] not in (b"MGR2", b"MGR3"):
+        return None
+    flags = int.from_bytes(payload[83:85], "little")
+    names = ("supported", "logging_enabled", "watchdog_enabled", "host_connected",
+             "reader_connected", "stalled", "recovering", "recovery_deferred",
+             "persistence_ready", "watchdog_auto", "logger_active")
+    result = {name: bool(flags & (1 << bit)) for bit, name in enumerate(names)}
+    result.update(stage=payload[85] & 3, backoff_step=(payload[85] >> 2) & 15,
+                  retry_seconds=int.from_bytes(payload[86:90], "little"),
+                  inactive_seconds=int.from_bytes(payload[90:94], "little"),
+                  auto_connected_seconds=int.from_bytes(payload[94:98], "little"))
+    return result
+
+
+def _watchdog_event(payload):
+    if payload[:4] != b"MGR3":
+        return None
+    code = payload[98]
+    epoch, uptime, sequence = struct.unpack_from("<III", payload, 99)
+    if not sequence:
+        if any(payload[98:111]):
+            raise ValueError("invalid empty USB watchdog event")
+        return None
+    reasons, action = code & 15, (code >> 4) & 7
+    if not reasons or not 1 <= action <= 4:
+        raise ValueError("invalid USB watchdog event reason/action")
+    names = ("host-absent", "reader-absent", "tx-stalled", "client-inactive")
+    actions = (None, "soft-recovery", "reenumerate", "reboot-requested", "reboot-cancelled")
+    return dict(reasons=reasons, reason_names=[name for bit, name in enumerate(names) if reasons & (1 << bit)],
+                action_code=action, action=actions[action], persisted=bool(code & 128),
+                epoch=epoch, uptime_seconds=uptime, sequence=sequence)
 
 
 def password_key(password):
@@ -54,18 +114,14 @@ def _extrema(raw):
 
 def decode_page(payload, password):
     from Crypto.Cipher import AES  # PyCryptodome; independent AES-SIV implementation
-    if not HEADER + TAG <= len(payload) <= 177 or payload[:4] != b"MGR1":
-        raise ValueError("invalid management payload")
-    page, pages, total, first, count = payload[78:83]
-    expected = max(1, (total + PER_PAGE - 1) // PER_PAGE)
-    if (total > MAX_KEYS or pages != expected or page >= pages or
-            first != page * PER_PAGE or first > total or
-            count != min(PER_PAGE, total - first) or len(payload) != HEADER + count * ENTRY + TAG):
+    header, canonical = _page_bounds(payload)
+    count = payload[82]
+    if len(payload) != canonical or len(payload) > 179:
         raise ValueError("invalid management page bounds")
     key = derive(password_key(password), "MeshCore-MGR1-SIV", payload[4:20])
     cipher = AES.new(key, AES.MODE_SIV)
-    cipher.update(payload[:HEADER])
-    private = cipher.decrypt_and_verify(payload[HEADER:-TAG], payload[-TAG:])
+    cipher.update(payload[:header])
+    private = cipher.decrypt_and_verify(payload[header:-TAG], payload[-TAG:])
     entries = []
     for i in range(count):
         token = private[i * ENTRY:i * ENTRY + 12]
@@ -77,7 +133,7 @@ def decode_page(payload, password):
 
 
 def mqtt_payload(message):
-    """Extract an MGR1 payload from the observer's MQTT PACKET/raw JSON.
+    """Extract an MGR1/MGR2/MGR3 payload from the observer's MQTT PACKET/raw JSON.
 
     Path length is encoded, not simply a byte count. Region transport codes
     precede it. Never trust the redundant MQTT payload_len/type fields.
@@ -103,14 +159,15 @@ def mqtt_payload(message):
     start = position + 1 + width * count
     payload = packet[start:]
     if (packet[0] >> 2) & 15 == 6:
-        if len(payload) < HEADER + TAG or payload[:4] != b"MGR1":
+        try:
+            _, canonical = _page_bounds(payload)
+        except ValueError:
             return None
-        canonical = HEADER + payload[82] * ENTRY + TAG
         padded = 3 + ((canonical - 3 + 15) // 16) * 16
         if len(payload) != padded or any(payload[canonical:]):
             return None
         payload = payload[:canonical]
-    return payload if payload[:4] == b"MGR1" else None
+    return payload if payload[:4] in (b"MGR1", b"MGR2", b"MGR3") else None
 
 
 def mqtt_reports(messages, password):
@@ -140,8 +197,8 @@ def mqtt_reports(messages, password):
 
 
 def decode_report(payloads, password):
-    if not payloads or len(payloads) > 6:
-        raise ValueError("one through six pages required")
+    if not payloads or len(payloads) > 9:
+        raise ValueError("one through nine pages required")
     # Check every page before indexing metadata, then reject omissions,
     # duplicates, mixed snapshots and unauthenticated public fields.
     decoded = [(p, decode_page(p, password)) for p in payloads]
@@ -149,7 +206,9 @@ def decode_report(payloads, password):
     first = decoded[0][0]
     if len(decoded) != first[79] or [p[78] for p, _ in decoded] != list(range(first[79])):
         raise ValueError("incomplete or duplicate report pages")
-    if any(p[:78] != first[:78] or p[79:81] != first[79:81] for p, _ in decoded):
+    header, _ = _layout(first)
+    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")
     fields = struct.unpack_from("<IIIII", first, 20)
     sequence, timestamp, firmware, bootloader, target = fields
@@ -157,7 +216,8 @@ def decode_report(payloads, password):
     radio = first[4:20]
     def version(value):
         return ".".join(str((value >> n) & 255) for n in (24, 16, 8, 0))
-    result = dict(radio_id=radio.hex(), sequence=sequence, timestamp=timestamp,
+    result = dict(protocol=first[:4].decode("ascii"),
+                  radio_id=radio.hex(), sequence=sequence, timestamp=timestamp,
                   firmware_version=version(firmware) if valid & 1 else None,
                   bootloader_version=version(bootloader) if valid & 2 else None,
                   target_id=f"{target:08x}" if valid & 4 else None,
@@ -171,6 +231,8 @@ def decode_report(payloads, password):
                   capability_bits=first[73], active_bits=first[74], known_bits=first[75],
                   partial_week=bool(valid & 16), partial_period=bool(valid & 32),
                   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])
     return result