"""Anker Power/Solix Cloud API class to handle a client connection session for an account.""" from asyncio import sleep from base64 import b64decode, b64encode import contextlib from datetime import datetime # TODO(COMPRESSION): from gzip import compress, decompress import hashlib import json import logging import os from pathlib import Path from random import randbytes, randrange import tempfile from types import SimpleNamespace from typing import Any import aiofiles import aiofiles.os from aiohttp import ClientSession, ClientTimeout from aiohttp.client_exceptions import ClientError from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import hashes, padding, serialization from cryptography.hazmat.primitives.asymmetric import ec from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.primitives.kdf.hkdf import HKDF from . import errors from .apitypes import ( API_COUNTRIES, API_ENDPOINTS, API_HEADERS, API_KEY_EXCHANGE, API_LOGIN, API_SERVERS, SolixDefaults, ) from .helpers import RequestCounter, generateTimestamp, getTimezoneGMTString, md5 _LOGGER: logging.Logger = logging.getLogger(__name__) class AnkerSolixClientSession: """Define the class to handle a client for Anker server authentication and API requests.""" # Public key of Anker Api servers _api_public_key_hex = "04c5c00c4f8d1197cc7c3167c52bf7acb054d722f0ef08dcd7e0883236e0d72a3868d9750cb47fa4619248f3d83f0f662671dadc6e2d31c2f41db0161651c7c076" def __init__( self, email: str, password: str, countryId: str, websession: ClientSession, logger: logging.Logger | None = None, ) -> None: """Initialize.""" self._countryId: str = countryId.upper() self._api_base: str | None = None for region, countries in API_COUNTRIES.items(): if self._countryId in countries: self._api_base = API_SERVERS.get(region) # default to EU server if not self._api_base: self._api_base = API_SERVERS.get("eu") self._email: str = email self._password: str = password self._session: ClientSession = websession self._loggedIn: bool = False self._testdir: str = str( (Path(__file__).parent / ".." / "examples" / "example1").resolve() ) # Flag for retry of any or certain error self._retry_attempt: bool | int = False # ensure folder for authentication caching exists auth_cache_dir = Path(__file__).parent / "authcache" if not os.access(auth_cache_dir.parent, os.W_OK): auth_cache_dir = Path(tempfile.gettempdir()) / "authcache" auth_cache_dir.mkdir(parents=True, exist_ok=True) # filename for authentication cache self._authFile: str = str(auth_cache_dir / f"{email}.json") self._authFileTime: float = 0 # Timezone format: 'GMT+01:00' self._timezone: str = getTimezoneGMTString() self._gtoken: str | None = None self._token: str | None = None self._token_expiration: datetime | None = None self._login_response: dict = {} self._request_delay: float = SolixDefaults.REQUEST_DELAY_DEF self._request_timeout: int = SolixDefaults.REQUEST_TIMEOUT_DEF self._last_request_time: datetime | None = None # define limit of same endpoint requests per minute self._endpoint_limit: int = SolixDefaults.ENDPOINT_LIMIT_DEF # Define authentication Encryption for password, using ECDH asymmetric key exchange for shared secret calculation, which must be used to encrypt the password using AES-256-CBC with seed of 16 # uncompressed public key from EU Anker server in the format 04 [32 byte x value] [32 byte y value] # Both, the EU and COM Anker server public key is the same and login response is provided for both upon an authentication request # However, if country ID assignment is to wrong server, no sites or devices will be listed for the authenticated account. # Create ECDH key pair using NIST P-256 curve SECP256R1 (identical to prime256v1) # get EllipticCurvePrivateKey self._private_key = ec.generate_private_key(ec.SECP256R1(), default_backend()) # get EllipticCurvePublicKey self._public_key = self._private_key.public_key() # get bytes of shared secret self._shared_key = self._private_key.exchange( ec.ECDH(), ec.EllipticCurvePublicKey.from_encoded_point( ec.SECP256R1(), bytes.fromhex(self._api_public_key_hex) ), ) # initialize logger for class if logger: self._logger = logger else: self._logger = _LOGGER self._logger.setLevel(logging.WARNING) if not self._logger.hasHandlers(): self._logger.addHandler(logging.StreamHandler()) # reset class variables self.nickname: str = "" self.mask_credentials: bool = True self.request_count: RequestCounter = RequestCounter() # Flag whether compression should be used (Actually not supported by Anker Power servers) self.compress_data: bool = False # handler for encryption self.encrypt_payload: bool = False self._eh: AnkerEncryptionHandler | None = None @property def email(self) -> str: """Get the email used for the active session.""" return self._email @property def countryId(self) -> str: """Get the country ID used for the active session.""" return self._countryId @property def server(self) -> str | None: """Get the server used for the active session.""" return self._api_base def logger(self, logger: logging.Logger | None = None) -> logging.Logger: """Get or set the logger for API client.""" if logger: self._logger = logger return self._logger def testDir(self, subfolder: str | None = None) -> str: """Get or set the subfolder for local API test files.""" if not subfolder or str(subfolder) == self._testdir: return self._testdir if not Path(subfolder).is_dir(): self._logger.error( "Specified test folder for api %s does not exist: %s", self.nickname, subfolder, ) else: self._testdir = str(subfolder) self._logger.info( "Set api %s test folder to: %s", self.nickname, self._testdir ) return self._testdir def logLevel(self, level: int | None = None) -> int: """Get or set the logger log level.""" if level is not None and isinstance(level, int): self._logger.setLevel(level) self._logger.info("Set api %s log level to: %s", self.nickname, level) return self._logger.getEffectiveLevel() def requestDelay(self, delay: float | None = None) -> float: """Get or set the api request delay in seconds.""" if ( delay is not None and isinstance(delay, float | int) and float(delay) != float(self._request_delay) ): self._request_delay = float( min( SolixDefaults.REQUEST_DELAY_MAX, max(SolixDefaults.REQUEST_DELAY_MIN, delay), ) ) self._logger.info( "Set api %s request delay to %.3f seconds", self.nickname, self._request_delay, ) return self._request_delay def requestTimeout(self, seconds: int | None = None) -> int: """Get or set the api request timeout in seconds.""" if ( seconds is not None and isinstance(seconds, float | int) and round(seconds) != self._request_timeout ): self._request_timeout = round( min( SolixDefaults.REQUEST_TIMEOUT_MAX, max(SolixDefaults.REQUEST_TIMEOUT_MIN, seconds), ) ) self._logger.info( "Set api %s request timeout to %s seconds", self.nickname, str(self._request_timeout), ) return self._request_timeout def endpointLimit(self, limit: int | None = None) -> int: """Get or set the api request limit per endpoint per minute.""" if ( limit is not None and isinstance(limit, float | int) and int(limit) != int(self._endpoint_limit) ): self._endpoint_limit = int(max(0, limit)) if self._endpoint_limit: self._logger.info( "Set api %s request limit to %s requests per endpoint per minute", self.nickname, self._endpoint_limit, ) else: self._logger.info( "Disabled api %s request limit and cleared %s throttled endpoints", self.nickname, len(self.request_count.throttled), ) self.request_count.throttled.clear() return self._endpoint_limit def generate_header(self) -> dict: """Generate common header fields for Api requests.""" # Start with fixed header fields header = API_HEADERS # {"content-type": "application/json", # "model-type": "DESKTOP", # "app-name": "anker_power", # "os-type": "android"} if self._countryId: header.update({"country": self._countryId}) if self._timezone: header.update({"timezone": self._timezone}) if self._token: header.update({"gtoken": self._gtoken, "x-auth-token": self._token}) if self.compress_data: header.update( { "accept-encoding": "gzip", # TODO(COMPRESSION): only response encoding seems to be accepted by servers # "content-type": "text/plain", # "content-encoding": "gzip", } ) return header async def _wait_delay( self, delay: float | None = None, endpoint: str | None = None ) -> None: """Wait at least for the defined Api request delay or for the provided delay in seconds since the last request occurred. If the endpoint is provided and a request limit is defined, the request will be throttled to avoid exceeding endpoint limit per minute. """ if delay is not None and isinstance(delay, float | int): delay = float( min( SolixDefaults.REQUEST_DELAY_MAX, max(SolixDefaults.REQUEST_DELAY_MIN, delay), ) ) else: delay = self._request_delay # throttle requests to same endpoint throttle = 0 if ( endpoint and delay == self._request_delay and self._endpoint_limit and endpoint in self.request_count.throttled ): same_requests = [ i for i in self.request_count.last_minute(details=True) if endpoint in i[1] ] # delay at least 1 minute from oldest request throttle = ( 65 - (datetime.now() - same_requests[0][0]).total_seconds() if len(same_requests) >= self._endpoint_limit else 0 ) if throttle: self._logger.warning( "Throttling next request of api %s for %.1f seconds to maintain request limit of %s for endpoint %s", self.nickname, throttle, self._endpoint_limit, endpoint, ) await sleep( max( 0, throttle, delay - (datetime.now() - self._last_request_time).total_seconds() if isinstance(self._last_request_time, datetime) else 0, ) ) async def async_authenticate(self, restart: bool = False) -> bool: """Authenticate with server and get an access token. If restart is not enforced, cached login data may be used to obtain previous token.""" if restart: self._token = None self._token_expiration = None self._gtoken = None self._loggedIn = False self._login_response = {} self._eh = None self._authFileTime = 0 self.nickname = "" # remove auth file if existing if Path(self._authFile).is_file(): with contextlib.suppress(Exception): Path(self._authFile).unlink() # First check if cached login response is available and login params can be filled, otherwise query server for new login tokens if Path(self._authFile).is_file(): data = await self.loadFromFile(self._authFile) self._authFileTime = Path(self._authFile).stat().st_mtime self._logger.debug( "Cached Login for %s from %s:", self.mask_values(self._email), datetime.fromtimestamp(self._authFileTime).isoformat(), ) self._logger.debug( "%s", self.mask_values( data, "user_id", "auth_token", "email", "geo_key", "ap_cloud_user_id", ), ) # clear retry attempt to allow retry for authentication refresh if isinstance(self._retry_attempt, bool): self._retry_attempt = False else: self._logger.debug("Fetching new Login credentials from server") now = datetime.now().astimezone() # set retry attempt to avoid retry on failed authentication self._retry_attempt = True auth_resp = await self.request( "post", API_LOGIN, json={ "ab": self._countryId, "client_secret_info": { # client public_key is uncompressed format of points in hex (0x04 + 32 Byte + 32 Byte) "public_key": self._rawPublicKey() }, "enc": 0, "email": self._email, # password is AES-256-CBC encrypted by the ECDH shared key derived from server public key and local private key "password": self._encryptApiData(self._password), # time_zone is offset in ms, e.g. 'GMT+01:00' => 3600000 "time_zone": round(datetime.utcoffset(now).total_seconds() * 1000), # transaction is Unix Timestamp in ms as string "transaction": generateTimestamp(in_ms=True), }, ) data = auth_resp.get("data", {}) self._logger.debug( "Login Response: %s", self.mask_values( data, "user_id", "auth_token", "email", "geo_key", "ap_cloud_user_id", ), ) self._loggedIn = True # Cache login response in file for reuse async with aiofiles.open(self._authFile, "w", encoding="utf-8") as authfile: await authfile.write(json.dumps(data, indent=2, skipkeys=True)) self._logger.debug("Response cached in file: %s", self._authFile) self._authFileTime = Path(self._authFile).stat().st_mtime # Update the login params self._login_response = dict(data) self._token = data.get("auth_token") self.nickname = data.get("nick_name") or "" if data.get("token_expires_at"): self._token_expiration = datetime.fromtimestamp( data.get("token_expires_at") ) else: self._token_expiration = None if data.get("user_id"): # gtoken is MD5 hash of user_id from login response self._gtoken = md5(data.get("user_id")) # reset retry flag upon valid authentication response for normal request retry attempts if isinstance(self._retry_attempt, bool): self._retry_attempt = False # Cached login: tokens are loaded above; mark session valid (HA relies on this too) self._loggedIn = bool(self._token and self._gtoken) else: self._gtoken = None self._loggedIn = False return self._loggedIn async def request( # noqa: C901 self, method: str, endpoint: str, *, headers: dict | None = None, json: dict | None = None, # pylint: disable=redefined-outer-name ) -> dict: """Handle all requests to the API. This is also called recursively by login requests if necessary.""" if not isinstance(headers, dict): headers = {} if not isinstance(json, dict): json = {} # check token expiration (7 days) if ( self._token_expiration and (self._token_expiration - datetime.now()).total_seconds() < 60 ): self._logger.warning( "WARNING: Access token expired, reloading cache%s", (" for " + str(self.nickname)) if self.nickname else "", ) self._authFileTime = 0 if not await self.async_authenticate(): self._logger.warning( "Cache reload after expiry failed; API re-login may be required" ) await self.async_authenticate(restart=True) # For non-Login requests, ensure authentication will be updated if not logged in yet or cached file was refreshed if endpoint != API_LOGIN and ( not self._loggedIn or ( Path(self._authFile).is_file() and self._authFileTime != Path(self._authFile).stat().st_mtime ) ): await self.async_authenticate() url: str = f"{self._api_base}/{endpoint}" # use required headers and merge provided/optional headers mergedHeaders = self.generate_header() mergedHeaders.update(headers) # TODO(ENCRYPTION): Handle payload encryption once known if self.encrypt_payload and self._login_response: if not self._eh and self._token: # init encryption handler self._eh = AnkerEncryptionHandler( login_response=self._login_response, session=self._session, logger=self._logger, ) if not self._eh.shared_secret: # Perform key exchange await self._eh.perform_key_exchange( api_base=self._api_base, headers=mergedHeaders ) # App Example request with encrypted payload # curl --request POST \ # --url https://ankerpower-api-eu.anker.com/power_service/v1/app/device/get_device_attrs \ # --compressed \ # --header 'accept-encoding: gzip' \ # --header 'app-name: anker_power' \ # --header 'app-version: 3.6.0' \ # --header 'content-type: text/plain' \ # --header 'gtoken: GTOKEN' \ # --header 'host: ankerpower-api-eu.anker.com' \ # --header 'language: en' \ # --header 'x-app-key: ' \ # --header 'x-auth-token: TOKEN' \ # --header 'x-encryption-info: algo_ecdh' \ # --header 'x-replay-info: replay' \ # --header 'x-key-ident: 487c4d59721f02c22112c46c1ea038bf' \ # --header 'x-request-once: cba657a9e6b9da45d89658b6a281ff97' \ # --header 'x-request-ts: 1745103679' \ # --header 'x-signature: d93f441a9951cc4f4069f13d21248764619e42336315dcc93022f93fddcc7cbb' \ # --data 'MTc0NTEwMzY3OTIyMjAwMIUJ7p7z1Zk4ra+8lqNwWY8l//5+uEyzztV+MglhAvvv3/xuqxYyn9heRVIRHje482N8BuZEVklJEnernTGFyEY=' # App Example response with encrypted payload # {"signature": "98444e7105fbdc29a4e969d8b24d6e757ea9c495c6c8158cefbecdaf2cbf72b0", # "msg": "success!", # "code": 0, # "trace_id": "9d0bd9c0a328bcfde9e60f5a33284977", # "data": "ybF4Sg9vCKt3V75DnvTddszD0m6qJH4ZhjcN4DQC2RMlvqCthrJ1qK4DLvJ7l8jwYYbMooaE2QXmtfGebt8d1wfzCnl3XqLhpM8dxXX7ghw="} # Mandatory extra api header arguments for key exchange # 'x-app-key': '' # empty is fine # 'x-encryption-info': 'algo_ecdh' # 'x-replay-info': 'replay' # 'x-auth-ts': [authentication timestamp] # Unix Timestamp in s as string # 'x-key-ident': [generated key ident] # 16 Byte MD5 hash # 'x-request-once': [generated request once] # 16 Byte MD5 hash # 'x-request-ts': [request timestamp] # Unix Timestamp in s as string # 'x-signature': [generated signature] # 32 Byte SHA256 hash # Response will return encrypted body and a signature field timestamp = generateTimestamp() mergedHeaders.update( self._eh.generate_x_header(timestamp=timestamp, data=json) | { "content-type": "text/plain", } ) self._logger.debug("Request Url: %s %s", method.upper(), url) self._logger.debug( "Request Headers: %s", self.mask_values(mergedHeaders, "x-auth-token", "gtoken"), ) if endpoint in [ API_LOGIN, API_KEY_EXCHANGE, API_ENDPOINTS["get_token_by_userid"], API_ENDPOINTS["get_shelly_status"], ]: body_text = str( self.mask_values( json, "user_id", "auth_token", "email", "geo_key", "token", "password", "ap_cloud_user_id", ) ) else: body_text = str(json) self._logger.debug("Request Body: %s", body_text) # enforce configured delay between any subsequent request await self._wait_delay(endpoint=endpoint) # uncompressed body must use json parameter, pre-compressed body must use data parameter data = {} # predefine response to handle TimeoutError like 522 timeouts from server resp = SimpleNamespace(status=0) try: # make the request, auto_decompression of body enabled by default async with self._session.request( method, url, headers=mergedHeaders, json=json, # TODO(COMPRESSION): only response encoding seems to be accepted by servers # json=None if self.compress_data else json, # data=compress(str(json).encode()) if self.compress_data else None, timeout=ClientTimeout(total=self._request_timeout), ) as resp: self._last_request_time = datetime.now() self.request_count.add( request_time=self._last_request_time, request_info=(f"{method.upper()} {url} {body_text}").strip(), ) # request handler has auto-decompression enabled self._logger.debug( "Api %s response received for request: %s %s", self.nickname, method, url, ) # print response headers self._logger.debug("Response Headers: %s", resp.headers) # get first the body text for usage in error detail logging if necessary body_text = await resp.text() resp.raise_for_status() # any response status >= 400 # get json data without strict checking for json content data = await resp.json(content_type=None) if not data: self._logger.error( "Api %s no data response for request: %s %s\nResponse Text: %s", self.nickname, method.upper(), url, body_text, ) raise ClientError( # noqa: TRY301 f"Api {self.nickname} no data response for request: {method.upper()} {url}" ) if endpoint == API_LOGIN: self._logger.debug( "Response Data: %s", self.mask_values( data, "user_id", "auth_token", "email", "geo_key", "ap_cloud_user_id", ), ) else: self._logger.debug("Response Data: %s", data) # valid client response at this point, mark login to avoid repeated authentication self._loggedIn = True # check the Api response status code in the data errors.raise_error(data) # reset retry flag for normal request retry attempts self._retry_attempt = False # TODO(ENCRYPTION): data field has to be decoded when encrypted and signature field in response if self.encrypt_payload and data.get("signature"): data["data"] = self._eh.decryptApiData(data.get("data")) return data # Exception from ClientSession based on standard response status codes except (ClientError, TimeoutError) as err: if isinstance(err, TimeoutError): resp.status = 522 body_text = "Timeout Error" # Prepare data dict for Api error lookup if not data: data = {} if not hasattr(data, "code"): data["code"] = resp.status if not hasattr(data, "msg"): data["msg"] = body_text if resp.status in [401, 403]: # Unauthorized or forbidden request self._logger.error( "Api %s Error %s for request: %s %s\nResponse Text: %s", self.nickname, err, method.upper(), url, body_text, ) # reattempt authentication with same credentials if cached token was kicked out # retry attempt is set if login response data were not cached to fail immediately if not self._retry_attempt: self._logger.warning( "Invalid Login, retrying authentication%s", (" for " + str(self.nickname)) if self.nickname else "", ) self._authFileTime = 0 if await self.async_authenticate(): return await self.request( method, endpoint, headers=headers, json=json ) if Path(self._authFile).is_file(): self._logger.error( "Login failed for %s: cached token invalid (mobile app " "may have replaced the API token). Copy fresh authcache " "from HA or restart after app logout.", self._email, ) elif await self.async_authenticate(restart=True): return await self.request( method, endpoint, headers=headers, json=json ) self._logger.error("Login failed for user %s", self._email) errors.raise_error(data, prefix=f"Login failed for user {self._email}") # catch error if Api code not defined raise errors.AuthorizationError( f"Login failed for user {self._email}" ) from err if resp.status == 429: # Too Many Requests for endpoint, repeat once after throttle delay and add endpoint to throttle if self._retry_attempt not in [True, 429] and self._endpoint_limit: self._retry_attempt = resp.status self.request_count.add_throttle(endpoint=endpoint) self._logger.warning( "Api %s exceeded request limit with %s known requests in last minute, throttle will be enabled for endpoint: %s", self.nickname, len( [ i for i in self.request_count.last_minute(details=True) if endpoint in i[1] ] ), endpoint, ) return await self.request( method, endpoint, headers=headers, json=json ) # Raise error if retry failed too, add stats to message self._logger.error( "Api %s Error %s for request: %s %s\nResponse Text: %s", self.nickname, err, method.upper(), url, body_text, ) errors.raise_error( data, prefix=f"Api {self.nickname} Too Many Requests: {self.request_count}", ) elif resp.status in [502, 504, 522]: # Server may be temporarily overloaded and does not respond # 502 is Gateway error # 504 is Gateway timeout error # 522 is Server timeout error if self._retry_attempt not in [True, 502, 504, 522]: self._retry_attempt = resp.status delay = randrange(2, 6) # random wait time 2-5 seconds self._logger.info( "Http error '%s', retrying request of api %s after delay of %s seconds for endpoint: %s", "Timeout Error" if isinstance(err, TimeoutError) else resp.status, self.nickname, delay, endpoint, ) await self._wait_delay(delay=delay) return await self.request( method, endpoint, headers=headers, json=json ) self._logger.error( "Api %s Error %s for request: %s %s\nResponse Text: %s", self.nickname, err, method.upper(), url, body_text, ) # raise Anker Solix error if code is known errors.raise_error(data) # raise Client error otherwise raise ClientError( f"Api {self.nickname} Request Error: {str(err) or 'Timeout'}, response={body_text}" ) from err except errors.AnkerSolixError as err: # Other Exception from API if isinstance(err, errors.BusyError): # Api fails to respond to standard query, repeat once after delay self._logger.error( "Api %s Busy Error %s for request: %s %s\nResponse Text: %s", self.nickname, err, method.upper(), url, body_text, ) if self._retry_attempt not in [True, 21105]: self._retry_attempt = 21105 delay = randrange(2, 6) # random wait time 2-5 seconds self._logger.warning( "Server busy, retrying request of api %s after delay of %s seconds for endpoint %s", self.nickname, delay, endpoint, ) await self._wait_delay(delay=delay) return await self.request( method, endpoint, headers=headers, json=json ) log = ( self._logger.warning if isinstance(err, errors.ItemNotFoundError) else self._logger.error ) log( "Api %s Error %s for request: %s %s\nResponse Text: %s", self.nickname, err, method.upper(), url, body_text, ) raise def _rawPublicKey(self) -> bytes: """Generate raw client public_key in uncompressed format of points in hex (0x04 + 32 Byte + 32 Byte).""" return self._public_key.public_bytes( encoding=serialization.Encoding.X962, format=serialization.PublicFormat.UncompressedPoint, ).hex() def _encryptApiData(self, raw_data: str) -> str: """Return Base64 encoded secret as utf-8 encoded string using the shared secret with seed of 16 for the encryption.""" # Password must be UTF-8 encoded and AES-256-CBC encrypted with block size of 16 # Create AES cipher cipher = Cipher( algorithms.AES(self._shared_key), modes.CBC(self._shared_key[:16]), backend=default_backend(), ) # Encrypt encryptor = cipher.encryptor() # Use default PKCS7 padding for incomplete AES blocks padder = padding.PKCS7(128).padder() raw_padded = padder.update(raw_data.encode()) + padder.finalize() return (b64encode(encryptor.update(raw_padded) + encryptor.finalize())).decode() def mask_values(self, data: dict | str, *args: str) -> dict | str: """Mask values in dictionary for provided keys or the given string.""" if self.mask_credentials: if isinstance(data, str): datacopy: dict = {"text": data} args: list = ["text"] else: datacopy = data.copy() for key in args: if old := datacopy.get(key): new = "" for idx in range(0, len(old), 16): new = new + ( f"{old[idx : idx + 2]}###masked###{old[idx + 14 : idx + 16]}" ) new = new[: len(old)] datacopy[key] = new if isinstance(data, str): return datacopy.get("text") return datacopy return data async def loadFromFile(self, filename: str | Path) -> dict: """Load json data from given file for testing.""" filename = str(filename) if self.mask_credentials: masked_filename = filename.replace( self._email, self.mask_values(self._email) ) else: masked_filename = filename try: if Path(filename).is_file(): async with aiofiles.open(filename, encoding="utf-8") as file: data = json.loads(await file.read() or "{}") self._logger.debug("Loaded JSON from file %s:", masked_filename) self._logger.debug( "Data: %s", self.mask_values( data, "user_id", "auth_token", "email", "geo_key", "token", "ap_cloud_user_id", ), ) self.request_count.add(request_info=f"LOAD {masked_filename}") return data except OSError as err: self._logger.error( "ERROR: Failed to load JSON from file %s\n%s", masked_filename, err ) return {} async def saveToFile(self, filename: str | Path, data: dict | None = None) -> bool: """Save json data to given file for testing.""" filename = str(filename) if self.mask_credentials: masked_filename = filename.replace( self._email, self.mask_values(self._email) ) else: masked_filename = filename if not data: data = {} try: async with aiofiles.open(filename, "w", encoding="utf-8") as file: await file.write(json.dumps(data, indent=2)) self._logger.debug("Saved JSON to file %s:", masked_filename) return True except OSError as err: self._logger.error( "ERROR: Failed to save JSON to file %s\n%s", masked_filename, err ) return False async def deleteModifiedFile(self, filename: str | Path) -> bool: """Delete given modified json file for testing.""" filename = str(filename) if "modified" not in filename: return False if self.mask_credentials: masked_filename = filename.replace( self._email, self.mask_values(self._email) ) else: masked_filename = filename try: await aiofiles.os.remove(masked_filename) self._logger.debug("Remove modified JSON file %s:", masked_filename) except OSError as err: self._logger.error( "ERROR: Failed to remove modified JSON file %s\n%s", masked_filename, err, ) return False return True async def get_mqtt_info(self) -> dict: r"""Get the Anker MQTT server info with account certificates from session. Example data: {"user_id": "1541fc0a3db5a23e3c4ee27c6ed0616444e1ab8c","app_name": "anker_power","thing_name": "1541fc0a3db5a23e3c4ee27c6ed0616444e1ab8c-anker_power", "certificate_id": "36167916173028037311485710012973829433","certificate_pem": "-----BEGIN CERTIFICATE-----\n\n-----END CERTIFICATE-----\n", "private_key": "-----BEGIN RSA PRIVATE KEY-----\n\n-----END RSA PRIVATE KEY-----\n","public_key": "", "endpoint_addr": "aiot-mqtt-eu.anker.com","aws_root_ca1_pem": "-----BEGIN CERTIFICATE-----\n\n-----END CERTIFICATE-----", "origin": "","pkcs12": ""} """ return (await self.request("post", API_ENDPOINTS["get_mqtt_info"])).get( "data" ) or {} def get_login_info(self, key: str) -> Any | None: """Get a certain key from authenticated client login info.""" return self._login_response.get(key) class AnkerEncryptionHandler: """Anker Solix encryption handler class. ATTENTION: This class is experimental and does not work yet since various x header field value generation is unknown. """ def __init__( self, login_response: dict, session: ClientSession, logger: logging.Logger | None = None, ) -> None: """Initialize the encryption handler.""" self._login_response = login_response self._session = session # Create ECDH key pair for encryption key exchange using NIST P-256 curve self.private_key = ec.generate_private_key(ec.SECP256R1()) self.public_key = self.private_key.public_key() self.server_public_key = None self.shared_secret = None # initialize logger for class if logger: self._logger = logger else: self._logger = _LOGGER self._logger.setLevel(logging.WARNING) if not self._logger.hasHandlers(): self._logger.addHandler(logging.StreamHandler()) def _generate_client_key(self, timestamp: str) -> str: """Generate the client public key in Anker's format.""" # Get raw public key bytes raw_public_key = self.public_key.public_bytes( encoding=serialization.Encoding.X962, format=serialization.PublicFormat.UncompressedPoint, ) # Construct the key with timestamp and marker as observed in Api communication key_data = bytearray() key_data.extend(timestamp.encode()) # First 16 bytes: timestamp key_data.append(0x1F) # Marker byte ? key_data.extend(raw_public_key) # Public key uncompressed point format in bytes # return base64 encoded string representation of key format return b64encode(key_data).decode() def generate_x_header(self, timestamp: str, data: dict) -> dict: """Generate extra header fields for encryption.""" # request_once using timestamp in s and random? data (to be adjusted as required) request_once = md5(timestamp.encode() + randbytes(16)) request_once = md5( timestamp.encode() + self._login_response.get("geo_key").encode() ) # key_ident using timestamp in ms and random? data (to be adjusted as required) key_ident = md5(timestamp.encode() + randbytes(16)) key_ident = md5( timestamp.encode() + self._login_response.get("auth_token").encode() ) # SHA256 signature using timestamp, request-once, key_ident and body? (to be adjusted as required) signature = hashlib.sha256( timestamp.encode() + request_once.encode() + key_ident.encode() + json.dumps(data).encode() ).hexdigest() return { "content-type": "text/plain", "x-app-key": "", # Can be empty "x-encryption-info": "algo_ecdh", "x-replay-info": "replay", "x-key-ident": key_ident, "x-request-once": request_once, "x-request-ts": timestamp, "x-signature": signature, } async def perform_key_exchange( self, api_base: str, auth_ts: str | None = None, headers: dict | None = None, ) -> str | None: """Perform the key exchange with Anker's server and return shared secret.""" if not isinstance(headers, dict): headers = {} timestamp = generateTimestamp() if not auth_ts: auth_ts = timestamp # Prepare request url = f"{api_base}/{API_KEY_EXCHANGE}" data = {"client_public_key": self._generate_client_key(timestamp=auth_ts)} # obtain encryption header fields and add/modify fields for key exchange request headers.update( self.generate_x_header(timestamp=timestamp, data=data) | { "content-type": "application/json", "x-auth-ts": auth_ts, } ) self._logger.debug("Request Url: %s %s", "POST", url) self._logger.debug( "Request Headers: %s", headers, ) self._logger.debug("Request Body: %s", str(data)) async with self._session.request( "post", url, headers=headers, json=data, timeout=ClientTimeout(total=self._request_timeout), ) as resp: try: self._logger.debug("AnkerEncryptionHandler request response received") # print response headers self._logger.debug("Response Headers: %s", resp.headers) # get first the body text for usage in error detail logging if necessary body_text = await resp.text() data = {} resp.raise_for_status() # any response status >= 400 # get json data without strict checking for json content data = await resp.json(content_type=None) if not data: self._logger.error("Response Text: %s", body_text) raise ClientError( # noqa: TRY301 f"No data response while requesting {API_KEY_EXCHANGE}" ) self._logger.debug("Response Data: %s", data) self.server_public_key = (data.get("data") or {}).get( "server_public_key" ) or None if not self.server_public_key: return None return self.derive_shared_key(self.server_public_key) # Exception from ClientSession based on standard response status codes except ClientError as err: # Prepare data dict for Api error lookup if not data: data = {} if not hasattr(data, "code"): data["code"] = resp.status if not hasattr(data, "msg"): data["msg"] = body_text # raise Client error otherwise raise ClientError( f"AnkerEncryptionHandler Key exchange failed: {err}, response={body_text}", ) from err def derive_shared_key(self, server_public_key_b64: str) -> bytes: """Derive the shared AES key from the server's public key.""" # Decode server's public key server_key_bytes = b64decode(server_public_key_b64) # Extract the actual key portion (first 96 bytes based on analysis, but may need to be adopted) key_portion = server_key_bytes[:96] # Load server's public key server_public_key = ec.EllipticCurvePublicKey.from_encoded_point( ec.SECP256R1(), key_portion ) # Compute shared secret shared_secret = self.private_key.exchange(ec.ECDH(), server_public_key) # Derive AES key using HKDF hkdf = HKDF( algorithm=hashes.SHA256(), length=32, salt=None, info=b"ecdh handshake" ) self.shared_secret = hkdf.derive(shared_secret) return self.shared_secret def decryptApiData(self, encrypted_payload: str) -> str: """Decrypt an encrypted payload using the derived shared AES key.""" # Decode base64 payload encrypted_data = b64decode(encrypted_payload) # Split IV and ciphertext by first 16 bytes iv = encrypted_data[:16] ciphertext = encrypted_data[16:] # Create AES cipher cipher = Cipher(algorithms.AES(self.shared_secret), modes.CBC(iv)) # Decrypt decryptor = cipher.decryptor() decrypted = decryptor.update(ciphertext) + decryptor.finalize() # Remove PKCS7 padding padding_length = decrypted[-1] return decrypted[:-padding_length].decode()