sblibs-pyswitchbot-368
Results
The prompt the agent sees
When an encrypted lock notification arrives while an expected device disconnection is in progress, the lock may already have cleared the initialization vector needed for decryption. The notification is then processed as normal, causing a `RuntimeError` instead of being safely ignored. In this state, lock status must not be updated from the late notification, and decryption must not fail when invoked with no initialization vector. Outside an expected disconnection, attempting to decrypt without an initialization vector should continue to report the decryption error.
What the tests pin: the device object tracks an expected disconnection in `_expected_disconnect` and the decryption vector in `_iv`. While `_expected_disconnect` is true and `_iv` is `None`, `_decrypt(data)` returns empty bytes (`b""`); otherwise, with `_iv` `None`, it raises `RuntimeError("Cannot decrypt: IV is None")` as before. When `_notification_handler` receives a lock notification while `_expected_disconnect` is true, it does not call `_update_lock_status` and logs, at debug level, a message containing "Ignoring lock notification during expected disconnect".
Hidden tests · 1 fail-to-pass, 89 pass-to-passrun after the agent submits, in a clean verifier
Test patch · 71 lines
diff --git a/tests/test_encrypted_device.py b/tests/test_encrypted_device.py
index 3ebfc886..1ed071a7 100644
--- a/tests/test_encrypted_device.py
+++ b/tests/test_encrypted_device.py
@@ -365,3 +365,22 @@ async def test_empty_data_encryption_decryption() -> None:
# Test empty decryption
decrypted = device._decrypt(bytearray())
assert decrypted == b""
+
+
+@pytest.mark.asyncio
+async def test_decrypt_with_none_iv_during_disconnect() -> None:
+ """Test that decryption returns empty bytes when IV is None during expected disconnect."""
+ device = create_encrypted_device()
+
+ # Simulate disconnection in progress
+ device._expected_disconnect = True
+ device._iv = None
+
+ # Should return empty bytes instead of raising
+ result = device._decrypt(bytearray(b"encrypted_data"))
+ assert result == b""
+
+ # Verify it still raises when not disconnecting
+ device._expected_disconnect = False
+ with pytest.raises(RuntimeError, match="Cannot decrypt: IV is None"):
+ device._decrypt(bytearray(b"encrypted_data"))
diff --git a/tests/test_lock.py b/tests/test_lock.py
index fc4f8337..6994e952 100644
--- a/tests/test_lock.py
+++ b/tests/test_lock.py
@@ -1,3 +1,4 @@
+import logging
from unittest.mock import AsyncMock, Mock, patch
import pytest
@@ -478,6 +479,34 @@ def test_notification_handler_not_enabled(model: str):
mock_super.assert_called_once()
+@pytest.mark.parametrize(
+ "model",
+ [
+ SwitchbotModel.LOCK,
+ SwitchbotModel.LOCK_LITE,
+ SwitchbotModel.LOCK_PRO,
+ SwitchbotModel.LOCK_ULTRA,
+ ],
+)
+def test_notification_handler_during_disconnect(
+ model: str, caplog: pytest.LogCaptureFixture
+) -> None:
+ """Test _notification_handler during expected disconnect."""
+ device = create_device_for_command_testing(model)
+ device._notifications_enabled = True
+ device._expected_disconnect = True
+ data = bytearray(b"\x0f\x00\x00\x00\x80\x00\x00\x00\x00\x00")
+ with (
+ patch.object(device, "_update_lock_status") as mock_update,
+ caplog.at_level(logging.DEBUG),
+ ):
+ device._notification_handler(0, data)
+ # Should not update lock status during disconnect
+ mock_update.assert_not_called()
+ # Should log debug message
+ assert "Ignoring lock notification during expected disconnect" in caplog.text
+
+
@pytest.mark.parametrize(
"model",
[
Reference fix · 2 files, +12 −0the upstream merge, used only for grading calibration
The agent could not see this: the repository holds one commit and the sandbox has no network. Leak audit.
switchbot/devices/device.py, switchbot/devices/lock.py
diff --git a/switchbot/devices/device.py b/switchbot/devices/device.py
index a80fb41c..1970ef92 100644
--- a/switchbot/devices/device.py
+++ b/switchbot/devices/device.py
@@ -955,6 +955,12 @@ def _decrypt(self, data: bytearray) -> bytes:
if len(data) == 0:
return b""
if self._iv is None:
+ if self._expected_disconnect:
+ _LOGGER.debug(
+ "%s: Cannot decrypt, IV is None during expected disconnect",
+ self.name,
+ )
+ return b""
raise RuntimeError("Cannot decrypt: IV is None")
decryptor = self._get_cipher().decryptor()
return decryptor.update(data) + decryptor.finalize()
diff --git a/switchbot/devices/lock.py b/switchbot/devices/lock.py
index c8be8915..1ac41f82 100644
--- a/switchbot/devices/lock.py
+++ b/switchbot/devices/lock.py
@@ -214,6 +214,12 @@ async def _disable_notifications(self) -> bool:
def _notification_handler(self, _sender: int, data: bytearray) -> None:
if self._notifications_enabled and self._check_command_result(data, 0, {0xF}):
+ if self._expected_disconnect:
+ _LOGGER.debug(
+ "%s: Ignoring lock notification during expected disconnect",
+ self.name,
+ )
+ return
self._update_lock_status(data)
else:
super()._notification_handler(_sender, data)
GLM-5.3 Flash · attempt 2 · solved · 15 steps · $0.002
Submitted diff · 34 lines
diff --git a/switchbot/devices/device.py b/switchbot/devices/device.py
index a80fb41..2bfd538 100644
--- a/switchbot/devices/device.py
+++ b/switchbot/devices/device.py
@@ -955,6 +955,11 @@ class SwitchbotEncryptedDevice(SwitchbotDevice):
if len(data) == 0:
return b""
if self._iv is None:
+ if self._expected_disconnect:
+ # The IV has already been cleared because we are
+ # expecting a disconnect; late notifications must
+ # not raise here.
+ return b""
raise RuntimeError("Cannot decrypt: IV is None")
decryptor = self._get_cipher().decryptor()
return decryptor.update(data) + decryptor.finalize()
diff --git a/switchbot/devices/lock.py b/switchbot/devices/lock.py
index c8be891..136a4dc 100644
--- a/switchbot/devices/lock.py
+++ b/switchbot/devices/lock.py
@@ -213,6 +213,13 @@ class SwitchbotLock(SwitchbotSequenceDevice, SwitchbotEncryptedDevice):
return not self._notifications_enabled
def _notification_handler(self, _sender: int, data: bytearray) -> None:
+ if self._expected_disconnect and self._check_command_result(data, 0, {0xF}):
+ _LOGGER.debug(
+ "%s: Ignoring lock notification during expected disconnect: %s",
+ self.name,
+ data,
+ )
+ return
if self._notifications_enabled and self._check_command_result(data, 0, {0xF}):
self._update_lock_status(data)
else:step 1
$ ls /repo; grep -rn "_expected_disconnect\|_iv\b" /repo --include=*.py -l
output · 13 lines
LICENSE MANIFEST.in PySwitchbot.egg-info README.md pyproject.toml requirements.txt requirements_dev.txt scripts setup.py switchbot tests /repo/switchbot/devices/device.py
1 more lines
/repo/tests/test_encrypted_device.py
step 2
$ grep -n "_iv\|_expected_disconnect\|_decrypt\|_notification_handler\|_update_lock_status" /repo/switchbot/devices/device.py | head -50
output · 17 lines
158: self._expected_disconnect = False 373: self._expected_disconnect = False 380: if self._expected_disconnect: 445: self._expected_disconnect = True 490: def _notification_handler(self, _sender: int, data: bytearray) -> None: 500: await self._client.start_notify(self._read_char, self._notification_handler) 768: self._iv: bytes | None = None 890: key[:2] + self._key_id + self._iv[0:2].hex() + self._encrypt(key[2:]) 901: return result[:1] + self._decrypt(result[4:]) 907: if self._iv is not None: 925: self._iv = result[4:] 933: self._iv = None
5 more lines
939: if self._iv is None: 942: algorithms.AES128(self._encryption_key), modes.CTR(self._iv) 949: if self._iv is None: 954: def _decrypt(self, data: bytearray) -> bytes: 957: if self._iv is None:
step 3
$ sed -n 480,520p /repo/switchbot/devices/device.py; sed -n 940,975p /repo/switchbot/devices/device.py
output · 31 lines
await self._execute_forced_disconnect()
raise
except BLEAK_RETRY_EXCEPTIONS as ex:
# Disconnect so we can reset state and try again
_LOGGER.debug(
"%s: RSSI: %s; Disconnecting due to error: %s", self.name, self.rssi, ex
)
await self._execute_forced_disconnect()
raise
def _notification_handler(self, _sender: int, data: bytearray) -> None:
"""Handle notification responses."""19 more lines
if self._notify_future and not self._notify_future.done():
self._notify_future.set_result(data)
return
_LOGGER.debug("%s: Received unsolicited notification: %s", self.name, data)
async def _start_notify(self) -> None:
"""Start notification."""
_LOGGER.debug("%s: Subscribe to notifications; RSSI: %s", self.name, self.rssi)
await
…[1878 chars omitted]…
nd uses the
data from the device instead.
"""
def update_from_advertisement(self, advertisement: SwitchBotAdvertisement) -> None:
super().update_from_advertisement(advertisement)
if self._client and self._client.is_connected:
# We do not consume the advertisement data if we are connected
# to the device. This is because the advertisement data is notstep 4
$ grep -rn "_decrypt\|_update_lock_status\|_notification_handler" /repo/switchbot/devices/lock.py | head;
output · 5 lines
215: def _notification_handler(self, _sender: int, data: bytearray) -> None: 217: self._update_lock_status(data) 219: super()._notification_handler(_sender, data) 221: def _update_lock_status(self, data: bytearray) -> None: 222: lock_data = self._parse_lock_data(self._decrypt(data[4:]), self._model)
step 5
$ sed -n 205,235p /repo/switchbot/devices/lock.py
output · 32 lines
return self._notifications_enabled
async def _disable_notifications(self) -> bool:
if not self._notifications_enabled:
return True
result = await self._send_command(COMMAND_DISABLE_NOTIFICATIONS)
if self._check_command_result(result, 0, COMMAND_RESULT_EXPECTED_VALUES):
self._notifications_enabled = False
return not self._notifications_enabled
def _notification_handler(self, _sender: int, data: bytearray) -> None:
if self._notifications_enabled and self._check_command_result(data, 0, {0xF}):20 more lines
self._update_lock_status(data)
else:
super()._notification_handler(_sender, data)
def _update_lock_status(self, data: bytearray) -> None:
lock_data = self._parse_lock_data(self._decrypt(data[4:]), self._model)
if self._update_parsed_data(lock_data):
# We
…[77 chars omitted]…
lly before we
# disconnect.
self._reset_disconnect_timer()
self._fire_callbacks()
@staticmethod
def _parse_lock_data(data: bytes, model: SwitchbotModel) -> dict[str, Any]:
if model == SwitchbotModel.LOCK:
return {
"calibration": bool(data[0] & 0b10000000),
"status": LockStatus((data[0] & 0b01110000) >> 4),step 6
$ cd /repo && python - <<'PY'
output · 3 lines
47:_LOGGER = logging.getLogger(__name__)
136: _LOGGER.warning("Invalid basic data received: %s", basic_data)
146: _LOGGER.debug(step 7
$ cd /repo && python -m pytest tests/test_encrypted_device.py -q 2>&1 | tail -3; ls tests | head -30
output · 23 lines
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html 16 passed, 16 warnings in 1.08s __init__.py __pycache__ test_adv_parser.py test_air_purifier.py test_base_cover.py test_blind_tilt.py test_bulb.py test_ceiling_light.py test_colormode_imports.py test_curtain.py
11 more lines
test_encrypted_device.py test_evaporative_humidifier.py test_fan.py test_helpers.py test_hub2.py test_hub3.py test_lock.py test_relay_switch.py test_roller_shade.py test_strip_light.py test_vacuum.py
step 8
$ cd /repo && python -m pytest tests/test_lock.py -q 2>&1 | tail -3 && python - <<'PY'
output · 6 lines
FAILED tests/test_lock.py::test_update_lock_status[WoLockPro] - RuntimeError:... FAILED tests/test_lock.py::test_update_lock_status[Lock Ultra] - RuntimeError... 36 failed, 77 passed, 106 warnings in 2.69s Traceback (most recent call last): File "<stdin>", line 16, in <module> AttributeError: property 'name' of 'SwitchbotLock' object has no setter
step 9
$ cd /repo && git stash && python -m pytest tests/test_lock.py -q 2>&1 | tail -3; git stash pop
output · 13 lines
Saved working directory and index state WIP on master: 5a8dc60 base FAILED tests/test_lock.py::test_update_lock_status[WoLockPro] - RuntimeError:... FAILED tests/test_lock.py::test_update_lock_status[Lock Ultra] - RuntimeError... 36 failed, 77 passed, 106 warnings in 2.68s On branch master Changes not staged for commit: (use "git add <file>..." to update what will be committed) (use "git restore <file>..." to discard changes in working directory) modified: switchbot/devices/device.py modified: switchbot/devices/lock.py no changes added to commit (use "git add" and/or "git commit -a")
1 more lines
Dropped refs/stash@{0} (31c320696581f69a421e808d0a986af917f7a9a9)step 10
Pre-existing failures, unchanged by my edit. Fixing my reproducer:
$ cd /repo && python - <<'PY'
output · 9 lines
Traceback (most recent call last):
File "<stdin>", line 16, in <module>
File "/repo/switchbot/devices/lock.py", line 219, in _notification_handler
self.name,
^^^^^^^^^
File "/repo/switchbot/devices/device.py", line 279, in name
return f"{self._device.name} ({self._device.address})"
^^^^^^^^^^^^
AttributeError: 'SwitchbotLock' object has no attribute '_device'step 11
$ cd /repo && python - <<'PY'
output · 1 lines
ok
step 12
$ cd /repo && python -m pytest tests/test_encrypted_device.py tests/test_lock.py -q 2>&1 | tail -2 && git diff
output · 30 lines
FAILED tests/test_lock.py::test_update_lock_status[Lock Ultra] - RuntimeError...
40 failed, 89 passed, 121 warnings in 3.36s
diff --git a/switchbot/devices/device.py b/switchbot/devices/device.py
index a80fb41..2bfd538 100644
--- a/switchbot/devices/device.py
+++ b/switchbot/devices/device.py
@@ -955,6 +955,11 @@ class SwitchbotEncryptedDevice(SwitchbotDevice):
if len(data) == 0:
return b""
if self._iv is None:
+ if self._expected_disconnect:
+ # The IV has already been cleared because we are18 more lines
+ # expecting a disconnect; late notifications must
+ # not raise here.
+ return b""
raise RuntimeError("Cannot decrypt: IV is None")
decryptor = self._get_cipher().decryptor()
return decryptor.update(data) + decryptor.finalize()
diff --git a/switchbot/devices/lock.
…[369 chars omitted]…
ed_disconnect and self._check_command_result(data, 0, {0xF}):
+ _LOGGER.debug(
+ "%s: Ignoring lock notification during expected disconnect: %s",
+ self.name,
+ data,
+ )
+ return
if self._notifications_enabled and self._check_command_result(data, 0, {0xF}):
self._update_lock_status(data)
else:step 13
$ cd /repo && python -m pytest tests/test_lock.py -q 2>&1 | grep FAILED | head -8
output · 8 lines
FAILED tests/test_lock.py::test_parse_basic_data - RuntimeError: There is no ... FAILED tests/test_lock.py::test_is_calibrated[WoLock] - RuntimeError: There i... FAILED tests/test_lock.py::test_is_calibrated[Lock Lite] - RuntimeError: Ther... FAILED tests/test_lock.py::test_is_calibrated[WoLockPro] - RuntimeError: Ther... FAILED tests/test_lock.py::test_is_calibrated[Lock Ultra] - RuntimeError: The... FAILED tests/test_lock.py::test_get_lock_status[WoLock] - RuntimeError: There... FAILED tests/test_lock.py::test_get_lock_status[Lock Lite] - RuntimeError: Th... FAILED tests/test_lock.py::test_get_lock_status[WoLockPro] - RuntimeError: Th...
step 14
$ cd /repo && python -m pytest tests/test_lock.py -q 2>&1 | grep FAILED | sort > /tmp/now.txt; git stash -q; python -m pytest tests/test_lock.py -q 2>&1 | grep FAILED | sort > /tmp/base.txt; git stash pop -q; diff /tmp/base.txt /tmp/now.txt && echo SAME
output · 1 lines
SAME
step 15
Failure sets are identical to the pre-change baseline (pre-existing environment failures). Fix is complete.
$ done: {"notes": "In SwitchbotEncryptedDevice._decrypt, return b\"\" instead of raising when _iv is None and _expected_disconnect is set. In SwitchbotLock._notification_handler, ignore lock notifications (log at debug \"Ignoring lock notification during expected disconnect\") while _expected_disconnect is true, skipping _update_lock_status. Reproducer passes; test_lock.py failures are identical to the pre-change baseline."}output · 1 lines
submitted.