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