LAN7800 EEPROM Programmer

"""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)