From 47833a6636231f83331c2fb067675f41c499df3e Mon Sep 17 00:00:00 2001 From: DAB-LABS <128871138+DAB-LABS@users.noreply.github.com> Date: Sat, 5 Sep 2026 04:54:19 +0000 Subject: [PATCH] Fix the issues found in the review of 1.0.0 I had 1.0.0 reviewed by an outside party before asking Home Assistant to switch to it. The review confirmed the library talks to devices exactly the way the original did, and found one real bug plus a few small things. This fixes them. No change to the wire format or the public API. Tested on Python 3.13 and 3.14 (252 tests) and live against an RM4 Pro. Technical details: - Replies are matched to their request by the packet counter the device echoes at offset 0x28. A reply for a request that already timed out is dropped; a reply whose counter matches nothing sent is still accepted, so firmware that does not echo the counter keeps working. - Concurrent callers hitting an expired session key share one re-authentication. The logged-out code (-2) now triggers re-auth. - Changing device.host reopens the socket. - aclose() during a request fails it immediately. - capture() treats ReadError (-10) as "nothing yet" alongside StorageError (-5); the CLI inherits this. - An abandoned capture() generator no longer blocks the next one; new capture_active property. - README: broadlink and python-broadlink cannot share an environment. - Changelog: carried-over device commits were squash-merged with Co-authored-by credit; oracle description corrected. - Version 1.0.1. --- CHANGELOG.md | 60 +++++++++++-- README.md | 13 ++- broadlink/device.py | 118 ++++++++++++++++++------- broadlink/remote.py | 185 ++++++++++++++++++++++++++-------------- pyproject.toml | 2 +- tests/test_capture.py | 83 ++++++++++++++++-- tests/test_transport.py | 144 +++++++++++++++++++++++++++++++ 7 files changed, 493 insertions(+), 112 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b1068bec..a26eef85 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,51 @@ All notable changes to this project are recorded here. The format follows Keep a Changelog; versions follow Semantic Versioning. +## 1.0.1 - 2026-09-05 + +Fixes from an independent review of 1.0.0, most of them in the transport. +None changes the wire format or the public API. + +### Fixed + +- A reply to a request that had already timed out could be delivered as the + reply to the next request on the same device, because the persistent + endpoint (new in 1.0.0) is not thrown away between calls the way the old + per-call socket was. Replies are now matched to their request by the + packet counter the device echoes at offset 0x28; a reply carrying the + counter of a request that already timed out is discarded, and a reply + whose counter matches nothing the device sent is still accepted, so + firmware that does not echo the counter is unaffected. Confirmed on an + RM4 Pro, which echoes it. +- `capture()` treated only `StorageError` (-5) as "nothing captured yet". + Some firmware answers `ReadError` (-10); both are now treated as "nothing + yet", matching what the original CLI and Home Assistant do while polling. + The CLI's `--learn` and `--rflearn` inherit the fix. +- Abandoning a capture generator without closing it (for example `break` + out of `async for` to take one code) no longer blocks the next + `capture()` on the same device: opening a new window closes an abandoned + one. Opening a window while another is actively being iterated still + raises `CaptureInProgressError`. A new read-only `Device.capture_active` + property reports whether a window is open. +- Re-authentication is now shared between concurrent callers: when several + requests hit an expired session key at once, the library authenticates + once and every caller retries, instead of one caller re-authenticating + and the others surfacing the raw error. The logged-out code (-2) now + triggers re-authentication as well, matching Home Assistant's own retry. +- Changing `device.host` after the endpoint is open now reopens it against + the new address instead of continuing to talk to the old one. +- `aclose()` while a request is in flight fails that request at once with + `ConnectionClosedError` instead of waiting out the timeout. + +### Documentation + +- The README explains that `broadlink` and `python-broadlink` install the + same package name and cannot coexist, and how to recover if both were + installed. +- The changelog no longer describes the carried-over device commits as + "intact" (they were squash-merged with `Co-authored-by` credit) and no + longer overstates what the oracle records. + ## 1.0.0 - 2026-09-05 This is the first release of `python-broadlink`, a maintained fork of @@ -39,6 +84,8 @@ history below starts at that fork point. #830). - `pulses_to_data` returns `bytes` (it returned a `bytearray`, against its own annotation). +- The device's request lock is now a private `_lock` that is actually + acquired; the unused public `Device.lock` attribute is gone. - Packaging moved to `pyproject.toml`; `setup.py` and the stale `requirements.txt` pin are gone. The distribution name is now `python-broadlink`; the import name stays `broadlink`. Python 3.13 or @@ -80,7 +127,8 @@ history below starts at that fork point. read by band and a capture is tagged from what it armed rather than the byte. - Devices, carried over from pull requests against the original repository - with their authors' commits intact: RM Max 0xAF8B (#838, Alexey Masolov); + with their authors credited (the changes were squash-merged with + `Co-authored-by` trailers naming each author): RM Max 0xAF8B (#838, Alexey Masolov); RM5 plus 0x5224 with a new `rm5plus` class (#831, Anil Daoud); RM mini 3 OEM 0xA544 (#823, Bartłomiej Nogaś); RM mini 3 CMCC 0x27C8 (#802, shuxin); LB26 R1 0xA517 (#812, techitapart); SP mini 3-AL 0x7D15 (#805, @@ -92,7 +140,9 @@ history below starts at that fork point. issue if either does not behave. - `cryptography` 43 or newer is required, the first release with wheels for Python 3.13 (supersedes mjg59/python-broadlink#749). -- A test suite. The `tests/oracle` package records the exact request bytes - every public method of every device class sends, and the results it - decodes from canned responses, so that later changes to the transport - can be checked byte for byte against the original behavior. +- A test suite. The `tests/oracle` package records, for every public method + of every device class, the request each one hands to the transport (its + packet type and plaintext payload) and the result it decodes from a canned + response, so that a later reimplementation can be checked against the + original method by method; the framing, encryption and checksum layer is + covered separately by `tests/test_transport.py`. diff --git a/README.md b/README.md index 3126066d..8edca5e6 100644 --- a/README.md +++ b/README.md @@ -60,9 +60,16 @@ Use pip3 to install the latest version of this module. pip3 install python-broadlink ``` -If the original `broadlink` distribution is also installed in the same -environment, remove it first (`pip3 uninstall broadlink`); both provide the -`broadlink` package. +Both this distribution and the original `broadlink` install a package named +`broadlink`, so only one can be present in an environment at a time. Pip +does not warn about this: installing one on top of the other appears to +succeed, and whichever was installed last is the one that `import broadlink` +finds. If both were installed, uninstall both (`pip3 uninstall broadlink +python-broadlink`) and reinstall this one, since `pip3 uninstall broadlink` +alone removes the shared files and leaves `python-broadlink` registered but +unimportable. This matters most where another package pins `broadlink`: +installing it into the same environment silently replaces this async +library with the original synchronous one. ## Basic functions diff --git a/broadlink/device.py b/broadlink/device.py index 2dc95e0d..4ac3988c 100644 --- a/broadlink/device.py +++ b/broadlink/device.py @@ -10,6 +10,7 @@ from __future__ import annotations import asyncio +import collections import random import socket from collections.abc import AsyncIterator @@ -30,8 +31,17 @@ HelloResponse = Tuple[int, Tuple[str, int], bytes, str, bool] # Device error codes that mean the session key is no longer accepted and a -# fresh auth() will fix it. -7: control key expired; -4012: control id error. -_REAUTH_CODES = {-7, -4012} +# fresh auth() will fix it. -2: logged out; -7: control key expired; +# -4012: control id error. +_REAUTH_CODES = {-2, -7, -4012} + +# How many timed-out request counters to remember, so that a reply to one +# of them arriving late is recognised and dropped instead of being taken as +# the answer to a later request. +_ABANDONED_MAX = 32 + +_CLOSED = (None, None) +"""Sentinel put on the receive queue when the endpoint is closed.""" class _Protocol(asyncio.DatagramProtocol): @@ -203,7 +213,10 @@ def __init__( self._lock: Optional[asyncio.Lock] = None self._transport: Optional[asyncio.DatagramTransport] = None self._protocol: Optional[_Protocol] = None - self._reauth_ok = True + self._endpoint_addr: Optional[Tuple[str, int]] = None + self._abandoned: collections.deque[int] = collections.deque(maxlen=_ABANDONED_MAX) + self._reauth_lock: Optional[asyncio.Lock] = None + self._auth_generation = 0 def __repr__(self) -> str: """Return a formal representation of the device.""" @@ -275,6 +288,7 @@ async def auth(self) -> bool: self.id = int.from_bytes(payload[:0x4], "little") self.update_aes(payload[0x04:0x14]) + self._auth_generation += 1 return True async def hello(self, local_ip_address=None) -> bool: @@ -363,17 +377,30 @@ def get_type(self) -> str: # -------------------------------------------------------- transport async def aclose(self) -> None: - """Close the device's endpoint. It is reopened on the next call.""" - if self._transport is not None: - self._transport.close() - self._transport = None - self._protocol = None + """Close the device's endpoint. It is reopened on the next call. + + A request in flight fails at once with ``ConnectionClosedError`` + rather than waiting out its timeout. + """ + transport, protocol = self._transport, self._protocol + self._transport = None + self._protocol = None + self._endpoint_addr = None + if transport is not None: + transport.close() + if protocol is not None: + protocol.queue.put_nowait(_CLOSED) # type: ignore[arg-type] async def _endpoint(self) -> tuple[asyncio.DatagramTransport, _Protocol]: + if self._transport is not None and self._endpoint_addr != self.host: + # The caller changed host; the connected socket points at the + # old address, so drop it. + await self.aclose() if self._transport is None or self._transport.is_closing(): self._transport, self._protocol = await _open_endpoint( remote_addr=self.host ) + self._endpoint_addr = self.host return self._transport, self._protocol # type: ignore[return-value] def _frame(self, packet_type: int, payload: bytes) -> bytes: @@ -419,28 +446,53 @@ def _validate(resp: bytes) -> bytes: return resp async def _exchange(self, packet: bytes) -> bytes: - """Send one frame and wait for one reply, resending on silence.""" + """Send one frame and wait for its reply, resending on silence. + + Replies carry the request's packet counter (offset 0x28), so a reply + is matched to the request by counter. A reply whose counter belongs + to a request that already timed out is dropped; one with a counter + this device has never sent is accepted, for firmware that may not + echo it. + """ transport, protocol = await self._endpoint() protocol.drain() loop = asyncio.get_running_loop() start = loop.time() timeout = self.timeout + count = int.from_bytes(packet[0x28:0x2A], "little") while True: transport.sendto(packet) - time_left = timeout - (loop.time() - start) - wait = min(DEFAULT_RETRY_INTVL, time_left) - try: - resp, _ = await asyncio.wait_for(protocol.queue.get(), max(wait, 0)) - except asyncio.TimeoutError: - if (loop.time() - start) >= timeout: - raise e.NetworkTimeoutError( - -4000, - "Network timeout", - f"No response received within {timeout}s", - ) from None - continue - return self._validate(resp) + resend_at = loop.time() + DEFAULT_RETRY_INTVL + while True: + now = loop.time() + if now - start >= timeout: + break + wait = min(resend_at, start + timeout) - now + try: + resp, _ = await asyncio.wait_for(protocol.queue.get(), max(wait, 0)) + except asyncio.TimeoutError: + if loop.time() - start >= timeout: + break + if loop.time() >= resend_at: + break # Resend. + continue + if resp is None: + raise e.ConnectionClosedError( + -4013, "Connection closed", "The device endpoint was closed" + ) + resp = self._validate(resp) + reply_count = int.from_bytes(resp[0x28:0x2A], "little") + if reply_count == count or reply_count not in self._abandoned: + return resp + # A late answer to a request we gave up on: keep waiting. + if loop.time() - start >= timeout: + self._abandoned.append(count) + raise e.NetworkTimeoutError( + -4000, + "Network timeout", + f"No response received within {timeout}s", + ) from None async def send_packet( self, packet_type: int, payload: bytes, *, _reauth: bool = True @@ -449,22 +501,24 @@ async def send_packet( If the device answers that the session key is no longer valid, the session is re-authenticated once and the request is sent again. + Concurrent callers that hit the same expired key share one + re-authentication and each retry once. """ if self._lock is None: self._lock = asyncio.Lock() + self._reauth_lock = asyncio.Lock() + generation = self._auth_generation async with self._lock: resp = await self._exchange(self._frame(packet_type, bytes(payload))) - if _reauth and self._reauth_ok: + if _reauth: code = int.from_bytes(resp[0x22:0x24], "little", signed=True) if code in _REAUTH_CODES: - self._reauth_ok = False - try: - await self.auth() - async with self._lock: - resp = await self._exchange( - self._frame(packet_type, bytes(payload)) - ) - finally: - self._reauth_ok = True + async with self._reauth_lock: # type: ignore[union-attr] + if self._auth_generation == generation: + await self.auth() + async with self._lock: + resp = await self._exchange( + self._frame(packet_type, bytes(payload)) + ) return resp diff --git a/broadlink/remote.py b/broadlink/remote.py index 9f905618..4527ccc6 100644 --- a/broadlink/remote.py +++ b/broadlink/remote.py @@ -4,6 +4,7 @@ import enum import struct import time +import weakref from dataclasses import dataclass, field from typing import AsyncIterator, Awaitable, Callable, List, Optional, Tuple @@ -214,7 +215,40 @@ def __init__(self, *args, **kwargs) -> None: # against the value it saw when it armed the device and re-arms # after any send, since the device has one front end for both. self._tx_generation = 0 - self._capture_open = False + # Weak reference to the async generator of the current capture + # window, if any. See _claim_window. + self._window: Optional[weakref.ReferenceType] = None + + @property + def capture_active(self) -> bool: + """True while a capture window is open on this device.""" + window = self._window() if self._window is not None else None + return window is not None and window.ag_frame is not None + + def _check_window(self) -> Optional[weakref.ReferenceType]: + """Refuse a new window while another is being iterated; return the + reference to a previous window that the new one should close.""" + old = self._window() if self._window is not None else None + if old is None or old.ag_frame is None: + return None + if old.ag_running: + raise e.CaptureInProgressError("A capture window is already open") + return self._window + + async def _claim_window(self, prev: Optional[weakref.ReferenceType]) -> None: + """Close a previous window whose consumer walked away from it. + + A window whose consumer is still iterating it is live and a new one + is refused at call time (``_check_window``). One the consumer broke + out of without closing the generator is not live; it is closed here + so that it cannot block the device forever. + """ + old = prev() if prev is not None else None + if old is None or old.ag_frame is None: + return + if old.ag_running: + raise e.CaptureInProgressError("A capture window is already open") + await old.aclose() async def _send(self, command: int, data: bytes = b"") -> bytes: """Send a packet to the device.""" @@ -266,10 +300,13 @@ def capture( and after every ``send_data`` on the same device. Closing the generator sends nothing further; the device times out by itself. Use ``contextlib.aclosing`` (or iterate to the end) so the window is - released promptly. Only one capture window can be open per device; - a second raises ``CaptureInProgressError``. + released promptly. Only one capture window can be open per device: + opening one while another is being iterated raises + ``CaptureInProgressError``, and opening one after breaking out of + another without closing it closes the old one. """ - return self._capture_loop( + prev = self._check_window() + gen = self._capture_loop( self.enter_learning, window, stop_after_first, @@ -277,7 +314,10 @@ def capture( rearm_interval, SignalKind.IR, None, + prev=prev, ) + self._window = weakref.ref(gen) + return gen async def _capture_loop( self, @@ -288,58 +328,60 @@ async def _capture_loop( rearm_interval: float, kind: SignalKind, frequency_mhz: Optional[float], + *, + prev: Optional[weakref.ReferenceType] = None, + claim: bool = True, ) -> AsyncIterator[CapturedSignal]: if window < 0: raise ValueError("window must be 0 (open-ended) or positive") if poll_interval <= 0 or rearm_interval <= 0: raise ValueError("poll_interval and rearm_interval must be positive") - if self._capture_open: - raise e.CaptureInProgressError("A capture window is already open") + if claim: + await self._claim_window(prev) - self._capture_open = True - try: - loop = asyncio.get_running_loop() - deadline = loop.time() + window if window else None - timeouts = 0 + loop = asyncio.get_running_loop() + deadline = loop.time() + window if window else None + timeouts = 0 + + await arm() + armed_at = loop.time() + generation = self._tx_generation - await arm() - armed_at = loop.time() - generation = self._tx_generation + while True: + now = loop.time() + if deadline is not None and now >= deadline: + return + delay = poll_interval + if deadline is not None: + delay = min(delay, deadline - now) + await asyncio.sleep(delay) - while True: - now = loop.time() - if deadline is not None and now >= deadline: + try: + data = await self.check_data() + except (e.StorageError, e.ReadError): + # "Nothing yet": -5 on the RM4 Pro, -10 on some older + # firmware (upstream's CLI and Home Assistant tolerate + # both). + data = b"" + except e.NetworkTimeoutError: + timeouts += 1 + if timeouts >= 3: + raise + generation = -1 # Re-arm; the device's state is unknown. + continue + timeouts = 0 + + if data: + yield CapturedSignal.from_packet(data, frequency_mhz, kind=kind) + if stop_after_first: return - delay = poll_interval - if deadline is not None: - delay = min(delay, deadline - now) - await asyncio.sleep(delay) - - try: - data = await self.check_data() - except e.StorageError: - data = b"" # The device's answer for "nothing yet". - except e.NetworkTimeoutError: - timeouts += 1 - if timeouts >= 3: - raise - generation = -1 # Re-arm; the device's state is unknown. - continue - timeouts = 0 - - if data: - yield CapturedSignal.from_packet(data, frequency_mhz, kind=kind) - if stop_after_first: - return - generation = -1 # One code per session: re-arm. - - now = loop.time() - if generation != self._tx_generation or now - armed_at >= rearm_interval: - await arm() - armed_at = loop.time() - generation = self._tx_generation - finally: - self._capture_open = False + generation = -1 # One code per session: re-arm. + + now = loop.time() + if generation != self._tx_generation or now - armed_at >= rearm_interval: + await arm() + armed_at = loop.time() + generation = self._tx_generation class rmpro(rmmini): @@ -369,7 +411,7 @@ async def cancel_sweep_frequency(self) -> None: """Cancel sweep frequency.""" await self._send(0x1E) - async def capture_rf( + def capture_rf( self, window: float = 30.0, *, @@ -389,24 +431,38 @@ async def capture_rf( The window, polling, re-arm and stop-after-first semantics are those of ``capture``; the sweep counts against the same ``window``. A - ``send_data`` during the sweep restarts it. Each ``CapturedSignal`` - carries the carrier in ``frequency_mhz``, which the packet itself - does not record. + ``send_data`` during the sweep restarts it. Closing the generator + during a sweep sends nothing; the device ends the sweep on its own. + Each ``CapturedSignal`` carries the carrier in ``frequency_mhz``, + which the packet itself does not record. """ - if self._capture_open: - raise e.CaptureInProgressError("A capture window is already open") + prev = self._check_window() + gen = self._capture_rf_loop( + window, frequency, stop_after_first, poll_interval, rearm_interval, + prev=prev, + ) + self._window = weakref.ref(gen) + return gen + + async def _capture_rf_loop( + self, + window: float, + frequency: Optional[float], + stop_after_first: bool, + poll_interval: float, + rearm_interval: float, + *, + prev: Optional[weakref.ReferenceType], + ) -> AsyncIterator[CapturedSignal]: if window < 0 or poll_interval <= 0: raise ValueError("window must be 0 or positive, poll_interval positive") + await self._claim_window(prev) loop = asyncio.get_running_loop() deadline = loop.time() + window if window else None if frequency is None: - self._capture_open = True - try: - frequency = await self._sweep(deadline, poll_interval) - finally: - self._capture_open = False + frequency = await self._sweep(deadline, poll_interval) if frequency is None: return if deadline is not None: @@ -418,10 +474,15 @@ async def arm() -> None: await self.find_rf_packet(frequency) kind = SignalKind.RF_315 if frequency < 400 else SignalKind.RF_433 - async for signal in self._capture_loop( - arm, window, stop_after_first, poll_interval, rearm_interval, kind, frequency - ): - yield signal + inner = self._capture_loop( + arm, window, stop_after_first, poll_interval, rearm_interval, kind, frequency, + claim=False, + ) + try: + async for signal in inner: + yield signal + finally: + await inner.aclose() async def _sweep( self, deadline: Optional[float], poll_interval: float diff --git a/pyproject.toml b/pyproject.toml index 76f1a2f7..c5224d2e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "python-broadlink" -version = "1.0.0" +version = "1.0.1" description = "Python API for controlling Broadlink devices" readme = "README.md" license = "MIT" diff --git a/tests/test_capture.py b/tests/test_capture.py index c4e4ce33..a6381fbf 100644 --- a/tests/test_capture.py +++ b/tests/test_capture.py @@ -50,6 +50,7 @@ def __init__(self, device: broadlink.Device, framing: str) -> None: self.sweeping = False self.sweep_answers: list[tuple[bool, float]] = [] self.timeouts_to_raise = 0 + self.nothing_yet_code = -5 # -10 on some older firmware device.send_packet = self.send_packet # type: ignore[method-assign] # -- what the test does to the device @@ -98,7 +99,7 @@ def handle(self, command: int, data: bytes) -> tuple[bytes, int | str]: self.timeouts_to_raise -= 1 return b"", "timeout" if self.pending is None: - return b"", -5 + return b"", self.nothing_yet_code code, self.pending = self.pending, None return code, 0 if command == CMD_SEND: @@ -166,7 +167,7 @@ async def go(): assert fake.commands[0][0] == CMD_LEARN assert fake.count(CMD_LEARN) == 1 assert fake.count(CMD_CHECK) >= 2 - assert device._capture_open is False + assert device.capture_active is False def test_capture_window_elapses_with_nothing(): @@ -175,7 +176,7 @@ def test_capture_window_elapses_with_nothing(): assert signals == [] assert fake.count(CMD_LEARN) == 1 assert fake.count(CMD_CHECK) >= 3 - assert device._capture_open is False + assert device.capture_active is False async def _collect(gen): @@ -287,7 +288,7 @@ async def go(): got = run(go()) assert len(got) == 1 - assert device._capture_open is False + assert device.capture_active is False # Closing sends nothing further to the device. assert fake.commands[-1][0] in (CMD_CHECK, CMD_LEARN) @@ -302,13 +303,77 @@ async def go(): await asyncio.sleep(2 * UNIT) with pytest.raises(e.CaptureInProgressError): await _collect(device.capture(window=1, **FAST)) - assert device._capture_open is True + assert device.capture_active is True task.cancel() with pytest.raises(asyncio.CancelledError): await task run(go()) - assert device._capture_open is False + assert device.capture_active is False + + +def test_abandoned_window_is_closed_by_the_next_one(): + """A consumer that breaks out of the loop without closing the generator + must not block the device; the next window closes the old one.""" + device, fake = make() + + async def go(): + asyncio.get_running_loop().create_task(press_later(fake, IR, 2 * UNIT)) + first = device.capture(window=1, stop_after_first=False, **FAST) + async for s in first: + got = s + break # Walk away without aclose(). + assert device.capture_active is True # The old generator is suspended. + assert not first.ag_running + asyncio.get_running_loop().create_task(press_later(fake, RF, 2 * UNIT)) + second = [s async for s in device.capture(window=1, **FAST)] + assert first.ag_frame is None # Closed by the second window. + return got, second + + got, second = run(go()) + assert got.packet == IR + assert [s.packet for s in second] == [RF] + assert device.capture_active is False + + +def test_unstarted_window_does_not_block(): + device, fake = make() + + async def go(): + _unused = device.capture(window=1, **FAST) # never iterated + asyncio.get_running_loop().create_task(press_later(fake, IR, 2 * UNIT)) + return [s async for s in device.capture(window=1, **FAST)] + + assert len(run(go())) == 1 + + +def test_older_firmware_read_error_means_nothing_yet(): + device, fake = make() + fake.nothing_yet_code = -10 # ReadError + + async def go(): + asyncio.get_running_loop().create_task(press_later(fake, IR, 3 * UNIT)) + return [s async for s in device.capture(window=1, **FAST)] + + signals = run(go()) + assert len(signals) == 1 + assert fake.count(CMD_CHECK) >= 2 + + +def test_capture_rf_abandoned_then_ir_window(): + device, fake = make() + + async def go(): + asyncio.get_running_loop().create_task(press_later(fake, RF, 2 * UNIT)) + rf = device.capture_rf(window=1, frequency=433.92, stop_after_first=False, **FAST) + async for _ in rf: + break + asyncio.get_running_loop().create_task(press_later(fake, IR, 2 * UNIT)) + return [s async for s in device.capture(window=1, **FAST)] + + signals = run(go()) + assert [s.kind for s in signals] == [SignalKind.IR] + assert device.capture_active is False def test_transport_timeouts_rearm_then_give_up(): @@ -327,7 +392,7 @@ async def go(): fake.timeouts_to_raise = 3 with pytest.raises(e.NetworkTimeoutError): run(_collect(device.capture(window=1, **FAST))) - assert device._capture_open is False + assert device.capture_active is False def test_capture_rejects_bad_arguments(): @@ -417,7 +482,7 @@ def test_capture_rf_sweep_that_never_locks_is_cancelled(): assert fake.count(CMD_CANCEL_SWEEP) == 1 assert fake.count(CMD_FIND_RF) == 0 assert fake.sweeping is False - assert device._capture_open is False + assert device.capture_active is False def test_send_during_sweep_restarts_it(): @@ -457,7 +522,7 @@ async def go(): await task run(go()) - assert device._capture_open is False + assert device.capture_active is False def test_rf_capture_is_only_on_pro_classes(): diff --git a/tests/test_transport.py b/tests/test_transport.py index 37b6447d..f1b59098 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -223,6 +223,89 @@ async def go(): assert run(go()) == 7 +def stamped(dev: Device, payload: bytes, count: int, error: int = 0) -> bytes: + """A response frame that echoes a packet counter, as real firmware does.""" + frame = bytearray(make_response(dev, payload, error)) + frame[0x28:0x2A] = count.to_bytes(2, "little") + checksum = sum(frame, 0xBEAF) - sum(frame[0x20:0x22]) & 0xFFFF + frame[0x20:0x22] = checksum.to_bytes(2, "little") + return bytes(frame) + + +def test_late_reply_to_timed_out_request_is_not_taken_as_next_reply(net): + """The defect that 0.19.0 could not have because it threw its socket away + after every call: a slow answer to request 1 arriving after request 1 + timed out must not be returned as the answer to request 2.""" + dev = fixed_device() + dev.timeout = 0.02 + + async def go(): + await dev._endpoint() + ep = net.endpoints[-1] + with pytest.raises(e.NetworkTimeoutError): + await dev.send_packet(0x6A, b"") # request 1, count 0x8001, no answer + # Its late reply lands while request 2 (count 0x8002) is waiting. + late = stamped(dev, bytes([1]) + bytes(15), 0x8001) + good = stamped(dev, bytes([2]) + bytes(15), 0x8002) + ep.replies = [late, good] and [] + ep.protocol.queue.put_nowait((late, HOST)) + + async def answer_later(): + await asyncio.sleep(0.005) + ep.protocol.queue.put_nowait((good, HOST)) + + asyncio.get_running_loop().create_task(answer_later()) + dev.timeout = 1 + resp = await dev.send_packet(0x6A, b"") + return dev.decrypt(resp[0x38:])[0] + + assert run(go()) == 2 + + +def test_reply_with_unknown_counter_is_accepted(net): + """Firmware that does not echo the counter must keep working.""" + dev = fixed_device() + + async def go(): + net.replies = [(stamped(dev, bytes([5]) + bytes(15), 0x0000), HOST)] + resp = await dev.send_packet(0x6A, b"") + return dev.decrypt(resp[0x38:])[0] + + assert run(go()) == 5 + + +def test_aclose_fails_inflight_request_fast(net): + dev = fixed_device() + dev.timeout = 5 + net.replies = [] + + async def go(): + task = asyncio.get_running_loop().create_task(dev.send_packet(0x6A, b"")) + await asyncio.sleep(0.01) + t0 = asyncio.get_running_loop().time() + await dev.aclose() + with pytest.raises(e.ConnectionClosedError): + await task + return asyncio.get_running_loop().time() - t0 + + assert run(go()) < 1.0 + + +def test_host_change_reopens_endpoint(net): + dev = fixed_device() + + async def go(): + net.replies = [(make_response(dev, b""), HOST)] + await dev.send_packet(0x6A, b"") + dev.host = ("192.0.2.99", 80) + net.replies = [(make_response(dev, b""), ("192.0.2.99", 80))] + await dev.send_packet(0x6A, b"") + + run(go()) + assert [ep.remote_addr for ep in net.endpoints] == [HOST, ("192.0.2.99", 80)] + assert net.endpoints[0].closed + + # ------------------------------------------------------------------------- auth @@ -311,6 +394,67 @@ async def go(): run(go()) +def test_concurrent_callers_share_one_reauth(net): + dev = fixed_device() + dev.id = 5 + session_key = bytes.fromhex("00112233445566778899aabbccddeeff") + fresh = fixed_device() + auth_reply = make_response(fresh, (0x42).to_bytes(4, "little") + session_key) + renewed = fixed_device() + renewed.update_aes(session_key) + counters = {"value": 1} + + async def go(): + await dev._endpoint() + ep = net.endpoints[-1] + authed = {"done": False} + + def sendto(data, addr=None): + ep.sent.append((bytes(data), addr or ep.remote_addr)) + ptype = int.from_bytes(data[0x26:0x28], "little") + if ptype == 0x65: + authed["done"] = True + reply = auth_reply + elif not authed["done"]: + reply = make_response(dev, b"", error=0xFFF9) # -7 expired + else: + n = counters["value"] + counters["value"] += 1 + reply = make_response(renewed, bytes([n]) + bytes(15)) + ep.protocol.queue.put_nowait((reply, ep.remote_addr)) + + ep.sendto = sendto + a, b = await asyncio.gather(dev.send_packet(0x6A, b"a"), dev.send_packet(0x6A, b"b")) + return ep, {dev.decrypt(a[0x38:])[0], dev.decrypt(b[0x38:])[0]} + + ep, values = run(go()) + types = [int.from_bytes(f[0x26:0x28], "little") for f, _ in ep.sent] + assert types.count(0x65) == 1 # exactly one auth despite two expired requests + assert values == {1, 2} + + +def test_logged_out_code_triggers_reauth(net): + dev = fixed_device() + session_key = bytes.fromhex("00112233445566778899aabbccddeeff") + fresh = fixed_device() + auth_reply = make_response(fresh, (0x42).to_bytes(4, "little") + session_key) + renewed = fixed_device() + renewed.update_aes(session_key) + + async def go(): + await dev._endpoint() + ep = net.endpoints[-1] + ep.replies = [ + (make_response(dev, b"", error=0xFFFE), HOST), # -2 logged out + (auth_reply, HOST), + (make_response(renewed, bytes([3]) + bytes(15)), HOST), + ] + resp = await dev.send_packet(0x6A, b"") + return dev.decrypt(resp[0x38:])[0] + + assert run(go()) == 3 + + # ------------------------------------------------------------------- discovery