improv_wifi: verify checksums, request device info; add README

README.md

@@ -0,0 +1,22 @@
+# firmware-tester
+
+Hardware-in-the-loop harness for [ESPresense](https://github.com/ESPresense/ESPresense) firmware, run by CrowCI (Woodpecker) against real ESP32 / C3 / C6 / S3 boards on the bench.
+
+## What is in here
+
+* `scripts/improv_wifi.py` — provisions WiFi over [Improv Serial](https://www.improv-wifi.com/serial/): requests device info (validates the frames and their checksums), sends the credentials, and expects the `provisioning` state.
+* `scripts/hil_monitor.py` — watches the serial console for `--duration` seconds and fails on crash patterns, restarts, missing scan results, the MQTT reconnect limit, a misbehaving `/json` under concurrent load, or a declining heap (see the exit codes at the top of the file). `--dma-bisect` runs the A/B browser-load heap test.
+* `scripts/test_*.py` — host-side tests for the monitor's heuristics (`python3 -m pytest scripts`).
+* `Dockerfile` — the `ghcr.io/espresense/firmware-tester` image (PlatformIO + the scripts) used by the Arduino-era pipeline. The ESP-IDF firmware runs these scripts from the `espressif/idf` image and clones this repo at run time.
+
+## Running by hand
+
+```sh
+pip install pyserial
+python3 scripts/improv_wifi.py --port /dev/ttyUSB0 --ssid MyWifi --password secret
+python3 scripts/hil_monitor.py --port /dev/ttyUSB0 --duration 180
+```
+
+## Releasing the image
+
+`gh release create vX.Y.Z --target <sha>` — `docker.yml` builds and pushes `:1`, `:X`, `:X.Y`; the pipeline pins the floating major tag.

scripts/improv_wifi.py

@@ -33,10 +33,33 @@ def build_wifi_packet(ssid: str, password: str) -> bytes:
     return payload + bytes([checksum])
 
 
+class BadChecksum(Exception):
+    pass
+
+
+def build_rpc_packet(command: int, data: bytes = b"") -> bytes:
+    """RPC command frame: command byte, data-length byte, data."""
+    rpc_data = bytes([command, len(data)]) + data
+    payload = b"IMPROV\x01\x03" + bytes([len(rpc_data)]) + rpc_data
+    return payload + bytes([sum(payload) & 0xFF])
+
+
+def decode_strings(data: bytes):
+    """Length-prefixed string list, as used by the RPC response payload."""
+    out, i = [], 0
+    while i < len(data):
+        n = data[i]
+        out.append(data[i + 1:i + 1 + n].decode("utf-8", "replace"))
+        i += 1 + n
+    return out
+
+
 def parse_improv_packet(buf: bytes):
     """Try to parse an Improv packet from a buffer.
 
-    Returns (packet_type, state_byte, consumed) or None.
+    Returns (packet_type, state_byte, consumed) or None. Raises BadChecksum when the
+    frame is complete but its checksum is wrong: a firmware whose serial path mangles
+    bytes (CRLF translation, dropped bytes) must fail here, not pass by luck.
     """
     idx = buf.find(b"IMPROV")
     if idx < 0 or len(buf) < idx + 10:
@@ -53,10 +76,39 @@ def parse_improv_packet(buf: bytes):
     if len(buf) < end:
         return None
 
+    expected = sum(buf[idx:end - 1]) & 0xFF
+    if buf[end - 1] != expected:
+        raise BadChecksum(f"Improv checksum {buf[end - 1]:#x}, expected {expected:#x}")
+
     state = buf[idx + 9]
     return (packet_type, state, end)
 
 
+def request_info(ser, timeout=10):
+    """Ask for device info first: its reply carries free text (names, versions), so it
+    exercises far more byte values than a bare state packet does."""
+    ser.write(build_rpc_packet(0x03))
+    ser.flush()
+    buf = b""
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        if ser.in_waiting:
+            buf += ser.read(ser.in_waiting)
+            while True:
+                result = parse_improv_packet(buf)
+                if not result:
+                    break
+                ptype, _, consumed = result
+                frame = buf[buf.find(b"IMPROV"):consumed]
+                buf = buf[consumed:]
+                if ptype == 0x04:  # RPC_Response: command, data length, strings
+                    print(f"Device info: {decode_strings(frame[11:-1])}")
+                    return True
+        time.sleep(0.1)
+    print("Timeout waiting for Improv device info", file=sys.stderr)
+    return False
+
+
 def main():
     parser = argparse.ArgumentParser(description="Provision WiFi via Improv Serial")
     parser.add_argument("--port", required=True, help="Serial port device path")
@@ -74,6 +126,15 @@ def main():
     time.sleep(5)
     ser.reset_input_buffer()
 
+    try:
+        if not request_info(ser, timeout=args.timeout):
+            ser.close()
+            return 1
+    except BadChecksum as e:
+        print(f"FAIL: {e}", file=sys.stderr)
+        ser.close()
+        return 1
+
     # Send WiFi credentials
     packet = build_wifi_packet(args.ssid, args.password)
     print(f"Sending WiFi credentials (SSID: {args.ssid})")
@@ -87,7 +148,12 @@ def main():
         if ser.in_waiting:
             buf += ser.read(ser.in_waiting)
             while True:
-                result = parse_improv_packet(buf)
+                try:
+                    result = parse_improv_packet(buf)
+                except BadChecksum as e:
+                    print(f"FAIL: {e}", file=sys.stderr)
+                    ser.close()
+                    return 1
                 if not result:
                     break
                 ptype, state, consumed = result