Use dedicated httpx Client

custom_components/polestar_api/config_flow.py

@@ -85,14 +85,18 @@ async def _test_credentials(
             password=password,
             client_session=get_async_client(self.hass),
         )
-        await api_client.async_init()
 
-        if found_vins := api_client.get_available_vins():
-            _LOGGER.debug("Found %d VINs for %s", len(found_vins), username)
-        else:
-            _LOGGER.warning("No VINs found for %s", username)
-            raise NoCarsFoundException
+        try:
+            await api_client.async_init()
 
-        if vin and vin not in found_vins:
-            _LOGGER.warning("VIN %s not found for %s", vin, username)
-            raise VinNotFoundException
+            if found_vins := api_client.get_available_vins():
+                _LOGGER.debug("Found %d VINs for %s", len(found_vins), username)
+            else:
+                _LOGGER.warning("No VINs found for %s", username)
+                raise NoCarsFoundException
+
+            if vin and vin not in found_vins:
+                _LOGGER.warning("VIN %s not found for %s", vin, username)
+                raise VinNotFoundException
+        finally:
+            await api_client.async_logout()

custom_components/polestar_api/polestar.py

@@ -8,7 +8,7 @@
 import httpx
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity import DeviceInfo
-from homeassistant.helpers.httpx_client import get_async_client
+from homeassistant.helpers.httpx_client import create_async_httpx_client
 from homeassistant.util import Throttle
 
 from .const import DEFAULT_SCAN_INTERVAL, DOMAIN as POLESTAR_API_DOMAIN
@@ -228,7 +228,7 @@ def __init__(
         self.polestar_api = PolestarApi(
             username=username,
             password=password,
-            client_session=get_async_client(hass),
+            client_session=create_async_httpx_client(hass),
             vins=[vin] if vin else None,
             unique_id=self.unique_id,
         )

custom_components/polestar_api/pypolestar/auth.py

@@ -3,13 +3,14 @@
 import logging
 import os
 from datetime import datetime, timedelta, timezone
-from urllib.parse import urljoin
+from urllib.parse import urljoin, urlparse
 
 import httpx
 
 from .const import (
     HTTPX_TIMEOUT,
     OIDC_CLIENT_ID,
+    OIDC_COOKIES,
     OIDC_PROVIDER_BASE_URL,
     OIDC_REDIRECT_URI,
     OIDC_SCOPE,
@@ -38,21 +39,41 @@ def __init__(
         self.username = username
         self.password = password
         self.client_session = client_session
+
         self.access_token = None
         self.id_token = None
         self.refresh_token = None
         self.token_lifetime = None
         self.token_expiry = None
+
         self.oidc_configuration = {}
+        self.oidc_provider = OIDC_PROVIDER_BASE_URL
+        self.oidc_code_verifier: str | None = None
+        self.oidc_state: str | None = None
+
         self.latest_call_code = None
         self.logger = _LOGGER.getChild(unique_id) if unique_id else _LOGGER
-        self.oidc_provider = OIDC_PROVIDER_BASE_URL
-        self.code_verifier: str | None = None
-        self.state: str | None = None
 
     async def async_init(self) -> None:
         await self.update_oidc_configuration()
 
+    async def async_logout(self) -> None:
+        self.logger.debug("Logout")
+
+        domain = urlparse(OIDC_PROVIDER_BASE_URL).hostname
+        for name in OIDC_COOKIES:
+            self.logger.debug("Delete cookie %s in domain %s", name, domain)
+            self.client_session.cookies.delete(name=name, domain=domain)
+
+        self.access_token = None
+        self.id_token = None
+        self.refresh_token = None
+        self.token_lifetime = None
+        self.token_expiry = None
+
+        self.oidc_code_verifier = None
+        self.oidc_state = None
+
     async def update_oidc_configuration(self) -> None:
         result = await self.client_session.get(
             urljoin(OIDC_PROVIDER_BASE_URL, "/.well-known/openid-configuration")
@@ -98,9 +119,9 @@ async def get_token(self, refresh=False) -> None:
                 "redirect_uri": OIDC_REDIRECT_URI,
                 **(
                     {
-                        "code_verifier": self.code_verifier,
+                        "code_verifier": self.oidc_code_verifier,
                     }
-                    if self.code_verifier
+                    if self.oidc_code_verifier
                     else {}
                 ),
             }
@@ -213,13 +234,13 @@ async def _get_code(self) -> str | None:
     async def _get_resume_path(self):
         """Get Resume Path from Polestar."""
 
-        self.state = self.get_state()
+        self.oidc_state = self.get_state()
 
         params = {
             "response_type": "code",
             "client_id": OIDC_CLIENT_ID,
             "redirect_uri": OIDC_REDIRECT_URI,
-            "state": self.state,
+            "state": self.oidc_state,
             "code_challenge_method": "S256",
             "code_challenge": self.get_code_challenge(),
             "scope": OIDC_SCOPE,
@@ -247,7 +268,7 @@ def get_code_verifier() -> str:
         return b64urlencode(os.urandom(32))
 
     def get_code_challenge(self) -> str:
-        self.code_verifier = self.get_code_verifier()
+        self.oidc_code_verifier = self.get_code_verifier()
         m = hashlib.sha256()
-        m.update(self.code_verifier.encode())
+        m.update(self.oidc_code_verifier.encode())
         return b64urlencode(m.digest())

custom_components/polestar_api/pypolestar/const.py

@@ -14,5 +14,6 @@
 OIDC_CLIENT_ID = "l3oopkc_10"
 OIDC_REDIRECT_URI = "https://www.polestar.com/sign-in-callback"
 OIDC_SCOPE = "openid profile email customer:attributes"
+OIDC_COOKIES = ["PF", "PF.PERSISTENT"]
 
 API_MYSTAR_V2_URL = "https://pc-api.polestar.com/eu-north-1/mystar-v2/"

custom_components/polestar_api/pypolestar/polestar.py

@@ -90,6 +90,9 @@ async def async_init(self, verbose: bool = False) -> None:
         ):
             self.logger.warning("Could not found configured VINs %s", missing_vins)
 
+    async def async_logout(self) -> None:
+        await self.auth.async_logout()
+
     def get_available_vins(self) -> list[str]:
         """Get list of all available VINs"""
         return list(self.available_vins)