zte dfu to edl 9008

dfu_to_edl.py

#!/usr/bin/env python3
"""
Send the SalesMultiDL "SwicthToEDL" DIAG command through the ZTE DFU bulk pipe.

Reverse-engineered source:
  BI_W_G_Qualcomm.dll
    CBIWcdmaQualcommService::SwicthToEDL -> FUN_10033210

Non-flashing: drains the unsolicited Sahara HELLO from 19d2:0112, writes one raw
DIAG payload to bulk OUT, and watches for re-enumeration as 05c6:9008.

Fixed usage (no options):
  python3 tool/dfu_to_edl.py
"""

import sys
import time

import usb.core
import usb.util


DFU_VID, DFU_PID = 0x19D2, 0x0112
EDL_VID, EDL_PID = 0x05C6, 0x9008
PAYLOAD = bytes.fromhex("fe00000008000000")
WAIT_S = 10.0
DRAIN_MS = 4000
QUIET_MS = 800


def open_bulk_device(vid: int, pid: int):
    dev = usb.core.find(idVendor=vid, idProduct=pid)
    if dev is None:
        raise RuntimeError(f"device {vid:04x}:{pid:04x} not found")

    try:
        if dev.is_kernel_driver_active(0):
            dev.detach_kernel_driver(0)
    except (NotImplementedError, usb.core.USBError):
        pass

    dev.set_configuration()
    intf = dev.get_active_configuration()[(0, 0)]
    ep_in = usb.util.find_descriptor(
        intf,
        custom_match=lambda e:
            usb.util.endpoint_direction(e.bEndpointAddress) == usb.util.ENDPOINT_IN,
    )
    ep_out = usb.util.find_descriptor(
        intf,
        custom_match=lambda e:
            usb.util.endpoint_direction(e.bEndpointAddress) == usb.util.ENDPOINT_OUT,
    )
    if ep_in is None or ep_out is None:
        raise RuntimeError("bulk IN/OUT endpoints not found")
    return dev, ep_in, ep_out


def read_optional(ep_in, size=0x4000, timeout=1000) -> bytes:
    try:
        return bytes(ep_in.read(size, timeout=timeout))
    except usb.core.USBError as exc:
        if getattr(exc, "errno", None) == 110 or "tim" in str(exc).lower():
            return b""
        raise


def drain_available(ep_in, timeout_ms: int, quiet_ms: int) -> int:
    total = 0
    quiet_deadline = time.time() + quiet_ms / 1000.0
    deadline = time.time() + timeout_ms / 1000.0
    while time.time() < deadline:
        data = read_optional(ep_in, timeout=min(quiet_ms, 500))
        if data:
            total += len(data)
            quiet_deadline = time.time() + quiet_ms / 1000.0
            print(f"[dfu-diag] drained {len(data)} bytes: {data[:32].hex()}")
        elif time.time() >= quiet_deadline:
            break
    return total


def wait_for_usb(vid: int, pid: int, seconds: float) -> bool:
    deadline = time.time() + seconds
    while time.time() < deadline:
        if usb.core.find(idVendor=vid, idProduct=pid) is not None:
            return True
        time.sleep(0.25)
    return False


def main() -> int:
    print(f"[dfu-diag] opening {DFU_VID:04x}:{DFU_PID:04x}")
    _dev, ep_in, ep_out = open_bulk_device(DFU_VID, DFU_PID)
    print(f"[dfu-diag] endpoints IN=0x{ep_in.bEndpointAddress:02x} "
          f"OUT=0x{ep_out.bEndpointAddress:02x}")

    total = drain_available(ep_in, DRAIN_MS, QUIET_MS)
    print(f"[dfu-diag] drain complete ({total} bytes)")

    print(f"[dfu-diag] write={PAYLOAD.hex()}")
    ep_out.write(PAYLOAD, timeout=5000)

    response = read_optional(ep_in, timeout=2000)
    print(f"[dfu-diag] response {len(response)} bytes: {response.hex()}"
          if response else "[dfu-diag] no immediate response")

    print(f"[dfu-diag] waiting for {EDL_VID:04x}:{EDL_PID:04x}")
    if wait_for_usb(EDL_VID, EDL_PID, WAIT_S):
        print("[dfu-diag] target EDL device detected")
        return 0

    print("[dfu-diag] target EDL device was not detected")
    return 1


if __name__ == "__main__":
    sys.exit(main())