Repository: mill1000/midea-msmart
Pull request #195 (2025-03-20 17:14:37)

Remove legacy CLI
Project has been around long enough that users should be familiar with the new cli `msmart-ng`

mill1000: Unit test failure is expected

Patch:
diff --git a/msmart/cli.py b/msmart/cli.py
index cb0dfd6..c23bde1 100644
--- a/msmart/cli.py
+++ b/msmart/cli.py
@@ -5,8 +5,8 @@ import logging
 from typing import NoReturn
 
 from msmart import __version__
-from msmart.cloud import Cloud, CloudError
-from msmart.const import CLOUD_CREDENTIALS, DEFAULT_CLOUD_REGION
+from msmart.cloud import CloudError, NetHomePlusCloud, SmartHomeCloud
+from msmart.const import DEFAULT_CLOUD_REGION
 from msmart.device import AirConditioner as AC
 from msmart.discover import Discover
 from msmart.lan import AuthenticationError
@@ -14,6 +14,8 @@ from msmart.utils import MideaIntEnum
 
 _LOGGER = logging.getLogger(__name__)
 
+# Use NetHome Plus cloud as default
+CLOUD_CREDENTIALS = NetHomePlusCloud.CLOUD_CREDENTIALS
 
 DEFAULT_CLOUD_ACCOUNT, DEFAULT_CLOUD_PASSWORD = CLOUD_CREDENTIALS[DEFAULT_CLOUD_REGION]
 
@@ -235,7 +237,11 @@ async def _download(args) -> None:
         exit(1)
 
     # Get cloud connection
-    cloud = Cloud(args.region, account=args.account, password=args.password)
+    cloud = SmartHomeCloud(
+        args.region,
+        account=args.account,
+        password=args.password
+    )
     try:
         await cloud.login()
     except CloudError as e:
@@ -294,7 +300,7 @@ def main() -> NoReturn:
         description="Command line utility for msmart-ng."
     )
     parser.add_argument("-v", "--version",
-                        action="version", version=f"msmart version: {__version__}")
+                        action="version", version=f"msmart-ng version: {__version__}")
     subparsers = parser.add_subparsers(title="Command", dest="command",
                                        required=True)
 
@@ -307,14 +313,11 @@ def main() -> NoReturn:
                                choices=CLOUD_CREDENTIALS.keys(),
                                default=DEFAULT_CLOUD_REGION)
     common_parser.add_argument("--account",
-                               help="Manually specify a MSmart username for cloud authentication.",
+                               help="Manually specify a username for cloud authentication.",
                                default=None)
     common_parser.add_argument("--password",
-                               help="Manually specify a MSmart password for cloud authentication.",
+                               help="Manually specify a password for cloud authentication.",
                                default=None)
