From c6dd264e0a8874978d645259e12a09d5e17da717 Mon Sep 17 00:00:00 2001 From: Teemu Rytilahti Date: Wed, 16 Nov 2022 07:55:08 +0100 Subject: [PATCH 1/5] Add functions to obtain miotspec information This extends the public API by adding the following new methods to MiCloud: * miot_get_specs() to get list of all available specs, containing device urns and version information * miot_get_spec(device_urn) to get a spec for the given device urn * miot_get_standard_types(type) to get list of standardized elements for a given type (devices, properties, actions, ..) * miot_get_standard_type_spec(type_urn) to get the spec for a standardized element The cli tool is extended with a new group ("miot") that can be used to access this functionality. --- micloud/cli.py | 52 ++++++++++++++++++++++++- micloud/micloud.py | 95 ++++++++++++++++++++++++++++++++++++++++++++-- setup.py | 2 +- 3 files changed, 144 insertions(+), 5 deletions(-) diff --git a/micloud/cli.py b/micloud/cli.py index cb8acc7..5ff577b 100644 --- a/micloud/cli.py +++ b/micloud/cli.py @@ -1,13 +1,24 @@ import click import json + +import micloud.micloud from micloud.micloud import MiCloud -@click.command() + +pass_micloud = click.make_pass_decorator(MiCloud, ensure=True) + +@click.group() +def cli(): + """Tool for fetching xiaomi cloud information.""" + + +@cli.command() @click.option('--username', '-u', prompt=True, help='Your Xiaomi username.') @click.option('--password', '-p', prompt=True, hide_input=True, confirmation_prompt=False) @click.option('--country', '-c', default='de', help='Language code of the server to query. Default: "de"') @click.option('--pretty', is_flag=True, help='Pretty print json output.') def get_devices(username, password, country, pretty): + """Get device information, including tokens.""" mc = MiCloud(username, password) mc.login() devices = mc.get_devices(country=country) @@ -15,3 +26,42 @@ def get_devices(username, password, country, pretty): click.echo(json.dumps(devices, indent=2, sort_keys=True)) else: click.echo(json.dumps(devices)) + + +@cli.group() +def miot(): + """Commands for miotspec fetching.""" + +@miot.command(name="specs") +@click.option("--status", type=str, default="released") +@pass_micloud +def miot_specs(micloud: MiCloud, status): + """Return all specs filtered by the given status.""" + click.echo(json.dumps(micloud.miot_get_specs(status=status))) + + +@miot.command(name="get-spec") +@click.argument("urn") +@pass_micloud +def miot_get_spec(micloud: MiCloud, urn): + """Return a device spec for the given URN.""" + click.echo(json.dumps(micloud.miot_get_spec(urn))) + + +@miot.command("types") +@click.argument("type", type=click.Choice(micloud.micloud.MIOT_STANDARD_TYPES)) +@pass_micloud +def miot_available_standard_types(micloud: MiCloud, type: str): + """Return available standard URNs for type. """ + click.echo(json.dumps(micloud.miot_get_standard_types(type))) + + +@miot.command("get-type-spec") +@click.argument("urn") +@pass_micloud +def miot_get_standard_type_spec(micloud: MiCloud, urn: str): + """Return a type spec for given type URN.""" + click.echo(json.dumps(micloud.miot_get_standard_type_spec(urn))) + +if __name__ == "__main__": + cli() diff --git a/micloud/micloud.py b/micloud/micloud.py index bb950a7..5c94a5a 100644 --- a/micloud/micloud.py +++ b/micloud/micloud.py @@ -18,9 +18,18 @@ from micloud.micloudexception import MiCloudAccessDenied, MiCloudException +MIOTSPEC_BASE_URL = "https://miot-spec.org/miot-spec-v2" +# A dictionary consisting of miot standard types and their singular form +MIOT_STANDARD_TYPES = {"devices": "device", + "services": "service", + "properties": "property", + "actions": "action", + "events": "event" +} + class MiCloud(): - def __init__(self, username, password): + def __init__(self, username=None, password=None): super().__init__() self.user_id = None self.service_token = None @@ -42,12 +51,12 @@ def __init__(self, username, password): self.default_server = 'de' # Sets default server to Europe. self.username = username self.password = password - if not self._check_credentials(): - raise MiCloudException("username or password can't be empty") self.client_id = miutils.get_random_string(6) + + def get_token(self): """Return the servie token if you have successfully logged in.""" return self.service_token @@ -63,6 +72,7 @@ def login(self): :return: True if login successful, False otherwise. """ if not self._check_credentials(): + logging.error("You need to define username and password to log in") return False if self.user_id and self.service_token: @@ -326,3 +336,82 @@ def request(self, url, params): logging.exception("Error while decrypting response of request to %s :%s", url, str(e)) except Exception as e: logging.exception("Error while executing request to %s :%s", url, str(e)) + + + def miot_get_specs(self, status="released"): + """Return information about all available miotspec implementations. + + Note that every model may appear multiple times using different version in the response. + + Use :meth:`miot_get_schema_for_urn` to download the miotspec schema file based on the urn. + + :param status: filter by status, "released", "debug", "preview", "all", defaults to "released" + """ + self._init_session(reset=True) + url = f"{MIOTSPEC_BASE_URL}/instances?status={status}" + + AVAILABLE_STATUSES = ["released", "debug", "preview", "all"] + if status not in AVAILABLE_STATUSES: + raise MiCloudException("Unknown release status %s" % status) + + logging.debug("Going to download specs listing with status %s" % status) + + response = self.session.get(url) + return response.json() + + + def miot_get_spec(self, device_urn: str): + """Return miotspec device schema for the given device URN. + + The returned dict contains information about all services, properties, actions, events etc. + the device has been reported to support. + + :meth:`miot_get_available_schemas` can be used to return a list of all available URNs. + """ + self._init_session(reset=True) + + logging.debug("Going to download a spec for %s" % device_urn) + + url = f"{MIOTSPEC_BASE_URL}/instance?type={device_urn}" + + response = self.session.get(url) + response.raise_for_status() + return response.json() + + + def miot_get_standard_types(self, type_: str): + """Return standardized URNs for a given type. + + The type can be either devices, services, properties, actions, or events. + """ + if type_ not in MIOT_STANDARD_TYPES: + raise MiCloudException("Invalid schema type requested: %s" % type_) + + self._init_session(reset=True) + url = f"{MIOTSPEC_BASE_URL}/spec/{type_}" + + logging.debug("Going to download definition for type %s" % type_) + + response = self.session.get(url) + response.raise_for_status() + return response.json() + + + def miot_get_standard_type_spec(self, type_urn: str): + """Return a schema for a standard type URN. + + The response depends on the requested type and contains metadata about + the elements the given standard type must and can implement. + """ + splitted_urn = type_urn.split(":") + spec_type = splitted_urn[2] + namespace = splitted_urn[1] + if namespace != "miot-spec-v2": + raise MiCloudException("Tried to fetch spec for non-standard namespace %s" % namespace) + + self._init_session(reset=True) + url = f"{MIOTSPEC_BASE_URL}/spec/{spec_type}?type={type_urn}" + + response = self.session.get(url) + response.raise_for_status() + return response.json() diff --git a/setup.py b/setup.py index e5a12fd..2e2c437 100644 --- a/setup.py +++ b/setup.py @@ -22,6 +22,6 @@ ], entry_points=''' [console_scripts] - micloud=micloud.cli:get_devices + micloud=micloud.cli:cli ''', ) From 206b1ef2ba08e41f1cbacdb0c020ee5deec7f798 Mon Sep 17 00:00:00 2001 From: Teemu Rytilahti Date: Wed, 16 Nov 2022 07:55:17 +0100 Subject: [PATCH 2/5] Fix invalid self-reference --- micloud/micloud.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/micloud/micloud.py b/micloud/micloud.py index 5c94a5a..33afbb1 100644 --- a/micloud/micloud.py +++ b/micloud/micloud.py @@ -91,7 +91,7 @@ def login(self): self.service_token = None if self.failed_logins > 10: logging.info("Repeated errors logging on to Xiaomi cloud. Cleaning stored cookies") - self.self._init_session(reset=True) + self._init_session(reset=True) return False except MiCloudAccessDenied as e: logging.info("Access denied when logging on to Xiaomi cloud (%s): %s", self.failed_logins, str(e)) @@ -99,7 +99,7 @@ def login(self): self.service_token = None if self.failed_logins > 10: logging.info("Repeated errors logging on to Xiaomi cloud. Cleaning stored cookies") - self.self._init_session(reset=True) + self._init_session(reset=True) raise e except: logging.exception("Unknown exception occurred!") From c18b40656e20fa6f683404b5b5033b2928fed2d5 Mon Sep 17 00:00:00 2001 From: Teemu Rytilahti Date: Thu, 24 Nov 2022 10:28:48 +0100 Subject: [PATCH 3/5] Move session creation to utils helper --- micloud/micloud.py | 24 +++++++----------------- micloud/miutils.py | 19 ++++++++++++++++++- 2 files changed, 25 insertions(+), 18 deletions(-) diff --git a/micloud/micloud.py b/micloud/micloud.py index 33afbb1..f5eda5e 100644 --- a/micloud/micloud.py +++ b/micloud/micloud.py @@ -6,7 +6,6 @@ # email sammy@ssvensson.se # ----------------------------------------------------------- -import http.client, http.cookies import json import hashlib import logging @@ -15,7 +14,8 @@ import requests from micloud import miutils -from micloud.micloudexception import MiCloudAccessDenied, MiCloudException +from .miutils import get_session +from .micloudexception import MiCloudAccessDenied, MiCloudException MIOTSPEC_BASE_URL = "https://miot-spec.org/miot-spec-v2" @@ -27,7 +27,7 @@ "events": "event" } -class MiCloud(): +class MiCloud: def __init__(self, username=None, password=None): super().__init__() @@ -40,8 +40,6 @@ def __init__(self, username=None, password=None): self.failed_logins = 0 - self.agent_id = miutils.get_random_agent_id() - self.useragent = "Android-7.1.1-1.0.0-ONEPLUS A3010-136-" + self.agent_id + " APP/xiaomi.smarthome APPV/62830" self.locale = locale.getdefaultlocale()[0] timezone = datetime.datetime.now(tzlocal.get_localzone()).strftime('%z') @@ -52,10 +50,6 @@ def __init__(self, username=None, password=None): self.username = username self.password = password - self.client_id = miutils.get_random_string(6) - - - def get_token(self): """Return the servie token if you have successfully logged in.""" @@ -135,12 +129,9 @@ def _login_request(self): def _init_session(self, reset=False): if not self.session or reset: - self.session = requests.Session() - self.session.headers.update({'User-Agent': self.useragent}) - self.session.cookies.update({ - 'sdkVersion': '3.8.6', - 'deviceId': self.client_id - }) + if self.session is not None: + self.session.close() + self.session = get_session() def _login_step1(self): @@ -298,9 +289,8 @@ def request(self, url, params): logging.debug("Send request: %s to %s", params['data'], url) - self.session = requests.Session() + self._init_session(reset=True) self.session.headers.update({ - 'User-Agent': self.useragent, 'Accept-Encoding': 'identity', 'x-xiaomi-protocal-flag-cli': 'PROTOCAL-HTTP2', 'content-type': 'application/x-www-form-urlencoded', diff --git a/micloud/miutils.py b/micloud/miutils.py index 62bd991..65d4d4b 100644 --- a/micloud/miutils.py +++ b/micloud/miutils.py @@ -10,10 +10,11 @@ import random import hashlib, hmac, base64 import string +import requests from urllib.parse import urlparse from Crypto.Cipher import ARC4 -from micloud.micloudexception import MiCloudException +from .micloudexception import MiCloudException def get_random_agent_id(): @@ -113,3 +114,19 @@ def decrypt_rc4(password, payload): r = ARC4.new(base64.b64decode(password)) r.encrypt(bytes(1024)) return r.encrypt(base64.b64decode(payload)) + + +def get_session(): + """Create client session with pre-defined common headers.""" + agent_id = get_random_agent_id() + client_id = get_random_string(6) + useragent = "Android-7.1.1-1.0.0-ONEPLUS A3010-136-" + agent_id + " APP/xiaomi.smarthome APPV/62830" + + session = requests.Session() + session.headers.update({'User-Agent': useragent}) + session.cookies.update({ + 'sdkVersion': '3.8.6', + 'deviceId': client_id + }) + + return session From 4662fd9eeb214e1239a299d3ae9462253afe2626 Mon Sep 17 00:00:00 2001 From: Teemu Rytilahti Date: Thu, 24 Nov 2022 10:29:11 +0100 Subject: [PATCH 4/5] Move miotspec into its own module --- micloud/cli.py | 37 +++++++------- micloud/micloud.py | 87 --------------------------------- micloud/miotspec.py | 116 ++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 136 insertions(+), 104 deletions(-) create mode 100644 micloud/miotspec.py diff --git a/micloud/cli.py b/micloud/cli.py index 5ff577b..396fa26 100644 --- a/micloud/cli.py +++ b/micloud/cli.py @@ -1,16 +1,22 @@ import click import json +import logging -import micloud.micloud from micloud.micloud import MiCloud +from .miotspec import MiotSpec, MIOT_STANDARD_TYPES -pass_micloud = click.make_pass_decorator(MiCloud, ensure=True) +pass_miotspec = click.make_pass_decorator(MiotSpec, ensure=True) @click.group() -def cli(): +@click.option("-d", "--debug", is_flag=True) +def cli(debug): """Tool for fetching xiaomi cloud information.""" + level = logging.INFO + if debug: + level = logging.DEBUG + logging.basicConfig(level=level) @cli.command() @click.option('--username', '-u', prompt=True, help='Your Xiaomi username.') @@ -32,36 +38,33 @@ def get_devices(username, password, country, pretty): def miot(): """Commands for miotspec fetching.""" + @miot.command(name="specs") @click.option("--status", type=str, default="released") -@pass_micloud -def miot_specs(micloud: MiCloud, status): +def miot_specs(status): """Return all specs filtered by the given status.""" - click.echo(json.dumps(micloud.miot_get_specs(status=status))) + click.echo(json.dumps(MiotSpec.get_specs(status=status))) @miot.command(name="get-spec") @click.argument("urn") -@pass_micloud -def miot_get_spec(micloud: MiCloud, urn): +def miot_get_spec(urn): """Return a device spec for the given URN.""" - click.echo(json.dumps(micloud.miot_get_spec(urn))) + click.echo(json.dumps(MiotSpec.get_spec_for_urn(urn))) @miot.command("types") -@click.argument("type", type=click.Choice(micloud.micloud.MIOT_STANDARD_TYPES)) -@pass_micloud -def miot_available_standard_types(micloud: MiCloud, type: str): +@click.argument("type", type=click.Choice(MIOT_STANDARD_TYPES)) +def miot_available_standard_types(type: str): """Return available standard URNs for type. """ - click.echo(json.dumps(micloud.miot_get_standard_types(type))) + click.echo(json.dumps(MiotSpec.get_standard_types(type))) @miot.command("get-type-spec") -@click.argument("urn") -@pass_micloud -def miot_get_standard_type_spec(micloud: MiCloud, urn: str): +@click.argument("urn", required=False) +def miot_get_standard_type_spec(urn: str): """Return a type spec for given type URN.""" - click.echo(json.dumps(micloud.miot_get_standard_type_spec(urn))) + click.echo(json.dumps(MiotSpec.get_standard_type_spec(urn))) if __name__ == "__main__": cli() diff --git a/micloud/micloud.py b/micloud/micloud.py index f5eda5e..348ee92 100644 --- a/micloud/micloud.py +++ b/micloud/micloud.py @@ -18,14 +18,6 @@ from .micloudexception import MiCloudAccessDenied, MiCloudException -MIOTSPEC_BASE_URL = "https://miot-spec.org/miot-spec-v2" -# A dictionary consisting of miot standard types and their singular form -MIOT_STANDARD_TYPES = {"devices": "device", - "services": "service", - "properties": "property", - "actions": "action", - "events": "event" -} class MiCloud: @@ -326,82 +318,3 @@ def request(self, url, params): logging.exception("Error while decrypting response of request to %s :%s", url, str(e)) except Exception as e: logging.exception("Error while executing request to %s :%s", url, str(e)) - - - def miot_get_specs(self, status="released"): - """Return information about all available miotspec implementations. - - Note that every model may appear multiple times using different version in the response. - - Use :meth:`miot_get_schema_for_urn` to download the miotspec schema file based on the urn. - - :param status: filter by status, "released", "debug", "preview", "all", defaults to "released" - """ - self._init_session(reset=True) - url = f"{MIOTSPEC_BASE_URL}/instances?status={status}" - - AVAILABLE_STATUSES = ["released", "debug", "preview", "all"] - if status not in AVAILABLE_STATUSES: - raise MiCloudException("Unknown release status %s" % status) - - logging.debug("Going to download specs listing with status %s" % status) - - response = self.session.get(url) - return response.json() - - - def miot_get_spec(self, device_urn: str): - """Return miotspec device schema for the given device URN. - - The returned dict contains information about all services, properties, actions, events etc. - the device has been reported to support. - - :meth:`miot_get_available_schemas` can be used to return a list of all available URNs. - """ - self._init_session(reset=True) - - logging.debug("Going to download a spec for %s" % device_urn) - - url = f"{MIOTSPEC_BASE_URL}/instance?type={device_urn}" - - response = self.session.get(url) - response.raise_for_status() - return response.json() - - - def miot_get_standard_types(self, type_: str): - """Return standardized URNs for a given type. - - The type can be either devices, services, properties, actions, or events. - """ - if type_ not in MIOT_STANDARD_TYPES: - raise MiCloudException("Invalid schema type requested: %s" % type_) - - self._init_session(reset=True) - url = f"{MIOTSPEC_BASE_URL}/spec/{type_}" - - logging.debug("Going to download definition for type %s" % type_) - - response = self.session.get(url) - response.raise_for_status() - return response.json() - - - def miot_get_standard_type_spec(self, type_urn: str): - """Return a schema for a standard type URN. - - The response depends on the requested type and contains metadata about - the elements the given standard type must and can implement. - """ - splitted_urn = type_urn.split(":") - spec_type = splitted_urn[2] - namespace = splitted_urn[1] - if namespace != "miot-spec-v2": - raise MiCloudException("Tried to fetch spec for non-standard namespace %s" % namespace) - - self._init_session(reset=True) - url = f"{MIOTSPEC_BASE_URL}/spec/{spec_type}?type={type_urn}" - - response = self.session.get(url) - response.raise_for_status() - return response.json() diff --git a/micloud/miotspec.py b/micloud/miotspec.py new file mode 100644 index 0000000..12a3b1a --- /dev/null +++ b/micloud/miotspec.py @@ -0,0 +1,116 @@ +import logging + +from .micloudexception import MiCloudException +from .miutils import get_session + +# A dictionary consisting of miot standard types and their singular form +MIOT_STANDARD_TYPES = { + "devices": "device", + "services": "service", + "properties": "property", + "actions": "action", + "events": "event", +} + +_LOGGER = logging.getLogger(__name__) + + +class MiotSpec: + """Simple wrapper for accessing miot-spec.org web service. + + You can pass an existing requests.Session object using *session* to each class method, + otherwise a new session is created using :meth:`micloud.miutils.get_session`. + + :meth:`get_specs` returns a list of all available specs with their version and urn information. + The latter can be used with :meth:`get_spec_for_urn` to obtain a full miotspec schema. + """ + + BASE_URL = "https://miot-spec.org/miot-spec-v2" + + @classmethod + def get_specs(cls, status="released", session=None): + """Return information about all available miotspec implementations. + + Note that every model may appear multiple times using different version in the response. + + Use :meth:`miot_get_schema_for_urn` to download the miotspec schema file based on the urn. + + :param status: filter by status, "released", "debug", "preview", "all", defaults to "released" + """ + if session is None: + session = get_session() + + url = f"{cls.BASE_URL}/instances?status={status}" + + AVAILABLE_STATUSES = ["released", "debug", "preview", "all"] + if status not in AVAILABLE_STATUSES: + raise MiCloudException("Unknown release status %s" % status) + + _LOGGER.debug("Going to download specs listing with status %s" % status) + + response = session.get(url) + return response.json() + + @classmethod + def get_spec_for_urn(cls, device_urn: str, session=None): + """Return miotspec device schema for the given device URN. + + The returned dict contains information about all services, properties, actions, events etc. + the device has been reported to support. + + :meth:`miot_get_available_schemas` can be used to return a list of all available URNs. + """ + if session is None: + session = get_session() + + _LOGGER.debug("Going to download a spec for %s" % device_urn) + + url = f"{cls.BASE_URL}/instance?type={device_urn}" + + response = session.get(url) + response.raise_for_status() + return response.json() + + @classmethod + def get_standard_types(cls, type_: str, session=None): + """Return standardized URNs for a given type. + + The type can be either devices, services, properties, actions, or events. + """ + if type_ not in MIOT_STANDARD_TYPES: + raise MiCloudException("Invalid schema type requested: %s" % type_) + + if session is None: + session = get_session() + + url = f"{cls.BASE_URL}/spec/{type_}" + + _LOGGER.debug("Going to download definition for type %s" % type_) + + response = session.get(url) + response.raise_for_status() + return response.json() + + @classmethod + def get_standard_type_spec(cls, type_urn: str, session=None): + """Return a schema for a standard type URN. + + The response depends on the requested type and contains metadata about + the elements the given standard type must and can implement. + """ + splitted_urn = type_urn.split(":") + spec_type = splitted_urn[2] + namespace = splitted_urn[1] + if namespace != "miot-spec-v2": + raise MiCloudException( + "Tried to fetch spec for non-standard namespace %s" % namespace + ) + + if session is None: + session = get_session() + + url = f"{cls.BASE_URL}/spec/{spec_type}?type={type_urn}" + + response = session.get(url) + response.raise_for_status() + return response.json() From a08a6ec7ccb866177cdfed34ce94c3b49d93faa0 Mon Sep 17 00:00:00 2001 From: Teemu Rytilahti Date: Thu, 24 Nov 2022 10:31:48 +0100 Subject: [PATCH 5/5] Remove pass_miotspec decorator from cli --- micloud/cli.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/micloud/cli.py b/micloud/cli.py index 396fa26..fc8e88f 100644 --- a/micloud/cli.py +++ b/micloud/cli.py @@ -6,8 +6,6 @@ from .miotspec import MiotSpec, MIOT_STANDARD_TYPES -pass_miotspec = click.make_pass_decorator(MiotSpec, ensure=True) - @click.group() @click.option("-d", "--debug", is_flag=True) def cli(debug):