From 6f3e6fccd338ab02e2e90e5be03d0f5e9bfdfb08 Mon Sep 17 00:00:00 2001 From: Ken Sanislo Date: Thu, 16 Apr 2026 10:15:50 -0700 Subject: [PATCH] Brute-force probe all 256 device_addr candidates on 0x50 handles MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- lib/logitech_receiver/base.py | 83 +++++++++++----------------- tests/logitech_receiver/test_base.py | 35 ++++++++---- 2 files changed, 55 insertions(+), 63 deletions(-) diff --git a/lib/logitech_receiver/base.py b/lib/logitech_receiver/base.py index 2c041fe4..f4447586 100644 --- a/lib/logitech_receiver/base.py +++ b/lib/logitech_receiver/base.py @@ -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 diff --git a/tests/logitech_receiver/test_base.py b/tests/logitech_receiver/test_base.py index e14164cf..4559fe3d 100644 --- a/tests/logitech_receiver/test_base.py +++ b/tests/logitech_receiver/test_base.py @@ -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))