Init commit

.gitignore

@@ -0,0 +1,6 @@
+.idea/
+venv/
+__pycache__/
+dist/
+**/*.egg-info/
+test.py

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2023 Andre Basche
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1 @@
+# pyhOn

pyhon/__init__.py

@@ -0,0 +1 @@
+from .api import HonConnection

pyhon/api.py

@@ -0,0 +1,130 @@
+import json
+import logging
+import secrets
+from datetime import datetime
+from typing import List
+
+import aiohttp as aiohttp
+
+import const
+from auth import HonAuth
+from device import HonDevice
+
+_LOGGER = logging.getLogger()
+
+
+class HonConnection:
+    def __init__(self, email, password) -> None:
+        super().__init__()
+        self._email = email
+        self._password = password
+        self._request_headers = {"Content-Type": "application/json"}
+        self._session = None
+        self._devices = []
+        self._mobile_id = secrets.token_hex(8)
+
+    async def __aenter__(self):
+        self._session = aiohttp.ClientSession()
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        await self._session.close()
+
+    @property
+    def devices(self) -> List[HonDevice]:
+        return self._devices
+
+    @property
+    async def _headers(self):
+        if "cognito-token" not in self._request_headers or "id-token" not in self._request_headers:
+            auth = HonAuth()
+            if await auth.authorize(self._email, self._password, self._mobile_id):
+                self._request_headers["cognito-token"] = auth.cognito_token
+                self._request_headers["id-token"] = auth.id_token
+            else:
+                raise PermissionError("Can't Login")
+        return self._request_headers
+
+    async def setup(self):
+        async with aiohttp.ClientSession() as session:
+            async with session.get(f"{const.API_URL}/commands/v1/appliance",
+                                   headers=await self._headers) as resp:
+                try:
+                    appliances = (await resp.json())["payload"]["appliances"]
+                    self._devices = [HonDevice(self, appliance) for appliance in appliances]
+                except json.JSONDecodeError:
+                    _LOGGER.error("No JSON Data after GET: %s", await resp.text())
+                    return False
+        return True
+
+    async def load_commands(self, device: HonDevice):
+        params = {
+            "applianceType": device.appliance_type_name,
+            "code": device.code,
+            "applianceModelId": device.appliance_model_id,
+            "firmwareId": "41",
+            "macAddress": device.mac_address,
+            "fwVersion": device.fw_version,
+            "os": const.OS,
+            "appVersion": const.APP_VERSION,
+            "series": device.series,
+        }
+        url = f"{const.API_URL}/commands/v1/retrieve"
+        async with self._session.get(url, params=params, headers=await self._headers) as response:
+            result = (await response.json()).get("payload", {})
+            if not result or result.pop("resultCode") != "0":
+                return {}
+            return result
+
+    async def load_attributes(self, device: HonDevice):
+        params = {
+            "macAddress": device.mac_address,
+            "applianceType": device.appliance_type_name,
+            "category": "CYCLE"
+        }
+        url = f"{const.API_URL}/commands/v1/context"
+        async with self._session.get(url, params=params, headers=await self._headers) as response:
+            return (await response.json()).get("payload", {})
+
+    async def load_statistics(self, device: HonDevice):
+        params = {
+            "macAddress": device.mac_address,
+            "applianceType": device.appliance_type_name
+        }
+        url = f"{const.API_URL}/commands/v1/statistics"
+        async with self._session.get(url, params=params, headers=await self._headers) as response:
+            return (await response.json()).get("payload", {})
+
+    async def send_command(self, device, command, parameters, ancillary_parameters):
+        now = datetime.utcnow().isoformat()
+        data = {
+            "macAddress": device.mac_address,
+            "timestamp": f"{now[:-3]}Z",
+            "commandName": command,
+            "transactionId": f"{device.mac_address}_{now[:-3]}Z",
+            "applianceOptions": device.commands_options,
+            "device": {
+                "mobileId": self._mobile_id,
+                "mobileOs": const.OS,
+                "osVersion": const.OS_VERSION,
+                "appVersion": const.APP_VERSION,
+                "deviceModel": const.DEVICE_MODEL
+            },
+            "attributes": {
+                "channel": "mobileApp",
+                "origin": "standardProgram",
+                "energyLabel": "0"
+            },
+            "ancillaryParameters": ancillary_parameters,
+            "parameters": parameters,
+            "applianceType": device.appliance_type_name
+        }
+        url = f"{const.API_URL}/commands/v1/send"
+        async with self._session.post(url, headers=await self._headers, json=data) as resp:
+            try:
+                json_data = await resp.json()
+            except json.JSONDecodeError:
+                return False
+            if json_data["payload"]["resultCode"] == "0":
+                return True
+        return False

pyhon/auth.py

@@ -0,0 +1,128 @@
+import json
+import logging
+import re
+import secrets
+import urllib
+from urllib import parse
+
+import aiohttp as aiohttp
+
+import const
+
+_LOGGER = logging.getLogger()
+
+
+class HonAuth:
+    def __init__(self) -> None:
+        self._framework = ""
+        self._cognito_token = ""
+        self._id_token = ""
+
+    @property
+    def cognito_token(self):
+        return self._cognito_token
+
+    @property
+    def id_token(self):
+        return self._id_token
+
+    async def _get_frontdoor_url(self, session, email, password):
+        data = {
+            "message": {
+                "actions": [
+                    {
+                        "id": "79;a",
+                        "descriptor": "apex://LightningLoginCustomController/ACTION$login",
+                        "callingDescriptor": "markup://c:loginForm",
+                        "params": {
+                            "username": email,
+                            "password": password,
+                            "startUrl": ""
+                        }
+                    }
+                ]
+            },
+            "aura.context": {
+                "mode": "PROD",
+                "fwuid": self._framework,
+                "app": "siteforce:loginApp2",
+                "loaded": {"APPLICATION@markup://siteforce:loginApp2": "YtNc5oyHTOvavSB9Q4rtag"},
+                "dn": [],
+                "globals": {},
+                "uad": False},
+            "aura.pageURI": f"SmartHome/s/login/?language={const.LANGUAGE}",
+            "aura.token": None}
+
+        params = {"r": 3, "other.LightningLoginCustom.login": 1}
+        async with session.post(
+                const.AUTH_API + "/s/sfsites/aura",
+                headers={"Content-Type": "application/x-www-form-urlencoded"},
+                data="&".join(f"{k}={json.dumps(v)}" for k, v in data.items()),
+                params=params
+        ) as response:
+            if response.status != 200:
+                _LOGGER.error("Unable to connect to the login service: %s\n%s", response.status, await response.text())
+                return ""
+            try:
+                text = await response.text()
+                return (await response.json())["events"][0]["attributes"]["values"]["url"]
+            except json.JSONDecodeError:
+                if framework := re.findall('clientOutOfSync.*?Expected: ([\\w-]+?) Actual: (.*?)"', text):
+                    self._framework, actual = framework[0]
+                    _LOGGER.warning('Framework update from "%s" to "%s"', self._framework, actual)
+                    return await self._get_frontdoor_url(session, email, password)
+                _LOGGER.error("Unable to retrieve the frontdoor URL. Message: " + text)
+                return ""
+
+    async def _prepare_login(self, session, email, password):
+        if not (frontdoor_url := await self._get_frontdoor_url(session, email, password)):
+            return False
+
+        async with session.get(frontdoor_url) as resp:
+            if resp.status != 200:
+                _LOGGER.error("Unable to connect to the login service: %s", resp.status)
+                return False
+
+        params = {"retURL": "/SmartHome/apex/CustomCommunitiesLanding"}
+        async with session.get(f"{const.AUTH_API}/apex/ProgressiveLogin", params=params) as resp:
+            if resp.status != 200:
+                _LOGGER.error("Unable to connect to the login service: %s", resp.status)
+                return False
+        return True
+
+    async def _login(self, session):
+        nonce = secrets.token_hex(16)
+        nonce = f"{nonce[:8]}-{nonce[8:12]}-{nonce[12:16]}-{nonce[16:20]}-{nonce[20:]}"
+        params = {
+            "response_type": "token+id_token",
+            "client_id": const.CLIENT_ID,
+            "redirect_uri": urllib.parse.quote(f"{const.APP}://mobilesdk/detect/oauth/done"),
+            "display": "touch",
+            "scope": "api openid refresh_token web",
+            "nonce": nonce
+        }
+        params = "&".join([f"{k}={v}" for k, v in params.items()])
+        async with session.get(f"{const.AUTH_API}/services/oauth2/authorize?{params}") as resp:
+            if id_token := re.findall("id_token=(.*?)&", await resp.text()):
+                self._id_token = id_token[0]
+                return True
+        return False
+
+    async def authorize(self, email, password, mobile_id):
+        async with aiohttp.ClientSession() as session:
+            if not await self._prepare_login(session, email, password):
+                return False
+            if not await self._login(session):
+                return False
+
+            post_headers = {"Content-Type": "application/json", "id-token": self._id_token}
+            data = {"appVersion": const.APP_VERSION, "mobileId": mobile_id, "osVersion": const.OS_VERSION,
+                    "os": const.OS, "deviceModel": const.DEVICE_MODEL}
+            async with session.post(f"{const.API_URL}/auth/v1/login", headers=post_headers, json=data) as resp:
+                try:
+                    json_data = await resp.json()
+                except json.JSONDecodeError:
+                    _LOGGER.error("No JSON Data after POST: %s", await resp.text())
+                    return False
+                self._cognito_token = json_data["cognitoUser"]["Token"]
+        return True

pyhon/commands.py

@@ -0,0 +1,42 @@
+from parameter import HonParameterFixed, HonParameterEnum, HonParameterRange
+
+
+class HonCommand:
+    def __init__(self, name, attributes, connector, device, multi=None):
+        self._connector = connector
+        self._device = device
+        self._name = name
+        self._description = attributes.get("description", "")
+        self._parameters = self._create_parameters(attributes.get("parameters", {}))
+        self._ancillary_parameters = self._create_parameters(attributes.get("ancillaryParameters", {}))
+        self._multi = multi
+
+    def _create_parameters(self, parameters):
+        result = {}
+        for parameter, attributes in parameters.items():
+            match attributes.get("typology"):
+                case "range":
+                    result[parameter] = HonParameterRange(parameter, attributes)
+                case "enum":
+                    result[parameter] = HonParameterEnum(parameter, attributes)
+                case "fixed":
+                    result[parameter] = HonParameterFixed(parameter, attributes)
+        return result
+
+    @property
+    def parameters(self):
+        return {key: parameter.value for key, parameter in self._parameters.items()}
+
+    @property
+    def ancillary_parameters(self):
+        return {key: parameter.value for key, parameter in self._ancillary_parameters.items()}
+
+    async def send(self):
+        return await self._connector.send_command(self._device, self._name, self.parameters,
+                                                  self.ancillary_parameters)
+
+    async def get_programs(self):
+        return self._multi
+
+    async def set_program(self, program):
+        self._device.commands[self._name] = self._multi[program]

pyhon/const.py

@@ -0,0 +1,10 @@
+AUTH_API = "https://he-accounts.force.com/SmartHome"
+API_URL = "https://api-iot.he.services"
+APP = "hon"
+# All seen id's (different accounts, different devices) are the same, so I guess this hash is static
+CLIENT_ID = "3MVG9QDx8IX8nP5T2Ha8ofvlmjLZl5L_gvfbT9.HJvpHGKoAS_dcMN8LYpTSYeVFCraUnV.2Ag1Ki7m4znVO6"
+APP_VERSION = "1.51.9"
+OS_VERSION = 31
+OS = "android"
+DEVICE_MODEL = "exynos9820"
+LANGUAGE = "en"