1
0

lock.py 13 KB

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