From 98c1d4a1e979f5e689be1491f174ee5447a78694 Mon Sep 17 00:00:00 2001 From: Mathew Beane Date: Sun, 4 Oct 2026 16:28:47 -0400 Subject: [PATCH] fix(provision): key saved state by board MAC, not serial port provision.py saved each board's settings under the serial port path. Ports get reused when boards are swapped (a second board on the same hub slot often comes up as the same /dev/cu.usbmodem*), so the next board inherited the previous board's node_id, target_port and channel without any sign (#1755). - On a real flash the script reads the base MAC with `esptool read_mac` and keeps state in mac-.json. The record also stores the MAC and the port it was last written from. - EUI-64 chips (C5, C6, H2) print an 8-byte "MAC:" line before "BASE MAC:"; the parser takes the base MAC on those and the single MAC line elsewhere. - Migration: if the board has no record yet and a port-keyed file from an earlier version exists, it is used once, with a notice that points at --reset, then removed after the save so no other board can pick it up. - --state and --dry-run still never open the port. --state shows the board last provisioned from that port, and --mac selects a board by hand. --dry-run without --mac keeps the port-keyed file, which the QEMU swarm test relies on (--port /dev/null --dry-run). - If the MAC can't be read, state stays keyed by port and the script says so. Tests: tests/test_provision_mac_state.py adds 18 tests, using placeholder MACs, with esptool, the NVS generator and the flasher stubbed. 15 of them fail on the base; the other 3 guard behaviour the base already has (no serial access on --dry-run, --reset, rejecting a bad --mac). The suite goes from 14 to 32 passing. Not verified on hardware: the real esptool read_mac call. The output shapes in the tests come from esptool 4.12 and 5.4.0 cmds.py. Refs #1755 Co-Authored-By: claude-flow --- firmware/esp32-csi-node/provision.py | 160 +++++++++++- .../tests/test_provision_mac_state.py | 246 ++++++++++++++++++ 2 files changed, 402 insertions(+), 4 deletions(-) create mode 100644 firmware/esp32-csi-node/tests/test_provision_mac_state.py diff --git a/firmware/esp32-csi-node/provision.py b/firmware/esp32-csi-node/provision.py index 88172fe4e..146b1c963 100644 --- a/firmware/esp32-csi-node/provision.py +++ b/firmware/esp32-csi-node/provision.py @@ -28,6 +28,11 @@ ADDITIVE-BY-DEFAULT (issue #391, #574 phase 1): 3. Generate + flash NVS from the merged state. 4. Write the merged state back to the state file. + State is keyed by the board's MAC, read with esptool, so a board plugged + into a port another board used doesn't inherit its node_id (#1755). + Port-keyed files from earlier versions move to the board's record the + first time that board is provisioned. + Net effect: partial reconfigure works the way users expect. Pass `--reset` to wipe both the state file AND the device NVS for first-time provisioning of a recycled board. @@ -44,6 +49,7 @@ import csv import io import json import os +import re import struct import subprocess import sys @@ -157,6 +163,114 @@ def save_state(port: str, state_dir: str, state: dict) -> str: return path +# --------------------------------------------------------------------------- +# Board identity (#1755) +# --------------------------------------------------------------------------- +# +# Serial port names get reused when boards are swapped, so a port-keyed state +# file hands one board's node_id to the next. State is keyed by the chip's +# base MAC instead. Each record also stores the MAC and the port it was last +# written from, so --state can find it without opening the port. + +STATE_MAC_KEY = "_chip_mac" +STATE_PORT_KEY = "_port" + +_ESPTOOL_MAC_RE = re.compile( + r"^\s*(BASE MAC|MAC):\s*((?:[0-9a-fA-F]{2}:){5}[0-9a-fA-F]{2})\s*$", re.MULTILINE +) + + +def normalize_mac(value: str): + """Return `value` as lowercase aa:bb:cc:dd:ee:ff, or None if it isn't a MAC.""" + parts = re.split(r"[:-]", value.strip()) + if len(parts) != 6 or not all(re.fullmatch(r"[0-9a-fA-F]{2}", p) for p in parts): + return None + return ":".join(p.lower() for p in parts) + + +def parse_esptool_mac(output: str): + """Return the base MAC from `esptool read_mac` output, or None. + + Most chips print one 6-byte "MAC:" line. EUI-64 chips (C5, C6, H2) print an + 8-byte "MAC:" line, then "BASE MAC:" with the 6-byte address. + """ + found = {label: mac.lower() for label, mac in _ESPTOOL_MAC_RE.findall(output)} + return found.get("BASE MAC") or found.get("MAC") + + +def read_chip_mac(port: str, baud: int, chip: str): + """Read the connected board's base MAC with esptool. None if it can't.""" + cmd = [ + sys.executable, "-m", "esptool", + "--chip", chip, + "--port", port, + "--baud", str(baud), + "read_mac", + ] + try: + result = subprocess.run(cmd, capture_output=True, text=True, timeout=30) + except (OSError, subprocess.SubprocessError): + return None + if result.returncode != 0: + return None + return parse_esptool_mac(result.stdout) + + +def mac_state_key(mac: str) -> str: + """State-file key for a board, e.g. mac-aabbccddeeff.""" + return "mac-" + mac.replace(":", "") + + +def boards_last_on_port(port: str, state_dir: str) -> list: + """MACs of boards whose state was last written from `port`, newest first.""" + if not os.path.isdir(state_dir): + return [] + found = [] + for name in os.listdir(state_dir): + if not (name.startswith("mac-") and name.endswith(".json")): + continue + data = load_state(name[:-len(".json")], state_dir) + mac = data.get(STATE_MAC_KEY) + if data.get(STATE_PORT_KEY) == port and mac: + found.append((os.path.getmtime(os.path.join(state_dir, name)), mac)) + return [mac for _, mac in sorted(found, reverse=True)] + + +def load_board_state(port: str, mac, state_dir: str): + """Return (prior state, legacy path) for the board `mac` on `port`. + + Without a MAC this is the old port-keyed lookup. With one, the board's own + record wins. If it has none but a port-keyed file from an earlier version + exists, that file is returned for migration along with its path. + """ + if mac is None: + return load_state(port, state_dir), None + prior = load_state(mac_state_key(mac), state_dir) + if prior: + return prior, None + legacy = _state_path_for(port, state_dir) + if os.path.isfile(legacy): + return load_state(port, state_dir), legacy + return {}, None + + +def save_board_state(port: str, mac, state_dir: str, state: dict, legacy=None) -> str: + """Persist `state` under the board's MAC (or the port if the MAC is unknown). + + A migrated port-keyed file is removed afterwards, so no other board on that + port can pick it up. + """ + if mac is None: + return save_state(port, state_dir, state) + state = dict(state) + state[STATE_MAC_KEY] = mac + state[STATE_PORT_KEY] = port + path = save_state(mac_state_key(mac), state_dir, state) + if legacy and os.path.isfile(legacy): + os.unlink(legacy) + return path + + def merge_state_into_args(args, prior: dict) -> dict: """Overlay `args` onto `prior` for every MERGEABLE_ATTRS attribute. @@ -365,12 +479,44 @@ def main(): "previously-staged keys should NOT be re-applied.") parser.add_argument("--state-dir", default=_default_state_dir(), help="Override the per-user state directory (default: per-OS user config dir).") + parser.add_argument("--mac", type=str, + help="Board MAC (AA:BB:CC:DD:EE:FF) whose state to use. Normally read " + "from the chip with esptool; --state and --dry-run never open the " + "port, so pass it there to pick a board.") parser.add_argument("--state", action="store_true", help="Print the merged state that WOULD be flashed for this port and exit. " "Useful for debugging which keys are about to land on the device.") args = parser.parse_args() + # --- Board identity (#1755) --- + # State is keyed by the chip's MAC so a board swapped onto a reused port + # doesn't inherit the previous board's node_id. + if args.mac is not None: + chip_mac = normalize_mac(args.mac) + if chip_mac is None: + parser.error(f"--mac must be in AA:BB:CC:DD:EE:FF format, got '{args.mac}'") + elif args.state: + # Inspection only: don't open the port; show the board last + # provisioned from it. + recent = boards_last_on_port(args.port, args.state_dir) + chip_mac = recent[0] if recent else None + if chip_mac: + print(f"Showing board {chip_mac}, last provisioned on {args.port}. " + f"Pass --mac to pick another.", file=sys.stderr) + elif args.dry_run: + # No board involved; keep the port-keyed state as before. + chip_mac = None + else: + chip_mac = read_chip_mac(args.port, args.baud, args.chip) + if chip_mac: + print(f"Board MAC: {chip_mac}") + else: + print(f"WARNING: could not read the board's MAC with esptool, so state " + f"stays keyed by port {args.port}. Pass --mac to key it by board.", + file=sys.stderr) + legacy = None + # --- Per-port state load + merge (additive-by-default, #391 / #574) --- if args.reset: path = _state_path_for(args.port, args.state_dir) @@ -379,7 +525,12 @@ def main(): print(f"--reset: removed state file {path}", file=sys.stderr) prior = {} else: - prior = load_state(args.port, args.state_dir) + prior, legacy = load_board_state(args.port, chip_mac, args.state_dir) + if legacy: + print(f"Using port-keyed state {legacy} for board {chip_mac}; it moves to " + f"the board's own record when state is next saved. If this isn't the " + f"board last provisioned on {args.port}, rerun with --reset.", + file=sys.stderr) merged = merge_state_into_args(args, prior) if args.state: @@ -403,11 +554,12 @@ def main(): ] if val is None or val == "" ] if wifi_trio_missing and not args.force_partial: + state_key = mac_state_key(chip_mac) if chip_mac else args.port parser.error( f"Missing required WiFi credentials after merging prior state: " f"{', '.join(wifi_trio_missing)}.\n" f"\n" - f" No per-port state file at {_state_path_for(args.port, args.state_dir)}\n" + f" No saved state at {_state_path_for(state_key, args.state_dir)}\n" f" and the CLI didn't include them. Either pass --ssid + --password + --target-ip\n" f" on this run, or add --force-partial to flash without WiFi.\n" ) @@ -504,7 +656,7 @@ def main(): f"write_flash 0x9000 {out}") # Persist merged state even on dry-run so a subsequent real flash from # this machine sees the same staged config. - path = save_state(args.port, args.state_dir, merged) + path = save_board_state(args.port, chip_mac, args.state_dir, merged, legacy) print(f"State persisted to {path}") return @@ -512,7 +664,7 @@ def main(): # Persist merged state after a successful flash so future partial # invocations from this machine merge on top of what's actually on the # device. This is the heart of the additive-by-default fix (#391/#574). - path = save_state(args.port, args.state_dir, merged) + path = save_board_state(args.port, chip_mac, args.state_dir, merged, legacy) print(f"State persisted to {path}") diff --git a/firmware/esp32-csi-node/tests/test_provision_mac_state.py b/firmware/esp32-csi-node/tests/test_provision_mac_state.py new file mode 100644 index 000000000..bc8880cb9 --- /dev/null +++ b/firmware/esp32-csi-node/tests/test_provision_mac_state.py @@ -0,0 +1,246 @@ +"""Tests for keying provision.py state by board MAC instead of serial port (#1755). + +main() runs in-process. esptool's MAC read, the NVS generator and the flasher +are all stubbed, so no serial port is opened and nothing is flashed. MACs and +credentials are placeholders. +""" + +import contextlib +import csv +import importlib.util +import io +import json +import os +import shutil +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +PROVISION_PATH = Path(__file__).resolve().parents[1] / "provision.py" +SPEC = importlib.util.spec_from_file_location("provision", PROVISION_PATH) +provision = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(provision) + +PORT = "/dev/cu.usbmodem-test" +OTHER_PORT = "/dev/cu.usbmodem-other" +MAC_A = "02:00:00:00:00:0a" +MAC_B = "02:00:00:00:00:0b" +CREDS = ("--ssid", "test-ssid", "--password", "fake-pass-not-real", + "--target-ip", "192.0.2.10") + +# Shapes of `esptool read_mac` output (esptool/cmds.py read_mac in 4.x and 5.x). +ESPTOOL_V5_S3 = "Connected to ESP32-S3 on /dev/x:\nMAC: 02:00:00:00:00:0a\n" +ESPTOOL_V4_S3 = "Chip is ESP32-S3\nMAC: 02:00:00:00:00:0a\nHard resetting via RTS pin...\n" +ESPTOOL_V5_C6 = ( + "MAC: 02:00:00:ff:fe:00:00:0a\n" + "BASE MAC: 02:00:00:00:00:0a\n" + "MAC_EXT: ff:fe\n" +) + + +class TestParseEsptoolMac(unittest.TestCase): + def test_six_byte_mac_line(self): + self.assertEqual(provision.parse_esptool_mac(ESPTOOL_V5_S3), MAC_A) + self.assertEqual(provision.parse_esptool_mac(ESPTOOL_V4_S3), MAC_A) + + def test_eui64_chips_use_the_base_mac(self): + # C5/C6 print an 8-byte "MAC:" line first; that isn't the base MAC. + self.assertEqual(provision.parse_esptool_mac(ESPTOOL_V5_C6), MAC_A) + + def test_no_mac_in_output(self): + self.assertIsNone(provision.parse_esptool_mac("A fatal error occurred\n")) + + def test_normalize_mac(self): + self.assertEqual(provision.normalize_mac("02-00-00-00-00-0A"), MAC_A) + self.assertIsNone(provision.normalize_mac("02:00:00:00:00")) + self.assertIsNone(provision.normalize_mac("02:00:00:00:00:zz")) + + +class TestReadChipMac(unittest.TestCase): + def test_runs_esptool_read_mac_on_the_port(self): + done = subprocess.CompletedProcess([], 0, stdout=ESPTOOL_V5_C6, stderr="") + with mock.patch.object(provision.subprocess, "run", return_value=done) as run: + self.assertEqual(provision.read_chip_mac(PORT, 115200, "auto"), MAC_A) + cmd = run.call_args.args[0] + self.assertIn("read_mac", cmd) + self.assertEqual(cmd[cmd.index("--port") + 1], PORT) + + def test_failure_returns_none(self): + failed = subprocess.CompletedProcess([], 2, stdout="", stderr="no port") + with mock.patch.object(provision.subprocess, "run", return_value=failed): + self.assertIsNone(provision.read_chip_mac(PORT, 115200, "auto")) + with mock.patch.object(provision.subprocess, "run", + side_effect=subprocess.TimeoutExpired("esptool", 30)): + self.assertIsNone(provision.read_chip_mac(PORT, 115200, "auto")) + + +class _MainCase(unittest.TestCase): + def setUp(self): + self.root = tempfile.mkdtemp(prefix="provision-mac-") + self.state_dir = os.path.join(self.root, "state") + self.flashed_csv = [] + + def tearDown(self): + shutil.rmtree(self.root, ignore_errors=True) + + def run_main(self, *argv, mac=None): + """Run main() with the board `mac` attached; return (code, stdout, stderr).""" + def fake_generate(csv_content, _size): + self.flashed_csv.append(csv_content) + return b"nvs" + + out, err = io.StringIO(), io.StringIO() + code = 0 + old_cwd = os.getcwd() + os.chdir(self.root) + try: + with mock.patch.object(sys, "argv", ["provision.py", *argv, + "--state-dir", self.state_dir]), \ + mock.patch.object(provision, "read_chip_mac", return_value=mac, + create=True) as read, \ + mock.patch.object(provision, "generate_nvs_binary", + side_effect=fake_generate), \ + mock.patch.object(provision, "flash_nvs"), \ + contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + self.read_mac = read + try: + provision.main() + except SystemExit as exc: + code = exc.code + finally: + os.chdir(old_cwd) + return code, out.getvalue(), err.getvalue() + + def last_flashed(self): + rows = csv.DictReader(io.StringIO(self.flashed_csv[-1])) + return {row["key"]: row["value"] for row in rows} + + def board_file(self, mac): + return provision._state_path_for(provision.mac_state_key(mac), self.state_dir) + + +class TestBoardKeyedState(_MainCase): + def test_second_board_on_same_port_does_not_inherit_node_id(self): + self.assertEqual(self.run_main("--port", PORT, *CREDS, "--node-id", "2", mac=MAC_A)[0], 0) + + code, _, _ = self.run_main("--port", PORT, *CREDS, mac=MAC_B) + + self.assertEqual(code, 0) + self.assertNotIn("node_id", self.last_flashed()) + + def test_second_board_without_credentials_is_not_given_the_first_boards(self): + self.run_main("--port", PORT, *CREDS, "--node-id", "2", mac=MAC_A) + + code, _, err = self.run_main("--port", PORT, "--node-id", "3", mac=MAC_B) + + self.assertEqual(code, 2) + self.assertIn("Missing required WiFi credentials", err) + + def test_board_keeps_its_state_on_a_different_port(self): + self.run_main("--port", PORT, *CREDS, "--node-id", "2", mac=MAC_A) + + code, _, _ = self.run_main("--port", OTHER_PORT, "--zone", "lab", mac=MAC_A) + + self.assertEqual(code, 0) + flashed = self.last_flashed() + self.assertEqual(flashed["node_id"], "2") + self.assertEqual(flashed["ssid"], "test-ssid") + + def test_record_carries_board_identity(self): + self.run_main("--port", PORT, *CREDS, mac=MAC_A) + + with open(self.board_file(MAC_A), encoding="utf-8") as f: + saved = json.load(f) + self.assertEqual(saved[provision.STATE_MAC_KEY], MAC_A) + self.assertEqual(saved[provision.STATE_PORT_KEY], PORT) + self.assertNotIn("_chip_mac", self.flashed_csv[-1]) + + def test_unreadable_mac_falls_back_to_port_keyed_state(self): + code, _, err = self.run_main("--port", PORT, *CREDS, mac=None) + + self.assertEqual(code, 0) + self.assertIn("could not read the board's MAC", err) + self.assertTrue(os.path.isfile(provision._state_path_for(PORT, self.state_dir))) + + def test_reset_ignores_the_boards_record(self): + self.run_main("--port", PORT, *CREDS, "--node-id", "2", mac=MAC_A) + + self.run_main("--port", PORT, "--reset", *CREDS, mac=MAC_A) + + self.assertNotIn("node_id", self.last_flashed()) + + def test_invalid_mac_flag_is_rejected(self): + code, _, err = self.run_main("--port", PORT, "--mac", "not-a-mac", "--state") + self.assertEqual(code, 2) + self.assertIn("--mac", err) + + +class TestMigration(_MainCase): + def write_legacy(self, port, state): + os.makedirs(self.state_dir, exist_ok=True) + path = provision._state_path_for(port, self.state_dir) + with open(path, "w", encoding="utf-8") as f: + json.dump(state, f) + return path + + def test_port_keyed_record_moves_to_the_board_and_is_retired(self): + legacy = self.write_legacy(PORT, { + "ssid": "test-ssid", "password": "fake-pass-not-real", + "target_ip": "192.0.2.10", "node_id": 5, + }) + + code, _, err = self.run_main("--port", PORT, "--zone", "lab", mac=MAC_A) + + self.assertEqual(code, 0) + self.assertIn("port-keyed state", err) + self.assertEqual(self.last_flashed()["node_id"], "5") + self.assertFalse(os.path.exists(legacy)) + with open(self.board_file(MAC_A), encoding="utf-8") as f: + self.assertEqual(json.load(f)["node_id"], 5) + + # The next board on that port starts clean. + self.run_main("--port", PORT, *CREDS, mac=MAC_B) + self.assertNotIn("node_id", self.last_flashed()) + + def test_board_record_wins_over_a_port_keyed_file(self): + self.run_main("--port", OTHER_PORT, *CREDS, "--node-id", "2", mac=MAC_A) + self.write_legacy(PORT, {"node_id": 9}) + + code, _, _ = self.run_main("--port", PORT, "--zone", "lab", mac=MAC_A) + + self.assertEqual(code, 0) + self.assertEqual(self.last_flashed()["node_id"], "2") + + +class TestNoSerialForInspection(_MainCase): + def test_state_shows_last_board_on_port_without_opening_it(self): + self.run_main("--port", PORT, *CREDS, "--node-id", "2", mac=MAC_A) + + code, out, err = self.run_main("--port", PORT, "--state", mac=MAC_B) + + self.assertEqual(code, 0) + self.read_mac.assert_not_called() + self.assertEqual(json.loads(out)["node_id"], 2) + self.assertIn(MAC_A, err) + + def test_state_with_mac_flag_selects_that_board(self): + self.run_main("--port", PORT, *CREDS, "--node-id", "2", mac=MAC_A) + self.run_main("--port", PORT, *CREDS, "--node-id", "3", mac=MAC_B) + + code, out, _ = self.run_main("--port", PORT, "--state", "--mac", MAC_A.upper()) + + self.assertEqual(code, 0) + self.assertEqual(json.loads(out)["node_id"], 2) + + def test_dry_run_does_not_open_the_port(self): + code, _, _ = self.run_main("--port", PORT, *CREDS, "--dry-run", mac=MAC_A) + + self.assertEqual(code, 0) + self.read_mac.assert_not_called() + + +if __name__ == "__main__": + unittest.main()