Brute-force probe all 256 device_addr candidates on 0x50 handles
The previous probe sent a single all-zeros frame and waited for the dongle to respond — but the dongle silently drops frames with the wrong device_addr, producing no response. Now send a valid ROOT.GetProtocolVersion request for every possible device_addr (0x00–0xFF). The dongle ignores the 255 wrong addresses and responds only to the correct one. The response carries the real address at byte[1]. 256 writes complete in under 100ms on USB; the read phase (3 x 500ms) catches the single response. This discovers the address during synchronous init, before the listener starts, eliminating the need for deferred init when the headset is already powered on.
This commit is contained in:
Ken Sanislo
parent
5fddfb67ea
commit
6f3e6fccd3
|
|
@ -338,19 +338,18 @@ def _centurion_frame_header(state: CenturionHandleState, cpl_length: int, flags:
|
|||
|
||||
_CENTURION_REPORT_IDS = (CENTURION_REPORT_ID, CENTURION_ADDRESSED_REPORT_ID)
|
||||
|
||||
# Per-iteration read timeout (ms) and total iterations for the 0x50 probe below.
|
||||
# Read timeout (ms) for the brute-force device_addr probe below.
|
||||
_CENTURION_PROBE_READ_TIMEOUT_MS = 500
|
||||
_CENTURION_PROBE_READ_ITERATIONS = 3
|
||||
|
||||
|
||||
def probe_centurion_device_addr(handle, state: CenturionHandleState) -> bool:
|
||||
"""Probe the device address byte for a 0x50-variant Centurion handle.
|
||||
"""Brute-force probe the device address byte for a 0x50-variant Centurion handle.
|
||||
|
||||
Sends a 64-byte all-zero frame with the detected report ID and reads back
|
||||
the first response. The device answers with an error/unsolicited frame
|
||||
whose byte[1] holds its device address. Without this probe, the very first
|
||||
real TX ships with device_addr=0x00, which stricter firmware silently
|
||||
drops — breaking dongle feature discovery before it starts.
|
||||
Sends a ROOT.GetProtocolVersion request for every possible device_addr
|
||||
(0x00–0xFF). The dongle silently ignores wrong addresses and responds
|
||||
only to the correct one. The response carries the real address at byte[1],
|
||||
which we extract and store on the handle state.
|
||||
|
||||
No-op for 0x51 (no device_addr byte) or when an address is already known.
|
||||
Returns True if the address was learned.
|
||||
|
|
@ -358,58 +357,40 @@ def probe_centurion_device_addr(handle, state: CenturionHandleState) -> bool:
|
|||
if state.report_id != CENTURION_ADDRESSED_REPORT_ID or state.device_addr is not None:
|
||||
return False
|
||||
ihandle = int(handle)
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug(
|
||||
"(%s) probing centurion device_addr (report_id=0x%02X, %d iters x %d ms)",
|
||||
handle,
|
||||
state.report_id,
|
||||
_CENTURION_PROBE_READ_ITERATIONS,
|
||||
_CENTURION_PROBE_READ_TIMEOUT_MS,
|
||||
)
|
||||
probe = bytes([state.report_id]) + b"\x00" * (CENTURION_FRAME_SIZE - 1)
|
||||
try:
|
||||
hidapi.write(ihandle, probe)
|
||||
except Exception as reason:
|
||||
logger.warning("(%s) centurion device_addr probe write failed: %s", handle, reason)
|
||||
return False
|
||||
for attempt in range(1, _CENTURION_PROBE_READ_ITERATIONS + 1):
|
||||
logger.info("(%s) probing centurion device_addr: brute-force 0x00-0xFF", handle)
|
||||
|
||||
# ROOT.GetProtocolVersion: feat_idx=0x00, func=0x10, 3 zero param bytes
|
||||
payload = bytes([0x00, 0x10, 0x00, 0x00, 0x00])
|
||||
cpl_length = len(payload) + 1 # +1 for flags byte
|
||||
write_failed = 0
|
||||
|
||||
# Send a ROOT query for every possible device_addr (256 frames).
|
||||
# The dongle ignores frames with the wrong address. Only the matching
|
||||
# one produces a response that we can read back.
|
||||
for addr in range(256):
|
||||
frame = struct.pack("!BBBB", CENTURION_ADDRESSED_REPORT_ID, addr, cpl_length, 0x00) + payload
|
||||
frame = frame + b"\x00" * (CENTURION_FRAME_SIZE - len(frame))
|
||||
try:
|
||||
hidapi.write(ihandle, frame)
|
||||
except Exception:
|
||||
write_failed += 1
|
||||
if write_failed > 3:
|
||||
logger.warning("(%s) centurion device_addr probe: too many write failures, aborting", handle)
|
||||
return False
|
||||
|
||||
# Read back the response — dongle only replied to the correct address.
|
||||
for _attempt in range(_CENTURION_PROBE_READ_ITERATIONS):
|
||||
try:
|
||||
data = hidapi.read(ihandle, CENTURION_FRAME_SIZE, _CENTURION_PROBE_READ_TIMEOUT_MS)
|
||||
except Exception as reason:
|
||||
logger.warning("(%s) centurion device_addr probe read failed: %s", handle, reason)
|
||||
return False
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
if data:
|
||||
logger.debug(
|
||||
"(%s) centurion probe attempt %d/%d: got %d bytes, head=%s",
|
||||
handle,
|
||||
attempt,
|
||||
_CENTURION_PROBE_READ_ITERATIONS,
|
||||
len(data),
|
||||
common.strhex(data[:4]) if len(data) >= 4 else common.strhex(data),
|
||||
)
|
||||
else:
|
||||
logger.debug(
|
||||
"(%s) centurion probe attempt %d/%d: read timeout (no data)",
|
||||
handle,
|
||||
attempt,
|
||||
_CENTURION_PROBE_READ_ITERATIONS,
|
||||
)
|
||||
if data and len(data) >= 2 and ord(data[:1]) == state.report_id:
|
||||
state.device_addr = ord(data[1:2])
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug(
|
||||
"(%s) probed centurion device addr 0x%02X on attempt %d",
|
||||
handle,
|
||||
state.device_addr,
|
||||
attempt,
|
||||
)
|
||||
logger.info("(%s) probed centurion device addr 0x%02X", handle, state.device_addr)
|
||||
return True
|
||||
logger.warning(
|
||||
"(%s) centurion device_addr probe timed out after %d attempts, subsequent TX will use 0x00",
|
||||
handle,
|
||||
_CENTURION_PROBE_READ_ITERATIONS,
|
||||
)
|
||||
|
||||
logger.warning("(%s) centurion device_addr brute-force probe got no response", handle)
|
||||
return False
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -316,7 +316,7 @@ class TestUnwrapCenturionFrame:
|
|||
|
||||
|
||||
class TestProbeCenturionDeviceAddr:
|
||||
"""Test probe_centurion_device_addr: write-then-read dance to learn device_addr."""
|
||||
"""Test probe_centurion_device_addr: brute-force write for all 256 addrs, then read."""
|
||||
|
||||
HANDLE = 101
|
||||
|
||||
|
|
@ -326,27 +326,26 @@ class TestProbeCenturionDeviceAddr:
|
|||
def teardown_method(self):
|
||||
base._centurion_handles.pop(self.HANDLE, None)
|
||||
|
||||
def test_learns_addr_from_first_frame(self):
|
||||
def test_learns_addr_from_response(self):
|
||||
state = CenturionHandleState(report_id=CENTURION_ADDRESSED_REPORT_ID)
|
||||
reply = bytes([0x50, 0x23, 0x03, 0x00]) + b"\x00" * 60
|
||||
with (
|
||||
mock.patch.object(base.hidapi, "write") as mock_write,
|
||||
mock.patch.object(base.hidapi, "read", return_value=reply) as mock_read,
|
||||
mock.patch.object(base.hidapi, "read", return_value=reply),
|
||||
):
|
||||
result = base.probe_centurion_device_addr(self.HANDLE, state)
|
||||
assert result is True
|
||||
assert state.device_addr == 0x23
|
||||
# probe write is a 64-byte all-zero frame with just the report ID
|
||||
mock_write.assert_called_once()
|
||||
_, wdata = mock_write.call_args[0]
|
||||
assert len(wdata) == base.CENTURION_FRAME_SIZE
|
||||
assert wdata[0] == CENTURION_ADDRESSED_REPORT_ID
|
||||
assert wdata[1:] == b"\x00" * (base.CENTURION_FRAME_SIZE - 1)
|
||||
mock_read.assert_called_once()
|
||||
# Should have written 256 probe frames (one per candidate addr)
|
||||
assert mock_write.call_count == 256
|
||||
# Each frame should be 64 bytes with report_id 0x50
|
||||
for call in mock_write.call_args_list:
|
||||
_, wdata = call[0]
|
||||
assert len(wdata) == base.CENTURION_FRAME_SIZE
|
||||
assert wdata[0] == CENTURION_ADDRESSED_REPORT_ID
|
||||
|
||||
def test_skips_non_matching_frames_until_match(self):
|
||||
state = CenturionHandleState(report_id=CENTURION_ADDRESSED_REPORT_ID)
|
||||
# first two reads return unrelated frames, third returns our 0x50 frame
|
||||
noise = b"\x11\xff" + b"\x00" * 62
|
||||
match = bytes([0x50, 0x42, 0x03, 0x00]) + b"\x00" * 60
|
||||
with (
|
||||
|
|
@ -391,7 +390,7 @@ class TestProbeCenturionDeviceAddr:
|
|||
mock_write.assert_not_called()
|
||||
mock_read.assert_not_called()
|
||||
|
||||
def test_handles_write_failure(self):
|
||||
def test_aborts_on_repeated_write_failure(self):
|
||||
state = CenturionHandleState(report_id=CENTURION_ADDRESSED_REPORT_ID)
|
||||
with (
|
||||
mock.patch.object(base.hidapi, "write", side_effect=OSError("no device")),
|
||||
|
|
@ -401,3 +400,15 @@ class TestProbeCenturionDeviceAddr:
|
|||
assert result is False
|
||||
assert state.device_addr is None
|
||||
mock_read.assert_not_called()
|
||||
|
||||
def test_write_frames_have_sequential_addrs(self):
|
||||
"""Verify each write uses a different device_addr from 0x00 to 0xFF."""
|
||||
state = CenturionHandleState(report_id=CENTURION_ADDRESSED_REPORT_ID)
|
||||
with (
|
||||
mock.patch.object(base.hidapi, "write") as mock_write,
|
||||
mock.patch.object(base.hidapi, "read", return_value=None),
|
||||
):
|
||||
base.probe_centurion_device_addr(self.HANDLE, state)
|
||||
assert mock_write.call_count == 256
|
||||
addrs_sent = [mock_write.call_args_list[i][0][1][1] for i in range(256)]
|
||||
assert addrs_sent == list(range(256))
|
||||
|
|
|
|||
Loading…
Reference in New Issue