From 458c360aff90b8e1285ffe24a90c37e52673ec1b Mon Sep 17 00:00:00 2001 From: rleidner Date: Sun, 31 May 2026 19:04:51 +0200 Subject: [PATCH 1/2] soc_vwid: use OIDC hybrid login flow --- packages/modules/vehicles/vwid/libvwid.py | 1016 ++++++++++++++------- 1 file changed, 664 insertions(+), 352 deletions(-) diff --git a/packages/modules/vehicles/vwid/libvwid.py b/packages/modules/vehicles/vwid/libvwid.py index 2018906f56..03a41fdad4 100755 --- a/packages/modules/vehicles/vwid/libvwid.py +++ b/packages/modules/vehicles/vwid/libvwid.py @@ -8,14 +8,14 @@ import hashlib from json import dumps as to_json from json import loads +import uuid import logging from random import randint, random -import re from urllib.parse import parse_qs, urljoin, urlparse -import aiohttp -# from aiohttp.hdrs import METH_GET, METH_POST, METH_PUT -import bs4 +from aiohttp import ClientSession, ClientTimeout, client_exceptions +from aiohttp.hdrs import METH_GET, METH_POST, METH_PUT +from bs4 import BeautifulSoup import jwt ANDROID_PACKAGE_NAME = "com.volkswagen.weconnect" @@ -24,14 +24,8 @@ BASE_AUTH = "https://identity.vwgroup.io" BASE_SESSION = "https://msg.volkswagen.de" BRAND = "VW" -USER_AGENT = "Volkswagen/3.51.1-android/14" -CLIENT = { - "Legacy": { - "CLIENT_ID": "a24fba63-34b3-4d43-b181-942111e6bda8@apps_vw-dilab_com", - "SCOPE": "openid profile badge cars dealers vin", - "TOKEN_TYPES": "code", - } -} +USER_AGENT = "Volkswagen/3.61.0-android/14" +CLIENT_ID = "a24fba63-34b3-4d43-b181-942111e6bda8@apps_vw-dilab_com" COUNTRY = "DE" HEADERS_SESSION = { "Connection": "keep-alive", @@ -39,7 +33,8 @@ "Accept-charset": "UTF-8", "Accept": "application/json", "User-Agent": USER_AGENT, - "tokentype": "IDK_TECHNICAL", + "x-platform": "android", + "x-assertion": "0", "x-android-package-name": ANDROID_PACKAGE_NAME, } HEADERS_AUTH = { @@ -48,12 +43,62 @@ "image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9", "Accept-Encoding": "gzip, deflate", "Content-Type": "application/x-www-form-urlencoded", + "User-Agent": USER_AGENT, + "x-platform": "android", + "x-assertion": "0", "x-android-package-name": ANDROID_PACKAGE_NAME, } MAX_RETRIES_ON_RATE_LIMIT = 3 TIMEOUT = timedelta(seconds=30) JWT_ALGORITHMS = ["RS256"] +CLIENT_TOKEN_TYPES = "code" +CLIENT_SCOPE = "openid profile badge cars dealers vin offline_access" + + +"""Custom exceptions for volkswagencarnet.""" + + +class VWError(Exception): + """Base exception for VW CarNet errors.""" + + pass + + +class AuthenticationError(VWError): + """Authentication failed.""" + + pass + + +class APIError(VWError): + """API request failed.""" + + pass + + +class SPINError(VWError): + """S-PIN related error.""" + + pass + + +class RedirectError(VWError): + """Redirect handling failed.""" + + pass + + +class RequestError(VWError): + """Request execution failed.""" + + pass + + +class TermsAndConditionsError(AuthenticationError): + """Terms and Conditions need to be accepted.""" + + pass class Services: @@ -482,23 +527,14 @@ def __init__( session, username, password, - fulldebug=False, country=COUNTRY, interval=timedelta(minutes=5), ) -> None: """Initialize.""" - self._x_client_id = None self._session = session - self._session_fulldebug = fulldebug self._session_headers = HEADERS_SESSION.copy() - self._session_base = BASE_SESSION self._session_auth_headers = HEADERS_AUTH.copy() - self._session_auth_base = BASE_AUTH self._session_refresh_interval = interval - - no_vin_key = "" - self._session_auth_ref_urls = {no_vin_key: BASE_SESSION} - self._session_spin_ref_urls = {no_vin_key: BASE_SESSION} self._session_logged_in = False self._session_first_update = False self._session_auth_username = username @@ -508,24 +544,21 @@ def __init__( self._vehicles = [] - _LOGGER.debug("Using service %s", self._session_base) - - self._jarCookie = "" - self._state = {} + self._jarCookie = None self._service_status = {} def _clear_cookies(self): - self._session._cookie_jar._cookies.clear() + self._session._cookie_jar._cookies.clear() # pylint: disable=protected-access # API Login - async def doLogin(self, tries: int = 3): + async def doLogin(self, tries: int = 1): """Login method, clean login.""" async with self._login_lock: _LOGGER.debug("Initiating new login") for i in range(tries): - self._session_logged_in = await self._login("Legacy") + self._session_logged_in = await self._login() if self._session_logged_in: break if i > tries: @@ -536,8 +569,7 @@ async def doLogin(self, tries: int = 3): if not self._session_logged_in: return False - _LOGGER.debug("Successfully logged in") - self._session_tokens["identity"] = self._session_tokens["Legacy"].copy() + _LOGGER.info("Successfully logged in") # Get list of vehicles from account _LOGGER.debug("Fetching vehicles associated with account") @@ -558,282 +590,315 @@ async def doLogin(self, tries: int = 3): await self.update() return True - async def get_openid_config(self): + async def get_openid_config(self) -> dict[str, str]: """Get OpenID config.""" - if self._session_fulldebug: - _LOGGER.debug("Requesting openid config") + _LOGGER.debug("Requesting openid config") req = await self._session.get( - url=f"{BASE_API}/login/v1/idk/openid-configuration" + url=f"{BASE_API}/auth/v1/idk/oidc/openid-configuration" ) if req.status != 200: _LOGGER.error("Failed to get OpenID configuration, status: %s", req.status) - raise Exception("OpenID configuration error") - return await req.json() - - async def get_authorization_page(self, authorization_endpoint, client): - """Get authorization page (login page).""" - if self._session_fulldebug: - _LOGGER.debug( - 'Requesting authorization page from "%s"', authorization_endpoint + raise AuthenticationError( + f"OpenID configuration error: status {req.status}" ) - self._session_auth_headers.pop("Referer", None) - self._session_auth_headers.pop("Origin", None) - _LOGGER.debug('Request headers: "%s"', self._session_auth_headers) + config = await req.json() + _LOGGER.debug("OpenID config: %s", config) + return config - 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[client].get("TOKEN_TYPES"), - "client_id": CLIENT[client].get("CLIENT_ID"), - "scope": CLIENT[client].get("SCOPE"), - }, - ) + async def get_authorization_page(self, authorization_endpoint: str) -> str: + """Fetch the Auth0 Universal Login page for credential submission. - # Check if the response contains a redirect location - location = req.headers.get("Location") - if not location: - raise Exception( - f"Missing 'Location' header, payload returned: {await req.content.read()}" - ) - - ref = urljoin(authorization_endpoint, location) - if "error" in ref: - parsed_query = parse_qs(urlparse(ref).query) - error_msg = parsed_query.get("error", ["Unknown error"])[0] - error_description = parsed_query.get( - "error_description", ["No description"] - )[0] - _LOGGER.exception("Authorization error: %s", error_description) - raise Exception(error_msg) - - # If redirected, fetch the new location - req = await self._session.get( - url=ref, headers=self._session_auth_headers, allow_redirects=False - ) + Hits the OIDC authorization endpoint with response_type=code id_token token + (hybrid flow). Auth0 redirects to the login form; we follow that single + redirect and return the HTML so the caller can extract the state token. + """ + _LOGGER.debug('Requesting authorization page from "%s"', authorization_endpoint) + self._session_auth_headers.pop("Referer", None) + self._session_auth_headers.pop("Origin", None) + + params = { + "redirect_uri": APP_URI, + # Hybrid flow: Auth0 returns code + id_token + access_token in the + # callback so no separate token exchange with the CARIAD BFF is needed. + "response_type": "code id_token token", + "client_id": CLIENT_ID, + "scope": CLIENT_SCOPE, + "nonce": uuid.uuid4().hex, + } - if req.status != 200: - raise Exception("Failed to fetch authorization endpoint") + req = await self._session.get( + url=authorization_endpoint, + headers=self._session_auth_headers, + allow_redirects=False, + params=params, + ) - return await req.text() + location = req.headers.get("Location") + if not location: + raise AuthenticationError( + f"Missing 'Location' header in authorization response. " + f"Status: {req.status}" + ) - except Exception as e: - _LOGGER.warning("Error during fetching authorization page: %s", str(e)) - raise + ref = urljoin(authorization_endpoint, location) + if "error" in ref: + parsed_query = parse_qs(urlparse(ref).query) + error_msg = parsed_query.get("error", ["Unknown error"])[0] + error_description = parsed_query.get( + "error_description", ["No description"] + )[0] + _LOGGER.info("Authorization error: %s", error_description) + raise AuthenticationError(f"{error_msg}: {error_description}") + + # Follow the redirect to the actual login page + req = await self._session.get( + url=ref, headers=self._session_auth_headers, allow_redirects=False + ) + if req.status != 200: + raise AuthenticationError(f"Failed to fetch login page (HTTP {req.status})") - def extract_form_data(self, page_content, form_id): - """Extract form data from a page.""" - soup = bs4.BeautifulSoup(page_content, "html.parser") - form = soup.find("form", id=form_id) - if form is None: - _LOGGER.debug(f"Form with ID '{form_id}' not found.") - return None - return { - input_field["name"]: input_field["value"] - for input_field in form.find_all("input", type="hidden") - } + return await req.text() - def extract_state_token(self, page_content): + def extract_state_token(self, page_content: str) -> str | None: """Extract state token from a page.""" - soup = bs4.BeautifulSoup(page_content, "html.parser") + soup = BeautifulSoup(page_content, "html.parser") state_input = soup.select_one('input[name="state"]') if not state_input or not state_input.get("value"): _LOGGER.debug("State token not found.") return None return state_input["value"] - def extract_password_form_data(self, soup): - """Extract password form data from a page.""" - pw_form = {} - for script in soup.find_all("script"): - if "src" in script.attrs or not script.string: - continue - script_text = script.string - - if "window._IDK" not in script_text: - continue # Skip scripts that don't contain relevant data - if re.match('"errorCode":"', script_text): - raise Exception("Error code found in script data.") - - pw_form["relayState"] = re.search( - '"relayState":"([a-f0-9]*)"', script_text - )[1] - pw_form["hmac"] = re.search('"hmac":"([a-f0-9]*)"', script_text)[1] - pw_form["email"] = re.search('"email":"([^"]*)"', script_text)[1] - pw_form["_csrf"] = re.search("csrf_token:\\s*'([^\"']*)'", script_text)[1] - - post_action = re.search('"postAction":\\s*"([^"\']*)"', script_text)[1] - client_id = re.search('"clientId":\\s*"([^"\']*)"', script_text)[1] - return pw_form, post_action, client_id - - raise Exception("Password form data not found in script.") - - async def post_form(self, session, url, headers, form_data, redirect=True): + async def post_form( + self, session, url: str, headers: dict, form_data: dict, redirect: bool = True + ) -> str: """Post a form and check for success.""" req = await session.post( url, headers=headers, data=form_data, allow_redirects=redirect ) + + # Redirect case if not redirect and req.status == 302: - return req.headers["Location"] - if req.status != 200: - raise Exception("Form POST request failed.") - return await req.text() + return req.headers.get("Location") + + # Handle explicit error 400 (form validation failure) + if req.status == 400: + page_content = await req.text() + soup = BeautifulSoup(page_content, "html.parser") + + # Try both username + password fields in one pass + for field_id in ("error-element-username", "error-element-password"): + span = soup.select_one(f'span[id="{field_id}"]') + if not span: + continue + + error_code = span.get("data-error-code") + if error_code == "wrong-email-credentials": + raise AuthenticationError("Wrong username or password") + + # Unknown 400 error + raise AuthenticationError( + "Login form validation failed with unknown 400 error" + ) + + # Any unexpected HTTP code + if req.status not in (200, 400): + raise RequestError( + f"Login form submission failed with HTTP {req.status}. " + "This might indicate incorrect credentials or a temporary service issue." + ) - async def handle_login_with_password(self, session, url, auth_headers, form_data): - """Handle login with email and password.""" - return await self.post_form(session, url, auth_headers, form_data, False) + # Normal success path + return await req.text() - async def follow_redirects(self, session, pw_url, redirect_location): + async def follow_redirects( + self, session, pw_url: str, redirect_location: str + ) -> str: """Handle redirects.""" ref = urljoin(pw_url, redirect_location) max_depth = 10 while not ref.startswith(APP_URI): if max_depth == 0: - raise Exception("Too many redirects") + raise RedirectError( + f"Too many redirects during login flow (max depth: {max_depth}). " + "This might indicate an authentication loop." + ) response = await session.get( url=ref, headers=self._session_auth_headers, allow_redirects=False ) + + # Check if we hit a terms and conditions page (HTTP 200 with no redirect) + if response.status == 200 and "Location" not in response.headers: + page_content = await response.text() + if ( + "termsAndConditions" in page_content + or '"page":"termsAndConditions"' in page_content + ): + _LOGGER.error( + "Terms and Conditions acceptance required. " + "Please log in to https://www.myvolkswagen.net/ and accept the updated terms." + ) + raise TermsAndConditionsError( + "Terms and Conditions must be accepted. " + "Please visit https://www.myvolkswagen.net/ to accept the updated terms and conditions, " + "then try logging in again." + ) + if "Location" not in response.headers: _LOGGER.warning("Failed to find next redirect location") - raise Exception("Redirect error") + raise RedirectError("Failed to find next redirect location") ref = urljoin(ref, response.headers["Location"]) max_depth -= 1 return ref - async def _login(self, client="Legacy"): - """Login function.""" + async def _get_authorization_code(self, openid_config: dict) -> tuple: + """Run the OIDC hybrid login flow and return the callback tokens. + + Uses response_type=code id_token token so identity.vwgroup.io (Auth0) + delivers access_token and id_token directly in the callback — these are + usable with the CARIAD BFF without a separate token exchange step. + + Returns: + Tuple of (auth_code_jwt, id_token, access_token) + """ + authorization_endpoint = openid_config["authorization_endpoint"] + auth_issuer = openid_config["issuer"] + + authorization_page = await self.get_authorization_page(authorization_endpoint) + + state_token = self.extract_state_token(authorization_page) + if not state_token: + _LOGGER.error( + "Unable to find valid login page. " + "Try logging in to the portal: https://www.myvolkswagen.net/" + ) + raise AuthenticationError("Invalid login page - missing state token") + + login_form = { + "username": self._session_auth_username, + "password": self._session_auth_password, + "state": state_token, + "action": "default", + } + login_url = f"{auth_issuer}/u/login?state={state_token}" + + redirect_location = await self.post_form( + self._session, + login_url, + self._session_auth_headers, + login_form, + False, + ) + + callback_url = await self.follow_redirects( + self._session, auth_issuer, redirect_location + ) + + parsed = urlparse(callback_url) + all_params = {**parse_qs(parsed.query), **parse_qs(parsed.fragment)} + _LOGGER.debug("Callback params keys: %s", list(all_params.keys())) + + auth_code = all_params.get("code", [None])[0] + id_token = all_params.get("id_token", [None])[0] + access_token = all_params.get("access_token", [None])[0] + + if not auth_code: + raise AuthenticationError("No authorization code in callback URL") + + return auth_code, id_token, access_token + + def _build_session_tokens( + self, auth_code: str, id_token: str, access_token: str + ) -> dict: + """Build the session token dict from the hybrid OIDC callback values. + + The hybrid flow (response_type=code id_token token) already delivers + access_token and id_token directly from identity.vwgroup.io. These + Auth0-issued tokens are valid for the CARIAD BFF, which validates them + against Auth0's public keys. No separate token exchange is required. + + Note: the authorization code (auth_code) is not used for a server-side + exchange — it is decoded here only for debug logging. + """ + try: + payload = jwt.decode(auth_code, options={"verify_signature": False}) + _LOGGER.debug("Auth code JWT payload: %s", payload) + except Exception: + _LOGGER.debug("Auth code is not a JWT, continuing without decode") + + return { + "access_token": access_token, + "id_token": id_token, + "token_type": "Bearer", + } + + async def _login(self) -> bool: + """Login function. + Returns: + True if login successful, False otherwise + """ try: # Clear cookies and reset headers self._clear_cookies() self._session_headers = HEADERS_SESSION.copy() self._session_auth_headers = HEADERS_AUTH.copy() - # Get OpenID configuration + # Get OpenID configuration (authorization_endpoint, issuer) openid_config = await self.get_openid_config() - authorization_endpoint = openid_config["authorization_endpoint"] - token_endpoint = openid_config["token_endpoint"] - auth_issuer = openid_config["issuer"] - # Get authorization page - authorization_page = await self.get_authorization_page( - authorization_endpoint, client + # Get authorization code and hybrid tokens from login flow + auth_code, id_token, access_token = await self._get_authorization_code( + openid_config ) - # Extract form data - mailform = self.extract_form_data(authorization_page, "emailPasswordForm") - state_token = self.extract_state_token(authorization_page) - if mailform: - _LOGGER.debug("Legacy authentication found., client=" + str(client)) - mailform["email"] = self._session_auth_username - pe_url = auth_issuer + bs4.BeautifulSoup( - authorization_page, "html.parser" - ).find("form", id="emailPasswordForm").get("action") - - # POST email - # https://identity.vwgroup.io/signin-service/v1/{CLIENT_ID}/login/identifier - self._session_auth_headers["Referer"] = authorization_endpoint - self._session_auth_headers["Origin"] = auth_issuer - response_text = await self.post_form( - self._session, pe_url, self._session_auth_headers, mailform - ) - - # Extract password form data - response_soup = bs4.BeautifulSoup(response_text, "html.parser") - pw_form, post_action, client_id = self.extract_password_form_data( - response_soup - ) - - # Add password to form data - pw_form["password"] = self._session_auth_password - pw_url = f"{auth_issuer}/signin-service/v1/{client_id}/{post_action}" - - # POST password - self._session_auth_headers["Referer"] = pe_url - redirect_location = await self.handle_login_with_password( - self._session, pw_url, self._session_auth_headers, pw_form - ) - - # Handle redirects and extract tokens - redirect_response = await self.follow_redirects( - self._session, pw_url, redirect_location - ) - jwt_auth_code = parse_qs(urlparse(redirect_response).query)["code"][0] - elif state_token: - _LOGGER.debug( - "Legacy authentication not found. Trying new authentication flow., client=" + str(client) - ) - - # Do login - login_form = {} - login_form["username"] = self._session_auth_username - login_form["password"] = self._session_auth_password - login_form["state"] = state_token - login_url = f"{auth_issuer}/u/login?state={state_token}" - - redirect_location = await self.post_form( - self._session, - login_url, - self._session_auth_headers, - login_form, - False, - ) + # Build session tokens from hybrid flow response (no server-side exchange needed) + tokens = self._build_session_tokens(auth_code, id_token, access_token) - # Handle redirects and extract tokens - redirect_response = await self.follow_redirects( - self._session, auth_issuer, redirect_location - ) - jwt_auth_code = parse_qs(urlparse(redirect_response).query)["code"][0] - else: + # Validate token structure + required_keys = ["access_token", "id_token", "token_type"] + if not all(key in tokens for key in required_keys): _LOGGER.error( - "Unable to find valid login page." - "Try logging in to the portal: https://www.myvolkswagen.net/" + "Invalid token response. Missing required keys. Got: %s", + list(tokens.keys()), ) + self._session_logged_in = False return False - # Exchange authorization code for tokens - token_body = { - "client_id": CLIENT[client].get("CLIENT_ID"), - "grant_type": "authorization_code", - "code": jwt_auth_code, - "redirect_uri": APP_URI, - } + # Store directly as "identity" + self._session_tokens["identity"] = tokens - # Token endpoint - token_response = await self.post_form( - self._session, token_endpoint, self._session_auth_headers, token_body + # Update authorization header + self._session_headers["Authorization"] = ( + "Bearer " + self._session_tokens["identity"]["access_token"] ) - # Store session tokens - self._session_tokens[client] = json_loads(token_response) - - # Verify tokens - if not await self.verify_tokens( - self._session_tokens[client].get("id_token", ""), "identity" - ): - _LOGGER.warning("User identity token could not be verified!") - else: - _LOGGER.debug("User identity token verified successfully, client=" + str(client)) + _LOGGER.debug("Successfully stored authentication tokens") # Mark session as logged in self._session_logged_in = True + return True + except (AuthenticationError, RequestError, RedirectError) as error: + _LOGGER.error("Authentication error during login: %s", error) + self._session_logged_in = False + return False + except client_exceptions.ClientError as error: + _LOGGER.error("Network error during login: %s", error) + self._session_logged_in = False + return False + except KeyError as error: + _LOGGER.error("Missing required data during login: %s", error) + self._session_logged_in = False + return False except Exception as error: - _LOGGER.error("Login failed: %s", error) + _LOGGER.error("Unexpected error during login: %s", error) self._session_logged_in = False return False - self._session_headers["Authorization"] = ( - "Bearer " + self._session_tokens[client]["access_token"] - ) - return True async def _handle_action_result(self, response_raw): response = await response_raw.json(loads=json_loads) if not response: - raise Exception("Invalid or no response") + raise APIError("Invalid or no response from action endpoint") if response == 429: return {"id": None, "state": "Throttled"} request_id = response.get("data", {}).get("requestID", 0) @@ -842,24 +907,18 @@ async def _handle_action_result(self, response_raw): async def terminate(self): """Log out from connect services.""" - _LOGGER.debug("Initiating logout") + _LOGGER.info("Initiating logout") await self.logout() async def logout(self): """Logout, revoke tokens.""" - _LOGGER.debug("logout - revoke token") self._session_headers.pop("Authorization", None) if self._session_logged_in: - if self._session_headers.get("identity", {}).get("identity_token"): - _LOGGER.debug("Revoking Identity Access Token") - - if self._session_headers.get("identity", {}).get("refresh_token"): - _LOGGER.debug("Revoking Identity Refresh Token") + if self._session_tokens.get("identity", {}).get("refresh_token"): + _LOGGER.info("Revoking Identity Refresh Token") params = {"token": self._session_tokens["identity"]["refresh_token"]} - await self.post( - "https://emea.bff.cariad.digital/login/v1/idk/revoke", data=params - ) + await self.post(f"{BASE_API}/auth/v1/idk/oidc/revoke", data=params) # HTTP methods to API async def _request(self, method, url, return_raw=False, **kwargs): @@ -872,7 +931,7 @@ async def _request(self, method, url, return_raw=False, **kwargs): method, url, headers=self._session_headers, - timeout=aiohttp.ClientTimeout(total=TIMEOUT.seconds), + timeout=ClientTimeout(total=TIMEOUT.seconds), cookies=self._jarCookie, raise_for_status=False, **kwargs, @@ -880,7 +939,7 @@ async def _request(self, method, url, return_raw=False, **kwargs): response.raise_for_status() # Update cookie jar - if self._jarCookie != "": + if self._jarCookie is not None: self._jarCookie.update(response.cookies) else: self._jarCookie = response.cookies @@ -894,7 +953,7 @@ async def _request(self, method, url, return_raw=False, **kwargs): res = response else: res = {"status_code": response.status} - elif response.status >= 200 or response.status <= 300: + elif 200 <= response.status < 300: res = await response.json(loads=json_loads) else: res = {} @@ -903,7 +962,7 @@ async def _request(self, method, url, return_raw=False, **kwargs): response.status, response.text, ) - except Exception: + except Exception: # pylint: disable=broad-exception-caught res = {} _LOGGER.debug( "Something went wrong [%s] response: %s", @@ -914,27 +973,20 @@ async def _request(self, method, url, return_raw=False, **kwargs): return response return res - if self._session_fulldebug: - _LOGGER.debug( - 'Request for "%s" returned with status code [%s], headers: %s, response: %s', - url, - response.status, - response.headers, - res, - ) - else: - _LOGGER.debug( - 'Request for "%s" returned with status code [%s]', - url, - response.status, - ) + _LOGGER.debug( + 'Request for "%s" returned with status code [%s], headers: %s, response: %s', + url, + response.status, + response.headers, + res, + ) if return_raw: res = response return res - except aiohttp.client_exceptions.ClientResponseError as httperror: + except client_exceptions.ClientResponseError as httperror: # Update service status - await self.update_service_status(url, httperror.code) + await self.update_service_status(url, httperror.status) raise httperror from None except Exception as error: # Update service status @@ -944,8 +996,8 @@ async def _request(self, method, url, return_raw=False, **kwargs): async def get(self, url, vin="", tries=0): """Perform a get query.""" try: - return await self._request(aiohttp.hdrs.METH_GET, url) - except aiohttp.client_exceptions.ClientResponseError as error: + return await self._request(METH_GET, url) + except client_exceptions.ClientResponseError as error: if error.status == 400: _LOGGER.error( 'Got HTTP 400 "Bad Request" from server, this request might be malformed or not implemented' @@ -964,11 +1016,11 @@ async def get(self, url, vin="", tries=0): await asyncio.sleep(delay) return await self.get(url, vin, tries + 1) elif error.status == 500: - _LOGGER.warning( + _LOGGER.debug( "Got HTTP 500 from server, service might be temporarily unavailable" ) elif error.status == 502: - _LOGGER.warning( + _LOGGER.debug( "Got HTTP 502 from server, this request might not be supported for this vehicle" ) else: @@ -980,10 +1032,10 @@ async def post(self, url, vin="", tries=0, return_raw=False, **data): try: if data: return await self._request( - aiohttp.hdrs.METH_POST, url, return_raw=return_raw, **data + METH_POST, url, return_raw=return_raw, **data ) - return await self._request(aiohttp.hdrs.METH_POST, url, return_raw=return_raw) - except aiohttp.client_exceptions.ClientResponseError as error: + return await self._request(METH_POST, url, return_raw=return_raw) + except client_exceptions.ClientResponseError as error: if error.status == 429 and tries < MAX_RETRIES_ON_RATE_LIMIT: delay = randint(1, 3 + tries * 2) _LOGGER.debug( @@ -999,9 +1051,9 @@ async def put(self, url, vin="", tries=0, return_raw=False, **data): """Perform a put query.""" try: if data: - return await self._request(aiohttp.hdrs.METH_PUT, url, return_raw=return_raw, **data) - return await self._request(aiohttp.hdrs.METH_PUT, url, return_raw=return_raw) - except aiohttp.client_exceptions.ClientResponseError as error: + return await self._request(METH_PUT, url, return_raw=return_raw, **data) + return await self._request(METH_PUT, url, return_raw=return_raw) + except client_exceptions.ClientResponseError as error: if error.status == 429 and tries < MAX_RETRIES_ON_RATE_LIMIT: delay = randint(1, 3 + tries * 2) _LOGGER.debug( @@ -1021,13 +1073,13 @@ async def update(self): _LOGGER.warning("Login for %s account failed!", BRAND) return False try: - if not await self.validate_tokens: - _LOGGER.warning( + if not await self.validate_tokens(): + _LOGGER.info( "Session expired. Initiating new login for %s account", BRAND ) if not await self.doLogin(): _LOGGER.warning("Login for %s account failed!", BRAND) - raise Exception(f"Login for {BRAND} account failed") + raise AuthenticationError(f"Login for {BRAND} account failed") else: _LOGGER.debug("Going to call vehicle updates") # Get all Vehicle objects and update in parallell @@ -1036,13 +1088,13 @@ async def update(self): await asyncio.gather(*updatelist) return True - except (OSError, LookupError, Exception) as error: + except (OSError, LookupError, Exception) as error: # pylint: disable=broad-exception-caught _LOGGER.warning("Could not update information: %s", error) return False async def getPendingRequests(self, vin): """Get status information for pending requests.""" - if not await self.validate_tokens: + if not await self.validate_tokens(): return False try: response = await self.get( @@ -1053,7 +1105,7 @@ async def getPendingRequests(self, vin): response["refreshTimestamp"] = datetime.now(UTC) return response - except Exception as error: + except Exception as error: # pylint: disable=broad-exception-caught _LOGGER.warning( "Could not fetch information for pending requests, error: %s", error ) @@ -1061,7 +1113,7 @@ async def getPendingRequests(self, vin): async def getOperationList(self, vin): """Collect operationlist for VIN, supported/licensed functions.""" - if not await self.validate_tokens: + if not await self.validate_tokens(): return False try: response = await self.get( @@ -1076,16 +1128,16 @@ async def getOperationList(self, vin): ) data = response else: - _LOGGER.warning("Could not fetch operation list: %s", response) + _LOGGER.info("Could not fetch operation list: %s", response) data = {"error": "unknown"} - except Exception as error: + except Exception as error: # pylint: disable=broad-exception-caught _LOGGER.warning("Could not fetch operation list, error: %s", error) data = {"error": "unknown"} return data async def getSelectiveStatus(self, vin, services): """Get status information for specified services.""" - if not await self.validate_tokens: + if not await self.validate_tokens(): return False try: response = await self.get( @@ -1096,8 +1148,8 @@ async def getSelectiveStatus(self, vin, services): for service in services: if not response.get(service): _LOGGER.debug( - "Did not receive return data for requested service %s." - " (This is expected for several service/car combinations)", + "Did not receive return data for requested service %s.\ + (This is expected for several service/car combinations)", service, ) @@ -1105,13 +1157,13 @@ async def getSelectiveStatus(self, vin, services): response.update({"refreshTimestamp": datetime.now(UTC)}) return response - except Exception as error: + except Exception as error: # pylint: disable=broad-exception-caught _LOGGER.warning("Could not fetch selectivestatus, error: %s", error) return False async def getVehicleData(self, vin): """Get car information like VIN, nickname, etc.""" - if not await self.validate_tokens: + if not await self.validate_tokens(): return False try: response = await self.get(f"{BASE_API}/vehicle/v2/vehicles", "") @@ -1122,13 +1174,109 @@ async def getVehicleData(self, vin): _LOGGER.warning("Could not fetch vehicle data for vin %s", vin) - except Exception as error: + except Exception as error: # pylint: disable=broad-exception-caught _LOGGER.warning("Could not fetch vehicle data, error: %s", error) return False + async def getParkingPosition(self, vin): + """Get information about the parking position.""" + if not await self.validate_tokens(): + return False + try: + response = await self.get( + f"{BASE_API}/vehicle/v1/vehicles/{vin}/parkingposition", "" + ) + + if "data" in response: + return {"isMoving": False, "parkingposition": response["data"]} + if response.get("status_code", {}): + if response.get("status_code", 0) == 204: + _LOGGER.debug( + "Seems car is moving, HTTP 204 received from parkingposition" + ) + return {"isMoving": True, "parkingposition": {}} + + _LOGGER.warning( + "Could not fetch parkingposition, HTTP status code: %s", + response.get("status_code"), + ) + else: + _LOGGER.info( + "Unhandled error while trying to fetch parkingposition data" + ) + except Exception as error: # pylint: disable=broad-exception-caught + _LOGGER.warning("Could not fetch parkingposition, error: %s", error) + return False + + async def getTripLast(self, vin): + """Get car information like VIN, nickname, etc.""" + if not await self.validate_tokens(): + return False + try: + response = await self.get( + f"{BASE_API}/vehicle/v1/trips/{vin}/shortterm/last", "" + ) + if "data" in response: + return {"trip_last": response["data"]} + + if response.get("status_code", 0) in [404, 502]: + _LOGGER.debug("No last trip data available for this vehicle") + else: + _LOGGER.warning( + "Could not fetch last trip data, server response: %s", response + ) + + except Exception as error: # pylint: disable=broad-exception-caught + _LOGGER.warning("Could not fetch last trip data, error: %s", error) + return False + + async def getTripRefuel(self, vin): + """Get information about the trip since last refuel""" + if not await self.validate_tokens(): + return False + try: + response = await self.get( + f"{BASE_API}/vehicle/v1/trips/{vin}/cyclic/last", "" + ) + if "data" in response: + return {"trip_refuel": response["data"]} + + if response.get("status_code", 0) in [404, 502]: + _LOGGER.debug("No refuel trip data available for this vehicle") + else: + _LOGGER.warning( + "Could not fetch refuel trip data, server response: %s", response + ) + + except Exception as error: # pylint: disable=broad-exception-caught + _LOGGER.warning("Could not fetch last trip data, error: %s", error) + return False + + async def getTripLongterm(self, vin): + """Get information about the trip last longterm""" + if not await self.validate_tokens(): + return False + try: + response = await self.get( + f"{BASE_API}/vehicle/v1/trips/{vin}/longterm/last", "" + ) + if "data" in response: + return {"trip_longterm": response["data"]} + + if response.get("status_code", 0) in [404, 502]: + _LOGGER.debug("No longterm trip data available for this vehicle") + else: + _LOGGER.warning( + "Could not fetch longterm trip data, server response: %s", response + ) + + except Exception as error: # pylint: disable=broad-exception-caught + _LOGGER.warning("Could not fetch last trip data, error: %s", error) + return False + async def wakeUpVehicle(self, vin): """Wake up vehicle to send updated data to VW Backend.""" - if not await self.validate_tokens: + if not await self.validate_tokens(): return False try: return await self.post( @@ -1137,7 +1285,7 @@ async def wakeUpVehicle(self, vin): return_raw=True, ) - except Exception as error: + except Exception as error: # pylint: disable=broad-exception-caught _LOGGER.warning("Could not refresh the data, error: %s", error) return False @@ -1146,15 +1294,15 @@ async def get_request_status(self, vin, requestId, actionId=""): if self.logged_in is False: if not await self.doLogin(): _LOGGER.warning("Login for %s account failed!", BRAND) - raise Exception(f"Login for {BRAND} account failed") + raise AuthenticationError(f"Login for {BRAND} account failed") try: - if not await self.validate_tokens: - _LOGGER.warning( + if not await self.validate_tokens(): + _LOGGER.info( "Session expired. Initiating new login for %s account", BRAND ) if not await self.doLogin(): _LOGGER.warning("Login for %s account failed!", BRAND) - raise Exception(f"Login for {BRAND} account failed") + raise AuthenticationError(f"Login for {BRAND} account failed") response = await self.getPendingRequests(vin) @@ -1179,7 +1327,7 @@ async def get_request_status(self, vin, requestId, actionId=""): status = result except Exception as error: _LOGGER.warning("Failure during get request status: %s", error) - raise Exception(f"Failure during get request status: {error}") from error + raise RequestError(f"Failure during get request status: {error}") from error else: return status @@ -1188,10 +1336,10 @@ async def check_spin_state(self): result = await self.get(f"{BASE_API}/vehicle/v1/spin/state") remainingTries = result.get("remainingTries", None) if remainingTries is None: - raise Exception("Couldn't determine S-PIN state.") + raise SPINError("Couldn't determine S-PIN state") if remainingTries < 3: - raise Exception( + raise SPINError( "Remaining tries for S-PIN is < 3. Bailing out for security reasons. " "To resume operation, please make sure the correct S-PIN has been set in the integration " "and then use the correct S-PIN once via the Volkswagen app." @@ -1199,13 +1347,207 @@ async def check_spin_state(self): return True + async def setClimater(self, vin, data, action): + """Execute climatisation actions.""" + action = "start" if action else "stop" + try: + response_raw = await self.post( + f"{BASE_API}/vehicle/v1/vehicles/{vin}/climatisation/{action}", + json=data, + return_raw=True, + ) + return await self._handle_action_result(response_raw) + except Exception as e: + raise APIError(f"Unknown error during setClimater: {str(e)}") from e + + async def setClimaterSettings(self, vin, data): + """Execute climatisation settings.""" + try: + response_raw = await self.put( + f"{BASE_API}/vehicle/v1/vehicles/{vin}/climatisation/settings", + json=data, + return_raw=True, + ) + return await self._handle_action_result(response_raw) + except Exception as e: + raise APIError(f"Unknown error during setClimaterSettings: {str(e)}") from e + + async def setAuxiliary(self, vin, data, action): + """Execute auxiliary climatisation actions.""" + action = "start" if action else "stop" + try: + response_raw = await self.post( + f"{BASE_API}/vehicle/v1/vehicles/{vin}/auxiliaryheating/{action}", + json=data, + return_raw=True, + ) + return await self._handle_action_result(response_raw) + except Exception as e: + raise APIError(f"Unknown error during setAuxiliary: {str(e)}") from e + + async def setWindowHeater(self, vin, action): + """Execute window heating actions.""" + action = "start" if action else "stop" + try: + response_raw = await self.post( + f"{BASE_API}/vehicle/v1/vehicles/{vin}/windowheating/{action}", + json={}, + return_raw=True, + ) + return await self._handle_action_result(response_raw) + except Exception as e: + raise APIError(f"Unknown error during setWindowHeater: {str(e)}") from e + + async def setCharging(self, vin, action): + """Execute charging actions.""" + action = "start" if action else "stop" + try: + response_raw = await self.post( + f"{BASE_API}/vehicle/v1/vehicles/{vin}/charging/{action}", + json={}, + return_raw=True, + ) + return await self._handle_action_result(response_raw) + except Exception as e: + raise APIError(f"Unknown error during setCharging: {str(e)}") from e + + async def setChargingSettings(self, vin, data): + """Execute charging actions.""" + try: + response_raw = await self.put( + f"{BASE_API}/vehicle/v1/vehicles/{vin}/charging/settings", + json=data, + return_raw=True, + ) + return await self._handle_action_result(response_raw) + except Exception as e: + raise APIError(f"Unknown error during setChargingSettings: {str(e)}") from e + + async def setChargingCareModeSettings(self, vin, data): + """Execute battery care mode actions.""" + try: + response_raw = await self.put( + f"{BASE_API}/vehicle/v1/vehicles/{vin}/charging/care/settings", + json=data, + return_raw=True, + ) + return await self._handle_action_result(response_raw) + except Exception as e: + raise APIError( + f"Unknown error during setChargingCareModeSettings: {str(e)}" + ) from e + + async def setReadinessBatterySupport(self, vin, data): + """Execute readiness battery support actions.""" + try: + response_raw = await self.put( + f"{BASE_API}/vehicle/v1/vehicles/{vin}/readiness/batterysupport", + json=data, + return_raw=True, + ) + return await self._handle_action_result(response_raw) + except Exception as e: + raise APIError( + f"Unknown error during setReadinessBatterySupport: {str(e)}" + ) from e + + async def setDepartureProfiles(self, vin, data): + """Execute departure timers actions.""" + try: + response_raw = await self.put( + f"{BASE_API}/vehicle/v1/vehicles/{vin}/departure/profiles", + json=data, + return_raw=True, + ) + return await self._handle_action_result(response_raw) + except Exception as e: + raise APIError( + f"Unknown error during setDepartureProfiles: {str(e)}" + ) from e + + async def setClimatisationTimers(self, vin, data): + """Execute climatisation timers actions.""" + try: + response_raw = await self.put( + f"{BASE_API}/vehicle/v1/vehicles/{vin}/climatisation/timers", + json=data, + return_raw=True, + ) + return await self._handle_action_result(response_raw) + except Exception as e: + raise APIError( + f"Unknown error during setClimatisationTimers: {str(e)}" + ) from e + + async def setAuxiliaryHeatingTimers(self, vin, data): + """Execute auxiliary heating timers actions.""" + try: + response_raw = await self.put( + f"{BASE_API}/vehicle/v1/vehicles/{vin}/auxiliaryheating/timers", + json=data, + return_raw=True, + ) + return await self._handle_action_result(response_raw) + except Exception as e: + raise APIError( + f"Unknown error during setAuxiliaryHeatingTimers: {str(e)}" + ) from e + + async def setDepartureTimers(self, vin, data): + """Execute departure timers actions.""" + try: + response_raw = await self.put( + f"{BASE_API}/vehicle/v1/vehicles/{vin}/departure/timers", + json=data, + return_raw=True, + ) + return await self._handle_action_result(response_raw) + except Exception as e: + raise APIError(f"Unknown error during setDepartureTimers: {str(e)}") from e + + async def setLock(self, vin, lock, spin): + """Remote lock and unlock actions.""" + await self.check_spin_state() + action = "lock" if lock else "unlock" + try: + response_raw = await self.post( + f"{BASE_API}/vehicle/v1/vehicles/{vin}/access/{action}", + json={"spin": spin}, + return_raw=True, + ) + return await self._handle_action_result(response_raw) + except Exception as e: + raise APIError(f"Unknown error during setLock: {str(e)}") from e + + async def setHonkAndFlash(self, vin, position): + """Remote Honk and Flash actions.""" + await self.check_spin_state() + try: + response_raw = await self.post( + f"{BASE_API}/vehicle/v1/vehicles/{vin}/honkandflash", + json={ + "userPosition": { + "longitude": position["lng"], + "latitude": position["lat"], + }, + "mode": "flash", + "duration_s": 15, + }, + return_raw=True, + ) + return await self._handle_action_result(response_raw) + except Exception as e: + raise APIError(f"Unknown error during setHonkAndFlash: {str(e)}") from e + # Token handling # - @property - async def validate_tokens(self): + async def validate_tokens(self) -> bool: """Validate expiry of tokens.""" - _LOGGER.debug("validate_tokens") - idtoken = self._session_tokens["identity"]["id_token"] - atoken = self._session_tokens["identity"]["access_token"] + try: + idtoken = self._session_tokens["identity"]["id_token"] + atoken = self._session_tokens["identity"]["access_token"] + except KeyError as error: + _LOGGER.warning("Token validation failed - missing token data: %s", error) + return False id_exp = jwt.decode( idtoken, options={"verify_signature": False, "verify_aud": False}, @@ -1223,7 +1565,7 @@ async def validate_tokens(self): # Check if tokens have expired, or expires now if now >= id_dt or now >= at_dt: - _LOGGER.warning("Tokens have expired. Try to fetch new tokens") + _LOGGER.debug("Tokens have expired. Try to fetch new tokens") if await self.refresh_tokens(): _LOGGER.debug("Successfully refreshed tokens") else: @@ -1237,40 +1579,8 @@ async def validate_tokens(self): return False return True - async def verify_tokens(self, token, type, client="Legacy"): - """Verify JWT against JWK(s).""" - _LOGGER.debug("verify_tokens, type=" + str(type) + ", client=" + str(client)) - if type == "identity": - req = await self._session.get(url="https://identity.vwgroup.io/v1/jwks") - keys = await req.json() - audience = [ - CLIENT[client].get("CLIENT_ID"), - "VWGMBB01DELIV1", - "https://api.vas.eu.dp15.vwg-connect.com", - "https://api.vas.eu.wcardp.io", - ] - else: - _LOGGER.debug("Not implemented") - return False - try: - pubkeys = {} - for jwk in keys["keys"]: - kid = jwk["kid"] - if jwk["kty"] == "RSA": - pubkeys[kid] = jwt.algorithms.RSAAlgorithm.from_jwk(to_json(jwk)) - - token_kid = jwt.get_unverified_header(token)["kid"] - - pubkey = pubkeys[token_kid] - jwt.decode(token, key=pubkey, algorithms=JWT_ALGORITHMS, audience=audience) - except Exception as error: - _LOGGER.exception("Failed to verify token, error: %s", error) - return False - return True - async def refresh_tokens(self): """Refresh tokens.""" - _LOGGER.debug("refresh_tokens") try: tHeaders = { "Accept-Encoding": "gzip, deflate, br", @@ -1283,31 +1593,35 @@ async def refresh_tokens(self): body = { "grant_type": "refresh_token", "refresh_token": self._session_tokens["identity"]["refresh_token"], - "client_id": CLIENT["Legacy"]["CLIENT_ID"], + "client_id": CLIENT_ID, } response = await self._session.post( - url="https://emea.bff.cariad.digital/login/v1/idk/token", + url=f"{BASE_API}/auth/v1/idk/oidc/token", headers=tHeaders, data=body, ) await self.update_service_status("token", response.status) if response.status == 200: tokens = await response.json() - # Verify Token - if not await self.verify_tokens(tokens["id_token"], "identity"): - _LOGGER.warning("Token could not be verified!") - _LOGGER.debug("refresh_tokens successful") + + if not tokens or "access_token" not in tokens: + _LOGGER.error("Invalid refresh token response: %s", tokens) + return False for token in tokens: self._session_tokens["identity"][token] = tokens[token] self._session_headers["Authorization"] = ( "Bearer " + self._session_tokens["identity"]["access_token"] ) + _LOGGER.debug("Successfully refreshed and updated tokens") else: + response_text = await response.text() _LOGGER.warning( - "Something went wrong when refreshing %s account tokens", BRAND + "Token refresh failed with status %s: %s", + response.status, + response_text, ) return False - except Exception as error: + except Exception as error: # pylint: disable=broad-exception-caught _LOGGER.warning("Could not refresh tokens: %s", error) return False else: @@ -1380,11 +1694,10 @@ def hash_spin(self, challenge, spin): spinArray.extend(byteChallenge) return hashlib.sha512(spinArray).hexdigest() - @property - async def validate_login(self): + async def validate_login(self) -> bool: """Check that we have a valid access token.""" try: - if not await self.validate_tokens: + if not await self.validate_tokens(): return False except OSError as error: _LOGGER.warning("Could not validate login: %s", error) @@ -1395,12 +1708,11 @@ async def validate_login(self): class vwid(): + # _LOGGER.setLevel(logging.DEBUG) connection = {} def __init__(self, session): self.session = session - # self.log = logging.getLogger(__name__) - # self.connection = {} def set_vin(self, vin): self.vin = vin @@ -1418,7 +1730,7 @@ async def get_status(self): # SOCERR-01: login problem, username, password wrong, account locked, etc. # SOCERR-02: vehicle not found in account, VIN wrong? try: - async with aiohttp.ClientSession(headers={'Connection': 'keep-alive'}) as session: + async with ClientSession(headers={'Connection': 'keep-alive'}) as session: _now = datetime.now(UTC).strftime('%Y-%m-%dT%H:%M:%SZ') data = {} data['charging'] = {} From 9c27398083989ca9289b3c6e68fd02aba771673e Mon Sep 17 00:00:00 2001 From: rleidner Date: Sun, 31 May 2026 19:57:49 +0200 Subject: [PATCH 2/2] fix pytest --- packages/modules/vehicles/vwid/libvwid.py | 28 ++++++++++++----------- 1 file changed, 15 insertions(+), 13 deletions(-) diff --git a/packages/modules/vehicles/vwid/libvwid.py b/packages/modules/vehicles/vwid/libvwid.py index 03a41fdad4..942d2a46cc 100755 --- a/packages/modules/vehicles/vwid/libvwid.py +++ b/packages/modules/vehicles/vwid/libvwid.py @@ -12,12 +12,14 @@ import logging from random import randint, random from urllib.parse import parse_qs, urljoin, urlparse - -from aiohttp import ClientSession, ClientTimeout, client_exceptions -from aiohttp.hdrs import METH_GET, METH_POST, METH_PUT -from bs4 import BeautifulSoup +import aiohttp +import bs4 import jwt +METH_GET = "GET" +METH_POST = "POST" +METH_PUT = "PUT" + ANDROID_PACKAGE_NAME = "com.volkswagen.weconnect" APP_URI = "weconnect://authenticated" BASE_API = "https://emea.bff.cariad.digital" @@ -661,7 +663,7 @@ async def get_authorization_page(self, authorization_endpoint: str) -> str: def extract_state_token(self, page_content: str) -> str | None: """Extract state token from a page.""" - soup = BeautifulSoup(page_content, "html.parser") + soup = bs4.BeautifulSoup(page_content, "html.parser") state_input = soup.select_one('input[name="state"]') if not state_input or not state_input.get("value"): _LOGGER.debug("State token not found.") @@ -683,7 +685,7 @@ async def post_form( # Handle explicit error 400 (form validation failure) if req.status == 400: page_content = await req.text() - soup = BeautifulSoup(page_content, "html.parser") + soup = bs4.BeautifulSoup(page_content, "html.parser") # Try both username + password fields in one pass for field_id in ("error-element-username", "error-element-password"): @@ -882,7 +884,7 @@ async def _login(self) -> bool: _LOGGER.error("Authentication error during login: %s", error) self._session_logged_in = False return False - except client_exceptions.ClientError as error: + except aiohttp.client_exceptions.ClientError as error: _LOGGER.error("Network error during login: %s", error) self._session_logged_in = False return False @@ -931,7 +933,7 @@ async def _request(self, method, url, return_raw=False, **kwargs): method, url, headers=self._session_headers, - timeout=ClientTimeout(total=TIMEOUT.seconds), + timeout=aiohttp.ClientTimeout(total=TIMEOUT.seconds), cookies=self._jarCookie, raise_for_status=False, **kwargs, @@ -984,7 +986,7 @@ async def _request(self, method, url, return_raw=False, **kwargs): if return_raw: res = response return res - except client_exceptions.ClientResponseError as httperror: + except aiohttp.client_exceptions.ClientResponseError as httperror: # Update service status await self.update_service_status(url, httperror.status) raise httperror from None @@ -997,7 +999,7 @@ async def get(self, url, vin="", tries=0): """Perform a get query.""" try: return await self._request(METH_GET, url) - except client_exceptions.ClientResponseError as error: + except aiohttp.client_exceptions.ClientResponseError as error: if error.status == 400: _LOGGER.error( 'Got HTTP 400 "Bad Request" from server, this request might be malformed or not implemented' @@ -1035,7 +1037,7 @@ async def post(self, url, vin="", tries=0, return_raw=False, **data): METH_POST, url, return_raw=return_raw, **data ) return await self._request(METH_POST, url, return_raw=return_raw) - except client_exceptions.ClientResponseError as error: + except aiohttp.client_exceptions.ClientResponseError as error: if error.status == 429 and tries < MAX_RETRIES_ON_RATE_LIMIT: delay = randint(1, 3 + tries * 2) _LOGGER.debug( @@ -1053,7 +1055,7 @@ async def put(self, url, vin="", tries=0, return_raw=False, **data): if data: return await self._request(METH_PUT, url, return_raw=return_raw, **data) return await self._request(METH_PUT, url, return_raw=return_raw) - except client_exceptions.ClientResponseError as error: + except aiohttp.client_exceptions.ClientResponseError as error: if error.status == 429 and tries < MAX_RETRIES_ON_RATE_LIMIT: delay = randint(1, 3 + tries * 2) _LOGGER.debug( @@ -1730,7 +1732,7 @@ async def get_status(self): # SOCERR-01: login problem, username, password wrong, account locked, etc. # SOCERR-02: vehicle not found in account, VIN wrong? try: - async with ClientSession(headers={'Connection': 'keep-alive'}) as session: + async with aiohttp.ClientSession(headers={'Connection': 'keep-alive'}) as session: _now = datetime.now(UTC).strftime('%Y-%m-%dT%H:%M:%SZ') data = {} data['charging'] = {}