install: stop fleet-converging on the OpenIPC default MAC

src/defib/cli/app.py

@@ -1654,6 +1654,12 @@ def install(
     tftp_port: int = typer.Option(69, "--tftp-port", help="TFTP server port"),
     nor_size: int = typer.Option(8, "--nor-size", help="NOR flash size in MB (8, 16, or 32)"),
     nand: bool = typer.Option(False, "--nand", help="Use NAND flash instead of NOR"),
+    wipe_env: bool = typer.Option(
+        False, "--wipe-env",
+        help="Erase the env partition during U-Boot flash (loses ethaddr; "
+             "default is to preserve env so MACs aren't reset to the OpenIPC "
+             "u-boot default 00:00:23:34:45:66).",
+    ),
     output: str = typer.Option("human", "--output", help="Output mode: human, json"),
     debug: bool = typer.Option(False, "-d", "--debug", help="Enable debug logging"),
 ) -> None:
@@ -1666,7 +1672,7 @@ def install(
     import asyncio
     asyncio.run(_install_async(
         chip, firmware, port, power_cycle, nic, host_ip, device_ip,
-        tftp_port, nor_size, nand, output, debug,
+        tftp_port, nor_size, nand, wipe_env, output, debug,
     ))
 
 
@@ -1738,6 +1744,7 @@ async def _install_async(
     tftp_port: int,
     nor_size: int,
     nand: bool,
+    wipe_env: bool,
     output: str,
     debug: bool,
 ) -> None:
@@ -1868,10 +1875,15 @@ async def _install_async(
     env_off, env_sz = layout["env"]
     # OpenIPC publishes raw U-Boot now (issue #73) — pad locally to the
     # boot partition size so the trailing flash is erased (0xFF), not
-    # left at whatever was previously written. We then erase boot+env
-    # together so the env partition gets cleared in the same operation.
+    # left at whatever was previously written.
     uboot_data = pad_to_size(uboot_raw, b_sz)
-    uboot_flash_size = b_sz + env_sz
+    # Default: erase only the boot partition. Erasing the env partition
+    # destroys any ethaddr that u-boot derived on a previous boot, which
+    # then has the OpenIPC compiled-in default 00:00:23:34:45:66 saved
+    # back in its place at the saveenv at the end of install — that's
+    # how multiple cameras converge on the same MAC.  --wipe-env opts
+    # back into the old behavior when a clean env is wanted.
+    uboot_flash_size = b_sz + env_sz if wipe_env else b_sz
 
     if output == "human":
         if len(uboot_raw) == len(uboot_data):
@@ -2250,6 +2262,37 @@ async def tftp_and_flash(
                     mtdparts_var = f"mtdpartsnor{nor_size}m"
                     await _cmd(f"run {mtdparts_var}", timeout=3.0)
                 await _cmd("setenv bootcmd ${bootcmdnor}", timeout=3.0)
+
+            # Rescue ethaddr before saveenv. OpenIPC u-boot's compiled-in
+            # default env carries ethaddr=00:00:23:34:45:66; if u-boot
+            # loaded that default (because the env partition was empty
+            # or just got erased by --wipe-env), saveenv would persist
+            # the bogus MAC and every fresh camera in a fleet would
+            # converge on it. Replace with a locally-administered random
+            # MAC if we see the default or nothing valid.
+            from defib.uboot_env import (
+                generate_locally_administered_mac,
+                is_unset_or_default_ethaddr,
+                parse_printenv_value,
+            )
+            eth_resp = await _cmd("printenv ethaddr", timeout=5.0)
+            current_eth = parse_printenv_value(eth_resp, "ethaddr")
+            if is_unset_or_default_ethaddr(current_eth):
+                new_mac = generate_locally_administered_mac()
+                if output == "human":
+                    if current_eth:
+                        console.print(
+                            f"  ethaddr was [yellow]{current_eth}[/yellow] "
+                            f"(OpenIPC default) — assigning [cyan]{new_mac}[/cyan]"
+                        )
+                    else:
+                        console.print(
+                            f"  ethaddr unset — assigning [cyan]{new_mac}[/cyan]"
+                        )
+                await _cmd(f"setenv ethaddr {new_mac}", timeout=3.0)
+            elif output == "human":
+                console.print(f"  ethaddr preserved: [cyan]{current_eth}[/cyan]")
+
             resp = await _cmd("saveenv", timeout=10.0)
             if output == "human":
                 console.print("  [green]Environment saved[/green]")

src/defib/uboot_env.py

@@ -0,0 +1,63 @@
+"""Helpers for U-Boot env handling during install.
+
+The OpenIPC u-boot binaries ship with a compiled-in default env that
+contains ``ethaddr=00:00:23:34:45:66``. When a camera boots with an
+empty NAND env partition, u-boot loads that default into RAM. If anyone
+then runs ``saveenv``, the bogus MAC is persisted to flash and from
+then on every boot reads the same MAC. Multiple cameras converging on
+``00:00:23:34:45:66`` is the visible symptom.
+
+The mitigation here: detect the default (or missing) ``ethaddr`` and
+replace it with a locally-administered random MAC before ``saveenv``.
+"""
+
+from __future__ import annotations
+
+import re
+import secrets
+
+# CONFIG_ETHADDR baked into OpenIPC u-boot's default env. Found in the
+# LZMA-compressed payload of u-boot-*-universal.bin (e.g. hi3516av200,
+# hi3516cv300). Cameras whose env partition was empty when u-boot first
+# saw them all converge on this MAC after the first saveenv.
+OPENIPC_DEFAULT_ETHADDR = "00:00:23:34:45:66"
+
+_MAC_RE = re.compile(r"^([0-9a-fA-F]{2}:){5}[0-9a-fA-F]{2}$")
+
+
+def is_unset_or_default_ethaddr(value: str | None) -> bool:
+    """True if `value` is missing, blank, malformed, or the OpenIPC default."""
+    if value is None:
+        return True
+    v = value.strip().lower()
+    if not v:
+        return True
+    if not _MAC_RE.match(v):
+        return True
+    return v == OPENIPC_DEFAULT_ETHADDR.lower()
+
+
+def generate_locally_administered_mac() -> str:
+    """Generate a random unicast, locally-administered MAC.
+
+    First octet has the locally-administered bit (bit 1) set and the
+    multicast bit (bit 0) cleared, per IEEE 802. The remaining five
+    octets are random. Always returns lowercase ``xx:xx:xx:xx:xx:xx``.
+    """
+    raw = bytearray(secrets.token_bytes(6))
+    # Bit 0 (LSB of first octet): 0 = unicast.
+    # Bit 1 (LSB+1):              1 = locally administered.
+    raw[0] = (raw[0] & 0xFC) | 0x02
+    return ":".join(f"{b:02x}" for b in raw)
+
+
+def parse_printenv_value(response: str, var: str) -> str | None:
+    """Pull the value of `var` out of a ``printenv VAR`` response.
+
+    U-Boot prints lines like ``ethaddr=00:00:23:34:45:66`` (no quotes,
+    one var per line). May be preceded/followed by prompt characters or
+    download-mode framing. Returns the value or None if not found.
+    """
+    pattern = re.compile(rf"(?m)^\s*{re.escape(var)}=(.+?)\s*$")
+    m = pattern.search(response)
+    return m.group(1).strip() if m else None

tests/test_uboot_env.py

@@ -0,0 +1,106 @@
+"""Tests for u-boot env helpers (ethaddr default detection + rescue MAC)."""
+
+import re
+
+from defib.uboot_env import (
+    OPENIPC_DEFAULT_ETHADDR,
+    generate_locally_administered_mac,
+    is_unset_or_default_ethaddr,
+    parse_printenv_value,
+)
+
+_MAC_RE = re.compile(r"^([0-9a-f]{2}:){5}[0-9a-f]{2}$")
+
+
+class TestIsUnsetOrDefaultEthaddr:
+    def test_none_is_default(self):
+        assert is_unset_or_default_ethaddr(None) is True
+
+    def test_empty_is_default(self):
+        assert is_unset_or_default_ethaddr("") is True
+        assert is_unset_or_default_ethaddr("   ") is True
+
+    def test_openipc_default(self):
+        assert is_unset_or_default_ethaddr("00:00:23:34:45:66") is True
+
+    def test_openipc_default_uppercase(self):
+        assert is_unset_or_default_ethaddr("00:00:23:34:45:66".upper()) is True
+
+    def test_malformed_is_default(self):
+        assert is_unset_or_default_ethaddr("not-a-mac") is True
+        assert is_unset_or_default_ethaddr("00:00:23:34:45") is True
+        assert is_unset_or_default_ethaddr("zz:zz:zz:zz:zz:zz") is True
+
+    def test_real_macs_not_default(self):
+        for mac in [
+            "00:12:31:5e:e0:d2",   # HiSilicon OUI, real av200 MAC
+            "02:ab:cd:ef:01:23",   # locally-administered
+            "aa:bb:cc:dd:ee:ff",
+        ]:
+            assert is_unset_or_default_ethaddr(mac) is False, mac
+
+
+class TestGenerateLocallyAdministeredMac:
+    def test_format(self):
+        mac = generate_locally_administered_mac()
+        assert _MAC_RE.match(mac), f"bad format: {mac}"
+
+    def test_locally_administered_bit_set(self):
+        # Run many to be sure the bit-twiddle is correct regardless of randomness.
+        for _ in range(200):
+            mac = generate_locally_administered_mac()
+            first = int(mac.split(":")[0], 16)
+            assert first & 0x02, f"locally-administered bit not set in {mac}"
+
+    def test_unicast_bit_clear(self):
+        for _ in range(200):
+            mac = generate_locally_administered_mac()
+            first = int(mac.split(":")[0], 16)
+            assert (first & 0x01) == 0, f"multicast bit set in {mac}"
+
+    def test_not_default(self):
+        # Should never collide with the OpenIPC default (locally-administered
+        # bit makes that physically impossible — 00:00:23 has bit 1 == 0).
+        for _ in range(200):
+            assert generate_locally_administered_mac() != OPENIPC_DEFAULT_ETHADDR
+
+    def test_uniqueness(self):
+        macs = {generate_locally_administered_mac() for _ in range(100)}
+        # 100 random MACs out of 2^46 possible → birthday collision negligible.
+        assert len(macs) == 100
+
+
+class TestParsePrintenvValue:
+    def test_simple(self):
+        assert parse_printenv_value("ethaddr=00:11:22:33:44:55\n", "ethaddr") == "00:11:22:33:44:55"
+
+    def test_default_value(self):
+        assert (
+            parse_printenv_value("ethaddr=00:00:23:34:45:66\n", "ethaddr")
+            == "00:00:23:34:45:66"
+        )
+
+    def test_with_prompt_around(self):
+        resp = "hisilicon # printenv ethaddr\nethaddr=02:aa:bb:cc:dd:ee\nhisilicon # "
+        assert parse_printenv_value(resp, "ethaddr") == "02:aa:bb:cc:dd:ee"
+
+    def test_missing(self):
+        # U-Boot reports "## Error: ..." for unset vars
+        assert parse_printenv_value("## Error: \"ethaddr\" not defined\n", "ethaddr") is None
+
+    def test_doesnt_match_substring(self):
+        # 'eth' shouldn't match 'ethaddr' line
+        assert parse_printenv_value("ethaddr=00:11:22:33:44:55\n", "eth") is None
+
+    def test_multiple_vars(self):
+        resp = "bootcmd=run abc\nethaddr=02:aa:bb:cc:dd:ee\nipaddr=192.168.1.10\n"
+        assert parse_printenv_value(resp, "ethaddr") == "02:aa:bb:cc:dd:ee"
+        assert parse_printenv_value(resp, "ipaddr") == "192.168.1.10"
+        assert parse_printenv_value(resp, "bootcmd") == "run abc"
+
+
+def test_default_const_is_what_we_observed():
+    # Pin the constant to what we actually decompressed out of OpenIPC u-boot
+    # binaries (hi3516av200 + hi3516cv300). If OpenIPC ever changes the
+    # baked-in default, this test breaks loudly so we know to update.
+    assert OPENIPC_DEFAULT_ETHADDR == "00:00:23:34:45:66"