"""libusb backend, driver-independent EEPROM access via USB vendor commands. Primary path when the LAN7800 has NO lan78xx netdev (so ethtool cannot reach it). Talks to the chip directly with libusb-1.0 via ctypes (no pyusb / pip install needed; uses the system libusb-1.0.so.0). Implements the EEPROM read protocol from docs/01 section B: read reg : bmRequestType 0xC0, bRequest 0xA1, wIndex=reg -> 4 bytes LE write reg : bmRequestType 0x40, bRequest 0xA0, wIndex=reg <- 4 bytes LE read byte A: write E2P_CMD = BUSY|READ|A ; poll BUSY clear ; read E2P_DATA[7:0] write byte : EWEN ; write E2P_DATA ; write E2P_CMD=BUSY|WRITE|A ; poll ; EWDS Runtime needs USB read/write access to 0424:7800 (a udev rule; see udev_rule()). Root is NOT required at runtime once the rule is installed. """ from __future__ import annotations import ctypes as C import time from .base import EepromBackend, BackendError from ..eeprom import LAN7800_VID, LAN7800_PID, MAX_EEPROM_SIZE # LAN78xx register map (docs/01) HW_CFG = 0x010 # On the LAN7800, the EEPROM interface pins (EECLK/EEDI/EEDO) are MUXED with the # LED0/LED1 outputs. While these LED enables are set, the EEPROM bus is NOT # connected -- reads return 0x00 and writes silently no-op. The in-kernel lan78xx # driver clears these before every EEPROM access and restores them after; we must # do the same or nothing reaches the chip. (lan78xx.h: HW_CFG_LEDx_EN_) HW_CFG_LED0_EN = 0x00100000 HW_CFG_LED1_EN = 0x00200000 E2P_CMD = 0x040 E2P_DATA = 0x044 E2P_BUSY = 0x80000000 # EPC_TIMEOUT status bit. On the LAN7800 this is bit 10 (0x00000400), NOT the # bit-30 value some LAN78xx docs list -- confirmed on hardware (RESEARCH). When # the EEPROM bus is unreachable (LED pin-mux still enabled, or no chip), the # command "completes" (BUSY clears) with this bit SET and E2P_DATA reads 0x00; # the OLD code ignored it and returned 0x00-as-blank. Each fresh E2P_CMD write # (addr <= 0x1FF, so bits 9/10 are 0) clears any stale timeout before the op. E2P_TIMEOUT = 0x00000400 E2P_CMD_READ = 0x0 << 28 E2P_CMD_EWEN = 0x2 << 28 E2P_CMD_WRITE = 0x3 << 28 E2P_CMD_EWDS = 0x1 << 28 E2P_ADDR_MASK = 0x1FF # USB vendor request constants VENDOR_READ_TYPE = 0xC0 VENDOR_WRITE_TYPE = 0x40 REQ_READ_REG = 0xA1 REQ_WRITE_REG = 0xA0 UDEV_RULE_PATH = "/etc/udev/rules.d/99-lan7800.rules" def udev_rule() -> str: # GROUP=plugdev + 0660 grants the container user (a plugdev member) rw on # the device node. NOTE: TAG+="uaccess"/MODE 0660 root:root does NOT work in # a headless container (uaccess needs a logind seat; 0660 root:root excludes # the user). Use 0666 instead if the running user is not in plugdev. return ('SUBSYSTEM=="usb", ATTR{idVendor}=="0424", ATTR{idProduct}=="7800", ' 'GROUP="plugdev", MODE="0660"') def udev_install_commands() -> list[str]: return [ f'echo \'{udev_rule()}\' | sudo tee {UDEV_RULE_PATH}', "sudo udevadm control --reload-rules", "sudo udevadm trigger --subsystem-match=usb --attr-match=idVendor=0424", "# then unplug + replug the adapter so the new node gets the rule", ] class _LibUSB: """Thin ctypes shim over the pieces of libusb-1.0 we use.""" def __init__(self): try: self.lib = C.CDLL("libusb-1.0.so.0") except OSError as e: raise BackendError(f"libusb-1.0 not available: {e}") L = self.lib L.libusb_init.argtypes = [C.POINTER(C.c_void_p)] L.libusb_open_device_with_vid_pid.restype = C.c_void_p L.libusb_open_device_with_vid_pid.argtypes = [C.c_void_p, C.c_uint16, C.c_uint16] L.libusb_close.argtypes = [C.c_void_p] L.libusb_exit.argtypes = [C.c_void_p] L.libusb_kernel_driver_active.argtypes = [C.c_void_p, C.c_int] L.libusb_detach_kernel_driver.argtypes = [C.c_void_p, C.c_int] L.libusb_attach_kernel_driver.argtypes = [C.c_void_p, C.c_int] L.libusb_claim_interface.argtypes = [C.c_void_p, C.c_int] L.libusb_release_interface.argtypes = [C.c_void_p, C.c_int] L.libusb_control_transfer.restype = C.c_int L.libusb_control_transfer.argtypes = [ C.c_void_p, C.c_uint8, C.c_uint8, C.c_uint16, C.c_uint16, C.POINTER(C.c_ubyte), C.c_uint16, C.c_uint] self.ctx = C.c_void_p() if L.libusb_init(C.byref(self.ctx)) != 0: raise BackendError("libusb_init failed") def open(self, vid, pid): h = self.lib.libusb_open_device_with_vid_pid(self.ctx, vid, pid) if not h: raise BackendError( f"cannot open USB {vid:04x}:{pid:04x} (device absent, or no " f"permission; install the udev rule and replug)") return C.c_void_p(h) def close(self, h): if h: self.lib.libusb_close(h) def exit(self): if self.ctx: self.lib.libusb_exit(self.ctx) def can_access() -> bool: """True if we can OPEN 0424:7800 (USB access granted), without claiming or detaching the kernel driver. A gentle probe for 'is the udev rule live yet'.""" try: u = _LibUSB() except BackendError: return False try: h = u.lib.libusb_open_device_with_vid_pid(u.ctx, LAN7800_VID, LAN7800_PID) if h: u.lib.libusb_close(C.c_void_p(h)) return True return False finally: u.exit() class UsbEepromBackend(EepromBackend): physical = True name = "libusb" def __init__(self, size: int = MAX_EEPROM_SIZE, iface: int = 0): self.size = size self.iface = iface self._usb = _LibUSB() self._h = None self._detached = False # -- lifecycle -------------------------------------------------------- def _ensure_open(self): if self._h is not None: return # Open the handle ONLY. We deliberately do NOT detach the kernel driver # or claim the interface: the LAN78xx register reads/writes go through # ep0 (the default control endpoint) via control transfers, which do NOT # require claiming an interface. Detach/claim would need CAP_NET_ADMIN # (i.e. root) and would disrupt the bound lan78xx netdev. Staying on ep0 # keeps the read fully sudo-free and non-disruptive. self._h = self._usb.open(LAN7800_VID, LAN7800_PID) def close(self): if self._h is not None: self._usb.close(self._h) self._h = None self._usb.exit() # -- register access -------------------------------------------------- def _read_reg(self, reg: int) -> int: buf = (C.c_ubyte * 4)() n = self._usb.lib.libusb_control_transfer( self._h, VENDOR_READ_TYPE, REQ_READ_REG, 0, reg, buf, 4, 2000) if n != 4: raise BackendError(f"read reg 0x{reg:03x} failed (libusb rc {n})") return buf[0] | (buf[1] << 8) | (buf[2] << 16) | (buf[3] << 24) def _write_reg(self, reg: int, val: int) -> None: buf = (C.c_ubyte * 4)(val & 0xFF, (val >> 8) & 0xFF, (val >> 16) & 0xFF, (val >> 24) & 0xFF) n = self._usb.lib.libusb_control_transfer( self._h, VENDOR_WRITE_TYPE, REQ_WRITE_REG, 0, reg, buf, 4, 2000) if n != 4: raise BackendError(f"write reg 0x{reg:03x} failed (libusb rc {n})") def _wait_not_busy(self, timeout_s: float = 1.0, poll_s: float = 0.0005) -> None: """Poll E2P_CMD until BUSY clears, bounded by wall-clock time (NOT a raw iteration count). A serial EEPROM cell-program cycle is ~5-10 ms; USB register reads are ~tens of us, so a fixed small iteration cap (the old 100 tries) could elapse in ~5 ms and give up mid-write. Sleep between polls and bound by real time so one write cycle always fits, and report the EPC_TIMEOUT status bit truthfully when the chip itself aborts.""" deadline = time.monotonic() + timeout_s while True: cmd = self._read_reg(E2P_CMD) if not (cmd & E2P_BUSY): if cmd & E2P_TIMEOUT: raise BackendError( "EEPROM unreachable (E2P_CMD EPC_TIMEOUT set): the LED " "pin-mux is still enabled, or no EEPROM device is " "present. The E2P operation did NOT complete -- refusing " "to report a false blank/success.") return if time.monotonic() >= deadline: raise BackendError( f"E2P_CMD BUSY never cleared within {timeout_s:.1f}s " f"(EEPROM timeout; last E2P_CMD=0x{cmd:08X})") time.sleep(poll_s) # -- EEPROM pin-mux gate ---------------------------------------------- def _begin_eeprom(self) -> int: """Connect the EEPROM bus by clearing the LED0/LED1 pin-mux enables in HW_CFG (they share pins with EECLK/EEDI/EEDO). Returns the prior HW_CFG so _end_eeprom can restore it. Mirrors the in-kernel lan78xx driver.""" saved = self._read_reg(HW_CFG) self._write_reg(HW_CFG, saved & ~(HW_CFG_LED0_EN | HW_CFG_LED1_EN)) return saved def _end_eeprom(self, saved: int) -> None: self._write_reg(HW_CFG, saved) # -- EEPROM byte ops -------------------------------------------------- def _read_byte(self, addr: int) -> int: self._write_reg(E2P_CMD, E2P_BUSY | E2P_CMD_READ | (addr & E2P_ADDR_MASK)) self._wait_not_busy() return self._read_reg(E2P_DATA) & 0xFF def _write_byte(self, addr: int, val: int) -> None: self._write_reg(E2P_CMD, E2P_BUSY | E2P_CMD_EWEN) # enable writes self._wait_not_busy() self._write_reg(E2P_DATA, val & 0xFF) self._write_reg(E2P_CMD, E2P_BUSY | E2P_CMD_WRITE | (addr & E2P_ADDR_MASK)) self._wait_not_busy() self._write_reg(E2P_CMD, E2P_BUSY | E2P_CMD_EWDS) # disable writes self._wait_not_busy() # -- backend API ------------------------------------------------------ def read_image(self) -> bytes: self._ensure_open() saved = self._begin_eeprom() try: return bytes(self._read_byte(a) for a in range(self.size)) finally: self._end_eeprom(saved) def write_bytes(self, offset: int, values: bytes) -> None: self._ensure_open() saved = self._begin_eeprom() try: for i, v in enumerate(values): self._write_byte(offset + i, v) finally: self._end_eeprom(saved) def program_image(self, data: bytes) -> None: """Program a whole EEPROM image sudo-free via ep0 control transfers. EWEN once, write every byte EXCEPT byte0, then write byte0 (the 0xA5 signature) LAST so a partially-written image is never treated as valid (anti-brick), then EWDS. No interface claim / kernel-driver detach. """ self._ensure_open() saved = self._begin_eeprom() try: self._write_reg(E2P_CMD, E2P_BUSY | E2P_CMD_EWEN) self._wait_not_busy() order = list(range(1, len(data))) + [0] # byte0 (signature) last for a in order: self._write_reg(E2P_DATA, data[a]) self._write_reg(E2P_CMD, E2P_BUSY | E2P_CMD_WRITE | (a & E2P_ADDR_MASK)) self._wait_not_busy() self._write_reg(E2P_CMD, E2P_BUSY | E2P_CMD_EWDS) self._wait_not_busy() finally: self._end_eeprom(saved)