Fix lock encryption key retrieval

requirements.txt

@@ -1,5 +1,4 @@
 bleak>=0.17.0
 bleak-retry-connector>=2.9.0
 cryptography>=38.0.3
-boto3>=1.20.24
 requests>=2.28.1

requirements_dev.txt

@@ -3,5 +3,4 @@ pytest-cov
 bleak>=0.17.0
 bleak-retry-connector>=3.4.0
 cryptography>=38.0.3
-boto3>=1.20.24
 requests>=2.28.1

setup.py

@@ -8,7 +8,6 @@ setup(
         "bleak-retry-connector>=3.4.0",
         "cryptography>=39.0.0",
         "pyOpenSSL>=23.0.0",
-        "boto3>=1.20.24",
         "requests>=2.28.1",
     ],
     version="0.44.1",

switchbot/api_config.py

@@ -1,13 +1,6 @@
 # Those values have been obtained from the following files in SwitchBot Android app
 # That's how you can verify them yourself
 # /assets/switchbot_config.json
-# /res/raw/amplifyconfiguration.json
-# /res/raw/awsconfiguration.json
 
-SWITCHBOT_APP_API_BASE_URL = "https://l9ren7efdj.execute-api.us-east-1.amazonaws.com"
-SWITCHBOT_APP_COGNITO_POOL = {
-    "PoolId": "us-east-1_x1fixo5LC",
-    "AppClientId": "66r90hdllaj4nnlne4qna0muls",
-    "AppClientSecret": "1v3v7vfjsiggiupkeuqvsovg084e3msbefpj9rgh611u30uug6t8",
-    "Region": "us-east-1",
-}
+SWITCHBOT_APP_API_BASE_URL = "api.switchbot.net"
+SWITCHBOT_APP_CLIENT_ID = "5nnwmhmsa9xxskm14hd85lm9bm"

switchbot/const.py

@@ -10,6 +10,14 @@ DEFAULT_RETRY_TIMEOUT = 1
 DEFAULT_SCAN_TIMEOUT = 5
 
 
