diff --git a/packages/modules/vehicles/byd/THIRD_PARTY_NOTICES.md b/packages/modules/vehicles/byd/THIRD_PARTY_NOTICES.md new file mode 100644 index 0000000000..cdd8a8854d --- /dev/null +++ b/packages/modules/vehicles/byd/THIRD_PARTY_NOTICES.md @@ -0,0 +1,58 @@ +# Third-party notices + +This module ports protocol logic from two MIT-licensed projects: +`crypto.py` (including `bangcle_tables.bin`) from +[pyBYD](https://github.com/jkaberg/pyBYD), and `device_fingerprint.py` from +[ioBroker.byd](https://github.com/TA2k/ioBroker.byd). + +## pyBYD + +``` +MIT License + +Copyright (c) 2026 Joel Kåberg + +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. +``` + +## ioBroker.byd + +``` +MIT License + +Copyright (c) 2026 TA2k + +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. +``` diff --git a/packages/modules/vehicles/byd/__init__.py b/packages/modules/vehicles/byd/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/packages/modules/vehicles/byd/api.py b/packages/modules/vehicles/byd/api.py new file mode 100644 index 0000000000..952e89b87a --- /dev/null +++ b/packages/modules/vehicles/byd/api.py @@ -0,0 +1,428 @@ +#!/usr/bin/env python3 +"""BYD-Cloud-API-Client für openWB. + +Protokoll (Login, Envelope-Aufbau, Feldnamen) reverse-engineered von +https://github.com/jkaberg/pyBYD (MIT) und https://github.com/TA2k/ioBroker.byd, +hier synchron mit `requests` statt `aiohttp`/`asyncio` nachgebaut, damit es unter +dem in openWB verwendeten Python 3.9 läuft (pyBYD selbst braucht Python 3.11+, +technisch aber nur wegen der genutzten Bibliotheken, nicht wegen des Protokolls). + +Ablauf pro Zyklus (fetch_soc): + 1. Falls noch keine oder eine abgelaufene Session für diesen Account im + Prozess-Cache liegt: Login (Passwort-basiert, AES-verschlüsseltes Payload). + 2. vehicleRealTimeRequest (Trigger) -> vehicleRealTimeResult (Poll, mit Sleep- + Intervall), bis SoC-Daten vorliegen oder die Versuche aufgebraucht sind. + 3. Bei Session-Fehlercode: einmalig neu einloggen und den Request wiederholen. + +Die Session (userId/signToken/encryToken) wird pro Account (username) im Prozess +zwischengespeichert, damit nicht bei jedem openWB-Zyklus neu eingeloggt wird. Mehrere +Fahrzeuge desselben Accounts teilen sich Session, Login-Lock und Geräte-Fingerprint, +da BYD pro Account nur eine aktive Session erlaubt. +""" +import json +import logging +import secrets +import threading +import time +from typing import Any, Dict, Optional + +from modules.common import req +from modules.common.component_state import CarState +from modules.vehicles.byd import crypto, device_fingerprint +from modules.vehicles.byd.config import BydConfiguration + +log = logging.getLogger(__name__) + +BASE_URL = "https://dilinkappoversea-eu.byd.auto" + +LOGIN_ENDPOINT = "/app/account/login" +REALTIME_TRIGGER_ENDPOINT = "/vehicleInfo/vehicle/vehicleRealTimeRequest" +REALTIME_POLL_ENDPOINT = "/vehicleInfo/vehicle/vehicleRealTimeResult" + +SESSION_EXPIRED_CODES = {"1002", "1005", "1010"} +ENDPOINT_NOT_SUPPORTED_CODES = {"1001"} +VEHICLE_UNREACHABLE_CODES = {"6002"} + +POLL_ATTEMPTS = 10 +POLL_INTERVAL_S = 1.5 + +_ENERGY_TYPE_CODES = {"ev": "0", "hybrid": "2"} + +# Prozess-weiter Session-Cache je Account: {username: {"user_id", "sign_token", +# "encry_token", "created_at"}} +_session_cache: Dict[str, Dict[str, Any]] = {} +_SESSION_TTL_S = 12 * 3600 + +# Jeder fehlgeschlagene Login verbraucht einen von wenigen Versuchen, bevor der BYD-Account +# gesperrt wird - Cooldown verhindert, dass jeder Zyklus (alle 5min) den Account weiter verbrennt. +_login_failure_cache: Dict[str, Dict[str, Any]] = {} +_LOGIN_FAILURE_COOLDOWN_S = 3600 + +# Fingerprint-Cache je Account, analog zu _session_cache +_fingerprint_cache: Dict[str, str] = {} + +# Lock je Account, da update_soc.py jedes Fahrzeug in einem eigenen Thread aktualisiert +_account_locks: Dict[str, threading.Lock] = {} +_locks_meta_lock = threading.Lock() + + +def _account_lock(username: str) -> threading.Lock: + with _locks_meta_lock: + lock = _account_locks.get(username) + if lock is None: + lock = threading.Lock() + _account_locks[username] = lock + return lock + + +def resolve_account_fingerprint(username: str, persisted: str) -> str: + """JSON-serialisierter Fingerprint für diesen Account, konsistent über alle Fahrzeuge + desselben Accounts hinweg. soc.fetch() persistiert das Ergebnis in der Fahrzeug-Config.""" + with _locks_meta_lock: + cached = _fingerprint_cache.get(username) + if cached: + return cached + fingerprint = persisted or json.dumps(device_fingerprint.generate()) + _fingerprint_cache[username] = fingerprint + return fingerprint + + +class BydApiError(Exception): + """API hat einen Fehlercode != 0 geliefert.""" + + +class BydAuthenticationError(BydApiError): + """Login fehlgeschlagen oder Session ungültig.""" + + +def _device(config: BydConfiguration) -> Dict[str, str]: + """Fallback für Direktaufrufe ohne soc.fetch() (zB Tests): generiert nur im + Speicher, ohne MQTT-Persistenz.""" + if not config.device_fingerprint: + fingerprint = device_fingerprint.generate() + config.device_fingerprint = json.dumps(fingerprint) + log.debug("BYD: Fingerprint generiert (ungespeichert): %s %s", + fingerprint["mobileBrand"], fingerprint["mobileModel"]) + return json.loads(config.device_fingerprint) + + +def _post_secure(endpoint: str, outer_payload: Dict[str, Any]) -> Dict[str, Any]: + """Bangcle-verschlüsselt das äußere Envelope, sendet es per POST und + entschlüsselt die Antwort wieder zu einem Dict.""" + encoded = crypto.bangcle_encode(json.dumps(outer_payload, separators=(",", ":"))) + session = req.get_http_session() + response = session.post( + f"{BASE_URL}{endpoint}", + data=json.dumps({"request": encoded}), + headers={ + "accept-encoding": "identity", + "content-type": "application/json; charset=UTF-8", + "user-agent": "okhttp/4.12.0", + }, + timeout=15, + ) + response.raise_for_status() + body = response.json() + response_str = body.get("response") + if not response_str: + raise BydApiError(f"{endpoint}: Antwort enthält kein 'response'-Feld") + + decoded_text = crypto.bangcle_decode(response_str).decode("utf-8").strip() + if decoded_text.startswith("F{") or decoded_text.startswith("F["): + decoded_text = decoded_text[1:] + return json.loads(decoded_text) + + +def _login(config: BydConfiguration) -> Dict[str, Any]: + log.debug("BYD: Login für %s", config.username) + now_ms = int(time.time() * 1000) + random_hex = secrets.token_hex(16).upper() + req_timestamp = str(now_ms) + device = _device(config) + imei_md5 = device["imeiMd5"] + + inner = { + "agreeStatus": "0", + "agreementType": "[1,2]", + "appInnerVersion": "323", + "appVersion": "3.2.3", + "deviceName": device["mobileBrand"] + device["mobileModel"], + "deviceType": device["deviceType"], + "imeiMD5": imei_md5, + "isAuto": "1", + "mobileBrand": device["mobileBrand"], + "mobileModel": device["mobileModel"], + "networkType": device["networkType"], + "osType": device["osType"], + "osVersion": device["osVersion"], + "random": random_hex, + "softType": "0", + "timeStamp": req_timestamp, + "timeZone": "Europe/Amsterdam", + } + encry_data = crypto.aes_encrypt_hex(json.dumps(inner, separators=(",", ":")), crypto.pwd_login_key(config.password)) + password_md5 = crypto.md5_hex(config.password) + + sign_fields = { + **inner, + "countryCode": config.country_code, + "functionType": "pwdLogin", + "identifier": config.username, + "identifierType": config.identifier_type, + "language": "en", + "reqTimestamp": req_timestamp, + } + sign = crypto.sha1_mixed(crypto.build_sign_string(sign_fields, password_md5)) + + outer = { + "countryCode": config.country_code, + "encryData": encry_data, + "functionType": "pwdLogin", + "identifier": config.username, + "identifierType": config.identifier_type, + "imeiMD5": imei_md5, + "isAuto": "1", + "language": "en", + "reqTimestamp": req_timestamp, + "sign": sign, + "signKey": config.password, + "ostype": device["ostype"], + "imei": device["imei"], + "mac": device["mac"], + "model": device["model"], + "sdk": device["sdk"], + "mod": device["mod"], + "serviceTime": str(int(time.time() * 1000)), + } + outer["checkcode"] = crypto.compute_checkcode(outer) + + response = _post_secure(LOGIN_ENDPOINT, outer) + if str(response.get("code")) != "0": + raise BydAuthenticationError( + f"Login fehlgeschlagen: code={response.get('code')} message={response.get('message')}") + + respond_data = response.get("respondData") + if not respond_data: + raise BydAuthenticationError("Login-Antwort enthält kein respondData") + + plaintext = crypto.aes_decrypt_utf8(respond_data, crypto.pwd_login_key(config.password)) + token = (json.loads(plaintext) or {}).get("token") or {} + if not token.get("userId") or not token.get("signToken") or not token.get("encryToken"): + raise BydAuthenticationError("Login-Antwort enthält keine vollständigen Token-Felder") + + return { + "user_id": str(token["userId"]), + "sign_token": str(token["signToken"]), + "encry_token": str(token["encryToken"]), + "created_at": time.monotonic(), + } + + +def _get_session(config: BydConfiguration, force_relogin: bool = False, + rejected_session: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: + cache_key = config.username + # Lock verhindert parallelen Login zweier Fahrzeuge desselben Accounts + with _account_lock(cache_key): + cached = _session_cache.get(cache_key) + if (not force_relogin and cached is not None + and (time.monotonic() - cached["created_at"]) < _SESSION_TTL_S): + return cached + if force_relogin and cached is not None and cached is not rejected_session: + # bereits von einem anderen Fahrzeug desselben Accounts erneuert -> verwenden statt + # erneut einzuloggen (nur eine aktive Session pro Account) + return cached + + failure = _login_failure_cache.get(cache_key) + if failure is not None: + age_s = time.monotonic() - failure["timestamp"] + if age_s < _LOGIN_FAILURE_COOLDOWN_S: + raise BydAuthenticationError( + f"BYD-Login für {config.username} wird für weitere " + f"{int(_LOGIN_FAILURE_COOLDOWN_S - age_s)}s nicht erneut versucht (letzter " + f"Versuch vor {int(age_s)}s fehlgeschlagen mit: {failure['message']}). Account/" + "Zugangsdaten prüfen, um keine Login-Fehlversuche zu verbrauchen.") + + try: + session = _login(config) + except BydAuthenticationError as exc: + _login_failure_cache[cache_key] = {"timestamp": time.monotonic(), "message": str(exc)} + raise + _session_cache[cache_key] = session + _login_failure_cache.pop(cache_key, None) + return session + + +def _post_token_json(endpoint: str, config: BydConfiguration, session: Dict[str, Any], + inner: Dict[str, str]) -> Any: + keys = crypto.session_keys(session["sign_token"], session["encry_token"]) + device = _device(config) + imei_md5 = device["imeiMd5"] + req_timestamp = str(int(time.time() * 1000)) + + encry_data = crypto.aes_encrypt_hex(json.dumps(inner, separators=(",", ":")), keys["content_key"]) + sign_fields = { + **inner, + "countryCode": config.country_code, + "identifier": session["user_id"], + "imeiMD5": imei_md5, + "language": "en", + "reqTimestamp": req_timestamp, + } + sign = crypto.sha1_mixed(crypto.build_sign_string(sign_fields, keys["sign_key"])) + + outer = { + "countryCode": config.country_code, + "encryData": encry_data, + "identifier": session["user_id"], + "imeiMD5": imei_md5, + "language": "en", + "reqTimestamp": req_timestamp, + "sign": sign, + "ostype": device["ostype"], + "imei": device["imei"], + "mac": device["mac"], + "model": device["model"], + "sdk": device["sdk"], + "mod": device["mod"], + "serviceTime": str(int(time.time() * 1000)), + } + outer["checkcode"] = crypto.compute_checkcode(outer) + + response = _post_secure(endpoint, outer) + code = str(response.get("code", "")) + if code != "0": + if code in SESSION_EXPIRED_CODES: + raise BydAuthenticationError(f"{endpoint}: Session abgelaufen (code={code})") + if code in ENDPOINT_NOT_SUPPORTED_CODES: + raise BydApiError(f"{endpoint}: von diesem Fahrzeug nicht unterstützt (code={code})") + if code in VEHICLE_UNREACHABLE_CODES: + raise BydApiError(f"{endpoint}: Fahrzeug nicht erreichbar, z.B. im Tiefschlaf (code={code})") + raise BydApiError(f"{endpoint} fehlgeschlagen: code={code} message={response.get('message')}") + + respond_data = response.get("respondData") + if not respond_data: + return {} + plaintext = crypto.aes_decrypt_utf8(respond_data, keys["content_key"]) + if not plaintext.strip(): + return {} + return json.loads(plaintext) + + +def _inner_base(config: BydConfiguration, request_serial: Optional[str] = None) -> Dict[str, str]: + device = _device(config) + inner = { + "deviceType": device["deviceType"], + "imeiMD5": device["imeiMd5"], + "networkType": device["networkType"], + "random": secrets.token_hex(16).upper(), + "timeStamp": str(int(time.time() * 1000)), + "version": "351", + } + inner["vin"] = config.vin + if request_serial: + inner["requestSerial"] = request_serial + return inner + + +def _is_realtime_ready(data: Dict[str, Any]) -> bool: + """Trigger-Antwort ist nur eine Platzhalter-Quittung mit zB elecPercent=0 statt + fehlend, "is not None" greift daher nicht. Kriterium wie pyBYDs is_ready_raw: + offline nie bereit, sonst zählt ein echter Zeitstempel/Reifendruck/Restreichweite.""" + if not data or data.get("onlineState") == 0: + return False + tire_fields = ("leftFrontTirepressure", "rightFrontTirepressure", + "leftRearTirepressure", "rightRearTirepressure") + if any(float(data.get(f) or 0) > 0 for f in tire_fields): + return True + if int(data.get("time") or 0) > 0: + return True + return float(data.get("enduranceMileage") or 0) > 0 + + +def _fetch_realtime(config: BydConfiguration, session: Dict[str, Any]) -> Dict[str, Any]: + """Trigger + Poll: löst die Fahrzeug-Abfrage aus und pollt dann bis zu + POLL_ATTEMPTS mal im Abstand von POLL_INTERVAL_S, bis Daten vorliegen.""" + trigger_inner = _inner_base(config) + trigger_inner["energyType"] = _ENERGY_TYPE_CODES.get(config.energy_type, "0") + trigger_inner["tboxVersion"] = "3" + result = _post_token_json(REALTIME_TRIGGER_ENDPOINT, config, session, trigger_inner) + request_serial = result.get("requestSerial") if isinstance(result, dict) else None + + if isinstance(result, dict) and _is_realtime_ready(result): + return result + + for attempt in range(1, POLL_ATTEMPTS + 1): + time.sleep(POLL_INTERVAL_S) + poll_inner = _inner_base(config, request_serial=request_serial) + poll_inner["energyType"] = _ENERGY_TYPE_CODES.get(config.energy_type, "0") + poll_inner["tboxVersion"] = "3" + try: + result = _post_token_json(REALTIME_POLL_ENDPOINT, config, session, poll_inner) + except BydAuthenticationError: + # durchreichen, damit fetch_soc()s Relogin-Pfad greift statt den Poll-Loop zu verbrauchen + raise + except BydApiError as exc: + log.debug("BYD: Poll-Versuch %d/%d fehlgeschlagen: %s", attempt, POLL_ATTEMPTS, exc) + continue + if isinstance(result, dict): + request_serial = result.get("requestSerial") or request_serial + if _is_realtime_ready(result): + return result + + raise BydApiError("Keine Realtime-Daten erhalten (Fahrzeug antwortet nicht rechtzeitig, evtl. im Tiefschlaf)") + + +def extract_soc(data: Dict[str, Any]) -> float: + value = data.get("elecPercent") + if value is None or float(value) < 0: + raise BydApiError("SoC (elecPercent) nicht in der Antwort enthalten") + return float(value) + + +def extract_range(data: Dict[str, Any]) -> Optional[float]: + value = data.get("enduranceMileage") + if value is None or float(value) < 0: + return None + return float(value) + + +def extract_soc_timestamp(data: Dict[str, Any]) -> Optional[float]: + # "time" ist der Messzeitpunkt des Fahrzeugs, nicht der Abrufzeitpunkt + value = data.get("time") + if value is None: + return None + return float(value) + + +def extract_odometer(data: Dict[str, Any]) -> Optional[float]: + value = data.get("totalMileage") + if value is None or float(value) < 0: + return None + return float(value) + + +def fetch_soc(config: BydConfiguration, vehicle: int) -> CarState: + if not config.username or not config.password: + raise BydApiError("Benutzername und Passwort müssen konfiguriert sein") + if not config.vin: + raise BydApiError("VIN muss konfiguriert sein") + + try: + session = _get_session(config) + try: + data = _fetch_realtime(config, session) + except BydAuthenticationError: + log.info("BYD: Session abgelaufen, neuer Login für Fahrzeug %s", vehicle) + session = _get_session(config, force_relogin=True, rejected_session=session) + data = _fetch_realtime(config, session) + except BydApiError: + raise + except Exception as exc: + raise BydApiError(f"Unerwarteter Fehler beim Abruf der BYD-Daten: {exc}") from exc + + return CarState( + soc=extract_soc(data), + range=extract_range(data), + soc_timestamp=extract_soc_timestamp(data), + odometer=extract_odometer(data), + ) diff --git a/packages/modules/vehicles/byd/api_test.py b/packages/modules/vehicles/byd/api_test.py new file mode 100644 index 0000000000..55781b2da4 --- /dev/null +++ b/packages/modules/vehicles/byd/api_test.py @@ -0,0 +1,135 @@ +import threading +import time +from unittest.mock import Mock + +import pytest + +from modules.vehicles.byd import api +from modules.vehicles.byd.config import BydConfiguration + + +@pytest.fixture(autouse=True) +def _clear_caches(): + api._session_cache.clear() + api._login_failure_cache.clear() + api._fingerprint_cache.clear() + api._account_locks.clear() + yield + + +def _config(username="user@example.com", vin="VIN123") -> BydConfiguration: + return BydConfiguration(username=username, password="pw", vin=vin) + + +def test_fetch_soc_requires_vin(): + with pytest.raises(api.BydApiError, match="VIN"): + api.fetch_soc(_config(vin=""), 1) + + +def test_fetch_soc_requires_credentials(): + with pytest.raises(api.BydApiError, match="Benutzername"): + api.fetch_soc(BydConfiguration(username="", password="", vin="VIN123"), 1) + + +class TestResolveAccountFingerprint: + def test_same_account_reuses_fingerprint(self): + first = api.resolve_account_fingerprint("acc1", "") + second = api.resolve_account_fingerprint("acc1", "") + assert first == second + + def test_different_accounts_get_different_fingerprints(self): + first = api.resolve_account_fingerprint("acc1", "") + second = api.resolve_account_fingerprint("acc2", "") + assert first != second + + def test_adopts_already_persisted_fingerprint(self): + result = api.resolve_account_fingerprint("acc1", '{"imei": "already-persisted"}') + assert result == '{"imei": "already-persisted"}' + + +class TestGetSession: + def test_cached_session_is_reused(self, monkeypatch): + login_mock = Mock(return_value={"user_id": "u", "sign_token": "s", + "encry_token": "e", "created_at": time.monotonic()}) + monkeypatch.setattr(api, "_login", login_mock) + + api._get_session(_config()) + api._get_session(_config()) + + login_mock.assert_called_once() + + def test_failed_login_blocks_retries_during_cooldown(self, monkeypatch): + monkeypatch.setattr(api, "_login", Mock(side_effect=api.BydAuthenticationError("bad credentials"))) + + with pytest.raises(api.BydAuthenticationError, match="bad credentials"): + api._get_session(_config()) + with pytest.raises(api.BydAuthenticationError, match="wird für weitere"): + api._get_session(_config()) + + def test_concurrent_logins_for_same_account_only_login_once(self, monkeypatch): + call_count = 0 + + def slow_login(config): + nonlocal call_count + call_count += 1 + time.sleep(0.05) + return {"user_id": "u", "sign_token": "s", "encry_token": "e", "created_at": time.monotonic()} + + monkeypatch.setattr(api, "_login", slow_login) + + threads = [threading.Thread(target=api._get_session, args=(_config(),)) for _ in range(5)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert call_count == 1 + + def test_force_relogin_reuses_session_already_refreshed_by_another_vehicle(self, monkeypatch): + rejected_session = {"user_id": "u", "sign_token": "old", "encry_token": "old", + "created_at": time.monotonic()} + refreshed_session = {"user_id": "u", "sign_token": "new", "encry_token": "new", + "created_at": time.monotonic()} + api._session_cache["user@example.com"] = refreshed_session + login_mock = Mock(return_value={"user_id": "u", "sign_token": "newer", "encry_token": "newer", + "created_at": time.monotonic()}) + monkeypatch.setattr(api, "_login", login_mock) + + result = api._get_session(_config(), force_relogin=True, rejected_session=rejected_session) + + assert result is refreshed_session + login_mock.assert_not_called() + + def test_force_relogin_logs_in_when_cached_session_is_still_the_rejected_one(self, monkeypatch): + rejected_session = {"user_id": "u", "sign_token": "old", "encry_token": "old", + "created_at": time.monotonic()} + api._session_cache["user@example.com"] = rejected_session + new_session = {"user_id": "u", "sign_token": "new", "encry_token": "new", + "created_at": time.monotonic()} + monkeypatch.setattr(api, "_login", Mock(return_value=new_session)) + + result = api._get_session(_config(), force_relogin=True, rejected_session=rejected_session) + + assert result is new_session + + +class TestFetchRealtimeAuthError: + def test_auth_error_during_poll_is_not_swallowed(self, monkeypatch): + monkeypatch.setattr(api, "_inner_base", Mock(return_value={})) + monkeypatch.setattr(api, "_post_token_json", Mock(side_effect=[ + {}, # trigger response, not yet ready + api.BydAuthenticationError("session expired"), + ])) + + with pytest.raises(api.BydAuthenticationError, match="session expired"): + api._fetch_realtime(_config(), {"user_id": "u", "sign_token": "s", "encry_token": "e"}) + + def test_other_api_errors_are_retried(self, monkeypatch): + monkeypatch.setattr(api, "_inner_base", Mock(return_value={})) + responses = [{}, api.BydApiError("transient"), {"onlineState": 1, "time": 123}] + monkeypatch.setattr(api, "_post_token_json", Mock(side_effect=responses)) + monkeypatch.setattr(api.time, "sleep", Mock()) + + result = api._fetch_realtime(_config(), {"user_id": "u", "sign_token": "s", "encry_token": "e"}) + + assert result == {"onlineState": 1, "time": 123} diff --git a/packages/modules/vehicles/byd/bangcle_tables.bin b/packages/modules/vehicles/byd/bangcle_tables.bin new file mode 100644 index 0000000000..fc72997d0a Binary files /dev/null and b/packages/modules/vehicles/byd/bangcle_tables.bin differ diff --git a/packages/modules/vehicles/byd/config.py b/packages/modules/vehicles/byd/config.py new file mode 100644 index 0000000000..edabe5f8c1 --- /dev/null +++ b/packages/modules/vehicles/byd/config.py @@ -0,0 +1,36 @@ +from dataclasses import dataclass + + +@dataclass +class BydConfiguration: + username: str = "" + password: str = "" + # erforderlich, auch bei nur einem Fahrzeug am Account + vin: str = "" + # ISO-Ländercode, wie in der BYD-App unter "Land/Region" hinterlegt (z.B. "DE", "NL"). + # Bestimmt u.a. den Login-Endpunkt/die Sprache der API-Antworten. + country_code: str = "DE" + # "ev" (rein elektrisch) oder "hybrid" (PHEV). Bei Hybriden liefert die API + # bei falscher Einstellung die EV-Reichweite/Verbrauchsfelder als 0 zurück. + energy_type: str = "ev" + # Login-Feld "identifierType", "0" = E-Mail (Standard). Bei manchen Accounts liefert + # der Login trotz korrekter Zugangsdaten "Email or password is incorrect" zurück - laut + # https://github.com/jkaberg/pyBYD/issues/71 evtl. abhängig vom Account-Typ ein anderer + # Wert noetig, aber (Stand dort) noch nicht bestätigt. Vorsicht: jeder Loginversuch + # zaehlt gegen das Fehlversuch-Limit des BYD-Accounts. + identifier_type: str = "0" + # Android-Geräte-Fingerprint (JSON), einmalig generiert und dauerhaft gespeichert, damit der + # Account immer als dasselbe Gerät auftritt - siehe soc.py. Nicht vom Nutzer editierbar. + device_fingerprint: str = "" + + +@dataclass +class Byd: + name: str = "BYD" + type: str = "byd" + official: bool = False + configuration: BydConfiguration = None + + def __post_init__(self): + if self.configuration is None: + self.configuration = BydConfiguration() diff --git a/packages/modules/vehicles/byd/crypto.py b/packages/modules/vehicles/byd/crypto.py new file mode 100644 index 0000000000..80716a184b --- /dev/null +++ b/packages/modules/vehicles/byd/crypto.py @@ -0,0 +1,333 @@ +#!/usr/bin/env python3 +"""Krypto-Bausteine für die BYD-Cloud-API. + +Synchron und ohne Fremdbibliotheken (außer `cryptography`, die in openWB +bereits vorhanden ist) nach Python 3.9 portiert. Der Bangcle-Whitebox-AES-Teil +(Tabellen-Lookup statt normalem AES-Key-Schedule, wie es die offizielle BYD-App +als App-Hardening nutzt) sowie die Ableitung der Login-/Sitzungs-Keys stammen +aus dem MIT-lizenzierten Projekt https://github.com/jkaberg/pyBYD, das dieses +Schema aus der BYD-App reverse-engineered hat. Die Tabellen-Binärdatei +(bangcle_tables.bin) wurde von dort unverändert übernommen. Lizenztext siehe +THIRD_PARTY_NOTICES.md. + +Zwei unabhängige Verschlüsselungs-Layer: + 1. Bangcle (Whitebox-AES-CBC, Zero-IV): verschlüsselt das komplette äußere + JSON-Envelope für den Transport ({"request": ""}). + 2. Standard-AES-128-CBC (Zero-IV): verschlüsselt/entschlüsselt innerhalb des + Envelopes die Felder "encryData" (Request) bzw. "respondData" (Response). + Der Schlüssel dafür wird beim Login aus dem Passwort abgeleitet, danach + aus dem vom Server erhaltenen encryToken (siehe session_keys()). +""" +import hashlib +import json +import struct +from pathlib import Path +from typing import Any, Dict, NamedTuple, Optional + +_ZERO_IV = b"\x00" * 16 +_TABLES_FILE = Path(__file__).parent / "bangcle_tables.bin" + +# --- Binärformat der Tabellendatei (siehe pyBYD._crypto.bangcle) --- +_MAGIC = b"BGTB" +_VERSION = 1 +_TABLE_COUNT = 8 +_HEADER_SIZE = 8 +_INDEX_ENTRY_SIZE = 8 +_TABLE_SPECS = [ + ("inv_round", 0x28000), ("inv_xor", 0x3C000), ("inv_first", 0x1000), + ("round", 0x28000), ("xor", 0x3C000), ("final", 0x1000), + ("perm_decrypt", 8), ("perm_encrypt", 8), +] + + +class BydCryptoError(Exception): + """Fehler beim Ver-/Entschlüsseln oder beim Laden der Bangcle-Tabellen.""" + + +class BangcleTables(NamedTuple): + inv_round: bytes + inv_xor: bytes + inv_first: bytes + round: bytes + xor: bytes + final: bytes + perm_decrypt: bytes + perm_encrypt: bytes + + +_tables_cache: Optional[BangcleTables] = None + + +def _load_tables() -> BangcleTables: + global _tables_cache + if _tables_cache is not None: + return _tables_cache + try: + data = _TABLES_FILE.read_bytes() + except FileNotFoundError as exc: + raise BydCryptoError(f"bangcle_tables.bin nicht gefunden unter {_TABLES_FILE}") from exc + + if data[:4] != _MAGIC or len(data) < _HEADER_SIZE + _TABLE_COUNT * _INDEX_ENTRY_SIZE: + raise BydCryptoError("bangcle_tables.bin ist beschädigt oder hat ein unbekanntes Format") + + tables = [] + for i, (name, expected_len) in enumerate(_TABLE_SPECS): + idx_off = _HEADER_SIZE + i * _INDEX_ENTRY_SIZE + offset, length = struct.unpack_from(" None: + for col in range(4): + for row in range(4): + out[col * 8 + row] = block[col + row * 4] + + +def _decrypt_block(tables: BangcleTables, block: bytes) -> bytes: + state = bytearray(32) + temp64 = bytearray(64) + output = bytearray(16) + _prepare_matrix(block, state) + + for rnd in range(9, 0, -1): + l21 = rnd * 4 + perm_ptr = 0 + for i in range(4): + perm = tables.perm_decrypt[perm_ptr] + base = i * 16 + for j in range(4): + u = (perm + j) & 3 + byte_val = state[i * 8 + u] + idx = byte_val + (i + (l21 + u) * 4) * 256 + value = struct.unpack_from("> 4) | ((bv >> 4) << 4)) & 0xFF + idx1 = (l2 + (i25 - 1)) * 0x100 + u27 + u6 = tables.inv_xor[idx1] & 0xF + idx2 = (l2 + i25) * 0x100 + u26 + bnew = tables.inv_xor[idx2] + u26 = (bnew & 0xF) << 4 + i25 += 2 + state[l9x + l21x * 8] = (u26 | u6) & 0xFF + off += 4 + i15 += 6 + + tmp32 = bytearray(state) + u8, u10, u12 = 1, 3, 2 + for row in range(4): + state[row] = tables.inv_first[tmp32[row] + row * 0x400] + r1 = u10 & 3 + state[8 + row] = tables.inv_first[tmp32[8 + r1] + r1 * 0x400 + 0x100] + r2 = u12 & 3 + state[0x10 + row] = tables.inv_first[tmp32[0x10 + r2] + r2 * 0x400 + 0x200] + r3 = u8 & 3 + state[0x18 + row] = tables.inv_first[tmp32[0x18 + r3] + r3 * 0x400 + 0x300] + u8 += 1 + u10 += 1 + u12 += 1 + + for col in range(4): + for row in range(4): + output[col + row * 4] = state[col * 8 + row] + return bytes(output) + + +def _encrypt_block(tables: BangcleTables, block: bytes) -> bytes: + state = bytearray(32) + temp64 = bytearray(64) + output = bytearray(16) + _prepare_matrix(block, state) + + for rnd in range(9): + l21 = rnd * 4 + perm_ptr = 0 + for i in range(4): + perm = tables.perm_encrypt[perm_ptr] + base = i * 16 + for j in range(4): + u = (perm + j) & 3 + byte_val = state[i * 8 + u] + idx = byte_val + (i + (l21 + u) * 4) * 256 + value = struct.unpack_from("> 4) | ((bv >> 4) << 4)) & 0xFF + idx1 = (l2 + (i25 - 1)) * 0x100 + u27 + u7 = tables.xor[idx1] & 0xF + idx2 = (l2 + i25) * 0x100 + u26 + bnew = tables.xor[idx2] + u26 = (bnew & 0xF) << 4 + i25 += 2 + state[l10 + l22 * 8] = (u26 | u7) & 0xFF + off += 4 + i16 += 6 + + tmp32 = bytearray(state) + u13, u9, u11, u8e = 3, 2, 1, 0 + for row in range(4): + r0 = (u8e + row) & 3 + state[row] = tables.final[tmp32[r0] + r0 * 0x400] + r1 = (u11 + row) & 3 + state[8 + row] = tables.final[tmp32[8 + r1] + r1 * 0x400 + 0x100] + r2 = (u9 + row) & 3 + state[0x10 + row] = tables.final[tmp32[0x10 + r2] + r2 * 0x400 + 0x200] + r3 = (u13 + row) & 3 + state[0x18 + row] = tables.final[tmp32[0x18 + r3] + r3 * 0x400 + 0x300] + + for col in range(4): + for row in range(4): + output[col + row * 4] = state[col * 8 + row] + return bytes(output) + + +def _add_pkcs7(data: bytes, block_size: int = 16) -> bytes: + pad_len = block_size - (len(data) % block_size) + return data + bytes([pad_len] * pad_len) + + +def _strip_pkcs7(data: bytes) -> bytes: + if not data: + return data + pad = data[-1] + if 0 < pad <= 16 and len(data) >= pad and all(b == pad for b in data[-pad:]): + return data[:-pad] + return data + + +def bangcle_encode(plaintext: str) -> str: + """Verschlüsselt den äußeren Envelope für den Transport ('F' + base64).""" + import base64 + tables = _load_tables() + padded = _add_pkcs7(plaintext.encode("utf-8")) + prev = bytearray(_ZERO_IV) + out = bytearray(len(padded)) + for off in range(0, len(padded), 16): + block = bytearray(padded[off:off + 16]) + for i in range(16): + block[i] ^= prev[i] + enc = _encrypt_block(tables, bytes(block)) + out[off:off + 16] = enc + prev[:] = enc + return "F" + base64.b64encode(bytes(out)).decode("ascii") + + +def bangcle_decode(envelope: str) -> bytes: + """Entschlüsselt eine Bangcle-Antwort zurück zu den rohen Bytes.""" + import base64 + tables = _load_tables() + cleaned = envelope.strip().replace("-", "+").replace("_", "/") + if cleaned.startswith("F"): + cleaned = cleaned[1:] + cleaned += "=" * (-len(cleaned) % 4) + ciphertext = base64.b64decode(cleaned) + if not ciphertext or len(ciphertext) % 16 != 0: + raise BydCryptoError(f"Bangcle-Ciphertext hat ungültige Länge {len(ciphertext)}") + + prev = bytearray(_ZERO_IV) + out = bytearray(len(ciphertext)) + for off in range(0, len(ciphertext), 16): + block = ciphertext[off:off + 16] + dec = bytearray(_decrypt_block(tables, block)) + for i in range(16): + dec[i] ^= prev[i] + out[off:off + 16] = dec + prev[:] = block + return _strip_pkcs7(bytes(out)) + + +# --- Standard-AES-128-CBC (Zero-IV) für encryData/respondData --- + +def aes_encrypt_hex(plaintext: str, key_hex: str) -> str: + from cryptography.hazmat.primitives import padding + from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + key = bytes.fromhex(key_hex) + padder = padding.PKCS7(128).padder() + padded = padder.update(plaintext.encode("utf-8")) + padder.finalize() + encryptor = Cipher(algorithms.AES(key), modes.CBC(_ZERO_IV)).encryptor() + return (encryptor.update(padded) + encryptor.finalize()).hex().upper() + + +def aes_decrypt_utf8(cipher_hex: str, key_hex: str) -> str: + from cryptography.hazmat.primitives import padding + from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + key = bytes.fromhex(key_hex) + ct = bytes.fromhex(cipher_hex) + decryptor = Cipher(algorithms.AES(key), modes.CBC(_ZERO_IV)).decryptor() + padded = decryptor.update(ct) + decryptor.finalize() + unpadder = padding.PKCS7(128).unpadder() + return (unpadder.update(padded) + unpadder.finalize()).decode("utf-8") + + +# --- Hashing / Signing --- + +def md5_hex(value: str) -> str: + return hashlib.md5(value.encode("utf-8")).hexdigest().upper() + + +def pwd_login_key(password: str) -> str: + """Login-AES-Key wird aus dem doppelt-MD5-gehashten Passwort abgeleitet.""" + return md5_hex(md5_hex(password)) + + +def sha1_mixed(value: str) -> str: + """SHA1 mit alternierender Groß-/Kleinschreibung + Entfernen führender Nullen + an geraden Positionen (Eigenheit des BYD-Signaturschemas).""" + digest = hashlib.sha1(value.encode("utf-8")).digest() + mixed = "".join( + f"{b:02x}".upper() if i % 2 == 0 else f"{b:02x}" + for i, b in enumerate(digest) + ) + return "".join(ch for j, ch in enumerate(mixed) if not (ch == "0" and j % 2 == 0)) + + +def build_sign_string(fields: Dict[str, str], password_or_key: str) -> str: + keys = sorted(fields.keys()) + joined = "&".join(f"{k}={'null' if fields[k] is None else fields[k]}" for k in keys) + return f"{joined}&password={password_or_key}" + + +def compute_checkcode(payload: Dict[str, Any]) -> str: + json_str = json.dumps(payload, separators=(",", ":"), ensure_ascii=False) + md5 = hashlib.md5(json_str.encode("utf-8")).hexdigest() + return md5[24:32] + md5[8:16] + md5[16:24] + md5[0:8] + + +def session_keys(sign_token: str, encry_token: str) -> Dict[str, str]: + """Leitet die beiden Sitzungs-Keys aus den beim Login erhaltenen Tokens ab.""" + return {"sign_key": md5_hex(sign_token), "content_key": md5_hex(encry_token)} diff --git a/packages/modules/vehicles/byd/device_fingerprint.py b/packages/modules/vehicles/byd/device_fingerprint.py new file mode 100644 index 0000000000..bc1925b989 --- /dev/null +++ b/packages/modules/vehicles/byd/device_fingerprint.py @@ -0,0 +1,76 @@ +#!/usr/bin/env python3 +"""Zufälliger, aber plausibler Android-Geräte-Fingerprint fürs BYD-Login. + +Jeder Account braucht eine eigene, stabile Geräte-Identität - sonst sehen viele +openWB-Nutzer aus BYDs Sicht wie dasselbe Gerät aus, das sich bei vielen Accounts +einloggt (Bot-Signal). Pool/Logik (gültige IMEI per Luhn, zufällige MAC) von +https://github.com/TA2k/ioBroker.byd (MIT), Lizenztext siehe THIRD_PARTY_NOTICES.md. +""" +import hashlib +import secrets +from typing import Any, Dict, List + +DEVICE_POOL: List[Dict[str, Any]] = [ + {"model": "SM-S928B", "mod": "samsung", "mobileBrand": "samsung", "mobileModel": "Galaxy S24 Ultra", + "tac_prefix": "35847512", "sdk_options": ["34", "35"], "os_options": ["14", "15"]}, + {"model": "SM-S911B", "mod": "samsung", "mobileBrand": "samsung", "mobileModel": "Galaxy S23", + "tac_prefix": "35294712", "sdk_options": ["33", "34", "35"], "os_options": ["13", "14", "15"]}, + {"model": "Pixel 8 Pro", "mod": "Google", "mobileBrand": "Google", "mobileModel": "Pixel 8 Pro", + "tac_prefix": "35396012", "sdk_options": ["34", "35"], "os_options": ["14", "15"]}, + {"model": "Pixel 7", "mod": "Google", "mobileBrand": "Google", "mobileModel": "Pixel 7", + "tac_prefix": "35396011", "sdk_options": ["33", "34", "35"], "os_options": ["13", "14", "15"]}, + {"model": "23127PN0CC", "mod": "Xiaomi", "mobileBrand": "Xiaomi", "mobileModel": "14", + "tac_prefix": "86758004", "sdk_options": ["34", "35"], "os_options": ["14", "15"]}, + {"model": "CPH2451", "mod": "OnePlus", "mobileBrand": "OnePlus", "mobileModel": "11", + "tac_prefix": "86543903", "sdk_options": ["33", "34"], "os_options": ["13", "14"]}, + {"model": "VOG-L29", "mod": "HUAWEI", "mobileBrand": "HUAWEI", "mobileModel": "P30 Pro", + "tac_prefix": "86069604", "sdk_options": ["31", "32"], "os_options": ["12", "12"]}, +] + + +def _luhn_check_digit(partial: str) -> str: + total = 0 + for i, ch in enumerate(partial): + digit = int(ch) + if i % 2 == 1: + digit *= 2 + if digit > 9: + digit -= 9 + total += digit + return str((10 - total % 10) % 10) + + +def _generate_imei(tac_prefix: str) -> str: + serial = "".join(str(secrets.randbelow(10)) for _ in range(6)) + partial = tac_prefix + serial + return partial + _luhn_check_digit(partial) + + +def _generate_mac() -> str: + first_byte = (secrets.randbelow(256) | 0x02) & 0xFE # locally administered, unicast + octets = [first_byte] + [secrets.randbelow(256) for _ in range(5)] + return ":".join(f"{b:02x}" for b in octets) + + +def generate() -> Dict[str, str]: + device = DEVICE_POOL[secrets.randbelow(len(DEVICE_POOL))] + idx = secrets.randbelow(len(device["sdk_options"])) + sdk = device["sdk_options"][idx] + os_type = device["os_options"][idx] + imei = _generate_imei(device["tac_prefix"]) + + return { + "ostype": "and", + "imei": imei, + "imeiMd5": hashlib.md5(imei.encode("utf-8")).hexdigest(), + "mac": _generate_mac(), + "model": device["model"], + "mod": device["mod"], + "mobileBrand": device["mobileBrand"], + "mobileModel": device["mobileModel"], + "deviceType": "0", + "networkType": "wifi", + "osType": os_type, + "osVersion": sdk, + "sdk": sdk, + } diff --git a/packages/modules/vehicles/byd/soc.py b/packages/modules/vehicles/byd/soc.py new file mode 100644 index 0000000000..db6934c158 --- /dev/null +++ b/packages/modules/vehicles/byd/soc.py @@ -0,0 +1,52 @@ +import json +import logging + +from dataclass_utils import asdict +from helpermodules.pub import Pub +from modules.common.abstract_device import DeviceDescriptor +from modules.common.abstract_vehicle import VehicleUpdateData +from modules.common.component_state import CarState +from modules.common.configurable_vehicle import ConfigurableVehicle +from modules.vehicles.byd import api +from modules.vehicles.byd.config import Byd + +log = logging.getLogger(__name__) + + +def fetch(vehicle_config: Byd, vehicle: int) -> CarState: + config = vehicle_config.configuration + + # Fingerprint ist account-, nicht fahrzeugbezogen - dauerhaft in der Fahrzeug-Config gespeichert + shared_fingerprint = api.resolve_account_fingerprint(config.username, config.device_fingerprint) + if config.device_fingerprint != shared_fingerprint: + is_new = not config.device_fingerprint + config.device_fingerprint = shared_fingerprint + if is_new: + fingerprint = json.loads(shared_fingerprint) + log.info("BYD: neuer Geräte-Fingerprint für Account %s generiert: %s %s (IMEI %s)", + config.username, fingerprint["mobileBrand"], fingerprint["mobileModel"], + fingerprint["imei"]) + Pub().pub(f"openWB/set/vehicle/{vehicle}/soc_module/config", asdict(vehicle_config)) + + return api.fetch_soc(config, vehicle) + + +def create_vehicle(vehicle_config: Byd, vehicle: int): + config = vehicle_config.configuration + if config.device_fingerprint: + # Account-Cache vor dem ersten Poll fuellen, damit die Reihenfolge mehrerer Fahrzeuge + # desselben Accounts keine Rolle spielt + api.resolve_account_fingerprint(config.username, config.device_fingerprint) + + def updater(vehicle_update_data: VehicleUpdateData) -> CarState: + return fetch(vehicle_config, vehicle) + + return ConfigurableVehicle(vehicle_config=vehicle_config, + component_updater=updater, + vehicle=vehicle, + # SoC während der Ladung über die Cloud-API abrufbar, + # daher keine manuelle Berechnung nötig + calc_while_charging=False) + + +device_descriptor = DeviceDescriptor(configuration_factory=Byd) diff --git a/packages/modules/vehicles/byd/soc_test.py b/packages/modules/vehicles/byd/soc_test.py new file mode 100644 index 0000000000..e9cf600c05 --- /dev/null +++ b/packages/modules/vehicles/byd/soc_test.py @@ -0,0 +1,27 @@ +import pytest + +from modules.vehicles.byd import api, soc +from modules.vehicles.byd.config import Byd, BydConfiguration + + +@pytest.fixture(autouse=True) +def _clear_caches(): + api._fingerprint_cache.clear() + yield + + +def test_create_vehicle_seeds_account_cache_with_persisted_fingerprint(): + config = Byd(configuration=BydConfiguration(username="user@example.com", vin="VIN1", + device_fingerprint='{"imei": "already-persisted"}')) + + soc.create_vehicle(config, 1) + + assert api._fingerprint_cache["user@example.com"] == '{"imei": "already-persisted"}' + + +def test_create_vehicle_does_not_seed_cache_without_persisted_fingerprint(): + config = Byd(configuration=BydConfiguration(username="user@example.com", vin="VIN1")) + + soc.create_vehicle(config, 1) + + assert "user@example.com" not in api._fingerprint_cache