1
0

lock.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403
  1. """Library to handle connection with Switchbot Lock."""
  2. from __future__ import annotations
  3. import asyncio
  4. import logging
  5. import time
  6. from typing import Any
  7. import aiohttp
  8. from bleak.backends.device import BLEDevice
  9. from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
  10. from ..api_config import SWITCHBOT_APP_API_BASE_URL, SWITCHBOT_APP_CLIENT_ID
  11. from ..const import (
  12. LockStatus,
  13. SwitchbotAccountConnectionError,
  14. SwitchbotApiError,
  15. SwitchbotAuthenticationError,
  16. SwitchbotModel,
  17. )
  18. from .device import SwitchbotDevice, SwitchbotOperationError
  19. COMMAND_HEADER = "57"
  20. COMMAND_GET_CK_IV = f"{COMMAND_HEADER}0f2103"
  21. COMMAND_LOCK_INFO = {
  22. SwitchbotModel.LOCK: f"{COMMAND_HEADER}0f4f8101",
  23. SwitchbotModel.LOCK_PRO: f"{COMMAND_HEADER}0f4f8102",
  24. }
  25. COMMAND_UNLOCK = {
  26. SwitchbotModel.LOCK: f"{COMMAND_HEADER}0f4e01011080",
  27. SwitchbotModel.LOCK_PRO: f"{COMMAND_HEADER}0f4e0101000080",
  28. }
  29. COMMAND_UNLOCK_WITHOUT_UNLATCH = {
  30. SwitchbotModel.LOCK: f"{COMMAND_HEADER}0f4e010110a0",
  31. SwitchbotModel.LOCK_PRO: f"{COMMAND_HEADER}0f4e01010000a0",
  32. }
  33. COMMAND_LOCK = {
  34. SwitchbotModel.LOCK: f"{COMMAND_HEADER}0f4e01011000",
  35. SwitchbotModel.LOCK_PRO: f"{COMMAND_HEADER}0f4e0101000000",
  36. }
  37. COMMAND_ENABLE_NOTIFICATIONS = f"{COMMAND_HEADER}0e01001e00008101"
  38. COMMAND_DISABLE_NOTIFICATIONS = f"{COMMAND_HEADER}0e00"
  39. MOVING_STATUSES = {LockStatus.LOCKING, LockStatus.UNLOCKING}
  40. BLOCKED_STATUSES = {LockStatus.LOCKING_STOP, LockStatus.UNLOCKING_STOP}
  41. REST_STATUSES = {LockStatus.LOCKED, LockStatus.UNLOCKED, LockStatus.NOT_FULLY_LOCKED}
  42. _LOGGER = logging.getLogger(__name__)
  43. COMMAND_RESULT_EXPECTED_VALUES = {1, 6}
  44. # The return value of the command is 1 when the command is successful.
  45. # The return value of the command is 6 when the command is successful but the battery is low.
  46. class SwitchbotLock(SwitchbotDevice):
  47. """Representation of a Switchbot Lock."""
  48. def __init__(
  49. self,
  50. device: BLEDevice,
  51. key_id: str,
  52. encryption_key: str,
  53. interface: int = 0,
  54. model: SwitchbotModel = SwitchbotModel.LOCK,
  55. **kwargs: Any,
  56. ) -> None:
  57. if len(key_id) == 0:
  58. raise ValueError("key_id is missing")
  59. elif len(key_id) != 2:
  60. raise ValueError("key_id is invalid")
  61. if len(encryption_key) == 0:
  62. raise ValueError("encryption_key is missing")
  63. elif len(encryption_key) != 32:
  64. raise ValueError("encryption_key is invalid")
  65. if model not in (SwitchbotModel.LOCK, SwitchbotModel.LOCK_PRO):
  66. raise ValueError("initializing SwitchbotLock with a non-lock model")
  67. self._iv = None
  68. self._cipher = None
  69. self._key_id = key_id
  70. self._encryption_key = bytearray.fromhex(encryption_key)
  71. self._notifications_enabled: bool = False
  72. self._model: SwitchbotModel = model
  73. super().__init__(device, None, interface, **kwargs)
  74. @staticmethod
  75. async def verify_encryption_key(
  76. device: BLEDevice,
  77. key_id: str,
  78. encryption_key: str,
  79. model: SwitchbotModel = SwitchbotModel.LOCK,
  80. **kwargs: Any,
  81. ) -> bool:
  82. try:
  83. lock = SwitchbotLock(
  84. device, key_id=key_id, encryption_key=encryption_key, model=model
  85. )
  86. except ValueError:
  87. return False
  88. try:
  89. lock_info = await lock.get_basic_info()
  90. except SwitchbotOperationError:
  91. return False
  92. return lock_info is not None
  93. @staticmethod
  94. async def api_request(
  95. session: aiohttp.ClientSession,
  96. subdomain: str,
  97. path: str,
  98. data: dict = None,
  99. headers: dict = None,
  100. ) -> dict:
  101. url = f"https://{subdomain}.{SWITCHBOT_APP_API_BASE_URL}/{path}"
  102. async with session.post(
  103. url,
  104. json=data,
  105. headers=headers,
  106. timeout=aiohttp.ClientTimeout(total=10),
  107. ) as result:
  108. if result.status > 299:
  109. raise SwitchbotApiError(
  110. f"Unexpected status code returned by SwitchBot API: {result.status}"
  111. )
  112. response = await result.json()
  113. if response["statusCode"] != 100:
  114. raise SwitchbotApiError(
  115. f"{response['message']}, status code: {response['statusCode']}"
  116. )
  117. return response["body"]
  118. # Old non-async method preserved for backwards compatibility
  119. @staticmethod
  120. def retrieve_encryption_key(device_mac: str, username: str, password: str):
  121. async def async_fn():
  122. async with aiohttp.ClientSession() as session:
  123. return await SwitchbotLock.async_retrieve_encryption_key(
  124. session, device_mac, username, password
  125. )
  126. return asyncio.run(async_fn())
  127. @staticmethod
  128. async def async_retrieve_encryption_key(
  129. session: aiohttp.ClientSession, device_mac: str, username: str, password: str
  130. ) -> dict:
  131. """Retrieve lock key from internal SwitchBot API."""
  132. device_mac = device_mac.replace(":", "").replace("-", "").upper()
  133. try:
  134. auth_result = await SwitchbotLock.api_request(
  135. session,
  136. "account",
  137. "account/api/v1/user/login",
  138. {
  139. "clientId": SWITCHBOT_APP_CLIENT_ID,
  140. "username": username,
  141. "password": password,
  142. "grantType": "password",
  143. "verifyCode": "",
  144. },
  145. )
  146. auth_headers = {"authorization": auth_result["access_token"]}
  147. except Exception as err:
  148. raise SwitchbotAuthenticationError(f"Authentication failed: {err}") from err
  149. try:
  150. userinfo = await SwitchbotLock.api_request(
  151. session, "account", "account/api/v1/user/userinfo", {}, auth_headers
  152. )
  153. if "botRegion" in userinfo and userinfo["botRegion"] != "":
  154. region = userinfo["botRegion"]
  155. else:
  156. region = "us"
  157. except Exception as err:
  158. raise SwitchbotAccountConnectionError(
  159. f"Failed to retrieve SwitchBot Account user details: {err}"
  160. ) from err
  161. try:
  162. device_info = await SwitchbotLock.api_request(
  163. session,
  164. f"wonderlabs.{region}",
  165. "wonder/keys/v1/communicate",
  166. {
  167. "device_mac": device_mac,
  168. "keyType": "user",
  169. },
  170. auth_headers,
  171. )
  172. return {
  173. "key_id": device_info["communicationKey"]["keyId"],
  174. "encryption_key": device_info["communicationKey"]["key"],
  175. }
  176. except Exception as err:
  177. raise SwitchbotAccountConnectionError(
  178. f"Failed to retrieve encryption key from SwitchBot Account: {err}"
  179. ) from err
  180. async def lock(self) -> bool:
  181. """Send lock command."""
  182. return await self._lock_unlock(
  183. COMMAND_LOCK[self._model], {LockStatus.LOCKED, LockStatus.LOCKING}
  184. )
  185. async def unlock(self) -> bool:
  186. """Send unlock command. If unlatch feature is enabled in EU firmware, also unlatches door"""
  187. return await self._lock_unlock(
  188. COMMAND_UNLOCK[self._model], {LockStatus.UNLOCKED, LockStatus.UNLOCKING}
  189. )
  190. async def unlock_without_unlatch(self) -> bool:
  191. """Send unlock command. This command will not unlatch the door."""
  192. return await self._lock_unlock(
  193. COMMAND_UNLOCK_WITHOUT_UNLATCH[self._model],
  194. {LockStatus.UNLOCKED, LockStatus.UNLOCKING, LockStatus.NOT_FULLY_LOCKED},
  195. )
  196. def _parse_basic_data(self, basic_data: bytes) -> dict[str, Any]:
  197. """Parse basic data from lock."""
  198. return {
  199. "battery": basic_data[1],
  200. "firmware": basic_data[2] / 10.0,
  201. }
  202. async def _lock_unlock(
  203. self, command: str, ignore_statuses: set[LockStatus]
  204. ) -> bool:
  205. status = self.get_lock_status()
  206. if status is None:
  207. await self.update()
  208. status = self.get_lock_status()
  209. if status in ignore_statuses:
  210. return True
  211. await self._enable_notifications()
  212. result = await self._send_command(command)
  213. status = self._check_command_result(result, 0, COMMAND_RESULT_EXPECTED_VALUES)
  214. # Also update the battery and firmware version
  215. if basic_data := await self._get_basic_info():
  216. self._last_full_update = time.monotonic()
  217. if len(basic_data) >= 3:
  218. self._update_parsed_data(self._parse_basic_data(basic_data))
  219. else:
  220. _LOGGER.warning("Invalid basic data received: %s", basic_data)
  221. self._fire_callbacks()
  222. return status
  223. async def get_basic_info(self) -> dict[str, Any] | None:
  224. """Get device basic status."""
  225. lock_raw_data = await self._get_lock_info()
  226. if not lock_raw_data:
  227. return None
  228. basic_data = await self._get_basic_info()
  229. if not basic_data:
  230. return None
  231. return self._parse_lock_data(lock_raw_data[1:]) | self._parse_basic_data(
  232. basic_data
  233. )
  234. def is_calibrated(self) -> Any:
  235. """Return True if lock is calibrated."""
  236. return self._get_adv_value("calibration")
  237. def get_lock_status(self) -> LockStatus:
  238. """Return lock status."""
  239. return self._get_adv_value("status")
  240. def is_door_open(self) -> bool:
  241. """Return True if door is open."""
  242. return self._get_adv_value("door_open")
  243. def is_unclosed_alarm_on(self) -> bool:
  244. """Return True if unclosed door alarm is on."""
  245. return self._get_adv_value("unclosed_alarm")
  246. def is_unlocked_alarm_on(self) -> bool:
  247. """Return True if lock unlocked alarm is on."""
  248. return self._get_adv_value("unlocked_alarm")
  249. def is_auto_lock_paused(self) -> bool:
  250. """Return True if auto lock is paused."""
  251. return self._get_adv_value("auto_lock_paused")
  252. def is_night_latch_enabled(self) -> bool:
  253. """Return True if Night Latch is enabled on EU firmware."""
  254. return self._get_adv_value("night_latch")
  255. async def _get_lock_info(self) -> bytes | None:
  256. """Return lock info of device."""
  257. _data = await self._send_command(
  258. key=COMMAND_LOCK_INFO[self._model], retry=self._retry_count
  259. )
  260. if not self._check_command_result(_data, 0, COMMAND_RESULT_EXPECTED_VALUES):
  261. _LOGGER.error("Unsuccessful, please try again")
  262. return None
  263. return _data
  264. async def _enable_notifications(self) -> bool:
  265. if self._notifications_enabled:
  266. return True
  267. result = await self._send_command(COMMAND_ENABLE_NOTIFICATIONS)
  268. if self._check_command_result(result, 0, COMMAND_RESULT_EXPECTED_VALUES):
  269. self._notifications_enabled = True
  270. return self._notifications_enabled
  271. async def _disable_notifications(self) -> bool:
  272. if not self._notifications_enabled:
  273. return True
  274. result = await self._send_command(COMMAND_DISABLE_NOTIFICATIONS)
  275. if self._check_command_result(result, 0, COMMAND_RESULT_EXPECTED_VALUES):
  276. self._notifications_enabled = False
  277. return not self._notifications_enabled
  278. def _notification_handler(self, _sender: int, data: bytearray) -> None:
  279. if self._notifications_enabled and self._check_command_result(data, 0, {0xF}):
  280. self._update_lock_status(data)
  281. else:
  282. super()._notification_handler(_sender, data)
  283. def _update_lock_status(self, data: bytearray) -> None:
  284. lock_data = self._parse_lock_data(self._decrypt(data[4:]))
  285. if self._update_parsed_data(lock_data):
  286. # We leave notifications enabled in case
  287. # the lock is operated manually before we
  288. # disconnect.
  289. self._reset_disconnect_timer()
  290. self._fire_callbacks()
  291. @staticmethod
  292. def _parse_lock_data(data: bytes) -> dict[str, Any]:
  293. return {
  294. "calibration": bool(data[0] & 0b10000000),
  295. "status": LockStatus((data[0] & 0b01110000) >> 4),
  296. "door_open": bool(data[0] & 0b00000100),
  297. "unclosed_alarm": bool(data[1] & 0b00100000),
  298. "unlocked_alarm": bool(data[1] & 0b00010000),
  299. }
  300. async def _send_command(
  301. self, key: str, retry: int | None = None, encrypt: bool = True
  302. ) -> bytes | None:
  303. if not encrypt:
  304. return await super()._send_command(key[:2] + "000000" + key[2:], retry)
  305. result = await self._ensure_encryption_initialized()
  306. if not result:
  307. _LOGGER.error("Failed to initialize encryption")
  308. return None
  309. encrypted = (
  310. key[:2] + self._key_id + self._iv[0:2].hex() + self._encrypt(key[2:])
  311. )
  312. result = await super()._send_command(encrypted, retry)
  313. return result[:1] + self._decrypt(result[4:])
  314. async def _ensure_encryption_initialized(self) -> bool:
  315. if self._iv is not None:
  316. return True
  317. result = await self._send_command(
  318. COMMAND_GET_CK_IV + self._key_id, encrypt=False
  319. )
  320. ok = self._check_command_result(result, 0, COMMAND_RESULT_EXPECTED_VALUES)
  321. if ok:
  322. self._iv = result[4:]
  323. return ok
  324. async def _execute_disconnect(self) -> None:
  325. await super()._execute_disconnect()
  326. self._iv = None
  327. self._cipher = None
  328. self._notifications_enabled = False
  329. def _get_cipher(self) -> Cipher:
  330. if self._cipher is None:
  331. self._cipher = Cipher(
  332. algorithms.AES128(self._encryption_key), modes.CTR(self._iv)
  333. )
  334. return self._cipher
  335. def _encrypt(self, data: str) -> str:
  336. if len(data) == 0:
  337. return ""
  338. encryptor = self._get_cipher().encryptor()
  339. return (encryptor.update(bytearray.fromhex(data)) + encryptor.finalize()).hex()
  340. def _decrypt(self, data: bytearray) -> bytes:
  341. if len(data) == 0:
  342. return b""
  343. decryptor = self._get_cipher().decryptor()
  344. return decryptor.update(data) + decryptor.finalize()