From 9b6ddc11136a35250442dd97be7dd23459be351f Mon Sep 17 00:00:00 2001 From: Alexis Santos Date: Sun, 26 Jul 2026 11:46:22 +0100 Subject: [PATCH 1/2] Add QR code login for the cloud interface Xiaomi serves a captcha on third-party password logins for many accounts, which micloud cannot answer, so `miiocli cloud` fails with MiCloudAccessDenied even when credentials are correct (#2038). micloud has had no release since 2022, so this cannot be fixed downstream. Xiaomi's QR code flow is unaffected by the captcha. This adds a small built-in backend implementing it: request a code, show it, long poll until the user approves it in the Xiaomi Home app, then call the signed device list endpoint. The backend deliberately exposes only login() and get_devices(country=...) and calls the same /home/device_list endpoint as micloud, so it returns identical records and CloudDeviceInfo parses them unchanged. Password login is untouched; QR is opt-in via `miiocli cloud --qr` or CloudInterface(use_qr=True), and password failures now point at it. It also never handles the account password at all. Notes on dependencies: - requests is now declared explicitly. It was already installed as a micloud dependency, so this adds nothing to an existing install, and declaring it means dropping micloud later does not break this path. - RC4 is implemented inline rather than imported. cryptography moved ARC4 to hazmat.decrepit in 43.0, so there is no import path valid across cryptography>=35, and pycryptodome is not a dependency. It is ~15 lines and is Xiaomi's transport obfuscation, not a security boundary. The request signing scheme follows Xiaomi-cloud-tokens-extractor (MIT). Tested with 18 new unit tests, and end to end against a real account and device: QR login, the signed device list call, and CloudDeviceInfo parsing of a live record. --- miio/cloud.py | 95 ++++++++--- miio/cloud_qr.py | 318 ++++++++++++++++++++++++++++++++++++ miio/tests/test_cloud_qr.py | 181 ++++++++++++++++++++ pyproject.toml | 1 + 4 files changed, 576 insertions(+), 19 deletions(-) create mode 100644 miio/cloud_qr.py create mode 100644 miio/tests/test_cloud_qr.py diff --git a/miio/cloud.py b/miio/cloud.py index d7e66d8ec..f6b30811f 100644 --- a/miio/cloud.py +++ b/miio/cloud.py @@ -18,6 +18,12 @@ if TYPE_CHECKING: from micloud import MiCloud # noqa: F401 +LOGIN_CAPTCHA_HINT = ( + "Login failed. Xiaomi serves a captcha on password logins for many " + "accounts, which micloud cannot answer. Try QR code login instead: " + "'miiocli cloud --qr list', or CloudInterface(use_qr=True)." +) + AVAILABLE_LOCALES = { "all": "All", "cn": "China", @@ -71,30 +77,63 @@ def raw_data(self): class CloudInterface: - """Cloud interface using micloud library. + """Cloud interface for obtaining a list of devices and their tokens. - You can use this to obtain a list of devices and their tokens. The :meth:`get_devices` takes the locale string (e.g., 'us') as an argument, defaulting to all known locales (accessible through :meth:`available_locales`). + Two login methods are available. Password login uses the ``micloud`` + library. QR code login is implemented locally and works when Xiaomi serves a + captcha on password login, which it currently does for many accounts. + Example:: ci = CloudInterface(username="foo", password=...) devs = ci.get_devices() for did, dev in devs.items(): print(dev) + + Or, without a password:: + + ci = CloudInterface(use_qr=True) + devs = ci.get_devices(locale="de") """ - def __init__(self, username, password): + def __init__(self, username=None, password=None, use_qr=False, on_qr=None): + """Initialize the interface. + + :param username: account name, for password login. + :param password: account password, for password login. + :param use_qr: authenticate by QR code instead, needing no password. + :param on_qr: called with ``(png_bytes, login_url)`` when using QR + login. Defaults to writing the image to a temporary file. + """ + if not use_qr and not (username and password): + raise CloudException( + "Either username and password, or use_qr=True, is required" + ) + self.username = username self.password = password - self._micloud = None + self.use_qr = use_qr + self._on_qr = on_qr + self._backend = None def _login(self): - if self._micloud is not None: + if self._backend is not None: _LOGGER.debug("Already logged in, skipping login") return + self._backend = self._qr_backend() if self.use_qr else self._micloud_backend() + + def _qr_backend(self): + from miio.cloud_qr import QrCodeLogin + + backend = QrCodeLogin(on_qr=self._on_qr) + backend.login() + return backend + + def _micloud_backend(self): try: from micloud import MiCloud # noqa: F811 from micloud.micloudexception import MiCloudAccessDenied @@ -103,12 +142,14 @@ def _login(self): "You need to install 'micloud' package to use cloud interface" ) from ex - self._micloud: MiCloud = MiCloud(username=self.username, password=self.password) + backend: MiCloud = MiCloud(username=self.username, password=self.password) try: # login() can either return False or raise an exception on failure - if not self._micloud.login(): - raise CloudException("Login failed") + if not backend.login(): + raise CloudException(LOGIN_CAPTCHA_HINT) except MiCloudAccessDenied as ex: - raise CloudException("Login failed") from ex + raise CloudException(LOGIN_CAPTCHA_HINT) from ex + + return backend def _parse_device_list(self, data, locale): """Parse device list response from micloud.""" @@ -139,7 +180,7 @@ def get_devices(self, locale: str | None = None) -> dict[str, CloudDeviceInfo]: self._login() if locale is not None and locale != "all": return self._parse_device_list( - self._micloud.get_devices(country=locale), locale=locale + self._backend.get_devices(country=locale), locale=locale ) all_devices: dict[str, CloudDeviceInfo] = {} @@ -153,18 +194,34 @@ def get_devices(self, locale: str | None = None) -> dict[str, CloudDeviceInfo]: @click.group(invoke_without_command=True) -@click.option("--username", prompt=True) -@click.option("--password", prompt=True, hide_input=True) +@click.option("--username", default=None, help="Account name for password login.") +@click.option("--password", default=None, help="Account password.") +@click.option( + "--qr", + "use_qr", + is_flag=True, + default=False, + help="Log in by scanning a QR code in the Xiaomi Home app, with no password. " + "Use this if password login fails with a captcha.", +) @click.pass_context -def cloud(ctx: click.Context, username, password): +def cloud(ctx: click.Context, username, password, use_qr): """Cloud commands.""" - try: - import micloud # noqa: F401 - except ImportError as ex: - _LOGGER.error("micloud is not installed, no cloud access available") - raise CloudException("install micloud for cloud access") from ex + if use_qr: + ctx.obj = CloudInterface(use_qr=True) + else: + try: + import micloud # noqa: F401 + except ImportError as ex: + _LOGGER.error("micloud is not installed, no cloud access available") + raise CloudException("install micloud for cloud access") from ex + + if not username: + username = click.prompt("Username") + if not password: + password = click.prompt("Password", hide_input=True) + ctx.obj = CloudInterface(username=username, password=password) - ctx.obj = CloudInterface(username=username, password=password) if ctx.invoked_subcommand is None: ctx.invoke(cloud_list) diff --git a/miio/cloud_qr.py b/miio/cloud_qr.py new file mode 100644 index 000000000..0de4f73c5 --- /dev/null +++ b/miio/cloud_qr.py @@ -0,0 +1,318 @@ +"""QR code login for the Xiaomi cloud. + +Xiaomi began serving a captcha on third-party password logins, which the +``micloud`` password flow cannot answer, so :class:`~miio.cloud.CloudInterface` +fails with ``MiCloudAccessDenied`` even when the credentials are correct +(see issue #2038). + +Xiaomi's QR code flow is unaffected. The account page issues a QR code, the +user scans it in the Xiaomi Home app, and a long poll returns the session. This +module implements that flow and the signed device-list call, exposing the same +two methods :class:`~miio.cloud.CloudInterface` needs from ``micloud``: +:meth:`login` and :meth:`get_devices`. + +It also never handles the account password, which removes a class of problem +rather than working around one. + +The request signing follows the scheme documented by +`Xiaomi-cloud-tokens-extractor `_ +(MIT). +""" + +import base64 +import hashlib +import json +import logging +import os +import secrets +import time +from collections.abc import Callable +from typing import Any + +from miio.exceptions import CloudException + +_LOGGER = logging.getLogger(__name__) + +LOGIN_URL = "https://account.xiaomi.com/longPolling/loginUrl" + +#: Seconds to wait for the user to scan, if Xiaomi does not say otherwise. +DEFAULT_SCAN_TIMEOUT = 300 + + +def _rc4(key: bytes, data: bytes) -> bytes: + """RC4, with the first 1024 keystream bytes discarded as Xiaomi expects. + + Implemented here rather than pulled from a crypto library on purpose: + ``cryptography`` moved ARC4 into ``hazmat.decrepit`` in 43.0, so it is not + reachable at a stable import path across the versions this package + supports, and RC4 is a handful of lines. This is Xiaomi's transport + obfuscation, not a security boundary. + """ + s = list(range(256)) + j = 0 + for i in range(256): + j = (j + s[i] + key[i % len(key)]) & 0xFF + s[i], s[j] = s[j], s[i] + + out = bytearray() + i = j = 0 + for byte in b"\x00" * 1024 + data: + i = (i + 1) & 0xFF + j = (j + s[i]) & 0xFF + s[i], s[j] = s[j], s[i] + out.append(byte ^ s[(s[i] + s[j]) & 0xFF]) + return bytes(out[1024:]) + + +def _strip_prefix(text: str) -> dict: + """Xiaomi prefixes its JSON responses with a fixed guard string.""" + return json.loads(text.replace("&&&START&&&", "")) + + +def print_qr_to_terminal(png: bytes, login_url: str) -> None: + """Default QR presenter: write the PNG to a temp file and print the URL. + + Flushed explicitly, because the caller then blocks on a long poll for up to + five minutes. Without the flush, a redirected or piped stdout would hold + the instructions in its buffer and the user would sit looking at nothing. + """ + import tempfile + + path = os.path.join(tempfile.gettempdir(), "miio-cloud-login-qr.png") + with open(path, "wb") as handle: + handle.write(png) + + _LOGGER.info("Login QR code written to %s", path) + print( # noqa: T201 + f"Scan this QR code with the Xiaomi Home app: {path}\n" + f"Or open this URL and sign in: {login_url}", + flush=True, + ) + + +class QrCodeLogin: + """Xiaomi cloud session authenticated by scanning a QR code. + + Drop-in for the parts of ``micloud.MiCloud`` that + :class:`~miio.cloud.CloudInterface` uses. + + Example:: + + backend = QrCodeLogin() + backend.login() + devices = backend.get_devices(country="de") + + :param on_qr: called with ``(png_bytes, login_url)`` when the code is + ready. Defaults to writing the PNG to a temporary file and printing + both paths. + :param scan_timeout: override how long to wait for the scan. + """ + + def __init__( + self, + on_qr: Callable[[bytes, str], None] | None = None, + scan_timeout: int | None = None, + ) -> None: + try: + import requests + except ImportError as ex: # pragma: no cover - requests is a dependency + raise CloudException( + "You need to install 'requests' to use the cloud interface" + ) from ex + + self._session = requests.Session() + self._on_qr = on_qr or print_qr_to_terminal + self._scan_timeout = scan_timeout + + agent_id = "".join(secrets.choice("ABCDE") for _ in range(13)) + self._agent = ( + f"Android-7.1.1-1.0.0-ONEPLUS A3010-136-{agent_id} " + "APP/xiaomi.smarthome APPV/62830" + ) + + self.user_id: str | None = None + self._ssecurity: str | None = None + self._service_token: str | None = None + + @staticmethod + def api_url(country: str) -> str: + """Return the API base for a locale. Mainland China has no prefix.""" + prefix = "" if country == "cn" else f"{country}." + return f"https://{prefix}api.io.mi.com/app" + + # -- login ----------------------------------------------------------- + + def login(self) -> bool: + """Run the QR login. Blocks until the code is scanned or expires.""" + import requests + + try: + response = self._session.get( + LOGIN_URL, + params={ + "_qrsize": "480", + "qs": "%3Fsid%3Dxiaomiio%26_json%3Dtrue", + "callback": "https://sts.api.io.mi.com/sts", + "_hasLogo": "false", + "sid": "xiaomiio", + "serviceParam": "", + "_locale": "en_GB", + "_dc": str(int(time.time() * 1000)), + }, + timeout=20, + ) + except requests.RequestException as ex: + raise CloudException(f"Unable to request a login QR code: {ex}") from ex + + data = _strip_prefix(response.text) if response.status_code == 200 else {} + if "qr" not in data: + raise CloudException("Xiaomi did not return a login QR code") + + timeout = self._scan_timeout or int(data.get("timeout", DEFAULT_SCAN_TIMEOUT)) + png = self._session.get(data["qr"], timeout=20).content + self._on_qr(png, data["loginUrl"]) + + session = self._long_poll(data["lp"], timeout) + self.user_id = session["userId"] + self._ssecurity = session["ssecurity"] + self._service_token = self._service_token_for(session["location"]) + _LOGGER.debug("Logged in as %s", self.user_id) + return True + + def _long_poll(self, url: str, timeout: int) -> dict: + """Wait for the user to approve the code in the Xiaomi Home app.""" + import requests + + started = time.monotonic() + while time.monotonic() - started < timeout: + try: + response = self._session.get(url, timeout=15) + except requests.Timeout: + continue # the long poll cycling is normal, keep waiting + except requests.RequestException as ex: + raise CloudException(f"Login polling failed: {ex}") from ex + if response.status_code == 200: + return _strip_prefix(response.text) + + raise CloudException("The QR code expired before it was scanned") + + def _service_token_for(self, location: str) -> str: + import requests + + try: + response = self._session.get( + location, + headers={"content-type": "application/x-www-form-urlencoded"}, + timeout=20, + ) + except requests.RequestException as ex: + raise CloudException(f"Unable to fetch the service token: {ex}") from ex + + token = response.cookies.get("serviceToken") + if not token: + raise CloudException("No service token in Xiaomi's response") + return token + + # -- signing --------------------------------------------------------- + + @staticmethod + def _signed_nonce(ssecurity: str, nonce: str) -> str: + digest = hashlib.sha256( + base64.b64decode(ssecurity) + base64.b64decode(nonce) + ).digest() + return base64.b64encode(digest).decode() + + @staticmethod + def _nonce() -> str: + minutes = int(time.time() * 1000) // 60000 + return base64.b64encode(os.urandom(8) + minutes.to_bytes(4, "big")).decode() + + @classmethod + def _signature(cls, url: str, signed_nonce: str, params: dict) -> str: + parts = ["POST", url.split("com")[1].replace("/app/", "/")] + parts += [f"{key}={value}" for key, value in params.items()] + parts.append(signed_nonce) + return base64.b64encode( + hashlib.sha1("&".join(parts).encode()).digest() # noqa: S324 + ).decode() + + @classmethod + def _seal(cls, signed_nonce: str, text: str) -> str: + return base64.b64encode( + _rc4(base64.b64decode(signed_nonce), text.encode()) + ).decode() + + @classmethod + def _unseal(cls, signed_nonce: str, text: str) -> bytes: + """Decrypt a response. + + The base64 decode is required; feeding the raw response body to RC4 + yields bytes that are not valid UTF-8. + """ + return _rc4(base64.b64decode(signed_nonce), base64.b64decode(text)) + + def _post(self, country: str, path: str, data: str) -> dict | None: + import requests + + if not (self._service_token and self._ssecurity): + raise CloudException("Not logged in") + + url = self.api_url(country) + path + nonce = self._nonce() + signed = self._signed_nonce(self._ssecurity, nonce) + + fields: dict[str, Any] = {"data": data} + fields["rc4_hash__"] = self._signature(url, signed, fields) + for key, value in list(fields.items()): + fields[key] = self._seal(signed, value) + fields["signature"] = self._signature(url, signed, fields) + fields["ssecurity"] = self._ssecurity + fields["_nonce"] = nonce + + try: + response = self._session.post( + url, + params=fields, + headers={ + "Accept-Encoding": "identity", + "User-Agent": self._agent, + "Content-Type": "application/x-www-form-urlencoded", + "x-xiaomi-protocal-flag-cli": "PROTOCAL-HTTP2", + "MIOT-ENCRYPT-ALGORITHM": "ENCRYPT-RC4", + }, + cookies={ + "userId": str(self.user_id), + "yetAnotherServiceToken": self._service_token, + "serviceToken": self._service_token, + "locale": "en_GB", + "channel": "MI_APP_STORE", + }, + timeout=25, + ) + except requests.RequestException as ex: + raise CloudException(f"Cloud request failed: {ex}") from ex + + if response.status_code != 200: + _LOGGER.debug("%s returned HTTP %s", path, response.status_code) + return None + try: + return json.loads(self._unseal(signed, response.text)) + except (json.JSONDecodeError, UnicodeDecodeError, ValueError): + _LOGGER.debug("Unable to decode the response for %s", path) + return None + + # -- devices --------------------------------------------------------- + + def get_devices(self, country: str = "cn") -> list[dict]: + """Return the raw device list for a locale. + + Deliberately calls the same endpoint as ``micloud.MiCloud.get_devices`` + and returns the same records, so this is a drop-in replacement and + :class:`~miio.cloud.CloudDeviceInfo` parses the result unchanged. + """ + response = self._post( + country, + "/home/device_list", + '{"getVirtualModel":false,"getHuamiDevices":0}', + ) + return ((response or {}).get("result") or {}).get("list") or [] diff --git a/miio/tests/test_cloud_qr.py b/miio/tests/test_cloud_qr.py new file mode 100644 index 000000000..0c52f7afd --- /dev/null +++ b/miio/tests/test_cloud_qr.py @@ -0,0 +1,181 @@ +import base64 +import json + +import pytest + +from miio.cloud import CloudInterface +from miio.cloud_qr import QrCodeLogin, _rc4, _strip_prefix +from miio.exceptions import CloudException + +KEY = base64.b64encode(b"0123456789abcdef").decode() + + +def test_rc4_is_symmetric(): + data = b"the same call encrypts and decrypts" + assert _rc4(base64.b64decode(KEY), _rc4(base64.b64decode(KEY), data)) == data + + +def _textbook_rc4(key: bytes, data: bytes) -> bytes: + """Reference RC4 with no keystream discard, for comparison only.""" + s = list(range(256)) + j = 0 + for i in range(256): + j = (j + s[i] + key[i % len(key)]) & 0xFF + s[i], s[j] = s[j], s[i] + out = bytearray() + i = j = 0 + for byte in data: + i = (i + 1) & 0xFF + j = (j + s[i]) & 0xFF + s[i], s[j] = s[j], s[i] + out.append(byte ^ s[(s[i] + s[j]) & 0xFF]) + return bytes(out) + + +def test_rc4_discards_the_first_1024_keystream_bytes(): + """Xiaomi skips 1024 keystream bytes before the payload. Without that skip + the ciphertext differs and the server rejects the request.""" + key = base64.b64decode(KEY) + payload = b"abc" + + assert len(_rc4(key, payload)) == len(payload) + assert _rc4(key, payload) != _textbook_rc4(key, payload) + + +def test_seal_unseal_round_trips(): + payload = json.dumps({"getVirtualModel": False, "getHuamiDevices": 0}) + assert QrCodeLogin._unseal(KEY, QrCodeLogin._seal(KEY, payload)).decode() == payload + + +def test_unseal_requires_the_base64_decode(): + """Feeding the raw response body to RC4 without base64-decoding it first + yields bytes that are not valid UTF-8, so responses silently fail to + parse.""" + sealed = QrCodeLogin._seal(KEY, '{"code": 0}') + assert QrCodeLogin._unseal(KEY, sealed).decode() == '{"code": 0}' + + not_decoded = _rc4(base64.b64decode(KEY), sealed.encode()) + with pytest.raises((UnicodeDecodeError, ValueError)): + json.loads(not_decoded) + + +def test_signed_nonce_is_deterministic(): + ssecurity = base64.b64encode(b"secret-secret-16").decode() + nonce = base64.b64encode(b"nonce-12").decode() + first = QrCodeLogin._signed_nonce(ssecurity, nonce) + assert first == QrCodeLogin._signed_nonce(ssecurity, nonce) + assert base64.b64decode(first) + + +def test_nonce_is_valid_base64_of_the_expected_length(): + raw = base64.b64decode(QrCodeLogin._nonce()) + assert len(raw) == 12 # 8 random bytes plus a 4-byte minute counter + + +@pytest.mark.parametrize( + ("country", "expected"), + [ + ("cn", "https://api.io.mi.com/app"), + ("de", "https://de.api.io.mi.com/app"), + ("us", "https://us.api.io.mi.com/app"), + ], +) +def test_api_url_per_locale(country, expected): + """Mainland China has no locale prefix, every other region does.""" + assert QrCodeLogin.api_url(country) == expected + + +def test_strip_prefix_handles_xiaomis_json_guard(): + assert _strip_prefix('&&&START&&&{"a": 1}') == {"a": 1} + assert _strip_prefix('{"a": 1}') == {"a": 1} + + +def test_post_before_login_raises(): + with pytest.raises(CloudException): + QrCodeLogin()._post("de", "/home/device_list", "{}") + + +def test_get_devices_tolerates_an_empty_response(mocker): + backend = QrCodeLogin() + mocker.patch.object(backend, "_post", return_value=None) + assert backend.get_devices(country="de") == [] + + +def test_get_devices_returns_the_device_list(mocker): + backend = QrCodeLogin() + mocker.patch.object( + backend, "_post", return_value={"result": {"list": [{"did": "1", "name": "x"}]}} + ) + assert backend.get_devices(country="de") == [{"did": "1", "name": "x"}] + + +# -- CloudInterface wiring ---------------------------------------------- + + +def test_cloud_interface_requires_credentials_or_qr(): + with pytest.raises(CloudException): + CloudInterface() + with pytest.raises(CloudException): + CloudInterface(username="foo") + + +def test_cloud_interface_accepts_qr_without_a_password(): + ci = CloudInterface(use_qr=True) + assert ci.use_qr is True + assert ci.password is None + + +def test_qr_login_is_used_when_requested(mocker): + backend = mocker.Mock() + login = mocker.patch("miio.cloud_qr.QrCodeLogin", return_value=backend) + + ci = CloudInterface(use_qr=True) + ci._login() + + login.assert_called_once() + backend.login.assert_called_once() + assert ci._backend is backend + + +def test_login_happens_only_once(mocker): + backend = mocker.Mock() + mocker.patch("miio.cloud_qr.QrCodeLogin", return_value=backend) + + ci = CloudInterface(use_qr=True) + ci._login() + ci._login() + + backend.login.assert_called_once() + + +def test_devices_are_parsed_from_the_qr_backend(mocker): + """The QR backend returns the same records as micloud, so the existing + parsing must work unchanged.""" + record = { + "localip": "192.168.1.2", + "token": "0" * 32, + "did": "123", + "mac": "aa:bb:cc:dd:ee:ff", + "name": "Robot", + "model": "xiaomi.vacuum.b108gl", + "desc": "Idle", + "parent_id": "", + "parent_model": "", + "ssid": "wifi", + "bssid": "aa:bb:cc:00:11:22", + "isOnline": True, + "rssi": -55, + } + backend = mocker.Mock() + backend.get_devices.return_value = [record] + mocker.patch("miio.cloud_qr.QrCodeLogin", return_value=backend) + + devices = CloudInterface(use_qr=True).get_devices(locale="de") + + assert len(devices) == 1 + info = devices["123_de"] + assert info.did == "123" + assert info.model == "xiaomi.vacuum.b108gl" + assert info.ip == "192.168.1.2" + assert info.is_online is True + assert info.is_child is False diff --git a/pyproject.toml b/pyproject.toml index 48482f1ea..2abc5dd9c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,6 +29,7 @@ dependencies = [ "platformdirs", "tqdm>=4,<5", "micloud>=0.6", + "requests>=2.28", "croniter>=1", "defusedxml>=0,<1", "pydantic>=2", From e32a5400e6a47b6d3c8a1e8c33fa0b69b4a5c167 Mon Sep 17 00:00:00 2001 From: Alexis Santos Date: Sun, 26 Jul 2026 11:53:21 +0100 Subject: [PATCH 2/2] Cover the QR login network paths codecov flagged the patch at 52%. The pure functions were tested but the network paths were not, which is where the real risk is. Adds 23 tests with a mocked session covering the full login walk, the long poll (including that a requests.Timeout is normal and must not abort it), service token retrieval, and the signed POST -- asserting the request is actually signed and encrypted rather than only that it returns. Also covers the untouched-but-now-reachable password path, the new CLI flag, and that --qr never prompts for credentials it will not use. cloud_qr.py goes from 48% to 100%. Drops the try/except around importing requests. It became dead code once requests was declared a dependency, so it is now a plain module-level import and the per-function imports are gone with it. --- miio/cloud_qr.py | 17 +- miio/tests/test_cloud_qr.py | 356 +++++++++++++++++++++++++++++++++++- 2 files changed, 357 insertions(+), 16 deletions(-) diff --git a/miio/cloud_qr.py b/miio/cloud_qr.py index 0de4f73c5..c699645d6 100644 --- a/miio/cloud_qr.py +++ b/miio/cloud_qr.py @@ -29,6 +29,8 @@ from collections.abc import Callable from typing import Any +import requests + from miio.exceptions import CloudException _LOGGER = logging.getLogger(__name__) @@ -113,13 +115,6 @@ def __init__( on_qr: Callable[[bytes, str], None] | None = None, scan_timeout: int | None = None, ) -> None: - try: - import requests - except ImportError as ex: # pragma: no cover - requests is a dependency - raise CloudException( - "You need to install 'requests' to use the cloud interface" - ) from ex - self._session = requests.Session() self._on_qr = on_qr or print_qr_to_terminal self._scan_timeout = scan_timeout @@ -144,8 +139,6 @@ def api_url(country: str) -> str: def login(self) -> bool: """Run the QR login. Blocks until the code is scanned or expires.""" - import requests - try: response = self._session.get( LOGIN_URL, @@ -181,8 +174,6 @@ def login(self) -> bool: def _long_poll(self, url: str, timeout: int) -> dict: """Wait for the user to approve the code in the Xiaomi Home app.""" - import requests - started = time.monotonic() while time.monotonic() - started < timeout: try: @@ -197,8 +188,6 @@ def _long_poll(self, url: str, timeout: int) -> dict: raise CloudException("The QR code expired before it was scanned") def _service_token_for(self, location: str) -> str: - import requests - try: response = self._session.get( location, @@ -252,8 +241,6 @@ def _unseal(cls, signed_nonce: str, text: str) -> bytes: return _rc4(base64.b64decode(signed_nonce), base64.b64decode(text)) def _post(self, country: str, path: str, data: str) -> dict | None: - import requests - if not (self._service_token and self._ssecurity): raise CloudException("Not logged in") diff --git a/miio/tests/test_cloud_qr.py b/miio/tests/test_cloud_qr.py index 0c52f7afd..a4f0e4b61 100644 --- a/miio/tests/test_cloud_qr.py +++ b/miio/tests/test_cloud_qr.py @@ -2,12 +2,49 @@ import json import pytest +import requests from miio.cloud import CloudInterface -from miio.cloud_qr import QrCodeLogin, _rc4, _strip_prefix +from miio.cloud_qr import ( + QrCodeLogin, + _rc4, + _strip_prefix, + print_qr_to_terminal, +) from miio.exceptions import CloudException KEY = base64.b64encode(b"0123456789abcdef").decode() +SSECURITY = base64.b64encode(b"secret-secret-16").decode() +NONCE = base64.b64encode(b"nonce-1234!!").decode() + +QR_PAYLOAD = { + "qr": "https://example.invalid/qr.png", + "loginUrl": "https://example.invalid/login", + "lp": "https://example.invalid/poll", + "timeout": 5, +} +SESSION_PAYLOAD = { + "userId": "42", + "ssecurity": SSECURITY, + "location": "https://example.invalid/sts", +} + + +class FakeResponse: + """Minimal stand-in for a requests response.""" + + def __init__(self, status_code=200, text="", content=b"", cookies=None): + self.status_code = status_code + self.text = text + self.content = content + self.cookies = cookies or {} + + +def logged_in(backend: QrCodeLogin) -> QrCodeLogin: + backend.user_id = "42" + backend._ssecurity = SSECURITY + backend._service_token = "service-token" + return backend def test_rc4_is_symmetric(): @@ -179,3 +216,320 @@ def test_devices_are_parsed_from_the_qr_backend(mocker): assert info.ip == "192.168.1.2" assert info.is_online is True assert info.is_child is False + + +# -- QR presentation ----------------------------------------------------- + + +def test_print_qr_to_terminal_writes_the_image_and_prints_both_routes( + tmp_path, monkeypatch, capsys +): + monkeypatch.setattr("tempfile.gettempdir", lambda: str(tmp_path)) + + print_qr_to_terminal(b"\x89PNG-ish", "https://example.invalid/login") + + written = tmp_path / "miio-cloud-login-qr.png" + assert written.read_bytes() == b"\x89PNG-ish" + + out = capsys.readouterr().out + assert str(written) in out + assert "https://example.invalid/login" in out + + +# -- login --------------------------------------------------------------- + + +def _login_session(mocker, backend, poll=None, sts_cookies=None): + """Route the four GETs login() makes to canned responses.""" + poll = poll if poll is not None else FakeResponse(200, json.dumps(SESSION_PAYLOAD)) + # `or` would be wrong here: an empty dict is the "no token" case under test. + if sts_cookies is None: + sts_cookies = {"serviceToken": "service-token"} + sts = FakeResponse(200, cookies=sts_cookies) + + def dispatch(url, **kwargs): + if "longPolling/loginUrl" in url: + return FakeResponse(200, "&&&START&&&" + json.dumps(QR_PAYLOAD)) + if url == QR_PAYLOAD["qr"]: + return FakeResponse(200, content=b"png-bytes") + if url == QR_PAYLOAD["lp"]: + return poll + return sts + + return mocker.patch.object(backend._session, "get", side_effect=dispatch) + + +def test_login_walks_the_whole_flow(mocker): + seen = {} + backend = QrCodeLogin(on_qr=lambda png, url: seen.update(png=png, url=url)) + _login_session(mocker, backend) + + assert backend.login() is True + + assert seen["png"] == b"png-bytes" + assert seen["url"] == QR_PAYLOAD["loginUrl"] + assert backend.user_id == "42" + assert backend._service_token == "service-token" + + +def test_login_raises_when_xiaomi_returns_no_qr(mocker): + backend = QrCodeLogin() + mocker.patch.object( + backend._session, "get", return_value=FakeResponse(200, json.dumps({})) + ) + with pytest.raises(CloudException, match="did not return a login QR"): + backend.login() + + +def test_login_raises_when_the_request_fails(mocker): + backend = QrCodeLogin() + mocker.patch.object( + backend._session, "get", side_effect=requests.ConnectionError("no route") + ) + with pytest.raises(CloudException, match="Unable to request a login QR"): + backend.login() + + +def test_login_raises_when_the_code_is_never_scanned(mocker): + """The long poll returns non-200 until the code expires.""" + backend = QrCodeLogin(on_qr=lambda *_: None, scan_timeout=0) + _login_session(mocker, backend, poll=FakeResponse(401)) + with pytest.raises(CloudException, match="expired"): + backend.login() + + +def test_login_raises_when_no_service_token_comes_back(mocker): + backend = QrCodeLogin(on_qr=lambda *_: None) + _login_session(mocker, backend, sts_cookies={}) + with pytest.raises(CloudException, match="No service token"): + backend.login() + + +def test_long_poll_keeps_waiting_through_timeouts(mocker): + """Xiaomi's long poll cycles; a timeout is normal and must not abort.""" + backend = QrCodeLogin() + responses = [ + requests.Timeout("cycle"), + FakeResponse(200, json.dumps(SESSION_PAYLOAD)), + ] + mocker.patch.object( + backend._session, + "get", + side_effect=lambda *a, **k: ( + (_ for _ in ()).throw(r) + if isinstance(r := responses.pop(0), Exception) + else r + ), + ) + assert backend._long_poll("https://example.invalid/poll", timeout=5) == ( + SESSION_PAYLOAD + ) + + +def test_long_poll_surfaces_a_connection_error(mocker): + backend = QrCodeLogin() + mocker.patch.object( + backend._session, "get", side_effect=requests.ConnectionError("down") + ) + with pytest.raises(CloudException, match="Login polling failed"): + backend._long_poll("https://example.invalid/poll", timeout=5) + + +# -- signed requests ----------------------------------------------------- + + +def test_post_signs_the_request_and_decrypts_the_reply(mocker): + backend = logged_in(QrCodeLogin()) + mocker.patch.object(QrCodeLogin, "_nonce", staticmethod(lambda: NONCE)) + signed = QrCodeLogin._signed_nonce(SSECURITY, NONCE) + + body = json.dumps({"code": 0, "result": {"list": []}}) + post = mocker.patch.object( + backend._session, + "post", + return_value=FakeResponse(200, QrCodeLogin._seal(signed, body)), + ) + + assert backend._post("de", "/home/device_list", "{}") == json.loads(body) + + kwargs = post.call_args.kwargs + assert kwargs["params"]["_nonce"] == NONCE + assert kwargs["params"]["ssecurity"] == SSECURITY + assert "signature" in kwargs["params"] + assert "rc4_hash__" in kwargs["params"] + # The payload must be encrypted, not sent in the clear. + assert kwargs["params"]["data"] != "{}" + assert kwargs["cookies"]["serviceToken"] == "service-token" + assert kwargs["headers"]["MIOT-ENCRYPT-ALGORITHM"] == "ENCRYPT-RC4" + assert post.call_args.args[0] == "https://de.api.io.mi.com/app/home/device_list" + + +def test_post_returns_none_on_a_non_200(mocker): + backend = logged_in(QrCodeLogin()) + mocker.patch.object(backend._session, "post", return_value=FakeResponse(500)) + assert backend._post("de", "/home/device_list", "{}") is None + + +def test_post_returns_none_when_the_reply_will_not_decode(mocker): + """A wrong key yields bytes that are not JSON; that must not raise.""" + backend = logged_in(QrCodeLogin()) + mocker.patch.object( + backend._session, "post", return_value=FakeResponse(200, "not-base64-!!") + ) + assert backend._post("de", "/home/device_list", "{}") is None + + +def test_post_surfaces_a_connection_error(mocker): + backend = logged_in(QrCodeLogin()) + mocker.patch.object( + backend._session, "post", side_effect=requests.ConnectionError("down") + ) + with pytest.raises(CloudException, match="Cloud request failed"): + backend._post("de", "/home/device_list", "{}") + + +def test_get_devices_reads_the_same_endpoint_as_micloud(mocker): + backend = logged_in(QrCodeLogin()) + post = mocker.patch.object(backend, "_post", return_value={"result": {"list": []}}) + + backend.get_devices(country="us") + + path = post.call_args.args[1] + assert path == "/home/device_list" + + +def test_service_token_fetch_surfaces_a_connection_error(mocker): + backend = QrCodeLogin() + mocker.patch.object( + backend._session, "get", side_effect=requests.ConnectionError("down") + ) + with pytest.raises(CloudException, match="Unable to fetch the service token"): + backend._service_token_for("https://example.invalid/sts") + + +# -- password login path ------------------------------------------------- + + +def test_password_login_failure_points_at_qr(mocker): + """The captcha is the usual cause now, so the error must say what to do.""" + from micloud.micloudexception import MiCloudAccessDenied + + micloud = mocker.Mock() + micloud.login.side_effect = MiCloudAccessDenied("Access denied") + mocker.patch("micloud.MiCloud", return_value=micloud) + + ci = CloudInterface(username="foo", password="bar") + with pytest.raises(CloudException, match="--qr"): + ci._login() + + +def test_password_login_failure_without_an_exception_also_points_at_qr(mocker): + """micloud.login() can return False instead of raising.""" + micloud = mocker.Mock() + micloud.login.return_value = False + mocker.patch("micloud.MiCloud", return_value=micloud) + + ci = CloudInterface(username="foo", password="bar") + with pytest.raises(CloudException, match="--qr"): + ci._login() + + +def test_password_login_still_works(mocker): + """QR is additive; the existing path must be untouched.""" + micloud = mocker.Mock() + micloud.login.return_value = True + factory = mocker.patch("micloud.MiCloud", return_value=micloud) + + ci = CloudInterface(username="foo", password="bar") + ci._login() + + factory.assert_called_once_with(username="foo", password="bar") + assert ci._backend is micloud + + +# -- CLI ----------------------------------------------------------------- + + +def test_cli_qr_flag_skips_the_credential_prompts(mocker): + """--qr must not ask for a username or password it will never use.""" + from click.testing import CliRunner + + from miio.cloud import cloud + + ci = mocker.Mock() + ci.get_devices.return_value = {} + factory = mocker.patch("miio.cloud.CloudInterface", return_value=ci) + + result = CliRunner().invoke(cloud, ["--qr", "list", "--locale", "de"], input="") + + assert result.exit_code == 0, result.output + factory.assert_called_once_with(use_qr=True) + assert "Username" not in result.output + assert "Password" not in result.output + + +def test_cli_password_path_prompts_when_not_given(mocker): + from click.testing import CliRunner + + from miio.cloud import cloud + + ci = mocker.Mock() + ci.get_devices.return_value = {} + factory = mocker.patch("miio.cloud.CloudInterface", return_value=ci) + + result = CliRunner().invoke(cloud, ["list", "--locale", "de"], input="user\npass\n") + + assert result.exit_code == 0, result.output + factory.assert_called_once_with(username="user", password="pass") + + +def test_cli_accepts_credentials_as_options(mocker): + from click.testing import CliRunner + + from miio.cloud import cloud + + ci = mocker.Mock() + ci.get_devices.return_value = {} + factory = mocker.patch("miio.cloud.CloudInterface", return_value=ci) + + result = CliRunner().invoke( + cloud, ["--username", "u", "--password", "p", "list", "--locale", "de"] + ) + + assert result.exit_code == 0, result.output + factory.assert_called_once_with(username="u", password="p") + + +def test_password_login_without_micloud_installed_explains_why(mocker): + mocker.patch.dict("sys.modules", {"micloud": None}) + ci = CloudInterface(username="foo", password="bar") + with pytest.raises(CloudException, match="micloud"): + ci._login() + + +def test_cli_without_micloud_installed_explains_why(mocker): + from click.testing import CliRunner + + from miio.cloud import cloud + + mocker.patch.dict("sys.modules", {"micloud": None}) + result = CliRunner().invoke(cloud, ["list", "--locale", "de"], input="u\np\n") + + assert result.exit_code != 0 + assert isinstance(result.exception, CloudException) + + +def test_cli_defaults_to_listing_when_no_subcommand_is_given(mocker): + """`miiocli cloud --qr` with no subcommand should still list.""" + from click.testing import CliRunner + + from miio.cloud import cloud + + ci = mocker.Mock() + ci.get_devices.return_value = {} + mocker.patch("miio.cloud.CloudInterface", return_value=ci) + + result = CliRunner().invoke(cloud, ["--qr"], input="de\n") + + assert result.exit_code == 0, result.output + ci.get_devices.assert_called_once()