diff --git a/packages/modules/vehicles/cupra/libcupra.py b/packages/modules/vehicles/cupra/libcupra.py index 9332ff2acd..93589b219e 100755 --- a/packages/modules/vehicles/cupra/libcupra.py +++ b/packages/modules/vehicles/cupra/libcupra.py @@ -8,6 +8,7 @@ import base64 import hashlib +from datetime import datetime, timezone from helpermodules.utils.error_handling import ImportErrorContext with ImportErrorContext(): import lxml.html @@ -16,7 +17,8 @@ LOGIN_BASE = "https://identity.vwgroup.io/oidc/v1" LOGIN_HANDLER_BASE = "https://identity.vwgroup.io" API_BASE = "https://ola.prod.code.seat.cloud.vwgroup.com" -CLIENT_ID = "99a5b77d-bd88-4d53-b4e5-a539c60694a3@apps_vw-dilab_com" +CLIENT_ID = "3c756d46-f1ba-4d78-9f9a-cff0d5292d51@apps_vw-dilab_com" +USER_AGENT = "OLACupra/2.16.0 (Android 14; Pixel 8; Google) Mobile" class cupra: @@ -97,6 +99,14 @@ def to_camel_case(s): return {to_camel_case(k): v for k, v in json.items()} + def token_headers(self): + basic_auth = base64.b64encode(f"{CLIENT_ID}:".encode('utf-8')).decode('utf-8') + return { + 'Content-Type': 'application/x-www-form-urlencoded', + 'Authorization': f'Basic {basic_auth}', + 'User-Agent': USER_AGENT, + } + async def connect(self, username, password): self.set_credentials(username, password) return (await self.reconnect()) @@ -107,13 +117,16 @@ async def reconnect(self): self.log.debug("Starting Cupra reconnect/auth flow") # Get authorize page - _scope = 'openid profile nickname birthdate phone' + _scope = ( + 'openid profile nickname birthdate phone mbb cars address ' + 'nationalIdentifier nationality profession badge driversLicense' + ) payload = { 'client_id': CLIENT_ID, 'scope': _scope, 'response_type': 'code', 'nonce': secrets.token_urlsafe(12), - 'redirect_uri': 'seat://oauth-callback', + 'redirect_uri': 'cupra://oauth-callback', 'state': str(uuid.uuid4()), 'code_challenge': code_challenge, 'code_challenge_method': 'S256' @@ -177,7 +190,7 @@ async def reconnect(self): url = response.headers['Location'] self.log.debug("Redirect: status=%s location=%s", response.status, url) - if (url.split(':')[0] == "seat"): + if (url.split(':')[0] == "cupra"): if not ('code' in url): self.log.error("Missing authorization code") return False @@ -197,22 +210,17 @@ async def reconnect(self): form = {} form['code'] = query['code'] form['client_id'] = CLIENT_ID - form['redirect_uri'] = "seat://oauth-callback" + form['redirect_uri'] = "cupra://oauth-callback" form['grant_type'] = 'authorization_code' form['code_verifier'] = code_verifier - headers = {} - headers['Content-Type'] = 'application/x-www-form-urlencoded' - headers['Authorization'] = 'BASIC OTlhNWI3N2QtYmQ4OC00ZDUzLWI0ZTUtYTUzOWM2MDY5NGEzQGFwcHNfdnctZGlsYWJfY29tOg==' - headers['User-Agent'] = ( - 'SEATApp/2.5.0 (com.seat.myseat.ola; build:202410171614; ' - 'iOS 15.8.3) Alamofire/5.7.0 Mobile' - ) + headers = self.token_headers() response = await self.session.post(API_BASE + '/authorization/api/v1/token', headers=headers, data=form) self.log.debug("Authorization code exchange status=%s", response.status) if response.status >= 400: - self.log.error("Login: Non-2xx response") + self.log.error("Login: Non-2xx response (%s), body=%s", + response.status, await response.text()) # Non 2xx response, failed return False self.tokens = self.convert_to_camel_case(await response.json()) @@ -233,16 +241,13 @@ async def refresh_tokens(self): form['client_id'] = CLIENT_ID form['grant_type'] = 'refresh_token' form['refresh_token'] = self.tokens["refreshToken"] - headers = {} - headers['Content-Type'] = 'application/x-www-form-urlencoded' - headers['User-Agent'] = ( - 'SEATApp/2.5.0 (com.seat.myseat.ola; build:202410171614; ' - 'iOS 15.8.3) Alamofire/5.7.0 Mobile' - ) + headers = self.token_headers() response = await self.session.post(API_BASE + '/authorization/api/v1/token', headers=headers, data=form) if response.status >= 400: + self.log.error("Refresh token exchange failed (%s), body=%s", + response.status, await response.text()) return False self.tokens = self.convert_to_camel_case(await response.json()) @@ -252,7 +257,12 @@ async def refresh_tokens(self): return True async def get_status(self): - status_url = f"{API_BASE}/vehicles/{self.vin}/charging/status" + self.headers['user-agent'] = USER_AGENT + self.headers['app-brand'] = 'cupra' + self.headers['app-market'] = 'android' + self.headers['app-version'] = '2.16.0' + status_url = f"{API_BASE}/v1/vehicles/{self.vin}/charging/status" + statusv2_url = f"{API_BASE}/v2/vehicles/{self.vin}/status" mileage_url = f"{API_BASE}/v1/vehicles/{self.vin}/mileage" response = await self.session.get(status_url, headers=self.headers) @@ -288,10 +298,22 @@ async def get_status(self): self.log.debug(f"Mileage data from Cupra API: {mileage_data}") odometer = mileage_data.get('mileageKm', None) + # Fetch additional status data from v2 endpoint + response = await self.session.get(statusv2_url, headers=self.headers) + if response.status >= 400: + self.log.error("Get status v2 failed") + else: + statusv2_data = await response.json() + self.log.debug(f"Status v2 data from Cupra API: {statusv2_data}") + # use current timestamp as a fallback, as the API field is missing + carCapturedTimestamp = statusv2_data.get('updatedAt', datetime.now(timezone.utc).isoformat()).split('.')[0] + if not carCapturedTimestamp.endswith('Z'): + carCapturedTimestamp += 'Z' + battery_value = { - 'currentSOC_pct': status_data['status']['battery']['currentSOC_pct'], - 'cruisingRangeElectric_km': status_data['status']['battery']['cruisingRangeElectric_km'], - 'carCapturedTimestamp': status_data['status']['battery']['carCapturedTimestamp'], + 'currentSOC_pct': status_data['battery']['currentSocPercentage'], + 'cruisingRangeElectric_km': status_data['battery']['estimatedRangeInKm'], + 'carCapturedTimestamp': status_data['battery'].get('updatedAt', carCapturedTimestamp), 'odometer': odometer, } diff --git a/packages/modules/vehicles/cupra/soc_test.py b/packages/modules/vehicles/cupra/soc_test.py index ade6ce013c..6b4ca994a9 100644 --- a/packages/modules/vehicles/cupra/soc_test.py +++ b/packages/modules/vehicles/cupra/soc_test.py @@ -7,7 +7,8 @@ from modules.vehicles.cupra import api from modules.vehicles.cupra.soc import create_vehicle from modules.vehicles.cupra.config import Cupra, CupraConfiguration -from modules.vehicles.cupra.libcupra import cupra as CupraApi +from modules.vehicles.cupra.libcupra import cupra as CupraApi, CLIENT_ID, USER_AGENT +import base64 class TestCupra: @@ -93,26 +94,38 @@ async def get_side_effect(*args, **kwargs): def cupra_instance(self, mock_session): instance = CupraApi(mock_session) instance.set_vin("test_vin") - instance.headers = {"Authorization": "Bearer test_token"} + instance.headers = { + "Authorization": "Bearer test_token", + "app-market": "android", + } return instance + def test_token_headers_use_dynamic_basic_auth_and_user_agent(self, cupra_instance): + headers = cupra_instance.token_headers() + expected_basic = base64.b64encode(f"{CLIENT_ID}:".encode("utf-8")).decode("utf-8") + + assert headers["Content-Type"] == "application/x-www-form-urlencoded" + assert headers["Authorization"] == f"Basic {expected_basic}" + assert headers["User-Agent"] == USER_AGENT + def test_get_status_success(self, cupra_instance, mock_session): # setup status_response_data = { - "status": { - "battery": { - "currentSOC_pct": 67, - "cruisingRangeElectric_km": 305, - "carCapturedTimestamp": "2026-03-29T11:20:00Z" - } + "battery": { + "currentSocPercentage": 67, + "estimatedRangeInKm": 305 } } mileage_response_data = { "mileageKm": 77889 } + statusv2_response_data = { + "updatedAt": "2026-03-29T11:20:00Z" + } responses = [ MockAiohttpResponse(status_response_data, 200), - MockAiohttpResponse(mileage_response_data, 200) + MockAiohttpResponse(mileage_response_data, 200), + MockAiohttpResponse(statusv2_response_data, 200) ] async def side_effect_func(*args, **kwargs): @@ -127,33 +140,39 @@ async def side_effect_func(*args, **kwargs): assert status['charging']['batteryStatus']['value']['cruisingRangeElectric_km'] == 305 assert status['charging']['batteryStatus']['value']['carCapturedTimestamp'] == "2026-03-29T11:20:00Z" assert status['charging']['batteryStatus']['value']['odometer'] == 77889 - assert mock_session.get.call_count == 2 + assert mock_session.get.call_count == 3 mock_session.get.assert_any_call( - "https://ola.prod.code.seat.cloud.vwgroup.com/vehicles/test_vin/charging/status", + "https://ola.prod.code.seat.cloud.vwgroup.com/v1/vehicles/test_vin/charging/status", headers=cupra_instance.headers ) mock_session.get.assert_any_call( "https://ola.prod.code.seat.cloud.vwgroup.com/v1/vehicles/test_vin/mileage", headers=cupra_instance.headers ) + mock_session.get.assert_any_call( + "https://ola.prod.code.seat.cloud.vwgroup.com/v2/vehicles/test_vin/status", + headers=cupra_instance.headers + ) def test_get_status_mileage_error_sets_odometer_none(self, cupra_instance, mock_session): # setup status_response_data = { - "status": { - "battery": { - "currentSOC_pct": 67, - "cruisingRangeElectric_km": 305, - "carCapturedTimestamp": "2026-03-29T11:20:00Z" - } + "battery": { + "currentSocPercentage": 67, + "estimatedRangeInKm": 305, + "updatedAt": "2026-03-29T11:20:00Z" } } mileage_error_response = { "error": "service_unavailable" } + statusv2_response_data = { + "error": "service_unavailable" + } responses = [ MockAiohttpResponse(status_response_data, 200), - MockAiohttpResponse(mileage_error_response, 503) + MockAiohttpResponse(mileage_error_response, 503), + MockAiohttpResponse(statusv2_response_data, 200) ] async def side_effect_func(*args, **kwargs):