diff --git a/volkswagencarnet/vw_connection.py b/volkswagencarnet/vw_connection.py index aac7fcc..944e8e1 100644 --- a/volkswagencarnet/vw_connection.py +++ b/volkswagencarnet/vw_connection.py @@ -4,10 +4,12 @@ from __future__ import annotations import asyncio +import base64 from datetime import UTC, datetime, timedelta import hashlib import logging from random import randint, random +import secrets from urllib.parse import parse_qs, urljoin, urlparse from typing import Dict, Optional @@ -138,26 +140,43 @@ class Connection: ) return await req.json() - async def get_authorization_page(self, authorization_endpoint: str) -> str: - """Get authorization page (login page).""" - # https://identity.vwgroup.io/oidc/v1/authorize?nonce={NONCE}&state={STATE}&response_type={TOKEN_TYPES}&scope={SCOPE}&redirect_uri={APP_URI}&client_id={CLIENT_ID} - # https://identity.vwgroup.io/oidc/v1/authorize?client_id={CLIENT_ID}&scope={SCOPE}&response_type={TOKEN_TYPES}&redirect_uri={APP_URI} + @staticmethod + def _generate_pkce_pair() -> tuple[str, str]: + """Return (code_verifier, code_challenge) for PKCE S256.""" + verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).rstrip(b"=").decode() + digest = hashlib.sha256(verifier.encode()).digest() + challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode() + return verifier, challenge + + async def get_authorization_page( + self, authorization_endpoint: str, extra_params: dict | None = None + ) -> tuple[str, str]: + """Get authorization page (login page). + + Returns: + (page_html, page_url) — the URL is needed as the POST target when + the form has no explicit action attribute. + """ _LOGGER.debug('Requesting authorization page from "%s"', authorization_endpoint) self._session_auth_headers.pop("Referer", None) self._session_auth_headers.pop("Origin", None) _LOGGER.debug('Request headers: "%s"', self._session_auth_headers) + params = { + "redirect_uri": APP_URI, + "response_type": CLIENT_TOKEN_TYPES, + "client_id": CLIENT_ID, + "scope": CLIENT_SCOPE, + } + if extra_params: + params.update(extra_params) + try: req = await self._session.get( url=authorization_endpoint, headers=self._session_auth_headers, allow_redirects=False, - params={ - "redirect_uri": APP_URI, - "response_type": CLIENT_TOKEN_TYPES, - "client_id": CLIENT_ID, - "scope": CLIENT_SCOPE, - }, + params=params, ) # Check if the response contains a redirect location @@ -177,15 +196,20 @@ class Connection: _LOGGER.info("Authorization error: %s", error_description) raise AuthenticationError(f"{error_msg}: {error_description}") - # If redirected, fetch the new location - req = await self._session.get( - url=ref, headers=self._session_auth_headers, allow_redirects=False - ) - - if req.status != 200: - raise AuthenticationError("Failed to fetch authorization endpoint") - - return await req.text() + # Follow redirects to the login page (may be more than one hop) + while True: + req = await self._session.get( + url=ref, headers=self._session_auth_headers, allow_redirects=False + ) + if req.status == 200: + return await req.text(), ref + if req.status in (301, 302, 303, 307, 308): + next_loc = req.headers.get("Location") + if not next_loc: + raise AuthenticationError("Missing Location header during login redirect chain") + ref = urljoin(ref, next_loc) + continue + raise AuthenticationError(f"Failed to fetch authorization endpoint: HTTP {req.status}") except Exception as e: _LOGGER.warning("Error during fetching authorization page: %s", str(e)) @@ -200,6 +224,26 @@ class Connection: return None return state_input["value"] + def extract_login_form_data(self, page_content: str) -> dict: + """Extract all hidden fields and submit button values from the login form.""" + soup = BeautifulSoup(page_content, "html.parser") + form = soup.select_one("form") + if not form: + return {} + form_data = {} + for inp in form.find_all("input"): + name = inp.get("name") + value = inp.get("value", "") + if name and inp.get("type") in ("hidden", None): + form_data[name] = value + # Include submit button value if it carries an action name (e.g. name="action" value="default") + for btn in form.find_all("button", type="submit"): + name = btn.get("name") + value = btn.get("value", "") + if name: + form_data[name] = value + return form_data + async def post_form( self, session, url: str, headers: dict, form_data: dict, redirect: bool = True ) -> str: @@ -227,7 +271,8 @@ class Connection: if error_code == "wrong-email-credentials": raise AuthenticationError("Wrong username or password") - # Unknown 400 error + # Unknown 400 error — log the response body to aid debugging + _LOGGER.debug("Login 400 response body: %s", page_content[:2000]) raise AuthenticationError( "Login form validation failed with unknown 400 error" ) @@ -303,10 +348,22 @@ class Connection: authorization_endpoint = openid_config["authorization_endpoint"] auth_issuer = openid_config["issuer"] - # Get authorization page - authorization_page = await self.get_authorization_page(authorization_endpoint) + # PKCE + nonce (nonce required when id_token appears in response_type) + code_verifier, code_challenge = self._generate_pkce_pair() + nonce = base64.urlsafe_b64encode(secrets.token_bytes(16)).rstrip(b"=").decode() + + # Get authorization page; returns (html, page_url) — use page_url as POST target + # because the form has no action attribute and implicitly submits to itself. + authorization_page, login_url = await self.get_authorization_page( + authorization_endpoint, + extra_params={ + "code_challenge": code_challenge, + "code_challenge_method": "S256", + "nonce": nonce, + }, + ) - # Extract form data + # Extract state token (still needed for the login URL) state_token = self.extract_state_token(authorization_page) if not state_token: @@ -316,13 +373,12 @@ class Connection: ) raise AuthenticationError("Invalid login page - missing state token") - # Do login - login_form = { - "username": self._session_auth_username, - "password": self._session_auth_password, - "state": state_token, - } - login_url = f"{auth_issuer}/u/login?state={state_token}" + # Build form from all hidden fields on the page so Auth0 gets every + # required field (e.g. csrf tokens, action), then overlay credentials. + login_form = self.extract_login_form_data(authorization_page) + login_form["username"] = self._session_auth_username + login_form["password"] = self._session_auth_password + login_form.setdefault("state", state_token) redirect_location = await self.post_form( self._session, @@ -337,36 +393,40 @@ class Connection: self._session, auth_issuer, redirect_location ) - jwt_auth_code = parse_qs(urlparse(redirect_response).query)["code"][0] - return jwt_auth_code + # Parse both query string and fragment — hybrid flow puts tokens in the fragment + all_params = { + **parse_qs(urlparse(redirect_response).query), + **parse_qs(urlparse(redirect_response).fragment), + } + jwt_auth_code = all_params["code"][0] + callback_id_token = all_params.get("id_token", [None])[0] + callback_access_token = all_params.get("access_token", [None])[0] + return jwt_auth_code, code_verifier, callback_id_token, callback_access_token async def _exchange_code_for_tokens( - self, auth_code: str, token_endpoint: str + self, auth_code: str, token_endpoint: str, code_verifier: str | None = None ) -> dict: - """Exchange authorization code for access tokens. - - Args: - auth_code: Authorization code from login flow - token_endpoint: Token endpoint URL - - Returns: - Dictionary containing tokens - - Raises: - AuthenticationError: If token exchange fails - """ + """Exchange authorization code for access tokens via the CARIAD BFF token endpoint.""" token_body = { "client_id": CLIENT_ID, "grant_type": "authorization_code", "code": auth_code, "redirect_uri": APP_URI, } + if code_verifier: + token_body["code_verifier"] = code_verifier + + token_headers = { + "Accept-Encoding": "gzip, deflate, br", + "Connection": "keep-alive", + "Content-Type": "application/x-www-form-urlencoded", + "User-Agent": USER_AGENT, + "x-android-package-name": ANDROID_PACKAGE_NAME, + } - # Token endpoint token_response = await self.post_form( - self._session, token_endpoint, self._session_auth_headers, token_body + self._session, token_endpoint, token_headers, token_body ) - return json_loads(token_response) async def _login(self) -> bool: @@ -385,11 +445,21 @@ class Connection: openid_config = await self.get_openid_config() token_endpoint = openid_config["token_endpoint"] - # Get authorization code - auth_code = await self._get_authorization_code(openid_config) - - # Exchange code for tokens - tokens = await self._exchange_code_for_tokens(auth_code, token_endpoint) + # Get authorization code; hybrid response_type also delivers tokens directly + auth_code, code_verifier, cb_id_token, cb_access_token = await self._get_authorization_code(openid_config) + + # Hybrid flow: tokens arrive in the callback URL — use them directly. + # The CARIAD BFF token endpoint requires a client_secret embedded in the + # official VW app, so we skip it when the callback already carries the tokens. + if cb_id_token and cb_access_token: + _LOGGER.debug("Using tokens from hybrid flow callback") + tokens = { + "access_token": cb_access_token, + "id_token": cb_id_token, + "token_type": "Bearer", + } + else: + tokens = await self._exchange_code_for_tokens(auth_code, token_endpoint, code_verifier) # Validate token structure required_keys = ["access_token", "id_token", "token_type"] diff --git a/volkswagencarnet/vw_const.py b/volkswagencarnet/vw_const.py index 17666a1..711fcf9 100644 --- a/volkswagencarnet/vw_const.py +++ b/volkswagencarnet/vw_const.py @@ -7,7 +7,7 @@ COUNTRY = "DE" # Data used in communication CLIENT_ID = "a24fba63-34b3-4d43-b181-942111e6bda8@apps_vw-dilab_com" CLIENT_SCOPE = "openid profile badge cars dealers vin" -CLIENT_TOKEN_TYPES = "code" +CLIENT_TOKEN_TYPES = "code id_token token" USER_AGENT = "Volkswagen/3.61.0-android/14" APP_URI = "weconnect://authenticated"