From b425d40dd092daa3ff2de68a710d5baaea3c4ae9 Mon Sep 17 00:00:00 2001 From: seaspotter Date: Tue, 13 Oct 2026 12:06:46 +0200 Subject: [PATCH] fix(vehicles/kia): port OneApp/CCI login flow to bypass IDPConnect WAF block --- packages/modules/vehicles/kia/api.py | 271 +++++++++++++++------- packages/modules/vehicles/kia/api_test.py | 141 +++++++++++ 2 files changed, 329 insertions(+), 83 deletions(-) diff --git a/packages/modules/vehicles/kia/api.py b/packages/modules/vehicles/kia/api.py index 016fbe1d35..a552e940f4 100644 --- a/packages/modules/vehicles/kia/api.py +++ b/packages/modules/vehicles/kia/api.py @@ -1,6 +1,7 @@ #!/usr/bin/env python3 import base64 +from datetime import datetime import json import uuid import hashlib @@ -89,6 +90,20 @@ def getString(param_id: str, brand: str) -> str: elif param_id == "redirect_uri": paramStr = getString("base_url", brand) + \ "/api/v1/user/oauth2/redirect" + elif param_id == "oneapp_client_id": + paramStr = "01b36c86-79e8-486c-8009-15f2ad88d670" + elif param_id == "oneapp_redirect_uri": + paramStr = "https://oneapp.kia.com/redirect" + elif param_id == "cci_api_url": + paramStr = "https://cci-api-eu.kia.com" + elif param_id == "cci_package_id": + paramStr = "com.kia.oneapp.eu" + elif param_id == "cci_client_name": + paramStr = "kia" + elif param_id == "cci_client_os_version": + paramStr = "27" + elif param_id == "cci_notification_provider": + paramStr = "IOS_APPSTORE" else: raise RuntimeError @@ -122,6 +137,20 @@ def getString(param_id: str, brand: str) -> str: elif param_id == "redirect_uri": paramStr = getString("base_url", brand) + \ "/api/v1/user/oauth2/token" + elif param_id == "oneapp_client_id": + paramStr = "4f4953b5-02e1-4dbc-8599-87e983ee1be5" + elif param_id == "oneapp_redirect_uri": + paramStr = "https://oneapp.hyundai.com/redirect" + elif param_id == "cci_api_url": + paramStr = "https://cci-api-eu.hyundai.com" + elif param_id == "cci_package_id": + paramStr = "com.hyundai.oneapp.eu" + elif param_id == "cci_client_name": + paramStr = "hyundai" + elif param_id == "cci_client_os_version": + paramStr = "18.7" + elif param_id == "cci_notification_provider": + paramStr = "APNS" else: raise RuntimeError @@ -365,7 +394,14 @@ def loadToken(user_id: str, password: str, vehicle: int) -> dict: "gcmVehicleId": "", "refreshToken": "", "gcmClientId": "", - "tokenType": "" + "tokenType": "", + "cciAccessToken": "", + "cciRefreshToken": "", + "nonCcsToken": "", + "nonCcsRefreshToken": "", + "exchangeableAccessToken": "", + "exchangeableRefreshToken": "", + "idToken": "", } except Exception: log.exception("kia.loadToken: token error") @@ -505,17 +541,18 @@ def getDeviceId(brand: str) -> dict: def getAuthCode(username: str, password: str, brand: str, cookies: dict) -> str: - log.info("Kia/Hyundai: Sending username/password (IDPConnect flow)") - - # The old Keycloak form-login (eu-account..com/auth/realms/...) - # was decommissioned by Hyundai/Kia. This uses the new IDPConnect - # OAuth2 flow with an RSA-encrypted password, mirroring the - # hyundai_kia_connect_api (KiaUvoApiEU) library used by Home Assistant. - # A dedicated requests.Session keeps the cookies across the redirect - # chain; the 'cookies' argument is kept only for call-site compatibility. + log.info("Kia/Hyundai: Sending username/password (OneApp/CCI flow)") + + # Die alte IDPConnect-Anmeldung über den "legacy" client_id wird von + # Kia/Hyundais WAF inzwischen serverseitig als "abusing request" blockiert + # (betrifft alle Nutzer dieses client_id, kein individuelles Account- + # Problem). Der OneApp-client_id umgeht das, siehe hyundai_kia_connect_api + # Issue #1273. Der resultierende Auth-Code wird anschließend über + # exchangeCciToken()/exchangeCcsToken() in einen auf den bestehenden + # ccapi:8080-Endpunkten gültigen Token getauscht. host = getString("login_form_host", brand) - client_id = getString("client_id", brand) - redirect_uri = getString("redirect_uri", brand) + client_id = getString("oneapp_client_id", brand) + redirect_uri = getString("oneapp_redirect_uri", brand) # The '_CCS_APP_AOS' suffix is REQUIRED — without it the authorize # endpoint answers "400 Bad Request". @@ -533,7 +570,12 @@ def getAuthCode(username: str, password: str, brand: str, '?response_type=code&client_id=' + client_id + '&redirect_uri=' + urlparse.quote(redirect_uri, safe='') + '&lang=en&state=ccsp&country=de') - session.get(auth_url, timeout=30, allow_redirects=True) + auth_resp = session.get(auth_url, timeout=30, allow_redirects=True) + if ('abusing' in auth_resp.text.lower() or + '/error?status=400' in auth_resp.url): + log.error("kia.getAuthCode: WAF hat die Anmeldung blockiert " + "('abusing request') - kein Account-Problem") + raise RuntimeError # Step 2: fetch the RSA public key (JWK) for password encryption certs = session.get(host + '/auth/api/v1/accounts/certs', timeout=30) @@ -588,40 +630,129 @@ def getAuthCode(username: str, password: str, brand: str, return auth_code +# ---------- OneApp/CCI token exchange ---------- + + +def getCciHeaders(brand: str, device_id: str, cci_access_token: str = "", + non_ccs_token: str = "", exchangeable_token: str = "", + content_type: str = "") -> dict: + offset = datetime.now().astimezone().strftime("%z") # e.g. +0200 + headers = { + 'client-id': getString("cci_package_id", brand), + 'client-name': getString("cci_client_name", brand), + 'client-version': '1.3.3', + 'client-os-code': 'ios', + 'client-os-version': getString("cci_client_os_version", brand), + 'client-device-id': device_id, + 'client-device-model': 'iPhone', + 'client-notification-provider-type': getString("cci_notification_provider", brand), + 'locale': 'DE', + 'Accept': 'application/json', + 'Accept-Language': 'de', + 'User-Agent': 'okhttp/3.12.12', + 'timezone': (offset[:3] + ':' + offset[3:]) if offset else '+00:00', + } + if non_ccs_token: + headers['Authentication'] = non_ccs_token + if cci_access_token: + headers['authorization'] = 'Bearer ' + cci_access_token.removeprefix('Bearer ').strip() + if exchangeable_token: + headers['exchangeable-token'] = exchangeable_token + headers['non-ccs-token'] = non_ccs_token + if content_type: + headers['Content-Type'] = content_type + else: + headers['Content-Length'] = '0' + return headers -def getAuthToken(auth_code: str, token: dict, brand: str) -> dict: - log.info("Kia/Hyundai: Requesting access token") - - response = "" - try: - # New IDPConnect token endpoint; requires client_secret instead - # of the old Basic-Auth 'basic_token' header. - url = getString("login_form_host", brand) + \ - '/auth/api/v2/user/oauth2/token' - data = urlparse.urlencode({ - 'grant_type': 'authorization_code', - 'code': auth_code, - 'redirect_uri': getString("redirect_uri", brand), - 'client_id': getString("client_id", brand), - 'client_secret': getString("client_secret", brand), - }) - headers = { - 'Content-type': 'application/x-www-form-urlencoded', - 'User-Agent': 'okhttp/3.12.12', - } - response = postHTTP(url=url, headers=headers, data=data) - - access_token = json.loads(response) - token["tokenType"] = access_token["token_type"] - token["accessToken"] = access_token["access_token"] - token["refreshToken"] = access_token["refresh_token"] - - except Exception: - log.exception("kia.getAuthToken: Login failed: " + response) - raise - log.debug("Kia/Hyundai: AuthToken = " + token["accessToken"][:8] + "[...]") +def exchangeCciToken(auth_code: str, device_id: str, brand: str) -> dict: + log.info("Kia/Hyundai: Exchanging OneApp auth code for CCI token") + cci_domain_api_url = getString("cci_api_url", brand) + '/domain/api/' + headers = getCciHeaders(brand, device_id) + response = requests.post(cci_domain_api_url + 'v1/auth/token', + params={'code': auth_code}, headers=headers, + timeout=30) + if response.status_code != 200: + log.error("kia.exchangeCciToken: CCI-Token-Austausch fehlgeschlagen, StatusCode: " + + str(response.status_code) + ': ' + response.text[:200]) + raise RuntimeError + cci = response.json() + return { + 'cciAccessToken': cci.get('accessToken', ''), + 'cciRefreshToken': cci.get('refreshToken', ''), + 'nonCcsToken': cci.get('nonCcsToken', ''), + 'exchangeableAccessToken': cci.get('exchangeableAccessToken', ''), + 'exchangeableRefreshToken': cci.get('exchangeableRefreshToken', ''), + 'nonCcsRefreshToken': cci.get('nonCcsRefreshToken', ''), + 'idToken': cci.get('idToken', ''), + } + + +def exchangeCcsToken(device_id: str, cci_access_token: str, non_ccs_token: str, + exchangeable_token: str, brand: str) -> str: + log.info("Kia/Hyundai: Exchanging CCI token for CCS token") + + cci_domain_api_url = getString("cci_api_url", brand) + '/domain/api/' + headers = getCciHeaders(brand, device_id, cci_access_token=cci_access_token, + non_ccs_token=non_ccs_token, + exchangeable_token=exchangeable_token) + response = requests.post(cci_domain_api_url + 'v1/auth/token-exchange', + params={'serviceType': 'CCS'}, headers=headers, + timeout=30) + if response.status_code != 200: + log.error("kia.exchangeCcsToken: CCS-Token-Austausch fehlgeschlagen, StatusCode: " + + str(response.status_code) + ': ' + response.text[:200]) + raise RuntimeError + data = response.json() + ccs_token = data.get('accessToken') or data.get('ccsAccessToken') or '' + if not ccs_token: + log.error("kia.exchangeCcsToken: kein accessToken in Antwort: " + response.text[:200]) + raise RuntimeError + return ccs_token + + +def refreshCciToken(token: dict, brand: str) -> dict: + log.info("Kia/Hyundai: CCI token refresh starting") + + device_id = token.get('deviceId', '') + cci_domain_api_url = getString("cci_api_url", brand) + '/domain/api/' + headers = getCciHeaders(brand, device_id, + cci_access_token=token.get('cciAccessToken', ''), + non_ccs_token=token.get('nonCcsToken', ''), + exchangeable_token=token.get('exchangeableAccessToken', ''), + content_type='application/json') + body = { + 'accessToken': token.get('cciAccessToken', '').removeprefix('Bearer '), + 'refreshToken': token.get('cciRefreshToken', ''), + 'exchangeableAccessToken': token.get('exchangeableAccessToken', ''), + 'exchangeableRefreshToken': token.get('exchangeableRefreshToken', ''), + 'nonCcsToken': token.get('nonCcsToken', ''), + 'nonCcsRefreshToken': token.get('nonCcsRefreshToken', ''), + 'idToken': token.get('idToken', ''), + } + response = requests.post(cci_domain_api_url + 'v2/auth/token-refresh', + json=body, headers=headers, timeout=30) + if response.status_code != 200: + log.error("kia.refreshCciToken: CCI-Token-Refresh fehlgeschlagen, StatusCode: " + + str(response.status_code) + ': ' + response.text[:200]) + raise RuntimeError + data = response.json() + token['cciAccessToken'] = data.get('accessToken', token.get('cciAccessToken', '')) + token['cciRefreshToken'] = data.get('refreshToken', token.get('cciRefreshToken', '')) + token['nonCcsToken'] = data.get('nonCcsToken', token.get('nonCcsToken', '')) + token['exchangeableAccessToken'] = data.get( + 'exchangeableAccessToken', token.get('exchangeableAccessToken', '')) + token['exchangeableRefreshToken'] = data.get( + 'exchangeableRefreshToken', token.get('exchangeableRefreshToken', '')) + token['nonCcsRefreshToken'] = data.get('nonCcsRefreshToken', token.get('nonCcsRefreshToken', '')) + token['idToken'] = data.get('idToken', token.get('idToken', '')) + + ccs_token = exchangeCcsToken(device_id, token['cciAccessToken'], token['nonCcsToken'], + token['exchangeableAccessToken'], brand) + token['tokenType'] = 'Bearer' + token['accessToken'] = ccs_token return token @@ -660,12 +791,19 @@ def requestToken(user_id: str, password: str, brand: str) -> dict: log.info("Kia/Hyundai: Token request starting") try: - # getCookies() is no longer needed: the IDPConnect getAuthCode() - # creates its own session. Kept the call out to avoid hitting the - # decommissioned Keycloak session endpoints. + # getCookies() is no longer needed: die OneApp/CCI getAuthCode() + # erstellt ihre eigene Session. Legacy IDPConnect-Login (client_id, + # getAuthToken) wird von Kia/Hyundais WAF blockiert (siehe #3410) - + # stattdessen OneApp-Auth-Code über CCI gegen einen CCS-Token tauschen, + # der von den bestehenden ccapi:8080-Endpunkten akzeptiert wird. token = getDeviceId(brand) auth_code = getAuthCode(user_id, password, brand, {}) - token = getAuthToken(auth_code, token, brand) + token.update(exchangeCciToken(auth_code, token["deviceId"], brand)) + token["accessToken"] = exchangeCcsToken( + token["deviceId"], token["cciAccessToken"], token["nonCcsToken"], + token["exchangeableAccessToken"], brand) + token["tokenType"] = "Bearer" + token["refreshToken"] = token["cciRefreshToken"] token["userHash"] = getUserHash(user_id, password) registerDevice(token, brand) except Exception: @@ -676,42 +814,9 @@ def requestToken(user_id: str, password: str, brand: str) -> dict: def refreshToken(token: dict, brand: str) -> dict: - log.info("Kia/Hyundai: Token refresh starting") - - response = "" - try: - # New IDPConnect refresh endpoint; requires client_secret. - url = getString("login_form_host", brand) + \ - '/auth/api/v2/user/oauth2/token' - data = urlparse.urlencode({ - 'grant_type': 'refresh_token', - 'refresh_token': token["refreshToken"], - 'client_id': getString("client_id", brand), - 'client_secret': getString("client_secret", brand), - }) - headers = { - 'Content-type': 'application/x-www-form-urlencoded', - 'User-Agent': 'okhttp/3.12.12', - } - - response = postHTTP(url=url, headers=headers, data=data) - - token_new = json.loads(response) - token["tokenType"] = token_new["token_type"] - token["accessToken"] = token_new["access_token"] - # IDPConnect may rotate the refresh token; keep the newest one. - if "refresh_token" in token_new: - token["refreshToken"] = token_new["refresh_token"] - - except Exception: - log.exception("kia.refreshToken: refresh token error: " + - str(response)) - raise - - log.debug("kia.refreshToken: New access token = " + - token["accessToken"][:8] + "...") - - return token + # Legacy IDPConnect-Refresh (client_id-basiert) ist vom selben WAF-Block + # betroffen wie der Login - CCI-Refresh ist der einzig noch funktionierende Weg + return refreshCciToken(token, brand) def getControlToken(pin: str, token: dict, brand: str) -> str: diff --git a/packages/modules/vehicles/kia/api_test.py b/packages/modules/vehicles/kia/api_test.py index ac97c66e6d..bfe13cc040 100644 --- a/packages/modules/vehicles/kia/api_test.py +++ b/packages/modules/vehicles/kia/api_test.py @@ -47,3 +47,144 @@ def test_target_soc_below_100_has_warning(self, monkeypatch): def test_target_soc_100_has_no_warning(self, monkeypatch): result = self._call(monkeypatch, _status_response(target_soc=100)) assert result.warning is None + + +class TestCciHeaders: + def test_bearer_prefix_is_not_duplicated(self, monkeypatch): + monkeypatch.setattr(api, "getString", Mock(return_value="x")) + headers = api.getCciHeaders("kia", "device-1", cci_access_token="Bearer abc123") + assert headers["authorization"] == "Bearer abc123" + + def test_no_body_sets_content_length_zero(self, monkeypatch): + monkeypatch.setattr(api, "getString", Mock(return_value="x")) + headers = api.getCciHeaders("kia", "device-1") + assert headers["Content-Length"] == "0" + assert "Content-Type" not in headers + + def test_exchangeable_token_sets_non_ccs_header(self, monkeypatch): + monkeypatch.setattr(api, "getString", Mock(return_value="x")) + headers = api.getCciHeaders("kia", "device-1", non_ccs_token="ncs", exchangeable_token="ex") + assert headers["exchangeable-token"] == "ex" + assert headers["non-ccs-token"] == "ncs" + assert headers["Authentication"] == "ncs" + + +class TestExchangeCciToken: + def test_raises_on_non_200(self, monkeypatch): + monkeypatch.setattr(api, "getString", Mock(return_value="https://cci-api-eu.kia.com")) + monkeypatch.setattr(api.requests, "post", Mock(return_value=Mock(status_code=400, text="bad request"))) + with pytest.raises(RuntimeError): + api.exchangeCciToken("auth-code", "device-1", "kia") + + def test_returns_expected_fields(self, monkeypatch): + monkeypatch.setattr(api, "getString", Mock(return_value="https://cci-api-eu.kia.com")) + response = Mock(status_code=200) + response.json.return_value = { + "accessToken": "cci-access", "refreshToken": "cci-refresh", + "nonCcsToken": "ncs", "exchangeableAccessToken": "ex-access", + "exchangeableRefreshToken": "ex-refresh", "nonCcsRefreshToken": "ncs-refresh", + "idToken": "id-1", + } + monkeypatch.setattr(api.requests, "post", Mock(return_value=response)) + result = api.exchangeCciToken("auth-code", "device-1", "kia") + assert result["cciAccessToken"] == "cci-access" + assert result["nonCcsToken"] == "ncs" + assert result["idToken"] == "id-1" + + +class TestExchangeCcsToken: + def test_raises_on_non_200(self, monkeypatch): + monkeypatch.setattr(api, "getString", Mock(return_value="https://cci-api-eu.kia.com")) + monkeypatch.setattr(api.requests, "post", Mock(return_value=Mock(status_code=403, text="forbidden"))) + with pytest.raises(RuntimeError): + api.exchangeCcsToken("device-1", "cci-access", "ncs", "ex-access", "kia") + + def test_raises_when_no_access_token_in_response(self, monkeypatch): + monkeypatch.setattr(api, "getString", Mock(return_value="https://cci-api-eu.kia.com")) + response = Mock(status_code=200, text="{}") + response.json.return_value = {} + monkeypatch.setattr(api.requests, "post", Mock(return_value=response)) + with pytest.raises(RuntimeError): + api.exchangeCcsToken("device-1", "cci-access", "ncs", "ex-access", "kia") + + def test_returns_ccs_token(self, monkeypatch): + monkeypatch.setattr(api, "getString", Mock(return_value="https://cci-api-eu.kia.com")) + response = Mock(status_code=200) + response.json.return_value = {"accessToken": "ccs-token-value"} + monkeypatch.setattr(api.requests, "post", Mock(return_value=response)) + result = api.exchangeCcsToken("device-1", "cci-access", "ncs", "ex-access", "kia") + assert result == "ccs-token-value" + + +class TestRefreshCciToken: + def _token(self): + return {"cciAccessToken": "old", "cciRefreshToken": "old-r", "nonCcsToken": "old-n", + "exchangeableAccessToken": "e", "exchangeableRefreshToken": "er", + "nonCcsRefreshToken": "nr", "idToken": "i", "deviceId": "d"} + + def test_raises_on_non_200(self, monkeypatch): + monkeypatch.setattr(api, "getString", Mock(return_value="https://cci-api-eu.kia.com")) + monkeypatch.setattr(api.requests, "post", Mock(return_value=Mock(status_code=401, text="unauthorized"))) + with pytest.raises(RuntimeError): + api.refreshCciToken(self._token(), "kia") + + def test_updates_token_and_reexchanges_ccs(self, monkeypatch): + monkeypatch.setattr(api, "getString", Mock(return_value="https://cci-api-eu.kia.com")) + refresh_response = Mock(status_code=200) + refresh_response.json.return_value = { + "accessToken": "new-cci-access", "refreshToken": "new-cci-refresh", "nonCcsToken": "new-ncs", + } + monkeypatch.setattr(api.requests, "post", Mock(return_value=refresh_response)) + monkeypatch.setattr(api, "exchangeCcsToken", Mock(return_value="new-ccs-token")) + + result = api.refreshCciToken(self._token(), "kia") + + assert result["cciAccessToken"] == "new-cci-access" + assert result["accessToken"] == "new-ccs-token" + assert result["tokenType"] == "Bearer" + + +class TestGetAuthCode: + def test_waf_abuse_block_raises(self, monkeypatch): + monkeypatch.setattr(api, "getString", Mock(return_value="x")) + fake_session = Mock() + fake_session.get.return_value = Mock( + text="Bad Request. It was classified as an abusing request and blocked.", + url="https://idpconnect-eu.kia.com/error?status=400") + monkeypatch.setattr(api.requests, "Session", Mock(return_value=fake_session)) + with pytest.raises(RuntimeError): + api.getAuthCode("user@example.com", "pw", "kia", {}) + + +class TestRequestToken: + def test_populates_ccs_and_cci_fields(self, monkeypatch): + monkeypatch.setattr(api, "getDeviceId", Mock(return_value={ + "deviceId": "dev-1", "gcmClientId": "c", "gcmVehicleId": "v"})) + monkeypatch.setattr(api, "getAuthCode", Mock(return_value="auth-code-1")) + monkeypatch.setattr(api, "exchangeCciToken", Mock(return_value={ + "cciAccessToken": "cci-access", "cciRefreshToken": "cci-refresh", + "nonCcsToken": "ncs", "exchangeableAccessToken": "ex-access", + "exchangeableRefreshToken": "ex-refresh", "nonCcsRefreshToken": "ncs-refresh", + "idToken": "id-1", + })) + monkeypatch.setattr(api, "exchangeCcsToken", Mock(return_value="ccs-token-value")) + monkeypatch.setattr(api, "registerDevice", Mock()) + + token = api.requestToken("user@example.com", "pw", "kia") + + assert token["accessToken"] == "ccs-token-value" + assert token["tokenType"] == "Bearer" + assert token["refreshToken"] == "cci-refresh" + assert token["cciAccessToken"] == "cci-access" + + +class TestRefreshToken: + def test_delegates_to_refresh_cci_token(self, monkeypatch): + refresh_mock = Mock(return_value={"accessToken": "new-ccs"}) + monkeypatch.setattr(api, "refreshCciToken", refresh_mock) + token = {"cciAccessToken": "old"} + + result = api.refreshToken(token, "kia") + + refresh_mock.assert_called_once_with(token, "kia") + assert result == {"accessToken": "new-ccs"}