-    common_parser.add_argument("--china",
-                               help="Use China server for discovery and authentication. Username and password must be specified.",
-                               action="store_true")
 
     # Setup discover parser
     discover_parser = subparsers.add_parser("discover",
@@ -390,41 +393,5 @@ def main() -> NoReturn:
     _run(parser.parse_args())
 
 
-def _legacy_main() -> NoReturn:
-    """Main entry point for legacy midea-discover command."""
-
-    async def _wrap_discover(args) -> None:
-        """Wrapper method to mimic legacy behavior."""
-        # Map old args to new names as needed
-        args.host = args.ip
-
-        # Output legacy information
-        _LOGGER.info("msmart version: %s", __version__)
-        _LOGGER.info(
-            "Only supports AC devices. Only supports MSmartHome and 美的美居.")
-
-        await _discover(args)
-
-    parser = argparse.ArgumentParser(
-        description="Discover Midea devices and print device information.",
-        formatter_class=argparse.ArgumentDefaultsHelpFormatter)
-    parser.add_argument(
-        "-d", "--debug", help="Enable debug logging.", action="store_true")
-    parser.add_argument(
-        "-a", "--account", help="MSmartHome or 美的美居 account username.", default=DEFAULT_CLOUD_ACCOUNT)
-    parser.add_argument(
-        "-p", "--password", help="MSmartHome or 美的美居 account password.", default=DEFAULT_CLOUD_PASSWORD)
-    parser.add_argument(
-        "-i", "--ip", help="IP address of a device. Useful if broadcasts don't work, or to query a single device.")
-    parser.add_argument(
-        "-c", "--count", help="Number of broadcast packets to send.", default=3, type=int)
-    parser.add_argument("--china", help="Use China server.",
-                        action="store_true")
-    parser.set_defaults(func=_wrap_discover)
-
-    # Run with args
-    _run(parser.parse_args())
-
-
 if __name__ == "__main__":
     main()
diff --git a/msmart/cloud.py b/msmart/cloud.py
index e69599f..dd0ad62 100644
--- a/msmart/cloud.py
+++ b/msmart/cloud.py
@@ -1,4 +1,4 @@
-"""Module for minimal Midea cloud API access."""
+"""Module for minimal Midea cloud access."""
 import hashlib
 import hmac
 import json
@@ -8,12 +8,13 @@ from asyncio import Lock
 from datetime import datetime, timezone
 from secrets import token_hex, token_urlsafe
 from typing import Any, Callable, Optional
+from urllib.parse import unquote_plus, urlencode, urlparse
 
 import httpx
 from Crypto.Cipher import AES
 from Crypto.Util import Padding
 
-from msmart.const import CLOUD_CREDENTIALS, DEFAULT_CLOUD_REGION, DeviceType
+from msmart.const import DEFAULT_CLOUD_REGION, DeviceType
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -36,36 +37,30 @@ class ApiError(CloudError):
         return f"Code: {self.code}, Message: {self.message}"
 
 
-class Cloud:
-    """Class for minimal Midea cloud API access."""
+class BaseCloud:
+    """Base class for minimal Midea cloud access."""
 
     # Misc constants for the API
+    APP_ID = ""
     CLIENT_TYPE = 1  # Android
     FORMAT = 2  # JSON
     LANGUAGE = "en_US"
-    APP_ID = "1010"
-    SRC = "1010"
     DEVICE_ID = token_hex(8)  # Random device ID
 
-    # Base URLs
-    BASE_URL = "https://mp-prod.appsmb.com"
-    BASE_URL_CHINA = "https://mp-prod.smartmidea.net"
-
     # Default number of request retries
     RETRIES = 3
 
+    CLOUD_CREDENTIALS = {}
+
     def __init__(self,
-                 region: str = DEFAULT_CLOUD_REGION,
+                 base_url: str,
+                 region: Optional[str],
+                 account: Optional[str],
+                 password: Optional[str],
                  *,
-                 account: Optional[str] = None,
-                 password: Optional[str] = None,
-                 use_china_server: bool = False,
                  get_async_client: Optional[
                      Callable[..., httpx.AsyncClient]] = None,
                  ) -> None:
-        # Allow override Chia server from environment
-        if os.getenv("MIDEA_CHINA_SERVER", "0") == "1":
-            use_china_server = True
 
         # Validate incoming credentials and region
         if account and password:
@@ -75,51 +70,40 @@ class Cloud:
             raise ValueError("Account and password must be specified.")
         else:
             try:
-                self._account, self._password = CLOUD_CREDENTIALS[region]
+                self._account, self._password = self.CLOUD_CREDENTIALS[region]
             except KeyError:
                 raise ValueError(f"Unknown cloud region '{region}'.")
 
-        # Attributes that holds the login information of the current user
+        self._api_lock = Lock()
+        self._base_url = base_url
         self._login_id = None
-        self._access_token = ""
         self._session = {}
 
-        self._api_lock = Lock()
-        self._security = _Security(use_china_server)
-
-        self._base_url = Cloud.BASE_URL_CHINA if use_china_server else Cloud.BASE_URL
-
         # Setup method for getting a client
         self._get_async_client = get_async_client if get_async_client else httpx.AsyncClient
 
-        _LOGGER.info("Using Midea cloud server: %s (China: %s).",
-                     self._base_url, use_china_server)
-
     def _timestamp(self) -> str:
         """Format a timestamp for the API."""
         return datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S")
 
     def _parse_response(self, response) -> Any:
-        """Parse a response from the API."""
-
-        _LOGGER.debug("API response: %s", response.text)
-        body = json.loads(response.text)
-
-        response_code = int(body["code"])
-        if response_code == 0:
-            return body["data"]
-
-        raise ApiError(body["msg"], code=response_code)
-
-    async def _post_request(self, url: str, headers: dict[str, Any],
-                            contents: str, retries: int = RETRIES) -> Optional[dict]:
-        """Post a request to the API."""
+        """Parse a response from the cloud."""
+        raise NotImplementedError
+
+    async def _post_request(self,
+                            url: str,
+                            headers: Optional[dict[str, Any]] = None,
+                            raw_data: Optional[str] = None,
+                            form_data: Optional[dict[str, Any]] = None,
+                            retries: int = RETRIES
+                            ) -> Optional[dict]:
+        """Post a request to the cloud."""
 
         async with self._get_async_client() as client:
             while retries > 0:
                 try:
                     # Post request and handle bad status code
-                    r = await client.post(url, headers=headers, content=contents, timeout=10.0)
+                    r = await client.post(url, headers=headers, content=raw_data, data=form_data, timeout=10.0)
                     r.raise_for_status()
 
                     # Parse the response
@@ -134,7 +118,119 @@ class Cloud:
                     raise CloudError("Request failed.") from e
 
     async def _api_request(self, endpoint: str, body: dict[str, Any]) -> Optional[dict]:
-        """Make a request to the Midea cloud return the results."""
+        """Make a request to the cloud and return the results."""
+        raise NotImplementedError
+
+    def _build_request_body(self, data: dict[str, Any]) -> dict[str, Any]:
+        """Build a request body."""
+
+        # Set up the initial body
+        body = {
+            "appId": self.APP_ID,
+            "src": self.APP_ID,
+            "format": BaseCloud.FORMAT,
+            "clientType": BaseCloud.CLIENT_TYPE,
+            "language": BaseCloud.LANGUAGE,
+            "deviceId": BaseCloud.DEVICE_ID,
+            "stamp": self._timestamp(),
+        }
+
+        # Add additional fields to the body
+        body.update(data)
+
+        return body
+
+    async def _get_login_id(self) -> str:
+        """Get a login ID for the cloud account."""
+
+        response = await self._api_request(
+            "/v1/user/login/id/get",
+            self._build_request_body(
+                {
+                    "loginAccount": self._account
+                }
+            )
+        )
+
+        # Assert response is not None since we should throw on errors
+        assert response is not None
+
+        login_id = response["loginId"]
+        _LOGGER.debug("Received loginId: %s", login_id)
+
+        return login_id
+
+    async def get_token(self, udpid: str) -> tuple[str, str]:
+        """Get token and key for the provided udpid."""
+
+        response = await self._api_request(
+            "/v1/iot/secure/getToken",
+            self._build_request_body(
+                {
+                    "udpid": udpid
+                }
+            )
+        )
+
+        # Assert response is not None since we should throw on errors
+        assert response is not None
+
+        for token in response["tokenlist"]:
+            if token["udpId"] == udpid:
+                return token["token"], token["key"]
+
+        # No matching udpId in the tokenlist
+        raise CloudError(f"No token/key found for udpid {udpid}.")
+
+
+class SmartHomeCloud(BaseCloud):
+    """Class for minimal Midea SmartHome cloud access."""
+
+    # Misc constants for the SmartHome cloud
+    APP_ID = "1010"
+
+    # Base URLs
+    BASE_URL = "https://mp-prod.appsmb.com"
+    BASE_URL_CHINA = "https://mp-prod.smartmidea.net"
+
+    CLOUD_CREDENTIALS = {
+        "DE": ("midea_eu@mailinator.com", "das_ist_passwort1"),
+        "KR": ("midea_sea@mailinator.com", "password_for_sea1"),
+        "US": ("midea@mailinator.com", "this_is_a_password1")
+    }
+
+    def __init__(self,
+                 region: str = DEFAULT_CLOUD_REGION,
+                 *,
+                 account: Optional[str] = None,
+                 password: Optional[str] = None,
+                 use_china_server: bool = False,
+                 **kwargs
+                 ) -> None:
+        # Allow override Chia server from environment
+        if os.getenv("MIDEA_CHINA_SERVER", "0") == "1":
+            use_china_server = True
+
+        base_url = SmartHomeCloud.BASE_URL_CHINA if use_china_server else SmartHomeCloud.BASE_URL
+        super().__init__(base_url, region, account, password, **kwargs)
+
+        self._access_token = ""
+        self._security = SmartHomeCloud._Security(use_china_server)
+
+    def _parse_response(self, response) -> Any:
+        """Parse a response from the cloud."""
+
+        _LOGGER.debug("Cloud response: %s", response.text)
+        body = json.loads(response.text)
+
+        response_code = int(body["code"])
+        if response_code == 0:
+            return body["data"]
+
+        raise ApiError(body["msg"], code=response_code)
+
+    async def _api_request(self, endpoint: str, body: dict[str, Any]) -> Optional[dict]:
+        """Make a request to the cloud and return the results."""
 
         # Encode body as JSON
         contents = json.dumps(body)
@@ -155,45 +251,23 @@ class Cloud:
 
         # Lock the API and post the request
         async with self._api_lock:
-            return await self._post_request(url, headers, contents)
+            return await self._post_request(url, headers=headers, raw_data=contents)
 
     def _build_request_body(self, data: dict[str, Any]) -> dict[str, Any]:
         """Build a request body."""
 
         # Set up the initial body
-        body = {
-            "appId": Cloud.APP_ID,
-            "format": Cloud.FORMAT,
-            "clientType": Cloud.CLIENT_TYPE,
-            "language": Cloud.LANGUAGE,
-            "src": Cloud.SRC,
-            "stamp": self._timestamp(),
-            "deviceId": Cloud.DEVICE_ID,
+        body = super()._build_request_body({
             "reqId": token_hex(16),
-        }
+        })
 
         # Add additional fields to the body
         body.update(data)
 
         return body
 
-    async def _get_login_id(self) -> str:
-        """Get a login ID for the cloud account."""
-
-        response = await self._api_request(
-            "/v1/user/login/id/get",
-            self._build_request_body(
-                {"loginAccount": self._account}
-            )
-        )
-
-        # Assert response is not None since we should throw on errors
-        assert response is not None
-
-        return response["loginId"]
-
     async def login(self, force: bool = False) -> None:
-        """Login to the cloud API."""
+        """Login to the cloud."""
 
         # Don't login if session already exists
         if self._session and not force:
@@ -202,24 +276,23 @@ class Cloud:
         # Get a login ID if we don't have one
         if self._login_id is None:
             self._login_id = await self._get_login_id()
-            _LOGGER.debug("Received loginId: %s", self._login_id)
 
         # Build the login data
         body = {
             "data": {
-                "platform": Cloud.FORMAT,
-                "deviceId": Cloud.DEVICE_ID,
+                "platform": BaseCloud.FORMAT,
+                "deviceId": BaseCloud.DEVICE_ID,
             },
             "iotData": {
-                "appId": Cloud.APP_ID,
-                "clientType": Cloud.CLIENT_TYPE,
-                "iampwd": self._security.encrypt_iam_password(self._login_id, self._password),
+                "appId": SmartHomeCloud.APP_ID,
+                "src": SmartHomeCloud.APP_ID,
+                "clientType": BaseCloud.CLIENT_TYPE,
                 "loginAccount": self._account,
+                "iampwd": self._security.encrypt_iam_password(self._login_id, self._password),
                 "password": self._security.encrypt_password(self._login_id, self._password),
                 "pushToken": token_urlsafe(120),
-                "reqId": token_hex(16),
-                "src": Cloud.SRC,
                 "stamp": self._timestamp(),
+                "reqId": token_hex(16),
             },
         }
 
@@ -233,24 +306,6 @@ class Cloud:
         self._access_token = response["mdata"]["accessToken"]
         _LOGGER.debug("Received accessToken: %s", self._access_token)
 
-    async def get_token(self, udpid: str) -> tuple[str, str]:
-        """Get token and key for the provided udpid."""
-
-        response = await self._api_request(
-            "/v1/iot/secure/getToken",
-            self._build_request_body({"udpid": udpid})
-        )
-
-        # Assert response is not None since we should throw on errors
-        assert response is not None
-
-        for token in response["tokenlist"]:
-            if token["udpId"] == udpid:
-                return token["token"], token["key"]
-
-        # No matching udpId in the tokenlist
-        raise CloudError(f"No token/key found for udpid {udpid}.")
-
     async def get_protocol_lua(self, device_type: DeviceType, sn: str) -> tuple[str, str]:
         """Fetch and decode the protocol Lua file."""
 
@@ -308,7 +363,7 @@ class Cloud:
 
         file_name = result["title"]
         url = result["url"]
-        async with self._get_async_client(verify=False) as client:
+        async with self._get_async_client() as client:
             try:
                 # Get file from server
                 r = await client.get(url, timeout=10.0)
@@ -319,80 +374,211 @@ class Cloud:
         file_data = r.content
         return (file_name, file_data)
 
+    class _Security:
+        """"Class for SmartHome cloud specific security."""
 
-class _Security:
-    """"Class for Midea cloud specific security."""
+        HMAC_KEY = "PROD_VnoClJI9aikS8dyy"
 
-    HMAC_KEY = "PROD_VnoClJI9aikS8dyy"
+        IOT_KEY = "meicloud"
+        LOGIN_KEY = "ac21b9f9cbfe4ca5a88562ef25e2b768"
 
-    IOT_KEY = "meicloud"
-    LOGIN_KEY = "ac21b9f9cbfe4ca5a88562ef25e2b768"
+        IOT_KEY_CHINA = "prod_secret123@muc"
+        LOGIN_KEY_CHINA = "ad0ee21d48a64bf49f4fb583ab76e799"
 
-    IOT_KEY_CHINA = "prod_secret123@muc"
-    LOGIN_KEY_CHINA = "ad0ee21d48a64bf49f4fb583ab76e799"
+        # MSmartHome
+        APP_KEY = "ac21b9f9cbfe4ca5a88562ef25e2b768"
 
-    # MSmartHome
-    APP_KEY = "ac21b9f9cbfe4ca5a88562ef25e2b768"
+        def __init__(self, use_china_server=False):
+            self._use_china_server = use_china_server
 
-    def __init__(self, use_china_server=False):
-        self._use_china_server = use_china_server
+        @property
+        def _iot_key(self) -> str:
+            """Get the IOT key for the appropriate server."""
+            return self.IOT_KEY_CHINA if self._use_china_server else self.IOT_KEY
 
-    @property
-    def _iot_key(self) -> str:
-        """Get the IOT key for the appropriate server."""
-        return _Security.IOT_KEY_CHINA if self._use_china_server else _Security.IOT_KEY
+        @property
+        def _login_key(self) -> str:
+            """Get the login key for the appropriate server."""
+            return self.LOGIN_KEY_CHINA if self._use_china_server else self.LOGIN_KEY
 
-    @property
-    def _login_key(self) -> str:
-        """Get the login key for the appropriate server."""
-        return _Security.LOGIN_KEY_CHINA if self._use_china_server else _Security.LOGIN_KEY
+        def sign(self, data: str, random: str) -> str:
+            """Generate a HMAC signature for the provided data and random data."""
+            msg = self._iot_key + data + random
 
-    def sign(self, data: str, random: str) -> str:
-        """Generate a HMAC signature for the provided data and random data."""
-        msg = self._iot_key + data + random
+            sign = hmac.new(self.HMAC_KEY.encode("ASCII"),
+                            msg.encode("ASCII"), hashlib.sha256)
+            return sign.hexdigest()
 
-        sign = hmac.new(self.HMAC_KEY.encode("ASCII"),
-                        msg.encode("ASCII"), hashlib.sha256)
-        return sign.hexdigest()
+        def encrypt_password(self, login_id: str, password: str) -> str:
+            """Encrypt the password for cloud password."""
+            # Hash the password
+            m1 = hashlib.sha256(password.encode("ASCII"))
 
-    def encrypt_password(self, login_id: str, password: str) -> str:
-        """Encrypt the password for cloud API password."""
-        # Hash the password
-        m1 = hashlib.sha256(password.encode("ASCII"))
+            # Create the login hash with the login ID + password hash + login key, then hash it all AGAIN
+            login_hash = login_id + m1.hexdigest() + self._login_key
+            m2 = hashlib.sha256(login_hash.encode("ASCII"))
 
-        # Create the login hash with the loginID + password hash + loginKey, then hash it all AGAIN
-        login_hash = login_id + m1.hexdigest() + self._login_key
-        m2 = hashlib.sha256(login_hash.encode("ASCII"))
+            return m2.hexdigest()
 
-        return m2.hexdigest()
+        def encrypt_iam_password(self, login_id: str, password: str) -> str:
+            """Encrypts password for cloud iampwd field."""
 
-    def encrypt_iam_password(self, login_id: str, password: str) -> str:
-        """Encrypts password for cloud API iampwd field."""
+            # Hash the password
+            m1 = hashlib.md5(password.encode("ASCII"))
 
-        # Hash the password
-        m1 = hashlib.md5(password.encode("ASCII"))
+            # Hash the password hash
+            m2 = hashlib.md5(m1.hexdigest().encode("ASCII"))
 
-        # Hash the password hash
-        m2 = hashlib.md5(m1.hexdigest().encode("ASCII"))
+            if self._use_china_server:
+                return m2.hexdigest()
+
+            login_hash = login_id + m2.hexdigest() + self._login_key
+            sha = hashlib.sha256(login_hash.encode("ASCII"))
+
+            return sha.hexdigest()
+
+        def _get_app_key_and_iv(self) -> tuple[bytes, bytes]:
+            hash = hashlib.sha256(self.APP_KEY.encode()).hexdigest()
+            return (hash[:16].encode(), hash[16:32].encode())
+
+        def encrypt_aes_app_key(self, data: bytes) -> bytes:
+            key, iv = self._get_app_key_and_iv()
+            cipher = AES.new(key, AES.MODE_CBC, iv=iv)
+            return cipher.encrypt(Padding.pad(data, 16))
+
+        def decrypt_aes_app_key(self, data: bytes) -> bytes:
+            key, iv = self._get_app_key_and_iv()
+            cipher = AES.new(key, AES.MODE_CBC, iv=iv)
+            return Padding.unpad(cipher.decrypt(data), 16)
 
-        if self._use_china_server:
-            return m2.hexdigest()
 
-        login_hash = login_id + m2.hexdigest() + self._login_key
-        sha = hashlib.sha256(login_hash.encode("ASCII"))
+class NetHomePlusCloud(BaseCloud):
+    """Class for minimal NetHome Plus cloud access."""
 
-        return sha.hexdigest()
+    # Misc constants for the NetHome Plus cloud
+    APP_ID = "1017"
 
-    def _get_app_key_and_iv(self) -> tuple[bytes, bytes]:
-        hash = hashlib.sha256(self.APP_KEY.encode()).hexdigest()
-        return (hash[:16].encode(), hash[16:32].encode())
+    BASE_URL = "https://mapp.appsmb.com"
 
-    def encrypt_aes_app_key(self, data: bytes) -> bytes:
-        key, iv = self._get_app_key_and_iv()
-        cipher = AES.new(key, AES.MODE_CBC, iv=iv)
-        return cipher.encrypt(Padding.pad(data, 16))
+    CLOUD_CREDENTIALS = {
+        "DE": ("nethome+de@mailinator.com", "password1"),
+        "KR": ("nethome+sea@mailinator.com", "password1"),
+        "US": ("nethome+us@mailinator.com", "password1")
+    }
 
-    def decrypt_aes_app_key(self, data: bytes) -> bytes:
-        key, iv = self._get_app_key_and_iv()
-        cipher = AES.new(key, AES.MODE_CBC, iv=iv)
-        return Padding.unpad(cipher.decrypt(data), 16)
+    def __init__(self,
+                 region: str = DEFAULT_CLOUD_REGION,
+                 *,
+                 account: Optional[str] = None,
+                 password: Optional[str] = None,
+                 **kwargs
+                 ) -> None:
+
+        super().__init__(NetHomePlusCloud.BASE_URL, region, account, password, **kwargs)
+
+        self._session_id = ""
+        self._security = NetHomePlusCloud._Security()
+
+    def _parse_response(self, response) -> Any:
+        """Parse a response from the cloud."""
+
+        _LOGGER.debug("Cloud response: %s", response.text)
+        body = json.loads(response.text)
+
+        response_code = int(body["errorCode"])
+        if response_code == 0:
+            return body["result"]
+
+        raise ApiError(body["msg"], code=response_code)
+
+    async def _api_request(self, endpoint: str, body: dict[str, Any]) -> Optional[dict]:
+        """Make a request to the cloud and return the results."""
+
+        # Sign the contents and add it to the body
+        body["sign"] = self._security.sign(endpoint, body)
+
+        # Build complete request URL
+        url = f"{self._base_url}{endpoint}"
+
+        # Lock the API and post the request
+        async with self._api_lock:
+            return await self._post_request(url, form_data=body)
+
+    def _build_request_body(self, data: dict[str, Any]) -> dict[str, Any]:
+        """Build a request body."""
+
+        # Set up the initial body
+        body = super()._build_request_body({
+            "sessionId": self._session_id
+        })
+
+        # Add additional fields to the body
+        body.update(data)
+
+        return body
+
+    async def login(self, force: bool = False) -> None:
+        """Login to the cloud."""
+
+        # Don't login if session already exists
+        if self._session and not force:
+            return
+
+        # Get a login ID if we don't have one
+        if self._login_id is None:
+            self._login_id = await self._get_login_id()
+
+        # Login and store the session
+        response = await self._api_request(
+            "/v1/user/login",
+            self._build_request_body(
+                {
+                    "loginAccount": self._account,
+                    "password": self._security.encrypt_password(self._login_id, self._password),
+                }
+            ))
+
+        # Assert response is not None since we should throw on errors
+        assert response is not None
+
+        self._session = response
+        self._session_id = response["sessionId"]
+        _LOGGER.debug("Received sessionId: %s", self._session_id)
+
+    async def get_protocol_lua(self, device_type: DeviceType, sn: str) -> tuple[str, str]:
+        """Fetch and decode the protocol Lua file."""
+        raise NotImplementedError
+
+    async def get_plugin(self, device_type: DeviceType, sn: str) -> tuple[str, bytes]:
+        """Request and download the device plugin."""
+        raise NotImplementedError
+
+    class _Security:
+        """"Class for NetHome Plus cloud specific security."""
+
+        # NetHome PLus
+        APP_KEY = "3742e9e5842d4ad59c2db887e12449f9"
+
+        def sign(self, url: str, data: dict[str, Any]) -> str:
+            """Generate a signature for the provided data and URL."""
+            # Get path portion of request
+            path = urlparse(url).path
+
+            # Sort request and create a query string
+            query = unquote_plus(urlencode(sorted(data.items())))
+
+            msg = path + query + self.APP_KEY
+
+            sign = hashlib.sha256(msg.encode("ASCII"))
+            return sign.hexdigest()
+
+        def encrypt_password(self, login_id: str, password: str) -> str:
+            """Encrypt the login password."""
+            # Hash the password
+            m1 = hashlib.sha256(password.encode("ASCII"))
+
+            # Create the login hash with the login ID + password hash + app key, then hash it all AGAIN
+            login_hash = login_id + m1.hexdigest() + self.APP_KEY
+            m2 = hashlib.sha256(login_hash.encode("ASCII"))
+
+            return m2.hexdigest()
diff --git a/msmart/const.py b/msmart/const.py
index b0b5df7..b2b093f 100644
--- a/msmart/const.py
+++ b/msmart/const.py
@@ -23,11 +23,6 @@ DEVICE_INFO_MSG = bytes([
 ])
 
 DEFAULT_CLOUD_REGION = "US"
-CLOUD_CREDENTIALS = {
-    "DE": ("midea_eu@mailinator.com", "das_ist_passwort1"),
-    "KR": ("midea_sea@mailinator.com", "password_for_sea1"),
-    "US": ("midea@mailinator.com", "this_is_a_password1")
-}
 
 
 class DeviceType(IntEnum):
diff --git a/msmart/discover.py b/msmart/discover.py
index 836f825..b0e6f39 100644
--- a/msmart/discover.py
+++ b/msmart/discover.py
@@ -8,7 +8,7 @@ from typing import Any, Callable, Optional, Type, cast
 
 import httpx
 
-from msmart.cloud import Cloud, CloudError
+from msmart.cloud import CloudError, NetHomePlusCloud
 from msmart.const import (DEFAULT_CLOUD_REGION, DEVICE_INFO_MSG, DISCOVERY_MSG,
                           DeviceType)
 from msmart.device import AirConditioner, Device
@@ -221,7 +221,7 @@ class Discover:
         return None
 
     @classmethod
-    async def _get_cloud(cls) -> Optional[Cloud]:
+    async def _get_cloud(cls) -> Optional[NetHomePlusCloud]:
         """Return a cloud connection, creating it if necessary."""
 
         # Lock should exist by now
@@ -230,11 +230,12 @@ class Discover:
         async with cls._lock:
             # Create cloud connection if nonexistent
             if cls._cloud is None:
-                cloud = Cloud(cls._region,
-                              account=cls._account,
-                              password=cls._password,
-                              get_async_client=cls._get_async_client
-                              )
+                cloud = NetHomePlusCloud(
+                    cls._region,
+                    account=cls._account,
+                    password=cls._password,
+                    get_async_client=cls._get_async_client
+                )
                 try:
                     await cloud.login()
                     cls._cloud = cloud
diff --git a/pyproject.toml b/pyproject.toml
index d271752..81915a7 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -33,7 +33,6 @@ Repository = "https://github.com/mill1000/midea-msmart"
 Issues = "https://github.com/mill1000/midea-msmart/issues"
 
 [project.scripts]
-midea-discover = "msmart.cli:_legacy_main"
 msmart-ng = "msmart.cli:main"
 
 [tool.setuptools]


Test patch:
diff --git a/msmart/tests/test_cloud.py b/msmart/tests/test_cloud.py
index a769f29..cfac1d7 100644
--- a/msmart/tests/test_cloud.py
+++ b/msmart/tests/test_cloud.py
@@ -1,7 +1,9 @@
+import logging
 import unittest
-from typing import Any, Optional
+from typing import Any, Optional, cast
 
-from msmart.cloud import ApiError, Cloud, CloudError
+from msmart.cloud import (ApiError, BaseCloud, CloudError, NetHomePlusCloud,
+                          SmartHomeCloud)
 from msmart.const import DEFAULT_CLOUD_REGION
 
 
@@ -9,23 +11,32 @@ class TestCloud(unittest.IsolatedAsyncioTestCase):
     # pylint: disable=protected-access
 
     async def _login(self,
-                     region: str = DEFAULT_CLOUD_REGION,
+                     class_name,
                      *,
+                     region: str = DEFAULT_CLOUD_REGION,
                      account: Optional[str] = None,
                      password: Optional[str] = None
-                     ) -> Cloud:
-        client = Cloud(region, account=account, password=password)
+                     ) -> BaseCloud:
+        client = class_name(region, account=account, password=password)
         await client.login()
 
         return client
 