+class SwitchbotApiError(RuntimeError):
+    """Raised when API call fails.
+
+    This exception inherits from RuntimeError to avoid breaking existing code
+    but will be changed to Exception in a future release.
+    """
+
+
 class SwitchbotAuthenticationError(RuntimeError):
     """Raised when authentication fails.
 

switchbot/devices/lock.py

@@ -1,24 +1,21 @@
 """Library to handle connection with Switchbot Lock."""
 from __future__ import annotations
 
-import base64
-import hashlib
-import hmac
 import json
 import logging
 import time
 from typing import Any
 
-import boto3
 import requests
 from bleak.backends.device import BLEDevice
 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
 
-from ..api_config import SWITCHBOT_APP_API_BASE_URL, SWITCHBOT_APP_COGNITO_POOL
+from ..api_config import SWITCHBOT_APP_API_BASE_URL, SWITCHBOT_APP_CLIENT_ID
 from ..const import (
     LockStatus,
     SwitchbotAccountConnectionError,
     SwitchbotAuthenticationError,
+    SwitchbotApiError,
 )
 from .device import SwitchbotDevice, SwitchbotOperationError
 
@@ -86,77 +83,67 @@ class SwitchbotLock(SwitchbotDevice):
 
         return lock_info is not None
 
+    @staticmethod
+    def api_request(subdomain: str, path: str, data: dict, headers: dict = None):
+        result = requests.post(
+            url=f"https://{subdomain}.{SWITCHBOT_APP_API_BASE_URL}/{path}",
+            headers=headers,
+            json=data,
+            timeout=5,
+        )
+
+        if result.status_code > 299:
+            raise SwitchbotApiError(f"Unexpected status code returned by SwitchBot API: {result.status_code}")
+
+        response = json.loads(result.content)
+        if response["statusCode"] != 100:
+            raise SwitchbotApiError(
+                f"{response['message']}, status code: {response['statusCode']}"
+            )
+
+        return response["body"]
+
     @staticmethod
     def retrieve_encryption_key(device_mac: str, username: str, password: str):
         """Retrieve lock key from internal SwitchBot API."""
         device_mac = device_mac.replace(":", "").replace("-", "").upper()
-        msg = bytes(username + SWITCHBOT_APP_COGNITO_POOL["AppClientId"], "utf-8")
-        secret_hash = base64.b64encode(
-            hmac.new(
-                SWITCHBOT_APP_COGNITO_POOL["AppClientSecret"].encode(),
-                msg,
-                digestmod=hashlib.sha256,
-            ).digest()
-        ).decode()
-
-        cognito_idp_client = boto3.client(
-            "cognito-idp", region_name=SWITCHBOT_APP_COGNITO_POOL["Region"]
-        )
+
         try:
-            auth_response = cognito_idp_client.initiate_auth(
-                ClientId=SWITCHBOT_APP_COGNITO_POOL["AppClientId"],
-                AuthFlow="USER_PASSWORD_AUTH",
-                AuthParameters={
-                    "USERNAME": username,
-                    "PASSWORD": password,
-                    "SECRET_HASH": secret_hash,
-                },
-            )
-        except cognito_idp_client.exceptions.NotAuthorizedException as err:
-            raise SwitchbotAuthenticationError(
-                f"Failed to authenticate: {err}"
-            ) from err
+            auth_result = SwitchbotLock.api_request("account", "account/api/v1/user/login", {
+                "clientId": SWITCHBOT_APP_CLIENT_ID,
+                "username": username,
+                "password": password,
+                "grantType": "password",
+                "verifyCode": ""
+            })
+            auth_headers = {"authorization": auth_result["access_token"]}
         except Exception as err:
             raise SwitchbotAuthenticationError(
-                f"Unexpected error during authentication: {err}"
+                f"Authentication failed: {err}"
             ) from err
 
-        if (
-            auth_response is None
-            or "AuthenticationResult" not in auth_response
-            or "AccessToken" not in auth_response["AuthenticationResult"]
-        ):
-            raise SwitchbotAuthenticationError("Unexpected authentication response")
+        try:
+            userinfo = SwitchbotLock.api_request("account", "account/api/v1/user/userinfo", {}, auth_headers)
+            region = userinfo["botRegion"]
+        except Exception as err:
+            raise SwitchbotAccountConnectionError(
+                f"Failed to retrieve SwitchBot Account user details: {err}"
+            ) from err
 
-        access_token = auth_response["AuthenticationResult"]["AccessToken"]
         try:
-            key_response = requests.post(
-                url=SWITCHBOT_APP_API_BASE_URL + "/developStage/keys/v1/communicate",
-                headers={"authorization": access_token},
-                json={
-                    "device_mac": device_mac,
-                    "keyType": "user",
-                },
-                timeout=10,
-            )
-        except requests.exceptions.RequestException as err:
+            device_info = SwitchbotLock.api_request(f"wonderlabs.{region}", "wonder/keys/v1/communicate", {
+                "device_mac": device_mac,
+                "keyType": "user",
+            }, auth_headers)
+
+            return {
+                "key_id": device_info["communicationKey"]["keyId"],
+                "encryption_key": device_info["communicationKey"]["key"],
+            }
+        except Exception as err:
             raise SwitchbotAccountConnectionError(
                 f"Failed to retrieve encryption key from SwitchBot Account: {err}"
             ) from err
-        if key_response.status_code > 299:
-            raise SwitchbotAuthenticationError(
-                f"Unexpected status code returned by SwitchBot Account API: {key_response.status_code}"
-            )
-        key_response_content = json.loads(key_response.content)
-        if key_response_content["statusCode"] != 100:
-            raise SwitchbotAuthenticationError(
-                f"Unexpected status code returned by SwitchBot API: {key_response_content['statusCode']}"
-            )
-
-        return {
-            "key_id": key_response_content["body"]["communicationKey"]["keyId"],
-            "encryption_key": key_response_content["body"]["communicationKey"]["key"],
-        }
 
     async def lock(self) -> bool:
         """Send lock command."""