Add PKCE and handle the passkey enrollment screen in the auth flow

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):
@@ -118,7 +120,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)",

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."""
@@ -51,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)
 

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."""
@@ -107,13 +113,25 @@ def __init__(
         credentials: Credentials,
         captcha: Captcha,
         leeway: int = 60,
+        *,
+        code_verifier: str | None = None,
     ):
         """Initialise the oauth2 client."""
         self.client = client
         self.credentials = credentials
         self.captcha = captcha
         self.leeway = leeway
         self.headers = {"User-Agent": USER_AGENT, "X-Client-ID": X_CLIENT_ID}
+        self.code_verifier: str | None = code_verifier
+
+    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 +166,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 +175,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 +195,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 +208,14 @@ async def fetch_authorization_code(self):
 
         else:
             try:
+                if self.code_verifier is None:
+                    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 +291,103 @@ 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.
+
+        :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.
 
@@ -323,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
 
@@ -371,6 +495,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 +510,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.