+
+class TestNetHomePlusCloud(TestCloud):
+    # pylint: disable=protected-access
+
+    async def _login(self, *args, **kwargs) -> NetHomePlusCloud:
+        client = await super()._login(NetHomePlusCloud, *args, **kwargs)
+        return cast(NetHomePlusCloud, client)
+
     async def test_login(self) -> None:
         """Test that we can login to the cloud."""
 
         client = await self._login()
 
         self.assertIsNotNone(client._session)
-        self.assertIsNotNone(client._access_token)
+        self.assertIsNotNone(client._session_id)
 
     async def test_login_exception(self) -> None:
         """Test that bad credentials raise an exception."""
@@ -37,7 +48,7 @@ class TestCloud(unittest.IsolatedAsyncioTestCase):
         """Test that an invalid region raise an exception."""
 
         with self.assertRaises(ValueError):
-            await self._login("NOT_A_REGION")
+            await self._login(region="NOT_A_REGION")
 
     async def test_invalid_credentials(self) -> None:
         """Test that invalid credentials raise an exception."""
@@ -74,7 +85,78 @@ class TestCloud(unittest.IsolatedAsyncioTestCase):
     async def test_connect_exception(self) -> None:
         """Test that an exception is thrown when the cloud connection fails."""
 
-        client = Cloud(DEFAULT_CLOUD_REGION)
+        client = NetHomePlusCloud(DEFAULT_CLOUD_REGION)
+
+        # Override URL to an invalid domain
+        client._base_url = "https://fake_server.invalid."
+
+        with self.assertRaises(CloudError):
+            await client.login()
+
+
+class TestSmartHomeCloud(TestCloud):
+    # pylint: disable=protected-access
+
+    async def _login(self, *args, **kwargs) -> SmartHomeCloud:
+        client = await super()._login(SmartHomeCloud, *args, **kwargs)
+        return cast(SmartHomeCloud, client)
+
+    async def test_login(self) -> None:
+        """Test that we can login to the cloud."""
+
+        client = await self._login()
+
+        self.assertIsNotNone(client._session)
+        self.assertIsNotNone(client._access_token)
+
+    async def test_login_exception(self) -> None:
+        """Test that bad credentials raise an exception."""
+
+        with self.assertRaises(ApiError):
+            await self._login(account="bad@account.com", password="not_a_password")
+
+    async def test_invalid_region(self) -> None:
+        """Test that an invalid region raise an exception."""
+
+        with self.assertRaises(ValueError):
+            await self._login(region="NOT_A_REGION")
+
+    async def test_invalid_credentials(self) -> None:
+        """Test that invalid credentials raise an exception."""
+
+        # Check that specifying only an account or password raises an error
+        with self.assertRaises(ValueError):
+            await self._login(account=None, password="some_password")
+
+        with self.assertRaises(ValueError):
+            await self._login(account="some_account", password=None)
+
+    async def test_get_token(self) -> None:
+        """Test that a token and key can be obtained from the cloud."""
+        # Get token tests disabled until we can solve the broken API
+        # DUMMY_UDPID = "4fbe0d4139de99dd88a0285e14657045"
+
+        # client = await self._login()
+        # token, key = await client.get_token(DUMMY_UDPID)
+
+        # self.assertIsNotNone(token)
+        # self.assertIsNotNone(key)
+
+    async def test_get_token_exception(self) -> None:
+        """Test that an exception is thrown when a token and key
+        can't be obtained from the cloud."""
+        # Get token tests disabled until we can solve the broken API
+        # BAD_UDPID = "NOT_A_UDPID"
+
+        # client = await self._login()
+
+        # with self.assertRaises(CloudError):
+        #     await client.get_token(BAD_UDPID)
+
+    async def test_connect_exception(self) -> None:
+        """Test that an exception is thrown when the cloud connection fails."""
+
+        client = SmartHomeCloud(DEFAULT_CLOUD_REGION)
 
         # Override URL to an invalid domain
         client._base_url = "https://fake_server.invalid."