Support Porsche mobile auth flow and Macan EV data

pyporscheconnectapi/account.py

@@ -4,12 +4,40 @@
 
 import logging
 
-from pyporscheconnectapi.connection import Connection
-from pyporscheconnectapi.vehicle import PorscheVehicle
+from .connection import Connection
+from .vehicle import PorscheVehicle
 
 _LOGGER = logging.getLogger(__name__)
 
 
+def _normalize_engine(vehicle: dict) -> str:
+    """Best-effort mapping of the portal vehicle payload to drivetrain type."""
+    model_type = vehicle.get("modelType", {})
+    if model_type.get("engine"):
+        return model_type["engine"]
+    description = str(vehicle.get("modelDescription", "")).lower()
+    if description in {"macan", "taycan"}:
+        return "BEV"
+    return "COMBUSTION"
+
+
+def _normalize_vehicle(vehicle: dict) -> dict:
+    """Normalize the portal vehicle payload to the legacy library shape."""
+    model_name = vehicle.get("modelDescription") or vehicle.get("modelName") or vehicle.get("vin", "Porsche")
+    return {
+        "vin": vehicle["vin"],
+        "name": model_name,
+        "modelName": model_name,
+        "modelType": {
+            "year": vehicle.get("modelYear") or vehicle.get("modelType", {}).get("year", "not available"),
+            "engine": _normalize_engine(vehicle),
+        },
+        "systemInfo": vehicle.get("systemInfo", {}),
+        "timestamp": vehicle.get("validFrom") or vehicle.get("timestamp"),
+        "portalVehicle": vehicle,
+    }
+
+
 class PorscheConnectAccount:
     """Establishes a connection to a Porsche Connect account."""
 
@@ -39,7 +67,7 @@ async def _init_vehicles(self) -> None:
                 _LOGGER.debug("Got vehicle %s", vehicle)
                 v = PorscheVehicle(
                     vin=vehicle["vin"],
-                    data=vehicle,
+                    data=_normalize_vehicle(vehicle),
                     status={},
                     connection=self.connection,
                 )

pyporscheconnectapi/cli.py

@@ -13,10 +13,10 @@
 
 import aiofiles
 
