diff --git a/CHANGELOG.md b/CHANGELOG.md index 4fcfd7b6..2413a7a3 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.3 - 2026-09-06 + +Fixes from a third review, this one of 1.0.2. No change to the wire +format. One small API change: `pulses` on a captured signal is a tuple. + +### Fixed + +- When the device answered that the session key had expired and the + re-authentication then failed (for example because the device had been + locked in the app), the call raised `AuthenticationError` from the + re-authentication instead of the error the device gave the request. The + original library never re-authenticated, so a program written against + it, Home Assistant's integration included, handles the request's own + error and never expected the other one. The failed re-authentication is + now logged and the request's original reply is returned, so the caller + sees the same `AuthorizationError` or `ConnectionClosedError` it always + did. +- The authentication generation was read before the request lock was + taken rather than under it, so a request queued behind an `auth()` + could observe a stale generation and skip a re-authentication it needed. +- `aclose()` racing an endpoint that was still being opened could leave + the new socket open and unreferenced. The open now notices the close + and fails with `EndpointClosedError`. +- After a new capture window gives the finalizer its turn, it re-checks + that no other window claimed the device in the meantime. +- `CapturedSignal` and `ParsedPacket` are frozen dataclasses, but they + held a list, so they could not be hashed or put in a set. `pulses` is + now a `tuple[int, ...]`. +- `check_error` unpacks the error code as little-endian explicitly + (`"> 8 # Checksum 2 position - transport, _ = await _open_endpoint(broadcast=True) - try: - transport.sendto(payload, (ip_address, DEFAULT_PORT)) - finally: - transport.close() + await send_setup_packet(bytes(payload), ip_address) diff --git a/broadlink/device.py b/broadlink/device.py index 2315f307..600356ea 100644 --- a/broadlink/device.py +++ b/broadlink/device.py @@ -56,10 +56,12 @@ def __init__(self) -> None: self.queue: asyncio.Queue[tuple[bytes, tuple[str, int]]] = asyncio.Queue() self.transport: asyncio.DatagramTransport | None = None - def connection_made(self, transport) -> None: # type: ignore[override] - self.transport = transport + def connection_made(self, transport: asyncio.BaseTransport) -> None: + """Keep the transport; the endpoint sends through it.""" + self.transport = transport # type: ignore[assignment] def datagram_received(self, data: bytes, addr: tuple[str, int]) -> None: + """Queue every datagram for the request that is waiting.""" self.queue.put_nowait((data, addr)) def error_received(self, exc: Exception) -> None: @@ -68,7 +70,7 @@ def error_received(self, exc: Exception) -> None: pass def connection_lost(self, exc: Exception | None) -> None: - pass + """Nothing to do; a waiting request is told through the queue.""" def drain(self) -> None: """Drop anything that arrived before the current request.""" @@ -161,6 +163,17 @@ async def scan( transport.close() +async def send_setup_packet( + payload: bytes, ip_address: str, port: int = DEFAULT_PORT +) -> None: + """Broadcast one Wi-Fi provisioning packet to a device in AP mode.""" + transport, _ = await _open_endpoint(broadcast=True) + try: + transport.sendto(payload, (ip_address, port)) + finally: + transport.close() + + async def ping(ip_address: str, port: int = DEFAULT_PORT) -> None: """Send a ping packet to an address. @@ -213,12 +226,13 @@ def __init__( self.aes = None self.update_aes(bytes.fromhex(self.__INIT_KEY)) - self._lock: asyncio.Lock | None = None + self._lock = asyncio.Lock() self._transport: asyncio.DatagramTransport | None = None self._protocol: _Protocol | None = None self._endpoint_addr: tuple[str, int] | None = None self._recent: collections.deque[int] = collections.deque(maxlen=_RECENT_MAX) - self._reauth_lock: asyncio.Lock | None = None + self._reauth_lock = asyncio.Lock() + self._closes = 0 # Bumped by aclose(); guards an open racing a close. self._auth_generation = 0 def __repr__(self) -> str: @@ -277,9 +291,6 @@ async def auth(self) -> bool: packet[0x2D] = 0x01 packet[0x30:0x36] = b"Test 1" - if self._lock is None: - self._lock = asyncio.Lock() - self._reauth_lock = asyncio.Lock() async with self._lock: self.id = 0 self.update_aes(bytes.fromhex(self.__INIT_KEY)) @@ -292,7 +303,7 @@ async def auth(self) -> bool: _LOGGER.debug("%s: authenticated, session id %d", self.host[0], self.id) return True - async def hello(self, local_ip_address=None) -> bool: + async def hello(self, local_ip_address: str | None = None) -> bool: """Send a hello message to the device. Device information is checked before updating name and lock status. @@ -385,6 +396,7 @@ async def aclose(self) -> None: A request in flight fails at once with ``ConnectionClosedError`` rather than waiting out its timeout. """ + self._closes += 1 transport, protocol = self._transport, self._protocol self._transport = None self._protocol = None @@ -401,7 +413,15 @@ async def _endpoint(self) -> tuple[asyncio.DatagramTransport, _Protocol]: # 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) + closes = self._closes + transport, protocol = await _open_endpoint(remote_addr=self.host) + if self._closes != closes: + # aclose() ran while the socket was being opened. + transport.close() + raise e.EndpointClosedError( + -4013, "Endpoint closed", "The device endpoint was closed" + ) + self._transport, self._protocol = transport, protocol self._endpoint_addr = self.host _LOGGER.debug("%s: endpoint opened", self.host[0]) return self._transport, self._protocol # type: ignore[return-value] @@ -507,27 +527,33 @@ async def _exchange(self, packet: bytes) -> bytes: f"No response received within {timeout}s", ) from None - async def send_packet(self, packet_type: int, payload: bytes) -> bytes: + async def send_packet(self, packet_type: int, payload: bytes | bytearray) -> bytes: """Send a packet to the device and return the raw response frame. 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. + re-authentication and each retry once. If that re-authentication + fails (for example the device has been locked in the app), the + original reply is returned unchanged, so the caller sees the same + error the original library raised and can run its own recovery. """ - if self._lock is None: - self._lock = asyncio.Lock() - self._reauth_lock = asyncio.Lock() - generation = self._auth_generation async with self._lock: + generation = self._auth_generation resp = await self._exchange(self._frame(packet_type, bytes(payload))) code = int.from_bytes(resp[0x22:0x24], "little", signed=True) if code in _REAUTH_CODES: _LOGGER.debug("%s: device answered %d, re-authenticating", self.host[0], code) - async with self._reauth_lock: # type: ignore[union-attr] + async with self._reauth_lock: if self._auth_generation == generation: - await self.auth() + try: + await self.auth() + except e.BroadlinkException as err: + _LOGGER.debug( + "%s: re-authentication failed: %s", self.host[0], err + ) + return resp async with self._lock: resp = await self._exchange(self._frame(packet_type, bytes(payload))) return resp diff --git a/broadlink/exceptions.py b/broadlink/exceptions.py index f5b56d10..7437d6e8 100644 --- a/broadlink/exceptions.py +++ b/broadlink/exceptions.py @@ -168,6 +168,6 @@ def exception(err_code: int) -> BroadlinkException: def check_error(error: bytes) -> None: """Raise exception if an error occurred.""" - error_code = struct.unpack("h", error)[0] + error_code = struct.unpack(" bool: + """True for the radio bands.""" return self is not SignalKind.IR @classmethod - def classify(cls, type_byte: int) -> "SignalKind": + def classify(cls, type_byte: int) -> Self: """Map a packet's raw first byte to a kind, tolerantly. The RF learn path returns bytes in the 0xB_ (433 MHz) and 0xD_ @@ -139,7 +141,7 @@ class ParsedPacket: kind: SignalKind repeat: int - pulses: list[int] + pulses: tuple[int, ...] type_byte: int @@ -152,7 +154,7 @@ def parse_packet(data: bytes, tick: float = TICK) -> ParsedPacket: if len(data) < 4: raise ValueError("Malformed data.") kind = SignalKind.classify(data[0x00]) - return ParsedPacket(kind, data[0x01], data_to_pulses(data, tick), data[0x00]) + return ParsedPacket(kind, data[0x01], tuple(data_to_pulses(data, tick)), data[0x00]) @dataclass(frozen=True) @@ -170,7 +172,7 @@ class CapturedSignal: packet: bytes kind: SignalKind - pulses: list[int] = field(repr=False) + pulses: tuple[int, ...] = field(repr=False) repeat: int = 0 frequency_mhz: float | None = None type_byte: int | None = None @@ -183,7 +185,7 @@ def from_packet( frequency_mhz: float | None = None, *, kind: SignalKind | None = None, - ) -> "CapturedSignal": + ) -> Self: """Build a signal from a device-returned packet. ``kind`` overrides the band read from the packet's type byte. A @@ -200,7 +202,7 @@ def from_packet( return cls( bytes(packet), kind, - data_to_pulses(packet), + tuple(data_to_pulses(packet)), packet[0x01], frequency_mhz, type_byte, @@ -259,6 +261,9 @@ async def _claim_window(self, new: weakref.ReferenceType) -> None: raise e.CaptureInProgressError( "A capture window is already open; close it with aclose() first" ) + if self._window is not prev: + # Another claimant got in during the two turns above. + raise e.CaptureInProgressError("A capture window is already open") self._window = new async def _send(self, command: int, data: bytes = b"") -> bytes: diff --git a/cli/broadlink_cli b/cli/broadlink_cli index 9512c7cb..7884f289 100644 --- a/cli/broadlink_cli +++ b/cli/broadlink_cli @@ -4,7 +4,7 @@ import asyncio import base64 import sys import time -from contextlib import aclosing +from contextlib import AsyncExitStack, aclosing from typing import List import broadlink @@ -96,6 +96,12 @@ args = parser.parse_args() async def main(): + async with AsyncExitStack() as stack: + await run_commands(stack) + + +async def run_commands(stack: AsyncExitStack): + """Run the requested commands; the device is closed with the stack.""" dev = None if args.device: @@ -110,6 +116,7 @@ async def main(): if args.host or args.device: dev = broadlink.gendevice(devtype, (host, DEFAULT_PORT), mac) + await stack.enter_async_context(dev) await dev.auth() if args.joinwifi: diff --git a/pyproject.toml b/pyproject.toml index cde5452c..af17ec80 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "python-broadlink" -version = "1.0.2" +version = "1.0.3" 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 82389f69..222c4ecd 100644 --- a/tests/test_capture.py +++ b/tests/test_capture.py @@ -168,7 +168,7 @@ async def go(): assert isinstance(sig, CapturedSignal) assert sig.packet == IR assert sig.kind is SignalKind.IR - assert sig.pulses == data_to_pulses(IR) + assert sig.pulses == tuple(data_to_pulses(IR)) assert sig.frequency_mhz is None assert fake.commands[0][0] == CMD_LEARN assert fake.count(CMD_LEARN) == 1 @@ -661,7 +661,7 @@ def test_parse_packet_round_trip(): parsed = parse_packet(packet) assert parsed.kind is kind assert parsed.repeat == 1 - assert parsed.pulses == data_to_pulses(packet) + assert parsed.pulses == tuple(data_to_pulses(packet)) for a, b in zip(pulses, parsed.pulses, strict=True): assert abs(a - b) <= 16 @@ -712,10 +712,16 @@ def test_signal_kind_flags(): assert SignalKind(0x26) is SignalKind.IR +def test_captured_signal_is_hashable(): + a = CapturedSignal.from_packet(RF, 433.92) + assert isinstance(hash(a), int) # a frozen value type belongs in a set + assert len({parse_packet(RF), parse_packet(RF)}) == 1 + + def test_captured_signal_from_packet(): sig = CapturedSignal.from_packet(RF, 433.92) assert sig.kind is SignalKind.RF_433 assert sig.repeat == 0 - assert sig.pulses == data_to_pulses(RF) + assert sig.pulses == tuple(data_to_pulses(RF)) assert sig.frequency_mhz == 433.92 assert sig.captured_at > 0 diff --git a/tests/test_transport.py b/tests/test_transport.py index 414bb044..770a45e0 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -75,7 +75,6 @@ async def __call__(self, local_addr=None, remote_addr=None, broadcast=False): def net(monkeypatch): fake = FakeNet() monkeypatch.setattr(device_module, "_open_endpoint", fake) - monkeypatch.setattr(broadlink, "_open_endpoint", fake) # Keep the retry loop from waiting on real time. monkeypatch.setattr(device_module, "DEFAULT_RETRY_INTVL", 0.005) return fake @@ -386,6 +385,30 @@ async def go(): assert run(go()) < 1.0 +def test_aclose_during_endpoint_open_does_not_leak(net, monkeypatch): + """aclose() landing while create_datagram_endpoint is still running must + not leave the freshly opened socket behind.""" + dev = fixed_device() + slow = net + + async def slow_open(**kwargs): + await asyncio.sleep(0.02) + return await slow(**kwargs) + + monkeypatch.setattr(device_module, "_open_endpoint", slow_open) + + async def go(): + task = asyncio.get_running_loop().create_task(dev.send_packet(0x6A, b"")) + await asyncio.sleep(0.005) # inside the slow open + await dev.aclose() + with pytest.raises(e.EndpointClosedError): + await task + + run(go()) + assert dev._transport is None + assert all(ep.closed for ep in net.endpoints) + + def test_host_change_reopens_endpoint(net): dev = fixed_device() @@ -475,18 +498,100 @@ async def go(): assert dev.decrypt(resp[0x38:])[0] == 9 -def test_reauth_is_not_attempted_twice(net): +def test_failed_reauth_returns_the_original_reply(net): + """When the library's own re-authentication fails, the caller must see + the reply the device gave to its request, exactly as 0.19.0 would have + shown it, so the caller's own recovery (Home Assistant's reauth flow) + still runs.""" dev = fixed_device() async def go(): await dev._endpoint() ep = net.endpoints[-1] expired = (make_response(dev, b"", error=0xFFF9), HOST) - ep.replies = [expired, expired] # request fails, auth fails - return await dev.send_packet(0x6A, b"") + ep.replies = [expired, expired] # request fails -7, auth fails -7 + resp = await dev.send_packet(0x6A, b"") + return ep, resp + ep, resp = run(go()) + assert int.from_bytes(resp[0x22:0x24], "little", signed=True) == -7 + types = [int.from_bytes(f[0x26:0x28], "little") for f, _ in ep.sent] + assert types == [0x6A, 0x65] # one auth attempt, no blind retry with pytest.raises(e.AuthorizationError): - run(go()) + e.check_error(resp[0x22:0x24]) + + +def test_locked_device_surfaces_as_the_original_error(net): + """Device locked in the app: request answered -7, auth answered -1. The + caller gets the -7 frame back (its check_error raises + AuthorizationError), and its own auth() call then sees the -1.""" + dev = fixed_device() + + async def go(): + await dev._endpoint() + ep = net.endpoints[-1] + + def sendto(data, addr=None): + ep.sent.append((bytes(data), addr or ep.remote_addr)) + ptype = int.from_bytes(data[0x26:0x28], "little") + error = 0xFFFF if ptype == 0x65 else 0xFFF9 # -1 to auth, -7 to requests + ep.protocol.queue.put_nowait((make_response(dev, b"", error=error), HOST)) + + ep.sendto = sendto + resp = await dev.send_packet(0x6A, b"") + code = int.from_bytes(resp[0x22:0x24], "little", signed=True) + with pytest.raises(e.AuthenticationError): + await dev.auth() + return ep, code + + ep, code = run(go()) + assert code == -7 + types = [int.from_bytes(f[0x26:0x28], "little") for f, _ in ep.sent] + assert types == [0x6A, 0x65, 0x65] + + +def test_request_queued_behind_an_auth_still_reauths_if_needed(net): + """The auth generation is read under the lock, so a request that was + queued while another caller's auth() ran, and still gets -7, performs + its own re-authentication instead of assuming the earlier one covers + it.""" + 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) + + async def go(): + await dev._endpoint() + ep = net.endpoints[-1] + auths = {"n": 0} + loop = asyncio.get_running_loop() + + 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: + auths["n"] += 1 + reply = auth_reply + elif auths["n"] < 2: + reply = make_response(dev, b"", error=0xFFF9) # still -7 after auth #1 + else: + reply = make_response(renewed, bytes([9]) + bytes(15)) + loop.call_later(0.002, ep.protocol.queue.put_nowait, (reply, ep.remote_addr)) + + ep.sendto = sendto + first_auth = loop.create_task(dev.auth()) + await asyncio.sleep(0) # let auth() take the lock first + resp = await dev.send_packet(0x6A, b"") + await first_auth + return ep, resp + + ep, resp = run(go()) + types = [int.from_bytes(f[0x26:0x28], "little") for f, _ in ep.sent] + assert types == [0x65, 0x6A, 0x65, 0x6A] + assert dev.decrypt(resp[0x38:])[0] == 9 def test_concurrent_callers_share_one_reauth(net):