From 241a90cbeaf0a286f6ecc375cd93ffbe9ab22110 Mon Sep 17 00:00:00 2001 From: Fredrik Ljunggren Date: Sun, 30 Aug 2026 08:34:44 +0000 Subject: [PATCH 1/3] Add PKCE and handle the passkey enrollment screen in the auth flow Auth0 has changed the Porsche login flow in two ways that the current implementation does not handle. The authorization request now sends a PKCE challenge (RFC 7636). The verifier is generated before /authorize and kept on the OAuth2Client so it survives a captcha round trip, since the challenge is bound to the Auth0 transaction created by that request. It is sent with the code exchange at the token endpoint. The resume URL returned by the Identifier First flow no longer redirects straight to the callback. Porsche can interleave an optional passkey enrollment screen, so the redirect chain is walked until the authorization code appears, declining enrollment when that screen shows up. PorscheExceptionError also carries the response body and request URL now, so failures at the token endpoint report what the server actually said instead of a bare status code. Co-authored-by: tietjen --- pyporscheconnectapi/connection.py | 6 +- pyporscheconnectapi/exceptions.py | 7 +- pyporscheconnectapi/oauth2.py | 143 ++++++++++++++++++++++++++++-- 3 files changed, 149 insertions(+), 7 deletions(-) diff --git a/pyporscheconnectapi/connection.py b/pyporscheconnectapi/connection.py index a19642f..177425f 100644 --- a/pyporscheconnectapi/connection.py +++ b/pyporscheconnectapi/connection.py @@ -118,7 +118,11 @@ async def request(self, method, url, **kwargs): # noqa: RET503 - loop body alwa except httpx.HTTPStatusError as exc: # noqa: PERF203 status = exc.response.status_code if status not in _RETRY_STATUS_CODES or attempt == _MAX_RETRIES: - raise PorscheExceptionError(status) from exc + raise PorscheExceptionError( + status, + response_body=exc.response.text[:1000] or None, + request_url=str(exc.request.url), + ) from exc delay = _compute_retry_delay(exc.response, attempt) _LOGGER.warning( "Transient HTTP %s on %s - retrying in %.1fs (attempt %d/%d)", diff --git a/pyporscheconnectapi/exceptions.py b/pyporscheconnectapi/exceptions.py index ada14b0..487582e 100644 --- a/pyporscheconnectapi/exceptions.py +++ b/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, *args, response_body=None, request_url=None, **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.""" diff --git a/pyporscheconnectapi/oauth2.py b/pyporscheconnectapi/oauth2.py index 49279c3..c1eedeb 100644 --- a/pyporscheconnectapi/oauth2.py +++ b/pyporscheconnectapi/oauth2.py @@ -4,9 +4,11 @@ import asyncio import base64 import binascii +import hashlib import json import logging import re +import secrets import time from typing import NamedTuple from urllib.parse import parse_qs, urljoin, urlparse @@ -34,6 +36,10 @@ _LOGGER = logging.getLogger(__name__) +# Auth0 screen that may be interleaved into the resume redirect chain +PASSKEY_ENROLLMENT_PATH = "/u/passkey-enrollment" +_MAX_RESUME_REDIRECTS = 10 + class Credentials(NamedTuple): """Store credentials for the Porsche Connect API.""" @@ -114,6 +120,19 @@ def __init__( self.captcha = captcha self.leeway = leeway self.headers = {"User-Agent": USER_AGENT, "X-Client-ID": X_CLIENT_ID} + # PKCE verifier for the authorization request currently in flight. It is + # created before /authorize and has to survive a captcha round trip, + # since the challenge is bound to that Auth0 transaction. + self.code_verifier: str | None = None + + def _generate_pkce_verifier(self) -> str: + """Generate a PKCE code verifier (RFC 7636 section 4.1).""" + return secrets.token_urlsafe(64) + + def _build_pkce_challenge(self, verifier: str) -> str: + """Derive the S256 code challenge from a verifier (RFC 7636 section 4.2).""" + digest = hashlib.sha256(verifier.encode("ascii")).digest() + return base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii") async def ensure_valid_token(self, token: OAuth2Token): """Ensure the access_token is valid, logging in or refreshing if necessary.""" @@ -148,6 +167,7 @@ async def fetch_authorization_code(self): _LOGGER.debug("Fetching authorization code.") # first request to get the code + self.code_verifier = self._generate_pkce_verifier() params = await self.get_and_extract_location_params( AUTHORIZATION_URL, params={ @@ -156,6 +176,8 @@ async def fetch_authorization_code(self): "redirect_uri": REDIRECT_URI, "audience": AUDIENCE, "scope": SCOPE, + "code_challenge": self._build_pkce_challenge(self.code_verifier), + "code_challenge_method": "S256", "state": "pyporscheconnectapi", }, ) @@ -174,10 +196,9 @@ async def fetch_authorization_code(self): resume_path = await self.login_with_identifier(params["state"][0]) # completed the Identifier First flow, now resume the auth code request - params = await self.get_and_extract_location_params( + authorization_code = await self.resume_authorization_code_flow( urljoin(f"https://{AUTHORIZATION_SERVER}", resume_path), ) - authorization_code = params.get("code", [None])[0] except httpx.HTTPStatusError as exc: raise PorscheExceptionError(exc.response.status_code) from exc @@ -188,11 +209,17 @@ async def fetch_authorization_code(self): else: try: + if self.code_verifier is None: + # The challenge was bound to the Auth0 transaction created by the + # /authorize request that produced this captcha. Generating a new + # verifier here would yield a code that cannot be exchanged. + msg = "PKCE_VERIFIER_MISSING_FOR_CAPTCHA_RESUME" + raise PorscheExceptionError(msg) + resume_path = await self.login_with_identifier(self.captcha.state) - params = await self.get_and_extract_location_params( + authorization_code = await self.resume_authorization_code_flow( urljoin(f"https://{AUTHORIZATION_SERVER}", resume_path), ) - authorization_code = params.get("code", [None])[0] except httpx.HTTPStatusError as exc: raise PorscheExceptionError(exc.response.status_code) from exc @@ -268,6 +295,107 @@ def _extract_captcha_image(self, html: str): return None + def _extract_universal_login_context(self, html: str) -> dict | None: + """Extract the Auth0 universal login context from the inline base64 payload.""" + match = re.search(r'atob\("([A-Za-z0-9+/=]+)"', html) + if not match: + return None + + try: + decoded = base64.b64decode(match.group(1)).decode("utf-8") + return json.loads(decoded) + except (ValueError, json.JSONDecodeError, binascii.Error) as exc: + _LOGGER.warning("Failed to parse Auth0 universal login context: %s", exc) + return None + + async def _skip_passkey_enrollment(self, url: str, html: str | None = None) -> str: + """Decline the optional passkey enrollment screen and return where to continue.""" + if html is None: + resp = await self.client.get( + url, + timeout=TIMEOUT, + headers=self.headers, + follow_redirects=False, + ) + resp.raise_for_status() + html = resp.text + + context = self._extract_universal_login_context(html) + if context is None: + msg = "PASSKEY_ENROLLMENT_CONTEXT_MISSING" + raise PorscheExceptionError(msg) + + transaction_state = context.get("transaction", {}).get("state") + if not transaction_state: + msg = "PASSKEY_ENROLLMENT_STATE_MISSING" + raise PorscheExceptionError(msg) + + data = dict(context.get("untrustedData", {}).get("submittedFormData") or {}) + data.update( + { + "state": transaction_state, + "action": "abort-passkey-enrollment", + "acul-sdk": "@auth0/auth0-acul-js@1.2.0", + }, + ) + + _LOGGER.debug("Declining passkey enrollment.") + resp = await self.client.post( + url, + data=data, + timeout=TIMEOUT, + headers=self.headers, + follow_redirects=False, + ) + if resp.status_code not in (302, 303): + msg = "PASSKEY_ENROLLMENT_SKIP_FAILED" + raise PorscheExceptionError(msg) + + return urljoin(url, resp.headers["Location"]) + + async def resume_authorization_code_flow(self, url: str) -> str: + """Follow the Auth0 redirect chain until the authorization code is returned. + + The resume URL used to redirect straight to the callback. Porsche now + interleaves optional screens - currently passkey enrollment - so the + chain has to be walked rather than read from a single Location header. + + :param url: resume URL returned by the Identifier First flow + :return: authorization code to be exchanged for an access token + """ + current_url = url + + for _ in range(_MAX_RESUME_REDIRECTS): + code = parse_qs(urlparse(current_url).query).get("code", [None])[0] + if code is not None: + return code + + resp = await self.client.get( + current_url, + timeout=TIMEOUT, + headers=self.headers, + follow_redirects=False, + ) + + if resp.status_code in (302, 303, 307, 308): + current_url = urljoin(str(resp.url), resp.headers["Location"]) + continue + + if resp.status_code == 200 and PASSKEY_ENROLLMENT_PATH in resp.url.path: + current_url = await self._skip_passkey_enrollment(str(resp.url), resp.text) + continue + + _LOGGER.error( + "Unexpected response %s at %s while resuming authorization.", + resp.status_code, + resp.url, + ) + msg = "Could not fetch authorization code" + raise PorscheExceptionError(msg) + + msg = "AUTHORIZATION_CODE_REDIRECT_LOOP" + raise PorscheExceptionError(msg) + async def login_with_identifier(self, state: str): """Log into the Identifier First flow. @@ -371,6 +499,8 @@ async def fetch_access_token(self, authorization_code): "code": authorization_code, "redirect_uri": REDIRECT_URI, } + if self.code_verifier is not None: + data["code_verifier"] = self.code_verifier try: _LOGGER.debug("Exchanging the authorization code for an access token.") @@ -384,7 +514,10 @@ async def fetch_access_token(self, authorization_code): resp.raise_for_status() return resp.json() except httpx.HTTPStatusError as exc: - raise PorscheExceptionError(exc.response.status_code) from exc + raise PorscheExceptionError( + exc.response.status_code, + response_body=exc.response.text[:1000] or None, + ) from exc async def refresh_token(self, refresh_token): """Use the provided refresh token to get a new access token. From 0a61e86755603a4f9f5df814a82be902ba043149 Mon Sep 17 00:00:00 2001 From: Fredrik Ljunggren Date: Sun, 30 Aug 2026 10:59:59 +0200 Subject: [PATCH 2/3] cleaned up comments --- pyporscheconnectapi/oauth2.py | 10 ---------- 1 file changed, 10 deletions(-) diff --git a/pyporscheconnectapi/oauth2.py b/pyporscheconnectapi/oauth2.py index c1eedeb..198204b 100644 --- a/pyporscheconnectapi/oauth2.py +++ b/pyporscheconnectapi/oauth2.py @@ -120,9 +120,6 @@ def __init__( self.captcha = captcha self.leeway = leeway self.headers = {"User-Agent": USER_AGENT, "X-Client-ID": X_CLIENT_ID} - # PKCE verifier for the authorization request currently in flight. It is - # created before /authorize and has to survive a captcha round trip, - # since the challenge is bound to that Auth0 transaction. self.code_verifier: str | None = None def _generate_pkce_verifier(self) -> str: @@ -210,9 +207,6 @@ async def fetch_authorization_code(self): else: try: if self.code_verifier is None: - # The challenge was bound to the Auth0 transaction created by the - # /authorize request that produced this captcha. Generating a new - # verifier here would yield a code that cannot be exchanged. msg = "PKCE_VERIFIER_MISSING_FOR_CAPTCHA_RESUME" raise PorscheExceptionError(msg) @@ -356,10 +350,6 @@ async def _skip_passkey_enrollment(self, url: str, html: str | None = None) -> s async def resume_authorization_code_flow(self, url: str) -> str: """Follow the Auth0 redirect chain until the authorization code is returned. - The resume URL used to redirect straight to the callback. Porsche now - interleaves optional screens - currently passkey enrollment - so the - chain has to be walked rather than read from a single Location header. - :param url: resume URL returned by the Identifier First flow :return: authorization code to be exchanged for an access token """ From fdba9d9ee4eeadae59efa13e209bf9d101fdcd9e Mon Sep 17 00:00:00 2001 From: Fredrik Ljunggren Date: Sun, 30 Aug 2026 17:30:58 +0200 Subject: [PATCH 3/3] carry the PKCE verifier across the captcha step --- pyporscheconnectapi/connection.py | 2 ++ pyporscheconnectapi/exceptions.py | 6 ++++-- pyporscheconnectapi/oauth2.py | 10 ++++++++-- 3 files changed, 14 insertions(+), 4 deletions(-) diff --git a/pyporscheconnectapi/connection.py b/pyporscheconnectapi/connection.py index 177425f..cc89b2a 100644 --- a/pyporscheconnectapi/connection.py +++ b/pyporscheconnectapi/connection.py @@ -59,6 +59,7 @@ def __init__( async_client=httpx.AsyncClient(), token=None, leeway: int = 60, + code_verifier: str | None = None, ) -> None: """Initialise the connection to the Porsche Connect API.""" if token is None: @@ -75,6 +76,7 @@ def __init__( Credentials(email, password), Captcha(captcha_code, state), leeway, + code_verifier=code_verifier, ) async def get_token(self): diff --git a/pyporscheconnectapi/exceptions.py b/pyporscheconnectapi/exceptions.py index 487582e..49bae97 100644 --- a/pyporscheconnectapi/exceptions.py +++ b/pyporscheconnectapi/exceptions.py @@ -56,13 +56,15 @@ class PorscheCaptchaRequiredError(PorscheExceptionError): captcha: str = None state: str = None + code_verifier: str = None - def __init__(self, captcha=None, state=None): + def __init__(self, captcha=None, state=None, code_verifier=None): """Initialize the captcha exception.""" if captcha is not None and state is not None: - _LOGGER.info("Initialising captcha exception: %s, %s", captcha, state) + _LOGGER.debug("Initialising captcha exception, state %s", state) self.captcha = captcha self.state = state + self.code_verifier = code_verifier super().__init__(captcha, state) diff --git a/pyporscheconnectapi/oauth2.py b/pyporscheconnectapi/oauth2.py index 198204b..e0dcd6a 100644 --- a/pyporscheconnectapi/oauth2.py +++ b/pyporscheconnectapi/oauth2.py @@ -113,6 +113,8 @@ def __init__( credentials: Credentials, captcha: Captcha, leeway: int = 60, + *, + code_verifier: str | None = None, ): """Initialise the oauth2 client.""" self.client = client @@ -120,7 +122,7 @@ def __init__( self.captcha = captcha self.leeway = leeway self.headers = {"User-Agent": USER_AGENT, "X-Client-ID": X_CLIENT_ID} - self.code_verifier: str | None = None + self.code_verifier: str | None = code_verifier def _generate_pkce_verifier(self) -> str: """Generate a PKCE code verifier (RFC 7636 section 4.1).""" @@ -441,7 +443,11 @@ async def login_with_identifier(self, state: str): raise PorscheExceptionError(msg) _LOGGER.debug("Parsed captcha image: %s...", str(captcha_img)[:100]) - raise PorscheCaptchaRequiredError(captcha=captcha_img, state=state) + raise PorscheCaptchaRequiredError( + captcha=captcha_img, + state=state, + code_verifier=self.code_verifier, + ) # 2. /u/login/password w/ password