Use certificate pinning for API requests

Author: lafriks

karcher/cli.py

@@ -12,7 +12,6 @@
 
 from karcher.exception import KarcherHomeException
 from karcher.karcher import KarcherHome
-from karcher.consts import Region
 
 try:
     from rich import print as echo

karcher/consts.py

@@ -116,3 +116,4 @@ class Product(str, Enum):
 PROTOCOL_VERSION = 'v1'
 APP_VERSION_CODE = 10004
 APP_VERSION_NAME = '1.0.4'
+SSL_CERTIFICATE_THUMBPRINT = bytes.fromhex('68:A3:68:92:5D:B3:DE:51:6A:80:64:FD:70:38:A3:49:45:D8:6E:DF:11:33:08:66:2C:87:85:A4:C9:F5:4A:10'.replace(':', ''))

karcher/karcher.py

@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: MIT
 # -----------------------------------------------------------
 
+import asyncio
 import collections
 import json
 import threading
@@ -14,8 +15,8 @@
 from .auth import Domains, Session
 from .countries import get_country_code, get_region_by_country
 from .consts import (
-    APP_VERSION_CODE, APP_VERSION_NAME, PROJECT_TYPE,
-    PROTOCOL_VERSION, REGION_URLS, ROBOT_PROPERTIES, TENANT_ID,
+    APP_VERSION_CODE, APP_VERSION_NAME, PROJECT_TYPE, PROTOCOL_VERSION,
+    REGION_URLS, ROBOT_PROPERTIES, SSL_CERTIFICATE_THUMBPRINT, TENANT_ID,
     Language, Region
 )
 from .device import Device, DeviceProperties
@@ -32,7 +33,11 @@ class KarcherHome:
     """Main class to access Karcher Home Robots API"""
 
     @classmethod
-    async def create(cls, country: str = 'GB', language: Language = Language.EN, session: aiohttp.ClientSession = None):
+    async def create(
+            cls,
+            country: str = 'GB',
+            language: Language = Language.EN,
+            session: aiohttp.ClientSession = None):
         """Create Karcher Home Robots API instance"""
 
         self = KarcherHome()
@@ -65,11 +70,13 @@ def __init__(self):
         self._mqtt = None
         self._device_props = {}
         self._wait_events = {}
+        self._http = None
+        self._http_external = False
 
     def __del__(self):
         """Destructor"""
 
-        self.close()
+        asyncio.run(self.close())
 
     async def close(self):
         """Close underlying connections"""
@@ -133,8 +140,7 @@ async def _request(self, method: str, url: str, **kwargs) -> aiohttp.ClientRespo
         headers['nonce'] = nonce
 
         kwargs['headers'] = headers
-        # TODO: Fix SSL
-        kwargs['verify_ssl'] = False
+        kwargs['ssl'] = aiohttp.Fingerprint(SSL_CERTIFICATE_THUMBPRINT)
         return await self._http.request(method, self._base_url + url, **kwargs)
 
     async def _download(self, url) -> bytes:

ruff.toml

@@ -4,3 +4,4 @@ extend-exclude = [
 
 [per-file-ignores]
 "tests/test_*.py" = ["E501"]
+"karcher/consts.py" = ["E501"]