diff --git a/firmware/esp32-csi-node/provision.py b/firmware/esp32-csi-node/provision.py index 96f407d53..f5ea2243a 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. @@ -46,6 +51,7 @@ import io import json import os import stat +import re import struct import subprocess import sys @@ -291,6 +297,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. @@ -514,6 +628,10 @@ 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. " @@ -537,6 +655,33 @@ def main(): print("WARNING: --password is visible in ps and shell history; " "use --password-file instead.", file=sys.stderr) cli_ssid, cli_password = args.ssid, args.password + # --- 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: @@ -548,7 +693,12 @@ def main(): print(f"--reset: ignoring 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) # A new --ssid with no password given: ask for it on a terminal instead of @@ -585,11 +735,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" ) @@ -691,7 +842,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()