Skip to content

Kia/Hyundai: X-Fingerprint-Signatur für CCI-Gate #4232

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

Merged
LKuemmel merged 1 commit into from
Nov 11, 2026
Merged

Kia/Hyundai: X-Fingerprint-Signatur für CCI-Gate #4232

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
121 changes: 105 additions & 16 deletions packages/modules/vehicles/kia/api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,10 @@
import base64
from datetime import datetime
import json
import re
import uuid
import hashlib
import hmac
import requests
import urllib.parse as urlparse
from urllib.parse import parse_qs
Expand Down Expand Up @@ -630,6 +632,66 @@ def getAuthCode(username: str, password: str, brand: str,

return auth_code


# ---------- CCI fingerprint gate (seit 2026-10-14) ----------
# Ohne X-Fingerprint-HMAC: 9035/9036/9037. Schema + Secrets aus
# hyundai_kia_connect_api#1352/#1355 (Android-OneApp-1.4.0-Build).

_FINGERPRINT_KEYS = {
'hyundai': ('OsZDfDuCyVX', 'yfqjCoA2JbX', '2MrLpGwNBC'),
'kia': ('yb4Z1n2nlKe', 'ukNilE7QEdb', 'KB34nStPVa'),
}

_GATE_REJECTION_CODES = frozenset({9035, 9036, 9037})


def computeXFingerprint(headers: dict, brand: str, body: str = "") -> str:
lower = {k.lower(): v for k, v in headers.items()}
identity = "".join(lower.get(name, "") for name in
('client-id', 'client-name', 'client-version', 'client-os-code'))
pieces = _FINGERPRINT_KEYS[brand]
msg = (identity + lower.get('client-os-version', '') + lower.get('x-timestamp', '') +
re.sub(r'\s+', '', body or '', flags=re.ASCII))
key = identity + ''.join(pieces)
return hmac.new(key.encode(), msg.encode(), hashlib.sha256).hexdigest()


def isGateRejection(status_code: int, text: str) -> bool:
if not 400 <= status_code < 500:
return False
try:
return json.loads(text).get('code') in _GATE_REJECTION_CODES
except (ValueError, AttributeError):
return False


def getCciFpHeaders(brand: str, device_id: str, body: str = "") -> dict:
# Gate prüft gegen diese Android-OneApp-1.4.0-Identität, nicht die
# iOS-Legacy-Header aus getCciHeaders() (bleiben für den Fallback).
offset = datetime.now().astimezone().strftime("%z")
headers = {
'client-id': getString("cci_package_id", brand),
'client-name': brand.capitalize() + 'EU',
'client-version': '1.4.0',
'client-os-code': 'android',
'client-os-version': '14',
'client-device-id': device_id,
'client-device-model': 'SM-S918B',
'client-notification-provider-type': 'FCM',
'locale': 'DE',
'timezone': (offset[:3] + ':' + offset[3:]) if offset else '+00:00',
'App-Request-Id': uuid.uuid4().hex[:24].upper() + '_' + device_id.replace('-', '') + '_AOS',
'X-Brand': brand.capitalize(),
'X-Region': 'eu',
'User-Agent': 'okhttp/4.12.0',
'Accept': '*/*',
'Accept-Language': 'de',
}
headers['X-Timestamp'] = str(int(time.time() * 1000))
headers['X-Fingerprint'] = computeXFingerprint(headers, brand, body)
return headers


# ---------- OneApp/CCI token exchange ----------


Expand Down Expand Up @@ -669,11 +731,18 @@ def getCciHeaders(brand: str, device_id: str, cci_access_token: str = "",
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',
cci_api_url = getString("cci_api_url", brand)
headers = getCciFpHeaders(brand, device_id)
response = requests.post(cci_api_url + '/auth/v1/token',
params={'code': auth_code}, headers=headers,
timeout=30)
if response.status_code != 200 and isGateRejection(response.status_code, response.text):
log.warning("kia.exchangeCciToken: Fingerprint-Gate hat signierte Anfrage abgelehnt "
"(HTTP %d), Fallback auf Legacy-Route.", response.status_code)
headers = getCciHeaders(brand, device_id)
response = requests.post(cci_api_url + '/domain/api/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])
Expand All @@ -694,13 +763,25 @@ 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',
cci_api_url = getString("cci_api_url", brand)
headers = getCciFpHeaders(brand, device_id)
headers['authorization'] = 'Bearer ' + cci_access_token.removeprefix('Bearer ').strip()
headers['Authentication'] = non_ccs_token
if exchangeable_token:
headers['exchangeable-token'] = exchangeable_token
headers['non-ccs-token'] = non_ccs_token
response = requests.post(cci_api_url + '/auth/v1/token-exchange',
params={'serviceType': 'CCS'}, headers=headers,
timeout=30)
if response.status_code != 200 and isGateRejection(response.status_code, response.text):
log.warning("kia.exchangeCcsToken: Fingerprint-Gate hat signierte Anfrage abgelehnt "
"(HTTP %d), Fallback auf Legacy-Route.", response.status_code)
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_api_url + '/domain/api/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])
Expand All @@ -717,12 +798,7 @@ 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')
cci_api_url = getString("cci_api_url", brand)
body = {
'accessToken': token.get('cciAccessToken', '').removeprefix('Bearer '),
'refreshToken': token.get('cciRefreshToken', ''),
Expand All @@ -732,8 +808,21 @@ def refreshCciToken(token: dict, brand: str) -> dict:
'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)
payload = json.dumps(body)
headers = getCciFpHeaders(brand, device_id, body=payload)
headers['Content-Type'] = 'application/json'
response = requests.post(cci_api_url + '/auth/v2/token-refresh',
data=payload, headers=headers, timeout=30)
if response.status_code != 200 and isGateRejection(response.status_code, response.text):
log.warning("kia.refreshCciToken: Fingerprint-Gate hat signierte Anfrage abgelehnt "
"(HTTP %d), Fallback auf Legacy-Route.", response.status_code)
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')
response = requests.post(cci_api_url + '/domain/api/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])
Expand Down
62 changes: 62 additions & 0 deletions packages/modules/vehicles/kia/api_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,36 @@ def test_exchangeable_token_sets_non_ccs_header(self, monkeypatch):
assert headers["Authentication"] == "ncs"


class TestCciFingerprint:
# Pinned live-verified vectors from hyundai_kia_connect_api#1352/#1355
# (test_eu_cci_fingerprint.py) - confirms our port matches byte-for-byte.
def test_matches_pinned_kia_vector(self):
headers = {
"client-id": "com.kia.oneapp.eu", "client-name": "KiaEU",
"client-version": "1.4.0", "client-os-code": "android",
"client-os-version": "14", "X-Timestamp": "1700000000123",
}
assert api.computeXFingerprint(headers, "kia") == \
"0a3ceccc83823ec0d2c5d39a15dbf722c813e6da48800a0156a7a8459db63337"

def test_matches_pinned_hyundai_vector(self):
headers = {
"client-id": "com.hyundai.oneapp.eu", "client-name": "HyundaiEU",
"client-version": "1.4.0", "client-os-code": "android",
"client-os-version": "14", "X-Timestamp": "1792000084084",
}
assert api.computeXFingerprint(headers, "hyundai") == \
"df04e3dadd52288ce5b280b368af261272556b5360adbc90a817a167526e2946"

def test_gate_rejection_codes(self):
for code in (9035, 9036, 9037):
assert api.isGateRejection(400, f'{{"code":{code},"message":"x"}}')
assert not api.isGateRejection(400, '{"code":9013,"message":"x"}')
assert not api.isGateRejection(500, '{"code":9035,"message":"x"}')
assert not api.isGateRejection(200, '{"code":9035,"message":"x"}')
assert not api.isGateRejection(502, "upstream html error")


class TestExchangeCciToken:
def test_raises_on_non_200(self, monkeypatch):
monkeypatch.setattr(api, "getString", Mock(return_value="https://cci-api-eu.kia.com"))
Expand All @@ -91,6 +121,38 @@ def test_returns_expected_fields(self, monkeypatch):
assert result["nonCcsToken"] == "ncs"
assert result["idToken"] == "id-1"

def test_uses_signed_endpoint_first(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"}
mock_post = Mock(return_value=response)
monkeypatch.setattr(api.requests, "post", mock_post)
api.exchangeCciToken("auth-code", "device-1", "kia")
assert mock_post.call_count == 1
assert mock_post.call_args.args[0] == "https://cci-api-eu.kia.com/auth/v1/token"
assert mock_post.call_args.kwargs["headers"]["X-Fingerprint"]

def test_falls_back_to_legacy_on_gate_rejection(self, monkeypatch):
monkeypatch.setattr(api, "getString", Mock(return_value="https://cci-api-eu.kia.com"))
gate_response = Mock(status_code=400, text='{"code":9035,"message":"x"}')
legacy_response = Mock(status_code=200)
legacy_response.json.return_value = {"accessToken": "legacy-access"}
mock_post = Mock(side_effect=[gate_response, legacy_response])
monkeypatch.setattr(api.requests, "post", mock_post)
result = api.exchangeCciToken("auth-code", "device-1", "kia")
assert result["cciAccessToken"] == "legacy-access"
assert mock_post.call_count == 2
assert mock_post.call_args_list[1].args[0] == "https://cci-api-eu.kia.com/domain/api/v1/auth/token"
assert "X-Fingerprint" not in mock_post.call_args_list[1].kwargs["headers"]

def test_non_gate_error_does_not_fall_back(self, monkeypatch):
monkeypatch.setattr(api, "getString", Mock(return_value="https://cci-api-eu.kia.com"))
mock_post = Mock(return_value=Mock(status_code=502, text="bad gateway"))
monkeypatch.setattr(api.requests, "post", mock_post)
with pytest.raises(RuntimeError):
api.exchangeCciToken("auth-code", "device-1", "kia")
assert mock_post.call_count == 1


class TestExchangeCcsToken:
def test_raises_on_non_200(self, monkeypatch):
Expand Down
Loading