Cupra SoC: Fix server communication

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,
         }
 

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):