-from pyporscheconnectapi.account import PorscheConnectAccount
-from pyporscheconnectapi.connection import Connection
-from pyporscheconnectapi.exceptions import PorscheCaptchaRequiredError, PorscheWrongCredentialsError
-from pyporscheconnectapi.remote_services import RemoteServices
+from .account import PorscheConnectAccount
+from .connection import Connection
+from .exceptions import PorscheCaptchaRequiredError, PorscheWrongCredentialsError
+from .remote_services import RemoteServices
 
 vehicle_commands = {
     "battery": "Prints the main battery level (BEV)",

pyporscheconnectapi/connection.py

@@ -5,10 +5,11 @@
 
 import asyncio
 import logging
+import uuid
 
 import httpx
 
-from .const import API_BASE_URL, TIMEOUT, USER_AGENT, X_CLIENT_ID
+from .const import API_BASE_URL, DCGW_BASE_URL, TIMEOUT, USER_AGENT, X_CLIENT_ID
 from .exceptions import PorscheExceptionError
 from .oauth2 import Captcha, Credentials, OAuth2Client, OAuth2Token
 
@@ -39,6 +40,8 @@ def __init__(
         captcha_code: str | None = None,
         state: str | None = None,
         async_client=httpx.AsyncClient(),
+        cookies: list[dict] | None = None,
+        code_verifier: str | None = None,
         token=None,
         leeway: int = 60,
     ) -> None:
@@ -47,15 +50,33 @@ def __init__(
             token = {}
         self.asyncClient = async_client
         self.token_lock = asyncio.Lock()
+        self.country_code = "de"
+        self.language_code = "de_DE"
+
+        if cookies:
+            for cookie in cookies:
+                self.asyncClient.cookies.set(
+                    cookie["name"],
+                    cookie["value"],
+                    domain=cookie.get("domain"),
+                    path=cookie.get("path", "/"),
+                )
 
         self.token = OAuth2Token(token)
 
-        self.headers = {"User-Agent": USER_AGENT, "X-Client-ID": X_CLIENT_ID}
+        self.headers = {
+            "User-Agent": USER_AGENT,
+            "X-Client-ID": X_CLIENT_ID,
+            "Accept-Language": "de-DE",
+            "Origin": "https://security.porsche.com",
+            "Referer": "https://security.porsche.com/",
+        }
 
         self.oauth2_client = OAuth2Client(
             self.asyncClient,
             Credentials(email, password),
             Captcha(captcha_code, state),
+            code_verifier,
             leeway,
         )
 
@@ -69,6 +90,18 @@ async def get(self, url, params=None):
         """Make a GET request to the Porsche Connect API."""
         return await self.request("GET", url, params=params)
 
+    async def portal_get(self, url, params=None):
+        """Make a GET request to the Porsche DCGW portal API."""
+        return await self.absolute_request("GET", f"{DCGW_BASE_URL}{url}", params=params)
+
+    async def get_portal_config(self):
+        """Fetch and cache portal localization/config data."""
+        config = await self.portal_get(f"/core/config/v1/{self.country_code}/{self.country_code}/")
+        localization = config.get("localization", {})
+        self.country_code = str(localization.get("countryCode", self.country_code)).lower()
+        self.language_code = localization.get("languageCode", self.language_code)
+        return config
+
     async def post(self, url, data=None, json=None):
         """Make a POST request to the Porsche Connect API."""
         return await self.request("POST", url, data=data, json=json)
@@ -83,20 +116,33 @@ async def delete(self, url, data=None, json=None):
 
     async def request(self, method, url, **kwargs):
         """Create a request to the Porsche Connect API."""
+        return await self.absolute_request(method, f"{API_BASE_URL}{url}", **kwargs)
+
+    async def absolute_request(self, method, url, **kwargs):
+        """Create a request to an absolute Porsche API URL."""
         try:
             async with self.token_lock:
                 await self.oauth2_client.ensure_valid_token(self.token)
+            headers = self.headers | {
+                "Authorization": f"Bearer {self.token.access_token}",
+                "X-TRACE-ID": f"PCCK-PORTAL-{uuid.uuid4()}",
+            }
             resp = await self.asyncClient.request(
                 method,
-                f"{API_BASE_URL}{url}",
-                headers=self.headers | {"Authorization": f"Bearer {self.token.access_token}"},
+                url,
+                headers=headers,
                 timeout=TIMEOUT,
                 **kwargs,
             )
             resp.raise_for_status()  # A common error seem to be: httpx.HTTPStatusError: Server error '504 Gateway Time-out'
             return resp.json()
         except httpx.HTTPStatusError as exc:
-            raise PorscheExceptionError(exc.response.status_code) from exc
+            response_text = exc.response.text[:1000] if exc.response.text else None
+            raise PorscheExceptionError(
+                exc.response.status_code,
+                response_body=response_text,
+                request_url=str(exc.request.url),
+            ) from exc
 
     async def close(self):
         """Close the asyncClient connection."""

pyporscheconnectapi/const.py

@@ -1,12 +1,13 @@
 """Client configuration constants."""
 
 AUTHORIZATION_SERVER = "identity.porsche.com"
-REDIRECT_URI = "my-porsche-app://auth0/callback"
+REDIRECT_URI = "https://security.porsche.com/auth/en-GB/app/callback"
 AUDIENCE = "https://api.porsche.com"
-CLIENT_ID = "XhygisuebbrqQ80byOuU5VncxLIm8E6H"
-X_CLIENT_ID = "41843fb4-691d-4970-85c7-2673e8ecef40"
-USER_AGENT = "pyporscheconnectapi/0.2.0"
+CLIENT_ID = "qIkoJqlAXvbj4R3j12ct3zdinPId0Zbl"
+X_CLIENT_ID = "09fcb5d8-d4ad-48e8-a0e8-a9c7cb1b9cbc"
+USER_AGENT = "de.porsche.one/18.26.09-row+162630 (android)"
 API_BASE_URL = "https://api.ppa.porsche.com/app"
+DCGW_BASE_URL = "https://dgw.p-fra.portal.aws.porsche.cloud"
 AUTHORIZATION_URL = f"https://{AUTHORIZATION_SERVER}/authorize"
 TOKEN_URL = f"https://{AUTHORIZATION_SERVER}/oauth/token"
 TIMEOUT = 90

pyporscheconnectapi/exceptions.py

@@ -9,9 +9,11 @@
 class PorscheExceptionError(Exception):
     """Class of Porsche API exceptions."""
 
-    def __init__(self, code=None, *args, **kwargs) -> None:
+    def __init__(self, code=None, response_body=None, request_url=None, *args, **kwargs) -> None:
         """Initialize exceptions for the Porsche API."""
         self.message = ""
+        self.response_body = response_body
+        self.request_url = request_url
         super().__init__(*args, **kwargs)
         if code is not None:
             self.code = code
@@ -41,6 +43,9 @@ def __init__(self, code=None, *args, **kwargs) -> None:
             elif self.code > 299:
                 self.message = f"UNKNOWN_ERROR_{self.code}"
 
+        if self.response_body:
+            self.message = f"{self.message}: {self.response_body}"
+
 
 class PorscheWrongCredentialsError(PorscheExceptionError):
     """Class of exceptions for incomplete credentials."""