Skip to content

feat: multi-region support (fixes 1501 "geelyos verify error" outside the EU) #3

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
alexandrepossebom wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat: multi-region support (fixes 1501 "geelyos verify error" outside the EU) #3

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 11 additions & 8 deletions custom_components/geely_global/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,21 +17,22 @@
from . import api as geely_api
from .api import GeelyApi, GeelyAuthError
from .const import (
APP_ID,
APP_SECRET,
CLIENT_ID,
CONF_CERT_PATH,
CONF_CIDPSSO_TOKEN,
CONF_DEVICE_ID,
CONF_KEY_PATH,
CONF_REGION,
CONF_USER_ID,
CONF_VEHICLE_MODEL_CODE,
CONF_VEHICLE_NICKNAME,
CONF_VEHICLE_SERIES,
CONF_VIN,
DEFAULT_REGION,
DOMAIN,
SCAN_INTERVAL_SECONDS,
SERIES_TO_FRIENDLY_NAME,
region_config,
)

_LOGGER = logging.getLogger(__name__)
Expand Down Expand Up @@ -133,12 +134,15 @@ async def _maybe_refetch_vehicle_metadata(hass: HomeAssistant, entry: ConfigEntr
have_series = entry.data.get(CONF_VEHICLE_SERIES) or entry.data.get(CONF_VEHICLE_MODEL_CODE)
if have and have_series:
return
app_host = region_config(entry.data.get(CONF_REGION) or DEFAULT_REGION)["app_host"]
try:
all_v = await hass.async_add_executor_job(
geely_api.list_vehicles,
entry.data.get(CONF_CIDPSSO_TOKEN),
entry.data.get(CONF_USER_ID),
entry.data.get("country_code", "IL"),
lambda: geely_api.list_vehicles(
entry.data.get(CONF_CIDPSSO_TOKEN),
entry.data.get(CONF_USER_ID),
entry.data.get("country_code", "IL"),
app_host=app_host,
)
)
except Exception as e: # noqa: BLE001
_LOGGER.debug("metadata refetch failed (non-fatal): %s", e)
Expand Down Expand Up @@ -203,8 +207,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
or "E245-J1"
)
api = GeelyApi(
app_id=APP_ID,
app_secret=APP_SECRET,
region=d.get(CONF_REGION) or DEFAULT_REGION,
user_id=d[CONF_USER_ID],
vin=d[CONF_VIN],
cidpsso_token=d[CONF_CIDPSSO_TOKEN],
Expand Down
95 changes: 64 additions & 31 deletions custom_components/geely_global/api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,9 +21,35 @@
from typing import Any
from urllib.parse import parse_qsl, quote, urlparse

from .const import DEFAULT_REGION, region_config

_LOGGER = logging.getLogger(__name__)


def region_from_login(login_data: dict, vin: str | None = None) -> str:
"""Derive the vehicle's telematics region from a cidpsso login response.

Order of preference:
1. the `tspInfo` entry whose `vin` matches (per-vehicle region),
2. any `tspInfo[].serviceRegion`,
3. `edgeInfo.code` (the account's primary vehicle region),
4. DEFAULT_REGION.

Deliberately ignores `masterInfo.code`, which is the *identity* center
(e.g. EU) and can differ from where the vehicle actually lives.
"""
tsp = login_data.get("tspInfo") or []
if vin:
for t in tsp:
if t.get("vin") == vin and t.get("serviceRegion"):
return t["serviceRegion"]
for t in tsp:
if t.get("serviceRegion"):
return t["serviceRegion"]
edge = login_data.get("edgeInfo") or {}
return edge.get("code") or DEFAULT_REGION


class GeelyAuthError(Exception):
"""Raised when the Geely server rejects our credentials.

Expand Down Expand Up @@ -159,8 +185,7 @@ class GeelyApi:
def __init__(
self,
*,
app_id: str,
app_secret: str,
region: str,
user_id: str,
vin: str,
cidpsso_token: str,
Expand All @@ -171,8 +196,13 @@ def __init__(
cert_path: str,
key_path: str,
) -> None:
self.app_id = app_id
self.app_secret = app_secret
cfg = region_config(region)
self.region = region
self.app_id = cfg["app_id"]
self.app_secret = cfg["app_secret"]
self.cert_host = cfg["cert_host"]
self.control_host = cfg["control_host"]
self.app_host = cfg["app_host"]
self.user_id = user_id
self.vin = vin
self.cidpsso_token = cidpsso_token
Expand Down Expand Up @@ -271,7 +301,7 @@ def _get_access_code(self) -> str:
ctx = _legacy_ctx()
body = json.dumps({"state": str(uuid.uuid4())}).encode()
req = urllib.request.Request(
"https://m-lcmsam-eu.geely.com/cidpsso/oauth2/v1/getCode",
f"https://{self.app_host}/cidpsso/oauth2/v1/getCode",
data=body, method="POST",
headers={
"token": self.cidpsso_token,
Expand All @@ -292,7 +322,7 @@ def refresh_jwt(self) -> dict:
ac = self._get_access_code()
body = json.dumps({"authCode": ac}).encode()
status, resp = self._mtls_send(
"apis.ecloudeu.com", "POST",
self.control_host, "POST",
"/auth/account/session/secure?identity_type=geelyos",
body,
)
Expand Down Expand Up @@ -329,7 +359,7 @@ def _authed_apis_call(self, method: str, path: str, body: bytes) -> dict:
in), auto-refresh the JWT once and retry. Only escalates to
GeelyAuthError when the cidpsso token itself has been revoked."""
status, resp = self._mtls_send(
"apis.ecloudeu.com", method, path, body,
self.control_host, method, path, body,
extra_headers=self._headers_with_jwt(),
)
j = json.loads(resp)
Expand All @@ -344,7 +374,7 @@ def _authed_apis_call(self, method: str, path: str, body: bytes) -> dict:
# cidpsso token also dead - needs reauth
raise
status, resp = self._mtls_send(
"apis.ecloudeu.com", method, path, body,
self.control_host, method, path, body,
extra_headers=self._headers_with_jwt(),
)
j = json.loads(resp)
Expand Down Expand Up @@ -383,23 +413,23 @@ def request_position_refresh(self) -> dict:
}
path = f"/remote-control/vehicle/telematics/{self.vin}"
status, resp = self._mtls_send(
"apis.ecloudeu.com", "PUT", path, json.dumps(body).encode(),
self.control_host, "PUT", path, json.dumps(body).encode(),
extra_headers=self._headers_with_jwt(),
)
return json.loads(resp)

