"""lan7800prog — LAN7800 EEPROM programmer CLI (Core v1, LAN7800 ONLY). Design contract (Adom AI-oriented CLI): * Every command prints a human line: `OK: ...`, or `ERROR: ...` + `Hint: ...`. * `--json` additionally emits one machine-readable object on stdout. * Reads are FREE. The ONLY commands that touch a physical adapter are `program` and `erase`; both refuse unless `--force-physical-write` is given. That flag is the safety catch — the operator/agent must have a cleared `mesh escalate` PROCEED before setting it. The webapp never sets it without an escalation. Field edits produce a .bin image (file->file). Hardware programming is a separate, explicit, gated step (load base -> edit -> preview .bin -> program). """ from __future__ import annotations import argparse import json import sys from .eeprom import (Eeprom, EepromError, parse_u16, LAN7800_VID, LAN7800_PID, SIG_VALUE, SIZES, MAX_EEPROM_SIZE) from .backends import (BinFileBackend, EthtoolBackend, UsbEepromBackend, BackendError, udev_rule, udev_install_commands) from .scan import scan_devices, diagnose_netdev from . import regs # ---- output helpers ------------------------------------------------------- def _emit(args, ok, payload=None, human=None, hint=None): if getattr(args, "json", False): obj = {"ok": ok} if payload: obj.update(payload) if not ok and human: obj["error"] = human if hint: obj["hint"] = hint print(json.dumps(obj)) else: if ok: if human: print(f"OK: {human}") else: print(f"ERROR: {human}", file=sys.stderr) if hint: print(f"Hint: {hint}", file=sys.stderr) return 0 if ok else 1 def _fail(args, human, hint=None): return _emit(args, False, human=human, hint=hint) # ---- image sourcing ------------------------------------------------------- def _load_image_from_in(args) -> Eeprom: if not getattr(args, "in_file", None): raise EepromError("this command needs --in ") data = BinFileBackend(args.in_file).read_image() return Eeprom.from_bytes(data) def _read_device_image(iface: str, sudo: bool) -> Eeprom: data = EthtoolBackend(iface, sudo=sudo).read_image() return Eeprom.from_bytes(data) def _describe(eep: Eeprom) -> dict: descs = [] lan7800_seen = False for d in eep.descriptors(): if d.present and d.vid == LAN7800_VID and d.pid == LAN7800_PID: lan7800_seen = True descs.append({ "kind": d.kind, "present": d.present, "looks_valid": d.looks_valid, "ptr_word": d.ptr_word, "byte_offset": d.byte_offset, "vid": (f"0x{d.vid:04X}" if d.vid is not None else None), "pid": (f"0x{d.pid:04X}" if d.pid is not None else None), "bcd_device": (f"0x{d.bcd_device:04X}" if d.bcd_device is not None else None), }) strings = [{"name": s.name, "present": s.present, "value": s.value, "byte_offset": s.byte_offset} for s in eep.strings()] return { "size": eep.size, "signature": f"0x{eep.signature:02X}", "signature_ok": eep.has_signature, "mac": eep.mac, "descriptors": descs, "strings": strings, "config_flags": [f"0x{v:08X}" for v in eep.config_flags()], "lan7800_identity": lan7800_seen, "regs": regs.describe_regs(eep), } # ---- commands ------------------------------------------------------------- def cmd_scan(args): """Enumerate attached LAN7800 adapters (FREE).""" devices = scan_devices(probe_eeprom=not args.no_probe) if not getattr(args, "json", False): if not devices: print("OK: no LAN7800 (0x0424/0x7800) adapters found") return 0 print(f"OK: {len(devices)} LAN7800 adapter(s):") for d in devices: iface = d["iface"] or "(no netdev bound)" print(f" usb {d['usb_path']} (bus {d['busnum']}/dev {d['devnum']}) " f"{iface} driver={d['driver'] or '-'}") print(f" MAC={d['mac'] or '?'} EEPROM={d['eeprom']} " f"product={d['product'] or '?'}") return 0 return _emit(args, True, {"count": len(devices), "devices": devices}) def cmd_read_device(args): """Read the CURRENTLY-CONNECTED LAN7800, auto-picking the backend. netdev present -> ethtool; else -> libusb (driver-independent). FREE read. """ devices = scan_devices(probe_eeprom=False) if not devices: return _fail(args, "no LAN7800 (0x0424/0x7800) adapter connected", hint="plug in the adapter, or run `scan`") dev = devices[0] data = None backend_used = None errors = [] # libusb is PRIMARY: no sudo, permissioned via the plugdev udev rule, and it # works even when this flapping adapter has no stable lan78xx netdev. ethtool # is only a fallback, tried when libusb is unavailable AND a netdev exists; # its "needs sudo" failure is NOT surfaced as the headline error when libusb # is the real path. try: backend_used = "libusb" be = UsbEepromBackend(size=args.size) try: data = be.read_image() finally: be.close() except BackendError as e: errors.append(f"libusb: {e}") data = None if data is None and dev.get("iface"): try: backend_used = f"ethtool:{dev['iface']}" data = EthtoolBackend(dev["iface"], sudo=args.sudo).read_image() except BackendError as e: errors.append(f"ethtool: {e}") data = None if data is None: payload = {"device": dev, "errors": errors, "udev_rule": udev_rule(), "udev_install_commands": udev_install_commands(), "netdev_diagnosis": diagnose_netdev(dev["usb_path"])} return _emit(args, False, payload, human="could not read the connected device (libusb primary)", hint="libusb needs the plugdev udev rule " "(udev_install_commands) then a replug; no sudo/root " "required. ethtool is only a fallback and needs a " "netdev + passwordless sudo.") eep = Eeprom.from_bytes(data) view = _describe(eep) view["backend"] = backend_used view["device"] = dev view["raw_head_hex"] = data[:64].hex() if not dev.get("iface"): view["netdev_diagnosis"] = diagnose_netdev(dev["usb_path"]) if args.out: BinFileBackend(args.out).write_image(data) view["out"] = args.out if not getattr(args, "json", False): print(f"OK: read {len(data)}B from {dev['usb_path']} via {backend_used}" f" sig={view['signature']} " f"({'valid' if view['signature_ok'] else 'blank/invalid'})") print(f" MAC: {view['mac']}") return 0 return _emit(args, True, view) def cmd_udev_rule(args): return _emit(args, True, {"path": "/etc/udev/rules.d/99-lan7800.rules", "rule": udev_rule(), "install_commands": udev_install_commands()}, human="udev rule for libusb access to 0x0424/0x7800 (run the " "install_commands as the human, then replug)") def cmd_new(args): size = args.size if size not in SIZES: return _fail(args, f"size must be one of {SIZES}") if args.blank: eep = Eeprom.blank(size) if args.sign: eep.set_signature() kind = "blank" + (" signed" if args.sign else "") else: eep = Eeprom.default_template(size) # signed default LAN7800 template kind = "default LAN7800 template" BinFileBackend(args.out).write_image(eep.to_bytes()) return _emit(args, True, {"out": args.out, "size": size, "kind": kind, "preview": _describe(eep)}, human=f"wrote {kind} {size}-byte image to {args.out}") def cmd_info(args): try: if args.iface: eep = _read_device_image(args.iface, args.sudo) source = f"iface:{args.iface}" else: eep = _load_image_from_in(args) source = f"file:{args.in_file}" except (EepromError, BackendError) as e: return _fail(args, str(e), hint="give --in or --iface ") view = _describe(eep) view["source"] = source if not getattr(args, "json", False): sig_note = ("valid" if view["signature_ok"] else "INVALID — device falls back to OTP/defaults") print(f"OK: {source} size={view['size']} sig={view['signature']} " f"({sig_note})") print(f" MAC: {view['mac']}") for d in view["descriptors"]: if d["present"]: print(f" {d['kind'].upper()} desc @0x{d['byte_offset']:X}: " f"VID={d['vid']} PID={d['pid']} bcdDevice={d['bcd_device']}" f"{'' if d['looks_valid'] else ' (!! not a std device desc)'}") else: print(f" {d['kind'].upper()} desc: absent") for s in view["strings"]: if s["present"]: print(f" str[{s['name']}]: {s['value']!r}") if not view["lan7800_identity"]: print(" note: no descriptor advertises VID/PID 0x0424/0x7800 — " "confirm this is a LAN7800 image (this tool is LAN7800-only)") return 0 return _emit(args, True, view) def _apply_edits(eep: Eeprom, args) -> list[str]: changes = [] if args.mac is not None: eep.set_mac(args.mac) changes.append(f"mac={eep.mac}") if args.vid is not None: v = parse_u16(args.vid) touched = eep.set_vid(v) changes.append(f"vid=0x{v:04X}({'+'.join(touched)})") if args.pid is not None: p = parse_u16(args.pid) touched = eep.set_pid(p) changes.append(f"pid=0x{p:04X}({'+'.join(touched)})") if args.bcd is not None: b = parse_u16(args.bcd) touched = eep.set_bcd_device(b) changes.append(f"bcdDevice=0x{b:04X}({'+'.join(touched)})") if args.serial is not None: eep.set_string_inplace("serial", args.serial) changes.append(f"serial={args.serial!r}") for pair in (args.set_string or []): if "=" not in pair: raise EepromError(f"--set-string needs name=value, got {pair!r}") name, value = pair.split("=", 1) eep.set_string_inplace(name.strip(), value) changes.append(f"str[{name.strip()}]={value!r}") for spec in (args.set_led or []): # index=mode idx, _, mode = spec.partition("=") regs.set_led_mode(eep, int(idx), int(mode, 0)) changes.append(f"led{int(idx)}.mode={int(mode, 0)}") for spec in (args.led_enable or []): # index=1|0 idx, _, on = spec.partition("=") regs.set_led_enable(eep, int(idx), on.strip() in ("1", "on", "true")) changes.append(f"led{int(idx)}.enable={on.strip()}") if args.led_blink is not None: regs.set_led_blink(eep, int(args.led_blink)) changes.append(f"led.blink={args.led_blink}") for spec in (args.set_gpio or []): # index=dir,drive,out idx, _, rest = spec.partition("=") parts = [p.strip() for p in rest.split(",")] kw = {} if len(parts) > 0 and parts[0]: kw["direction"] = parts[0] if len(parts) > 1 and parts[1]: kw["drive"] = parts[1] if len(parts) > 2 and parts[2] != "": kw["output_value"] = int(parts[2]) regs.set_gpio(eep, int(idx), **kw) changes.append(f"gpio{int(idx)}={rest}") for spec in (args.gpio_wake or []): # index=en,pol idx, _, rest = spec.partition("=") en, _, pol = rest.partition(",") regs.set_gpio(eep, int(idx), wake_enabled=en.strip() in ("1", "on", "true"), wake_active_high=(pol.strip() in ("1", "on", "true")) if pol else None) changes.append(f"gpio{int(idx)}.wake={rest}") for spec in (args.set_configflag or []): # index=hexval e.g. 0=0x00000001 idx, _, val = spec.partition("=") eep.set_config_flag(int(idx), int(val, 0)) changes.append(f"configflag{int(idx)}={int(val, 0):#010x}") for spec in (args.set_raw or []): # offset=hexbytes e.g. 0x48=0103 off, _, hexb = spec.partition("=") regs.set_region(eep, int(off, 0), bytes.fromhex(hexb.strip())) changes.append(f"raw[{off}]={hexb.strip()}") if args.ensure_signature: eep.set_signature() changes.append("signature=0xA5") return changes def cmd_edit(args): try: eep = _load_image_from_in(args) if args.in_file else Eeprom.blank(args.size) changes = _apply_edits(eep, args) except EepromError as e: return _fail(args, str(e), hint="VID/PID/strings need a base --in image that already " "has descriptor blocks; MAC/signature work on blanks too") if not changes: return _fail(args, "no edits requested", hint="pass --mac/--vid/--pid/--bcd/--serial/--set-string/" "--ensure-signature") out = args.out or args.in_file if not out: return _fail(args, "nowhere to write", hint="give --out ") BinFileBackend(out).write_image(eep.to_bytes()) payload = {"out": out, "changes": changes, "preview": _describe(eep)} return _emit(args, True, payload, human=f"applied [{', '.join(changes)}] -> {out}") def cmd_read(args): """Read a device EEPROM to a .bin (FREE).""" try: data = EthtoolBackend(args.iface, sudo=args.sudo).read_image() except BackendError as e: return _fail(args, str(e)) BinFileBackend(args.out).write_image(data) eep = Eeprom.from_bytes(data) return _emit(args, True, {"out": args.out, "bytes": len(data), "preview": _describe(eep)}, human=f"read {len(data)} bytes from {args.iface} -> {args.out}") def _read_connected_image(args) -> bytes: """Read the connected device: libusb by default, ethtool if --iface given.""" if getattr(args, "iface", None): return EthtoolBackend(args.iface, sudo=args.sudo).read_image() be = UsbEepromBackend() try: return be.read_image() finally: be.close() def cmd_verify(args): """Read device and compare against an expected image (FREE).""" try: expected = BinFileBackend(args.image).read_image() actual = _read_connected_image(args) except BackendError as e: return _fail(args, str(e)) n = min(len(expected), len(actual)) diffs = [{"offset": i, "expected": expected[i], "actual": actual[i]} for i in range(n) if expected[i] != actual[i]] if len(expected) != len(actual): diffs.append({"offset": None, "note": f"length differs expected={len(expected)} actual={len(actual)}"}) ok = not diffs return _emit(args, ok, {"match": ok, "diff_count": len(diffs), "diffs": diffs[:32]}, human=("device matches image" if ok else f"{len(diffs)} byte(s) differ"), hint=None if ok else "re-program or investigate the mismatched offsets") # ---- GATED hardware writers ---------------------------------------------- def _refuse_physical(args, action): tgt = getattr(args, "iface", None) or "the connected 0424:7800 (libusb)" return _emit( args, False, {"gated": True, "action": action}, human=f"REFUSED: {action} writes to {tgt}", hint="physical writes are gated. The operator/agent must clear a " "mesh escalate PROCEED, then re-run with --force-physical-write.") def cmd_program(args): if not args.force_physical_write: return _refuse_physical(args, "program") used = f"ethtool:{args.iface}" if args.iface else "libusb" try: image = BinFileBackend(args.image).read_image() eep = Eeprom.from_bytes(image) if not eep.has_signature and not args.allow_unsigned: return _fail(args, "image byte0 != 0xA5 (unsigned)", hint="run `edit --ensure-signature` first, or pass " "--allow-unsigned to write it as-is") if args.iface: EthtoolBackend(args.iface, sudo=args.sudo).write_bytes(0, image) else: be = UsbEepromBackend(size=len(image)) # sudo-free, signature last try: be.program_image(image) finally: be.close() except (BackendError, EepromError) as e: return _fail(args, str(e)) result = {"backend": used, "bytes": len(image), "verified": None} if not args.no_verify: try: actual = _read_connected_image(args) except BackendError as e: return _fail(args, f"programmed but could not read back to verify: {e}") n = min(len(image), len(actual)) diffs = [i for i in range(n) if image[i] != actual[i]] result["verified"] = (not diffs and len(image) == len(actual)) result["diff_count"] = len(diffs) if not result["verified"]: return _emit(args, False, result, human=f"programmed but VERIFY FAILED ({len(diffs)} bytes)", hint="do not trust this image; re-read and re-program") return _emit(args, True, result, human=f"programmed {len(image)} bytes via {used}" + ("" if args.no_verify else " (verified)") + "; replug/USB-reset to reload") def cmd_erase(args): if not args.force_physical_write: return _refuse_physical(args, "erase") used = f"ethtool:{args.iface}" if args.iface else "libusb" payload = bytes([0x00]) if args.mode == "signature" else bytes([0xFF] * args.size) off = 0 try: if args.iface: EthtoolBackend(args.iface, sudo=args.sudo).write_bytes(off, payload) else: be = UsbEepromBackend(size=args.size) try: be.write_bytes(off, payload) finally: be.close() except BackendError as e: return _fail(args, str(e)) touched = ("byte0 signature cleared (recoverable fallback)" if args.mode == "signature" else f"wrote 0xFF across {args.size} bytes") return _emit(args, True, {"backend": used, "mode": args.mode}, human=f"erase ({args.mode}) via {used}: {touched}; replug to reload") def _bulk_old_identity(args, dev): """Best-effort read of a connected unit's CURRENT MAC/serial for the record.""" try: cur = Eeprom.from_bytes(_read_connected_image(args)) if cur.has_signature: sv = cur.string("serial") return cur.mac, (sv.value if sv.present else "") except (BackendError, EepromError): pass return (dev.get("mac") if dev else None), "" def cmd_bulk(args): """Bulk-program many adapters with an auto-incrementing MAC/serial and a resumable manifest. Dry-run is the default; real writes are gated.""" from . import bulk try: seq = bulk.sequence(args.base, args.count, step=args.step, local_admin=args.local_admin, scheme=args.scheme) manifest = bulk.Manifest(args.manifest) except EepromError as e: return _fail(args, str(e)) start = manifest.next_index() remaining = [u for u in seq if u["index"] >= start and not manifest.has_mac(u["mac"])] # DRY-RUN is the default AND the forced behaviour whenever a real write is # not explicitly authorised -- so `bulk` never touches hardware by accident. if args.dry_run or not args.force_physical_write: for u in remaining: manifest.append(index=u["index"], usb_path="", old_mac="", old_serial="", new_mac=u["mac"], new_serial=u["serial"], verify="DRYRUN") preview = "\n".join( f" [{u['index']:>3}] MAC {u['mac']} serial {u['serial']}" for u in remaining) or " (nothing to do; manifest already complete)" return _emit(args, True, {"mode": "dry-run", "count": args.count, "step": args.step, "start_index": start, "planned": remaining, "manifest": manifest.csv_path}, human=(f"DRY-RUN preview: {len(remaining)} unit(s) " f"(resume from index {start}); no device touched.\n" f"{preview}\nManifest -> {manifest.csv_path}"), hint="real burn: escalate a mesh session-scoped allow, then " "re-run with --force-physical-write") # REAL burn (gated). One unit per invocation by default (replug + re-run); # --loop prompts between units in a single run for bench use. try: tmpl = (Eeprom.from_bytes(BinFileBackend(args.template).read_image()) if args.template else Eeprom.default_template()) except (BackendError, EepromError) as e: return _fail(args, f"template load failed: {e}") to_do = remaining if args.loop else remaining[:1] burned = [] for u in to_do: if args.loop: try: input(f"\n>> Insert unit for index {u['index']} " f"(MAC {u['mac']}); press Enter (Ctrl-C to stop)... ") except (EOFError, KeyboardInterrupt): break devs = scan_devices(probe_eeprom=False) if not devs: return _fail(args, f"no LAN7800 attached for index {u['index']}") dev = devs[0] old_mac, old_serial = _bulk_old_identity(args, dev) # NOT-SWAPPED GUARD: if the connected unit's EEPROM already carries the # identity we just wrote to the previous unit, barrett almost certainly # did not swap adapters. Refuse to re-burn the same physical unit. last = manifest.last_burned_mac() if last and old_mac and old_mac.upper() == last.upper(): return _emit(args, False, {"same_unit": True, "current_mac": old_mac, "next_index": u["index"], "manifest": manifest.csv_path}, human=(f"This looks like the unit you just programmed " f"(MAC {old_mac}). Unplug it and insert the next " f"blank adapter, then retry."), hint="not-swapped guard: refusing to re-burn the same unit") try: img = bulk.build_image(tmpl, u["mac"], u["serial"]) be = UsbEepromBackend(size=len(img)) try: be.program_image(img) # LED-mux-gated, signature LAST finally: be.close() actual = _read_connected_image(args) ok = bytes(actual) == bytes(img) except (BackendError, EepromError) as e: manifest.append(index=u["index"], usb_path=dev.get("usb_path"), old_mac=old_mac, old_serial=old_serial, new_mac=u["mac"], new_serial=u["serial"], verify="FAIL") return _fail(args, f"index {u['index']} program failed: {e}") manifest.append(index=u["index"], usb_path=dev.get("usb_path"), old_mac=old_mac, old_serial=old_serial, new_mac=u["mac"], new_serial=u["serial"], verify="PASS" if ok else "FAIL") burned.append({"index": u["index"], "mac": u["mac"], "serial": u["serial"], "verified": ok}) if not ok: return _emit(args, False, {"burned": burned, "manifest": manifest.csv_path}, human=f"index {u['index']} VERIFY FAILED after write") return _emit(args, True, {"mode": "burn", "burned": burned, "manifest": manifest.csv_path}, human=(f"burned {len(burned)} unit(s); manifest {manifest.csv_path}" + ("" if args.loop else "; replug next unit + re-run"))) # ---- argument parser ------------------------------------------------------ def build_parser(): p = argparse.ArgumentParser( prog="lan7800prog", description="LAN7800 EEPROM programmer (Core v1, LAN7800 only)") p.add_argument("--json", action="store_true", help="emit machine-readable JSON") sub = p.add_subparsers(dest="cmd", required=True) def add_target(sp, iface=True, infile=True): if infile: sp.add_argument("--in", dest="in_file", help="base .bin image") if iface: sp.add_argument("--iface", help="network interface of the LAN7800") sp.add_argument("--no-sudo", dest="sudo", action="store_false", default=True, help="do not prefix ethtool with sudo") sp = sub.add_parser("scan", help="enumerate attached LAN7800 adapters (FREE)") sp.add_argument("--no-probe", action="store_true", help="skip the ethtool EEPROM signature probe") sp.set_defaults(func=cmd_scan) sp = sub.add_parser("read-device", help="read the connected LAN7800 (auto: netdev->ethtool " "else libusb) (FREE)") sp.add_argument("--out", help="also write the image to this .bin") sp.add_argument("--size", type=int, default=MAX_EEPROM_SIZE, choices=SIZES) sp.add_argument("--no-sudo", dest="sudo", action="store_false", default=True) sp.set_defaults(func=cmd_read_device) sp = sub.add_parser("udev-rule", help="print the udev rule needed for libusb device access") sp.set_defaults(func=cmd_udev_rule) sp = sub.add_parser("new", help="create a default LAN7800 template .bin " "(or --blank for an empty image)") sp.add_argument("--out", required=True) sp.add_argument("--size", type=int, default=MAX_EEPROM_SIZE, choices=SIZES) sp.add_argument("--blank", action="store_true", help="all-zero image instead of the default template") sp.add_argument("--sign", action="store_true", help="with --blank: lay down the 0xA5 signature") sp.set_defaults(func=cmd_new) sp = sub.add_parser("info", help="parse & print an image or a live device") add_target(sp) sp.set_defaults(func=cmd_info) sp = sub.add_parser("edit", help="edit fields, write a new .bin (no hardware)") sp.add_argument("--in", dest="in_file", help="base .bin (required for VID/PID/strings)") sp.add_argument("--out", help="output .bin (defaults to --in)") sp.add_argument("--size", type=int, default=MAX_EEPROM_SIZE, choices=SIZES) sp.add_argument("--mac") sp.add_argument("--vid") sp.add_argument("--pid") sp.add_argument("--bcd", help="bcdDevice / DID") sp.add_argument("--serial") sp.add_argument("--set-string", action="append", help="name=value (manufacturer/product/serial/configuration/interface)") sp.add_argument("--set-led", action="append", metavar="I=MODE", help="LED index=mode(0..15), e.g. 0=10") sp.add_argument("--led-enable", action="append", metavar="I=1|0") sp.add_argument("--led-blink", help="blink rate code 0=2.5Hz 1=5Hz 2=10Hz 3=20Hz") sp.add_argument("--set-gpio", action="append", metavar="I=dir,drive,out", help="e.g. 0=out,push-pull,1") sp.add_argument("--gpio-wake", action="append", metavar="I=en,pol", help="e.g. 0=1,1") sp.add_argument("--set-configflag", action="append", metavar="I=HEX", help="Config Flags 0..3, e.g. 0=0x00000001") sp.add_argument("--set-raw", action="append", metavar="OFF=HEX", help="raw bytes, e.g. 0x48=0103 (authoritative fallback)") sp.add_argument("--ensure-signature", action="store_true") sp.set_defaults(func=cmd_edit) sp = sub.add_parser("read", help="read device EEPROM -> .bin (FREE)") sp.add_argument("--iface", required=True) sp.add_argument("--out", required=True) sp.add_argument("--no-sudo", dest="sudo", action="store_false", default=True) sp.set_defaults(func=cmd_read) sp = sub.add_parser("verify", help="compare connected device against a .bin (FREE; " "libusb by default, --iface for ethtool)") sp.add_argument("--iface", help="use ethtool via this iface (default: libusb)") sp.add_argument("--image", required=True) sp.add_argument("--no-sudo", dest="sudo", action="store_false", default=True) sp.set_defaults(func=cmd_verify) sp = sub.add_parser("program", help="GATED: write a .bin to the connected device + verify " "(libusb by default, --iface for ethtool)") sp.add_argument("--iface", help="use ethtool via this iface (default: libusb)") sp.add_argument("--image", required=True) sp.add_argument("--force-physical-write", action="store_true", help="required; set only after a cleared mesh escalation") sp.add_argument("--allow-unsigned", action="store_true") sp.add_argument("--no-verify", action="store_true") sp.add_argument("--no-sudo", dest="sudo", action="store_false", default=True) sp.set_defaults(func=cmd_program) sp = sub.add_parser("erase", help="GATED: clear signature or full 0xFF erase") sp.add_argument("--iface", help="use ethtool via this iface (default: libusb)") sp.add_argument("--mode", choices=["signature", "full"], default="signature") sp.add_argument("--size", type=int, default=MAX_EEPROM_SIZE, choices=SIZES) sp.add_argument("--force-physical-write", action="store_true") sp.add_argument("--no-sudo", dest="sudo", action="store_false", default=True) sp.set_defaults(func=cmd_erase) sp = sub.add_parser( "bulk", help="program many adapters with an auto-incrementing identity " "(dry-run by default; real writes gated)") sp.add_argument("--base", required=True, help="base MAC, e.g. 00:80:0F:78:00:00 (OUI not hardcoded)") sp.add_argument("--count", type=int, required=True, help="number of units") sp.add_argument("--step", type=int, default=1, help="MAC increment (default 1)") sp.add_argument("--manifest", required=True, help="manifest path (.csv; a .json is written alongside)") sp.add_argument("--scheme", default="mac-hex", help="serial scheme (default mac-hex: iSerial = MAC hex)") sp.add_argument("--local-admin", dest="local_admin", action="store_true", help="force the locally-administered 02: bit on the base MAC") sp.add_argument("--template", help="base .bin (needs a serial slot); default: built-in template") sp.add_argument("--dry-run", dest="dry_run", action="store_true", help="compute+log the sequence, write DRYRUN rows; no device") sp.add_argument("--loop", action="store_true", help="interactive: prompt to insert each unit within one run") sp.add_argument("--force-physical-write", dest="force_physical_write", action="store_true", help="authorize REAL writes (needs a mesh session-scoped allow)") sp.add_argument("--iface", help="use ethtool via this iface instead of libusb") sp.add_argument("--no-sudo", dest="sudo", action="store_false", default=True) sp.set_defaults(func=cmd_bulk) return p def main(argv=None): parser = build_parser() args = parser.parse_args(argv) try: return args.func(args) except (EepromError, BackendError) as e: return _fail(args, str(e)) except (ValueError, KeyError) as e: # user-input parse errors (int(), bytes.fromhex, dict lookups) — never # leak a Python traceback / server paths to the caller or the webapp. return _fail(args, f"invalid input: {e}", hint="check numeric/hex arguments and field names") except BrokenPipeError: return 0 except Exception as e: # last-resort: clean message, no traceback return _fail(args, f"unexpected error: {type(e).__name__}: {e}") if __name__ == "__main__": sys.exit(main())