def vehicle_status_state(self) -> dict:
path = f"/remote-control/vehicle/status/state/{self.vin}"
status, resp = self._mtls_send(
"apis.ecloudeu.com", "GET", path, b"",
self.control_host, "GET", path, b"",
extra_headers=self._headers_with_jwt(),
)
return json.loads(resp)

def charging_reservation(self) -> dict:
path = f"/remote-control/charging/reservation/{self.vin}"
status, resp = self._mtls_send(
"apis.ecloudeu.com", "GET", path, b"",
self.control_host, "GET", path, b"",
extra_headers=self._headers_with_jwt(),
)
return json.loads(resp)
Expand All @@ -415,7 +445,7 @@ def charge_server_get(self, biz_type: str) -> dict:
"""
path = f"/charge-server/ecarx_charge_set/{self.vin}?bizType={biz_type}"
status, resp = self._mtls_send(
"apis.ecloudeu.com", "GET", path, b"",
self.control_host, "GET", path, b"",
extra_headers=self._headers_with_jwt(),
)
return json.loads(resp)
Expand Down Expand Up @@ -454,7 +484,7 @@ def scheduled_charging_set(self, *, command: str, start_time: str,
body_bytes = json.dumps(body, separators=(",", ":")).encode()
path = f"/charge-server/ecarx_charge_set/{self.vin}"
status, resp = self._mtls_send(
"apis.ecloudeu.com", "POST", path, body_bytes,
self.control_host, "POST", path, body_bytes,
extra_headers=self._headers_with_jwt(),
)
j = json.loads(resp)
Expand Down Expand Up @@ -484,7 +514,7 @@ def control(self, service_id: str, parameters: list[dict] | None = None,
body = json.dumps(body_dict, separators=(",", ":")).encode()
path = f"/remote-control/vehicle/telematics/{self.vin}"
status, resp = self._mtls_send(
"apis.ecloudeu.com", "PUT", path, body,
self.control_host, "PUT", path, body,
extra_headers=self._headers_with_jwt(),
)
j = json.loads(resp)
Expand Down Expand Up @@ -523,7 +553,7 @@ def rapid_climate(self, *, ac: bool, temp: str, heat_seats: list[str] | None,
body_bytes = json.dumps(body, separators=(",", ":")).encode()
path = f"/charge-server/ecarx_charge_set/{self.vin}"
status, resp = self._mtls_send(
"apis.ecloudeu.com", "POST", path, body_bytes,
self.control_host, "POST", path, body_bytes,
extra_headers=self._headers_with_jwt(),
)
j = json.loads(resp)
Expand All @@ -546,7 +576,7 @@ def fetch_capabilities(self) -> list[dict]:
"?pageSize=2000&pageIndex=1&vehicleType=0&sortField=&direction="
)
status, resp = self._mtls_send(
"apis.ecloudeu.com", "GET", path, b"",
self.control_host, "GET", path, b"",
extra_headers=self._headers_with_jwt(),
)
try:
Expand All @@ -558,9 +588,9 @@ def fetch_capabilities(self) -> list[dict]:

# ---------- Cert provisioning (one-time during config_flow) ----------

def _sign_request_for_api_ecloudeu(app_id: str, app_secret: str,
def _sign_cert_request(app_id: str, app_secret: str,
method: str, url: str, body: bytes) -> dict:
"""Standalone signer for /auth/cert/* on api.ecloudeu.com (no mTLS)."""
"""Standalone signer for /auth/cert/* (single-auth, no mTLS)."""
p = urlparse(url)
nonce = _make_nonce()
ts_ms = int(time.time() * 1000)
Expand Down Expand Up @@ -588,8 +618,8 @@ def _sign_request_for_api_ecloudeu(app_id: str, app_secret: str,
}


def provision_user_cert(*, app_id: str, app_secret: str, user_id: str,
cidpsso_token: str, cert_out_path: str,
def provision_user_cert(*, app_id: str, app_secret: str, cert_host: str,
user_id: str, cidpsso_token: str, cert_out_path: str,
key_out_path: str) -> tuple[str, str]:
"""Generate EC P-256 keypair + CSR, send through /auth/cert/info + /file,
save signed cert + key. Returns (cert_path, key_path)."""
Expand Down Expand Up @@ -618,12 +648,12 @@ def provision_user_cert(*, app_id: str, app_secret: str, user_id: str,

# 2. POST /auth/cert/info → checkCode
body = json.dumps({"checkValue": user_id}, separators=(',', ':')).encode()
headers = _sign_request_for_api_ecloudeu(
headers = _sign_cert_request(
app_id, app_secret, "POST",
"https://api.ecloudeu.com/auth/cert/info", body)
f"https://{cert_host}/auth/cert/info", body)
ctx = _legacy_ctx()
req = urllib.request.Request(
"https://api.ecloudeu.com/auth/cert/info",
f"https://{cert_host}/auth/cert/info",
data=body, method="POST", headers=headers)
with urllib.request.urlopen(req, context=ctx, timeout=20) as resp:
j = json.loads(resp.read())
Expand All @@ -632,19 +662,19 @@ def provision_user_cert(*, app_id: str, app_secret: str, user_id: str,
check_code = j["data"]["checkCode"]

# 3. POST /auth/cert/file → signed cert
device_for_cert = hashlib.sha256(f"{user_id}_geely_ex5_ha".encode()).hexdigest()
device_for_cert = hashlib.sha256(f"{user_id}_home_assistant_ha".encode()).hexdigest()
body = json.dumps({
"csr": csr_pem,
"identityType": "geelyos",
"accessToken": cidpsso_token,
"deviceId": device_for_cert,
"checkCode": check_code,
}, separators=(',', ':')).encode()
headers = _sign_request_for_api_ecloudeu(
headers = _sign_cert_request(
app_id, app_secret, "POST",
"https://api.ecloudeu.com/auth/cert/file", body)
f"https://{cert_host}/auth/cert/file", body)
req = urllib.request.Request(
"https://api.ecloudeu.com/auth/cert/file",
f"https://{cert_host}/auth/cert/file",
data=body, method="POST", headers=headers)
with urllib.request.urlopen(req, context=ctx, timeout=30) as resp:
j = json.loads(resp.read())
Expand All @@ -665,8 +695,9 @@ def provision_user_cert(*, app_id: str, app_secret: str, user_id: str,

# ---------- cidpsso login (for config_flow) ----------

# The login/OTP host is global and region-independent. The per-region app_host
# (cidpcar vehicle list) is resolved from the region config instead.
LOGIN_HOST = "https://access-app-global.geely.com"
APP_HOST = "https://m-lcmsam-eu.geely.com"


def _ios_headers(token: str | None = None, user_id: str | None = None,
Expand Down Expand Up @@ -803,12 +834,14 @@ def cidpsso_login(email: str, otp: str, country_code: str = "IL", *,


def list_vehicles(cidpsso_token: str, user_id: str | None = None,
country_code: str = "IL", *,
country_code: str = "IL", *, app_host: str | None = None,
idfa: str | None = None, idfv: str | None = None) -> list[dict]:
"""List vehicles for the logged-in account. Returns the `data` list
from /cidpcar/vehicleOwner/v2/controlCars."""
from /cidpcar/vehicleOwner/v2/controlCars. `app_host` is the region's
cidpcar host; falls back to the default region when omitted."""
host = app_host or region_config(DEFAULT_REGION)["app_host"]
s = _legacy_session()
r = s.get(f"{APP_HOST}/cidpcar/vehicleOwner/v2/controlCars",
r = s.get(f"https://{host}/cidpcar/vehicleOwner/v2/controlCars",
headers=_ios_headers(token=cidpsso_token, user_id=user_id,
country_code=country_code,
idfa=idfa, idfv=idfv),
Expand Down
Loading